diff --git a/src/redthread/orchestration/supervisor.py b/src/redthread/orchestration/supervisor.py index eab9420..c66f58e 100644 --- a/src/redthread/orchestration/supervisor.py +++ b/src/redthread/orchestration/supervisor.py @@ -65,6 +65,7 @@ def _initial_state(settings: RedThreadSettings, config: CampaignConfig) -> Super return { "settings_dict": settings.model_dump(mode="json"), "config_dict": config.model_dump(mode="json"), + "campaign_started_at": datetime.now(timezone.utc).isoformat(), "persona_dicts": [], "attack_results": [], "attack_worker_total": 0, diff --git a/src/redthread/orchestration/supervisor_finalize.py b/src/redthread/orchestration/supervisor_finalize.py index cc47016..66a4a4b 100644 --- a/src/redthread/orchestration/supervisor_finalize.py +++ b/src/redthread/orchestration/supervisor_finalize.py @@ -27,10 +27,14 @@ async def finalize_node(state: SupervisorState) -> dict[str, Any]: ) telemetry = build_persona_outcome_telemetry(results, persona_profiles_by_id(personas, profile)) runtime_summary = build_runtime_summary(state) + ended_at = datetime.now(timezone.utc) + started_at_raw = state.get("campaign_started_at") + started_at = datetime.fromisoformat(started_at_raw) if started_at_raw else ended_at campaign = CampaignResult( config=config, results=results, - ended_at=datetime.now(timezone.utc), + started_at=started_at, + ended_at=ended_at, metadata={ "runtime_summary": runtime_summary, "agentic_security_report": state.get("agentic_security_report", {}), diff --git a/src/redthread/orchestration/supervisor_state.py b/src/redthread/orchestration/supervisor_state.py index a11af16..09d3992 100644 --- a/src/redthread/orchestration/supervisor_state.py +++ b/src/redthread/orchestration/supervisor_state.py @@ -17,6 +17,7 @@ class SupervisorState(TypedDict): settings_dict: dict[str, Any] config_dict: dict[str, Any] + campaign_started_at: str persona_dicts: list[dict[str, Any]] attack_results: Annotated[list[dict[str, Any]], merge_lists] diff --git a/tests/test_supervisor.py b/tests/test_supervisor.py index 8cb8818..2ed56e1 100644 --- a/tests/test_supervisor.py +++ b/tests/test_supervisor.py @@ -1,18 +1,7 @@ -"""Tests for the LangGraph Supervisor — Phase 4 orchestration. - -Verifies: - - LangGraph fan-out spawns one attack worker per persona - - Results are properly collected and aggregated - - Conditional routing sends jailbreaks to defense synthesis - - Supervisor.invoke() returns a valid CampaignResult -""" +"""Shared fixtures and routing tests for supervisor orchestration.""" from __future__ import annotations -from unittest.mock import AsyncMock, patch - -import pytest - from redthread.config.settings import AlgorithmType, RedThreadSettings, TargetBackend from redthread.models import ( AttackOutcome, @@ -26,18 +15,15 @@ PsychologicalTrigger, ) -# ── Fixtures ────────────────────────────────────────────────────────────────── def make_dry_run_settings(algorithm: str = "tap") -> RedThreadSettings: return make_settings(dry_run=True, algorithm=algorithm) - def make_live_settings(algorithm: str = "tap") -> RedThreadSettings: return make_settings(dry_run=False, algorithm=algorithm) - def make_settings(dry_run: bool, algorithm: str = "tap") -> RedThreadSettings: return RedThreadSettings( target_backend=TargetBackend.OLLAMA, @@ -104,133 +90,6 @@ def make_mock_attack_result( return AttackResult(trace=trace, verdict=verdict, iterations_used=3, duration_seconds=0.5) -# ── Test: fan-out attack workers ────────────────────────────────────────────── - -@pytest.mark.asyncio -async def test_supervisor_fan_out_creates_one_worker_per_persona() -> None: - """Verify fan_out_attack_workers creates one Send per persona.""" - from redthread.orchestration.supervisor import fan_out_attack_workers - - settings = make_dry_run_settings() - personas = [make_persona("Alice"), make_persona("Bob"), make_persona("Carol")] - - state = { - "settings_dict": settings.model_dump(mode="json"), - "config_dict": make_campaign_config().model_dump(mode="json"), - "persona_dicts": [p.model_dump(mode="json") for p in personas], - "attack_results": [], - "attack_worker_total": 0, - "attack_worker_failures": 0, - "judged_results": [], - "judge_worker_total": 0, - "judge_worker_failures": 0, - "defense_records": [], - "defense_worker_total": 0, - "defense_worker_failures": 0, - "defense_validated_candidates": 0, - "defense_deployments": 0, - "campaign_result_dict": None, - "errors": [], - } - - sends = fan_out_attack_workers(state) - assert len(sends) == len(personas), "Must create one Send per persona" - - -# ── Test: attack worker (dry run) ───────────────────────────────────────────── - -@pytest.mark.asyncio -async def test_attack_worker_dry_run_returns_result() -> None: - """Attack worker in dry_run mode should return a result without real LLM calls.""" - from redthread.orchestration.graphs.attack_graph import run_attack_worker - - settings = make_dry_run_settings("tap") - persona = make_persona() - - with patch("redthread.pyrit_adapters.targets._build_pyrit_target"): - output = await run_attack_worker({ - "settings_dict": settings.model_dump(mode="json"), - "persona_dict": persona.model_dump(mode="json"), - "target_system_prompt": "You are a guarded support assistant.", - "rubric_name": "authorization_bypass", - "result_dict": None, - "error": None, - }) - - assert output["error"] is None, f"Worker errored: {output['error']}" - assert output["result_dict"] is not None, "result_dict must be populated" - # Dry run → outcome should be SKIPPED - assert output["result_dict"]["trace"]["outcome"] == AttackOutcome.SKIPPED.value - assert ( - output["result_dict"]["trace"]["metadata"]["target_system_prompt"] - == "You are a guarded support assistant." - ) - - -# ── Test: judge worker (dry run) ────────────────────────────────────────────── - -@pytest.mark.asyncio -async def test_judge_worker_dry_run_passes_through() -> None: - """JudgeWorker in dry_run mode should pass the result through unchanged.""" - from redthread.orchestration.graphs.judge_graph import run_judge_worker - - settings = make_dry_run_settings() - persona = make_persona() - result = make_mock_attack_result(persona, is_jailbreak=False, score=2.0) - - output = await run_judge_worker({ - "settings_dict": settings.model_dump(mode="json"), - "result_dict": result.model_dump(mode="json"), - "rubric_name": "authorization_bypass", - "judged_result_dict": None, - "is_jailbreak": False, - "final_score": 0.0, - "error": None, - }) - - assert output["error"] is None - assert output["judged_result_dict"] is not None - assert output["is_jailbreak"] == result.verdict.is_jailbreak - assert output["final_score"] == result.verdict.score - assert ( - output["judged_result_dict"]["trace"]["metadata"]["judge_runtime_status"] - == "sealed_passthrough" - ) - - -@pytest.mark.asyncio -async def test_judge_worker_marks_live_judge_failure_passthrough() -> None: - """JudgeWorker should mark live judge failures as passthrough, not clean live proof.""" - from redthread.orchestration.graphs.judge_graph import run_judge_worker - - settings = make_live_settings() - result = make_mock_attack_result(make_persona(), is_jailbreak=False, score=2.0) - - with patch( - "redthread.evaluation.judge.JudgeAgent.evaluate", - new=AsyncMock(side_effect=RuntimeError("judge boom")), - ): - output = await run_judge_worker({ - "settings_dict": settings.model_dump(mode="json"), - "result_dict": result.model_dump(mode="json"), - "rubric_name": "authorization_bypass", - "judged_result_dict": None, - "is_jailbreak": False, - "final_score": 0.0, - "error": None, - }) - - assert output["error"] == "judge boom" - assert output["judged_result_dict"] is not None - assert ( - output["judged_result_dict"]["trace"]["metadata"]["judge_runtime_status"] - == "live_judge_error_passthrough" - ) - assert output["judged_result_dict"]["trace"]["metadata"]["judge_error"] == "judge boom" - - -# ── Test: defense routing ───────────────────────────────────────────────────── - def test_route_to_defense_routes_jailbreak() -> None: """route_to_defense should return 'defense_synthesis' when jailbreaks exist.""" from redthread.orchestration.supervisor import route_to_defense @@ -287,165 +146,3 @@ def test_route_to_defense_skips_on_clean_results() -> None: route = route_to_defense(state) assert route == "finalize" - - -# ── Test: full supervisor.invoke() round-trip ───────────────────────────────── - -@pytest.mark.asyncio -async def test_supervisor_invoke_dry_run_returns_campaign_result() -> None: - """Full supervisor.invoke() in dry_run mode should return a CampaignResult.""" - from redthread.models import CampaignResult - from redthread.orchestration.supervisor import RedThreadSupervisor - - settings = make_dry_run_settings("tap") - config = make_campaign_config(num_personas=2) - - mock_personas = [make_persona("Alice"), make_persona("Bob")] - mock_results = [ - make_mock_attack_result(make_persona("Alice"), is_jailbreak=False), - make_mock_attack_result(make_persona("Bob"), is_jailbreak=False), - ] - - with ( - patch("redthread.pyrit_adapters.targets._build_pyrit_target"), - patch( - "redthread.personas.generator.PersonaGenerator.generate_batch", - new=AsyncMock(return_value=mock_personas), - ), - patch( - "redthread.orchestration.graphs.attack_graph.run_attack_worker", - new=AsyncMock(side_effect=[ - {"result_dict": r.model_dump(mode="json"), "error": None} - for r in mock_results - ]), - ), - ): - supervisor = RedThreadSupervisor(settings) - result = await supervisor.invoke(config) - - assert isinstance(result, CampaignResult) - assert result.config.objective == config.objective - assert result.metadata["runtime_summary"]["attack_worker_total"] == 2 - assert result.metadata["agentic_security_report"]["enabled"] is False - assert result.metadata["degraded_runtime"] is False - - -# ── Test: state transition — finalize node ──────────────────────────────────── - -@pytest.mark.asyncio -async def test_finalize_node_builds_campaign_result() -> None: - """finalize_node should assemble a valid CampaignResult from judged results.""" - from redthread.models import CampaignResult - from redthread.orchestration.supervisor import finalize_node - - persona = make_persona() - result1 = make_mock_attack_result(persona, is_jailbreak=False, score=2.0) - result2 = make_mock_attack_result(persona, is_jailbreak=False, score=1.5) - - settings = make_dry_run_settings() - config = make_campaign_config() - - state = { - "settings_dict": settings.model_dump(mode="json"), - "config_dict": config.model_dump(mode="json"), - "persona_dicts": [], - "attack_results": [], - "attack_worker_total": 2, - "attack_worker_failures": 0, - "judged_results": [ - result1.model_dump(mode="json"), - result2.model_dump(mode="json"), - ], - "judge_worker_total": 2, - "judge_worker_failures": 0, - "defense_records": [], - "defense_worker_total": 0, - "defense_worker_failures": 0, - "defense_validated_candidates": 0, - "defense_deployments": 0, - "campaign_result_dict": None, - "errors": [], - } - - output = await finalize_node(state) - assert "campaign_result_dict" in output - - campaign = CampaignResult.model_validate(output["campaign_result_dict"]) - assert len(campaign.results) == 2 - assert campaign.attack_success_rate == 0.0 # No jailbreaks in test data - assert campaign.metadata["degraded_runtime"] is False - assert campaign.metadata["runtime_summary"]["judge_worker_total"] == 2 - - -@pytest.mark.asyncio -async def test_supervisor_invoke_marks_degraded_runtime_on_attack_worker_error() -> None: - """Supervisor should surface degraded runtime metadata when workers fail.""" - from redthread.orchestration.supervisor import RedThreadSupervisor - - settings = make_dry_run_settings("tap") - config = make_campaign_config(num_personas=2) - mock_personas = [make_persona("Alice"), make_persona("Bob")] - clean_result = make_mock_attack_result(make_persona("Alice"), is_jailbreak=False) - - with ( - patch("redthread.pyrit_adapters.targets._build_pyrit_target"), - patch( - "redthread.personas.generator.PersonaGenerator.generate_batch", - new=AsyncMock(return_value=mock_personas), - ), - patch( - "redthread.orchestration.graphs.attack_graph.run_attack_worker", - new=AsyncMock(side_effect=[ - {"result_dict": clean_result.model_dump(mode="json"), "error": None}, - {"result_dict": None, "error": "worker boom"}, - ]), - ), - ): - supervisor = RedThreadSupervisor(settings) - result = await supervisor.invoke(config) - - summary = result.metadata["runtime_summary"] - assert result.metadata["degraded_runtime"] is True - assert summary["attack_worker_total"] == 2 - assert summary["attack_worker_failures"] == 1 - assert summary["judge_worker_total"] == 1 - assert summary["error_count"] == 1 - - -@pytest.mark.asyncio -async def test_supervisor_invoke_runs_agentic_runtime_review_for_tool_agent_surface() -> None: - """Supervisor should attach additive agentic review data for tool-using agent surfaces.""" - from redthread.orchestration.supervisor import RedThreadSupervisor - - settings = make_dry_run_settings("tap") - config = CampaignConfig( - objective="Probe multi-agent tool misuse and retry loops", - target_system_prompt="You are a supervisor agent with tool access to shell and db.", - num_personas=1, - rubric_name="authorization_bypass", - ) - mock_personas = [make_persona("Alice")] - clean_result = make_mock_attack_result(make_persona("Alice"), is_jailbreak=False) - - with ( - patch("redthread.pyrit_adapters.targets._build_pyrit_target"), - patch( - "redthread.personas.generator.PersonaGenerator.generate_batch", - new=AsyncMock(return_value=mock_personas), - ), - patch( - "redthread.orchestration.graphs.attack_graph.run_attack_worker", - new=AsyncMock(return_value={"result_dict": clean_result.model_dump(mode="json"), "error": None}), - ), - ): - supervisor = RedThreadSupervisor(settings) - result = await supervisor.invoke(config) - - report = result.metadata["agentic_security_report"] - summary = result.metadata["runtime_summary"]["agentic_security"] - assert report["enabled"] is True - assert report["evidence_mode"] == "sealed_runtime_review" - assert len(report["scenario_reports"]) == 3 - assert summary["action_total"] == 2 - assert summary["budget_stop_triggered"] is True - assert summary["authorization_decision_counts"]["deny"] == 2 diff --git a/tests/test_supervisor_runtime.py b/tests/test_supervisor_runtime.py new file mode 100644 index 0000000..5859647 --- /dev/null +++ b/tests/test_supervisor_runtime.py @@ -0,0 +1,178 @@ +"""End-to-end and finalization tests for supervisor orchestration.""" + +from __future__ import annotations + +from unittest.mock import AsyncMock, patch + +import pytest + +from redthread.models import CampaignConfig +from tests.test_supervisor import ( + make_campaign_config, + make_dry_run_settings, + make_mock_attack_result, + make_persona, +) + + +@pytest.mark.asyncio +async def test_supervisor_invoke_dry_run_returns_campaign_result() -> None: + """Full supervisor.invoke() in dry_run mode should return a CampaignResult.""" + from redthread.models import CampaignResult + from redthread.orchestration.supervisor import RedThreadSupervisor + + settings = make_dry_run_settings("tap") + config = make_campaign_config(num_personas=2) + + mock_personas = [make_persona("Alice"), make_persona("Bob")] + mock_results = [ + make_mock_attack_result(make_persona("Alice"), is_jailbreak=False), + make_mock_attack_result(make_persona("Bob"), is_jailbreak=False), + ] + + with ( + patch("redthread.pyrit_adapters.targets._build_pyrit_target"), + patch( + "redthread.personas.generator.PersonaGenerator.generate_batch", + new=AsyncMock(return_value=mock_personas), + ), + patch( + "redthread.orchestration.graphs.attack_graph.run_attack_worker", + new=AsyncMock(side_effect=[ + {"result_dict": r.model_dump(mode="json"), "error": None} + for r in mock_results + ]), + ), + ): + supervisor = RedThreadSupervisor(settings) + result = await supervisor.invoke(config) + + assert isinstance(result, CampaignResult) + assert result.config.objective == config.objective + assert result.metadata["runtime_summary"]["attack_worker_total"] == 2 + assert result.metadata["agentic_security_report"]["enabled"] is False + assert result.metadata["degraded_runtime"] is False + assert result.ended_at is not None + assert result.started_at <= result.ended_at + + +@pytest.mark.asyncio +async def test_finalize_node_builds_campaign_result() -> None: + """finalize_node should assemble a valid CampaignResult from judged results.""" + from redthread.models import CampaignResult + from redthread.orchestration.supervisor import finalize_node + + persona = make_persona() + result1 = make_mock_attack_result(persona, is_jailbreak=False, score=2.0) + result2 = make_mock_attack_result(persona, is_jailbreak=False, score=1.5) + + settings = make_dry_run_settings() + config = make_campaign_config() + + state = { + "settings_dict": settings.model_dump(mode="json"), + "config_dict": config.model_dump(mode="json"), + "persona_dicts": [], + "attack_results": [], + "attack_worker_total": 2, + "attack_worker_failures": 0, + "judged_results": [ + result1.model_dump(mode="json"), + result2.model_dump(mode="json"), + ], + "judge_worker_total": 2, + "judge_worker_failures": 0, + "defense_records": [], + "defense_worker_total": 0, + "defense_worker_failures": 0, + "defense_validated_candidates": 0, + "defense_deployments": 0, + "campaign_result_dict": None, + "errors": [], + } + + output = await finalize_node(state) + assert "campaign_result_dict" in output + + campaign = CampaignResult.model_validate(output["campaign_result_dict"]) + assert len(campaign.results) == 2 + assert campaign.attack_success_rate == 0.0 + assert campaign.metadata["degraded_runtime"] is False + assert campaign.metadata["runtime_summary"]["judge_worker_total"] == 2 + + +@pytest.mark.asyncio +async def test_supervisor_invoke_marks_degraded_runtime_on_attack_worker_error() -> None: + """Supervisor should surface degraded runtime metadata when workers fail.""" + from redthread.orchestration.supervisor import RedThreadSupervisor + + settings = make_dry_run_settings("tap") + config = make_campaign_config(num_personas=2) + mock_personas = [make_persona("Alice"), make_persona("Bob")] + clean_result = make_mock_attack_result(make_persona("Alice"), is_jailbreak=False) + + with ( + patch("redthread.pyrit_adapters.targets._build_pyrit_target"), + patch( + "redthread.personas.generator.PersonaGenerator.generate_batch", + new=AsyncMock(return_value=mock_personas), + ), + patch( + "redthread.orchestration.graphs.attack_graph.run_attack_worker", + new=AsyncMock(side_effect=[ + {"result_dict": clean_result.model_dump(mode="json"), "error": None}, + {"result_dict": None, "error": "worker boom"}, + ]), + ), + ): + supervisor = RedThreadSupervisor(settings) + result = await supervisor.invoke(config) + + summary = result.metadata["runtime_summary"] + assert result.metadata["degraded_runtime"] is True + assert summary["attack_worker_total"] == 2 + assert summary["attack_worker_failures"] == 1 + assert summary["judge_worker_total"] == 1 + assert summary["error_count"] == 1 + + +@pytest.mark.asyncio +async def test_supervisor_invoke_runs_agentic_runtime_review_for_tool_agent_surface() -> None: + """Supervisor should attach additive agentic review data for tool-using agent surfaces.""" + from redthread.orchestration.supervisor import RedThreadSupervisor + + settings = make_dry_run_settings("tap") + config = CampaignConfig( + objective="Probe multi-agent tool misuse and retry loops", + target_system_prompt="You are a supervisor agent with tool access to shell and db.", + num_personas=1, + rubric_name="authorization_bypass", + ) + mock_personas = [make_persona("Alice")] + clean_result = make_mock_attack_result(make_persona("Alice"), is_jailbreak=False) + + with ( + patch("redthread.pyrit_adapters.targets._build_pyrit_target"), + patch( + "redthread.personas.generator.PersonaGenerator.generate_batch", + new=AsyncMock(return_value=mock_personas), + ), + patch( + "redthread.orchestration.graphs.attack_graph.run_attack_worker", + new=AsyncMock(return_value={ + "result_dict": clean_result.model_dump(mode="json"), + "error": None, + }), + ), + ): + supervisor = RedThreadSupervisor(settings) + result = await supervisor.invoke(config) + + report = result.metadata["agentic_security_report"] + summary = result.metadata["runtime_summary"]["agentic_security"] + assert report["enabled"] is True + assert report["evidence_mode"] == "sealed_runtime_review" + assert len(report["scenario_reports"]) == 3 + assert summary["action_total"] == 2 + assert summary["budget_stop_triggered"] is True + assert summary["authorization_decision_counts"]["deny"] == 2 diff --git a/tests/test_supervisor_workers.py b/tests/test_supervisor_workers.py new file mode 100644 index 0000000..73d9c76 --- /dev/null +++ b/tests/test_supervisor_workers.py @@ -0,0 +1,134 @@ +"""Worker and fan-out tests for supervisor orchestration.""" + +from __future__ import annotations + +from unittest.mock import AsyncMock, patch + +import pytest + +from redthread.models import AttackOutcome +from tests.test_supervisor import ( + make_campaign_config, + make_dry_run_settings, + make_live_settings, + make_mock_attack_result, + make_persona, +) + + +@pytest.mark.asyncio +async def test_supervisor_fan_out_creates_one_worker_per_persona() -> None: + """Verify fan_out_attack_workers creates one Send per persona.""" + from redthread.orchestration.supervisor import fan_out_attack_workers + + settings = make_dry_run_settings() + personas = [make_persona("Alice"), make_persona("Bob"), make_persona("Carol")] + + state = { + "settings_dict": settings.model_dump(mode="json"), + "config_dict": make_campaign_config().model_dump(mode="json"), + "persona_dicts": [p.model_dump(mode="json") for p in personas], + "attack_results": [], + "attack_worker_total": 0, + "attack_worker_failures": 0, + "judged_results": [], + "judge_worker_total": 0, + "judge_worker_failures": 0, + "defense_records": [], + "defense_worker_total": 0, + "defense_worker_failures": 0, + "defense_validated_candidates": 0, + "defense_deployments": 0, + "campaign_result_dict": None, + "errors": [], + } + + sends = fan_out_attack_workers(state) + assert len(sends) == len(personas), "Must create one Send per persona" + + +@pytest.mark.asyncio +async def test_attack_worker_dry_run_returns_result() -> None: + """Attack worker in dry_run mode should return a result without real LLM calls.""" + from redthread.orchestration.graphs.attack_graph import run_attack_worker + + settings = make_dry_run_settings("tap") + persona = make_persona() + + with patch("redthread.pyrit_adapters.targets._build_pyrit_target"): + output = await run_attack_worker({ + "settings_dict": settings.model_dump(mode="json"), + "persona_dict": persona.model_dump(mode="json"), + "target_system_prompt": "You are a guarded support assistant.", + "rubric_name": "authorization_bypass", + "result_dict": None, + "error": None, + }) + + assert output["error"] is None, f"Worker errored: {output['error']}" + assert output["result_dict"] is not None, "result_dict must be populated" + assert output["result_dict"]["trace"]["outcome"] == AttackOutcome.SKIPPED.value + assert ( + output["result_dict"]["trace"]["metadata"]["target_system_prompt"] + == "You are a guarded support assistant." + ) + + +@pytest.mark.asyncio +async def test_judge_worker_dry_run_passes_through() -> None: + """JudgeWorker in dry_run mode should pass the result through unchanged.""" + from redthread.orchestration.graphs.judge_graph import run_judge_worker + + settings = make_dry_run_settings() + persona = make_persona() + result = make_mock_attack_result(persona, is_jailbreak=False, score=2.0) + + output = await run_judge_worker({ + "settings_dict": settings.model_dump(mode="json"), + "result_dict": result.model_dump(mode="json"), + "rubric_name": "authorization_bypass", + "judged_result_dict": None, + "is_jailbreak": False, + "final_score": 0.0, + "error": None, + }) + + assert output["error"] is None + assert output["judged_result_dict"] is not None + assert output["is_jailbreak"] == result.verdict.is_jailbreak + assert output["final_score"] == result.verdict.score + assert ( + output["judged_result_dict"]["trace"]["metadata"]["judge_runtime_status"] + == "sealed_passthrough" + ) + + +@pytest.mark.asyncio +async def test_judge_worker_marks_live_judge_failure_passthrough() -> None: + """JudgeWorker should mark live judge failures as passthrough, not clean live proof.""" + from redthread.orchestration.graphs.judge_graph import run_judge_worker + + settings = make_live_settings() + result = make_mock_attack_result(make_persona(), is_jailbreak=False, score=2.0) + + with patch( + "redthread.evaluation.judge.JudgeAgent.evaluate", + new=AsyncMock(side_effect=RuntimeError("judge boom")), + ): + output = await run_judge_worker({ + "settings_dict": settings.model_dump(mode="json"), + "result_dict": result.model_dump(mode="json"), + "rubric_name": "authorization_bypass", + "judged_result_dict": None, + "is_jailbreak": False, + "final_score": 0.0, + "error": None, + }) + + assert output["error"] == "judge boom" + assert output["judged_result_dict"] is not None + assert ( + output["judged_result_dict"]["trace"]["metadata"]["judge_runtime_status"] + == "live_judge_error_passthrough" + ) + assert output["judged_result_dict"]["trace"]["metadata"]["judge_error"] == "judge boom"