Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 17 additions & 3 deletions pyrit/memory/memory_models.py
Original file line number Diff line number Diff line change
Expand Up @@ -1816,6 +1816,8 @@ class ScenarioResultEntry(Base):
scenario_identifier (dict): Canonical scenario identity (class name, version,
techniques, datasets, resolved params, objective target / scorer children).
objective_target_identifier (dict): Identifier for the target being evaluated in the scenario.
Required: this is the denormalized filter key that target-based queries match on, so a
scenario result without one could never be retrieved by target.
objective_scorer_identifier (dict): Optional identifier for the scorer used to evaluate results.
scenario_run_state (str): Current execution state of the scenario
(one of CREATED, IN_PROGRESS, COMPLETED, FAILED, CANCELLED).
Expand Down Expand Up @@ -1875,6 +1877,11 @@ def __init__(self, *, entry: ScenarioResult) -> None:

Args:
entry (ScenarioResult): The scenario result object to convert into a database entry.

Raises:
ValueError: If ``entry`` has no ``objective_target_identifier``. The denormalized target
column is the key that target-based queries filter on, so a result without one would
be persisted as a row those queries could never return.
"""
self.id = entry.id
self.scenario_name = entry.scenario_name
Expand All @@ -1891,10 +1898,17 @@ def __init__(self, *, entry: ScenarioResult) -> None:
self.scenario_identifier = scenario_identifier.model_dump()
self.scenario_identifier_hash = scenario_identifier.hash

# Convert ComponentIdentifier to dict for JSON storage
# Convert ComponentIdentifier to dict for JSON storage. The target is required: it is the
# denormalized key that target-based queries filter on, so persisting a result without one
# would write a row that those queries can never return.
target_identifier = entry.objective_target_identifier
target_identifier_dict = target_identifier.model_dump() if target_identifier else None
self.objective_target_identifier = target_identifier_dict # type: ignore[ty:invalid-assignment]
if target_identifier is None:
raise ValueError(
"objective_target_identifier is required to persist a ScenarioResult. "
f"Scenario '{entry.scenario_name}' produced a result with no objective target; "
"a scenario must declare and resolve objective_target before its result is stored."
)
self.objective_target_identifier = target_identifier.model_dump()
# Always recompute eval_hash before dumping so the stored JSON carries the
# freshly computed value for DB-level filtering (never a value from storage).
scorer_identifier = entry.objective_scorer_identifier
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -1011,3 +1011,22 @@ def test_get_scenario_results_by_target_identifier_filter_no_match(
],
)
assert len(results) == 0


def test_add_scenario_result_without_objective_target_raises(sqlite_instance: MemoryInterface):
"""Persisting a targetless scenario result fails loudly instead of writing an unqueryable row."""
attack_result = create_attack_result("conv_1", "Objective 1")
sqlite_instance.add_attack_results_to_memory(attack_results=[attack_result])

scenario = make_scenario_result(
scenario_name="Targetless Scenario",
scenario_version=1,
objective_target_identifier=None,
attack_results={"Attack1": [attack_result]},
objective_scorer_identifier=get_mock_scorer_identifier(),
)

with pytest.raises(ValueError, match="objective_target_identifier is required"):
sqlite_instance.add_scenario_results_to_memory(scenario_results=[scenario])

assert sqlite_instance.get_scenario_results(scenario_name="Targetless Scenario") == []
6 changes: 6 additions & 0 deletions tests/unit/memory/test_memory_models.py
Original file line number Diff line number Diff line change
Expand Up @@ -797,3 +797,9 @@ def test_str(self):
entry = ScenarioResultEntry(entry=sr)
s = str(entry)
assert "test_scenario" in s

def test_init_without_objective_target_raises(self):
"""The denormalized target column is the filter key, so a targetless result is rejected."""
sr = self._make_scenario_result(objective_target_identifier=None)
with pytest.raises(ValueError, match="objective_target_identifier is required"):
ScenarioResultEntry(entry=sr)
Loading