Repository navigation
BUG: Canonical scenario group counts are overwritten by unmatched attempts #3699
Description
Describe the bug
Scenario history's per-group counts lose a planned success when an unplanned attempt shares the same attack name and technique configuration. Overall counts remain correct, but the canonical group counts are overwritten.
At pyrit/memory/memory_interface.py:7334-7338, an unmatched group's raw key is converted into the canonical configuration hash. At lines 7376-7379, run_groups[canonical_gid] = counts replaces an existing group's counts instead of combining the disjoint attempt populations. The hierarchy read model consumes these counts at pyrit/backend/services/scenario_progress_read_model.py:1148-1160.
Reproduced on freshly fetched main at d3446e045960b5ebee4641aa70fdeb03af186587 on October 31, 2026. Introduced in #3167. Focused searches found no dedicated correction. #3611 concerns related hierarchy presentation, not this memory aggregation.
Steps/Code to Reproduce
Run in the project's Python environment. This uses synthetic records and a disposable SQLite database. The target is an identity record, not an instantiated target or network endpoint.
import asyncio
import tempfile
import uuid
from pathlib import Path
from unittest.mock import patch
from pyrit.common import path as paths
from pyrit.common.utils import to_sha256
from pyrit.models import (
SCENARIO_RUN_PLAN_METADATA_KEY, AttackOutcome, AttackResult,
ScenarioIdentifier, ScenarioResult, ScenarioRunPlan,
ScenarioRunPlanAtomicGroup, ScenarioRunPlanSeedGroup, ScenarioRunState,
TargetIdentifier,
)
from pyrit.models.identifiers.component_identifier import config_hash
async def main_async():
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
with patch.object(paths, "DB_DATA_PATH", root):
from pyrit.memory import SQLiteMemory
memory = SQLiteMemory(db_path=root / "synthetic.db")
try:
group_id = config_hash({"atomic_attack_name": "attack", "technique_eval_hash": "eval"})
plan = ScenarioRunPlan(
scenario_registry_name="synthetic",
atomic_groups=[ScenarioRunPlanAtomicGroup(
id=group_id, atomic_attack_name="attack", display_group="attack",
technique_eval_hash="eval", seed_group_ids=["planned"],
)],
seed_groups=[ScenarioRunPlanSeedGroup(
id="planned", objective="Objective", objective_sha256=to_sha256("Objective"),
)],
)
run = ScenarioResult(
scenario_identifier=ScenarioIdentifier(
class_name="SyntheticScenario", class_module="synthetic", version=1,
objective_target=TargetIdentifier(
class_name="SyntheticTarget", class_module="synthetic",
),
),
attack_results={}, scenario_run_state=ScenarioRunState.COMPLETED,
metadata={SCENARIO_RUN_PLAN_METADATA_KEY: plan.model_dump(mode="json")},
)
await memory.add_scenario_results_to_memory_async(scenario_results=[run])
await memory.add_attack_results_to_memory_async(attack_results=[
AttackResult(
conversation_id=str(uuid.uuid4()), objective=objective, outcome=outcome,
attribution_parent_id=str(run.id),
attribution_data={
"parent_collection": "attack", "parent_eval_hash": "eval",
"seed_group_id": seed, "result_role": "target_facing",
},
)
for seed, objective, outcome in [
("planned", "Objective", AttackOutcome.SUCCESS),
("outside", "Outside plan", AttackOutcome.ERROR),
]
])
result = (await memory.get_scenario_history_aggregates_async(
scenario_result_ids=[str(run.id)], plan_scenario_ids=[str(run.id)],
))[str(run.id)]
group = result.group_counts[group_id]
print("overall:", result.completed_units, result.successful_units,
result.target_facing_attempts)
print("group:", group.completed, group.succeeded,
group.producer_counts.target_facing.attempts)
finally:
await memory.dispose_engine_async()
asyncio.run(main_async())Expected Results
The group retains its one completed success and includes both producer attempts:
overall: 1 1 2
group: 1 1 2
The unplanned error should not become an additional planned unit.
Actual Results
overall: 1 1 2
group: 0 0 1
Suggested fix: combine counts for raw groups that resolve to the same canonical identity without changing planned-unit filtering or double-counting aliases. Add a regression using the canonical configuration hash as the persisted plan's group ID. Existing parity tests pass, but the similar two-attempt test uses a different plan group ID.
Screenshots
N/A.
Versions
Windows 11; Python 3.14.4; PyRIT 1.2.0.dev0, editable main at the SHA above. SQLAlchemy 2.1.1; Pydantic 2.14.1. Browser: N/A. SQLite reproduction confirmed; SQL Server was not tested.
pyrit.show_versions() dependency output (local executable path omitted):
pyrit: 1.2.0.dev0
Cython: None
numpy: 2.4.6
openai: 2.54.0
packaging: 26.3
pip: None
scipy: 1.17.1
setuptools: 84.2.1
sqlite3: None
torch: None
transformers: 5.20.2
Activity