Query Patterns

This guide covers the two read operations in mempill: query_memory (current belief and time travel) and query_audit (ledger forensics). For the write path see Quickstart (Python) and the language guides.

query_memory

query_memory computes the canonical belief for a (subject, predicate) pair from the full claim history at read time. The fold is deterministic (I8) — given the same claim history, it always returns the same result.

Pattern 1 — current belief

The most common query: what does the engine believe right now?

Python
Rust

result = engine.query_memory({
    "agent_id": "my-agent",
    "subject": "user",
    "predicate": "city",
    # as_of_tx_time omitted = now
})

belief = result["belief"]
print(belief["status"])  # "Resolved" | "TimingUncertain" | "Contested" | "NoBelief"

use mempill_core::application::QueryMemoryRequest;

let result = engine.query_memory(QueryMemoryRequest {
    agent_id: AgentId("my-agent".into()),
    subject: "user".into(),
    predicate: "city".into(),
    as_of_tx_time: None,
}).await?;

println!("status={:?}", result.belief.status);

Pattern 2 — point-in-time (time travel)

Read the engine’s belief as it was at a past transaction time. Useful for auditing, debugging “what did the agent think yesterday?”, or reproducible report generation:

Python
Rust

result = engine.query_memory({
    "agent_id": "my-agent",
    "subject": "acme:ceo",
    "predicate": "held_by",
    "as_of_tx_time": "2024-01-01T00:00:00Z",  # ISO-8601 UTC
})

use chrono::{Duration, Utc};

let one_week_ago = Utc::now() - Duration::weeks(1);
let result = engine.query_memory(QueryMemoryRequest {
    agent_id: agent.clone(),
    subject: "acme:ceo".into(),
    predicate: "held_by".into(),
    as_of_tx_time: Some(one_week_ago),
}).await?;

Current limitation: as_of_tx_time rewinds both transaction time and valid-time selection to the same instant. True separate bi-temporal querying (a distinct valid_at parameter) is planned for a future release.

Pattern 3 — extracting the belief value

The belief is a nested structure. The deep path is: belief → primary → fact → value

belief = result["belief"]
status = belief["status"]

if status in ("Resolved", "TimingUncertain"):
    primary = belief["primary"]
    value = primary["fact"]["value"]          # the authoritative value
    conf = primary["confidence"]["value_confidence"]
    vt = primary.get("valid_time") or {}
    vt_start = vt.get("start")               # ISO-8601 string or None
    vt_end = vt.get("end") or "open"
    claim_ref = primary["claim_ref"]          # UUID string of the winning claim
    print(f"{value!r}  conf={conf}  [{vt_start} .. {vt_end}]")

elif status == "Contested":
    # No primary — all candidates are in alternatives.
    for alt in belief.get("alternatives") or []:
        print(alt["fact"]["value"])

elif status == "NoBelief":
    print("No claim found for this (subject, predicate)")

Belief status reference

Status	`primary`	`alternatives`	Meaning
`"Resolved"`	present	empty	Single winner after valid-time fold
`"TimingUncertain"`	present	empty	Only one claim; no valid_time supplied — ordered by tx-time
`"Contested"`	`None`	2+ entries	Multiple claims; engine cannot pick a winner
`"NoBelief"`	`None`	empty	No claim exists for this (subject, predicate)

Pattern 4 — handling Contested beliefs

When status == "Contested", primary is None. Both (all) values are preserved in alternatives. Do not pick one silently — surface the uncertainty to the user or trigger oracle resolution.

if belief["status"] == "Contested":
    alts = [a["fact"]["value"] for a in belief.get("alternatives") or []]
    # Option A: surface to user
    print(f"Uncertain: {alts}")
    # Option B: trigger reconciliation
    engine.reconcile({
        "agent_id": "my-agent",
        "subject_lines": [("user", "city")],
    })

After reconcile, query_memory again. If an oracle is wired and resolved the conflict, status will be "Resolved". See Writing an Oracle.

Pattern 5 — valid-time queries

When you ingested claims with explicit valid_time, the fold selects the claim whose window contains the query instant. Use this for temporal succession queries:

# "Who is CEO today?"
q_now = engine.query_memory({"agent_id": "a", "subject": "acme:ceo", "predicate": "held_by"})
print(q_now["belief"]["primary"]["fact"]["value"])  # Bob (current)

# "Who was CEO in February 2022?"
q_past = engine.query_memory({
    "agent_id": "a",
    "subject": "acme:ceo",
    "predicate": "held_by",
    "as_of_tx_time": "2022-02-01T00:00:00Z",
})
print(q_past["belief"]["primary"]["fact"]["value"])  # Alice (past)

See Concepts: Valid-Time Succession.

history() — full claim timeline

history() returns every claim ever ingested for a (subject, predicate) subject-line, ordered oldest→newest. Use it to inspect how a belief evolved over time — who held the role before, when each value was valid, and which entry is current.

from mempill import open_in_memory, remember, history, recall, RememberOptions

engine = open_in_memory()
agent = "my-agent"

# Ingest a succession: Alice → John → Bob (non-overlapping windows)
remember(engine, agent, "acme:ceo", "held_by", "Alice",
         RememberOptions(valid_from="2018-01-01", valid_until="2021-01-01"))
remember(engine, agent, "acme:ceo", "held_by", "John",
         RememberOptions(valid_from="2021-01-01", valid_until="2024-01-01"))
remember(engine, agent, "acme:ceo", "held_by", "Bob",
         RememberOptions(valid_from="2024-01-01"))

# Retrieve the full timeline
h = history(engine, agent, "acme:ceo", "held_by")

for e in h:
    print(e.value, e.status, e.valid_from, "→", e.valid_until or "open")
# Alice  Superseded  2018-01-01 → 2021-01-01
# John   Superseded  2021-01-01 → 2024-01-01
# Bob    Current     2024-01-01 → open

use mempill::{open_default_in_memory, remember, history, recall, RememberOptions};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let engine = open_default_in_memory()?;
    let agent = "my-agent";

    // Ingest a succession: Alice → John → Bob (non-overlapping windows)
    remember(&engine, agent, "acme:ceo", "held_by", "Alice",
             RememberOptions::new().valid_from("2018-01-01").valid_until("2021-01-01")).await?;
    remember(&engine, agent, "acme:ceo", "held_by", "John",
             RememberOptions::new().valid_from("2021-01-01").valid_until("2024-01-01")).await?;
    remember(&engine, agent, "acme:ceo", "held_by", "Bob",
             RememberOptions::new().valid_from("2024-01-01")).await?;

    // Retrieve the full timeline
    let h = history(&engine, agent, "acme:ceo", "held_by").await?;

    for e in &h.entries {
        println!("{:?}  {:?}  {:?} → {:?}",
            e.value.as_str(), e.status, e.valid_from, e.valid_until);
    }
    // Some("Alice")  Superseded  Some(2018-01-01) → Some(2021-01-01)
    // Some("John")   Superseded  Some(2021-01-01) → Some(2024-01-01)
    // Some("Bob")    Current     Some(2024-01-01) → None (open)

    Ok(())
}

history().current() agrees with recall()

The current() helper returns the single Current entry and is guaranteed to agree with recall() — both use the same canonical fold:

Python
Rust

h = history(engine, agent, "acme:ceo", "held_by")

current_entry = h.current()       # HistoryEntry or None
recall_result = recall(engine, agent, "acme:ceo", "held_by")

assert current_entry.value == recall_result.as_str()  # "Bob" == "Bob"
assert not h.is_empty()

let h = history(&engine, agent, "acme:ceo", "held_by").await?;
let r = recall(&engine, agent, "acme:ceo", "held_by").await?;

// history().current() value matches recall() primary
assert_eq!(
    h.current().and_then(|e| e.value.as_str()),
    r.as_str(),
);
assert!(!h.is_empty());

HistoryEntry fields

Field	Type	Description
`value`	any	The asserted value for this claim
`status`	`"Current"` \| `"Superseded"`	Whether this entry is the live belief
`valid_from`	string \| None	RFC 3339 start of the valid-time window, or None
`valid_until`	string \| None	Effective end of the slot, or None (open-ended)
`provenance`	string	Human-readable label, e.g. `"External/UserAsserted"`
`value_confidence`	float	Confidence in this claim’s value (0.0–1.0)
`claim_ref`	string (UUID)	Identifies the underlying claim

Limitation: valid-time as-of (v0.3)

history() always returns the timeline as it exists now. There is no valid_at parameter yet — point-in-time valid-time queries (“who was CEO as of 2022-06-01 according to what we knew then?”) require separate bi-temporal axes that are planned for v0.3. Today, as_of_tx_time on query_memory rewinds transaction time and valid-time selection simultaneously but does not expose them independently.

query_audit

The audit ledger is an append-only record of every disposition event. Use query_audit for provenance forensics, compliance, and debugging.

AuditQueryRequest shape

Python
Rust

audit = engine.query_audit({
    "agent_id": "my-agent",   # required
    "claim_ref": None,        # str | None — filter to a specific claim UUID
    "from_tx_time": None,     # str | None — ISO-8601 UTC lower bound
    "limit": 100,             # int — max entries to return
})

use mempill_core::application::AuditQueryRequest;

let audit = engine.query_audit(AuditQueryRequest {
    agent_id: AgentId("my-agent".into()),
    claim_ref: None,           // or Some(claim_ref) to filter
    from_tx_time: None,
    limit: 100,
}).await?;

Pattern — full agent history

audit = engine.query_audit({
    "agent_id": "my-agent",
    "claim_ref": None,
    "from_tx_time": None,
    "limit": 500,
})

for entry in audit["entries"]:
    print(entry["event_kind"], entry["recorded_at"], entry["claim_ref"][:8])

Pattern — history for a specific claim

# First get the claim_ref from an ingest or query response:
resp = engine.ingest_claim({...})
claim_ref = resp["claim_ref"]

audit = engine.query_audit({
    "agent_id": "my-agent",
    "claim_ref": claim_ref,
    "from_tx_time": None,
    "limit": 50,
})

Pattern — recent events (time-bounded)

from datetime import datetime, timezone, timedelta

one_hour_ago = (datetime.now(timezone.utc) - timedelta(hours=1)).isoformat()

audit = engine.query_audit({
    "agent_id": "my-agent",
    "claim_ref": None,
    "from_tx_time": one_hour_ago,
    "limit": 200,
})

LedgerEventKind values

Event	Meaning
`ClaimCommitted`	Claim written and committed
`ValidityAsserted`	Valid-time bounds recorded or updated
`AdjudicationRequested`	Oracle was called; pending row created
`AdjudicationResolved`	`submit_adjudication` resolved the conflict
`RecallReEntryDetected`	Amplification Guard caught a RecallReEntry echo
`Quarantined`	Claim quarantined (provenance or amplification guard)
`DependentFlaggedPendingReview`	Claim whose source was invalidated
`AdjudicationExpired`	Pending adjudication TTL expired; returned to Contested

Understanding the belief shape from audit

Audit entries carry claim_ref but not subject, predicate, or value — the ledger is a disposition event log, not a content store. Correlate back to ingestion records in your application using the claim_ref returned from ingest_claim.

The mempill-demo adapter (memory_mempill.py) keeps a session-local registry dict mapping claim_ref → ClaimMeta for exactly this purpose.

Python (Advanced) — full dict shapes, TypedDicts, async
Rust (Advanced) — typed request structs, valid_time, cardinality
Writing an Oracle — resolving Contested beliefs
Concepts: Contested and Dispositions — 12-state model
Concepts: Temporal Validity Problem — why valid_time matters