Decide what to remember
Govern claim formation with ClaimPolicy and the events-versus-state rule.
Use ClaimPolicy to make one domain decision about every structured memory
candidate. The default policy is conservative: authoritative observations can
be accepted, while agent and summary inference is held until corroborated.
Route current state to live context
Current state owned by another system must not become a durable claim. Return
reference_live_state for it; use observe for events and decisions that
pgmem may safely retain.
from pgmem import (
Claim, ClaimCandidate, ClaimPolicy, DefaultClaimPolicy, MemoryDecision,
MemoryDecisionAction,
)
class BillingPolicy:
def __init__(self) -> None:
self.default = DefaultClaimPolicy()
def decide(self, candidate: ClaimCandidate, *, competing_claims: list[Claim]) -> MemoryDecision:
if candidate.predicate == "subscription.status":
return MemoryDecision(
action=MemoryDecisionAction.reference_live_state,
rule="billing_system_owns_current_state",
reason="fetch current subscription status from the billing system",
)
return self.default.decide(candidate, competing_claims=competing_claims)Wire it with PgMem.create(dsn, claim_policy=BillingPolicy()). Then form an
auditable event claim:
claim, decision = await session.observe(
subject=SubjectRef(kind="subscription", identifier="sub_123"),
predicate="subscription.cancellation_requested",
value=True,
authority=Authority.application,
evidence=[Evidence(kind="billing_event", reference="evt_123", authority=Authority.application)],
)The returned decision explains whether the candidate was accepted, held,
corroborated, superseded, discarded, or redirected to live state. Query the
authoritative ledger with mem.store.claims.recall(scope=..., query=...).