[WIP] Replace duck-typing with typed AttackResult model

hackagent/attacks/base.py

@@ -2,7 +2,9 @@
 # SPDX-License-Identifier: Apache-2.0
 
 import abc
-from typing import Any, Dict
+from typing import Any, Dict, List
+
+from hackagent.attacks.types import AttackResult
 
 
 class BaseAttack(abc.ABC):
@@ -64,7 +66,7 @@ def _setup(self):
         pass
 
     @abc.abstractmethod
-    def run(self, **kwargs: Any) -> Any:
+    def run(self, **kwargs: Any) -> List[AttackResult]:
         """
         Executes the attack logic.
 
@@ -80,12 +82,7 @@ def run(self, **kwargs: Any) -> Any:
                 - target_model: The model to attack
 
         Returns:
-            Attack-specific results. The format varies by implementation but
-            typically includes:
-            - adversarial_examples: Generated adversarial inputs
-            - success_metrics: Attack success rates and statistics
-            - detailed_results: Comprehensive result data (e.g., pandas DataFrame)
-            - attack_report: Summary of attack performance
+            A list of :class:`~hackagent.attacks.types.AttackResult` instances.
 
         Raises:
             NotImplementedError: If the method is not implemented by a subclass.

hackagent/attacks/orchestrator.py

@@ -56,6 +56,11 @@
     DEFAULT_REMOTE_ROLE_ENDPOINT,
 )
 from hackagent.server.storage.enums import StatusEnum
+from hackagent.attacks.types import (
+    AttackResult,
+    attack_results_to_rows,
+    flatten_run_result,
+)
 
 if TYPE_CHECKING:
     from hackagent.agent import HackAgent
@@ -1446,32 +1451,14 @@ def _get_attack_impl_kwargs(
             "agent_router": agent_router,
         }
 
-    @staticmethod
-    def _normalize_attack_results(results: Any) -> List[Dict[str, Any]]:
-        """Normalize heterogeneous attack outputs into a list of row dicts."""
-        if results is None:
-            return []
-        if isinstance(results, list):
-            return results
-        if isinstance(results, dict):
-            evaluated = results.get("evaluated")
-            if isinstance(evaluated, list):
-                return evaluated
-            for key in ("rows", "results", "data", "items"):
-                value = results.get(key)
-                if isinstance(value, list):
-                    return value
-            return []
-        return []
-
     def _execute_local_attack(
         self,
         attack_id: str,
         run_id: str,
         attack_params: Dict[str, Any],
         attack_config: Dict[str, Any],
         run_config_override: Optional[Dict[str, Any]],
-    ) -> Any:
+    ) -> List[AttackResult]:
         """
         Execute attack locally using technique implementation.
 
@@ -1560,7 +1547,7 @@ def _execute_local_attack(
                     f"goal_batch_workers={goal_batch_workers} (parallel goals per batch)"
                 )
 
-                all_results: List[Dict[str, Any]] = []
+                all_results: List[AttackResult] = []
                 batch_timings: List[float] = []
 
                 for batch_idx, (batch_start_idx, batch_goals) in enumerate(batches):
@@ -1576,11 +1563,7 @@ def _execute_local_attack(
                         # Global run status is finalized once in execute().
                         attack_impl.config["_suppress_run_status_updates"] = True
                         batch_params = {**attack_params, "goals": batch_goals}
-                        # attack_impl.run() may return a dict (e.g. baseline's
-                        # {"evaluated": [...], "summary": [...]}) rather than a
-                        # flat row list — normalize before aggregating, otherwise
-                        # extend() below would iterate the dict's *keys*.
-                        batch_results = self._normalize_attack_results(
+                        batch_results = flatten_run_result(
                             attack_impl.run(**batch_params)
                         )
                     else:
@@ -1591,7 +1574,7 @@ def _run_single_goal(
                             goal_idx_goal: Tuple[int, str],
                             _batch_label: str = batch_label,
                             _batch_start_idx: int = batch_start_idx,
-                        ) -> Tuple[int, List[Dict[str, Any]]]:
+                        ) -> Tuple[int, List[AttackResult]]:
                             goal_idx, goal = goal_idx_goal
 
                             # Label thread for _BatchContextFilter
@@ -1614,16 +1597,14 @@ def _run_single_goal(
                             }
                             local_impl = self.attack_impl_class(**local_impl_kwargs)
                             goal_params = {**attack_params, "goals": [goal]}
-                            # Normalize here too — same dict-vs-list return shape
-                            # concern as the sequential path above.
-                            goal_results = self._normalize_attack_results(
+                            goal_results = flatten_run_result(
                                 local_impl.run(**goal_params)
                             )
 
                             logger.info(f"Goal done ({len(goal_results)} results)")
                             return goal_idx, goal_results
 
-                        per_goal_results: Dict[int, List[Dict[str, Any]]] = {}
+                        per_goal_results: Dict[int, List[AttackResult]] = {}
 
                         # Install a LogRecordFactory so *all* log records,
                         # regardless of logger/handler routing, get the batch
@@ -1677,7 +1658,7 @@ def _batch_record_factory(*args, **kwargs):
                 )
                 return all_results
 
-            results = attack_impl.run(**attack_params)
+            results = flatten_run_result(attack_impl.run(**attack_params))
             logger.info(f"{self.attack_type} attack completed")
             return results
         finally:
@@ -1870,7 +1851,7 @@ def execute(
                 attack_config=attack_config,
                 run_config_override=effective_run_config,
             )
-            normalized_results = self._normalize_attack_results(results)
+            normalized_results = attack_results_to_rows(results)
 
             # =========================
             # RUN EVALUATION PIPELINE
@@ -1904,10 +1885,11 @@ def execute(
                     # (setting is_success/best_score per row) rather than via
                     # a separate judge pass, so it only needs the shared
                     # post-processing (default-filling + sync/ASR logging).
-                    final_results = evaluator._postprocess_inline_judge_results(
-                        normalized_results, attack_label="PAIR"
+                    final_results = flatten_run_result(
+                        evaluator._postprocess_inline_judge_results(
+                            normalized_results, attack_label="PAIR"
+                        )
                     )
-                    final_results = self._normalize_attack_results(final_results)
                     evaluator.prepare_and_sync(final_results, run_id)
                     logger.info("PAIR judge evaluation pipeline completed")
                 else:
@@ -1924,8 +1906,9 @@ def execute(
                     )
 
                     # Run evaluation pipeline
-                    final_results = evaluator.run_full_evaluation(normalized_results)
-                    final_results = self._normalize_attack_results(final_results)
+                    final_results = flatten_run_result(
+                        evaluator.run_full_evaluation(normalized_results)
+                    )
 
                     # Sync metrics to backend
                     evaluator.prepare_and_sync(final_results, run_id)
@@ -1942,7 +1925,7 @@ def execute(
                     error=e,
                     logger=logger,
                 )
-                final_results = results  # fallback
+                final_results = normalized_results  # fallback
                 if _tui_event_bus is not None:
                     _tui_event_bus.emit(
                         "step_ended",

hackagent/attacks/techniques/advprefix/attack.py

@@ -20,6 +20,7 @@
 from hackagent.server.client import AuthenticatedClient
 from hackagent.router.router import AgentRouter
 from hackagent.attacks.techniques.base import BaseAttack
+from hackagent.attacks.types import AttackResult, rows_to_attack_results
 from hackagent.attacks.shared.tui import with_tui_logging
 
 # Import step execution functions from same package
@@ -269,7 +270,7 @@ def _get_pipeline_steps(self):
         ]
 
     @with_tui_logging(logger_name="hackagent.attacks", level=logging.INFO)
-    def run(self, goals: List[str]) -> List[Dict]:
+    def run(self, goals: Optional[List[str]] = None, **kwargs) -> List[AttackResult]:
         """
         Executes the full prefix generation pipeline.
 
@@ -285,6 +286,7 @@ def run(self, goals: List[str]) -> List[Dict]:
             List of dictionaries containing the final selected prefixes,
             or empty list if no prefixes were generated.
         """
+        goals = goals or []
         if not goals:
             return []
 
@@ -343,7 +345,7 @@ def run(self, goals: List[str]) -> List[Dict]:
             # Finalize pipeline-level tracking
             coordinator.finalize_pipeline(results)
 
-            return results if results is not None else []
+            return rows_to_attack_results(results)
 
         except Exception:
             # Crash-safe: mark all unfinalized goals as failed

hackagent/attacks/techniques/autodan_turbo/attack.py

@@ -5,10 +5,11 @@
 import copy
 import logging
 import os
-from typing import Any, Dict, List
+from typing import Any, Dict, List, Optional
 
 from hackagent.attacks.shared.tui import with_tui_logging
 from hackagent.attacks.techniques.base import BaseAttack
+from hackagent.attacks.types import AttackResult, rows_to_attack_results
 
 from . import autodan_eval as evaluation, lifelong, warm_up
 from .config import DEFAULT_AUTODAN_TURBO_CONFIG, AutoDANTurboConfig
@@ -152,7 +153,7 @@ def _get_pipeline_steps(self):
         return []  # Managed manually in run() (like PAIR)
 
     @with_tui_logging(logger_name="hackagent.attacks", level=logging.INFO)
-    def run(self, goals: List[str]) -> List[Dict[str, Any]]:
+    def run(self, goals: Optional[List[str]] = None, **kwargs) -> List[AttackResult]:
         """Execute full AutoDAN-Turbo pipeline.
 
         Pipeline mapping to paper/integration:
@@ -169,6 +170,7 @@ def run(self, goals: List[str]) -> List[Dict[str, Any]]:
         Raises:
             Exception: Re-raises any runtime failure after coordinator finalization.
         """
+        goals = goals or []
         if not goals:
             return []
 
@@ -379,7 +381,7 @@ def run(self, goals: List[str]) -> List[Dict[str, Any]]:
             os.makedirs(output_dir, exist_ok=True)
             strategy_lib.save(f"{output_dir}/strategy_library")
 
-            return results
+            return rows_to_attack_results(results)
 
         except Exception:
             coordinator.finalize_on_error("AutoDAN-Turbo failed")

hackagent/attacks/techniques/base.py

@@ -31,6 +31,7 @@
 from hackagent.logger import get_logger
 from typing import Any, Dict, List, Optional
 
+from hackagent.attacks.types import AttackResult
 from hackagent.router.tracking import StepTracker, TrackingCoordinator
 
 logger = get_logger(__name__)
@@ -451,7 +452,7 @@ def _get_pipeline_steps(self) -> List[Dict]:
         pass
 
     @abc.abstractmethod
-    def run(self, **kwargs) -> Any:
+    def run(self, **kwargs) -> List[AttackResult]:
         """
         Execute the attack technique.
 
@@ -460,12 +461,12 @@ def run(self, **kwargs) -> Any:
         2. Define pipeline with self._get_pipeline_steps()
         3. Execute pipeline with self._execute_pipeline()
         4. Finalize with coordinator.finalize_all_goals() and coordinator.finalize_pipeline()
-        5. Return results
+        5. Return results as ``list[AttackResult]``
 
         Args:
             **kwargs: Technique-specific parameters (e.g., goals, prompts)
 
         Returns:
-            Attack results (format varies by implementation)
+            A list of :class:`~hackagent.attacks.types.AttackResult` instances.
         """
         pass

hackagent/attacks/techniques/baseline/attack.py

@@ -16,6 +16,7 @@
 from hackagent.router.router import AgentRouter
 from hackagent.attacks.techniques.base import BaseAttack
 from hackagent.attacks.shared.tui import with_tui_logging
+from hackagent.attacks.types import AttackResult, rows_to_attack_results
 
 from . import generation
 from .config import DEFAULT_BASELINE_CONFIG
@@ -159,18 +160,19 @@ def _build_step_args(
         return args
 
     @with_tui_logging(logger_name="hackagent.attacks", level=logging.INFO)
-    def run(self, goals: List[str]) -> Dict[str, Any]:
+    def run(self, goals: Optional[List[str]] = None, **kwargs) -> List[AttackResult]:
         """
         Execute baseline attack (direct goal submission).
 
         Args:
             goals: List of goal strings to send directly.
 
         Returns:
-            Dictionary with 'evaluated' and 'summary' DataFrames.
+            A list of :class:`~hackagent.attacks.types.AttackResult` instances.
         """
+        goals = goals or []
         if not goals:
-            return {"evaluated": [], "summary": []}
+            return []
 
         coordinator = self._initialize_coordinator(
             attack_type="Baseline",
@@ -187,7 +189,9 @@ def success_check(output):
                 return output and isinstance(output, dict)
 
             coordinator.finalize_pipeline(results, success_check)
-            return results if results else {"evaluated": [], "summary": []}
+            return rows_to_attack_results(
+                results if results else {"evaluated": [], "summary": []}
+            )
 
         except Exception as e:
             self.logger.error(f"Pipeline failed: {e}", exc_info=True)

hackagent/attacks/techniques/bon/attack.py

@@ -39,6 +39,7 @@
 from hackagent.server.client import AuthenticatedClient
 from hackagent.router.router import AgentRouter
 from hackagent.attacks.techniques.base import BaseAttack
+from hackagent.attacks.types import AttackResult, rows_to_attack_results
 from hackagent.attacks.shared.tui import with_tui_logging
 
 from hackagent.attacks.evaluator.evaluation_step import BaseEvaluationStep
@@ -206,7 +207,7 @@ def _get_pipeline_steps(self) -> List[Dict]:
     # ------------------------------------------------------------------
 
     @with_tui_logging(logger_name="hackagent.attacks", level=logging.INFO)
-    def run(self, goals: List[str]) -> List[Dict]:
+    def run(self, goals: Optional[List[str]] = None, **kwargs) -> List[AttackResult]:
         """Execute the full BoN attack pipeline.
 
         The generation step performs the multi-step BoN search **and** inline
@@ -220,6 +221,7 @@ def run(self, goals: List[str]) -> List[Dict]:
         Returns:
             List of result dictionaries, or empty list if no goals provided.
         """
+        goals = goals or []
         if not goals:
             return []
 
@@ -273,7 +275,7 @@ def run(self, goals: List[str]) -> List[Dict]:
             coordinator.log_summary()
             coordinator.finalize_pipeline(results)
 
-            return results if results is not None else []
+            return rows_to_attack_results(results)
 
         except Exception:
             coordinator.finalize_on_error("BoN pipeline failed with exception")

hackagent/attacks/techniques/cipherchat/attack.py

@@ -16,6 +16,7 @@
 
 from hackagent.attacks.shared.tui import with_tui_logging
 from hackagent.attacks.techniques.base import BaseAttack
+from hackagent.attacks.types import AttackResult, rows_to_attack_results
 from hackagent.server.client import AuthenticatedClient
 from hackagent.router.router import AgentRouter
 
@@ -160,7 +161,8 @@ def _get_pipeline_steps(self) -> List[Dict]:
         ]
 
     @with_tui_logging(logger_name="hackagent.attacks", level=logging.INFO)
-    def run(self, goals: List[str]) -> List[Dict]:
+    def run(self, goals: Optional[List[str]] = None, **kwargs) -> List[AttackResult]:
+        goals = goals or []
         if not goals:
             return []
 
@@ -203,7 +205,7 @@ def run(self, goals: List[str]) -> List[Dict]:
             coordinator.finalize_all_goals(results)
             coordinator.log_summary()
             coordinator.finalize_pipeline(results)
-            return results if results is not None else []
+            return rows_to_attack_results(results)
 
         except Exception:
             coordinator.finalize_on_error("CipherChat pipeline failed with exception")