crp.security¶

Auto-generated reference for the crp.security subpackage.

`security`¶

`crp.security` ¶

`ComplianceAuditTrail` ¶

Tamper-evident, HMAC-signed compliance audit trail (§7.14).

Provides an immutable, append-only log with cryptographic chaining that detects any modification to historical entries.

EU AI Act Art. 12: "High-risk AI systems shall technically allow for the automatic recording of events (logs) over the lifetime of the system."

ISO 42001 A.6.2.8: Organizations must maintain records of AI system development, deployment, operation, monitoring, and decommissioning.

Usage::

trail = ComplianceAuditTrail(signing_key=session_key)
trail.record(
    ComplianceEventType.DATA_INGESTED,
    session_id="abc-123",
    data={"source": "user-input", "size_bytes": 4096},
)
# Verify chain integrity
valid, broken_at = trail.verify_chain()
assert valid
# Export for regulatory review
export = trail.export()

`durable_path` `property` ¶

Path of the durable append log, when enabled.

`transcript_store` `property` ¶

The attached transcript store, or None when capture is disabled.

`entry_count` `property` ¶

Number of entries currently stored in the trail.

`enable_durable_append(path, fsync_every=1)` ¶

Enable crash-durable per-event JSONL appending (audit U-C9).

Every subsequently recorded entry is appended (and, per policy, fsync'd) so a process crash cannot lose the trail recorded so far. Idempotent: enabling twice keeps the first writer.

`flush_durable()` ¶

Flush + fsync the durable writer (clean close path).

`close_durable()` ¶

Flush, close and detach the durable writer (clean close path).

Must run BEFORE any code rewrites durable_path atomically: on Windows an open append handle on the target makes os.replace fail with WinError 5 (access denied), and any record() after this point would write to a closed file. After close, :meth:record skips durable appending (in-memory chain is authoritative for the final export).

`attach_transcript_store(store)` ¶

Attach a :class:crp.security.transcript_store.TranscriptStore.

Enables :meth:verify_transcripts and transcript-aware exports on this trail. Called by the dispatch router the first time a transcript is captured for a session.

`record(event_type, session_id='', data=None)` ¶

Append a tamper-evident entry to the audit trail.

Thread-safe. Each entry is chained to the previous via HMAC.

Parameters:

Name	Type	Description	Default
`event_type`	`ComplianceEventType \| str`	Compliance event type enum or custom string.	required
`session_id`	`str`	Session identifier. Falls back to the trail default.	`''`
`data`	`dict[str, Any] \| None`	Arbitrary event payload.	`None`

Returns:

Type	Description
`AuditEntry`	The newly created `AuditEntry`.

`verify_chain(verify_receipts=False, public_key_pem=None)` ¶

Verify the integrity of the entire audit trail.

Parameters:

Name	Type	Description	Default
`verify_receipts`	`bool`	Additionally verify each entry's Ed25519 `receipt` (audit_v2). Every entry must carry a valid receipt; entries without one fail verification.	`False`
`public_key_pem`	`bytes \| None`	PEM public key for receipt verification. Falls back to the trail's configured :class:`ActionSigner` public key.	`None`

Returns:

Type	Description
`bool`	`(is_valid, broken_at_sequence)`. `broken_at_sequence` is `-1`
`int`	when the chain is valid, otherwise the sequence of the first broken entry.

Raises:

Type	Description
`ValueError`	If `verify_receipts` is True but no public key is available (neither argument nor configured signer).

`verify_file(path, public_key_pem=None)` `classmethod` ¶

Offline verification of an exported audit JSONL file (A-PR-5 / I-C5).

Recomputes each line's content hash from the exported fields - so ANY edit to the file is detected, unlike the keyless recompute-the-chain check integrators used to perform (the chain is plain SHA-256 over public fields and fully re-derivable by an editor) - validates the previous_hash linkage, and, when public_key_pem is supplied, requires a valid Ed25519 receipt on every entry.

Parameters:

Name	Type	Description	Default
`path`	`str`	Path to an exported `<sid>.audit.jsonl` file.	required
`public_key_pem`	`bytes \| None`	Optional Ed25519 public key (PEM) for strict audit_v2 receipt verification of every entry.	`None`

Returns:

Type	Description
`bool`	`(is_valid, broken_at_line)` - 0-based JSONL line index of the
`int`	first problem, or -1 when the file is valid.

`verify_transcripts(*, session_id=None)` ¶

Verify stored transcripts against this trail (schema 2.1.0).

Companion to :meth:verify_chain: re-hashes stored transcript content and confirms it matches the TRANSCRIPT_RECORDED events and - for unredacted transcripts - the LLM_CALL_COMPLETED hashes already chained here. An edited transcript flags at the exact window.

Returns:

Type	Description
`Any`	class:`crp.security.transcript_store.TranscriptVerification`.

Raises:

Type	Description
`ValueError`	If no transcript store is attached (capture disabled).

`query(*, event_type=None, session_id=None, since=None, until=None)` ¶

Query audit entries with optional filters.

Parameters:

Name	Type	Description	Default
`event_type`	`str \| ComplianceEventType \| None`	Filter by event type.	`None`
`session_id`	`str \| None`	Filter by session ID.	`None`
`since`	`float \| None`	Include entries with timestamp >= this value.	`None`
`until`	`float \| None`	Include entries with timestamp <= this value.	`None`

Returns:

Type	Description
`list[AuditEntry]`	Matching `AuditEntry` objects in chronological order.

`export(*, include_signatures=True, since=None)` ¶

Export the full audit trail for regulatory review.

Parameters:

Name	Type	Description	Default
`include_signatures`	`bool`	Whether to include HMAC signatures in the export.	`True`
`since`	`float \| None`	Only export entries with timestamp >= this value.	`None`

Returns:

Type	Description
`dict[str, Any]`	Structured document suitable for compliance auditors, including
`dict[str, Any]`	chain integrity status and summary statistics.

`export_jsonl(*, since=None, include_transcripts=False)` ¶

Export as JSONL (one entry per line) for log aggregation.

Parameters:

Name	Type	Description	Default
`since`	`float \| None`	Only export entries with timestamp >= this value.	`None`
`include_transcripts`	`bool`	When True (and a transcript store is attached), append the stored window transcripts as additional `record_type="transcript"` lines so the bundle is self-contained: chain + verdicts + viewable query–response pairs (schema 2.1.0).	`False`

Returns:

Type	Description
`str`	Newline-delimited JSON string of all matching entries.

`export_ndjson(*, since=None)` ¶

Alias for :meth:export_jsonl - NDJSON is the primary format (§4.1).

Parameters:

Name	Type	Description	Default
`since`	`float \| None`	Only export entries with timestamp >= this value.	`None`

Returns:

Type	Description
`str`	Newline-delimited JSON string of all matching entries.

`export_ocsf(*, since=None, provider='', model='', include_transcripts=False)` ¶

Export events in OCSF API Activity format for SIEMs (§4.2).

Parameters:

Name	Type	Description	Default
`since`	`float \| None`	Only export entries with timestamp >= this value.	`None`
`provider`	`str`	Provider identifier for the destination endpoint field.	`''`
`model`	`str`	Model identifier for the destination endpoint field.	`''`
`include_transcripts`	`bool`	When True (and a transcript store is attached), TRANSCRIPT_RECORDED-derived records additionally carry the transcript reference and content hashes in `unmapped` (the transcript content itself stays in the attached store / :meth:`export_jsonl` bundle).	`False`

Returns:

Type	Description
`list[dict[str, Any]]`	List of OCSF `class_uid=6003` records. The HMAC and risk level
`list[dict[str, Any]]`	are carried in the `unmapped` extension.

`export_sarif(*, since=None)` ¶

Export events as a SARIF 2.1.0 log for GitHub integration (§4.3).

Parameters:

Name	Type	Description	Default
`since`	`float \| None`	Only export entries with timestamp >= this value.	`None`

Returns:

Type	Description
`dict[str, Any]`	SARIF 2.1.0 log dict. Non-INFO events become SARIF results; INFO
`dict[str, Any]`	events are emitted at `note` level. Used by CRP Scan (CRP-SPEC-013).

`ComplianceEventType` ¶

Bases: str, Enum

Compliance-grade event types beyond standard CRP events (§7.14.1).

`SessionBindingManager` ¶

HMAC-SHA256 session binding with OS keyring + zero-config fallback (§7.1).

Usage

mgr = SessionBindingManager() binding = mgr.create_session("session-123") sig = mgr.sign_request(b"payload") assert mgr.verify_request_signature(b"payload", sig)

`binding` `property` ¶

Return the binding.

`session_key` `property` ¶

Return current session key (raises if no session).

`key_version` `property` ¶

Current key version number.

`create_session(session_id='')` ¶

Create a new session with fresh nonce and derived key (§6A.1, §6A.2).

session_nonce: 32 bytes of cryptographic randomness
session_key: HMAC-SHA256(master_secret, nonce || session_id)

`sign_request(payload)` ¶

Compute HMAC-SHA256 signature over payload (§6A.3).

`verify_request_signature(payload, signature)` ¶

Constant-time HMAC verification (§6A.3).

Uses hmac.compare_digest to prevent timing attacks. Also checks previous key versions during rotation window.

`rotate_secret()` ¶

Rotate master secret with graceful rollover (§audit H2).

Generates a fresh 256-bit master secret, re-derives the session key, and keeps the previous key(s) valid for verification during the rotation window.

Returns:

Type	Description
`SessionBinding`	Updated SessionBinding with the new key material.

Raises:

Type	Description
`RuntimeError`	If no active session exists.

`store_to_keyring(service_name='crp-sdk')` ¶

Store master secret in OS keyring (DPAPI/Keychain/kernel).

Returns True if stored, False if keyring unavailable.

`from_keyring(service_name='crp-sdk')` `classmethod` ¶

Load master secret from OS keyring. Returns None if unavailable.

`Checkpoint` `dataclass` ¶

Inline human-in-the-loop declaration (SPEC-033 §3).

Attributes:

Name	Type	Description
`checkpoint_id`	`str`	Unique identifier for this checkpoint instance.
`trigger`	`CheckpointTrigger`	Condition that caused the checkpoint to fire.
`timeout`	`int`	Seconds before auto-action is taken.
`on_timeout`	`CheckpointTimeoutAction`	Action if human does not respond in time.
`on_reject`	`CheckpointRejectAction`	Action if human rejects.
`context`	`dict[str, Any]`	Arbitrary context dict for the reviewer UI.

`wait_for_resolution()` `async` ¶

Block until the checkpoint is resolved by a human reviewer.

If timeout expires, auto-resolves per on_timeout.

Returns:

Type	Description
`CheckpointResolution`	A `CheckpointResolution` - never `None` (Invariant 10).

Raises:

Type	Description
`TimeoutError`	Internally caught and converted to auto-resolution.

`resolve(resolution)` ¶

Called by a human reviewer (or webhook) to resolve this checkpoint.

Parameters:

Name	Type	Description	Default
`resolution`	`CheckpointResolution`	The reviewer's decision and optional edited output.	required

Returns:

Type	Description
`None`	None. Sets the internal resolved event so awaiting tasks unblock.

`decorate(when='always', timeout=300, on_timeout='escalate', on_reject='fallback')` `classmethod` ¶

Decorator factory: @checkpoint(when="risk >= HIGH").

Parameters:

Name	Type	Description	Default
`when`	`str`	Trigger condition string.	`'always'`
`timeout`	`int`	Seconds to await human review.	`300`
`on_timeout`	`str`	Action if the checkpoint times out.	`'escalate'`
`on_reject`	`str`	Action if the reviewer rejects.	`'fallback'`

Returns:

Type	Description
`Callable`	A decorator that wraps the function with checkpoint gating.

Raises:

Type	Description
`CRPError`	If the reviewer rejects and `on_reject` is `"halt"`.

`CheckpointRejectAction` ¶

Bases: str, Enum

What happens when a checkpoint is rejected by the human reviewer (SPEC-033 §3).

`CheckpointResolution` `dataclass` ¶

The result of a human reviewer resolving a checkpoint (SPEC-033 §3).

Attributes:

Name	Type	Description
`action`	`CheckpointResolutionAction`	Reviewer decision (approve/reject/edit).
`reviewer`	`str`	Identifier of the reviewer.
`timestamp`	`float`	Unix timestamp of the resolution.
`edited_output`	`str`	Optional replacement text when action is EDIT.
`audit_event`	`dict[str, Any] \| None`	Optional extra fields for the audit trail.

`to_audit_dict()` ¶

Produce an audit-trail compatible event dict (SPEC-011).

Returns:

Type	Description
`dict[str, Any]`	Dict with checkpoint resolution fields ready for `ComplianceAuditTrail`.

`CheckpointResolutionAction` ¶

Bases: str, Enum

The human reviewer's decision (SPEC-033 §3).

`CheckpointTimeoutAction` ¶

Bases: str, Enum

What happens when a checkpoint times out awaiting human response (SPEC-033 §3).

`CheckpointTrigger` ¶

Bases: str, Enum

What causes a checkpoint to fire (SPEC-033 §3).

`ClarificationAction` ¶

Bases: str, Enum

The user's response to a clarification request.

`ClarificationRequest` `dataclass` ¶

A request for user input raised by a CLARIFY operation or an oversight gate.

`to_dict()` ¶

Render for streaming / audit / a reviewer UI.

`ClarificationResolution` `dataclass` ¶

The outcome of resolving a clarification request.

`approved` `property` ¶

True when the resolution is an affirmative answer (for oversight gates).

`AIRiskLevel` ¶

Bases: str, Enum

EU AI Act risk classification levels (Art. 6) (§7.15.1).

`AISystemCategory` ¶

Bases: str, Enum

Categories of AI system use cases relevant to risk classification.

`ComplianceReporter` ¶

Generate compliance status reports (§7.15.3).

Maps CRP's native security controls to EU AI Act articles and ISO 42001 clauses, reporting implementation status for each.

Usage::

reporter = ComplianceReporter()
report = reporter.generate_report(session_stats={...})
print(report["summary"]["compliance_score"])

`generate_report(session_stats=None, risk_assessment=None)` ¶

Generate a comprehensive compliance status report.

Parameters:

Name	Type	Description	Default
`session_stats`	`dict[str, Any] \| None`	Optional runtime/session statistics to include.	`None`
`risk_assessment`	`RiskAssessment \| None`	Optional `RiskAssessment` to attach.	`None`

Returns:

Type	Description
`dict[str, Any]`	Dict with EU AI Act and ISO 42001 control lists, implementation
`dict[str, Any]`	counts, compliance percentages, and summary score.

`generate_technical_documentation(transparency=None, risk_assessment=None)` ¶

Generate EU AI Act Art. 11 technical documentation.

Parameters:

Name	Type	Description	Default
`transparency`	`TransparencyDeclaration \| None`	Optional transparency declaration to embed.	`None`
`risk_assessment`	`RiskAssessment \| None`	Optional risk assessment to embed.	`None`

Returns:

Type	Description
`dict[str, Any]`	Structured documentation dict suitable for submission to
`dict[str, Any]`	national competent authorities.

`RiskAssessment` `dataclass` ¶

AI system risk assessment result (§7.15.1).

EU AI Act Art. 9: Providers must establish a risk management system for the entire lifecycle of the high-risk AI system.

`to_dict()` ¶

Serialise the risk assessment to a JSON-safe dict.

Returns:

Type	Description
`dict[str, Any]`	Dict representation including risk level, category, factors,
`dict[str, Any]`	mitigations, and residual risks.

`RiskClassifier` ¶

Classify AI system risk level per EU AI Act (§7.15.1).

Helps users determine their obligation level based on how they deploy CRP within their AI system.

CRP itself is a context management tool - typically MINIMAL or LIMITED risk. However, if CRP is integrated into a high-risk AI system (e.g., employment screening, credit scoring), the overall system inherits the higher classification.

Usage::

classifier = RiskClassifier()
assessment = classifier.assess(
    category=AISystemCategory.CONTEXT_MANAGEMENT,
    intended_purpose="Managing context for a customer support chatbot",
    processes_personal_data=True,
)
print(f"Risk level: {assessment.risk_level.value}")

`assess(category=AISystemCategory.CONTEXT_MANAGEMENT, intended_purpose='', processes_personal_data=False, makes_automated_decisions=False, affects_fundamental_rights=False, safety_critical=False, profiles_individuals=False)` ¶

Perform risk assessment based on EU AI Act criteria.

Parameters:

Name	Type	Description	Default
`category`	`AISystemCategory`	AI system use-case category.	`CONTEXT_MANAGEMENT`
`intended_purpose`	`str`	Human-readable description of the system's purpose.	`''`
`processes_personal_data`	`bool`	Whether the system processes personal data.	`False`
`makes_automated_decisions`	`bool`	Whether decisions are automated.	`False`
`affects_fundamental_rights`	`bool`	Whether outputs affect fundamental rights.	`False`
`safety_critical`	`bool`	Whether the system is safety-critical.	`False`
`profiles_individuals`	`bool`	Whether individuals are profiled.	`False`

Returns:

Type	Description
`RiskAssessment`	`RiskAssessment` with level, mitigations, and residual risks.

`TransparencyDeclaration` `dataclass` ¶

Transparency declaration for AI system users (§7.15.2).

EU AI Act Art. 13: Providers must ensure that high-risk AI systems are designed and developed in such a way that their operation is sufficiently transparent to enable deployers to interpret the system's output and use it appropriately.

`to_dict()` ¶

Serialise the transparency declaration to a JSON-safe dict.

Returns:

Type	Description
`dict[str, Any]`	Dict representation suitable for disclosure dashboards or regulators.

`ConsentManager` ¶

Manage consent state for data processing purposes (§7.13.1).

EU AI Act Art. 13: Transparency requires clear communication about how data is processed and for what purposes.

Usage::

cm = ConsentManager("session-123")
cm.grant(ProcessingPurpose.ANALYTICS, reason="User opted in")
if cm.check(ProcessingPurpose.ANALYTICS):
    # Process analytics data
    ...
cm.withdraw(ProcessingPurpose.ANALYTICS, reason="User opted out")

`state` `property` ¶

Return the state.

`grant(purpose, reason='', expires_hours=0.0)` ¶

Grant consent for a processing purpose.

Parameters:

Name	Type	Description	Default
`purpose`	`ProcessingPurpose`	Processing purpose being consented to.	required
`reason`	`str`	Human-readable reason for the grant.	`''`
`expires_hours`	`float`	Optional expiry in hours; 0 means no expiry.	`0.0`

Returns:

Type	Description
`ConsentRecord`	The created `ConsentRecord`.

`deny(purpose, reason='')` ¶

Deny consent for a processing purpose.

Parameters:

Name	Type	Description	Default
`purpose`	`ProcessingPurpose`	Processing purpose being denied.	required
`reason`	`str`	Human-readable reason for the denial.	`''`

Returns:

Type	Description
`ConsentRecord`	The created `ConsentRecord`.

`withdraw(purpose, reason='')` ¶

Withdraw previously granted consent.

Parameters:

Name	Type	Description	Default
`purpose`	`ProcessingPurpose`	Processing purpose being withdrawn.	required
`reason`	`str`	Human-readable reason for the withdrawal.	`''`

Returns:

Type	Description
`ConsentRecord`	The created `ConsentRecord`.

`check(purpose)` ¶

Check if consent is granted for a purpose.

Parameters:

Name	Type	Description	Default
`purpose`	`ProcessingPurpose`	Processing purpose to check.	required

Returns:

Type	Description
`bool`	True if consent is granted. Required purposes return True by default;
`bool`	opt-in purposes return False until explicitly granted.

`check_required(purpose)` ¶

Check consent and raise if denied for a required purpose.

Parameters:

Name	Type	Description	Default
`purpose`	`ProcessingPurpose`	Processing purpose to check.	required

Returns:

Type	Description
`bool`	True if consent is granted.

Raises:

Type	Description
`ConsentRequiredError`	If the purpose is required and consent is missing.

`to_dict()` ¶

Export consent state for compliance reporting.

Returns:

Type	Description
`dict[str, Any]`	Serialised consent state dict.

`HumanOversightController` ¶

Implements human oversight controls (EU AI Act Art. 14) (§7.13.4).

ISO 42001 A.6.2.3: Organizations must establish processes for human oversight of AI systems appropriate to the risk level.

Usage::

hoc = HumanOversightController(OversightConfig(
    level=HumanOversightLevel.APPROVAL,
    require_approval_for_dispatch=True,
))

# Before dispatch:
if hoc.requires_approval("dispatch"):
    approval = hoc.request_approval("dispatch", {"task": "..."})
    # ... wait for human approval ...
    hoc.record_decision(approval.event_id, approved=True)

`config` `property` ¶

Return the config.

`level` `property` ¶

Return the level.

`requires_approval(operation)` ¶

Check if an operation requires human approval.

Parameters:

Name	Type	Description	Default
`operation`	`str`	One of "dispatch", "ingest", "export", "deletion".	required

Returns:

Type	Description
`bool`	True if the configured oversight level and per-operation flags
`bool`	require explicit human approval.

`check_autonomous_limit()` ¶

Check if the autonomous dispatch limit has been reached.

Returns:

Type	Description
`bool`	True if more autonomous dispatches are allowed (or no limit is set).

`record_autonomous_dispatch()` ¶

Record an autonomous dispatch for limit tracking.

Returns:

Type	Description
`None`	None. Increments the internal autonomous dispatch counter.

`request_approval(operation, details=None)` ¶

Create an approval request event.

Parameters:

Name	Type	Description	Default
`operation`	`str`	Operation requiring approval.	required
`details`	`dict[str, Any] \| None`	Arbitrary context for the reviewer.	`None`

Returns:

Type	Description
`OversightEvent`	The created `OversightEvent`.

`record_decision(event_id, approved, approved_by='', reason='')` ¶

Record a human oversight decision.

Parameters:

Name	Type	Description	Default
`event_id`	`str`	ID of the original approval request.	required
`approved`	`bool`	True if approved, False if denied.	required
`approved_by`	`str`	Identifier of the reviewer.	`''`
`reason`	`str`	Optional reason for the decision.	`''`

Returns:

Type	Description
`OversightEvent`	The created `OversightEvent`.

`record_halt(operation, reason, details=None)` ¶

Record a halt event (system stopped due to policy).

Parameters:

Name	Type	Description	Default
`operation`	`str`	Operation that was halted.	required
`reason`	`str`	Human-readable reason for the halt.	required
`details`	`dict[str, Any] \| None`	Additional context.	`None`

Returns:

Type	Description
`OversightEvent`	The created `OversightEvent`.

`should_halt_on_injection()` ¶

Check if processing should halt when injection is detected.

Returns:

Type	Description
`bool`	True if `halt_on_injection_detection` is enabled.

`should_halt_on_pii()` ¶

Check if processing should halt when PII is detected.

Returns:

Type	Description
`bool`	True if `halt_on_pii_detection` is enabled.

`to_dict()` ¶

Export oversight state for compliance reporting.

Returns:

Type	Description
`dict[str, Any]`	Dict summarising oversight level, autonomous dispatch counts, and
`dict[str, Any]`	approval/denial/halt event counts.

`HumanOversightLevel` ¶

Bases: str, Enum

Levels of human oversight (EU AI Act Art. 14) (§7.13.4).

`ProcessingPurpose` ¶

Bases: str, Enum

Data processing purposes - must be declared before processing (§7.13.2).

EU AI Act Art. 10: Data governance requires clear purpose limitation.

`ProcessingRecordKeeper` ¶

Maintain GDPR Article 30 processing records (§7.13.3).

Every data processing activity within a CRP session is recorded with its purpose, legal basis, data categories, and retention.

Usage::

keeper = ProcessingRecordKeeper("session-123")
keeper.record(
    purpose=ProcessingPurpose.FACT_EXTRACTION,
    data_categories=["text_input"],
    legal_basis="legitimate_interest",
    input_size_bytes=4096,
)
records = keeper.export()

`activity_count` `property` ¶

Number of recorded processing activities.

`record(purpose, data_categories, legal_basis='legitimate_interest', input_size_bytes=0, output_size_bytes=0, automated_decision=False, human_oversight=False, retention_period='session')` ¶

Record a data processing activity.

Parameters:

Name	Type	Description	Default
`purpose`	`ProcessingPurpose`	Processing purpose for the activity.	required
`data_categories`	`list[str]`	Categories of data processed (e.g. ["text_input"]).	required
`legal_basis`	`str`	GDPR legal basis string.	`'legitimate_interest'`
`input_size_bytes`	`int`	Size of input data in bytes.	`0`
`output_size_bytes`	`int`	Size of output data in bytes.	`0`
`automated_decision`	`bool`	Whether automated decision-making occurred.	`False`
`human_oversight`	`bool`	Whether human oversight was available.	`False`
`retention_period`	`str`	Retention descriptor (e.g. "session", "30d").	`'session'`

Returns:

Type	Description
`ProcessingActivity`	The created `ProcessingActivity`.

`export()` ¶

Export all processing records for regulatory review.

Returns:

Type	Description
`list[dict[str, Any]]`	List of dicts representing every recorded activity.

`summary()` ¶

Summarize processing activities for compliance dashboard.

Returns:

Type	Description
`dict[str, Any]`	Dict with totals, per-purpose counts, input/output byte sums, and
`dict[str, Any]`	oversight/decision counts.

`CustomSafetyRule` `dataclass` ¶

A user-defined safety rule registered as a first-class citizen (SPEC-033).

Attributes:

Name	Type	Description
`name`	`str`	Unique rule identifier.
`check_fn`	`Any`	Callable that performs the safety check. Type is kept as `Any` to avoid heavy import dependencies in the control plane module.
`description`	`str`	Human-readable explanation of what the rule detects.
`default`	`Any`	Default value for the rule when registered.

`SafetyControlPlane` `dataclass` ¶

Single place from which all CRP safety is seen, tuned, and extended (SPEC-033).

Attributes:

Name	Type	Description
`registry`	`dict[str, SafetyCapability]`	Dict of capability-name → SafetyCapability.
`manifest`	`SafetyManifest`	The SafetyManifest that drives settings.
`coverage`	`SafetyCoverageMap`	The SafetyCoverageMap (capabilities + out-of-scope).
`checkpoints`	`SafetyCoverageMap`	Active checkpoint instances awaiting resolution.

`get_capability(name)` ¶

Retrieve a capability from the registry.

Parameters:

Name	Type	Description	Default
`name`	`str`	Capability identifier.	required

Returns:

Type	Description
`SafetyCapability \| None`	The matching `SafetyCapability` or `None` if not registered.

`list_capabilities()` ¶

Return all registered capabilities.

Returns:

Type	Description
`list[SafetyCapability]`	List of every built-in, addable, and custom capability in the registry.

`tune(name, value)` ¶

Change the current value of a capability and sync to Manifest.

Parameters:

Name	Type	Description	Default
`name`	`str`	Capability identifier.	required
`value`	`Any`	New current value. Should normally be within `allowed_range`.	required

Returns:

Type	Description
`None`	None. Logs a warning if the capability is unknown.

`register_rule(rule)` ¶

Register a custom safety rule as a first-class citizen.

Parameters:

Name	Type	Description	Default
`rule`	`CustomSafetyRule`	Custom rule definition including name, check function, and default.	required

Returns:

Type	Description
`None`	None. The rule is mirrored into the registry and coverage map.

`create_checkpoint(trigger='always', timeout=300, on_timeout='escalate', on_reject='fallback', context=None)` ¶

Create and track a new human-in-the-loop checkpoint.

Parameters:

Name	Type	Description	Default
`trigger`	`str`	Condition that fires the checkpoint. One of the `CheckpointTrigger` values or a custom string.	`'always'`
`timeout`	`int`	Seconds to wait for human review before auto-resolution.	`300`
`on_timeout`	`str`	Action taken when the checkpoint times out.	`'escalate'`
`on_reject`	`str`	Action taken when the human reviewer rejects.	`'fallback'`
`context`	`dict[str, Any] \| None`	Arbitrary dict passed to the reviewer UI/webhook.	`None`

Returns:

Type	Description
`Checkpoint`	The created `Checkpoint` instance, tracked by ID internally.

`resolve_checkpoint(checkpoint_id, resolution)` ¶

Resolve a checkpoint by ID (called by reviewer or webhook).

Parameters:

Name	Type	Description	Default
`checkpoint_id`	`str`	UUID of the checkpoint returned by `create_checkpoint`.	required
`resolution`	`CheckpointResolution`	Human reviewer's decision and optional edited output.	required

Returns:

Type	Description
`None`	None. Logs an error if the checkpoint ID is unknown.

`get_surface_map()` ¶

Return the complete safety surface as a dict - for UI dashboards.

Returns:

Type	Description
`dict[str, Any]`	Dict with `registry`, `manifest`, `coverage`, and
`dict[str, Any]`	`active_checkpoints` keys.

`show()` ¶

Return a human-readable printout of the entire safety surface.

Returns:

Type	Description
`str`	Multi-line string listing every capability's current/default values,
`str`	allowed ranges, effects, and the explicit out-of-scope list.

`SafetyCapability` `dataclass` ¶

One entry in the Safety Registry / Coverage Map (SPEC-033, SPEC-034).

Attributes:

Name	Type	Description
`name`	`str`	Unique capability identifier.
`description`	`str`	What the capability evaluates.
`spec`	`str`	Specification that defines the capability (e.g. "005", "033").
`default`	`Any`	Factory/default value.
`current`	`Any`	Active value after tuning.
`allowed_range`	`list[Any] \| None`	Optional list or range of permitted values.
`effect`	`str`	Human-readable effect when the capability triggers.
`addable`	`bool`	True if the capability comes from SPEC-034 addable rules.

`to_dict()` ¶

Serialise to dict for dashboard/export.

Returns:

Type	Description
`dict[str, Any]`	JSON-safe dict representation of this capability.

`SafetyCoverageMap` `dataclass` ¶

The complete map of detectable risks + explicit out-of-scope list (SPEC-034).

The out-of-scope list is shown in the Control Plane too - honesty is a feature.

`register(capability)` ¶

Add or overwrite a capability in the map.

Parameters:

Name	Type	Description	Default
`capability`	`SafetyCapability`	Capability descriptor to register.	required

Returns:

Type	Description
`None`	None.

`get(name)` ¶

Lookup a capability by name.

Parameters:

Name	Type	Description	Default
`name`	`str`	Capability identifier.	required

Returns:

Type	Description
`SafetyCapability \| None`	The matching `SafetyCapability` or `None`.

`list_addable()` ¶

Return only the addable rules (SPEC-034).

Returns:

Type	Description
`list[SafetyCapability]`	List of capabilities where `addable` is True.

`to_dict()` ¶

Serialise for dashboard rendering or config export.

Returns:

Type	Description
`dict[str, Any]`	Dict with `capabilities` and `out_of_scope` keys.

`EmbeddingDefense` ¶

SQ8 quantization + XOR salting for embedding protection (§7.11).

Usage

defense = EmbeddingDefense() protected = defense.protect([0.1, 0.2, -0.3, ...]) recovered = defense.recover(protected)

recovered ≈ original (within quantization error)¶

export_state: embeddings are stripped¶

safe_data = defense.strip_embeddings_for_export(state_dict)

`protect(embedding, salt=None)` ¶

Apply SQ8 quantization + XOR salting (§6H.1, §6H.2).

SQ8: Maps float32 range [min, max] → int8 [-128, 127]. XOR: Applies 4-byte repeating XOR mask to quantized bytes.

`recover(protected)` ¶

Recover embedding from SQ8 + XOR protected form.

Returns approximate original values (quantization introduces error).

`strip_embeddings_for_export(state_dict)` `staticmethod` ¶

Strip all embeddings from state dict for export (§6H.3).

export_state() must export text only - no embeddings.

`ProtectedEmbedding` `dataclass` ¶

Embedding with SQ8 quantization and XOR salt applied.

`to_dict()` ¶

Serialize the protected embedding to a base64-encoded dict.

`from_dict(data)` `classmethod` ¶

Create a new instance from a dictionary.

Parameters:

Name	Type	Description	Default
`data`	`dict[str, Any]`	The data value.	required

Returns:

Type	Description
`ProtectedEmbedding`	`ProtectedEmbedding`.

`EncryptedBlob` `dataclass` ¶

Encrypted data container.

`to_dict()` ¶

Serialize the encrypted blob to a base64-encoded dict.

`from_dict(data)` `classmethod` ¶

Create a new instance from a dictionary.

Parameters:

Name	Type	Description	Default
`data`	`dict[str, str]`	The data value.	required

Returns:

Type	Description
`EncryptedBlob`	`EncryptedBlob`.

`StateEncryptor` ¶

AES-256-GCM encryption for cold state and event logs (§7.3).

Usage

enc = StateEncryptor(session_key) blob = enc.encrypt_cold_state(data_bytes) plaintext = enc.decrypt_cold_state(blob)

`encrypt_cold_state(data)` ¶

Encrypt cold state data (§6C.2).

`decrypt_cold_state(blob)` ¶

Decrypt cold state data (§6C.2).

`encrypt_event_log(data)` ¶

Encrypt event log data (§6C.3).

`decrypt_event_log(blob)` ¶

Decrypt event log data (§6C.3).

`InjectionDetector` ¶

Advisory injection detection - NEVER blocks, only reports (§7.5).

CRITICAL DESIGN CONSTRAINT: This detector NEVER modifies input text and NEVER prevents processing. It only produces advisory flags that are reported to QualityReport.security_flags.

Detection layers (ensembled automatically): Layer 1: Regex pattern library (always active) Layer 2: ML classifier (auto-detected, optional) - CRP safety DeBERTa: crp-safety-deberta-v1 safe/unsafe (CRPv6 Phase A) - prompt-injection-detector: TF-IDF + Logistic Regression (~1MB, MIT) - ProtectAI DeBERTa v2: ONNX transformer (~350MB, Apache 2.0)

Usage

detector = InjectionDetector() report = detector.scan("ignore all previous instructions") if report.has_flags: quality_report.security_flags = report.security_flags

`ml_enabled` `property` ¶

Whether ML-based injection detection is available.

`ml_backend` `property` ¶

Name of the ML backend in use, or 'none'.

`scan(text)` ¶

Scan text for injection patterns.

Returns advisory report - NEVER blocks (§6E.1). Ensembles regex patterns with ML classifier when available.

`InjectionReport` `dataclass` ¶

Result of injection detection - advisory only.

`has_flags` `property` ¶

Return whether this object has flags.

`highest_confidence` `property` ¶

Return the highest confidence.

`security_flags` `property` ¶

Return flag strings for QualityReport.security_flags (§6E.3).

`InjectionType` ¶

Bases: str, Enum

Categories of detected injection patterns.

`FactIntegrityChain` ¶

Tamper-evident chain of fact hashes (§7.7).

Maintains an ordered hash chain. Chain signature is computed via HMAC(session_key, hash_N ‖ ... ‖ hash_0).

Usage

chain = FactIntegrityChain(session_key) chain.add_fact("f1", "capital of France is Paris") sig = chain.chain_signature() assert chain.verify_chain(sig)

`size` `property` ¶

Return the current size count.

`add_fact(fact_id, text)` ¶

Hash a fact and append to the chain.

`get_hash(fact_id)` ¶

Get stored hash for a fact by ID.

`verify_fact(fact_id, text)` ¶

Verify a single fact's hash matches the chain.

`chain_signature()` ¶

Compute HMAC(session_key, hash_N ‖ ... ‖ hash_0) (§6B.3).

Hash chain is concatenated in reverse order (newest first).

`verify_chain(expected_signature)` ¶

Verify the full chain signature matches.

`verify_spot_check(fact_texts, sample_ratio=0.1)` ¶

Spot-check 10% of facts on cold load (§6B.4).

Parameters:

Name	Type	Description	Default
`fact_texts`	`dict[str, str]`	{fact_id: current_text} for verification	required
`sample_ratio`	`float`	fraction of chain to sample (default 10%)	`0.1`

Returns:

Type	Description
`tuple[int, int, list[str]]`	(checked, failures, failed_ids)

`verify_for_envelope(fact_ids, fact_texts)` ¶

Verify all facts before including in envelope (§6B.5).

Returns (all_valid, failed_ids).

`to_dict()` ¶

Serialize the integrity chain entries to a dict.

`export_for_verification()` ¶

Export chain data for external verification (§audit M13).

Returns a dict containing all chain entries and (if session key is set) the chain signature, suitable for independent audit.

`verify_external(export_data, fact_texts)` `staticmethod` ¶

Verify an exported chain against provided fact texts (§audit M13).

This is a static method usable without access to the session key - it re-hashes each fact and compares to the stored hash.

Returns (all_valid, failed_fact_ids).

`from_dict(data, session_key=None)` `classmethod` ¶

Create a new instance from a dictionary.

Parameters:

Name	Type	Description	Default
`data`	`dict[str, Any]`	The data value.	required
`session_key`	`bytes \| None`	The session key value.	`None`

Returns:

Type	Description
`FactIntegrityChain`	`FactIntegrityChain`.

`KillIncident` `dataclass` ¶

Forensic record of a kill-switch fire event.

`to_dict()` ¶

Serialize incident to a JSON-safe dict.

`KillSwitch` ¶

Auditable emergency stop for a CRP session or agent.

Attributes:

Name	Type	Description
`state`	`KillSwitchState`	Current `KillSwitchState`.
`audit_callback`	`KillSwitchState`	Optional callable `fn(incident: KillIncident)` that receives every fired incident (e.g., to append to an audit trail).
`on_fire`	`KillSwitchState`	Optional callable `fn(incident: KillIncident)` invoked when the switch fires; useful to terminate the caller's runtime loop.

`state` `property` ¶

Current state of the kill-switch.

`is_fired` `property` ¶

True once the switch has fired.

`incidents` `property` ¶

All incidents recorded by this switch (typically one after firing).

`arm()` ¶

Re-arm a disarmed switch. Cannot re-arm a fired switch.

`disarm()` ¶

Disarm the switch - it will not fire until re-armed.

`fire(session_id, reason, triggered_by, snapshot=None)` ¶

Fire the kill-switch and record an incident.

Parameters:

Name	Type	Description	Default
`session_id`	`str`	Identifier of the guarded session.	required
`reason`	`str`	Canonical or human-readable reason for firing.	required
`triggered_by`	`str`	Identifier of the subsystem / user that triggered it.	required
`snapshot`	`dict[str, Any] \| None`	Arbitrary serialisable context captured at the moment of firing (e.g., last window id, safety budget, trust score).	`None`

Returns:

Type	Description
`KillIncident`	The recorded `KillIncident`. If the switch was already fired or
`KillIncident`	disarmed, a synthetic incident is returned without side effects.

`latest_incident()` ¶

Return the most recent incident, or None if the switch never fired.

`KillSwitchReason` ¶

Bases: str, Enum

Canonical reasons a kill-switch can fire.

`KillSwitchState` ¶

Bases: str, Enum

Kill-switch lifecycle states.

`DataClassification` ¶

Bases: IntEnum

Data sensitivity classification (§7.12.1).

Higher values = more sensitive. Controls encryption, retention, access requirements, and audit verbosity.

`DataLineageTracker` ¶

Track data provenance and transformation history (§7.12.5).

ISO 42001 A.6.2.6: Data management requires tracking data origin, quality, and transformations throughout the AI lifecycle.

Usage::

tracker = DataLineageTracker()
tracker.record("fact-1", "extraction", "user-input", DataClassification.INTERNAL)
tracker.add_transformation("fact-1", "quarantine_promoted")

`record(data_id, origin, source_label='', classification=DataClassification.INTERNAL, parent_ids=None)` ¶

Record a new data lineage entry.

`add_transformation(data_id, transformation)` ¶

Record a transformation applied to data. Returns True if found.

`reclassify(data_id, new_classification)` ¶

Update classification level for a data item. Returns True if found.

`get_lineage(data_id)` ¶

Get lineage entry for a data item.

`get_by_classification(level)` ¶

Get all entries at or above a classification level.

`to_dict()` ¶

Export lineage state for audit/compliance reporting.

`ErasureManager` ¶

Handles right-to-erasure requests (GDPR Article 17) (§7.12.4).

Tracks erasure requests, ensures they are fulfilled, and maintains an audit trail of what was deleted and when.

Usage::

em = ErasureManager()
req = em.create_request("user-hash-abc", scope="session")
# ... caller erases the actual data ...
em.complete_request(req.request_id, items_erased=42)

`create_request(requester_hash, scope='session', target_ids=None)` ¶

Create a new erasure request.

Parameters:

Name	Type	Description	Default
`requester_hash`	`str`	SHA-256 hash of the requester's identity.	required
`scope`	`str`	"session" (all data in current session), "all" (all data across sessions), "specific_facts" (only listed fact IDs).	`'session'`
`target_ids`	`list[str] \| None`	Specific fact IDs when scope="specific_facts".	`None`

Returns:

Type	Description
`ErasureRequest`	ErasureRequest with a unique request_id.

`complete_request(request_id, items_erased=0)` ¶

Mark an erasure request as completed.

Returns True if request was found and completed.

`pending_requests()` ¶

Return all pending (incomplete) erasure requests.

`to_dict()` ¶

Export erasure state for audit/compliance reporting.

`PIIScanner` ¶

Detect PII patterns in text (§7.12.2).

IMPORTANT: This scanner is advisory. It NEVER modifies, redacts, or blocks content (Axiom 9 - output integrity). It reports findings for the user to act upon.

Usage::

scanner = PIIScanner()
result = scanner.scan("Contact john@example.com for details.")
if result.has_pii:
    print(f"Found PII: {result.pii_types_found}")

`scan(text)` ¶

Scan text for PII patterns.

Returns PIIScanResult with detections. Text of matches is NEVER stored - only SHA-256 hashes for audit trail purposes.

`RetentionManager` ¶

Manages data retention and automatic purging (§7.12.3).

EU AI Act Art. 12: Record-keeping with defined retention periods. ISO 42001 A.6.2.8: Records management with lifecycle tracking.

Usage::

rm = RetentionManager()
rm.register("fact-123", DataClassification.RESTRICTED, "user-input")
expired = rm.get_expired()  # Returns IDs ready for purging
rm.mark_purged("fact-123")

`tracked_count` `property` ¶

Return the current tracked count.

`active_count` `property` ¶

Return the current active count.

`register(data_id, classification, source_label='')` ¶

Register a data item for retention tracking.

`get_expired()` ¶

Return IDs of all expired, non-purged data items.

`mark_purged(data_id)` ¶

Mark a data item as purged. Returns True if found.

`enforce()` ¶

Run retention enforcement: return IDs of newly expired items.

The caller is responsible for actually deleting the data. This method only identifies what needs purging.

`get_record(data_id)` ¶

Get retention record for a data item.

`to_dict()` ¶

Export retention state for audit/compliance reporting.

`IngestQuarantine` ¶

1-window quarantine with confidence penalty and batch poisoning detection (§7.8).

Workflow: 1. Incoming facts go into quarantine with 0.7× confidence 2. After 1 window, cross-reference against extraction-derived facts 3. Matching facts are promoted (confidence restored) 4. Non-matching facts are rejected 5. If >30% of a batch fails, quarantine entire batch

Usage

q = IngestQuarantine() q.quarantine_facts([...], "w-1", source="user_input")

... next window processes ...¶

report = q.validate_and_promote("w-2", extraction_fact_texts)

`quarantine_count` `property` ¶

Number of facts currently in quarantine (not promoted or rejected).

`history` `property` ¶

Return the history.

`quarantine_fact(fact_id, original_confidence, window_id, source_label='', fact_text='')` ¶

Place a single fact into quarantine with 0.7× confidence penalty (§6F.1).

`quarantine_facts(facts, window_id, source_label='')` ¶

Quarantine a batch of facts.

Facts can be (fact_id, confidence) or (fact_id, confidence, text).

`get_penalized_confidence(fact_id)` ¶

Get quarantine-penalized confidence for a fact.

`is_quarantined(fact_id)` ¶

Check if a fact is in active quarantine.

`validate_and_promote(current_window_id, extraction_fact_texts, similarity_threshold=0.5)` ¶

Cross-reference validation: promote or reject quarantined facts (§6F.2).

Facts quarantined in an earlier window are validated against extraction-derived facts. Text overlap > threshold → promote.

Parameters:

Name	Type	Description	Default
`current_window_id`	`str`	Current window being processed	required
`extraction_fact_texts`	`dict[str, str]`	{fact_id: text} from extraction pipeline	required
`similarity_threshold`	`float`	Word overlap threshold for cross-reference	`0.5`

Returns:

Type	Description
`QuarantineReport`	QuarantineReport with promotion/rejection counts

`get_active_entries()` ¶

Return all currently quarantined (non-promoted, non-rejected) entries.

`clear()` ¶

Clear all quarantine state.

`QuarantineReport` `dataclass` ¶

Result of cross-reference validation pass.

`Permission` ¶

Bases: str

Named permission strings.

`RateLimitConfig` `dataclass` ¶

Rate limiting configuration (§6G.2).

`RBACEnforcer` ¶

Role-based access control + rate limiting (§7.10).

Usage

rbac = RBACEnforcer(role=Role.OPERATOR) result = rbac.check_permission("dispatch") if not result.allowed: raise PermissionError(result.reason) result = rbac.check_rate_limit("dispatch") if not result.allowed: raise RateLimitError(result.reason) rbac.record_dispatch()

`role` `property` `writable` ¶

Return the role.

`session_tokens_used` `property` ¶

Return the session tokens used.

`check_permission(permission)` ¶

Check if current role has the given permission.

`has_permission(permission)` ¶

Return True if the current role has permission.

`check_rate_limit(operation, payload_bytes=0)` ¶

Check if the operation is within rate limits.

`record_dispatch(tokens_used=0)` ¶

Record a dispatch operation for rate limiting.

`record_ingest(payload_bytes)` ¶

Record an ingest operation for rate limiting.

`record_tokens(count)` ¶

Record token consumption.

`reset_limits()` ¶

Reset all rate limit counters.

`Role` ¶

Bases: IntEnum

RBAC roles with increasing privilege (§6G.1).

`SafetyManifest` `dataclass` ¶

The one config that drives all CRP safety - code and dashboard (SPEC-033 §2).

Attributes:

Name	Type	Description
`profile`	`str`	Named profile ("balanced", "strict", "medical", "financial") or "custom" when any field deviates from a named profile.
`settings`	`dict[str, Any]`	Dict of capability-name → current value.
`checkpoints`	`list[dict[str, Any]]`	List of checkpoint declarations (inline HITL rules).
`custom_rules`	`list[str]`	List of paths to custom rule modules.

`get(name, default=None)` ¶

Read a setting by name.

Parameters:

Name	Type	Description	Default
`name`	`str`	Setting key.	required
`default`	`Any`	Value returned if the setting is absent.	`None`

Returns:

Type	Description
`Any`	Current setting value or `default`.

`set(name, value)` ¶

Write a setting - marks profile as custom if it deviates.

Parameters:

Name	Type	Description	Default
`name`	`str`	Setting key.	required
`value`	`Any`	New value to store.	required

Returns:

Type	Description
`None`	None. Updates `profile` to `"custom"` when the value differs
`None`	from the named profile defaults.

`compute_hash()` ¶

Return a deterministic hash of this manifest for CRP-Config-Hash.

Returns:

Type	Description
`str`	Hex-encoded BLAKE2b digest of the canonical settings JSON.

Raises:

Type	Description
`ValueError`	If the settings cannot be serialised (rare).

`to_dict()` ¶

Serialise the manifest to a JSON-safe dict.

Returns:

Type	Description
`dict[str, Any]`	Dict with `profile`, `settings`, `checkpoints`, and `custom_rules`.

`from_dict(data)` `classmethod` ¶

Restore a SafetyManifest from a serialised dict.

Parameters:

Name	Type	Description	Default
`data`	`dict[str, Any]`	Dict produced by `to_dict`.	required

Returns:

Type	Description
`SafetyManifest`	Reconstructed `SafetyManifest` instance.

`SessionTokenPayload` `dataclass` ¶

Decoded CRP session-token payload (SPEC-007 §2.2).

`to_json_obj()` ¶

Serialize the payload to a JSON-serializable dict.

`from_json_obj(obj)` `classmethod` ¶

Create a new instance from a JSON-compatible object.

Parameters:

Name	Type	Description	Default
`obj`	`dict[str, object]`	The obj value.	required

Returns:

Type	Description
`SessionTokenPayload`	`SessionTokenPayload`.

`TokenStatus` ¶

Bases: str, Enum

Validation outcomes (SPEC-007 §5).

`TokenValidation` `dataclass` ¶

Result of :func:validate_token.

`ok` `property` ¶

Return whether the ok condition holds.

`http_status` `property` ¶

Return the HTTP status.

`IndicatorOfCompromise` `dataclass` ¶

One observed behavioral anomaly.

`to_dict()` ¶

Serialize IoC to a dict.

`TrustAction` ¶

Bases: str

Graduated trust action recommendation.

`TrustActions` ¶

Canonical trust actions.

`TrustDecision` `dataclass` ¶

Outcome of a trust observation.

`to_dict()` ¶

Serialize decision to a dict.

`TrustMonitor` ¶

Runtime behavioral trust monitor with graduated response.

Usage::

monitor = TrustMonitor(session_id="s-1", kill_switch=ks)
decision = monitor.observe({"tool": "http", "output": "..."})

`trust_score` `property` ¶

Current trust score in [0, 1].

`observe(observation)` ¶

Evaluate an observation and update trust.

Parameters:

Name	Type	Description	Default
`observation`	`dict[str, Any]`	Dict describing the event. Recognised keys: `tool` (str), `output` (str), `input` (str), `action` (str), `risk_level` (str). Any unknown key is stringified and scanned for patterns.	required

Returns:

Type	Description
`TrustDecision`	A `TrustDecision` with the recommended action and detected IoCs.

`summary()` ¶

Return a JSON-safe summary of the monitor state.

`TrustMonitorConfig` `dataclass` ¶

Configuration for the trust monitor.

`InputValidator` ¶

Structural input validation - Layer 1, cannot be disabled (§7.4).

This validator ALWAYS runs on all input. It cannot be turned off. It performs structural sanitization without modifying semantic content.

Usage

validator = InputValidator() result = validator.validate("input text") if result.valid: use(result.sanitized_text)

`validate(text, mime_type=None, metadata=None)` ¶

Validate and sanitize input text (§7.4).

Steps: 1. Size check (50 MB limit) 2. Unicode NFC normalization 3. Null byte stripping 4. Control character stripping (keep \n, \t, \r) 5. MIME type validation (if provided) 6. Metadata key count check (≤ 50)

`validate_metadata(metadata)` ¶

Validate and truncate metadata keys (§6D.4).

Returns (sanitized_metadata, warnings).

`ValidationResult` `dataclass` ¶

Result of input validation.

`resolve_clarification(request, handler)` ¶

Resolve a clarification request with a graceful fallback (Invariant 10).

Never raises to the caller: a missing or failing handler degrades to SKIP so the agent continues best-effort rather than crashing the end user's session.

`get_default_control_plane()` ¶

Return the singleton default control plane (lazy initialised).

Returns:

Type	Description
`SafetyControlPlane`	The module-level default `SafetyControlPlane` instance, creating it
`SafetyControlPlane`	on first call.

`compute_fact_hash(text)` ¶

Compute integrity hash for a fact's text (§6B.2).

Uses BLAKE3 (~1μs) when available, SHA-256 fallback.

`build_token(payload, master_key)` ¶

Serialise + sign payload, returning payload_b64.signature_b64.

Raises:

Type	Description
`ValueError`	if the encoded payload exceeds 4096 bytes (SPEC-007 §2.4).

`derive_signing_key(master_key, session_id)` ¶

Derive the per-session 256-bit signing key (SPEC-007 §3.1).

`format_set_session_header(token, payload, *, max_age=None)` ¶

Render the CRP-Set-Session header value (SPEC-007 §2.1).

`issue_token(*, session_id, master_key, window=1, quality_history=None, safety_budget=1.0, chain_tip='', continuation_id='', dag_pattern='LINEAR', strategy='', policy_hash='', ckf_hash='', scope='', nonce='', key_version=None, lifetime=DEFAULT_TOKEN_LIFETIME, now=None)` ¶

Convenience: build a fresh payload with iat/exp and sign it.

`parse_token(token)` ¶

Split + decode a token's payload without verifying its signature.

`validate_token(token, *, master_keys, expected_chain_tip=None, expected_scope=None, presented_policy_hash=None, policy_is_more_restrictive=None, presented_nonce=None, now=None)` ¶

Full session-token validation (SPEC-007 §5).

Parameters:

Name	Type	Description	Default
`token`	`str`	the raw `CRP-Session-Token` value.	required
`master_keys`	`bytes \| list[bytes]`	the gateway master key(s). A list allows trying both the current and previous key during master-key rotation (SPEC-007 §6).	required
`expected_chain_tip`	`str \| None`	gateway's recorded HMAC chain tip for the session. Mismatch → 409 (stale/replayed token).	`None`
`expected_scope`	`str \| None`	API-key prefix from the request `Authorization`. Mismatch → 403.	`None`
`presented_policy_hash`	`str \| None`	hash of a `CRP-Safety-Policy` presented with the request. If it differs from the token's `pol` it MUST be more restrictive (set policy_is_more_restrictive) else → 403.	`None`
`policy_is_more_restrictive`	`bool \| None`	caller-computed tightening check.	`None`
`presented_nonce`	`str \| None`	`CRP-Safety-Nonce` from the request. Mismatch → 400.	`None`
`now`	`int \| None`	override current epoch seconds (for testing).	`None`

Returns:

Type	Description
`TokenValidation`	class:`TokenValidation`.

`security.audit_anchor`¶

`crp.security.audit_anchor` ¶

External anchoring for the compliance audit trail (§7.14, SPEC-011).

Anchoring commits a trail to a single Merkle root (see crp.security.audit_merkle) and publishes that commitment to an external medium, so later tampering with the local trail is detectable by anyone holding the anchor. The flow mirrors RFC 3161 timestamping:

compute the Merkle root over all trail entry_hash values,
build an :class:AnchorRecord (root, entry count, ISO-8601 timestamp, method) - the "timestamp token",
sign the canonical record bytes with an Ed25519 :class:ActionSigner when one is available (unsigned in zero-dependency mode),
persist via a pluggable :class:AnchorTransport.

LocalFileTransport is the v1 anchor transport; a TSA HTTP transport (e.g. an RFC 3161 Time-Stamp Authority client) can slot in by implementing the same two-method interface without touching the anchor/verify logic.

`AnchorRecord` `dataclass` ¶

A signed, timestamped commitment to a trail's Merkle root.

Attributes:

Name	Type	Description
`root_hash`	`str`	Merkle root over the trail's entry hashes (hex).
`entry_count`	`int`	Number of trail entries covered by the root.
`anchored_at`	`str`	ISO-8601 UTC timestamp of anchoring.
`method`	`str`	Anchor transport method tag (e.g. `"local-file"`).
`version`	`str`	Anchor record schema version.
`signature`	`str`	Hex Ed25519 signature over :meth:`tbs_bytes` ("" when unsigned).
`signer_public_key`	`str`	PEM public key for signature verification ("" when unsigned).

`tbs_bytes()` ¶

Canonical to-be-signed bytes (RFC-3161-style: fixed field set, sorted keys, compact separators - signature fields excluded).

`to_dict()` ¶

Serialise the record to a JSON-safe dict.

`from_dict(data)` `classmethod` ¶

Deserialise a record; raises ValueError on missing fields.

`AnchorTransport` ¶

Bases: ABC

Transport that persists and retrieves anchor records.

Implementations: :class:LocalFileTransport (v1). A TSA HTTP client (RFC 3161) or transparency-log submitter slots in here later.

`write(record)` `abstractmethod` ¶

Persist record to the external medium.

`read()` `abstractmethod` ¶

Retrieve the previously persisted record.

`LocalFileTransport` `dataclass` ¶

Bases: AnchorTransport

v1 anchor transport: a JSON file on local (or mounted) storage.

`write(record)` ¶

Write the record as pretty-printed JSON (atomic-ish via direct write).

`read()` ¶

Read and parse the record file; raises ValueError if malformed.

`anchor_hashes(leaf_hashes, transport, signer=None)` ¶

Anchor an arbitrary sequence of 64-char hex leaf hashes.

Generic form of :func:anchor_trail - works over any hash-linked ledger (compliance trail, manifest ledger, …) without requiring a :class:ComplianceAuditTrail instance. transport may be an :class:AnchorTransport or a filesystem path (wrapped in :class:LocalFileTransport).

`verify_hashes_anchor(leaf_hashes, anchor_path, transport=None)` ¶

Verify an arbitrary leaf-hash sequence against a previously written anchor.

Same contract as :func:verify_anchor but for any hash-linked ledger: recomputes the Merkle root over leaf_hashes and compares root + count (and the Ed25519 signature when present). Tail truncation or extension of the sequence fails verification. Fails closed on any problem.

`anchor_trail(trail, transport, signer=None)` ¶

Anchor trail via transport; returns the persisted record.

Parameters:

Name	Type	Description	Default
`trail`	`ComplianceAuditTrail`	The compliance audit trail to commit to.	required
`transport`	`AnchorTransport`	Where the anchor record goes.	required
`signer`	`ActionSigner \| None`	Ed25519 signer for the record. When omitted, one is created (`CRP_AUDIT_SIGNING_KEY` env or ephemeral) if `cryptography` is installed; in zero-dependency mode the record is unsigned.	`None`

`anchor_to_file(trail, path, signer=None)` ¶

Convenience wrapper: anchor trail to a local JSON file at path.

`verify_audit_file(anchor_path, audit_path)` ¶

Verify a persisted <sid>.audit.jsonl file against its anchor (U-C9).

Used on session resume, where the live trail is a fresh empty chain and cannot be anchored-compared directly: recomputes the Merkle root over the file's per-line entry_hash values and compares it (and the entry count) to the anchored record. When the record is signed, the Ed25519 signature must also verify - an unverifiable signed record fails closed.

Returns False on any mismatch, tampering, or unreadable input.

`verify_anchor(trail, anchor_path, transport=None)` ¶

Verify trail against a previously written anchor.

Recomputes the Merkle root over the current trail and compares it (and the entry count) to the anchored record. When the record is signed, the Ed25519 signature must also verify - a signed record whose signature cannot be checked (malformed, wrong key, or cryptography unavailable) fails closed.

Returns False on any mismatch, tampering, or unreadable anchor.

`security.audit_merkle`¶

`crp.security.audit_merkle` ¶

Merkle inclusion proofs for the compliance audit trail (§7.14, SPEC-011).

A :class:MerkleTree is built over the SHA-256 entry_hash values of a :class:~crp.security.audit_trail.ComplianceAuditTrail, giving:

a single root() commitment over the whole trail (what gets externally anchored - see crp.security.audit_anchor), and
O(log n) inclusion_proof(index) / verify_inclusion receipts that let an auditor confirm one entry is part of an anchored root without seeing the rest of the trail.

Hashing scheme (stdlib hashlib only, deterministic):

Leaves are the entry hashes themselves (64-char lowercase hex of SHA-256).
Internal node = sha256(bytes.fromhex(left) + bytes.fromhex(right)).
An odd node at any level is promoted unchanged to the next level (never duplicated), so proof sides are unambiguous.
The empty tree has root "" and no valid proofs.

Domain separation (§6.4.2, opt-in): pass domain=b"my-app" to hash leaves as sha256(domain + b"\x00" + leaf) and internal nodes as sha256(domain + b"\x01" + left + right). This closes the second-preimage gap between leaf pre-images and internal nodes; verifiers must pass the same domain. The default (None) preserves the original scheme for backward compatibility with already-anchored roots.

Consistency proofs (§6.4.2): :meth:MerkleTree.consistency_proof proves that a previously anchored root over m leaves is a prefix commitment of the current tree over n leaves - i.e. the log only ever appended. This is what makes tail truncation (delete the last N entries and present the remainder) detectable between two successive anchors; a verifier calls :meth:MerkleTree.verify_consistency with the old root/count, the new root/count, and the proof.

Proof encoding: a proof is a list of "<side>:<sibling_hex>" strings, ordered leaf-to-root, where side is "left" or "right" - the position of the sibling relative to the running hash. The encoding is self-contained: :func:MerkleTree.verify_inclusion needs no index or leaf count, which makes proofs safe to ship as standalone receipts. Consistency proofs are plain sibling-hash lists consumed from the end.

`MerkleTree` ¶

Binary Merkle tree over hex-encoded SHA-256 leaf hashes.

Parameters:

Name	Type	Description	Default
`leaf_hashes`	`list[str]`	Leaf digests (64-char hex, e.g. audit `entry_hash` values) in trail order.	required
`domain`	`bytes \| None`	Optional domain-separation tag (§6.4.2). When set, leaves and internal nodes are re-hashed under the domain; verifiers must pass the same tag. Default preserves the legacy scheme.	`None`

Raises:

Type	Description
`ValueError`	If any leaf is not a 64-char hex string.

`leaf_count` `property` ¶

Number of leaves the tree was built from.

`root()` ¶

Merkle root hex digest; "" for an empty tree.

`inclusion_proof(index)` ¶

Return the sibling-hash proof for the leaf at index.

The proof is ordered leaf-to-root; each element is "<side>:<sibling_hex>" where side is the sibling's position relative to the running hash.

Raises:

Type	Description
`IndexError`	If `index` is out of range or the tree is empty.

`verify_inclusion(leaf_hash, proof, root, *, domain=None)` `staticmethod` ¶

Verify that leaf_hash is included under root given proof.

Parameters:

Name	Type	Description	Default
`leaf_hash`	`str`	The claimed leaf digest (64-char hex).	required
`proof`	`list[str]`	Sibling list as produced by :meth:`inclusion_proof`.	required
`root`	`str`	The Merkle root to check against.	required
`domain`	`bytes \| None`	Domain-separation tag used to build the tree (§6.4.2); must match the tree's tag.	`None`

Returns:

Type	Description
`bool`	True iff replaying the proof from the leaf reproduces `root`.
`bool`	Any malformed element, bad hex, or mismatch returns False.

`consistency_proof(old_count)` ¶

Proof that the first old_count leaves commit to a tree whose root an auditor already holds - i.e. this tree is a pure extension of the anchored one (append-only evolution, §6.4.2, RFC 6962 §2.1.2).

Combined with :meth:verify_consistency, this detects tail truncation: an anchor at old_count leaves plus the current root at n leaves can only be bridged by a valid proof if every entry up to the anchor is still present, untouched, in order.

The proof is a flat list of subtree-root hashes (the RFC 6962 PROOF(m, D[n])); :meth:verify_consistency consumes it from the END (each recursion level appends its commitment last).

Raises:

Type	Description
`IndexError`	If `old_count` is out of range.

`verify_consistency(old_root, old_count, new_root, new_count, proof, *, domain=None)` `staticmethod` ¶

Verify that old_root commits to the first old_count leaves of the tree whose root is new_root over new_count leaves.

Replays proof (consumed from the end, as produced by :meth:consistency_proof), recomputing both the old portion's root and the full root. Verification succeeds only when every proof element is consumed, the recomputed full root equals new_root, AND the recomputed old-portion root equals old_root - so a truncated, extended, reordered, or substituted chain fails closed.

`security.audit_trail`¶

`crp.security.audit_trail` ¶

Tamper-evident compliance audit trail - HMAC-signed, immutable (§7.14).

Upgrades the observability audit log (§8.9) with: - HMAC-SHA256 chained signatures (tamper-evident) - Compliance-specific event types (data access, consent, erasure) - Structured export for regulatory review - Retention management for audit records - ISO 42001 / EU AI Act Article 12 compliant record-keeping

EU AI Act: Art. 12 (record-keeping), Art. 13 (transparency) ISO 42001: A.6.2.8 (records management), 9.1 (performance monitoring)

`ActionSigner` ¶

Ed25519 signer for per-entry audit receipts (audit_v2).

Each audit entry's entry_hash is signed, producing a non-repudiable receipt verifiable against the signer's public key alone - unlike the symmetric HMAC chain, a verifier needs no shared secret.

Key resolution order

explicit private_key_path argument, then
CRP_AUDIT_SIGNING_KEY env var (path to a PEM private key), then
an ephemeral keypair generated for the process lifetime.

Parameters:

Name	Type	Description	Default
`private_key_path`	`str \| None`	Path to a PEM-encoded Ed25519 private key.	`None`

Raises:

Type	Description
`ImportError`	If `cryptography` is not installed.

`public_key_pem()` ¶

PEM-encoded Ed25519 public key (safe to distribute to verifiers).

`sign(entry_hash)` ¶

Sign an audit entry hash; returns the Ed25519 signature as hex.

`verify(public_key_pem, entry_hash, signature_hex)` `staticmethod` ¶

Verify a receipt against a PEM public key.

Returns False (never raises) on any malformed input, wrong key, or invalid signature.

Raises:

Type	Description
`ImportError`	If `cryptography` is not installed.

`ComplianceEventType` ¶

Bases: str, Enum

Compliance-grade event types beyond standard CRP events (§7.14.1).

`AuditEntry` `dataclass` ¶

Single tamper-evident audit log entry (§7.14.2).

Each entry includes an HMAC-SHA256 signature chained from the previous entry, creating a tamper-evident log. Modification of any entry invalidates all subsequent signatures.

`to_dict()` ¶

Serialise the audit entry to a JSON-safe dict.

Returns:

Type	Description
`dict[str, Any]`	Dict representation including tamper-evidence fields.

`ComplianceAuditTrail` ¶

Tamper-evident, HMAC-signed compliance audit trail (§7.14).

Provides an immutable, append-only log with cryptographic chaining that detects any modification to historical entries.

EU AI Act Art. 12: "High-risk AI systems shall technically allow for the automatic recording of events (logs) over the lifetime of the system."

ISO 42001 A.6.2.8: Organizations must maintain records of AI system development, deployment, operation, monitoring, and decommissioning.

Usage::

trail = ComplianceAuditTrail(signing_key=session_key)
trail.record(
    ComplianceEventType.DATA_INGESTED,
    session_id="abc-123",
    data={"source": "user-input", "size_bytes": 4096},
)
# Verify chain integrity
valid, broken_at = trail.verify_chain()
assert valid
# Export for regulatory review
export = trail.export()

`durable_path` `property` ¶

Path of the durable append log, when enabled.

`transcript_store` `property` ¶

The attached transcript store, or None when capture is disabled.

`entry_count` `property` ¶

Number of entries currently stored in the trail.

`enable_durable_append(path, fsync_every=1)` ¶

Enable crash-durable per-event JSONL appending (audit U-C9).

Every subsequently recorded entry is appended (and, per policy, fsync'd) so a process crash cannot lose the trail recorded so far. Idempotent: enabling twice keeps the first writer.

`flush_durable()` ¶

Flush + fsync the durable writer (clean close path).

`close_durable()` ¶

Flush, close and detach the durable writer (clean close path).

Must run BEFORE any code rewrites durable_path atomically: on Windows an open append handle on the target makes os.replace fail with WinError 5 (access denied), and any record() after this point would write to a closed file. After close, :meth:record skips durable appending (in-memory chain is authoritative for the final export).

`attach_transcript_store(store)` ¶

Attach a :class:crp.security.transcript_store.TranscriptStore.

Enables :meth:verify_transcripts and transcript-aware exports on this trail. Called by the dispatch router the first time a transcript is captured for a session.

`record(event_type, session_id='', data=None)` ¶

Append a tamper-evident entry to the audit trail.

Thread-safe. Each entry is chained to the previous via HMAC.

Parameters:

Name	Type	Description	Default
`event_type`	`ComplianceEventType \| str`	Compliance event type enum or custom string.	required
`session_id`	`str`	Session identifier. Falls back to the trail default.	`''`
`data`	`dict[str, Any] \| None`	Arbitrary event payload.	`None`

Returns:

Type	Description
`AuditEntry`	The newly created `AuditEntry`.

`verify_chain(verify_receipts=False, public_key_pem=None)` ¶

Verify the integrity of the entire audit trail.

Parameters:

Name	Type	Description	Default
`verify_receipts`	`bool`	Additionally verify each entry's Ed25519 `receipt` (audit_v2). Every entry must carry a valid receipt; entries without one fail verification.	`False`
`public_key_pem`	`bytes \| None`	PEM public key for receipt verification. Falls back to the trail's configured :class:`ActionSigner` public key.	`None`

Returns:

Type	Description
`bool`	`(is_valid, broken_at_sequence)`. `broken_at_sequence` is `-1`
`int`	when the chain is valid, otherwise the sequence of the first broken entry.

Raises:

Type	Description
`ValueError`	If `verify_receipts` is True but no public key is available (neither argument nor configured signer).

`verify_file(path, public_key_pem=None)` `classmethod` ¶

Offline verification of an exported audit JSONL file (A-PR-5 / I-C5).

Recomputes each line's content hash from the exported fields - so ANY edit to the file is detected, unlike the keyless recompute-the-chain check integrators used to perform (the chain is plain SHA-256 over public fields and fully re-derivable by an editor) - validates the previous_hash linkage, and, when public_key_pem is supplied, requires a valid Ed25519 receipt on every entry.

Parameters:

Name	Type	Description	Default
`path`	`str`	Path to an exported `<sid>.audit.jsonl` file.	required
`public_key_pem`	`bytes \| None`	Optional Ed25519 public key (PEM) for strict audit_v2 receipt verification of every entry.	`None`

Returns:

Type	Description
`bool`	`(is_valid, broken_at_line)` - 0-based JSONL line index of the
`int`	first problem, or -1 when the file is valid.

`verify_transcripts(*, session_id=None)` ¶

Verify stored transcripts against this trail (schema 2.1.0).

Companion to :meth:verify_chain: re-hashes stored transcript content and confirms it matches the TRANSCRIPT_RECORDED events and - for unredacted transcripts - the LLM_CALL_COMPLETED hashes already chained here. An edited transcript flags at the exact window.

Returns:

Type	Description
`Any`	class:`crp.security.transcript_store.TranscriptVerification`.

Raises:

Type	Description
`ValueError`	If no transcript store is attached (capture disabled).

`query(*, event_type=None, session_id=None, since=None, until=None)` ¶

Query audit entries with optional filters.

Parameters:

Name	Type	Description	Default
`event_type`	`str \| ComplianceEventType \| None`	Filter by event type.	`None`
`session_id`	`str \| None`	Filter by session ID.	`None`
`since`	`float \| None`	Include entries with timestamp >= this value.	`None`
`until`	`float \| None`	Include entries with timestamp <= this value.	`None`

Returns:

Type	Description
`list[AuditEntry]`	Matching `AuditEntry` objects in chronological order.

`export(*, include_signatures=True, since=None)` ¶

Export the full audit trail for regulatory review.

Parameters:

Name	Type	Description	Default
`include_signatures`	`bool`	Whether to include HMAC signatures in the export.	`True`
`since`	`float \| None`	Only export entries with timestamp >= this value.	`None`

Returns:

Type	Description
`dict[str, Any]`	Structured document suitable for compliance auditors, including
`dict[str, Any]`	chain integrity status and summary statistics.

`export_jsonl(*, since=None, include_transcripts=False)` ¶

Export as JSONL (one entry per line) for log aggregation.

Parameters:

Name	Type	Description	Default
`since`	`float \| None`	Only export entries with timestamp >= this value.	`None`
`include_transcripts`	`bool`	When True (and a transcript store is attached), append the stored window transcripts as additional `record_type="transcript"` lines so the bundle is self-contained: chain + verdicts + viewable query–response pairs (schema 2.1.0).	`False`

Returns:

Type	Description
`str`	Newline-delimited JSON string of all matching entries.

`export_ndjson(*, since=None)` ¶

Alias for :meth:export_jsonl - NDJSON is the primary format (§4.1).

Parameters:

Name	Type	Description	Default
`since`	`float \| None`	Only export entries with timestamp >= this value.	`None`

Returns:

Type	Description
`str`	Newline-delimited JSON string of all matching entries.

`export_ocsf(*, since=None, provider='', model='', include_transcripts=False)` ¶

Export events in OCSF API Activity format for SIEMs (§4.2).

Parameters:

Name	Type	Description	Default
`since`	`float \| None`	Only export entries with timestamp >= this value.	`None`
`provider`	`str`	Provider identifier for the destination endpoint field.	`''`
`model`	`str`	Model identifier for the destination endpoint field.	`''`
`include_transcripts`	`bool`	When True (and a transcript store is attached), TRANSCRIPT_RECORDED-derived records additionally carry the transcript reference and content hashes in `unmapped` (the transcript content itself stays in the attached store / :meth:`export_jsonl` bundle).	`False`

Returns:

Type	Description
`list[dict[str, Any]]`	List of OCSF `class_uid=6003` records. The HMAC and risk level
`list[dict[str, Any]]`	are carried in the `unmapped` extension.

`export_sarif(*, since=None)` ¶

Export events as a SARIF 2.1.0 log for GitHub integration (§4.3).

Parameters:

Name	Type	Description	Default
`since`	`float \| None`	Only export entries with timestamp >= this value.	`None`

Returns:

Type	Description
`dict[str, Any]`	SARIF 2.1.0 log dict. Non-INFO events become SARIF results; INFO
`dict[str, Any]`	events are emitted at `note` level. Used by CRP Scan (CRP-SPEC-013).

`atomic_write_text(path, text)` ¶

Atomically and durably write text to path (audit U-C9).

tmp-file + fsync + os.replace + directory fsync: a crash mid-write can never leave a truncated file at the destination (the previous pre-6.2.2 close-time open(path, "w") export could, and the result looked authoritative).

`validate_incident_payload(event_type, data)` ¶

Validate an INCIDENT_* event payload against its canonical shape.

Parameters:

Name	Type	Description	Default
`event_type`	`ComplianceEventType \| str`	One of the four INCIDENT_* event types (enum or value str).	required
`data`	`dict[str, Any]`	Event payload to validate.	required

Returns:

Type	Description
`list[str]`	A list of problems (empty when the payload conforms). Problems are
`list[str]`	strings of the form `"missing field: <name>"` or
`list[str]`	`"bad type: <name> (expected <t>, got <t>)"`. Non-incident event
`list[str]`	types return an empty list (no shape constraint).

`security.binding`¶

`crp.security.binding` ¶

Session binding - HMAC-SHA256 key derivation and request verification (§7.1).

Provides session-scoped cryptographic binding: - 32-byte session nonce - 256-bit HMAC-derived session key - Constant-time request signature verification - OS keyring storage with zero-config fallback

`SessionBinding` `dataclass` ¶

Cryptographic session binding parameters.

`SessionBindingManager` ¶

HMAC-SHA256 session binding with OS keyring + zero-config fallback (§7.1).

Usage

mgr = SessionBindingManager() binding = mgr.create_session("session-123") sig = mgr.sign_request(b"payload") assert mgr.verify_request_signature(b"payload", sig)

`binding` `property` ¶

Return the binding.

`session_key` `property` ¶

Return current session key (raises if no session).

`key_version` `property` ¶

Current key version number.

`create_session(session_id='')` ¶

Create a new session with fresh nonce and derived key (§6A.1, §6A.2).

session_nonce: 32 bytes of cryptographic randomness
session_key: HMAC-SHA256(master_secret, nonce || session_id)

`sign_request(payload)` ¶

Compute HMAC-SHA256 signature over payload (§6A.3).

`verify_request_signature(payload, signature)` ¶

Constant-time HMAC verification (§6A.3).

Uses hmac.compare_digest to prevent timing attacks. Also checks previous key versions during rotation window.

`rotate_secret()` ¶

Rotate master secret with graceful rollover (§audit H2).

Generates a fresh 256-bit master secret, re-derives the session key, and keeps the previous key(s) valid for verification during the rotation window.

Returns:

Type	Description
`SessionBinding`	Updated SessionBinding with the new key material.

Raises:

Type	Description
`RuntimeError`	If no active session exists.

`store_to_keyring(service_name='crp-sdk')` ¶

Store master secret in OS keyring (DPAPI/Keychain/kernel).

Returns True if stored, False if keyring unavailable.

`from_keyring(service_name='crp-sdk')` `classmethod` ¶

Load master secret from OS keyring. Returns None if unavailable.

`security.checkpoint`¶

`crp.security.checkpoint` ¶

Checkpoint primitive - inline human-in-the-loop for CRP (SPEC-033 §3, SPEC-034).

A Checkpoint is like a breakpoint for human judgment. Declare anywhere:

@client.checkpoint(when="risk >= HIGH")
def generate_medical_advice(prompt):
    ...

Or inline:

result = client.complete(prompt)
await client.checkpoint("approve this output before sending to user")

On timeout or reject, the Checkpoint NEVER leaves the end user with a raw error. It always provides a graceful fallback (Invariant 10).

`CheckpointTrigger` ¶

Bases: str, Enum

What causes a checkpoint to fire (SPEC-033 §3).

`CheckpointTimeoutAction` ¶

Bases: str, Enum

What happens when a checkpoint times out awaiting human response (SPEC-033 §3).

`CheckpointRejectAction` ¶

Bases: str, Enum

What happens when a checkpoint is rejected by the human reviewer (SPEC-033 §3).

`CheckpointResolutionAction` ¶

Bases: str, Enum

The human reviewer's decision (SPEC-033 §3).

`CheckpointResolution` `dataclass` ¶

The result of a human reviewer resolving a checkpoint (SPEC-033 §3).

Attributes:

Name	Type	Description
`action`	`CheckpointResolutionAction`	Reviewer decision (approve/reject/edit).
`reviewer`	`str`	Identifier of the reviewer.
`timestamp`	`float`	Unix timestamp of the resolution.
`edited_output`	`str`	Optional replacement text when action is EDIT.
`audit_event`	`dict[str, Any] \| None`	Optional extra fields for the audit trail.

`to_audit_dict()` ¶

Produce an audit-trail compatible event dict (SPEC-011).

Returns:

Type	Description
`dict[str, Any]`	Dict with checkpoint resolution fields ready for `ComplianceAuditTrail`.

`Checkpoint` `dataclass` ¶

Inline human-in-the-loop declaration (SPEC-033 §3).

Attributes:

Name	Type	Description
`checkpoint_id`	`str`	Unique identifier for this checkpoint instance.
`trigger`	`CheckpointTrigger`	Condition that caused the checkpoint to fire.
`timeout`	`int`	Seconds before auto-action is taken.
`on_timeout`	`CheckpointTimeoutAction`	Action if human does not respond in time.
`on_reject`	`CheckpointRejectAction`	Action if human rejects.
`context`	`dict[str, Any]`	Arbitrary context dict for the reviewer UI.

`wait_for_resolution()` `async` ¶

Block until the checkpoint is resolved by a human reviewer.

If timeout expires, auto-resolves per on_timeout.

Returns:

Type	Description
`CheckpointResolution`	A `CheckpointResolution` - never `None` (Invariant 10).

Raises:

Type	Description
`TimeoutError`	Internally caught and converted to auto-resolution.

`resolve(resolution)` ¶

Called by a human reviewer (or webhook) to resolve this checkpoint.

Parameters:

Name	Type	Description	Default
`resolution`	`CheckpointResolution`	The reviewer's decision and optional edited output.	required

Returns:

Type	Description
`None`	None. Sets the internal resolved event so awaiting tasks unblock.

`decorate(when='always', timeout=300, on_timeout='escalate', on_reject='fallback')` `classmethod` ¶

Decorator factory: @checkpoint(when="risk >= HIGH").

Parameters:

Name	Type	Description	Default
`when`	`str`	Trigger condition string.	`'always'`
`timeout`	`int`	Seconds to await human review.	`300`
`on_timeout`	`str`	Action if the checkpoint times out.	`'escalate'`
`on_reject`	`str`	Action if the reviewer rejects.	`'fallback'`

Returns:

Type	Description
`Callable`	A decorator that wraps the function with checkpoint gating.

Raises:

Type	Description
`CRPError`	If the reviewer rejects and `on_reject` is `"halt"`.

`security.clarify`¶

`crp.security.clarify` ¶

Clarification bridge - the CLARIFY operation's human-in-the-loop (CRP-SPEC-033/034 + v5).

When a positioned agent does not know something, or is about to use a gated capability, it raises a :class:ClarificationRequest to the user instead of guessing. This is the synchronous companion to the async :class:crp.security.checkpoint.Checkpoint, designed for the positioned-tool-loop.

Invariant 10 (never leave the user with a raw error): resolve_clarification ALWAYS returns a resolution - if no handler is registered, or the handler raises, it degrades gracefully to a SKIP (best-effort continuation), never an exception.

`ClarificationAction` ¶

Bases: str, Enum

The user's response to a clarification request.

`ClarificationRequest` `dataclass` ¶

A request for user input raised by a CLARIFY operation or an oversight gate.

`to_dict()` ¶

Render for streaming / audit / a reviewer UI.

`ClarificationResolution` `dataclass` ¶

The outcome of resolving a clarification request.

`approved` `property` ¶

True when the resolution is an affirmative answer (for oversight gates).

`resolve_clarification(request, handler)` ¶

Resolve a clarification request with a graceful fallback (Invariant 10).

Never raises to the caller: a missing or failing handler degrades to SKIP so the agent continues best-effort rather than crashing the end user's session.

`security.compliance`¶

`crp.security.compliance` ¶

EU AI Act + ISO 42001 compliance framework (§7.15).

Implements

AI system risk classification (EU AI Act Art. 6)
Transparency declarations (EU AI Act Art. 13)
Technical documentation generation (EU AI Act Art. 11)
Compliance status reporting (EU AI Act Art. 9, ISO 42001 9.1)
AI impact assessment (ISO 42001 A.6.2.4)
Quality management system integration (EU AI Act Art. 17)

EU AI Act: Art. 6 (classification), Art. 9-17 (high-risk requirements) ISO 42001: 4-10 (full AIMS lifecycle), A.6.2 (AI-specific controls)

`AIRiskLevel` ¶

Bases: str, Enum

EU AI Act risk classification levels (Art. 6) (§7.15.1).

`AISystemCategory` ¶

Bases: str, Enum

Categories of AI system use cases relevant to risk classification.

`RiskAssessment` `dataclass` ¶

AI system risk assessment result (§7.15.1).

EU AI Act Art. 9: Providers must establish a risk management system for the entire lifecycle of the high-risk AI system.

`to_dict()` ¶

Serialise the risk assessment to a JSON-safe dict.

Returns:

Type	Description
`dict[str, Any]`	Dict representation including risk level, category, factors,
`dict[str, Any]`	mitigations, and residual risks.

`RiskClassifier` ¶

Classify AI system risk level per EU AI Act (§7.15.1).

Helps users determine their obligation level based on how they deploy CRP within their AI system.

CRP itself is a context management tool - typically MINIMAL or LIMITED risk. However, if CRP is integrated into a high-risk AI system (e.g., employment screening, credit scoring), the overall system inherits the higher classification.

Usage::

classifier = RiskClassifier()
assessment = classifier.assess(
    category=AISystemCategory.CONTEXT_MANAGEMENT,
    intended_purpose="Managing context for a customer support chatbot",
    processes_personal_data=True,
)
print(f"Risk level: {assessment.risk_level.value}")

`assess(category=AISystemCategory.CONTEXT_MANAGEMENT, intended_purpose='', processes_personal_data=False, makes_automated_decisions=False, affects_fundamental_rights=False, safety_critical=False, profiles_individuals=False)` ¶

Perform risk assessment based on EU AI Act criteria.

Parameters:

Name	Type	Description	Default
`category`	`AISystemCategory`	AI system use-case category.	`CONTEXT_MANAGEMENT`
`intended_purpose`	`str`	Human-readable description of the system's purpose.	`''`
`processes_personal_data`	`bool`	Whether the system processes personal data.	`False`
`makes_automated_decisions`	`bool`	Whether decisions are automated.	`False`
`affects_fundamental_rights`	`bool`	Whether outputs affect fundamental rights.	`False`
`safety_critical`	`bool`	Whether the system is safety-critical.	`False`
`profiles_individuals`	`bool`	Whether individuals are profiled.	`False`

Returns:

Type	Description
`RiskAssessment`	`RiskAssessment` with level, mitigations, and residual risks.

`TransparencyDeclaration` `dataclass` ¶

Transparency declaration for AI system users (§7.15.2).

EU AI Act Art. 13: Providers must ensure that high-risk AI systems are designed and developed in such a way that their operation is sufficiently transparent to enable deployers to interpret the system's output and use it appropriately.

`to_dict()` ¶

Serialise the transparency declaration to a JSON-safe dict.

Returns:

Type	Description
`dict[str, Any]`	Dict representation suitable for disclosure dashboards or regulators.

`ComplianceControl` `dataclass` ¶

Single compliance control status.

`ComplianceReporter` ¶

Generate compliance status reports (§7.15.3).

Maps CRP's native security controls to EU AI Act articles and ISO 42001 clauses, reporting implementation status for each.

Usage::

reporter = ComplianceReporter()
report = reporter.generate_report(session_stats={...})
print(report["summary"]["compliance_score"])

`generate_report(session_stats=None, risk_assessment=None)` ¶

Generate a comprehensive compliance status report.

Parameters:

Name	Type	Description	Default
`session_stats`	`dict[str, Any] \| None`	Optional runtime/session statistics to include.	`None`
`risk_assessment`	`RiskAssessment \| None`	Optional `RiskAssessment` to attach.	`None`

Returns:

Type	Description
`dict[str, Any]`	Dict with EU AI Act and ISO 42001 control lists, implementation
`dict[str, Any]`	counts, compliance percentages, and summary score.

`generate_technical_documentation(transparency=None, risk_assessment=None)` ¶

Generate EU AI Act Art. 11 technical documentation.

Parameters:

Name	Type	Description	Default
`transparency`	`TransparencyDeclaration \| None`	Optional transparency declaration to embed.	`None`
`risk_assessment`	`RiskAssessment \| None`	Optional risk assessment to embed.	`None`

Returns:

Type	Description
`dict[str, Any]`	Structured documentation dict suitable for submission to
`dict[str, Any]`	national competent authorities.

`security.consent`¶

`crp.security.consent` ¶

Consent & data rights management - GDPR + EU AI Act transparency (§7.13).

Implements

Consent state management (opt-in, opt-out, withdrawal)
Purpose limitation (data processing tied to declared purposes)
Processing records (GDPR Article 30 / EU AI Act Art. 12)
Data portability support (export in standard format)
Human oversight controls (EU AI Act Art. 14)

EU AI Act: Art. 13 (transparency), Art. 14 (human oversight), Art. 12 (record-keeping) ISO 42001: A.6.2.3 (human oversight), A.6.2.5 (data collection), A.6.2.7 (data subject rights)

`ConsentStatus` ¶

Bases: str, Enum

Consent states (§7.13.1).

`ProcessingPurpose` ¶

Bases: str, Enum

Data processing purposes - must be declared before processing (§7.13.2).

EU AI Act Art. 10: Data governance requires clear purpose limitation.

`ConsentRecord` `dataclass` ¶

Records a consent decision (§7.13.1).

`ConsentState` `dataclass` ¶

Aggregate consent state for a session.

`is_granted(purpose)` ¶

Check if consent is currently granted for a purpose.

Parameters:

Name	Type	Description	Default
`purpose`	`ProcessingPurpose`	Processing purpose to check.	required

Returns:

Type	Description
`bool`	True if consent is granted and not expired. Required purposes
`bool`	return True by default.

`denied_purposes()` ¶

Return purposes that have been explicitly denied or withdrawn.

Returns:

Type	Description
`list[ProcessingPurpose]`	List of purposes with `DENIED` or `WITHDRAWN` status.

`to_dict()` ¶

Serialise the consent state to a JSON-safe dict.

Returns:

Type	Description
`dict[str, Any]`	Dict with session id, creation time, and purpose records.

`ConsentManager` ¶

Manage consent state for data processing purposes (§7.13.1).

EU AI Act Art. 13: Transparency requires clear communication about how data is processed and for what purposes.

Usage::

cm = ConsentManager("session-123")
cm.grant(ProcessingPurpose.ANALYTICS, reason="User opted in")
if cm.check(ProcessingPurpose.ANALYTICS):
    # Process analytics data
    ...
cm.withdraw(ProcessingPurpose.ANALYTICS, reason="User opted out")

`state` `property` ¶

Return the state.

`grant(purpose, reason='', expires_hours=0.0)` ¶

Grant consent for a processing purpose.

Parameters:

Name	Type	Description	Default
`purpose`	`ProcessingPurpose`	Processing purpose being consented to.	required
`reason`	`str`	Human-readable reason for the grant.	`''`
`expires_hours`	`float`	Optional expiry in hours; 0 means no expiry.	`0.0`

Returns:

Type	Description
`ConsentRecord`	The created `ConsentRecord`.

`deny(purpose, reason='')` ¶

Deny consent for a processing purpose.

Parameters:

Name	Type	Description	Default
`purpose`	`ProcessingPurpose`	Processing purpose being denied.	required
`reason`	`str`	Human-readable reason for the denial.	`''`

Returns:

Type	Description
`ConsentRecord`	The created `ConsentRecord`.

`withdraw(purpose, reason='')` ¶

Withdraw previously granted consent.

Parameters:

Name	Type	Description	Default
`purpose`	`ProcessingPurpose`	Processing purpose being withdrawn.	required
`reason`	`str`	Human-readable reason for the withdrawal.	`''`

Returns:

Type	Description
`ConsentRecord`	The created `ConsentRecord`.

`check(purpose)` ¶

Check if consent is granted for a purpose.

Parameters:

Name	Type	Description	Default
`purpose`	`ProcessingPurpose`	Processing purpose to check.	required

Returns:

Type	Description
`bool`	True if consent is granted. Required purposes return True by default;
`bool`	opt-in purposes return False until explicitly granted.

`check_required(purpose)` ¶

Check consent and raise if denied for a required purpose.

Parameters:

Name	Type	Description	Default
`purpose`	`ProcessingPurpose`	Processing purpose to check.	required

Returns:

Type	Description
`bool`	True if consent is granted.

Raises:

Type	Description
`ConsentRequiredError`	If the purpose is required and consent is missing.

`to_dict()` ¶

Export consent state for compliance reporting.

Returns:

Type	Description
`dict[str, Any]`	Serialised consent state dict.

`ConsentRequiredError` ¶

Bases: Exception

Raised when a required consent is missing.

Required purposes cannot be disabled without degrading core CRP functionality.

`ProcessingActivity` `dataclass` ¶

Records a single data processing activity (GDPR Art. 30) (§7.13.3).

EU AI Act Art. 12: Systems must automatically record events relevant for identifying risks and substantial modifications.

`ProcessingRecordKeeper` ¶

Maintain GDPR Article 30 processing records (§7.13.3).

Every data processing activity within a CRP session is recorded with its purpose, legal basis, data categories, and retention.

Usage::

keeper = ProcessingRecordKeeper("session-123")
keeper.record(
    purpose=ProcessingPurpose.FACT_EXTRACTION,
    data_categories=["text_input"],
    legal_basis="legitimate_interest",
    input_size_bytes=4096,
)
records = keeper.export()

`activity_count` `property` ¶

Number of recorded processing activities.

`record(purpose, data_categories, legal_basis='legitimate_interest', input_size_bytes=0, output_size_bytes=0, automated_decision=False, human_oversight=False, retention_period='session')` ¶

Record a data processing activity.

Parameters:

Name	Type	Description	Default
`purpose`	`ProcessingPurpose`	Processing purpose for the activity.	required
`data_categories`	`list[str]`	Categories of data processed (e.g. ["text_input"]).	required
`legal_basis`	`str`	GDPR legal basis string.	`'legitimate_interest'`
`input_size_bytes`	`int`	Size of input data in bytes.	`0`
`output_size_bytes`	`int`	Size of output data in bytes.	`0`
`automated_decision`	`bool`	Whether automated decision-making occurred.	`False`
`human_oversight`	`bool`	Whether human oversight was available.	`False`
`retention_period`	`str`	Retention descriptor (e.g. "session", "30d").	`'session'`

Returns:

Type	Description
`ProcessingActivity`	The created `ProcessingActivity`.

`export()` ¶

Export all processing records for regulatory review.

Returns:

Type	Description
`list[dict[str, Any]]`	List of dicts representing every recorded activity.

`summary()` ¶

Summarize processing activities for compliance dashboard.

Returns:

Type	Description
`dict[str, Any]`	Dict with totals, per-purpose counts, input/output byte sums, and
`dict[str, Any]`	oversight/decision counts.

`HumanOversightLevel` ¶

Bases: str, Enum

Levels of human oversight (EU AI Act Art. 14) (§7.13.4).

`OversightConfig` `dataclass` ¶

Human oversight configuration (§7.13.4).

EU AI Act Art. 14: High-risk AI systems must be designed to allow effective human oversight during their period of use.

`OversightEvent` `dataclass` ¶

Records a human oversight event.

`HumanOversightController` ¶

Implements human oversight controls (EU AI Act Art. 14) (§7.13.4).

ISO 42001 A.6.2.3: Organizations must establish processes for human oversight of AI systems appropriate to the risk level.

Usage::

hoc = HumanOversightController(OversightConfig(
    level=HumanOversightLevel.APPROVAL,
    require_approval_for_dispatch=True,
))

# Before dispatch:
if hoc.requires_approval("dispatch"):
    approval = hoc.request_approval("dispatch", {"task": "..."})
    # ... wait for human approval ...
    hoc.record_decision(approval.event_id, approved=True)

`config` `property` ¶

Return the config.

`level` `property` ¶

Return the level.

`requires_approval(operation)` ¶

Check if an operation requires human approval.

Parameters:

Name	Type	Description	Default
`operation`	`str`	One of "dispatch", "ingest", "export", "deletion".	required

Returns:

Type	Description
`bool`	True if the configured oversight level and per-operation flags
`bool`	require explicit human approval.

`check_autonomous_limit()` ¶

Check if the autonomous dispatch limit has been reached.

Returns:

Type	Description
`bool`	True if more autonomous dispatches are allowed (or no limit is set).

`record_autonomous_dispatch()` ¶

Record an autonomous dispatch for limit tracking.

Returns:

Type	Description
`None`	None. Increments the internal autonomous dispatch counter.

`request_approval(operation, details=None)` ¶

Create an approval request event.

Parameters:

Name	Type	Description	Default
`operation`	`str`	Operation requiring approval.	required
`details`	`dict[str, Any] \| None`	Arbitrary context for the reviewer.	`None`

Returns:

Type	Description
`OversightEvent`	The created `OversightEvent`.

`record_decision(event_id, approved, approved_by='', reason='')` ¶

Record a human oversight decision.

Parameters:

Name	Type	Description	Default
`event_id`	`str`	ID of the original approval request.	required
`approved`	`bool`	True if approved, False if denied.	required
`approved_by`	`str`	Identifier of the reviewer.	`''`
`reason`	`str`	Optional reason for the decision.	`''`

Returns:

Type	Description
`OversightEvent`	The created `OversightEvent`.

`record_halt(operation, reason, details=None)` ¶

Record a halt event (system stopped due to policy).

Parameters:

Name	Type	Description	Default
`operation`	`str`	Operation that was halted.	required
`reason`	`str`	Human-readable reason for the halt.	required
`details`	`dict[str, Any] \| None`	Additional context.	`None`

Returns:

Type	Description
`OversightEvent`	The created `OversightEvent`.

`should_halt_on_injection()` ¶

Check if processing should halt when injection is detected.

Returns:

Type	Description
`bool`	True if `halt_on_injection_detection` is enabled.

`should_halt_on_pii()` ¶

Check if processing should halt when PII is detected.

Returns:

Type	Description
`bool`	True if `halt_on_pii_detection` is enabled.

`to_dict()` ¶

Export oversight state for compliance reporting.

Returns:

Type	Description
`dict[str, Any]`	Dict summarising oversight level, autonomous dispatch counts, and
`dict[str, Any]`	approval/denial/halt event counts.

`security.control_plane`¶

`crp.security.control_plane` ¶

Safety Control Plane (SCP) - single place for all CRP safety (SPEC-033).

The SCP unifies existing scattered safety mechanisms under one catalogue

Registry: every capability, its default, its current setting, its effect
Manifest: the one config that drives code + dashboard
Checkpoints: inline human-in-the-loop declarations
Coverage Map: what CRP detects AND what it explicitly does not detect

Usage::

from crp.security.control_plane import SafetyControlPlane

scp = SafetyControlPlane()
scp.show()                       # human-readable printout
capability = scp.get_capability("require_grounding")
scp.tune("require_grounding", 0.85)
scp.register_rule(my_custom_rule)
surface = scp.get_surface_map()  # for dashboard UI rendering

`CustomSafetyRule` `dataclass` ¶

A user-defined safety rule registered as a first-class citizen (SPEC-033).

Attributes:

Name	Type	Description
`name`	`str`	Unique rule identifier.
`check_fn`	`Any`	Callable that performs the safety check. Type is kept as `Any` to avoid heavy import dependencies in the control plane module.
`description`	`str`	Human-readable explanation of what the rule detects.
`default`	`Any`	Default value for the rule when registered.

`SafetyControlPlane` `dataclass` ¶

Single place from which all CRP safety is seen, tuned, and extended (SPEC-033).

Attributes:

Name	Type	Description
`registry`	`dict[str, SafetyCapability]`	Dict of capability-name → SafetyCapability.
`manifest`	`SafetyManifest`	The SafetyManifest that drives settings.
`coverage`	`SafetyCoverageMap`	The SafetyCoverageMap (capabilities + out-of-scope).
`checkpoints`	`SafetyCoverageMap`	Active checkpoint instances awaiting resolution.

`get_capability(name)` ¶

Retrieve a capability from the registry.

Parameters:

Name	Type	Description	Default
`name`	`str`	Capability identifier.	required

Returns:

Type	Description
`SafetyCapability \| None`	The matching `SafetyCapability` or `None` if not registered.

`list_capabilities()` ¶

Return all registered capabilities.

Returns:

Type	Description
`list[SafetyCapability]`	List of every built-in, addable, and custom capability in the registry.

`tune(name, value)` ¶

Change the current value of a capability and sync to Manifest.

Parameters:

Name	Type	Description	Default
`name`	`str`	Capability identifier.	required
`value`	`Any`	New current value. Should normally be within `allowed_range`.	required

Returns:

Type	Description
`None`	None. Logs a warning if the capability is unknown.

`register_rule(rule)` ¶

Register a custom safety rule as a first-class citizen.

Parameters:

Name	Type	Description	Default
`rule`	`CustomSafetyRule`	Custom rule definition including name, check function, and default.	required

Returns:

Type	Description
`None`	None. The rule is mirrored into the registry and coverage map.

`create_checkpoint(trigger='always', timeout=300, on_timeout='escalate', on_reject='fallback', context=None)` ¶

Create and track a new human-in-the-loop checkpoint.

Parameters:

Name	Type	Description	Default
`trigger`	`str`	Condition that fires the checkpoint. One of the `CheckpointTrigger` values or a custom string.	`'always'`
`timeout`	`int`	Seconds to wait for human review before auto-resolution.	`300`
`on_timeout`	`str`	Action taken when the checkpoint times out.	`'escalate'`
`on_reject`	`str`	Action taken when the human reviewer rejects.	`'fallback'`
`context`	`dict[str, Any] \| None`	Arbitrary dict passed to the reviewer UI/webhook.	`None`

Returns:

Type	Description
`Checkpoint`	The created `Checkpoint` instance, tracked by ID internally.

`resolve_checkpoint(checkpoint_id, resolution)` ¶

Resolve a checkpoint by ID (called by reviewer or webhook).

Parameters:

Name	Type	Description	Default
`checkpoint_id`	`str`	UUID of the checkpoint returned by `create_checkpoint`.	required
`resolution`	`CheckpointResolution`	Human reviewer's decision and optional edited output.	required

Returns:

Type	Description
`None`	None. Logs an error if the checkpoint ID is unknown.

`get_surface_map()` ¶

Return the complete safety surface as a dict - for UI dashboards.

Returns:

Type	Description
`dict[str, Any]`	Dict with `registry`, `manifest`, `coverage`, and
`dict[str, Any]`	`active_checkpoints` keys.

`show()` ¶

Return a human-readable printout of the entire safety surface.

Returns:

Type	Description
`str`	Multi-line string listing every capability's current/default values,
`str`	allowed ranges, effects, and the explicit out-of-scope list.

`get_default_control_plane()` ¶

Return the singleton default control plane (lazy initialised).

Returns:

Type	Description
`SafetyControlPlane`	The module-level default `SafetyControlPlane` instance, creating it
`SafetyControlPlane`	on first call.

`security.coverage`¶

`crp.security.coverage` ¶

Safety Coverage Map - addable rules registry and explicit out-of-scope list (SPEC-034 §11).

The Coverage Map answers two questions honestly

What risks can CRP detect? (the capabilities)
What risks can CRP NOT detect? (the out-of-scope list)

Honesty is a feature - the Control Plane shows both lists so users have accurate expectations.

`AddableRule` ¶

Metadata for a safety rule that can be registered in the Coverage Map (SPEC-034).

Attributes:

Name	Type	Description
`name`		Unique rule identifier.
`description`		What the rule detects and why it matters.
`default`		Default setting (e.g. "on", "warn").
`allowed_values`		Valid settings for this rule.
`effect`		Human-readable description of the rule's effect when triggered.

`SafetyCapability` `dataclass` ¶

One entry in the Safety Registry / Coverage Map (SPEC-033, SPEC-034).

Attributes:

Name	Type	Description
`name`	`str`	Unique capability identifier.
`description`	`str`	What the capability evaluates.
`spec`	`str`	Specification that defines the capability (e.g. "005", "033").
`default`	`Any`	Factory/default value.
`current`	`Any`	Active value after tuning.
`allowed_range`	`list[Any] \| None`	Optional list or range of permitted values.
`effect`	`str`	Human-readable effect when the capability triggers.
`addable`	`bool`	True if the capability comes from SPEC-034 addable rules.

`to_dict()` ¶

Serialise to dict for dashboard/export.

Returns:

Type	Description
`dict[str, Any]`	JSON-safe dict representation of this capability.

`SafetyCoverageMap` `dataclass` ¶

The complete map of detectable risks + explicit out-of-scope list (SPEC-034).

The out-of-scope list is shown in the Control Plane too - honesty is a feature.

`register(capability)` ¶

Add or overwrite a capability in the map.

Parameters:

Name	Type	Description	Default
`capability`	`SafetyCapability`	Capability descriptor to register.	required

Returns:

Type	Description
`None`	None.

`get(name)` ¶

Lookup a capability by name.

Parameters:

Name	Type	Description	Default
`name`	`str`	Capability identifier.	required

Returns:

Type	Description
`SafetyCapability \| None`	The matching `SafetyCapability` or `None`.

`list_addable()` ¶

Return only the addable rules (SPEC-034).

Returns:

Type	Description
`list[SafetyCapability]`	List of capabilities where `addable` is True.

`to_dict()` ¶

Serialise for dashboard rendering or config export.

Returns:

Type	Description
`dict[str, Any]`	Dict with `capabilities` and `out_of_scope` keys.

`security.embedding_defense`¶

`crp.security.embedding_defense` ¶

Embedding defense - SQ8 quantization, XOR salting, no export (§7.11).

Protects stored embeddings from extraction: - SQ8 quantization: float32 → int8 (reduces precision, saves memory) - XOR salting: 4-byte salt per embedding (masks raw values) - No embedding export: export_state() exports text only

`ProtectedEmbedding` `dataclass` ¶

Embedding with SQ8 quantization and XOR salt applied.

`to_dict()` ¶

Serialize the protected embedding to a base64-encoded dict.

`from_dict(data)` `classmethod` ¶

Create a new instance from a dictionary.

Parameters:

Name	Type	Description	Default
`data`	`dict[str, Any]`	The data value.	required

Returns:

Type	Description
`ProtectedEmbedding`	`ProtectedEmbedding`.

`EmbeddingDefense` ¶

SQ8 quantization + XOR salting for embedding protection (§7.11).

Usage

defense = EmbeddingDefense() protected = defense.protect([0.1, 0.2, -0.3, ...]) recovered = defense.recover(protected)

recovered ≈ original (within quantization error)¶

export_state: embeddings are stripped¶

safe_data = defense.strip_embeddings_for_export(state_dict)

`protect(embedding, salt=None)` ¶

Apply SQ8 quantization + XOR salting (§6H.1, §6H.2).

SQ8: Maps float32 range [min, max] → int8 [-128, 127]. XOR: Applies 4-byte repeating XOR mask to quantized bytes.

`recover(protected)` ¶

Recover embedding from SQ8 + XOR protected form.

Returns approximate original values (quantization introduces error).

`strip_embeddings_for_export(state_dict)` `staticmethod` ¶

Strip all embeddings from state dict for export (§6H.3).

export_state() must export text only - no embeddings.

`security.encryption`¶

`crp.security.encryption` ¶

Encryption at rest - AES-256-GCM for cold state and event log (§7.3).

Uses HKDF key diversification so different data classes (cold_storage, event_log) use different derived keys from the same session key.

Requires the cryptography package for production use (AES-256-GCM). A fallback XOR cipher is available for dep-free environments but logs a loud warning - it provides only obfuscation, not real encryption.

`EncryptedBlob` `dataclass` ¶

Encrypted data container.

`to_dict()` ¶

Serialize the encrypted blob to a base64-encoded dict.

`from_dict(data)` `classmethod` ¶

Create a new instance from a dictionary.

Parameters:

Name	Type	Description	Default
`data`	`dict[str, str]`	The data value.	required

Returns:

Type	Description
`EncryptedBlob`	`EncryptedBlob`.

`StateEncryptor` ¶

AES-256-GCM encryption for cold state and event logs (§7.3).

Usage

enc = StateEncryptor(session_key) blob = enc.encrypt_cold_state(data_bytes) plaintext = enc.decrypt_cold_state(blob)

`encrypt_cold_state(data)` ¶

Encrypt cold state data (§6C.2).

`decrypt_cold_state(blob)` ¶

Decrypt cold state data (§6C.2).

`encrypt_event_log(data)` ¶

Encrypt event log data (§6C.3).

`decrypt_event_log(blob)` ¶

Decrypt event log data (§6C.3).

`security.injection`¶

`crp.security.injection` ¶

Injection detection - Layer 2, advisory only, NEVER blocks (§7.5).

Detects prompt injection patterns and reports them as advisory flags. CRITICAL: This detector NEVER modifies input and NEVER blocks processing. It only sets security_flags on the quality report for upstream inspection.

Detection layers (ensembled): 1. Regex pattern library - 21 compiled patterns across 6 categories 2. ML classifier (optional) - CRP safety DeBERTa (crp-safety-deberta-v1, CRPv6 Phase A), prompt-injection-detector (TF-IDF + LR), or ProtectAI DeBERTa v2 (transformer-based). Zero-config: auto-detected.

`InjectionType` ¶

Bases: str, Enum

Categories of detected injection patterns.

`InjectionFlag` `dataclass` ¶

Advisory flag for a detected injection pattern.

`InjectionReport` `dataclass` ¶

Result of injection detection - advisory only.

`has_flags` `property` ¶

Return whether this object has flags.

`highest_confidence` `property` ¶

Return the highest confidence.

`security_flags` `property` ¶

Return flag strings for QualityReport.security_flags (§6E.3).

`InjectionDetector` ¶

Advisory injection detection - NEVER blocks, only reports (§7.5).

CRITICAL DESIGN CONSTRAINT: This detector NEVER modifies input text and NEVER prevents processing. It only produces advisory flags that are reported to QualityReport.security_flags.

Detection layers (ensembled automatically): Layer 1: Regex pattern library (always active) Layer 2: ML classifier (auto-detected, optional) - CRP safety DeBERTa: crp-safety-deberta-v1 safe/unsafe (CRPv6 Phase A) - prompt-injection-detector: TF-IDF + Logistic Regression (~1MB, MIT) - ProtectAI DeBERTa v2: ONNX transformer (~350MB, Apache 2.0)

Usage

detector = InjectionDetector() report = detector.scan("ignore all previous instructions") if report.has_flags: quality_report.security_flags = report.security_flags

`ml_enabled` `property` ¶

Whether ML-based injection detection is available.

`ml_backend` `property` ¶

Name of the ML backend in use, or 'none'.

`scan(text)` ¶

Scan text for injection patterns.

Returns advisory report - NEVER blocks (§6E.1). Ensembles regex patterns with ML classifier when available.

`security.integrity`¶

`crp.security.integrity` ¶

Fact integrity - BLAKE3/SHA-256 hash chain + HMAC verification (§7.2, §7.7).

Provides a tamper-evident chain of fact hashes: - Per-fact hash: BLAKE3 (~1μs) with SHA-256 fallback - Hash chain: ordered sequence of fact hashes - Chain signature: HMAC(session_key, hash_N ‖ ... ‖ hash_0) - Spot-check verification: 10% sample on cold load - Full verification on envelope construction

`ChainEntry` `dataclass` ¶

Single entry in the fact integrity chain.

`FactIntegrityChain` ¶

Tamper-evident chain of fact hashes (§7.7).

Maintains an ordered hash chain. Chain signature is computed via HMAC(session_key, hash_N ‖ ... ‖ hash_0).

Usage

chain = FactIntegrityChain(session_key) chain.add_fact("f1", "capital of France is Paris") sig = chain.chain_signature() assert chain.verify_chain(sig)

`size` `property` ¶

Return the current size count.

`add_fact(fact_id, text)` ¶

Hash a fact and append to the chain.

`get_hash(fact_id)` ¶

Get stored hash for a fact by ID.

`verify_fact(fact_id, text)` ¶

Verify a single fact's hash matches the chain.

`chain_signature()` ¶

Compute HMAC(session_key, hash_N ‖ ... ‖ hash_0) (§6B.3).

Hash chain is concatenated in reverse order (newest first).

`verify_chain(expected_signature)` ¶

Verify the full chain signature matches.

`verify_spot_check(fact_texts, sample_ratio=0.1)` ¶

Spot-check 10% of facts on cold load (§6B.4).

Parameters:

Name	Type	Description	Default
`fact_texts`	`dict[str, str]`	{fact_id: current_text} for verification	required
`sample_ratio`	`float`	fraction of chain to sample (default 10%)	`0.1`

Returns:

Type	Description
`tuple[int, int, list[str]]`	(checked, failures, failed_ids)

`verify_for_envelope(fact_ids, fact_texts)` ¶

Verify all facts before including in envelope (§6B.5).

Returns (all_valid, failed_ids).

`to_dict()` ¶

Serialize the integrity chain entries to a dict.

`export_for_verification()` ¶

Export chain data for external verification (§audit M13).

Returns a dict containing all chain entries and (if session key is set) the chain signature, suitable for independent audit.

`verify_external(export_data, fact_texts)` `staticmethod` ¶

Verify an exported chain against provided fact texts (§audit M13).

This is a static method usable without access to the session key - it re-hashes each fact and compares to the stored hash.

Returns (all_valid, failed_fact_ids).

`from_dict(data, session_key=None)` `classmethod` ¶

Create a new instance from a dictionary.

Parameters:

Name	Type	Description	Default
`data`	`dict[str, Any]`	The data value.	required
`session_key`	`bytes \| None`	The session key value.	`None`

Returns:

Type	Description
`FactIntegrityChain`	`FactIntegrityChain`.

`compute_fact_hash(text)` ¶

Compute integrity hash for a fact's text (§6B.2).

Uses BLAKE3 (~1μs) when available, SHA-256 fallback.

`security.kill_switch`¶

`crp.security.kill_switch` ¶

Emergency kill-switch + forensic incident recorder (SPEC-033 §3.4, SPEC-012 §5).

A kill-switch is an auditable emergency stop. Once fired, the session it guards is terminated, no further agent actions are dispatched, and a tamper-evident incident record is produced for post-mortem analysis. It is the last line of defense in the graduated safety response (trust decay → constrain → gate → kill).

`KillSwitchState` ¶

Bases: str, Enum

Kill-switch lifecycle states.

`KillSwitchReason` ¶

Bases: str, Enum

Canonical reasons a kill-switch can fire.

`KillIncident` `dataclass` ¶

Forensic record of a kill-switch fire event.

`to_dict()` ¶

Serialize incident to a JSON-safe dict.

`KillSwitch` ¶

Auditable emergency stop for a CRP session or agent.

Attributes:

Name	Type	Description
`state`	`KillSwitchState`	Current `KillSwitchState`.
`audit_callback`	`KillSwitchState`	Optional callable `fn(incident: KillIncident)` that receives every fired incident (e.g., to append to an audit trail).
`on_fire`	`KillSwitchState`	Optional callable `fn(incident: KillIncident)` invoked when the switch fires; useful to terminate the caller's runtime loop.

`state` `property` ¶

Current state of the kill-switch.

`is_fired` `property` ¶

True once the switch has fired.

`incidents` `property` ¶

All incidents recorded by this switch (typically one after firing).

`arm()` ¶

Re-arm a disarmed switch. Cannot re-arm a fired switch.

`disarm()` ¶

Disarm the switch - it will not fire until re-armed.

`fire(session_id, reason, triggered_by, snapshot=None)` ¶

Fire the kill-switch and record an incident.

Parameters:

Name	Type	Description	Default
`session_id`	`str`	Identifier of the guarded session.	required
`reason`	`str`	Canonical or human-readable reason for firing.	required
`triggered_by`	`str`	Identifier of the subsystem / user that triggered it.	required
`snapshot`	`dict[str, Any] \| None`	Arbitrary serialisable context captured at the moment of firing (e.g., last window id, safety budget, trust score).	`None`

Returns:

Type	Description
`KillIncident`	The recorded `KillIncident`. If the switch was already fired or
`KillIncident`	disarmed, a synthetic incident is returned without side effects.

`latest_incident()` ¶

Return the most recent incident, or None if the switch never fired.

`security.privacy`¶

`crp.security.privacy` ¶

Privacy controls - data classification, PII detection, retention, erasure (§7.12).

Implements

Data sensitivity classification (PUBLIC → CRITICAL)
PII detection with configurable patterns
Data minimization enforcement
Retention policies with automatic expiry
Right to erasure (GDPR Article 17 / EU AI Act transparency)
Data lineage tracking

EU AI Act: Art. 10 (data governance), Art. 12 (record-keeping), Art. 13 (transparency) ISO 42001: A.6.2.4 (impact assessment), A.6.2.6 (data management), A.6.2.7 (data subject rights)

`DataClassification` ¶

Bases: IntEnum

Data sensitivity classification (§7.12.1).

Higher values = more sensitive. Controls encryption, retention, access requirements, and audit verbosity.

`PIIDetection` `dataclass` ¶

Result of PII scan on a piece of text.

`PIIScanResult` `dataclass` ¶

Aggregate result of PII scanning.

`has_pii` `property` ¶

Return whether this object has PII.

`pii_types_found` `property` ¶

Return the PII types found.

`highest_classification` `property` ¶

Return the highest data classification implied by detected PII.

`to_dict()` ¶

Serialize the PII scan result to a dict.

`PIIScanner` ¶

Detect PII patterns in text (§7.12.2).

IMPORTANT: This scanner is advisory. It NEVER modifies, redacts, or blocks content (Axiom 9 - output integrity). It reports findings for the user to act upon.

Usage::

scanner = PIIScanner()
result = scanner.scan("Contact john@example.com for details.")
if result.has_pii:
    print(f"Found PII: {result.pii_types_found}")

`scan(text)` ¶

Scan text for PII patterns.

Returns PIIScanResult with detections. Text of matches is NEVER stored - only SHA-256 hashes for audit trail purposes.

`RetentionPolicy` `dataclass` ¶

Data retention policy configuration (§7.12.3).

Defines how long different classifications of data are retained and what happens at expiry.

`RetentionRecord` `dataclass` ¶

Tracks retention status for a piece of data.

`RetentionManager` ¶

Manages data retention and automatic purging (§7.12.3).

EU AI Act Art. 12: Record-keeping with defined retention periods. ISO 42001 A.6.2.8: Records management with lifecycle tracking.

Usage::

rm = RetentionManager()
rm.register("fact-123", DataClassification.RESTRICTED, "user-input")
expired = rm.get_expired()  # Returns IDs ready for purging
rm.mark_purged("fact-123")

`tracked_count` `property` ¶

Return the current tracked count.

`active_count` `property` ¶

Return the current active count.

`register(data_id, classification, source_label='')` ¶

Register a data item for retention tracking.

`get_expired()` ¶

Return IDs of all expired, non-purged data items.

`mark_purged(data_id)` ¶

Mark a data item as purged. Returns True if found.

`enforce()` ¶

Run retention enforcement: return IDs of newly expired items.

The caller is responsible for actually deleting the data. This method only identifies what needs purging.

`get_record(data_id)` ¶

Get retention record for a data item.

`to_dict()` ¶

Export retention state for audit/compliance reporting.

`ErasureRequest` `dataclass` ¶

Tracks a right-to-erasure (data deletion) request.

`ErasureManager` ¶

Handles right-to-erasure requests (GDPR Article 17) (§7.12.4).

Tracks erasure requests, ensures they are fulfilled, and maintains an audit trail of what was deleted and when.

Usage::

em = ErasureManager()
req = em.create_request("user-hash-abc", scope="session")
# ... caller erases the actual data ...
em.complete_request(req.request_id, items_erased=42)

`create_request(requester_hash, scope='session', target_ids=None)` ¶

Create a new erasure request.

Parameters:

Name	Type	Description	Default
`requester_hash`	`str`	SHA-256 hash of the requester's identity.	required
`scope`	`str`	"session" (all data in current session), "all" (all data across sessions), "specific_facts" (only listed fact IDs).	`'session'`
`target_ids`	`list[str] \| None`	Specific fact IDs when scope="specific_facts".	`None`

Returns:

Type	Description
`ErasureRequest`	ErasureRequest with a unique request_id.

`complete_request(request_id, items_erased=0)` ¶

Mark an erasure request as completed.

Returns True if request was found and completed.

`pending_requests()` ¶

Return all pending (incomplete) erasure requests.

`to_dict()` ¶

Export erasure state for audit/compliance reporting.

`DataLineageEntry` `dataclass` ¶

Tracks the origin and transformations of a piece of data.

`DataLineageTracker` ¶

Track data provenance and transformation history (§7.12.5).

ISO 42001 A.6.2.6: Data management requires tracking data origin, quality, and transformations throughout the AI lifecycle.

Usage::

tracker = DataLineageTracker()
tracker.record("fact-1", "extraction", "user-input", DataClassification.INTERNAL)
tracker.add_transformation("fact-1", "quarantine_promoted")

`record(data_id, origin, source_label='', classification=DataClassification.INTERNAL, parent_ids=None)` ¶

Record a new data lineage entry.

`add_transformation(data_id, transformation)` ¶

Record a transformation applied to data. Returns True if found.

`reclassify(data_id, new_classification)` ¶

Update classification level for a data item. Returns True if found.

`get_lineage(data_id)` ¶

Get lineage entry for a data item.

`get_by_classification(level)` ¶

Get all entries at or above a classification level.

`to_dict()` ¶

Export lineage state for audit/compliance reporting.

`classification_requirements(level)` ¶

Return minimum security requirements for a data classification level.

`security.quarantine`¶

`crp.security.quarantine` ¶

Ingest quarantine - anti-poisoning with 1-window quarantine (§7.8).

Facts from untrusted sources are quarantined for 1 window with a 0.7× confidence penalty. Cross-reference validation promotes or rejects them. Batch poisoning detection: >30% failures → quarantine entire batch.

`QuarantineEntry` `dataclass` ¶

A fact held in quarantine.

`QuarantineReport` `dataclass` ¶

Result of cross-reference validation pass.

`IngestQuarantine` ¶

1-window quarantine with confidence penalty and batch poisoning detection (§7.8).

Workflow: 1. Incoming facts go into quarantine with 0.7× confidence 2. After 1 window, cross-reference against extraction-derived facts 3. Matching facts are promoted (confidence restored) 4. Non-matching facts are rejected 5. If >30% of a batch fails, quarantine entire batch

Usage

q = IngestQuarantine() q.quarantine_facts([...], "w-1", source="user_input")

... next window processes ...¶

report = q.validate_and_promote("w-2", extraction_fact_texts)

`quarantine_count` `property` ¶

Number of facts currently in quarantine (not promoted or rejected).

`history` `property` ¶

Return the history.

`quarantine_fact(fact_id, original_confidence, window_id, source_label='', fact_text='')` ¶

Place a single fact into quarantine with 0.7× confidence penalty (§6F.1).

`quarantine_facts(facts, window_id, source_label='')` ¶

Quarantine a batch of facts.

Facts can be (fact_id, confidence) or (fact_id, confidence, text).

`get_penalized_confidence(fact_id)` ¶

Get quarantine-penalized confidence for a fact.

`is_quarantined(fact_id)` ¶

Check if a fact is in active quarantine.

`validate_and_promote(current_window_id, extraction_fact_texts, similarity_threshold=0.5)` ¶

Cross-reference validation: promote or reject quarantined facts (§6F.2).

Facts quarantined in an earlier window are validated against extraction-derived facts. Text overlap > threshold → promote.

Parameters:

Name	Type	Description	Default
`current_window_id`	`str`	Current window being processed	required
`extraction_fact_texts`	`dict[str, str]`	{fact_id: text} from extraction pipeline	required
`similarity_threshold`	`float`	Word overlap threshold for cross-reference	`0.5`

Returns:

Type	Description
`QuarantineReport`	QuarantineReport with promotion/rejection counts

`get_active_entries()` ¶

Return all currently quarantined (non-promoted, non-rejected) entries.

`clear()` ¶

Clear all quarantine state.

`security.rbac`¶

`crp.security.rbac` ¶

RBAC & rate limiting - role-based access control + per-session limits (§7.10).

Roles: OBSERVER (read-only), OPERATOR (dispatch + ingest), ADMIN (full). Rate limits: 60 req/min dispatch, 100 MB/min ingest, per-session token cap.

`Role` ¶

Bases: IntEnum

RBAC roles with increasing privilege (§6G.1).

`Permission` ¶

Bases: str

Named permission strings.

`RateLimitConfig` `dataclass` ¶

Rate limiting configuration (§6G.2).

`RateLimitState` `dataclass` ¶

Sliding-window rate limit tracking.

`AccessResult` `dataclass` ¶

Result of an RBAC check.

`RBACEnforcer` ¶

Role-based access control + rate limiting (§7.10).

Usage

rbac = RBACEnforcer(role=Role.OPERATOR) result = rbac.check_permission("dispatch") if not result.allowed: raise PermissionError(result.reason) result = rbac.check_rate_limit("dispatch") if not result.allowed: raise RateLimitError(result.reason) rbac.record_dispatch()

`role` `property` `writable` ¶

Return the role.

`session_tokens_used` `property` ¶

Return the session tokens used.

`check_permission(permission)` ¶

Check if current role has the given permission.

`has_permission(permission)` ¶

Return True if the current role has permission.

`check_rate_limit(operation, payload_bytes=0)` ¶

Check if the operation is within rate limits.

`record_dispatch(tokens_used=0)` ¶

Record a dispatch operation for rate limiting.

`record_ingest(payload_bytes)` ¶

Record an ingest operation for rate limiting.

`record_tokens(count)` ¶

Record token consumption.

`reset_limits()` ¶

Reset all rate limit counters.

`security.safety_manifest`¶

`crp.security.safety_manifest` ¶

Safety Manifest - the single config that drives code + dashboard (SPEC-033 §2).

The Manifest is the one source of truth that both the Safety Control Plane and any dashboard UI read and write. Change it in code and the dashboard reflects it; change it in the dashboard and the file updates.

`SafetyManifest` `dataclass` ¶

The one config that drives all CRP safety - code and dashboard (SPEC-033 §2).

Attributes:

Name	Type	Description
`profile`	`str`	Named profile ("balanced", "strict", "medical", "financial") or "custom" when any field deviates from a named profile.
`settings`	`dict[str, Any]`	Dict of capability-name → current value.
`checkpoints`	`list[dict[str, Any]]`	List of checkpoint declarations (inline HITL rules).
`custom_rules`	`list[str]`	List of paths to custom rule modules.

`get(name, default=None)` ¶

Read a setting by name.

Parameters:

Name	Type	Description	Default
`name`	`str`	Setting key.	required
`default`	`Any`	Value returned if the setting is absent.	`None`

Returns:

Type	Description
`Any`	Current setting value or `default`.

`set(name, value)` ¶

Write a setting - marks profile as custom if it deviates.

Parameters:

Name	Type	Description	Default
`name`	`str`	Setting key.	required
`value`	`Any`	New value to store.	required

Returns:

Type	Description
`None`	None. Updates `profile` to `"custom"` when the value differs
`None`	from the named profile defaults.

`compute_hash()` ¶

Return a deterministic hash of this manifest for CRP-Config-Hash.

Returns:

Type	Description
`str`	Hex-encoded BLAKE2b digest of the canonical settings JSON.

Raises:

Type	Description
`ValueError`	If the settings cannot be serialised (rare).

`to_dict()` ¶

Serialise the manifest to a JSON-safe dict.

Returns:

Type	Description
`dict[str, Any]`	Dict with `profile`, `settings`, `checkpoints`, and `custom_rules`.

`from_dict(data)` `classmethod` ¶

Restore a SafetyManifest from a serialised dict.

Parameters:

Name	Type	Description	Default
`data`	`dict[str, Any]`	Dict produced by `to_dict`.	required

Returns:

Type	Description
`SafetyManifest`	Reconstructed `SafetyManifest` instance.

`security.session_token`¶

`crp.security.session_token` ¶

CRP Session Token v3 - signed, stateless session relay (CRP-SPEC-007).

The session token carries signed session state in the CRP-Set-Session response header. The client echoes it back via CRP-Session-Token. Any gateway instance sharing the master key can validate the signature and resume the session - no shared session store required.

Pipeline:

:func:build_token - serialise + HMAC-sign a :class:SessionTokenPayload.
:func:format_set_session_header - render the CRP-Set-Session wire value.
:func:parse_token - split + decode a token without verifying.
:func:validate_token - full validation (signature, expiry, chain tip, scope, policy-hash tightening, nonce) per SPEC-007 §5.

Signing (SPEC-007 §3)::

session_signing_key = HKDF-SHA256(IKM=master_key, salt=sid_bytes,
                                  info="crp-session-sign-v3", L=32)
header    = base64url({"alg":"HS256","typ":"CRP"})
payload   = base64url(json(payload_object))
signature = HMAC-SHA256(header + "." + payload, session_signing_key)
token     = payload + "." + base64url(signature)

`SessionTokenPayload` `dataclass` ¶

Decoded CRP session-token payload (SPEC-007 §2.2).

`to_json_obj()` ¶

Serialize the payload to a JSON-serializable dict.

`from_json_obj(obj)` `classmethod` ¶

Create a new instance from a JSON-compatible object.

Parameters:

Name	Type	Description	Default
`obj`	`dict[str, object]`	The obj value.	required

Returns:

Type	Description
`SessionTokenPayload`	`SessionTokenPayload`.

`TokenStatus` ¶

Bases: str, Enum

Validation outcomes (SPEC-007 §5).

`TokenValidation` `dataclass` ¶

Result of :func:validate_token.

`ok` `property` ¶

Return whether the ok condition holds.

`http_status` `property` ¶

Return the HTTP status.

`derive_signing_key(master_key, session_id)` ¶

Derive the per-session 256-bit signing key (SPEC-007 §3.1).

`build_token(payload, master_key)` ¶

Serialise + sign payload, returning payload_b64.signature_b64.

Raises:

Type	Description
`ValueError`	if the encoded payload exceeds 4096 bytes (SPEC-007 §2.4).

`issue_token(*, session_id, master_key, window=1, quality_history=None, safety_budget=1.0, chain_tip='', continuation_id='', dag_pattern='LINEAR', strategy='', policy_hash='', ckf_hash='', scope='', nonce='', key_version=None, lifetime=DEFAULT_TOKEN_LIFETIME, now=None)` ¶

Convenience: build a fresh payload with iat/exp and sign it.

`format_set_session_header(token, payload, *, max_age=None)` ¶

Render the CRP-Set-Session header value (SPEC-007 §2.1).

`parse_token(token)` ¶

Split + decode a token's payload without verifying its signature.

`validate_token(token, *, master_keys, expected_chain_tip=None, expected_scope=None, presented_policy_hash=None, policy_is_more_restrictive=None, presented_nonce=None, now=None)` ¶

Full session-token validation (SPEC-007 §5).

Parameters:

Name	Type	Description	Default
`token`	`str`	the raw `CRP-Session-Token` value.	required
`master_keys`	`bytes \| list[bytes]`	the gateway master key(s). A list allows trying both the current and previous key during master-key rotation (SPEC-007 §6).	required
`expected_chain_tip`	`str \| None`	gateway's recorded HMAC chain tip for the session. Mismatch → 409 (stale/replayed token).	`None`
`expected_scope`	`str \| None`	API-key prefix from the request `Authorization`. Mismatch → 403.	`None`
`presented_policy_hash`	`str \| None`	hash of a `CRP-Safety-Policy` presented with the request. If it differs from the token's `pol` it MUST be more restrictive (set policy_is_more_restrictive) else → 403.	`None`
`policy_is_more_restrictive`	`bool \| None`	caller-computed tightening check.	`None`
`presented_nonce`	`str \| None`	`CRP-Safety-Nonce` from the request. Mismatch → 400.	`None`
`now`	`int \| None`	override current epoch seconds (for testing).	`None`

Returns:

Type	Description
`TokenValidation`	class:`TokenValidation`.

`security.transcript_store`¶

`crp.security.transcript_store` ¶

Content-addressed transcript store for the compliance audit chain (§7.14).

The tamper-evident audit trail (:mod:crp.security.audit_trail) records hashes and telemetry for LLM interactions - never the content. Every window mints WindowNode.raw_output_id but, before this module, nothing persisted the raw output against that identifier. TranscriptStore closes that gap: it persists the assembled message array and raw output per window, keyed by raw_output_id and content hash, reusing the truncated SHA-256 hashes already chained in LLM_CALL_COMPLETED so :func:verify_transcripts can prove stored transcript content matches the chain - and flag tampering at the exact window.

Capture is OFF by default (CRP_AUDIT_CAPTURE_TRANSCRIPTS / audit_capture_transcripts); when disabled the store is never touched and no TRANSCRIPT_RECORDED events are emitted (zero overhead by default).

Design reference: docs/FIX_NOTE_transcript_capture.md (schema 2.1.0).

`TranscriptRecord` `dataclass` ¶

One window's persisted transcript (messages + raw output).

Content hashes are computed over the stored (possibly redacted) content so verification always compares like with like; the TRANSCRIPT_RECORDED audit event carries the same 16-char prefixes chained by LLM_CALL_COMPLETED.

`to_dict()` ¶

Serialize to a JSON-safe dict.

`from_dict(data)` `classmethod` ¶

Reconstruct a record from :meth:to_dict output.

`recompute_hashes()` ¶

Re-hash the content as stored on disk; used by tamper verification.

`expected_output_hash()` ¶

Hash the stored output is expected to carry (stored variant when truncated).

`TranscriptStore` ¶

Content-addressed store for per-window transcripts (§7.14).

Records are persisted as one JSON file per raw_output_id under storage_dir plus a manifest.jsonl index. When storage_dir is None the store is memory-only (useful for tests and ephemeral sessions).

Size cap: the cap is enforced on the FINAL serialized record size (encode-then-measure). When a record exceeds max_bytes, stored content is shrunk - messages first (they largely duplicate system_prompt / task_input), then output, then system_prompt, then task_input - halving the current field each pass until the serialized blob fits. Truncation is byte-accurate (multi-byte UTF-8 is never split) and truncated=True is set. Content hashes of the FULL pre-truncation content are kept on the record and on the audit event, so the chain linkage survives truncation - the full output remains re-derivable from the caller.

`storage_dir` `property` ¶

On-disk directory, or None for memory-only storage.

`max_bytes` `property` ¶

Per-transcript serialized byte cap.

`store(record)` ¶

Persist record; returns the transcript reference (file path or id).

Applies the byte cap on the FINAL serialized size: content fields are shrunk (messages first - they largely duplicate system_prompt / task_input - then output, then system_prompt, then task_input) until the encoded JSON fits max_bytes. Full-content hashes are computed BEFORE truncation and stored on the record.

`get(raw_output_id)` ¶

Fetch a transcript by raw_output_id - resolves the dangling hook.

`list_ids()` ¶

All stored transcript ids (memory + on-disk manifest).

`verify_entry(record)` ¶

Re-hash stored content and report fields that no longer match.

Returns a list of human-readable mismatch descriptions (empty when the record is intact). This detects post-hoc tampering with the stored transcript at this exact window.

`TranscriptVerification` `dataclass` ¶

Outcome of :func:verify_transcripts.

`hash_text(text)` ¶

Full SHA-256 hex digest of text (utf-8).

`hash_messages(messages)` ¶

SHA-256 of the canonical JSON serialization of the message array.

`redact_text(text, scan_result)` ¶

Replace PII detection spans with [REDACTED:<type>] markers.

The scanner is advisory and never stores raw PII - only hashes and positions - so redaction is performed by span replacement over the caller's copy, never inside the scanner.

Parameters:

Name	Type	Description	Default
`text`	`str`	Original text (system prompt, task input, or output).	required
`scan_result`	`PIIScanResult`	Result of :meth:`PIIScanner.scan` on text.	required

Returns:

Type	Description
`str`	Redacted text. When no PII was detected the input is returned
`str`	unchanged.

`verify_transcripts(store, trail, *, session_id=None)` ¶

Prove stored transcripts match the compliance audit chain (§7.14).

For every TRANSCRIPT_RECORDED event in trail (optionally filtered by session_id):

The stored record must exist (by raw_output_id) - otherwise the event is listed in missing.
The record's stored content is re-hashed and compared against the hashes recorded on the TRANSCRIPT_RECORDED event itself (input_hash / output_hash) - an edited transcript file flags here, at the exact window_id.
When the event was NOT redacted, the stored system_prompt / task_input are additionally compared against the 16-char hashes in the window's LLM_CALL_COMPLETED entry, reusing the hashes already chained there. Redacted transcripts skip this step by design (the stored content is intentionally different from what was hashed for the chain).

valid is True only when nothing is missing and no mismatch occurred.

`new_raw_output_id()` ¶

Mint a raw-output identifier (kept here so store and router agree).

`security.trust_monitor`¶

`crp.security.trust_monitor` ¶

Runtime trust monitor with trust decay and graduated response (SPEC-033 §3.5).

Observes agent behavior (tool calls, I/O strings, out-of-pattern actions) and computes a session trust score that decays when indicators of compromise (IoCs) are detected. The monitor recommends one of four actions:

allow - trust is healthy, continue normally. constrain - minor anomaly; tighten policy (e.g., require grounding). gate - significant anomaly; halt for checkpoint / human review. kill - trust collapsed; fire the kill-switch.

The monitor is intentionally lightweight and rule-first: it runs synchronously in the governance loop and never blocks on external ML inference.

`TrustAction` ¶

Bases: str

Graduated trust action recommendation.

`TrustActions` ¶

Canonical trust actions.

`IndicatorOfCompromise` `dataclass` ¶

One observed behavioral anomaly.

`to_dict()` ¶

Serialize IoC to a dict.

`TrustDecision` `dataclass` ¶

Outcome of a trust observation.

`to_dict()` ¶

Serialize decision to a dict.

`TrustMonitorConfig` `dataclass` ¶

Configuration for the trust monitor.

`TrustMonitor` ¶

Runtime behavioral trust monitor with graduated response.

Usage::

monitor = TrustMonitor(session_id="s-1", kill_switch=ks)
decision = monitor.observe({"tool": "http", "output": "..."})

`trust_score` `property` ¶

Current trust score in [0, 1].

`observe(observation)` ¶

Evaluate an observation and update trust.

Parameters:

Name	Type	Description	Default
`observation`	`dict[str, Any]`	Dict describing the event. Recognised keys: `tool` (str), `output` (str), `input` (str), `action` (str), `risk_level` (str). Any unknown key is stringified and scanned for patterns.	required

Returns:

Type	Description
`TrustDecision`	A `TrustDecision` with the recommended action and detected IoCs.

`summary()` ¶

Return a JSON-safe summary of the monitor state.

`security.validation`¶

`crp.security.validation` ¶

Input validation - Layer 1 structural validation, CANNOT be disabled (§7.4).

Enforces: - Size limit: 50 MB - Unicode NFC normalization - Null byte stripping - Control character stripping (preserves \n, \t, \r) - MIME type validation - Metadata key count ≤ 50

`ValidationResult` `dataclass` ¶

Result of input validation.

`InputValidator` ¶

Structural input validation - Layer 1, cannot be disabled (§7.4).

This validator ALWAYS runs on all input. It cannot be turned off. It performs structural sanitization without modifying semantic content.

Usage

validator = InputValidator() result = validator.validate("input text") if result.valid: use(result.sanitized_text)

`validate(text, mime_type=None, metadata=None)` ¶

Validate and sanitize input text (§7.4).

Steps: 1. Size check (50 MB limit) 2. Unicode NFC normalization 3. Null byte stripping 4. Control character stripping (keep \n, \t, \r) 5. MIME type validation (if provided) 6. Metadata key count check (≤ 50)

`validate_metadata(metadata)` ¶

Validate and truncate metadata keys (§6D.4).

Returns (sanitized_metadata, warnings).

crp.security¶

security¶

crp.security ¶

ComplianceAuditTrail ¶

durable_path property ¶

transcript_store property ¶

entry_count property ¶

enable_durable_append(path, fsync_every=1) ¶

flush_durable() ¶

close_durable() ¶

attach_transcript_store(store) ¶

record(event_type, session_id='', data=None) ¶

verify_chain(verify_receipts=False, public_key_pem=None) ¶

verify_file(path, public_key_pem=None) classmethod ¶

verify_transcripts(*, session_id=None) ¶

query(*, event_type=None, session_id=None, since=None, until=None) ¶

export(*, include_signatures=True, since=None) ¶

export_jsonl(*, since=None, include_transcripts=False) ¶

export_ndjson(*, since=None) ¶

export_ocsf(*, since=None, provider='', model='', include_transcripts=False) ¶

export_sarif(*, since=None) ¶

ComplianceEventType ¶

SessionBindingManager ¶

binding property ¶

session_key property ¶

key_version property ¶

create_session(session_id='') ¶

sign_request(payload) ¶

verify_request_signature(payload, signature) ¶

rotate_secret() ¶

store_to_keyring(service_name='crp-sdk') ¶

from_keyring(service_name='crp-sdk') classmethod ¶

Checkpoint dataclass ¶

wait_for_resolution() async ¶

resolve(resolution) ¶

decorate(when='always', timeout=300, on_timeout='escalate', on_reject='fallback') classmethod ¶

CheckpointRejectAction ¶

CheckpointResolution dataclass ¶

to_audit_dict() ¶

CheckpointResolutionAction ¶

CheckpointTimeoutAction ¶

CheckpointTrigger ¶

ClarificationAction ¶

ClarificationRequest dataclass ¶

to_dict() ¶

ClarificationResolution dataclass ¶

approved property ¶

AIRiskLevel ¶

AISystemCategory ¶

ComplianceReporter ¶

generate_report(session_stats=None, risk_assessment=None) ¶

generate_technical_documentation(transparency=None, risk_assessment=None) ¶

RiskAssessment dataclass ¶

to_dict() ¶

RiskClassifier ¶

assess(category=AISystemCategory.CONTEXT_MANAGEMENT, intended_purpose='', processes_personal_data=False, makes_automated_decisions=False, affects_fundamental_rights=False, safety_critical=False, profiles_individuals=False) ¶

TransparencyDeclaration dataclass ¶

to_dict() ¶

ConsentManager ¶

state property ¶

grant(purpose, reason='', expires_hours=0.0) ¶

deny(purpose, reason='') ¶

withdraw(purpose, reason='') ¶

check(purpose) ¶

check_required(purpose) ¶

to_dict() ¶

HumanOversightController ¶

config property ¶

level property ¶

requires_approval(operation) ¶

check_autonomous_limit() ¶

record_autonomous_dispatch() ¶

request_approval(operation, details=None) ¶

record_decision(event_id, approved, approved_by='', reason='') ¶

record_halt(operation, reason, details=None) ¶

should_halt_on_injection() ¶

should_halt_on_pii() ¶

to_dict() ¶

HumanOversightLevel ¶

ProcessingPurpose ¶

`security`¶

`crp.security` ¶

`ComplianceAuditTrail` ¶

`durable_path` `property` ¶

`transcript_store` `property` ¶

`entry_count` `property` ¶

`enable_durable_append(path, fsync_every=1)` ¶

`flush_durable()` ¶

`close_durable()` ¶

`attach_transcript_store(store)` ¶

`record(event_type, session_id='', data=None)` ¶

`verify_chain(verify_receipts=False, public_key_pem=None)` ¶

`verify_file(path, public_key_pem=None)` `classmethod` ¶

`verify_transcripts(*, session_id=None)` ¶

`query(*, event_type=None, session_id=None, since=None, until=None)` ¶

`export(*, include_signatures=True, since=None)` ¶

`export_jsonl(*, since=None, include_transcripts=False)` ¶

`export_ndjson(*, since=None)` ¶

`export_ocsf(*, since=None, provider='', model='', include_transcripts=False)` ¶

`export_sarif(*, since=None)` ¶

`ComplianceEventType` ¶

`SessionBindingManager` ¶

`binding` `property` ¶

`session_key` `property` ¶

`key_version` `property` ¶

`create_session(session_id='')` ¶

`sign_request(payload)` ¶

`verify_request_signature(payload, signature)` ¶

`rotate_secret()` ¶

`store_to_keyring(service_name='crp-sdk')` ¶

`from_keyring(service_name='crp-sdk')` `classmethod` ¶

`Checkpoint` `dataclass` ¶

`wait_for_resolution()` `async` ¶

`resolve(resolution)` ¶

`decorate(when='always', timeout=300, on_timeout='escalate', on_reject='fallback')` `classmethod` ¶

`CheckpointRejectAction` ¶

`CheckpointResolution` `dataclass` ¶

`to_audit_dict()` ¶

`CheckpointResolutionAction` ¶

`CheckpointTimeoutAction` ¶

`CheckpointTrigger` ¶

`ClarificationAction` ¶

`ClarificationRequest` `dataclass` ¶

`to_dict()` ¶

`ClarificationResolution` `dataclass` ¶

`approved` `property` ¶

`AIRiskLevel` ¶

`AISystemCategory` ¶

`ComplianceReporter` ¶

`generate_report(session_stats=None, risk_assessment=None)` ¶

`generate_technical_documentation(transparency=None, risk_assessment=None)` ¶

`RiskAssessment` `dataclass` ¶

`to_dict()` ¶

`RiskClassifier` ¶

`assess(category=AISystemCategory.CONTEXT_MANAGEMENT, intended_purpose='', processes_personal_data=False, makes_automated_decisions=False, affects_fundamental_rights=False, safety_critical=False, profiles_individuals=False)` ¶

`TransparencyDeclaration` `dataclass` ¶

`to_dict()` ¶

`ConsentManager` ¶

`state` `property` ¶

`grant(purpose, reason='', expires_hours=0.0)` ¶

`deny(purpose, reason='')` ¶

`withdraw(purpose, reason='')` ¶

`check(purpose)` ¶

`check_required(purpose)` ¶

`to_dict()` ¶

`HumanOversightController` ¶

`config` `property` ¶

`level` `property` ¶

`requires_approval(operation)` ¶

`check_autonomous_limit()` ¶

`record_autonomous_dispatch()` ¶

`request_approval(operation, details=None)` ¶

`record_decision(event_id, approved, approved_by='', reason='')` ¶

`record_halt(operation, reason, details=None)` ¶

`should_halt_on_injection()` ¶

`should_halt_on_pii()` ¶

`to_dict()` ¶

`HumanOversightLevel` ¶

`ProcessingPurpose` ¶

`ProcessingRecordKeeper` ¶