FIX: export partial adversarial benchmark results

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

Copilot-Session: 5b5e066d-8e77-4ba9-8fc8-56f4745ef365

Author: hannahwestra25

.azuredevops/adversarial-benchmark.yml

@@ -276,21 +276,27 @@ jobs:
                 "$artifact_dir/scenario-result.json"
             )"
 
+            export_exit_code=0
+            if [[ -n "$result_id" ]]; then
+              echo "export-results" > "$artifact_dir/phase.txt"
+              set +e
+              uv run python -m build_scripts.export_adversarial_benchmark_result \
+                --scenario-result-id "$result_id" \
+                --output-dir "$artifact_dir"
+              export_exit_code=$?
+              set -e
+              echo "$export_exit_code" > "$artifact_dir/export-exit-code.txt"
+            fi
+
             if [[ "$benchmark_exit_code" -eq 0 ]]; then
               if [[ -z "$result_id" || "$run_status" != "COMPLETED" ]]; then
                 echo "Benchmark exited successfully without a completed persisted scenario result." >&2
                 exit 3
               fi
-
-              echo "export-results" > "$artifact_dir/phase.txt"
-              uv run pyrit_scan \
-                scenario-results "$result_id" \
-                --view overview \
-                > "$artifact_dir/overview.txt" 2>&1
-              uv run pyrit_scan \
-                scenario-results "$result_id" \
-                --view attacks \
-                > "$artifact_dir/attacks.txt" 2>&1
+              if [[ "$export_exit_code" -ne 0 ]]; then
+                echo "Failed to export completed benchmark result." >&2
+                exit "$export_exit_code"
+              fi
             fi
 
             if [[ "$benchmark_exit_code" -eq 0 ]]; then
@@ -367,6 +373,7 @@ jobs:
               "last_phase": read_text("phase.txt"),
               "execution_exit_code": read_text("execution-exit-code.txt"),
               "benchmark_exit_code": read_text("benchmark-exit-code.txt"),
+              "export_exit_code": read_text("export-exit-code.txt"),
               "objective_target": os.environ["OBJECTIVE_TARGET_INPUT"],
               "adversarial_targets": os.environ["ADVERSARIAL_TARGETS_INPUT"].split(),
               "technique_preset": preset,

build_scripts/export_adversarial_benchmark_result.py

@@ -0,0 +1,80 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT license.
+
+"""Export readable partial or completed adversarial benchmark results from SQLite."""
+
+import argparse
+import asyncio
+import contextlib
+from pathlib import Path
+
+from pyrit.cli._output import print_attacks_table
+from pyrit.cli._results import build_attacks_table_payload
+from pyrit.memory import CentralMemory
+from pyrit.models import ScenarioResult
+from pyrit.output.scenario_result.pretty import PrettyScenarioResultMemoryPrinter
+from pyrit.output.sink import FileSink
+from pyrit.setup import SQLITE, initialize_pyrit_async
+
+
+async def _load_result_async(*, scenario_result_id: str) -> ScenarioResult:
+    """Load one persisted scenario result, regardless of terminal state."""
+    await initialize_pyrit_async(
+        memory_db_type=SQLITE,
+        load_defaults=False,
+        env_files=[],
+        silent=True,
+    )
+    results = CentralMemory.get_memory_instance().get_scenario_results(
+        scenario_result_ids=[scenario_result_id],
+    )
+    if not results:
+        raise ValueError(f"Scenario result '{scenario_result_id}' was not found in SQLite memory.")
+    return results[0]
+
+
+async def _write_overview_async(*, result: ScenarioResult, output_dir: Path) -> None:
+    """Write the existing scenario overview without terminal color codes."""
+    printer = PrettyScenarioResultMemoryPrinter(
+        sink=FileSink(path=output_dir / "overview.txt"),
+        enable_colors=False,
+    )
+    await printer.write_async(result)
+
+
+def _write_attacks(*, result: ScenarioResult, output_dir: Path) -> None:
+    """Write machine-readable and console-style partial attack tables."""
+    payload = build_attacks_table_payload(
+        result=result,
+        scenario_result_id=str(result.id),
+    )
+    (output_dir / "attacks.json").write_text(payload.model_dump_json(indent=2), encoding="utf-8")
+    with open(output_dir / "attacks.txt", "w", encoding="utf-8") as output:
+        with contextlib.redirect_stdout(output):
+            print_attacks_table(payload=payload)
+
+
+async def _export_async(*, scenario_result_id: str, output_dir: Path) -> None:
+    """Export all readable result views."""
+    result = await _load_result_async(scenario_result_id=scenario_result_id)
+    output_dir.mkdir(parents=True, exist_ok=True)
+    await _write_overview_async(result=result, output_dir=output_dir)
+    await asyncio.to_thread(_write_attacks, result=result, output_dir=output_dir)
+
+
+def main() -> None:
+    """Run the result exporter."""
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--scenario-result-id", required=True)
+    parser.add_argument("--output-dir", type=Path, required=True)
+    args = parser.parse_args()
+    asyncio.run(
+        _export_async(
+            scenario_result_id=args.scenario_result_id,
+            output_dir=args.output_dir,
+        )
+    )
+
+
+if __name__ == "__main__":
+    main()