Skip to content

How-to: configure Privacy Runtime

Wire PIIDetector, PermissionManager, DataClassifier, and AuditChain into the pipeline so every capsule is enforced against LGPD/GDPR-aligned controls.

What gets enforced

Rule Where Behavior
Rule 21 (privacy by design) Phase 3 — Automated Gates PIIDetector.has_pii(context) blocks any real PII (email, CPF, CNPJ, SSN, phone, credit card, API keys, passwords)
Rule 22 (purpose binding) Phase 3 + SkillInvoker Constraints.purpose must be a valid Purpose enum; SkillInvoker checks RBAC per call
Rule 23 (data classification) Phase 3 DataClassifier runs on context; if tier ≥ CONFIDENTIAL, metadata.data_tier must be declared
Rule 9 (PII redaction) Phase 4 (defense-in-depth) Auto-sanitize PII in context before executor; emits SANITIZE audit entry
Rule 24 (audit chain inviolable) Every phase + every DENY HMAC-chained append-only log

Quickstart

from src.ai.metrics import AgentMetrics
from src.pipeline import (
    BudgetTracker, CheckpointStore, ContinuationStore, LockManager,
    PipelineOrchestrator, VerificationAgent,
)
from src.privacy import AuditChain

# Privacy Runtime: just attach AuditChain — Phase 3 auto-loads PIIDetector + DataClassifier
audit = AuditChain(path="_framework/audit/chain.jsonl")
metrics = AgentMetrics(base_dir="_framework/observability")
ckpts = CheckpointStore(base_dir="_framework/checkpoints")

orch = PipelineOrchestrator(
    budget=BudgetTracker(capsule_id="...", limit_tokens=50_000, limit_dollars=2.0),
    locks=LockManager(),
    verifier=VerificationAgent(),
    audit_chain=audit,
    metrics=metrics,
    checkpoint_store=ckpts,
)

That's it. Phase 3 already wires PIIDetector + DataClassifier automatically.

Adding RBAC at skill-invocation boundary

For skill-level access control (Rule 22 — purpose binding per call):

from src.privacy import DataTier, PermissionManager, Purpose
from src.skills.invoker import SkillInvoker
from src.ai.model_mgmt import ModelManager

inv = SkillInvoker(
    model_manager=ModelManager.default(),
    metrics=metrics,
    permissions=PermissionManager(),   # default policy matrix
    audit_chain=audit,                 # DENY entries get appended here
)

# Code-writer trying to read RESTRICTED data → denied (max_tier=INTERNAL)
result = inv.invoke(
    skill=skill_manifest,
    agent_id="code-writer",
    inputs={},
    data_tier=DataTier.RESTRICTED,
    purpose=Purpose.FEATURE_BUILD,
)
assert not result.success
assert "permission_denied" in result.error

The default policy matrix:

Role max_tier allowed_purposes
operator RESTRICTED all
auditor-haiku RESTRICTED AUDIT, DSAR
cortex CONFIDENTIAL FEATURE_BUILD, ANALYTICS
code-writer INTERNAL FEATURE_BUILD, DEBUG
test-runner INTERNAL FEATURE_BUILD
file-operator CONFIDENTIAL FEATURE_BUILD, DEBUG

Override by passing your own PermissionManager(policy=[...]).

DSAR (Data Subject Access Request)

The framework supports ACCESS and ERASURE per LGPD Art. 18 / GDPR Art. 15-17:

from src.privacy import AuditChain, DeletionCascade, DSARHandler
from src.privacy.dsar import DSARAction
from src.ai.memory import EpisodicStore, ProceduralStore, SemanticStore

ep, sm, pr = EpisodicStore(), SemanticStore(), ProceduralStore()
audit = AuditChain()
deletion = DeletionCascade(episodic=ep, semantic=sm, procedural=pr, audit=audit)

dsar = DSARHandler(
    episodic=ep, semantic=sm, procedural=pr,
    audit=audit, deletion=deletion,
)

# Subject requests all data about them
req = dsar.new_request(subject="eric@example.com", action=DSARAction.ACCESS)
resp = dsar.handle(req)
print(f"Found {resp.found_records} records")

# Subject requests erasure
req2 = dsar.new_request(subject="eric@example.com", action=DSARAction.ERASURE)
resp2 = dsar.handle(req2)
print(f"Erased {resp2.affected_records} records")

Erasure cascades through all 3 memory stores + emits an audit entry per deletion.

TTL retention sweep

RetentionScheduler.sweep() walks each store and deletes records past their TTL:

from src.privacy.retention import RetentionScheduler

scheduler = RetentionScheduler(episodic=ep, semantic=sm, procedural=pr, audit=audit)
report = scheduler.sweep()
print(report)  # {"scanned": N, "expired": M, "errors": []}

Default TTLs: episodic=90 days, procedural=180 days, semantic=None (never expires). Override with custom RetentionPolicy list.

Verifying the audit chain

ok, errors = AuditChain(path="_framework/audit/chain.jsonl").verify()
assert ok, errors

Any tampering (edit, reorder, delete) breaks the HMAC chain. Requires AUDIT_HMAC_KEY env var in production.

What blocks a capsule (cheat sheet)

Context contains Halt reason
"eric@example.com" absolute_rules_violated:[21]
"CPF: 123.456.789-00" absolute_rules_violated:[21]
"password: hunter2" absolute_rules_violated:[21]
"api_key=...16+chars" absolute_rules_violated:[21]
"internal-only — do not share" (no data_tier) absolute_rules_violated:[23]
Constraints.purpose="bogus" absolute_rules_violated:[22]

See also