FIX: preserve child-result links when execution fails in SequentialAttack

doc/code/executor/0_executor.md

@@ -76,6 +76,25 @@ configs (prepended conversations, multimodal seeds, next-turn messages, memory l
 
 The category pages above each walk through their executors with short runnable examples.
 
+## Partial execution and stored results
+
+`AttackExecutor` raises execution failures by default. With `return_partial_on_failure=True`,
+it returns an `AttackExecutorResult` containing completed results and `incomplete_objectives`.
+Each incomplete entry retains the original exception; `raise_if_incomplete()` re-raises it.
+Cancellation still propagates rather than becoming an incomplete objective.
+
+An incomplete execution may have saved an error result, or a completed result whose database
+commit acknowledgement failed. `incomplete_result_ids` aligns with `incomplete_objectives` and
+contains each confirmed stored result ID, or `None` when no row was confirmed. After a failed
+write, PyRIT checks the allocated result ID without retrying the write or creating another row.
+If that check also fails, the failure is logged and the original execution/persistence error
+still propagates.
+
+`SequentialAttack` keeps its ordered child links in
+`metadata["child_attack_result_ids"]`, including when the parent fails. Nested sequences link
+their direct children, not their grandchildren. These links survive database round-trips;
+they include confirmed child rows only, never IDs for children that were not saved.
+
 ## When do you actually need a new executor class?
 
 Most of an executor's behavior comes from its *configuration and data*, not from new code. So before

pyrit/executor/attack/compound/sequential_attack.py

@@ -250,6 +250,13 @@ async def _teardown_async(self, *, context: AttackContext[AttackParameters]) ->
 
     async def _perform_async(self, *, context: AttackContext[AttackParameters]) -> SequentialAttackResult:
         results: list[AttackResult] = []
+        child_ids: list[str] = []
+        context._error_result_metadata.update(
+            {
+                self.CHILD_ATTACK_RESULT_IDS_KEY: child_ids,
+                self.COMPLETION_POLICY_KEY: self._completion_policy.value,
+            }
+        )
 
         for attempt_index, child_attack in enumerate(self._child_attacks, start=1):
             labels = {**context.memory_labels, **dict(child_attack.memory_labels)}
@@ -260,8 +267,10 @@ async def _perform_async(self, *, context: AttackContext[AttackParameters]) -> S
                 memory_labels=labels,
                 attribution=attribution,
                 expectation=context.params.expectation,
+                child_result_ids=child_ids,
             )
             results.append(result)
+            child_ids.append(result.attack_result_id)
             if self._should_stop_after(result=result):
                 break
 
@@ -294,6 +303,7 @@ async def _run_child_attack_async(
         *,
         child_attack: SequentialChildAttack,
         memory_labels: dict[str, str],
+        child_result_ids: list[str],
         attribution: AttackResultAttribution | None = None,
         expectation: ScoringExpectation | None = None,
     ) -> AttackResult:
@@ -316,6 +326,8 @@ async def _run_child_attack_async(
                 parent linkage.
             expectation (ScoringExpectation | None): Explicit scoring input forwarded unchanged.
                 Omission leaves the child's seed preparation and objective fallback in control.
+            child_result_ids (list[str]): This execution's ordered child links. A confirmed
+                persisted result from a failed dispatch is appended before its exception is re-raised.
 
         Returns:
             AttackResult: The ``AttackResult`` produced by the inner
@@ -335,12 +347,15 @@ async def _run_child_attack_async(
             adversarial_chat=child_attack.adversarial_chat,
             objective_scorer=child_attack.objective_scorer,
             memory_labels=memory_labels,
+            return_partial_on_failure=True,
             attribution=attribution,
             **expectation_override,
         )
         if executor_result.completed_results:
             return executor_result.completed_results[0]
         if executor_result.incomplete_objectives:
+            if executor_result.incomplete_result_ids and executor_result.incomplete_result_ids[0] is not None:
+                child_result_ids.append(executor_result.incomplete_result_ids[0])
             raise executor_result.incomplete_objectives[0][1]
         raise RuntimeError(  # pragma: no cover - defensive
             "AttackExecutor returned neither completed nor incomplete results."

pyrit/executor/attack/core/attack_executor.py

@@ -56,6 +56,23 @@ class AttackExecutorResult(Generic[AttackResultT]):
     list that produced ``completed_results[i]``.  When some inputs fail, this lets
     callers correlate results back to the specific input that produced them.
     """
+    incomplete_result_ids: list[str | None] = field(default_factory=list)
+    """Confirmed persisted result ID for each ``incomplete_objectives`` entry.
+
+    ``None`` means no stored row was confirmed, including parameter-build failures.
+    Executor-produced containers align this list with ``incomplete_objectives``;
+    manually constructed containers may omit it for backward compatibility.
+    """
+
+    def __post_init__(self) -> None:
+        """
+        Validate the alignment of supplied failure result IDs.
+
+        Raises:
+            ValueError: If supplied IDs do not align with incomplete objectives.
+        """
+        if self.incomplete_result_ids and len(self.incomplete_result_ids) != len(self.incomplete_objectives):
+            raise ValueError("incomplete_result_ids length must match incomplete_objectives length")
 
     def __iter__(self) -> Iterator[AttackResultT]:
         """
@@ -400,14 +417,18 @@ async def _execute_with_params_list_async(
             raise ValueError(
                 f"attributions length ({len(attributions)}) must match params_list length ({len(params_list)})"
             )
+        persisted_result_ids: list[str | None] = [None] * len(params_list)
 
         async def run_one_async(index: int, params: AttackParameters) -> AttackStrategyResultT:
             async with semaphore:
                 context = attack._context_type(params=params)
                 task_attribution = attributions[index] if attributions is not None else None
                 if task_attribution is not None:
                     context._attribution = task_attribution
-                return await attack.execute_with_context_async(context=context)
+                try:
+                    return await attack.execute_with_context_async(context=context)
+                finally:
+                    persisted_result_ids[index] = context._persisted_attack_result_id
 
         tasks = [run_one_async(i, p) for i, p in enumerate(params_list)]
         results_or_exceptions = await asyncio.gather(*tasks, return_exceptions=True)
@@ -417,6 +438,7 @@ async def run_one_async(index: int, params: AttackParameters) -> AttackStrategyR
             results_or_exceptions=list(results_or_exceptions),
             return_partial_on_failure=return_partial_on_failure,
             input_indices=input_indices,
+            persisted_result_ids=persisted_result_ids,
         )
 
     def _process_execution_results(
@@ -426,6 +448,7 @@ def _process_execution_results(
         results_or_exceptions: list[Any],
         return_partial_on_failure: bool,
         input_indices: Sequence[int] | None = None,
+        persisted_result_ids: Sequence[str | None] | None = None,
     ) -> AttackExecutorResult[AttackStrategyResultT]:
         """
         Process results from parallel execution into an AttackExecutorResult.
@@ -435,26 +458,32 @@ def _process_execution_results(
             results_or_exceptions: Results or exceptions from asyncio.gather.
             return_partial_on_failure: Whether to return partial results on failure.
             input_indices: Original input positions corresponding to ``objectives``.
+            persisted_result_ids: Confirmed stored row IDs corresponding to ``objectives``.
 
         Returns:
             AttackExecutorResult with completed and incomplete results.
 
         Raises:
             BaseException: If return_partial_on_failure=False and any failed.
-            ValueError: If input_indices length doesn't match objectives length.
+            ValueError: If input-index or result-ID lengths don't match objectives.
         """
         completed: list[AttackStrategyResultT] = []
         incomplete: list[tuple[str, BaseException]] = []
         completed_indices: list[int] = []
+        incomplete_result_ids: list[str | None] = []
         source_indices = list(input_indices) if input_indices is not None else list(range(len(objectives)))
         if len(source_indices) != len(objectives):
             raise ValueError("input_indices length must match objectives length")
+        result_ids = list(persisted_result_ids) if persisted_result_ids is not None else [None] * len(objectives)
+        if len(result_ids) != len(objectives):
+            raise ValueError("persisted_result_ids length must match objectives length")
 
         self._raise_first_fatal_exception(results_or_exceptions=results_or_exceptions)
 
         for i, (objective, result) in enumerate(zip(objectives, results_or_exceptions, strict=True)):
             if isinstance(result, BaseException):
                 incomplete.append((objective, result))
+                incomplete_result_ids.append(result_ids[i])
             else:
                 completed.append(result)
                 completed_indices.append(source_indices[i])
@@ -463,6 +492,7 @@ def _process_execution_results(
             completed_results=completed,
             incomplete_objectives=incomplete,
             input_indices=completed_indices,
+            incomplete_result_ids=incomplete_result_ids,
         )
 
         if not return_partial_on_failure:
@@ -493,17 +523,21 @@ def _merge_parameter_build_failures(
         if not build_failures:
             return execution_result
 
-        incomplete_by_index: dict[int, tuple[str, BaseException]] = {
-            index: (objective, error) for index, objective, error in build_failures
+        incomplete_by_index: dict[int, tuple[str, BaseException, str | None]] = {
+            index: (objective, error, None) for index, objective, error in build_failures
         }
         completed_indices = set(execution_result.input_indices)
-        execution_failures = iter(execution_result.incomplete_objectives)
+        result_ids = execution_result.incomplete_result_ids or [None] * len(execution_result.incomplete_objectives)
+        execution_failures = iter(zip(execution_result.incomplete_objectives, result_ids, strict=True))
         for input_index in successful_input_indices:
             if input_index not in completed_indices:
-                incomplete_by_index[input_index] = next(execution_failures)
+                (objective, error), result_id = next(execution_failures)
+                incomplete_by_index[input_index] = (objective, error, result_id)
 
+        ordered_failures = [incomplete_by_index[index] for index in sorted(incomplete_by_index)]
         return AttackExecutorResult(
             completed_results=execution_result.completed_results,
-            incomplete_objectives=[incomplete_by_index[index] for index in sorted(incomplete_by_index)],
+            incomplete_objectives=[(objective, error) for objective, error, _ in ordered_failures],
             input_indices=execution_result.input_indices,
+            incomplete_result_ids=[result_id for _, _, result_id in ordered_failures],
         )

pyrit/executor/attack/core/attack_strategy.py

@@ -35,6 +35,7 @@
     AttackResult,
     AttackResultMetadata,
     AttackResultRole,
+    AttackResultSelection,
     ComponentIdentifier,
     ConversationReference,
     ConverterIdentifier,
@@ -183,6 +184,8 @@ class AttackContext(StrategyContext, ABC, Generic[AttackParamsT]):
     _prepended_conversation_override: list[Message] | None = None
     _memory_labels_override: dict[str, str] | None = None
     _error_result_persistence_error: Exception | None = field(default=None, init=False, repr=False)
+    _error_result_metadata: dict[str, Any] = field(default_factory=dict, init=False, repr=False, compare=False)
+    _persisted_attack_result_id: str | None = field(default=None, init=False, repr=False, compare=False)
     _persist_attack_result: bool = field(default=True, init=False, repr=False, compare=False)
     _objective_target_conversation_lifecycle: _ObjectiveTargetConversationLifecycle | None = field(
         default=None,
@@ -437,6 +440,23 @@ async def _persist_result_async(self, *, result: AttackStrategyResultT) -> None:
         """
         (await self._memory.add_attack_results_to_memory_async(attack_results=[result]))
 
+    async def _confirm_result_persistence_async(
+        self, *, context: AttackStrategyContextT, attack_result_id: str
+    ) -> None:
+        """Confirm an uncertain write by its allocated ID without retrying persistence."""
+        try:
+            stored = await self._memory.get_attack_results_async(
+                attack_result_ids=[attack_result_id],
+                result_selection=AttackResultSelection.ALL_RESULTS,
+            )
+        except Exception:
+            self._logger.exception(
+                "Unable to confirm persisted attack result %s after a failed write", attack_result_id
+            )
+        else:
+            if any(result.attack_result_id == attack_result_id for result in stored):
+                context._persisted_attack_result_id = attack_result_id
+
     @staticmethod
     def _apply_attribution(
         *,
@@ -564,6 +584,7 @@ async def _on_error_async(
             error_traceback="".join(traceback.format_exception(type(error), error, error.__traceback__)),
             retry_events=retry_events,
             total_retries=len(retry_events),
+            metadata=dict(context._error_result_metadata),
         )
 
         end_time = time.perf_counter()
@@ -579,6 +600,11 @@ async def _on_error_async(
             (await self._memory.add_attack_results_to_memory_async(attack_results=[error_result]))
         except Exception as persistence_error:
             context._error_result_persistence_error = persistence_error
+            await self._confirm_result_persistence_async(
+                context=context, attack_result_id=error_result.attack_result_id
+            )
+        else:
+            context._persisted_attack_result_id = error_result.attack_result_id
 
         self._logger.error(f"Attack failed with {type(error).__name__}: {error}")
 
@@ -880,8 +906,10 @@ async def execute_with_context_async(self, *, context: AttackStrategyContextT) -
         Raises:
             ExceptionGroup: If attack execution and recording its error result both fail.
         """
-        self._validate_scoring_expectation(context=context)
         context._error_result_persistence_error = None
+        context._error_result_metadata.clear()
+        context._persisted_attack_result_id = None
+        self._validate_scoring_expectation(context=context)
         context._attack_result_id = str(uuid.uuid4())
         context._result_role = self.RESULT_ROLE
         lifecycle = _ObjectiveTargetConversationLifecycle(
@@ -906,7 +934,14 @@ async def execute_with_context_async(self, *, context: AttackStrategyContextT) -
             context._objective_target_conversation_lifecycle = None
 
         if context._persist_attack_result:
-            (await self._default_event_handler._persist_result_async(result=result))
+            try:
+                await self._default_event_handler._persist_result_async(result=result)
+            except Exception:
+                await self._default_event_handler._confirm_result_persistence_async(
+                    context=context, attack_result_id=result.attack_result_id
+                )
+                raise
+            context._persisted_attack_result_id = result.attack_result_id
         return result
 
     def _validate_scoring_expectation(self, *, context: AttackStrategyContextT) -> None:

pyrit/scenario/core/atomic_attack.py

@@ -408,6 +408,7 @@ async def run_async(
                 completed_results=completed_results,
                 incomplete_objectives=untyped_results.incomplete_objectives,
                 input_indices=untyped_results.input_indices,
+                incomplete_result_ids=untyped_results.incomplete_result_ids,
             )
 
             # Enrich atomic_attack_identifier with seed identifiers