Skip to main content

​
Control Path Pipeline

Every memory operation in Memproof — whether it is a remember, update, forget, or search — passes through the same deterministic pipeline. The pipeline has six stages, each emitting an immutable event to the Audit Trail.

​
Pipeline Stages

​
Stage Details

1

received

The orchestrator accepts the request and generates a unique operation_id (format: op-{hex16}). An idempotency check runs first — if the same idempotency key was already processed, the cached response is returned. The received event is emitted to the ledger with the operation’s scope.Data produced: operation_id, validated request model
2

risk_assessed

The Risk Engine scores the operation across five weighted factors: operation type, PII detection, secrets detection, source trust, and scope anomalies. A separate assess_content_flags call produces boolean flags (contains_pii, contains_secret) used by the policy engine.Data produced: RiskAssessment (score, level, factors, scorer), content flags
3

policy_decided

The Policy Engine evaluates YAML rules in priority order against the operation type, risk assessment, scope, context, and content flags. The first matching rule wins, producing a PolicyDecision with one of four actions.Data produced: PolicyDecision (action, reason_codes, matched_rule_ids, policy_version)
4

approval_requested (conditional)

This stage only runs when the policy decision is require_approval. The Approval Broker sends the request to an external approval system (Attesta) or queues it for manual review. Three outcomes are possible:
  • Synchronous approval: the broker returns immediately and the pipeline continues.
  • Async/pending: the operation is stored with status pending_approval and the response is returned to the caller. A later call to mp.approve() or mp.deny() completes the operation.
  • Denied: the pipeline emits a blocked event and raises ApprovalDeniedError.
5

provider_attempted

The orchestrator delegates to the memory adapter to execute the actual backend operation (create_memory, update_memory, delete_memory, or search_memories). If the adapter raises an AdapterError, the operation status is set to failed and a ProviderUnavailableError is raised.Data produced: MemoryRecord (on success)
6

committed / blocked

The terminal stage. On success, a committed event is emitted and the operation record is stored with status committed. On policy denial, quarantine, or approval rejection, a blocked event is emitted instead.Data produced: final MemoryOperationResponse or OperationStatusResponse

​
Operation Statuses

Each operation ends in one of these statuses:
StatusMeaning
committedSuccessfully persisted to the memory backend
blockedDenied by policy or rejected during approval
quarantinedHeld for review; payload stored in quarantine
pending_approvalAwaiting human or system approval
failedAdapter error during provider execution

​
Idempotency

Every mutation endpoint (remember, update, forget) requires an idempotency key. If you do not provide one, Memproof generates a random key. The idempotency check runs before the pipeline begins:
  • Same key, same payload: returns the cached response.
  • Same key, different payload: raises ConflictError.
result = await mp.remember(
    content="user prefers dark mode",
    scope={...},
    context={...},
    idempotency_key="idem-unique-abc123",
)

​
Approval Flow

When a policy decision returns require_approval, pending operations can be resolved later: 
# Approve a pending operation
await mp.approve(operation_id="op-abc123", actor_id="admin@acme.com")

# Or deny it
await mp.deny(operation_id="op-abc123", actor_id="admin@acme.com", notes="Not authorized")
Operations in pending_approval status hold the deferred request payload in memory. Ensure approval decisions are resolved promptly to avoid unbounded memory growth.

​
Querying Operation Status

You can check the current status of any tracked operation at any time: 
status = mp.get_operation_status("op-abc123")
# status.status: committed | blocked | pending_approval | quarantined | failed
# status.decision: the PolicyDecision that was made
# status.risk_assessment: the RiskAssessment produced