fix: preserve campaign timestamp ordering

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,

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", {}),

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]

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