pyrit.memory

Provide functionality for storing and retrieving conversation history and embeddings.

This package defines the core MemoryInterface and concrete implementations for different storage backends.

Functions¶

data_serializer_factory¶

data_serializer_factory(data_type: PromptDataType, value: str | None = None, extension: str | None = None, category: AllowedCategories) → DataTypeSerializer

Create a DataTypeSerializer instance.

Returns:

• DataTypeSerializer — An instance of the appropriate serializer.

Raises:

• ValueError — If the category is not provided or invalid.

get_lazy_dir¶

get_lazy_dir(module_globals: dict[str, Any], exports: Mapping[str, LazyExport]) → list[str]

Return package attributes and unresolved public exports.

Returns:

• list[str] — list[str]: Sorted package attribute names.

resolve_lazy_export¶

resolve_lazy_export(name: str, module_name: str, module_globals: dict[str, Any], exports: Mapping[str, LazyExport]) → Any

Resolve and cache one package export.

Returns:

• Any — The resolved public value.

Raises:

• AttributeError — If name is not a declared lazy export.

set_message_piece_sha256_async¶

set_message_piece_sha256_async(message_piece: MessagePiece) → None

Compute and assign SHA256 hash values for a message piece’s original and converted payloads.

Async because blob payloads may need to be fetched. Must be called explicitly after the message piece is constructed and its values are finalized.

set_seed_sha256_async¶

set_seed_sha256_async(seed: Seed[Any]) → None

Compute and assign the SHA256 hash value for a seed’s value.

Should be called after the seed value is serialized to text, as file paths used in the value may have changed from local to memory storage paths. Async due to blob retrieval.

AttackResultEntry¶

Bases: Base

Represents the attack result data in the database.

Constructor Parameters:

Methods:

filter_json_serializable_metadata¶

filter_json_serializable_metadata(metadata: dict[str, Any]) → dict[str, Any]

Filter a dictionary to only include JSON-serializable values.

This function iterates through the metadata dictionary and keeps only values that can be serialized to JSON, discarding any non-serializable objects.

Returns:

• dict[str, Any] — Dictionary with only JSON-serializable values

get_attack_result¶

get_attack_result() → AttackResult

Convert this database entry back into an AttackResult object.

Returns:

• AttackResult — The reconstructed attack result including related conversations and scores.

AttackResultKeysetCursor¶

Bases: NamedTuple

Keyset (seek) anchor identifying the last attack result on a page.

Captures the recency ordering tuple — the row timestamp (the attack’s last-updated time; see AttackResultEntry.timestamp) and the unique attack_result_id — so the next page can seek to rows strictly after this one under the timestamp DESC, id DESC ordering. Unlike a numeric offset — which shifts every later row when a row is inserted or deleted between page loads — a keyset anchor stays pinned to its row, so inserts and deletions elsewhere no longer skip or duplicate results at the page boundary.

Methods:

from_attack_result¶

from_attack_result(result: AttackResult) → AttackResultKeysetCursor

Build the keyset anchor for result (typically the last row of a page).

Returns:

• AttackResultKeysetCursor — Anchor capturing the result’s recency sort key.

AudioPathDataTypeSerializer¶

Bases: DataTypeSerializer

Serializer for audio path values stored on disk.

Constructor Parameters:

Methods:

data_on_disk¶

data_on_disk() → bool

Indicate whether this serializer persists data on disk.

Returns:

• bool — Always True for audio path serializers.

AzureBlobStorageIO¶

Bases: StorageIO

Implementation of StorageIO for Azure Blob Storage.

Constructor Parameters:

Methods:

create_directory_if_not_exists_async¶

create_directory_if_not_exists_async(directory_path: Path | str) → None

Log a no-op directory creation for Azure Blob Storage.

delete_file_async¶

delete_file_async(path: Path | str) → None

Delete one blob created by the caller.

is_file_async¶

is_file_async(path: Path | str) → bool

Check whether the path refers to a file (blob) in Azure Blob Storage.

Returns:

• bool — True when the blob exists and has non-zero content size.

parse_blob_url¶

parse_blob_url(file_path: str) → tuple[str, str]

Parse a blob URL to extract the container and blob name.

Returns:

• tuple[str, str] — tuple[str, str]: Container name and blob name.

Raises:

• ValueError — If file_path is not a valid blob URL.

path_exists_async¶

path_exists_async(path: Path | str) → bool

Check whether a given path exists in the Azure Blob Storage container.

Returns:

• bool — True when the path exists.

read_file_async¶

read_file_async(path: Path | str) → bytes

Asynchronously reads the content of a file (blob) from Azure Blob Storage.

If the provided path is a full URL (e.g., https://account.blob.core.windows.net/container/dir1/dir2/sample.png), it extracts the relative blob path (e.g., dir1/dir2/sample.png) to correctly access the blob. If a relative path is provided, it will use it as-is.

Returns:

• bytes — The content of the file (blob) as bytes.

write_file_async¶

write_file_async(path: Path | str, data: bytes) → None

Write data to Azure Blob Storage at the specified path.

If the provided path is a full URL, the blob name is extracted from it. If a relative path is provided, it is used as the blob name directly.

AzureSQLMemory¶

Bases: MemoryInterface

A class to manage conversation memory using Azure SQL Server as the backend database. It leverages SQLAlchemy Base models for creating tables and provides CRUD operations to interact with the tables.

This class encapsulates the setup of the database connection, table creation based on SQLAlchemy models, and session management to perform database operations.

Constructor Parameters:

Methods:

dispose_loop_resources_async¶

dispose_loop_resources_async() → None

Close the current loop’s engine and its owned Azure credential.

BinaryPathDataTypeSerializer¶

Bases: DataTypeSerializer

Serializer for generic binary path values stored on disk.

Constructor Parameters:

Methods:

data_on_disk¶

data_on_disk() → bool

Indicate whether this serializer persists data on disk.

Returns:

• bool — Always True for binary path serializers.

CentralMemory¶

Provide a centralized memory instance across the framework. The provided memory instance will be reused for future calls.

Examples:

• Attack Configuration

• Azure SQL Memory

• Benchmarks

• Combining & Stacking Scorers

• File-write scoring

• Float-Scale Scorers

• Generating Datasets

• Loading Built-in Datasets

• Memory

• Memory Labels and Advanced Memory Queries

• Output Module

• Scoring

• Seed Database Management

• Trace-backed tool-call scoring

• Workflows

Methods:

get_memory_instance¶

get_memory_instance() → MemoryInterface

Return a centralized memory instance.

Returns:

• MemoryInterface — The singleton memory instance.

Raises:

• ValueError — If the central memory instance has not been set.

set_memory_instance¶

set_memory_instance(passed_memory: MemoryInterface) → None

Set a provided memory instance as the central instance for subsequent calls.

DataTypeSerializer¶

Bases: abc.ABC

Abstract base class for data type normalizers.

Responsible for reading and saving multi-modal data types to local disk or Azure Storage Account.

Methods:

data_on_disk¶

data_on_disk() → bool

Indicate whether the data is stored on disk.

Returns:

• bool — True when data is persisted on disk.

get_data_filename_async¶

get_data_filename_async(file_name: str | None = None) → Path | str

Generate or retrieve a unique filename for the data file.

Returns:

• Path | str — Path | str: Full storage path for the generated data file.

Raises:

• TypeError — If the serializer is not configured for on-disk data.

• RuntimeError — If required data subdirectory information is missing.

get_extension¶

get_extension(file_path: str) → str | None

Get the file extension from the file path.

Returns:

• str | None — str | None: File extension (including dot) or None if unavailable.

get_mime_type¶

get_mime_type(file_path: str) → str | None

Get the MIME type of the file path.

Returns:

• str | None — str | None: MIME type if detectable; otherwise None.

get_sha256_async¶

get_sha256_async() → str

Compute SHA256 hash for this serializer’s current value.

Returns:

• str — Hex digest of the computed SHA256 hash.

Raises:

• FileNotFoundError — If on-disk data path does not exist.

• ValueError — If in-memory data cannot be converted to bytes.

read_data_async¶

read_data_async() → bytes

Read data from storage.

Returns:

• bytes — The data read from storage.

Raises:

• TypeError — If the serializer does not represent on-disk data.

• RuntimeError — If no value is set.

• FileNotFoundError — If the referenced file does not exist.

read_data_base64_async¶

read_data_base64_async() → str

Read data from storage and return it as a base64 string.

Returns:

• str — Base64-encoded data.

save_b64_image_async¶

save_b64_image_async(data: str | bytes, output_filename: str | None = None) → None

Save a base64-encoded image to storage.

Raises:

• RuntimeError — If storage IO is not initialized.

save_data_async¶

save_data_async(data: bytes, output_filename: str | None = None) → None

Save data to storage.

Raises:

• RuntimeError — If storage IO is not initialized.

save_formatted_audio_async¶

save_formatted_audio_async(data: bytes, num_channels: int = 1, sample_width: int = 2, sample_rate: int = 16000, output_filename: str | None = None) → None

Save PCM16 or similarly formatted audio data to storage.

Raises:

• RuntimeError — If storage IO is not initialized.

DiskStorageIO¶

Bases: StorageIO

Implementation of StorageIO for local disk storage.

Methods:

create_directory_if_not_exists_async¶

create_directory_if_not_exists_async(path: Path | str) → None

Asynchronously creates a directory if it doesn’t exist on the local disk.

delete_file_async¶

delete_file_async(path: Path | str) → None

Delete one file, not its containing directory.

is_file_async¶

is_file_async(path: Path | str) → bool

Check whether the given path is a file (not a directory).

Returns:

• bool — True if the path is a file, False otherwise.

path_exists_async¶

path_exists_async(path: Path | str) → bool

Check whether a path exists on the local disk.

Returns:

• bool — True if the path exists, False otherwise.

read_file_async¶

read_file_async(path: Path | str) → bytes

Asynchronously reads a file from the local disk.

Returns:

• bytes — The content of the file.

write_file_async¶

write_file_async(path: Path | str, data: bytes) → None

Asynchronously writes data to a file on the local disk.

EmbeddingDataEntry¶

Bases: Base

Represents the embedding data associated with conversation entries in the database. Each embedding is linked to a specific conversation entry via an id.

ErrorDataTypeSerializer¶

Bases: DataTypeSerializer

Serializer for error payloads stored as in-memory text.

Constructor Parameters:

Methods:

data_on_disk¶

data_on_disk() → bool

Indicate whether this serializer persists data on disk.

Returns:

• bool — Always False for error serializers.

ImagePathDataTypeSerializer¶

Bases: DataTypeSerializer

Serializer for image path values stored on disk.

Constructor Parameters:

Methods:

data_on_disk¶

data_on_disk() → bool

Indicate whether this serializer persists data on disk.

Returns:

• bool — Always True for image path serializers.

MemoryEmbedding¶

The MemoryEmbedding class is responsible for encoding the memory embeddings.

Constructor Parameters:

Methods:

generate_embedding_memory_data¶

generate_embedding_memory_data(message_piece: MessagePiece) → EmbeddingDataEntry

Generate metadata for a message piece.

Returns:

• EmbeddingDataEntry — The generated metadata.

Raises:

• ValueError — If the message piece is not of type text.

MemoryInterface¶

Bases: abc.ABC

Abstract interface for conversation memory storage systems.

This interface defines the contract for storing and retrieving chat messages and conversation history. Implementations can use different storage backends such as files, databases, or cloud storage services.

Constructor Parameters:

Methods:

add_attack_results_to_memory¶

add_attack_results_to_memory(attack_results: Sequence[AttackResult]) → None

Use add_attack_results_to_memory_async.

This synchronous API is deprecated and can block the caller.

add_attack_results_to_memory_async¶

add_attack_results_to_memory_async(attack_results: Sequence[AttackResult]) → None

Insert a list of attack results into the memory storage. The database model automatically calculates objective_sha256 for consistency.

Raises:

• SQLAlchemyError — If the database transaction fails.

add_conversation_branches_to_attack¶

add_conversation_branches_to_attack(attack_result_id: str, conversations: Sequence[Conversation], message_pieces: Sequence[MessagePiece], source_conversation: Conversation | None = None, new_attack: AttackResult | None = None, expected_fields: Mapping[str, Any] | None = None, update_fields: Mapping[str, Any] | None = None, request_fingerprint: str | None = None) → bool

Use add_conversation_branches_to_attack_async.

This synchronous API is deprecated and can block the caller.

Returns:

• bool — The operation result.

add_conversation_branches_to_attack_async¶

add_conversation_branches_to_attack_async(attack_result_id: str, conversations: Sequence[Conversation], message_pieces: Sequence[MessagePiece], source_conversation: Conversation | None = None, new_attack: AttackResult | None = None, expected_fields: Mapping[str, Any] | None = None, update_fields: Mapping[str, Any] | None = None, request_fingerprint: str | None = None) → bool

Atomically store initial or related conversations, pieces, and attack references.

The caller prepares the copies. This method only persists them, preserving the usual conversation and message insertion invariants. A supplied source must still be an active objective conversation when the transaction acquires the attack’s write lock. Supply new_attack to create the attack in the same transaction. An optional request fingerprint makes retries with the same conversation IDs idempotent.

Returns:

• bool — True for an insert; False for an identical retry or a missing destination.

Raises:

• ValueError — If the source is unrelated, branch IDs repeat, or pieces belong elsewhere.

• AttackStateConflictError — An expected field or creation identity changed.

• SQLAlchemyError — If persistence fails; the complete preparation is rolled back.

add_conversation_retry¶

add_conversation_retry(conversation_id: str, sequence: int, reason: ConversationRetryReason) → None

Use add_conversation_retry_async.

This synchronous API is deprecated and can block the caller.

add_conversation_retry_async¶

add_conversation_retry_async(conversation_id: str, sequence: int, reason: ConversationRetryReason) → None

Append a retry record to the conversation-scoped metadata for conversation_id.

Records that a turn had to be retried (e.g. because its response failed JSON validation and was rolled back out of memory). The conversation’s Conversations row is updated in place; if no row exists yet it is created. This is distinct from the insert-only _insert_conversation.

Raises:

• SQLAlchemyError — If the database update fails; the transaction is rolled back first.

add_conversation_to_memory¶

add_conversation_to_memory(conversation: Conversation) → None

Use add_conversation_to_memory_async.

This synchronous API is deprecated and can block the caller.

add_conversation_to_memory_async¶

add_conversation_to_memory_async(conversation: Conversation) → None

Register a conversation in memory, recording its conversation-scoped metadata.

A conversation is a first-class entity held with a single target. Build a Conversation when it is created and call this once (before, or independently of, adding its messages) to record the target it is held with. Message writes (add_message_to_memory / add_message_pieces_to_memory) deliberately do not take a target, so that conversation ownership is expressed in a single place rather than threaded through every write.

Registration is idempotent only for an identical conversation: re-registering the same conversation_id with the same target identity is a no-op, even across PyRIT versions (so repeated per-turn registration is safe). Re-registering an existing conversation_id with a different target is a conflict and raises ValueError -- a conversation is held with exactly one target and is never re-targeted.

The caller supplies ownership through conversation.attack_result_id. Registering a conversation that belongs to a different execution raises ValueError.

Raises:

• ValueError — If conversation_id is empty, or if a conversation with the same id already exists with a different target or belongs to a different attack execution.

add_finding_async¶

add_finding_async(finding: Finding, evidence: FindingEvidence | None = None) → Finding

Save a finding, and optionally its evidence, in one transaction.

Returns:

• Finding — The saved finding, or the finding already saved with this ID.

Raises:

• IntegrityError — If the insert fails for a reason other than a taken ID.

add_finding_evidence_async¶

add_finding_evidence_async(evidence: FindingEvidence) → FindingEvidence

Link a saved attack result to a finding, if the finding still exists.

The insert runs before the existence check so the write lock is held when the finding is checked, and a concurrent delete can’t leave an orphan.

Returns:

• FindingEvidence — The saved evidence, or the existing link for the same finding and attack result.

Raises:

• LookupError — If the finding does not exist.

• IntegrityError — If the insert fails for a reason other than a duplicate link.

add_message_pieces_to_memory¶

add_message_pieces_to_memory(message_pieces: Sequence[MessagePiece]) → None

Use add_message_pieces_to_memory_async.

This synchronous API is deprecated and can block the caller.

add_message_pieces_to_memory_async¶

add_message_pieces_to_memory_async(message_pieces: Sequence[MessagePiece]) → None

Insert a list of message pieces into the memory storage.

Pieces flagged via MessagePiece.not_in_memory = True are silently filtered out so callers don’t need to track persistence policy themselves. Every remaining piece must carry a non-empty conversation_id (the memory layer never invents one -- see _validate_persistable_conversation_ids).

Conversation-scoped metadata (the target a conversation is held with) is not recorded here; register it once via add_conversation_to_memory when the conversation is created.

This is a template method: subclasses implement only the backend-specific _add_message_pieces_to_memory and inherit the filtering and validation steps so no subclass can forget to run them.

add_message_to_memory¶

add_message_to_memory(request: Message) → None

Use add_message_to_memory_async.

This synchronous API is deprecated and can block the caller.

add_message_to_memory_async¶

add_message_to_memory_async(request: Message) → None

Insert a list of message pieces into the memory storage.

Automatically updates the sequence to be the next number in the conversation. If necessary, generates embedding data for applicable entries

add_operation_async¶

add_operation_async(operation: Operation) → Operation

Save an operation. Names are unique, ignoring case and surrounding whitespace.

Returns:

• Operation — The saved operation, or the existing one when the name is taken.

Raises:

• IntegrityError — If the insert fails for a reason other than a taken name.

add_scenario_results_to_memory¶

add_scenario_results_to_memory(scenario_results: Sequence[ScenarioResult]) → None

Use add_scenario_results_to_memory_async.

This synchronous API is deprecated and can block the caller.

add_scenario_results_to_memory_async¶

add_scenario_results_to_memory_async(scenario_results: Sequence[ScenarioResult]) → None

Insert a list of scenario results into the memory storage.

Raises:

• SQLAlchemyError — If a scenario result or identifier graph cannot be persisted.

add_scores_to_memory¶

add_scores_to_memory(scores: Sequence[Score], observations: Sequence[Observation] = (), intermediate_scores: Sequence[Score] = ()) → None

Use add_scores_to_memory_async.

This synchronous API is deprecated and can block the caller.

add_scores_to_memory_async¶

add_scores_to_memory_async(scores: Sequence[Score], observations: Sequence[Observation] = (), intermediate_scores: Sequence[Score] = ()) → None

Prepare file-backed loose content, then persist scores and observations.

intermediate_scores are results of nested scorers at any depth. They share the transaction but are excluded from default score queries.

Raises:

• ValueError — If an observation does not match its scored content.

add_seed_datasets_to_memory_async¶

add_seed_datasets_to_memory_async(datasets: Sequence[SeedDataset], added_by: str) → None

Insert a list of seed datasets into the memory storage.

add_seed_groups_to_memory_async¶

add_seed_groups_to_memory_async(prompt_groups: Sequence[SeedGroup], added_by: str | None = None) → None

Insert a list of seed groups into the memory storage.

Raises:

• ValueError — If a seed group does not have at least one seed.

• ValueError — If seed group IDs are inconsistent within the same seed group.

add_seeds_to_memory_async¶

add_seeds_to_memory_async(seeds: Sequence[Seed[Any]], added_by: str | None = None) → None

Insert a list of seeds into the memory storage.

Seeds with the same value hash, dataset, and canonical conditions already present in storage are skipped. Duplicates within seeds are all inserted, because the check looks at what storage held when the call started. Groups introducing new objective criteria retain the other seeds in the same group even when their content is already stored in another group. When a stored seed is copied to a new group, its caller-visible ID is updated after successful insertion.

Raises:

• ValueError — If the ‘added_by’ attribute is not set for each prompt.

cleanup¶

cleanup() → None

Ensure cleanup on process exit.

delete_conversation_pieces_after_sequence¶

delete_conversation_pieces_after_sequence(conversation_id: str, sequence: int) → int

Use delete_conversation_pieces_after_sequence_async.

This synchronous API is deprecated and can block the caller.

Returns:

• int — The operation result.

delete_conversation_pieces_after_sequence_async¶

delete_conversation_pieces_after_sequence_async(conversation_id: str, sequence: int) → int

Delete all message pieces in a conversation whose sequence is greater than sequence.

Rolls a conversation back to a baseline so a failed turn can be resent on a clean history. Dependent EmbeddingData rows for the deleted pieces are removed first to avoid orphaned foreign keys. Pieces at or below sequence (e.g. the system prompt and any prior good turns) are left intact.

Returns:

• int — The number of PromptMemoryEntries deleted.

Raises:

• SQLAlchemyError — If the deletion fails; the transaction is rolled back first.

delete_finding_async¶

delete_finding_async(operation_id: uuid.UUID, finding_id: uuid.UUID) → None

Permanently remove a finding within its operation.

Raises:

• LookupError — If the finding does not belong to the operation.

• RuntimeError — If the database does not return a cursor result.

delete_finding_evidence_async¶

delete_finding_evidence_async(finding_id: uuid.UUID, evidence_id: uuid.UUID) → None

Unlink evidence from a finding. The attack result is kept.

Raises:

• LookupError — If the evidence does not belong to the finding.

• RuntimeError — If the database does not return a cursor result.

disable_embedding¶

disable_embedding() → None

Disable embedding functionality for the memory interface.

Sets the memory_embedding attribute to None, disabling any embedding operations.

dispose_engine¶

dispose_engine() → None

Use dispose_engine_async.

This synchronous API is deprecated and can block the caller.

dispose_engine_async¶

dispose_engine_async() → None

Dispose async and legacy resources after operations have finished.

Raises:

• RuntimeError — If another event loop still owns memory resources.

dispose_loop_resources_async¶

dispose_loop_resources_async() → None

Close resources before their owning event loop stops.

duplicate_conversation¶

duplicate_conversation(conversation_id: str, attack_result_id: str | None = None) → str

Use duplicate_conversation_async.

This synchronous API is deprecated and can block the caller.

Returns:

• str — The operation result.

duplicate_conversation_async¶

duplicate_conversation_async(conversation_id: str, attack_result_id: str | None = None) → str

Duplicate a conversation for reuse.

This can be useful when an attack strategy requires branching out from a particular point in the conversation. One cannot continue both branches with the same conversation ID since that would corrupt the memory. Instead, one needs to duplicate the conversation and continue with the new conversation ID.

Returns:

• str — The uuid for the new conversation.

duplicate_conversation_excluding_last_turn¶

duplicate_conversation_excluding_last_turn(conversation_id: str, attack_result_id: str | None = None) → str

Use duplicate_conversation_excluding_last_turn_async.

This synchronous API is deprecated and can block the caller.

Returns:

• str — The operation result.

duplicate_conversation_excluding_last_turn_async¶

duplicate_conversation_excluding_last_turn_async(conversation_id: str, attack_result_id: str | None = None) → str

Duplicate a conversation, excluding the last turn. In this case, last turn is defined as before the last user request (e.g. if there is half a turn, it just removes that half).

This can be useful when an attack strategy requires back tracking the last prompt/response pair.

Returns:

• str — The uuid for the new conversation.

duplicate_messages¶

duplicate_messages(messages: Sequence[Message]) → tuple[str, Sequence[MessagePiece]]

Use duplicate_messages_async.

This synchronous API is deprecated and can block the caller.

Returns:

• tuple[str, Sequence[MessagePiece]] — tuple[str, Sequence[MessagePiece]]: The operation result.

duplicate_messages_async¶

duplicate_messages_async(messages: Sequence[Message]) → tuple[str, Sequence[MessagePiece]]

Duplicate messages with a new conversation ID.

Each duplicated piece gets a fresh id and timestamp while preserving original_prompt_id for tracking lineage.

Returns:

• tuple[str, Sequence[MessagePiece]] — Tuple of (new_conversation_id, duplicated_message_pieces).

enable_embedding¶

enable_embedding(embedding_model: Any | None = None) → None

Enable embedding functionality for the memory interface.

Raises:

• ValueError — If no embedding model is provided and required environment

get_all_embeddings¶

get_all_embeddings() → Sequence[EmbeddingDataEntry]

Use get_all_embeddings_async.

This synchronous API is deprecated and can block the caller.

Returns:

• Sequence[EmbeddingDataEntry] — Sequence[EmbeddingDataEntry]: The operation result.

get_all_embeddings_async¶

get_all_embeddings_async() → Sequence[EmbeddingDataEntry]

Load all EmbeddingData from the memory storage handler.

Returns:

• Sequence[EmbeddingDataEntry] — Sequence[EmbeddingDataEntry]: All stored embedding entries.

get_atomic_attack_identifiers¶

get_atomic_attack_identifiers(identifier_hashes: Sequence[str] | None = None, class_name: str | None = None, attack_technique_identifier_hash: str | None = None) → Sequence[AtomicAttackIdentifier]

Use get_atomic_attack_identifiers_async.

This synchronous API is deprecated and can block the caller.

Returns:

• Sequence[AtomicAttackIdentifier] — Sequence[AtomicAttackIdentifier]: The operation result.

get_atomic_attack_identifiers_async¶

get_atomic_attack_identifiers_async(identifier_hashes: Sequence[str] | None = None, class_name: str | None = None, attack_technique_identifier_hash: str | None = None) → Sequence[AtomicAttackIdentifier]

Retrieve atomic attack identifiers using exact normalized-column filters.

Returns:

• Sequence[AtomicAttackIdentifier] — Sequence[AtomicAttackIdentifier]: Matching identifiers ordered by content hash.

get_attack_identifiers¶

get_attack_identifiers(identifier_hashes: Sequence[str] | None = None, class_name: str | None = None, adversarial_system_prompt: str | None = None, adversarial_seed_prompt: str | None = None, objective_target_hash: str | None = None, adversarial_chat_hash: str | None = None, objective_scorer_hash: str | None = None) → Sequence[AttackIdentifier]

Use get_attack_identifiers_async.

This synchronous API is deprecated and can block the caller.

Returns:

• Sequence[AttackIdentifier] — Sequence[AttackIdentifier]: The operation result.

get_attack_identifiers_async¶

get_attack_identifiers_async(identifier_hashes: Sequence[str] | None = None, class_name: str | None = None, adversarial_system_prompt: str | None = None, adversarial_seed_prompt: str | None = None, objective_target_hash: str | None = None, adversarial_chat_hash: str | None = None, objective_scorer_hash: str | None = None) → Sequence[AttackIdentifier]

Retrieve attack identifiers using exact normalized-column filters.

Returns:

• Sequence[AttackIdentifier] — Sequence[AttackIdentifier]: Matching identifiers ordered by content hash.

get_attack_result_conversations_async¶

get_attack_result_conversations_async(attack_result_id: str) → list[Conversation]

Read the conversations owned by one attack execution.

These include its objective conversation and any adversarial, scoring, converter and branch conversations created while it ran. A child attack’s conversations belong to the child’s result.

Returns:

• list[Conversation] — The owned conversations’ metadata, ordered by conversation ID.

get_attack_result_for_conversation_async¶

get_attack_result_for_conversation_async(conversation_id: str) → AttackResult | None

Read the newest result that references a main or pruned conversation.

This also works for upgraded history without a conversation-to-attack link. Internal conversation references are not included.

Returns:

• AttackResult | None — AttackResult | None: The newest referring result, or None if none exists.

get_attack_results¶

get_attack_results(attack_result_ids: Sequence[str] | None = None, conversation_id: str | None = None, objective: str | None = None, objective_sha256: Sequence[str] | None = None, outcome: str | None = None, attack_classes: Sequence[str] | None = None, atomic_attack_eval_hashes: Sequence[str] | None = None, converter_classes: Sequence[str] | None = None, converter_classes_match: Literal['all', 'any'] = 'all', has_converters: bool | None = None, include_scenario_attacks: bool = True, labels: Mapping[str, str | Sequence[str]] | None = None, operator: str | Sequence[str] | None = None, operation: str | Sequence[str] | None = None, operation_ids: Sequence[uuid.UUID] | None = None, targeted_harm_categories: Sequence[str] | None = None, identifier_filters: Sequence[IdentifierFilter] | None = None, scenario_result_id: str | None = None, result_selection: AttackResultSelection = AttackResultSelection.LATEST_PER_CONVERSATION, min_turns: int | None = None, max_turns: int | None = None, limit: int | None = None, after: AttackResultKeysetCursor | None = None) → Sequence[AttackResult]

Use get_attack_results_async.

This synchronous API is deprecated and can block the caller.

Returns:

• Sequence[AttackResult] — Sequence[AttackResult]: The operation result.

get_attack_results_async¶

get_attack_results_async(attack_result_ids: Sequence[str] | None = None, conversation_id: str | None = None, objective: str | None = None, objective_sha256: Sequence[str] | None = None, outcome: str | None = None, attack_classes: Sequence[str] | None = None, atomic_attack_eval_hashes: Sequence[str] | None = None, converter_classes: Sequence[str] | None = None, converter_classes_match: Literal['all', 'any'] = 'all', has_converters: bool | None = None, include_scenario_attacks: bool = True, labels: Mapping[str, str | Sequence[str]] | None = None, operator: str | Sequence[str] | None = None, operation: str | Sequence[str] | None = None, operation_ids: Sequence[uuid.UUID] | None = None, targeted_harm_categories: Sequence[str] | None = None, identifier_filters: Sequence[IdentifierFilter] | None = None, scenario_result_id: str | None = None, result_selection: AttackResultSelection = AttackResultSelection.LATEST_PER_CONVERSATION, min_turns: int | None = None, max_turns: int | None = None, limit: int | None = None, after: AttackResultKeysetCursor | None = None) → Sequence[AttackResult]

Retrieve a list of AttackResult objects based on the specified filters.

Returns:

• Sequence[AttackResult] — Sequence[AttackResult]: A list of AttackResult objects that match the specified filters.

Raises:

• ValueError — If result_selection is not a supported selection mode.

• ValueError — If any label key contains characters outside the allowlist [A-Za-z0-9_.-]+.

• ValueError — If limit or after is combined with attack_result_ids or objective_sha256 (id-batched lookups do not support SQL pagination).

get_attack_technique_identifiers¶

get_attack_technique_identifiers(identifier_hashes: Sequence[str] | None = None, class_name: str | None = None, attack_identifier_hash: str | None = None) → Sequence[AttackTechniqueIdentifier]

Use get_attack_technique_identifiers_async.

This synchronous API is deprecated and can block the caller.

Returns:

• Sequence[AttackTechniqueIdentifier] — Sequence[AttackTechniqueIdentifier]: The operation result.

get_attack_technique_identifiers_async¶

get_attack_technique_identifiers_async(identifier_hashes: Sequence[str] | None = None, class_name: str | None = None, attack_identifier_hash: str | None = None) → Sequence[AttackTechniqueIdentifier]

Retrieve attack technique identifiers using exact normalized-column filters.

Returns:

• Sequence[AttackTechniqueIdentifier] — Sequence[AttackTechniqueIdentifier]: Matching identifiers ordered by content hash.

get_conversation_messages¶

get_conversation_messages(conversation_id: str) → MutableSequence[Message]

Use get_conversation_messages_async.

This synchronous API is deprecated and can block the caller.

Returns:

• MutableSequence[Message] — MutableSequence[Message]: The operation result.

get_conversation_messages_async¶

get_conversation_messages_async(conversation_id: str) → MutableSequence[Message]

Retrieve a list of Message objects that have the specified conversation ID.

Returns:

• MutableSequence[Message] — MutableSequence[Message]: A list of chat memory entries with the specified conversation ID.

Raises:

• ValueError — If conversation_id is empty or None. A falsy id would cause the underlying get_message_pieces filter to be skipped, silently returning pieces from every conversation in memory.

get_conversation_metadata_async¶

get_conversation_metadata_async(conversation_id: str) → Conversation | None

Read the stored metadata for one conversation.

Returns:

• Conversation | None — Conversation metadata, or None if the conversation is not stored.

get_conversation_stats¶

get_conversation_stats(conversation_ids: Sequence[str]) → dict[str, ConversationStats]

Use get_conversation_stats_async.

This synchronous API is deprecated and can block the caller.

Returns:

• dict[str, ConversationStats] — dict[str, ‘ConversationStats’]: The operation result.

get_conversation_stats_async¶

get_conversation_stats_async(conversation_ids: Sequence[str]) → dict[str, ConversationStats]

Return lightweight aggregate statistics for one or more conversations.

Computes per-conversation message count (distinct sequence numbers), a truncated last-message preview, the first non-empty labels dict, and the earliest message timestamp using efficient SQL aggregation instead of loading full pieces.

Returns:

• dict[str, ConversationStats] — Mapping from conversation_id to ConversationStats.

• dict[str, ConversationStats] — Conversations with no pieces are omitted from the result.

get_converter_identifiers¶

get_converter_identifiers(identifier_hashes: Sequence[str] | None = None, class_name: str | None = None, supported_input_types: Sequence[str] | None = None, supported_output_types: Sequence[str] | None = None, converter_target_hash: str | None = None, sub_converter_hash: str | None = None) → Sequence[ConverterIdentifier]

Use get_converter_identifiers_async.

This synchronous API is deprecated and can block the caller.

Returns:

• Sequence[ConverterIdentifier] — Sequence[ConverterIdentifier]: The operation result.

get_converter_identifiers_async¶

get_converter_identifiers_async(identifier_hashes: Sequence[str] | None = None, class_name: str | None = None, supported_input_types: Sequence[str] | None = None, supported_output_types: Sequence[str] | None = None, converter_target_hash: str | None = None, sub_converter_hash: str | None = None) → Sequence[ConverterIdentifier]

Retrieve converter identifiers using exact normalized-column filters.

Returns:

• Sequence[ConverterIdentifier] — Sequence[ConverterIdentifier]: Matching identifiers ordered by content hash.

get_finding_async¶

get_finding_async(operation_id: uuid.UUID, finding_id: uuid.UUID) → Finding | None

Read a finding only within its owning operation.

Returns:

• Finding | None — Finding | None: The matching finding, if present.

get_finding_evidence_async¶

get_finding_evidence_async(finding_id: uuid.UUID, limit: int, offset: int) → list[FindingEvidence]

Read one page of a finding’s evidence, most recently linked first.

Returns:

• list[FindingEvidence] — list[FindingEvidence]: The evidence on the page.

Raises:

• ValueError — If the page bounds are invalid.

get_finding_evidence_by_source_async¶

get_finding_evidence_by_source_async(finding_id: uuid.UUID, attack_result_id: uuid.UUID) → FindingEvidence | None

Read the evidence linking a finding to a saved attack result.

Returns:

• FindingEvidence | None — FindingEvidence | None: The evidence, if linked.

get_finding_evidence_counts_async¶

get_finding_evidence_counts_async(finding_ids: Sequence[uuid.UUID]) → dict[uuid.UUID, int]

Count evidence for several findings in one query.

Returns:

• dict[uuid.UUID, int] — dict[uuid.UUID, int]: Counts for findings that have evidence.

get_finding_severity_counts_async¶

get_finding_severity_counts_async() → dict[uuid.UUID, dict[FindingSeverity, int]]

Count every operation’s findings by severity in one query.

Returns:

• dict[uuid.UUID, dict[FindingSeverity, int]] — dict[uuid.UUID, dict[FindingSeverity, int]]: Counts for operations with findings.

get_findings_async¶

get_findings_async(operation_id: uuid.UUID | None = None, limit: int = 20, offset: int = 0, title_query: str | None = None) → list[Finding]

Read one page of findings, highest severity first, then newest first.

Returns:

• list[Finding] — list[Finding]: The findings on the page.

Raises:

• ValueError — If the page bounds are invalid.

get_message_pieces¶

get_message_pieces(role: str | None = None, conversation_id: str | uuid.UUID | None = None, prompt_ids: Sequence[str | uuid.UUID] | None = None, labels: dict[str, str] | None = None, prompt_metadata: dict[str, str | int] | None = None, sent_after: datetime | None = None, sent_before: datetime | None = None, original_values: Sequence[str] | None = None, converted_values: Sequence[str] | None = None, data_type: str | None = None, not_data_type: str | None = None, converted_value_sha256: Sequence[str] | None = None, identifier_filters: Sequence[IdentifierFilter] | None = None, attack_result_id: str | None = None) → Sequence[MessagePiece]

Use get_message_pieces_async.

This synchronous API is deprecated and can block the caller.

Returns:

• Sequence[MessagePiece] — Sequence[MessagePiece]: The operation result.

get_message_pieces_async¶

get_message_pieces_async(role: str | None = None, conversation_id: str | uuid.UUID | None = None, prompt_ids: Sequence[str | uuid.UUID] | None = None, labels: dict[str, str] | None = None, prompt_metadata: dict[str, str | int] | None = None, sent_after: datetime | None = None, sent_before: datetime | None = None, original_values: Sequence[str] | None = None, converted_values: Sequence[str] | None = None, data_type: str | None = None, not_data_type: str | None = None, converted_value_sha256: Sequence[str] | None = None, identifier_filters: Sequence[IdentifierFilter] | None = None, attack_result_id: str | None = None) → Sequence[MessagePiece]

Retrieve a list of MessagePiece objects based on the specified filters.

Returns:

• Sequence[MessagePiece] — Sequence[MessagePiece]: A list of MessagePiece objects that match the specified filters.

Raises:

• Exception — If there is an error retrieving the prompts, an exception is logged and an empty list is returned.

get_observations¶

get_observations(observation_ids: Sequence[uuid.UUID | str]) → list[Observation]

Use get_observations_async.

This synchronous API is deprecated and can block the caller.

Returns:

• list[Observation] — list[Observation]: The operation result.

get_observations_async¶

get_observations_async(observation_ids: Sequence[uuid.UUID | str]) → list[Observation]

Load observations by ID.

Returns:

• list[Observation] — list[Observation]: Stored observations in the requested order. Missing IDs are omitted.

get_operations_async¶

get_operations_async(operation_ids: Sequence[uuid.UUID] | None = None, name: str | None = None) → list[Operation]

Read operations ordered by name, optionally matching IDs or a name regardless of case and spacing.

Returns:

• list[Operation] — list[Operation]: The matching operations.

get_prompt_scores¶

get_prompt_scores(role: str | None = None, conversation_id: str | uuid.UUID | None = None, prompt_ids: Sequence[str | uuid.UUID] | None = None, labels: dict[str, str] | None = None, prompt_metadata: dict[str, str | int] | None = None, sent_after: datetime | None = None, sent_before: datetime | None = None, original_values: Sequence[str] | None = None, converted_values: Sequence[str] | None = None, data_type: str | None = None, not_data_type: str | None = None, converted_value_sha256: Sequence[str] | None = None, include_intermediate: bool = False) → Sequence[Score]

Use get_prompt_scores_async.

This synchronous API is deprecated and can block the caller.

Returns:

• Sequence[Score] — Sequence[Score]: The operation result.

get_prompt_scores_async¶

get_prompt_scores_async(role: str | None = None, conversation_id: str | uuid.UUID | None = None, prompt_ids: Sequence[str | uuid.UUID] | None = None, labels: dict[str, str] | None = None, prompt_metadata: dict[str, str | int] | None = None, sent_after: datetime | None = None, sent_before: datetime | None = None, original_values: Sequence[str] | None = None, converted_values: Sequence[str] | None = None, data_type: str | None = None, not_data_type: str | None = None, converted_value_sha256: Sequence[str] | None = None, include_intermediate: bool = False) → Sequence[Score]

Retrieve scores attached to message pieces based on the specified filters.

Returns:

• Sequence[Score] — Sequence[Score]: A list of scores extracted from the message pieces.

get_request_from_response¶

get_request_from_response(response: Message) → Message

Use get_request_from_response_async.

This synchronous API is deprecated and can block the caller.

Returns:

• Message — The operation result.

get_request_from_response_async¶

get_request_from_response_async(response: Message) → Message

Retrieve the request that produced the given response.

Returns:

• Message — The corresponding message object.

Raises:

• ValueError — If the response is not from an assistant role or has no preceding request.

get_scenario_attack_result_deltas¶

get_scenario_attack_result_deltas(scenario_result_id: str, cursor: AttackResultKeysetCursor | None = None, limit: int = 100) → tuple[list[ScenarioAttackResultDelta], bool]

Use get_scenario_attack_result_deltas_async.

This synchronous API is deprecated and can block the caller.

Returns:

• tuple[list[ScenarioAttackResultDelta], bool] — tuple[list[ScenarioAttackResultDelta], bool]: The operation result.

get_scenario_attack_result_deltas_async¶

get_scenario_attack_result_deltas_async(scenario_result_id: str, cursor: AttackResultKeysetCursor | None = None, limit: int = 100) → tuple[list[ScenarioAttackResultDelta], bool]

Return bounded scenario-linked result deltas in ascending keyset order.

This projection selects only progress fields and the linked objective score. It never hydrates ORM relationships, prompt rows, or a full ScenarioResult.

Returns:

• tuple[list[ScenarioAttackResultDelta], bool] — tuple[list[ScenarioAttackResultDelta], bool]: The page and whether more rows exist.

Raises:

• ValueError — If the limit or cursor identifiers are invalid.

get_scenario_history_aggregates¶

get_scenario_history_aggregates(scenario_result_ids: Sequence[str], plan_scenario_ids: Sequence[str] = ()) → dict[str, ScenarioHistoryAggregate]

Use get_scenario_history_aggregates_async.

This synchronous API is deprecated and can block the caller.

Returns:

• dict[str, ScenarioHistoryAggregate] — dict[str, ScenarioHistoryAggregate]: The operation result.

get_scenario_history_aggregates_async¶

get_scenario_history_aggregates_async(scenario_result_ids: Sequence[str], plan_scenario_ids: Sequence[str] = ()) → dict[str, ScenarioHistoryAggregate]

Return one attempt aggregate per requested scenario run.

Persisted attempts are grouped into logical work units before being counted, so retries and errored re-runs of the same objective collapse into a single unit. For every scenario listed in plan_scenario_ids the persisted run plan resolves those units: attempts are matched to their planned atomic group and seed group (remapping objective-hash attribution onto the planned seed group ID), and attempts that resolve to no planned unit are excluded from the counters. Scenarios outside plan_scenario_ids keep the persisted attribution as the unit identity and count every unit, which is the legacy behavior for runs without a usable plan.

Returns:

• dict[str, ScenarioHistoryAggregate] — dict[str, ScenarioHistoryAggregate]: One aggregate per requested scenario ID.

get_scenario_identifiers¶

get_scenario_identifiers(identifier_hashes: Sequence[str] | None = None, class_name: str | None = None, version: int | None = None, techniques: Sequence[str] | None = None, datasets: Sequence[str] | None = None, objective_target_hash: str | None = None, objective_scorer_hash: str | None = None) → Sequence[ScenarioIdentifier]

Use get_scenario_identifiers_async.

This synchronous API is deprecated and can block the caller.

Returns:

• Sequence[ScenarioIdentifier] — Sequence[ScenarioIdentifier]: The operation result.

get_scenario_identifiers_async¶

get_scenario_identifiers_async(identifier_hashes: Sequence[str] | None = None, class_name: str | None = None, version: int | None = None, techniques: Sequence[str] | None = None, datasets: Sequence[str] | None = None, objective_target_hash: str | None = None, objective_scorer_hash: str | None = None) → Sequence[ScenarioIdentifier]

Retrieve scenario identifiers using exact normalized-column filters.

Returns:

• Sequence[ScenarioIdentifier] — Sequence[ScenarioIdentifier]: Matching identifiers ordered by content hash.

get_scenario_result_header¶

get_scenario_result_header(scenario_result_id: str) → ScenarioResult | None

Use get_scenario_result_header_async.

This synchronous API is deprecated and can block the caller.

Returns:

• ScenarioResult | None — ScenarioResult | None: The operation result.

get_scenario_result_header_async¶

get_scenario_result_header_async(scenario_result_id: str) → ScenarioResult | None

Return one ScenarioResult header without hydrating linked attack results.

get_scenario_results¶

get_scenario_results(scenario_result_ids: Sequence[str] | None = None, scenario_name: str | None = None, scenario_version: int | None = None, pyrit_version: str | None = None, added_after: datetime | None = None, added_before: datetime | None = None, labels: dict[str, str] | None = None, objective_target_endpoint: str | None = None, objective_target_model_name: str | None = None, identifier_filters: Sequence[IdentifierFilter] | None = None, limit: int | None = None) → Sequence[ScenarioResult]

Use get_scenario_results_async.

This synchronous API is deprecated and can block the caller.

Returns:

• Sequence[ScenarioResult] — Sequence[ScenarioResult]: The operation result.

get_scenario_results_async¶

get_scenario_results_async(scenario_result_ids: Sequence[str] | None = None, scenario_name: str | None = None, scenario_version: int | None = None, pyrit_version: str | None = None, added_after: datetime | None = None, added_before: datetime | None = None, labels: dict[str, str] | None = None, objective_target_endpoint: str | None = None, objective_target_model_name: str | None = None, identifier_filters: Sequence[IdentifierFilter] | None = None, limit: int | None = None) → Sequence[ScenarioResult]

Retrieve a list of ScenarioResult objects based on the specified filters.

Results are always ordered by completion_time descending (most recent first).

Returns:

• Sequence[ScenarioResult] — Sequence[ScenarioResult]: A list of ScenarioResult objects that match the specified filters, ordered by completion_time descending.

get_scenario_run_history_page¶

get_scenario_run_history_page(scenario_names: Sequence[str] | None = None, statuses: Sequence[str] | None = None, labels: Mapping[str, str | Sequence[str]] | None = None, operation_ids: Sequence[uuid.UUID] | None = None, cursor: ScenarioHistoryKeysetCursor | None = None, limit: int = 100) → tuple[list[ScenarioHistoryRunRecord], dict[str, ScenarioHistoryAggregate], bool]

Use get_scenario_run_history_page_async.

This synchronous API is deprecated and can block the caller.

Returns:

• tuple[list[ScenarioHistoryRunRecord], dict[str, ScenarioHistoryAggregate], bool] — tuple[list[ScenarioHistoryRunRecord], dict[str, ScenarioHistoryAggregate], bool]: The operation result.

get_scenario_run_history_page_async¶

get_scenario_run_history_page_async(scenario_names: Sequence[str] | None = None, statuses: Sequence[str] | None = None, labels: Mapping[str, str | Sequence[str]] | None = None, operation_ids: Sequence[uuid.UUID] | None = None, cursor: ScenarioHistoryKeysetCursor | None = None, limit: int = 100) → tuple[list[ScenarioHistoryRunRecord], dict[str, ScenarioHistoryAggregate], bool]

Return one descending scenario-history page and its database-side attempt metrics.

Only selected ScenarioResult columns and the linked AttackResult columns required for aggregate counts are read. Full ORM result objects and their relationships are never hydrated, and attempt metrics are reduced to one aggregate row per history row inside the database.

Returns:

• tuple[list[ScenarioHistoryRunRecord], dict[str, ScenarioHistoryAggregate], bool] — tuple[list[ScenarioHistoryRunRecord], dict[str, ScenarioHistoryAggregate], bool]: Page headers, attempt aggregates keyed by scenario ID, and whether another page exists.

Raises:

• ValueError — If the limit, cursor ID, or label keys are invalid.

get_scenario_run_state_page¶

get_scenario_run_state_page(states: Sequence[ScenarioRunState], after_id: str | None = None, limit: int = 500) → tuple[list[ScenarioRunStateRecord], bool]

Use get_scenario_run_state_page_async.

This synchronous API is deprecated and can block the caller.

Returns:

• tuple[list[ScenarioRunStateRecord], bool] — tuple[list[ScenarioRunStateRecord], bool]: The operation result.

get_scenario_run_state_page_async¶

get_scenario_run_state_page_async(states: Sequence[ScenarioRunState], after_id: str | None = None, limit: int = 500) → tuple[list[ScenarioRunStateRecord], bool]

Return one bounded ID/state page without hydrating ScenarioResults or AttackResults.

Returns:

• tuple[list[ScenarioRunStateRecord], bool] — tuple[list[ScenarioRunStateRecord], bool]: State records and whether another page exists.

Raises:

• ValueError — If the limit or cursor ID is invalid.

get_scorable_content¶

get_scorable_content(content_ids: Sequence[uuid.UUID | str]) → dict[uuid.UUID, ContentScorable]

Use get_scorable_content_async.

This synchronous API is deprecated and can block the caller.

Returns:

• dict[uuid.UUID, ContentScorable] — dict[uuid.UUID, ContentScorable]: The operation result.

get_scorable_content_async¶

get_scorable_content_async(content_ids: Sequence[uuid.UUID | str]) → dict[uuid.UUID, ContentScorable]

Load the loose content that stored scores are anchored on.

Returns:

• dict[uuid.UUID, ContentScorable] — dict[uuid.UUID, ContentScorable]: The content by id, omitting ids with no row.

get_scorable_content_hashes¶

get_scorable_content_hashes(content_ids: Sequence[uuid.UUID | str]) → dict[uuid.UUID, str]

Use get_scorable_content_hashes_async.

This synchronous API is deprecated and can block the caller.

Returns:

• dict[uuid.UUID, str] — dict[uuid.UUID, str]: The operation result.

get_scorable_content_hashes_async¶

get_scorable_content_hashes_async(content_ids: Sequence[uuid.UUID | str]) → dict[uuid.UUID, str]

Load the immutable content digest for stored score anchors.

Returns:

• dict[uuid.UUID, str] — dict[uuid.UUID, str]: SHA-256 digests by content ID.

get_scorer_identifiers¶

get_scorer_identifiers(identifier_hashes: Sequence[str] | None = None, class_name: str | None = None, scorer_type: str | None = None, score_aggregator: str | None = None, prompt_target_hash: str | None = None) → Sequence[ScorerIdentifier]

Use get_scorer_identifiers_async.

This synchronous API is deprecated and can block the caller.

Returns:

• Sequence[ScorerIdentifier] — Sequence[ScorerIdentifier]: The operation result.

get_scorer_identifiers_async¶

get_scorer_identifiers_async(identifier_hashes: Sequence[str] | None = None, class_name: str | None = None, scorer_type: str | None = None, score_aggregator: str | None = None, prompt_target_hash: str | None = None) → Sequence[ScorerIdentifier]

Retrieve scorer identifiers using exact normalized-column filters.

Returns:

• Sequence[ScorerIdentifier] — Sequence[ScorerIdentifier]: Matching identifiers ordered by content hash.

get_scores¶

get_scores(score_ids: Sequence[str] | None = None, score_type: str | None = None, score_category: str | None = None, sent_after: datetime | None = None, sent_before: datetime | None = None, identifier_filters: Sequence[IdentifierFilter] | None = None, include_intermediate: bool = False) → Sequence[Score]

Use get_scores_async.

This synchronous API is deprecated and can block the caller.

Returns:

• Sequence[Score] — Sequence[Score]: The operation result.

get_scores_async¶

get_scores_async(score_ids: Sequence[str] | None = None, score_type: str | None = None, score_category: str | None = None, sent_after: datetime | None = None, sent_before: datetime | None = None, identifier_filters: Sequence[IdentifierFilter] | None = None, include_intermediate: bool = False) → Sequence[Score]

Retrieve a list of Score objects based on the specified filters.

Returns:

• Sequence[Score] — Sequence[Score]: A list of Score objects that match the specified filters.

get_seed_dataset_names¶

get_seed_dataset_names() → Sequence[str]

Use get_seed_dataset_names_async.

This synchronous API is deprecated and can block the caller.

Returns:

• Sequence[str] — Sequence[str]: The operation result.

get_seed_dataset_names_async¶

get_seed_dataset_names_async() → Sequence[str]

Return a list of all seed dataset names in the memory storage.

Returns:

• Sequence[str] — Sequence[str]: A list of unique dataset names.

get_seed_dataset_summaries¶

get_seed_dataset_summaries() → Sequence[SeedDatasetSummary]

Use get_seed_dataset_summaries_async.

This synchronous API is deprecated and can block the caller.

Returns:

• Sequence[SeedDatasetSummary] — Sequence[SeedDatasetSummary]: The operation result.

get_seed_dataset_summaries_async¶

get_seed_dataset_summaries_async() → Sequence[SeedDatasetSummary]

Return aggregate metadata for datasets already loaded in memory.

The queries intentionally project only dataset metadata and counts. Seed values and media are never hydrated, so callers can build dataset cards without materializing every prompt or fetching providers.

Named dataset grouping and metadata matching stay on the original collated dataset_name column. NULL and empty names are aggregated separately into the unnamed population so their shared logical groups are counted exactly once.

Returns:

• Sequence[SeedDatasetSummary] — Sequence[SeedDatasetSummary]: One summary for each stored dataset, including

• Sequence[SeedDatasetSummary] — a single deterministic entry for seeds without a dataset name.

get_seed_example_async¶

get_seed_example_async(dataset_name: str | None, example_id: uuid.UUID) → list[SeedRecord]

Read one complete logical seed example from one dataset.

Seeds are read as in get_seed_examples_async.

Returns:

• list[SeedRecord] — list[SeedRecord]: The stored members, objectives first. The list is empty if the

• list[SeedRecord] — dataset does not contain it.

get_seed_examples_async¶

get_seed_examples_async(dataset_name: str | None, limit: int, after: DecodedKeysetCursor | None = None, data_types: Sequence[PromptDataType] | None = None, harm_categories: Sequence[str] | None = None, seed_types: Sequence[SeedType] | None = None, value_search: str | None = None) → tuple[dict[uuid.UUID, list[SeedRecord]], int, DecodedKeysetCursor | None]

Read one page of complete logical seed examples from one dataset.

A logical example is all seeds in the dataset that share a prompt_group_id, or one seed without a group. Examples are ordered by their earliest date_added, then by example ID, both descending, using the canonical textual UUID order on every backend. Values inside one filter use OR, different filters use AND, and any member can satisfy a filter. Members are stored-record projections: configurations remain raw text, and no templates are rendered or referenced files loaded.

Returns:

• dict[uuid.UUID, list[SeedRecord]] — tuple[dict[uuid.UUID, list[SeedRecord]], int, DecodedKeysetCursor | None]: The stored members of each

• int — example keyed by example ID in page order, objectives first; the number of examples that

• DecodedKeysetCursor | None — match the filters; and the sort key of the last example when more examples follow.

get_seed_groups¶

get_seed_groups(value: str | None = None, value_sha256: Sequence[str] | None = None, dataset_name: str | None = None, dataset_name_pattern: str | None = None, data_types: Sequence[str] | None = None, harm_categories: Sequence[str] | None = None, added_by: str | None = None, authors: Sequence[str] | None = None, groups: Sequence[str] | None = None, source: str | None = None, seed_type: SeedType | None = None, origin: SeedOrigin | None = None, parameters: Sequence[str] | None = None, metadata: dict[str, str | int] | None = None, prompt_group_ids: Sequence[uuid.UUID] | None = None, group_length: Sequence[int] | None = None) → Sequence[SeedGroup]

Use get_seed_groups_async.

This synchronous API is deprecated and can block the caller.

Returns:

• Sequence[SeedGroup] — Sequence[SeedGroup]: The operation result.

get_seed_groups_async¶

get_seed_groups_async(value: str | None = None, value_sha256: Sequence[str] | None = None, dataset_name: str | None = None, dataset_name_pattern: str | None = None, data_types: Sequence[str] | None = None, harm_categories: Sequence[str] | None = None, added_by: str | None = None, authors: Sequence[str] | None = None, groups: Sequence[str] | None = None, source: str | None = None, seed_type: SeedType | None = None, origin: SeedOrigin | None = None, parameters: Sequence[str] | None = None, metadata: dict[str, str | int] | None = None, prompt_group_ids: Sequence[uuid.UUID] | None = None, group_length: Sequence[int] | None = None) → Sequence[SeedGroup]

Retrieve groups of seed prompts based on the provided filtering criteria.

Returns:

• Sequence[SeedGroup] — Sequence[SeedGroup]: A list of SeedGroup objects that match the filtering criteria.

get_seed_identifiers¶

get_seed_identifiers(identifier_hashes: Sequence[str] | None = None, class_name: str | None = None, value: str | None = None, value_sha256: str | None = None, data_type: str | None = None, dataset_name: str | None = None, is_general_technique: bool | None = None) → Sequence[SeedIdentifier]

Use get_seed_identifiers_async.

This synchronous API is deprecated and can block the caller.

Returns:

• Sequence[SeedIdentifier] — Sequence[SeedIdentifier]: The operation result.

get_seed_identifiers_async¶

get_seed_identifiers_async(identifier_hashes: Sequence[str] | None = None, class_name: str | None = None, value: str | None = None, value_sha256: str | None = None, data_type: str | None = None, dataset_name: str | None = None, is_general_technique: bool | None = None) → Sequence[SeedIdentifier]

Retrieve seed identifiers using exact normalized-column filters.

Returns:

• Sequence[SeedIdentifier] — Sequence[SeedIdentifier]: Matching identifiers ordered by content hash.

get_seeds¶

get_seeds(value: str | None = None, value_sha256: Sequence[str] | None = None, dataset_name: str | None = None, dataset_name_pattern: str | None = None, data_types: Sequence[str] | None = None, harm_categories: Sequence[str] | None = None, added_by: str | None = None, authors: Sequence[str] | None = None, groups: Sequence[str] | None = None, source: str | None = None, seed_type: SeedType | None = None, origin: SeedOrigin | None = None, parameters: Sequence[str] | None = None, metadata: dict[str, str | int] | None = None, prompt_group_ids: Sequence[uuid.UUID] | None = None) → Sequence[Seed[Any]]

Use get_seeds_async.

This synchronous API is deprecated and can block the caller.

Returns:

• Sequence[Seed[Any]] — Sequence[Seed]: The operation result.

get_seeds_async¶

get_seeds_async(value: str | None = None, value_sha256: Sequence[str] | None = None, dataset_name: str | None = None, dataset_name_pattern: str | None = None, data_types: Sequence[str] | None = None, harm_categories: Sequence[str] | None = None, added_by: str | None = None, authors: Sequence[str] | None = None, groups: Sequence[str] | None = None, source: str | None = None, seed_type: SeedType | None = None, origin: SeedOrigin | None = None, parameters: Sequence[str] | None = None, metadata: dict[str, str | int] | None = None, prompt_group_ids: Sequence[uuid.UUID] | None = None) → Sequence[Seed[Any]]

Retrieve a list of seed prompts based on the specified filters.

List filters match substrings within decoded elements, not their JSON representation. Substring matching retains the backend’s LIKE case and wildcard semantics.

Returns:

• Sequence[Seed[Any]] — Sequence[Seed]: A list of seeds (e.g., SeedPrompt, SeedObjective, SeedSimulatedConversation) matching the criteria.

get_session¶

get_session() → Session

Use get_session_async.

This synchronous API is deprecated and can block the caller.

Returns:

• Session — The operation result.

get_session_async¶

get_session_async() → AsyncSession

Create an independent async session.

The caller must close the session with an async context manager.

Returns:

• AsyncSession — A session owned by the current event loop.

Raises:

• NotImplementedError — If a custom sync session hook has not been migrated.

get_target_identifiers¶

get_target_identifiers(identifier_hashes: Sequence[str] | None = None, class_name: str | None = None, endpoint: str | None = None, model_name: str | None = None, underlying_model_name: str | None = None, temperature: float | None = None, top_p: float | None = None, max_requests_per_minute: int | None = None, supported_auth_modes: Sequence[str] | None = None) → Sequence[TargetIdentifier]

Use get_target_identifiers_async.

This synchronous API is deprecated and can block the caller.

Returns:

• Sequence[TargetIdentifier] — Sequence[TargetIdentifier]: The operation result.

get_target_identifiers_async¶

get_target_identifiers_async(identifier_hashes: Sequence[str] | None = None, class_name: str | None = None, endpoint: str | None = None, model_name: str | None = None, underlying_model_name: str | None = None, temperature: float | None = None, top_p: float | None = None, max_requests_per_minute: int | None = None, supported_auth_modes: Sequence[str] | None = None) → Sequence[TargetIdentifier]

Retrieve target identifiers using exact normalized-column filters.

Returns:

• Sequence[TargetIdentifier] — Sequence[TargetIdentifier]: Matching identifiers ordered by content hash.

get_unique_attack_attribution¶

get_unique_attack_attribution() → dict[str, list[str]]

Use get_unique_attack_attribution_async.

This synchronous API is deprecated and can block the caller.

Returns:

• dict[str, list[str]] — dict[str, list[str]]: The operation result.

get_unique_attack_attribution_async¶

get_unique_attack_attribution_async() → dict[str, list[str]]

Return unique dedicated operator and operation values from indexed columns.

get_unique_attack_class_names¶

get_unique_attack_class_names() → list[str]

Use get_unique_attack_class_names_async.

This synchronous API is deprecated and can block the caller.

Returns:

• list[str] — list[str]: The operation result.

get_unique_attack_class_names_async¶

get_unique_attack_class_names_async() → list[str]

Return sorted unique attack class names from all stored attack results.

Extracts class_name from the atomic_attack_identifier JSON column via a database-level DISTINCT query.

Returns:

• list[str] — Sorted list of unique attack class name strings.

get_unique_attack_labels¶

get_unique_attack_labels(operator: Sequence[str] | None = None, operation: Sequence[str] | None = None, labels: Mapping[str, str | Sequence[str]] | None = None) → dict[str, list[str]]

Use get_unique_attack_labels_async.

This synchronous API is deprecated and can block the caller.

Returns:

• dict[str, list[str]] — dict[str, list[str]]: The operation result.

get_unique_attack_labels_async¶

get_unique_attack_labels_async(operator: Sequence[str] | None = None, operation: Sequence[str] | None = None, labels: Mapping[str, str | Sequence[str]] | None = None) → dict[str, list[str]]

Return unique arbitrary labels, optionally narrowed by indexed attribution first.

Returns:

• dict[str, list[str]] — dict[str, list[str]]: Mapping of label keys to sorted lists of

• dict[str, list[str]] — unique values.

get_unique_converter_class_names¶

get_unique_converter_class_names() → list[str]

Use get_unique_converter_class_names_async.

This synchronous API is deprecated and can block the caller.

Returns:

• list[str] — list[str]: The operation result.

get_unique_converter_class_names_async¶

get_unique_converter_class_names_async() → list[str]

Return sorted unique converter class names used across all attack results.

Extracts class_name values from the nested request_converters array within the atomic_attack_identifier JSON column via a database-level query.

Returns:

• list[str] — Sorted list of unique converter class name strings.

get_unique_scenario_labels¶

get_unique_scenario_labels() → dict[str, list[str]]

Use get_unique_scenario_labels_async.

This synchronous API is deprecated and can block the caller.

Returns:

• dict[str, list[str]] — dict[str, list[str]]: The operation result.

get_unique_scenario_labels_async¶

get_unique_scenario_labels_async() → dict[str, list[str]]

Return all unique label values across scenario results.

initialize_async¶

initialize_async() → None

Initialize the schema once without blocking the caller’s event loop.

Raises:

• RuntimeError — If initialization is already in progress.

print_schema¶

print_schema() → None

Use print_schema_async.

This synchronous API is deprecated and can block the caller.

print_schema_async¶

print_schema_async() → None

Print the schema without blocking the caller’s event loop.

promote_attack_conversation¶

promote_attack_conversation(attack_result_id: str, conversation_id: str) → bool

Use promote_attack_conversation_async.

This synchronous API is deprecated and can block the caller.

Returns:

• bool — The operation result.

promote_attack_conversation_async¶

promote_attack_conversation_async(attack_result_id: str, conversation_id: str) → bool

Promote an existing related conversation without losing concurrent branch additions.

Returns:

• bool — False when the attack no longer exists.

Raises:

• ValueError — If the requested conversation is not part of this attack.

• SQLAlchemyError — If the transaction fails.

remove_seed_groups_from_memory¶

remove_seed_groups_from_memory(value: str | None = None, exact: bool = True, value_sha256: Sequence[str] | None = None, dataset_name: str | None = None, dataset_name_pattern: str | None = None, data_types: Sequence[str] | None = None, harm_categories: Sequence[str] | None = None, added_by: str | None = None, authors: Sequence[str] | None = None, groups: Sequence[str] | None = None, source: str | None = None, seed_type: SeedType | None = None, origin: SeedOrigin | None = None, parameters: Sequence[str] | None = None, metadata: dict[str, str | int] | None = None, prompt_group_ids: Sequence[uuid.UUID] | None = None) → int

Use remove_seed_groups_from_memory_async.

This synchronous API is deprecated and can block the caller.

Returns:

• int — The operation result.

remove_seed_groups_from_memory_async¶

remove_seed_groups_from_memory_async(value: str | None = None, exact: bool = True, value_sha256: Sequence[str] | None = None, dataset_name: str | None = None, dataset_name_pattern: str | None = None, data_types: Sequence[str] | None = None, harm_categories: Sequence[str] | None = None, added_by: str | None = None, authors: Sequence[str] | None = None, groups: Sequence[str] | None = None, source: str | None = None, seed_type: SeedType | None = None, origin: SeedOrigin | None = None, parameters: Sequence[str] | None = None, metadata: dict[str, str | int] | None = None, prompt_group_ids: Sequence[uuid.UUID] | None = None) → int

Delete groups of seed prompts based on the provided filtering criteria.

Unlike remove_seeds_from_memory, which deletes only the individual seeds that match, this removes every seed that shares a prompt_group_id with any matching seed. This preserves group integrity: filtering by a single modality (e.g. data_types=[“image_path”]) or attribute removes the whole group rather than leaving a partial group behind. It accepts the same filtering parameters as get_seeds (plus an exact flag). It is recommended to call get_seed_groups with the same filters first to preview which groups will be removed. At least one filter must be provided to prevent accidental deletion of all seeds. The deletion runs in a single transaction, so it works consistently across SQLite and Azure SQL.

Only seeds that belong to a group are affected: a matching seed with no prompt_group_id (for example, one added individually via add_seeds_to_memory_async rather than as part of a group) is skipped, since it has no group to expand. Use remove_seeds_from_memory to delete ungrouped seeds.

Only database records are removed. For file-backed seeds (image_path, audio_path, video_path) the serialized file on disk is left in place; delete those files separately if they are no longer needed.

Returns:

• int — The number of seeds removed across all affected groups.

Raises:

• ValueError — If no filters are provided.

• SQLAlchemyError — If the database deletion fails (the transaction is rolled back).

remove_seeds_from_memory¶

remove_seeds_from_memory(value: str | None = None, exact: bool = True, value_sha256: Sequence[str] | None = None, dataset_name: str | None = None, dataset_name_pattern: str | None = None, data_types: Sequence[str] | None = None, harm_categories: Sequence[str] | None = None, added_by: str | None = None, authors: Sequence[str] | None = None, groups: Sequence[str] | None = None, source: str | None = None, seed_type: SeedType | None = None, origin: SeedOrigin | None = None, parameters: Sequence[str] | None = None, metadata: dict[str, str | int] | None = None, prompt_group_ids: Sequence[uuid.UUID] | None = None) → int

Use remove_seeds_from_memory_async.

This synchronous API is deprecated and can block the caller.

Returns:

• int — The operation result.

remove_seeds_from_memory_async¶

remove_seeds_from_memory_async(value: str | None = None, exact: bool = True, value_sha256: Sequence[str] | None = None, dataset_name: str | None = None, dataset_name_pattern: str | None = None, data_types: Sequence[str] | None = None, harm_categories: Sequence[str] | None = None, added_by: str | None = None, authors: Sequence[str] | None = None, groups: Sequence[str] | None = None, source: str | None = None, seed_type: SeedType | None = None, origin: SeedOrigin | None = None, parameters: Sequence[str] | None = None, metadata: dict[str, str | int] | None = None, prompt_group_ids: Sequence[uuid.UUID] | None = None) → int

Delete a list of seed prompts based on the specified filters.

Accepts the same filtering parameters as get_seeds (plus an exact flag). It is recommended to call get_seeds with the same filters first to preview which seeds will be removed. At least one filter must be provided to prevent accidental deletion of all seeds. The deletion runs in a single transaction, so it works consistently across SQLite and Azure SQL.

Only database records are removed. For file-backed seeds (image_path, audio_path, video_path) the serialized file on disk is left in place; delete those files separately if they are no longer needed.

Returns:

• int — The number of seeds removed.

Raises:

• ValueError — If no filters are provided.

• SQLAlchemyError — If the database deletion fails (the transaction is rolled back).

replace_seeds_for_dataset_async¶

replace_seeds_for_dataset_async(dataset_name: str, seeds: Sequence[Seed[Any]], added_by: str | None = None) → int

Atomically replace all stored seeds for a dataset with a new set.

Every existing SeedPromptEntries row for dataset_name is deleted and the provided seeds are inserted in a single transaction and commit; if the insert fails the delete is rolled back with it, so the previously stored seeds are preserved. Seeds are prepared (media serialized, SHA256 computed) before the transaction opens. Deduplication is intentionally skipped: this is a full replace, so the provided seeds are stored as given.

The isolation guarantee is the database transaction boundary: a reader that queries after the commit sees the complete new set. This holds on the file-backed SQLite and Azure SQL backends, where each session has its own connection. The in-memory SQLite backend shares a single connection across all sessions, so it does not isolate concurrent sessions from one another; callers that need to read a dataset while it is being replaced should use a file-backed or Azure SQL backend. RefreshDatasets replaces datasets sequentially, so it does not rely on cross-session isolation.

SeedPromptEntries has no dependent foreign keys, so no related rows are removed first. Deleting media-backed seeds (image_path, audio_path, video_path) removes only the database rows; any serialized media files they reference are left on disk. This matches every other seed-delete path and results in disk bloat, not data loss.

Returns:

• int — The number of SeedPromptEntries deleted before the new seeds were inserted.

Raises:

• ValueError — If dataset_name is empty, seeds is empty, or any seed’s dataset_name does not match dataset_name.

• SQLAlchemyError — If the replacement fails; the transaction is rolled back first.

require_operation_async¶

require_operation_async(operation_id: uuid.UUID) → Operation

Read a saved Operation by ID without creating or resolving a name.

Returns:

• Operation — The saved record.

Raises:

• LookupError — If the Operation does not exist.

reset_database¶

reset_database() → None

Use reset_database_async.

This synchronous API is deprecated and can block the caller.

reset_database_async¶

reset_database_async() → None

Drop and rebuild the schema using the blocking-only Alembic tooling.

resolve_operation_id_async¶

resolve_operation_id_async(operation_id: uuid.UUID | str | None, operation_name: str | None = None) → uuid.UUID | None

Return the saved Operation for an ID or a deprecated Operation name, without creating one.

An ID takes precedence over a name. Names match regardless of case and surrounding whitespace. Until 1.4.0, a name that matches no saved Operation is kept as a plain name rather than rejected.

TODO(PyRIT 1.4): Remove operation_name with Operation name attribution.

Returns:

• uuid.UUID | None — uuid.UUID | None: The saved Operation’s ID, or None when no ID is given and no saved Operation has the name.

Raises:

• ValueError — If the ID is malformed.

• LookupError — If the ID does not belong to a saved Operation.

try_update_scenario_run_state¶

try_update_scenario_run_state(scenario_result_id: str, expected_states: Collection[ScenarioRunState], scenario_run_state: ScenarioRunState, error_message: str | None = None, error_type: str | None = None) → bool

Use try_update_scenario_run_state_async.

This synchronous API is deprecated and can block the caller.

Returns:

• bool — The operation result.

try_update_scenario_run_state_async¶

try_update_scenario_run_state_async(scenario_result_id: str, expected_states: Collection[ScenarioRunState], scenario_run_state: ScenarioRunState, error_message: str | None = None, error_type: str | None = None) → bool

Update the run state only when the stored state is one of expected_states.

The compare and the write are a single UPDATE so a run that reached a terminal state on another thread is not overwritten. A read followed by update_scenario_run_state cannot give that guarantee because scenario preparation and cancellation run on different threads.

Returns:

• bool — True if the row was updated, False if it was missing or in another state.

Raises:

• ValueError — If expected_states is empty.

update_attack_result¶

update_attack_result(conversation_id: str, update_fields: dict[str, Any]) → bool

Use update_attack_result_async.

This synchronous API is deprecated and can block the caller.

Returns:

• bool — The operation result.

update_attack_result_async¶

update_attack_result_async(conversation_id: str, update_fields: dict[str, Any]) → bool

Update specific fields of an existing AttackResultEntry identified by conversation_id.

This method queries for the raw database entry by conversation_id and updates the specified fields in place, avoiding the creation of duplicate rows.

Returns:

• bool — True if the update was successful, False if the entry was not found.

Raises:

• ValueError — If update_fields is empty.

update_attack_result_by_id¶

update_attack_result_by_id(attack_result_id: str, update_fields: dict[str, Any]) → bool

Use update_attack_result_by_id_async.

This synchronous API is deprecated and can block the caller.

Returns:

• bool — The operation result.

update_attack_result_by_id_async¶

update_attack_result_by_id_async(attack_result_id: str, update_fields: dict[str, Any]) → bool

Update specific fields of an existing AttackResultEntry identified by its primary key.

Returns:

• bool — True if the update was successful, False if the entry was not found.

update_attack_result_conditionally_async¶

update_attack_result_conditionally_async(attack_result_id: str, expected_fields: Mapping[str, Any], update_fields: Mapping[str, Any], conversation_target: ComponentIdentifier | None = None, expected_conversation_pieces: Mapping[str, Sequence[MessagePiece]] | None = None) → bool

Compare and update prepared fields and conversation targets in one transaction.

Returns:

• bool — True after a successful update.

Raises:

• AttackStateConflictError — The attack changed or a conversation uses another target.

update_finding_async¶

update_finding_async(operation_id: uuid.UUID, finding_id: uuid.UUID, request: FindingCreate) → Finding

Replace a finding’s editable fields without changing identity or creation time.

Returns:

• Finding — The updated finding.

Raises:

• LookupError — If the finding does not belong to the operation.

• RuntimeError — If the database does not return a cursor result.

update_operation_async¶

update_operation_async(operation_id: uuid.UUID, description: str) → Operation

Replace an operation’s description. The last save wins; the name and identity are unchanged.

Returns:

• Operation — The updated operation.

Raises:

• LookupError — If the Operation does not exist.

update_prompt_entries_by_conversation_id¶

update_prompt_entries_by_conversation_id(conversation_id: str, update_fields: dict[str, Any]) → bool

Use update_prompt_entries_by_conversation_id_async.

This synchronous API is deprecated and can block the caller.

Returns:

• bool — The operation result.

update_prompt_entries_by_conversation_id_async¶

update_prompt_entries_by_conversation_id_async(conversation_id: str, update_fields: dict[str, Any]) → bool

Update prompt entries for a given conversation ID with the specified field values.

Returns:

• bool — True if the update was successful, False otherwise.

Raises:

• ValueError — If update_fields is empty or not provided.

• ValueError — If an entry is immutable observation evidence.

update_prompt_metadata_by_conversation_id¶

update_prompt_metadata_by_conversation_id(conversation_id: str, prompt_metadata: dict[str, str | int]) → bool

Use update_prompt_metadata_by_conversation_id_async.

This synchronous API is deprecated and can block the caller.

Returns:

• bool — The operation result.

update_prompt_metadata_by_conversation_id_async¶

update_prompt_metadata_by_conversation_id_async(conversation_id: str, prompt_metadata: dict[str, str | int]) → bool

Update the metadata of prompt entries in memory for a given conversation ID.

Returns:

• bool — True if the update was successful, False otherwise.

update_scenario_metadata¶

update_scenario_metadata(scenario_result_id: str, metadata: dict[str, Any]) → None

Use update_scenario_metadata_async.

This synchronous API is deprecated and can block the caller.

update_scenario_metadata_async¶

update_scenario_metadata_async(scenario_result_id: str, metadata: dict[str, Any]) → None

Replace the scenario_metadata JSON blob on an existing scenario result.

Used by the scenario layer to persist first-run state (e.g. objective_hashes) that resume needs to replay. Performs a targeted UPDATE so it doesn’t clobber other columns.

Raises:

• ValueError — If the scenario result is not found.

update_scenario_metadata_fields¶

update_scenario_metadata_fields(scenario_result_id: str, fields: Mapping[str, Any]) → None

Use update_scenario_metadata_fields_async.

This synchronous API is deprecated and can block the caller.

update_scenario_metadata_fields_async¶

update_scenario_metadata_fields_async(scenario_result_id: str, fields: Mapping[str, Any]) → None

Merge selected fields into persisted scenario metadata in one transaction.

Raises:

• ValueError — If the scenario result is not found.

update_scenario_result_async¶

update_scenario_result_async(scenario_result: ScenarioResult) → None

Persist an updated scenario result.

update_scenario_run_state¶

update_scenario_run_state(scenario_result_id: str, scenario_run_state: ScenarioRunState, error_message: str | None = None, error_type: str | None = None) → None

Use update_scenario_run_state_async.

This synchronous API is deprecated and can block the caller.

update_scenario_run_state_and_metadata_fields¶

update_scenario_run_state_and_metadata_fields(scenario_result_id: str, scenario_run_state: ScenarioRunState, metadata_fields: Mapping[str, Any], error_message: str | None = None, error_type: str | None = None) → None

Use update_scenario_run_state_and_metadata_fields_async.

This synchronous API is deprecated and can block the caller.

update_scenario_run_state_and_metadata_fields_async¶

update_scenario_run_state_and_metadata_fields_async(scenario_result_id: str, scenario_run_state: ScenarioRunState, metadata_fields: Mapping[str, Any], error_message: str | None = None, error_type: str | None = None) → None

Update run state and merge scenario metadata in one transaction.

Raises:

• ValueError — If the scenario result is not found.

update_scenario_run_state_async¶

update_scenario_run_state_async(scenario_result_id: str, scenario_run_state: ScenarioRunState, error_message: str | None = None, error_type: str | None = None) → None

Update the run state of an existing scenario result.

Performs a targeted UPDATE of only the state/error columns instead of rebuilding the entire ScenarioResultEntry row.

Raises:

• ValueError — If the scenario result is not found.

PromptMemoryEntry¶

Bases: Base

Represents the prompt data.

Because of the nature of database and sql alchemy, type ignores are abundant :)

Constructor Parameters:

Methods:

get_message_piece¶

get_message_piece() → MessagePiece

Convert this database entry back into a MessagePiece object.

Returns:

• MessagePiece — The reconstructed message piece with all its data.

SQLiteMemory¶

Bases: MemoryInterface

A memory interface that uses SQLite as the backend database.

This class provides functionality to insert, query, and manage conversation data using SQLite. It supports both file-based and in-memory databases.

Cancellation finalizes active cursors and closes interrupted connections before returning to the caller. Session cleanup also finishes under repeated cancellation. Failed disconnects and session rollbacks preserve the original cancellation and expose cleanup failures as its cause.

Note: this is replacing the old DuckDB implementation.

Constructor Parameters:

Examples:

• Basic Memory Programming Usage

• SQLite Memory

Methods:

get_all_table_models¶

get_all_table_models() → list[type[Base]]

Return a list of all table models used in the database by inspecting the Base registry.

Returns:

• list[type[Base]] — list[Base]: A list of SQLAlchemy model classes.

get_session_async¶

get_session_async() → AsyncSession

Create a session with cancellation-safe SQLite cleanup.

In-memory sessions also have exclusive access to the shared database.

Returns:

• AsyncSession — A session that finishes cleanup before releasing exclusive access.

Raises:

• NotImplementedError — If a custom sync session hook has not been migrated.

• RuntimeError — If this thread already holds a synchronous session.

ScenarioHistoryAggregate¶

Attempt metrics for one scenario run, aggregated by the database.

Counters are computed over logical work units — every persisted attempt is first resolved to the planned unit it belongs to, so retried and errored attempts never inflate unit counts. The metrics query produces one aggregate per history row; a separate projection returns only distinct technique names, never one row per unit.

Methods:

empty¶

empty(scenario_result_id: str) → ScenarioHistoryAggregate

Build the zero-valued aggregate for a run with no persisted attempts.

Returns:

• ScenarioHistoryAggregate — Aggregate with all counters set to zero.

ScenarioHistoryKeysetCursor¶

Bases: NamedTuple

Descending keyset anchor for scenario history.

ScenarioHistoryRunRecord¶

Lightweight persisted scenario header for one history row.

ScenarioRunStateRecord¶

Lightweight persisted ID/state projection used for startup reconciliation.

SeedEntry¶

Bases: Base

Represents the raw prompt or prompt template data as found in open datasets.

Note: This is different from the PromptMemoryEntry which is the processed prompt data. SeedPrompt merely reflects basic prompts before plugging into attacks, running through models with corresponding attack strategies, and applying converters. PromptMemoryEntry captures the processed prompt data before and after the above steps.

Constructor Parameters:

Methods:

get_seed¶

get_seed() → SeedPrompt | SeedObjective | SeedSimulatedConversation

Convert this database entry back into a Seed object.

Returns:

• SeedPrompt | SeedObjective | SeedSimulatedConversation — SeedPrompt | SeedObjective | SeedSimulatedConversation: The reconstructed seed object.

Raises:

• ValueError — If persisted conditions are invalid or attached to a non-objective seed, or a simulated conversation record cannot be rebuilt, for example when it names a prompt file that is not present on this machine.

get_seed_record¶

get_seed_record() → SeedRecord

Project stored fields without reconstructing an executable seed.

Returns:

• SeedRecord — Stored content, identifiers, and metadata, including raw configuration text.

StorageIO¶

Bases: ABC

Abstract interface for storage systems (local disk, Azure Storage Account, etc.).

Methods:

create_directory_if_not_exists_async¶

create_directory_if_not_exists_async(path: Path | str) → None

Asynchronously creates a directory or equivalent in the storage system if it doesn’t exist.

delete_file_async¶

delete_file_async(path: Path | str) → None

Delete one caller-owned file; custom storage must implement cleanup explicitly.

is_file_async¶

is_file_async(path: Path | str) → bool

Asynchronously checks if the path refers to a file (not a directory or container).

path_exists_async¶

path_exists_async(path: Path | str) → bool

Asynchronously checks if a file or blob exists at the given path.

read_file_async¶

read_file_async(path: Path | str) → bytes

Asynchronously reads the file (or blob) from the given path.

write_file_async¶

write_file_async(path: Path | str, data: bytes) → None

Asynchronously writes data to the given path.

SupportedContentType¶

Bases: Enum

All supported content types for uploading blobs to provided storage account container. See all options here: https://www.iana.org/assignments/media-types/media-types.xhtml.

Examples:

• Workflows

TextDataTypeSerializer¶

Bases: DataTypeSerializer

Serializer for text and text-like prompt values that stay in-memory.

Constructor Parameters:

Methods:

data_on_disk¶

data_on_disk() → bool

Indicate whether this serializer persists data on disk.

Returns:

• bool — Always False for text serializers.

URLDataTypeSerializer¶

Bases: DataTypeSerializer

Serializer for URL values and URL-backed local file references.

Constructor Parameters:

Methods:

data_on_disk¶

data_on_disk() → bool

Indicate whether this serializer persists data on disk.

Returns:

• bool — True for non-http values, False for URL values.

VideoPathDataTypeSerializer¶

Bases: DataTypeSerializer

Serializer for video path values stored on disk.

Constructor Parameters:

Methods:

data_on_disk¶

data_on_disk() → bool

Indicate whether this serializer persists data on disk.

Returns:

• bool — Always True for video path serializers.