FIX: CLI Usability Updates

pyrit/backend/services/scenario_run_service.py

@@ -18,6 +18,8 @@
 from pyrit.memory import CentralMemory
 from pyrit.models import AttackOutcome, ScenarioResult, ScenarioRunState
 from pyrit.models.catalog.scenario import (
+    AttackErrorSummary,
+    AttackRetrySummary,
     RunScenarioRequest,
     ScenarioRunSummary,
 )
@@ -597,6 +599,38 @@ def _build_response_from_db(self, *, scenario_result: ScenarioResult) -> Scenari
         completed_attacks = total_attacks
         techniques_used = scenario_result.get_techniques_used()
 
+        # Surface per-attack errors and retry pressure regardless of overall run status:
+        # a COMPLETED scenario can still hide errored objectives or rate-limit retries.
+        failed_attacks: list[AttackErrorSummary] = []
+        attack_retries: list[AttackRetrySummary] = []
+        total_retries = 0
+        for atomic_attack_name, results in scenario_result.attack_results.items():
+            for attack_result in results:
+                retries = getattr(attack_result, "total_retries", 0)
+                if isinstance(retries, int):
+                    total_retries += retries
+
+                retry_events = getattr(attack_result, "retry_events", None)
+                if isinstance(retry_events, list) and retry_events:
+                    attack_retries.append(
+                        AttackRetrySummary(
+                            attack_result_id=str(attack_result.attack_result_id),
+                            atomic_attack_name=atomic_attack_name,
+                            retries=retry_events,
+                        )
+                    )
+
+                if attack_result.outcome == AttackOutcome.ERROR:
+                    failed_attacks.append(
+                        AttackErrorSummary(
+                            atomic_attack_name=atomic_attack_name,
+                            objective=attack_result.objective,
+                            error_type=attack_result.error_type,
+                            error_message=attack_result.error_message,
+                            total_retries=retries if isinstance(retries, int) else 0,
+                        )
+                    )
+
         return ScenarioRunSummary(
             scenario_result_id=scenario_result_id,
             scenario_name=scenario_result.scenario_name,
@@ -610,6 +644,9 @@ def _build_response_from_db(self, *, scenario_result: ScenarioResult) -> Scenari
             total_attacks=total_attacks,
             completed_attacks=completed_attacks,
             objective_achieved_rate=scenario_result.objective_achieved_rate(),
+            failed_attacks=failed_attacks,
+            attack_retries=attack_retries,
+            total_retries=total_retries,
             labels=scenario_result.labels,
             completed_at=scenario_result.completion_time,
         )

pyrit/cli/_output.py

@@ -50,6 +50,18 @@ def _header(text: str) -> None:
     _cprint(f"\n  {text}", color="cyan", bold=True)
 
 
+def _truncate_text(text: str, max_length: int) -> str:
+    """
+    Truncate *text* to *max_length* characters, appending an ellipsis when cut.
+
+    Returns:
+        str: The original text, or a truncated copy ending in ``...``.
+    """
+    if len(text) <= max_length:
+        return text
+    return text[: max_length - 3].rstrip() + "..."
+
+
 def _wrap(*, text: str, indent: str, width: int = 78) -> str:
     """
     Word-wrap *text* with the given *indent*.
@@ -262,6 +274,54 @@ def print_dataset_list(*, items: list[dict[str, Any]]) -> None:
 # ---------------------------------------------------------------------------
 
 
+def _format_retry_location(retry: Any) -> str:
+    """
+    Build a human-readable "on <component>, endpoint <url>" clause for a retry.
+
+    Returns:
+        str: The location clause, or an empty string when no context is available.
+    """
+    bits: list[str] = []
+    if retry.component_role:
+        role = retry.component_role.replace("_", " ")
+        bits.append(f"{role} {retry.component_name}" if retry.component_name else role)
+    if retry.endpoint:
+        bits.append(f"endpoint {retry.endpoint}")
+    return " on " + ", ".join(bits) if bits else ""
+
+
+def print_scenario_retry_warnings(*, run: ScenarioRunSummary, seen_attack_ids: set[str]) -> None:
+    """
+    Print retry warnings for attack results seen for the first time.
+
+    Called during polling so retries stream to the console as each attack result
+    lands. ``seen_attack_ids`` is mutated to de-duplicate across polls.
+
+    Args:
+        run: ``ScenarioRunSummary`` from ``GET /api/scenarios/runs/{id}``.
+        seen_attack_ids: Attack-result IDs already printed; updated in place.
+    """
+    new_attacks = [a for a in run.attack_retries if a.attack_result_id not in seen_attack_ids]
+    if not new_attacks:
+        return
+
+    # Finalize the in-place progress line (written with `\r`, no newline) so these
+    # warnings become persistent scrollback above the next progress redraw.
+    print()
+    for attack in new_attacks:
+        seen_attack_ids.add(attack.attack_result_id)
+        for retry in attack.retries:
+            exc = retry.exception_type or "error"
+            message = (retry.exception_message or "").strip().splitlines()
+            if message:
+                exc = f"{exc}: {_truncate_text(message[0], 160)}"
+            location = _format_retry_location(retry)
+            _cprint(
+                f"  ! retry #{retry.attempt_number} [{attack.atomic_attack_name}]{location}: {exc}",
+                color="yellow",
+            )
+
+
 def print_scenario_run_progress(*, run: ScenarioRunSummary, total_techniques: int = 0) -> None:
     """
     Print a single-line progress update (overwrites the current line).
@@ -277,19 +337,17 @@ def print_scenario_run_progress(*, run: ScenarioRunSummary, total_techniques: in
 
     parts: list[str] = []
 
+    # The bar tracks techniques completed / total, which is the only ratio we can
+    # honestly compute mid-run: the server only knows about attacks already persisted,
+    # so an attacks-based bar would always read 100%.
     if effective_total > 0:
-        parts.append(f"techniques: {techniques_done}/{effective_total}")
-    elif techniques_done > 0:
-        parts.append(f"techniques: {techniques_done}")
-
-    if run.total_attacks > 0:
-        pct = int((run.completed_attacks / run.total_attacks) * 100)
+        pct = int((techniques_done / effective_total) * 100)
         bar_width = 30
-        filled = int(bar_width * run.completed_attacks / run.total_attacks)
+        filled = int(bar_width * techniques_done / effective_total)
         bar = "█" * filled + "░" * (bar_width - filled)
-        parts.append(f"[{bar}] {run.completed_attacks}/{run.total_attacks} attacks ({pct}%)")
+        parts.append(f"[{bar}] techniques: {techniques_done}/{effective_total} ({pct}%)")
     else:
-        parts.append(f"attacks: {run.completed_attacks}")
+        parts.append(f"techniques: {techniques_done}")
 
     parts.append(f"success rate: {run.objective_achieved_rate}%")
     parts.append(run.status.value)
@@ -310,16 +368,29 @@ def print_scenario_run_summary(*, run: ScenarioRunSummary) -> None:
     print(f"\nScenario: {run.scenario_name}")
     print(f"  Result ID:      {run.scenario_result_id}")
     print(f"  Status:         {run.status.value}")
-    print(f"  Total Attacks:  {run.total_attacks}")
-    print(f"  Completed:      {run.completed_attacks}")
+    # Count of individual attack-result records persisted (one per technique x objective
+    # that ran), not a planned total. It stops growing wherever a failed run halted.
+    print(f"  Attack Results: {run.total_attacks}")
     print(f"  Success Rate:   {run.objective_achieved_rate}%")
 
+    if run.total_retries:
+        print(f"  Retries:        {run.total_retries} (endpoint-stress signal)")
+
     if run.error:
         print(f"  Error:          {run.error}")
 
     if run.techniques_used:
         print(f"  Techniques:     {', '.join(run.techniques_used)}")
 
+    if run.failed_attacks:
+        print(f"\n  Failed Attacks ({len(run.failed_attacks)}):")
+        for failed in run.failed_attacks:
+            error_type = failed.error_type or "Error"
+            message = (failed.error_message or "").strip().splitlines()
+            detail = _truncate_text(message[0], 200) if message else "no detail"
+            retry_note = f" [{failed.total_retries} retries]" if failed.total_retries else ""
+            print(f"    - {failed.atomic_attack_name}{retry_note}: {error_type}: {detail}")
+
 
 # ---------------------------------------------------------------------------
 # Scenario run detail (full results via output module)

pyrit/cli/_server_launcher.py

@@ -237,19 +237,46 @@ async def start_async(
 
             exit_code = self._process.poll()
             if exit_code is not None:
+                self._print_log_tail()
                 raise RuntimeError(
                     f"Server process exited with code {exit_code} during startup. See logs: {self._log_path}"
                 )
 
             if await self.probe_health_async(base_url=base_url):
-                print(f"Server ready (PID {self._pid})")
+                print(f"Server ready (PID {self._pid}). Logs: {self._log_path}")
                 return base_url
 
+        self._print_log_tail()
         raise RuntimeError(
             f"pyrit_backend did not become healthy within {startup_timeout}s. "
             f"Check the server logs ({self._log_path}) or start it manually with: pyrit_backend"
         )
 
+    def _read_log_tail(self, *, max_lines: int = 20) -> str:
+        """
+        Read the last ``max_lines`` lines of the backend log file.
+
+        Returns:
+            str: The tail of the log, or an empty string when the log is
+            unavailable or empty.
+        """
+        if not self._log_path:
+            return ""
+        try:
+            with open(self._log_path, encoding="utf-8", errors="replace") as handle:
+                lines = handle.readlines()
+        except OSError:
+            return ""
+        return "".join(lines[-max_lines:]).rstrip()
+
+    def _print_log_tail(self) -> None:
+        """Echo the tail of the backend log to stderr, if any is available."""
+        tail = self._read_log_tail()
+        if tail:
+            print(f"\n--- pyrit_backend log ({self._log_path}) ---", file=sys.stderr)
+            print(tail, file=sys.stderr)
+            print("--- end of log ---", file=sys.stderr)
+
     # ------------------------------------------------------------------
     # Stop
     # ------------------------------------------------------------------

pyrit/cli/pyrit_scan.py

@@ -682,8 +682,10 @@ async def _poll_until_terminal_async(
 
     terminal_states = {ScenarioRunState.COMPLETED, ScenarioRunState.FAILED, ScenarioRunState.CANCELLED}
 
+    seen_retry_attack_ids: set[str] = set()
     while True:
         run = await client.get_scenario_run_async(scenario_result_id=scenario_result_id)
+        _output.print_scenario_retry_warnings(run=run, seen_attack_ids=seen_retry_attack_ids)
         _output.print_scenario_run_progress(run=run, total_techniques=total_techniques)
         if run.status in terminal_states:
             return run

pyrit/models/catalog/__init__.py

@@ -17,6 +17,8 @@
     RegisteredInitializer,
 )
 from pyrit.models.catalog.scenario import (
+    AttackErrorSummary,
+    AttackRetrySummary,
     RegisteredScenario,
     RunScenarioRequest,
     ScenarioRunSummary,
@@ -26,6 +28,8 @@
 )
 
 __all__ = [
+    "AttackErrorSummary",
+    "AttackRetrySummary",
     "RegisteredInitializer",
     "RegisteredScenario",
     "RunScenarioRequest",

pyrit/models/catalog/scenario.py

@@ -20,6 +20,7 @@
 
 from pyrit.models.parameter import Parameter
 from pyrit.models.results.scenario_result import ScenarioRunState
+from pyrit.models.retry_event import RetryEvent
 
 # Authoritative set of dataset seed filters exposed over the run request surface. Each entry
 # is used verbatim as a ``MemoryInterface.get_seeds`` keyword argument, so a filter key IS the
@@ -115,6 +116,26 @@ def _validate_dataset_filters(cls, value: dict[str, list[str]] | None) -> dict[s
         return value
 
 
+class AttackErrorSummary(BaseModel):
+    """A single errored attack result surfaced in a run summary."""
+
+    atomic_attack_name: str = Field(..., description="Atomic-attack cell that errored")
+    objective: str = Field("", description="Objective that was being attempted")
+    error_type: str | None = Field(None, description="Exception class name")
+    error_message: str | None = Field(None, description="Exception message")
+    total_retries: int = Field(0, ge=0, description="Retry attempts recorded for this attack")
+
+
+class AttackRetrySummary(BaseModel):
+    """Retry events recorded for one attack result, for near-real-time CLI display."""
+
+    attack_result_id: str = Field(..., description="Stable ID of the attack result (used to de-duplicate)")
+    atomic_attack_name: str = Field(..., description="Atomic-attack cell that retried")
+    retries: list[RetryEvent] = Field(
+        default_factory=list, description="Retry attempts, each with component role/name, endpoint, and exception"
+    )
+
+
 class ScenarioRunSummary(BaseModel):
     """Response for a scenario run (status + result details)."""
 
@@ -130,5 +151,16 @@ class ScenarioRunSummary(BaseModel):
     total_attacks: int = Field(0, ge=0, description="Total number of attack results persisted for this run")
     completed_attacks: int = Field(0, ge=0, description="Number of attacks that reached a terminal outcome")
     objective_achieved_rate: int = Field(0, ge=0, le=100, description="Success rate as percentage (0-100)")
+    failed_attacks: list[AttackErrorSummary] = Field(
+        default_factory=list,
+        description="Individual attack results that errored, surfaced regardless of overall run status",
+    )
+    attack_retries: list[AttackRetrySummary] = Field(
+        default_factory=list,
+        description="Per-attack retry events, surfaced as each attack result lands so the CLI can stream warnings",
+    )
+    total_retries: int = Field(
+        0, ge=0, description="Total retry attempts recorded across all attack results (endpoint-stress signal)"
+    )
     labels: dict[str, str] = Field(default_factory=dict, description="Labels attached to this run")
     completed_at: datetime | None = Field(None, description="When the scenario finished")

pyrit/models/results/scenario_result.py

@@ -155,13 +155,18 @@ def objective_scorer_identifier(self) -> ScorerIdentifier | None:
 
     def get_techniques_used(self) -> list[str]:
         """
-        Get the list of techniques used in this scenario.
+        Get the list of techniques present in the results.
+
+        Results are aggregated by display group so each technique is counted once,
+        even when it fans out into multiple atomic-attack cells (e.g. across
+        datasets or targets in a matrix scenario). When no ``display_group_map`` is
+        set, atomic-attack names are returned unchanged.
 
         Returns:
-            list[str]: Atomic attack technique names present in the results.
+            list[str]: Technique (display-group) names that produced results.
 
         """
-        return list(self.attack_results.keys())
+        return list(self.get_display_groups().keys())
 
     def get_display_groups(self) -> dict[str, list[AttackResult]]:
         """

pyrit/scenario/core/atomic_attack.py

@@ -361,8 +361,8 @@ async def run_async(
             return results
 
         except Exception as e:
-            logger.error(f"Atomic attack execution failed: {str(e)}")
-            raise ValueError(f"Failed to execute atomic attack: {str(e)}") from e
+            logger.error(f"Atomic attack '{self.atomic_attack_name}' execution failed: {str(e)}")
+            raise ValueError(f"Failed to execute atomic attack '{self.atomic_attack_name}': {str(e)}") from e
 
     def _enrich_atomic_attack_identifiers(self, *, results: AttackExecutorResult[AttackResult]) -> None:
         """

pyrit/scenario/core/attack_technique_factory.py

@@ -395,6 +395,26 @@ def adversarial_chat(self) -> PromptTarget | None:
         """The adversarial chat target baked into this factory, or None."""
         return self._adversarial_chat
 
+    def resolve_adversarial_chat(self) -> PromptTarget | None:
+        """
+        Resolve the adversarial chat target an ``AtomicAttack`` needs to expand this technique.
+
+        A baked ``adversarial_chat`` always wins. Otherwise, when the technique's seed group
+        carries a simulated conversation (built via ``with_simulated_conversation``), the default
+        adversarial target is resolved lazily here — mirroring how ``create()`` resolves the target
+        for the attack's own ``attack_adversarial_config``. Techniques without a simulated
+        conversation seed do not need one and return ``None``.
+
+        Returns:
+            PromptTarget | None: The adversarial chat target for the ``AtomicAttack``, or ``None``
+            when the technique does not drive a simulated conversation.
+        """
+        if self._adversarial_chat is not None:
+            return self._adversarial_chat
+        if self._seed_technique is not None and self._seed_technique.has_simulated_conversation:
+            return get_default_adversarial_target()
+        return None
+
     @property
     def uses_adversarial(self) -> bool:
         """Whether this technique drives an adversarial chat during execution."""

pyrit/scenario/core/matrix_atomic_attack_builder.py

@@ -346,7 +346,7 @@ def build(
                             attack_technique=attack_technique,
                             seed_groups=compatible_groups,
                             adversarial_chat=(
-                                target_instance if target_instance is not None else factory.adversarial_chat
+                                target_instance if target_instance is not None else factory.resolve_adversarial_chat()
                             ),
                             objective_scorer=cast("TrueFalseScorer", self._objective_scorer),
                             memory_labels=self._memory_labels,