Verification & memory

How do I reject a fabricated citation without paying for a model call?

verify_claim checks that the citation literally exists in the source before it prompts the reviewer, so a quote the model invented never reaches the reviewer at all. The recipe is reviewer.call_count == 0: the reviewer was scripted to rubber-stamp anything and never got the chance. model_accepted is None, not False, and the distinction is load-bearing — None means never consulted, False means consulted and unconvinced — which tells you which rejections in a batch were free. The anchor also refuses citations under MIN_CITATION_CHARS (12) before doing anything else, since "the" appears in every document and anchors nothing.

from grapharc.runtime.verify import verify_claim
from grapharc.testing import ScriptedChatModel

SOURCE = """GraphARC wraps LangGraph with typed state contracts.
Budgets put hard ceilings on iterations, tokens, and wall-clock time.
"""

# A reviewer that would approve anything — the point is that it is never asked.
reviewer = ScriptedChatModel(responses=['{"supported": true, "reason": "looks right"}'])

verdict = verify_claim(
    reviewer,
    text="GraphARC was benchmarked at 10x faster than raw LangGraph",
    citation="benchmarked at 10x faster",
    source_text=SOURCE,
)

print("accepted:      ", verdict.accepted)
print("anchor_ok:     ", verdict.anchor_ok)
print("model_accepted:", verdict.model_accepted)
print("reason:        ", verdict.reason)
print("reviewer calls:", reviewer.call_count)

accepted:       False
anchor_ok:      False
model_accepted: None
reason:         citation does not exist in the source (deterministic anchor)
reviewer calls: 0

How do I keep a line-wrapped source from causing a false reject?

Real sources have newlines mid-sentence; a model quoting one perfectly still writes a space where the file has a \n, and exact matching would reject correct work. The anchor collapses whitespace — and nothing else: the citation is split on whitespace and rejoined as a regex with \s+ between tokens (_find_span). Every non-space character must still match exactly, so the latitude is bounded to the one artifact that actually causes false rejects. A paraphrase is not "close enough" — it is a different string, and the anchor's job is existence, not similarity.

quotes = [
    "Every node declares which state fields it may write",   # newline -> space
    "Every node  declares   which state fields",              # extra spaces
    "Each node declares which state fields it may write",     # paraphrase
    "Every node declares which state field it may write",     # one letter off
]

for quote in quotes:
    v = verify_claim(
        reviewer, text="Nodes declare their writes", citation=quote, source_text=SOURCE
    )
    print(f"{v.anchor_ok!s:<5}  {quote!r}")

True   'Every node declares which state fields it may write'
True   'Every node  declares   which state fields'
False  'Each node declares which state fields it may write'
False  'Every node declares which state field it may write'
verbatim in source: False
reviewer calls:     2

How do I catch a quote lifted out of a negated sentence?

The anchor proves a quote exists, not that it means what the claim says — "benchmarked at 10x faster" really does appear in "was never benchmarked at 10x faster". So the reviewer is shown a mechanically extracted window of CONTEXT_WINDOW_CHARS (240) either side of the match, sliced straight out of source_text. Two properties fall out: it is mechanical, so no part of the author's conversation can leak in and the reviewer cannot be persuaded by the reasoning that produced the claim; and it is bounded, so verifying against a 400 KB document costs a fixed-size prompt. The reviewer's entire context is one HumanMessage — no system prompt, no history.

anchor_ok: True | accepted: False
reason:    the sentence negates the quote
--- the one message the reviewer received ---
You are verifying a single claim against its cited evidence. Judge ONLY whether the evidence supports the claim. The surrounding source context is included: if it negates, reverses, or otherwise contradicts what the quote alone implies, the claim is NOT supported.
Claim: GraphARC was benchmarked at 10x faster than raw LangGraph
Evidence (verbatim from source): benchmarked at 10x faster than raw LangGraph
Surrounding source context: GraphARC was never benchmarked at 10x faster than raw LangGraph; no such measurement exists.
Reply with JSON: {"supported": true|false, "reason": "..."}

What happens when the reviewer replies with something that isn't a verdict?

Every ambiguity resolves toward rejection — which matters because bool("false") is True in Python, so a model quoting its boolean would otherwise be an accept. Only a real JSON true accepts: 1 does not, "true" does not, and an unparseable reply fails closed. The implementation uses isinstance(raw_supported, bool) rather than a truthiness test. The cost is real — a model that reliably answers "supported": "yes" has every claim rejected — and that is the deliberate trade: a systematic false-reject is visible in the stats, a systematic false-accept is not.

for reply in ['sounds right to me!', '{"supported": "false"}', '{"supported": "true"}',
              '{"supported": 1}', '{"verdict": true}', '{"supported": true}']:
    reviewer = ScriptedChatModel(responses=[reply])
    v = verify_claim(reviewer, text="Budgets bound work",
                     citation=CITATION, source_text=SOURCE)
    print(f"{v.accepted!s:<5} {reply!r:<26} {v.reason or '(reviewer gave no reason)'}")

False 'sounds right to me!'      reviewer reply unparseable — failing closed
False '{"supported": "false"}'   reviewer 'supported' was not a JSON boolean — failing closed
False '{"supported": "true"}'    reviewer 'supported' was not a JSON boolean — failing closed
False '{"supported": 1}'         reviewer 'supported' was not a JSON boolean — failing closed
False '{"verdict": true}'        reviewer reply unparseable — failing closed
True  '{"supported": true}'      (reviewer gave no reason)

How do I tell whether my verifier is actually any good?

A verifier that rejects everything has zero false accepts and zero value; evaluate_verdicts scores against known ground truth and keeps the two error types apart. VerifierStats is four counters (true_accepts, true_rejects, false_accepts, false_rejects), not a score, because the two error types have different costs and different owners: a false accept ships a wrong claim, a false reject burns a retry loop and eventually makes people turn verification off. Track a single "accuracy" number and a good verifier and a rejects-all verifier look similar; track four cells and true_accepts == 0 with a non-empty positive class is the tell.

good = ScriptedChatModel(responses=[
    '{"supported": true,  "reason": "the source says exactly this"}',
    '{"supported": false, "reason": "ceilings bound work, they do not speed it up"}',
])
paranoid = ScriptedChatModel(responses=['{"supported": false, "reason": "no"}'],
                             on_exhausted="repeat")

print("good verifier: ", run(good).model_dump())
print("rejects-all:   ", run(paranoid).model_dump())

good verifier:  {'true_accepts': 1, 'true_rejects': 2, 'false_accepts': 0, 'false_rejects': 0}
rejects-all:    {'true_accepts': 0, 'true_rejects': 2, 'false_accepts': 0, 'false_rejects': 1}

How do I make sure the reviewer is genuinely a different model?

There are two enforcement points and they are not equally strong. build_stage5 catches only author is reviewer — two separate instances of the same weights sail through. The stronger check is different_providers, which compares the vendor each gateway spec reaches (the model author when the id names one, anthropic/… vs openai/…; the backend's own vendor otherwise), so two Anthropic models on OpenRouter come back False, as does a Claude-CLI author paired with an Anthropic model over OpenRouter. Correlated agreement follows training lineage, not model names, and not which API you reached the model through. The CLI warns on a shared vendor rather than making you write the check.

from grapharc.examples.stage5_verifier import build_stage5
from grapharc.gateway import different_providers
from grapharc.testing import ScriptedChatModel

model = ScriptedChatModel(responses=["x"])
try:
    build_stage5(model, model)
except ValueError as exc:
    print("refused:", exc)

pairs = [
    ("openrouter/anthropic/claude-haiku-4.5", "openrouter/anthropic/claude-sonnet-4.5"),
    ("openrouter/anthropic/claude-haiku-4.5", "openrouter/openai/gpt-4o-mini"),
    ("claude-cli/claude-sonnet-5", "openrouter/openai/gpt-4o-mini"),
]
for a, b in pairs:
    print(f"different_providers({a!r}, {b!r}) -> {different_providers(a, b)}")

grapharc demo stage5 --model openrouter/anthropic/claude-haiku-4.5 \
                    --reviewer-model openrouter/openai/gpt-4o-mini

How do I wire verification into a graph?

build_stage5 is a working three-node graph — draft, verify, report — and reading grapharc/examples/stage5_verifier.py is the fastest way to see how verify_claim sits inside a node. Invoked with a scripted author emitting three claims, it returns accepted, rejected, and the verdicts list; three claims cost only two reviewer calls because the fabricated citation dies at the anchor. The Verdict list is kept in state, so a downstream node can route on why something was rejected rather than only on the boolean.

result = build_stage5(author, reviewer).invoke({"source_text": DEMO_SOURCE})

print("accepted:", result["accepted"])
print("rejected:", result["rejected"])
print("author calls:", author.call_count, "| reviewer calls:", reviewer.call_count)
for v in result["verdicts"]:
    print(f"  anchor_ok={v.anchor_ok!s:<5} model_accepted={v.model_accepted!s:<5} {v.claim_text}")

accepted: ['GraphARC enforces typed state contracts']
rejected: ['GraphARC is 10x faster', 'Budgets make graphs run faster']
author calls: 1 | reviewer calls: 2
  anchor_ok=True  model_accepted=True  GraphARC enforces typed state contracts
  anchor_ok=False model_accepted=None  GraphARC is 10x faster
  anchor_ok=True  model_accepted=False Budgets make graphs run faster

How do I record a fact so I can audit where it came from?

A Claim is a subject-predicate-object triple plus provenance; source is required, and run_id/confidence are optional but worth filling in. Lookups go through _normalize: NFKC, casefold, then every non-word character collapsed to a single space — so case and trailing punctuation do not split an entity, but punctuation inside a name does ("GraphARC" → grapharc, "Graph-ARC" → graph arc, different entities). Normalization is Unicode-aware rather than ASCII-only precisely so 東京 and 北京 do not both collapse to the empty key. observed_at comes from a strictly increasing clock at microsecond resolution so no two claims a process writes can tie — not cosmetic, since ties in a reverse=True sort are not reversed by Python's stable sort, which once made the two backends return dead ends in opposite orders from identical input.

from grapharc.memory import Claim, MemoryStore

store = MemoryStore()
claim = store.add(
    Claim(
        subject="GraphARC",
        predicate="depends on",
        object="LangGraph",
        source="README.md#L3",       # required: where it came from
        run_id="run-12",             # optional: which run learned it
        confidence=0.9,
    )
)

for lookup in ("GraphARC", "  grapharc  ", "GRAPHARC!", "Graph-ARC"):
    print(f"current({lookup!r:<14}) -> {[c.object for c in store.current(lookup)]}")

id:          bf12abda3985
observed_at: 2026-07-28T04:27:25.893671+00:00
is_current:  True
key:         ('grapharc', 'depends on')
current('GraphARC'    ) -> ['LangGraph']
current('  grapharc  ') -> ['LangGraph']
current('GRAPHARC!'   ) -> ['LangGraph']
current('Graph-ARC'   ) -> []

How do I correct a fact without losing what we used to believe?

supersede never deletes: the old claim stays, marked, pointing at its replacement — the "run #51 sees that run #37 replaced run #12's fact" story end to end. current returns one claim, history(subject, predicate) returns the whole chain oldest-first with each link's own source and run, and dead_ends(subject) returns exactly what was retracted and by what. Three preconditions live in one shared validate_supersede that both backends call: already-superseded and self-supersede raise ValueError, an unknown id raises KeyError. Self-supersede raises because when each backend implemented it separately, MemoryStore and SQLiteMemoryStore disagreed and both ended up with a claim whose superseded_by pointed at itself — invisible to current, self-referential in dead_ends.

store.add(fact("v1", "30s", "config/prod.yaml@2026-05", "run-12"))
store.supersede("v1", fact("v2", "10s", "config/prod.yaml@2026-06", "run-37"))
store.supersede("v2", fact("v3", "15s", "incident-4412 postmortem", "run-51"))

print("current: ", [(c.id, c.object) for c in store.current("auth-service")])
print("dead ends:", [(c.id, c.object, "->", c.superseded_by)
                     for c in store.dead_ends("auth-service")])
for c in store.history("auth-service", "times out after"):
    print(f"  {c.id} {c.object:<4} source={c.source:<26} run={c.run_id} "
          f"superseded_by={c.superseded_by}")

current:  [('v3', '15s')]
dead ends: [('v1', '30s', '->', 'v2'), ('v2', '10s', '->', 'v3')]
already superseded: ValueError: claim 'v1' was already superseded by 'v2'
self-supersede: ValueError: claim 'v3' cannot supersede itself — a correction is a new claim with its own id, so the old one survives to be pointed at
unknown claim: KeyError: "unknown claim 'nope'"

How do I keep memory across processes?

MemoryStore is a dict and dies with the interpreter; SQLiteMemoryStore implements the same ClaimStore protocol against a file, so swapping is a one-line change and the difference only shows up across a process boundary — which is how to test it (writer process, reader process that supersedes, then a third parent process reading both). The store opens in autocommit and wraps supersede in BEGIN IMMEDIATE, so its read-then-write cannot interleave with another process's — a deferred transaction takes the write lock at first write, letting two processes both read a claim as live and both supersede it. It runs synchronous = FULL under WAL, because losing the last few commits on a host failure would defeat the one thing this backend is for.

WRITER = """
import os, sys
from grapharc.memory import Claim, SQLiteMemoryStore

with SQLiteMemoryStore(sys.argv[1]) as store:
    store.add(Claim(id="v1", subject="auth-service", predicate="times out after",
                    object="30s", source="config/prod.yaml", run_id="run-12"))
    print(f"writer pid={os.getpid()}: wrote v1")
"""

writer pid=478023: wrote v1
reader pid=478024: found v1 = '30s' from config/prod.yaml (run run-12)
reader pid=478024: superseded v1 -> v2
parent: [('v2', '10s')]
parent dead ends: [('v1', '30s')]
both satisfy ClaimStore: True True

Can I query the claim graph in Cypher?

LadybugMemoryStore is the third backend, implementing the same ClaimStore protocol against LadybugDB — an embedded property-graph database with Cypher, forked from Kuzu after Apple acquired and closed it; embedded means the same deal SQLite offers, a path on disk with no server. The difference is what gets stored: the other two keep claims as rows and rebuild the graph in Python (ClaimIndex scans the whole corpus every time you construct one), while this backend stores the edges — superseded_by is a SUPERSEDED_BY edge rather than a column, and each claim's subject and object are Entity nodes. So a correction chain is a path you can walk, and the escape hatch from the seven-method protocol is a query rather than a scan. Install with pip install 'grapharc[ladybug]'.

# Not executed by the docs test — needs the `ladybug` extra, which CI does not install.
from grapharc.memory import Claim, LadybugMemoryStore

with LadybugMemoryStore("memory.lbdb") as store:
    old = store.add(Claim(subject="auth-service", predicate="times out after",
                          object="30s", source="config/prod.yaml", run_id="run-12"))
    store.supersede(old.id, Claim(subject="auth-service", predicate="times out after",
                                  object="5s", source="incident-114", run_id="run-37"))

    for was, now, source in store.cypher(
        "MATCH (old:Claim)-[:SUPERSEDED_BY]->(new:Claim) "
        "RETURN old.object, new.object, new.source"
    ):
        print(f"{was} -> {now} ({source})")

    store.cypher(
        "MATCH (:Claim {subject: $s})-[:MENTIONS]->(e:Entity)<-[:ABOUT]-(next:Claim) "
        "RETURN next.subject, next.predicate, next.object",
        {"s": "auth-service"},
    )

How do I find the claims relevant to a question?

search ranks by relevance and returns why each claim was returned; two ways in — name entities you already know are relevant, or pass free text. Scores land in two bands that cannot overlap: naming an entity is worth a flat 1.0 because the caller knows the subject matters, and the text channels contribute at most 0.9, so a named subject always outranks a claim found only by its words. A hop multiplies the parent's score by 0.45, so a neighbour lands below the claim it came from and reads as supporting evidence rather than the answer; hops, via and channel are on every ScoredClaim so a node can show its work rather than presenting a second-hand fact as a direct hit. Incoming edges are followed too, which is how billing-service depends on auth-service shows up — the "who calls me" answer.

print("--- entities=['auth-service'] ---")
for hit in search(store, entities=["auth-service"]):
    c = hit.claim
    print(f"{hit.score:.3f} hop={hit.hops} via={hit.via!s:<5} {hit.channel:<8} "
          f"{c.id} {c.subject} {c.predicate} {c.object}")

--- entities=['auth-service'] ---
1.000 hop=0 via=None  entity   a2 auth-service owned by platform team
1.000 hop=0 via=None  entity   a1 auth-service depends on redis
0.450 hop=1 via=a2    graph    p1 platform team on call rotation PagerDuty schedule P7
0.450 hop=1 via=a2    graph    b1 billing-service depends on auth-service
0.450 hop=1 via=a1    graph    r2 redis has known issue connection leak under TLS
0.450 hop=1 via=a1    graph    r1 redis runs version 7.2.4
--- query='connection leak' (no entity named) ---
0.900 hop=0 via=None  lexical  r2 redis has known issue connection leak under TLS
0.405 hop=1 via=r2    graph    a1 auth-service depends on redis

What if the query is misspelled, or there is no query at all?

Three behaviours to know before you debug an empty result list. No query and no entity returns nothing, not the whole store — there is nothing to be relevant to, and handing a node an arbitrary slice of memory labelled "context" is worse than handing it none. HashingEmbedder is a spelling channel, not a semantic one: hashed word tokens plus character n-grams, so authservice finds auth-service where BM25 scores a flat zero, while redsi still finds nothing because two transposed characters in a five-letter word leave too little n-gram overlap to clear MIN_SIMILARITY (0.15); it will never relate "car" to "automobile" — for meaning, inject a real embedder through the Embedder protocol and lower lexical_weight from its 0.6 default. known_entities returns normalized keys (hence auth service), so use it to decide what to look up, not to render.

# Build the index once, query it many times. Otherwise every search rebuilds it.
index = ClaimIndex.from_store(store)
for query in ("connection leak", "who depends on auth-service"):
    print(f"{query!r:<32} -> {[h.claim.id for h in search(store, query=query, index=index)]}")

print("search(store) with neither     ->", search(store))
print("'authservice' + HashingEmbedder->",
      [h.claim.id for h in search(store, query="authservice", embedder=HashingEmbedder())])
print("known entities:", sorted(known_entities(store)))

'connection leak'                -> ['r2', 'a1']
'who depends on auth-service'    -> ['a1', 'b1', 'r2']
search(store) with neither     -> []
'authservice' lexical only     -> []
'authservice' + HashingEmbedder-> ['a1', 'b1', 'r2']
'redsi' + HashingEmbedder      -> []
known entities: ['auth service', 'billing service', 'redis']

Why did my reused index stop finding new claims?

Because an index is a snapshot and it does not notice writes — a claim added after ClaimIndex.from_store is simply absent from results. This fails silently: no exception, just a fact quietly missing from a node's context. There is deliberately no ClaimIndex.add, because term statistics change with every write and an index a few claims stale does not error, it ranks wrongly. Rebuild after writing, or drop index= and eat the O(claims) rebuild.

store.add(Claim(id="a", subject="redis", predicate="runs version", object="7.2", source="tf"))
index = ClaimIndex.from_store(store)

store.add(Claim(id="b", subject="redis", predicate="has issue", object="leak", source="i-1"))

print("stale index:", [h.claim.id for h in search(store, entities=["redis"], index=index)])
print("rebuilt:    ", [h.claim.id for h in
                       search(store, entities=["redis"], index=ClaimIndex.from_store(store))])

stale index: ['a']
rebuilt:     ['b', 'a']

How do I fit memory into a prompt without blowing the context window?

render_context produces the brief a node actually puts in its prompt: "Known facts (with provenance)", "Contradictions — same subject and predicate, different values", and "Superseded — do not re-derive these", three sections in priority order with provenance on every line. The (related via auth-service) tag marks a hop-1 fact so the node can tell a direct answer from a neighbour. _fit drops whole entries from the tail and never truncates mid-line — half a claim with half its provenance is worse than no claim — and a section whose header fits but whose first entry does not is dropped entirely, so a header never appears over nothing; the omission note is the one line never dropped, because a brief that silently omits facts is worse than a short one.

print(render_context(store, entities=["auth-service"]))
print("=== same brief, max_tokens=90 ===")
print(render_context(store, entities=["auth-service"], max_tokens=90))

Known facts (with provenance):
- auth-service deployed to eu-central-1 [source: terraform@2026, observed: 2026-07-28T04:29:22.317681+00:00]
- auth-service times out after 10s [source: helm/values.yaml, observed: ...]
- auth-service depends on redis [source: arch.md, observed: ...]
- redis runs version 7.2.4 [source: infra/redis.tf, observed: ...] (related via auth-service)

Contradictions — same subject and predicate, different values:
- auth-service times out after: '30s' (t1, config/prod.yaml) | '10s' (t2, helm/values.yaml)

Superseded — do not re-derive these:
- auth-service deployed to eu-west-1 (superseded by m2)
- auth-service listens on port 8080 (superseded by p2)

=== same brief, max_tokens=90 ===
Known facts (with provenance):
- auth-service deployed to eu-central-1 [source: terraform@2026, observed: ...]
- auth-service listens on port 8443 [source: ingress.yaml, observed: ...]
(+7 lines omitted to fit a 90-token budget)

How do I keep the "already tried that" section from growing forever?

Every correction adds a dead end about the same subject, so on a long-lived project the superseded section grows without bound; it is capped by count (max_dead_ends) and says what it dropped. retrieve_dead_ends returns (kept, omitted) and orders newest correction first, because the oldest dead ends are least likely to be the one a node is about to walk into — and since the cap discards the tail, that ordering decides which corrections the node sees at all. The sort key is three fields (correction time, then observation time, then id) so it is a total order and both backends render the same section from the same claims; ties used to be left to list.sort's stability, which preserves rather than reverses input order, so reverse=True quietly handed back tied groups oldest-first. retrieve_dead_ends also takes query=, re-ranking by relevance first and correction time second, scored against the dead ends alone rather than the whole corpus.

kept, omitted = retrieve_dead_ends(store, entities=["auth-service"], max_dead_ends=2)
print("shown:", [(k.id, k.object) for k in kept], "omitted:", omitted)
print(render_context(store, entities=["auth-service"], max_dead_ends=2))

shown: [('v4', '15s'), ('v3', '20s')] omitted: 2

Known facts (with provenance):
- auth-service times out after 10s [source: incident-4412, observed: 2026-07-28T04:29:33.999234+00:00]

Superseded — do not re-derive these:
- auth-service times out after 15s (superseded by v5)
- auth-service times out after 20s (superseded by v4)
- (+2 older corrections omitted)

How do I notice that memory contradicts itself?

supersede needs you to already know the id of the claim being replaced, but an agent has a fact in hand and no idea what the store already believes — add_and_detect closes that gap, storing the incoming claim regardless and returning the conflicts it found, each with .describe(). Detection and resolution are separate calls on purpose: the test is structural — same normalized (subject, predicate), different normalized object — so a genuinely multi-valued predicate like depends on reads as disagreement, and auto-superseding would delete the second half of a multi-valued fact inside the one subsystem whose promise is that facts are never destroyed. So it reports, and a caller with a basis for a decision calls supersede_conflicting.

stored, conflicts = add_and_detect(store, incoming)
print("stored anyway:", stored.id)
for conflict in conflicts:
    print("conflict:", conflict.describe())

# Resolution is a separate, deliberate call.
supersede_conflicting(store, conflicts[0])

for group in unresolved_contradictions(store):
    print("flagged:", group.subject, group.predicate,
          [claim.object for claim in group.claims])

stored anyway: t2
conflict: auth-service times out after: '10s' (from incident-4412) contradicts '30s' (from config/prod.yaml, claim t1)
after resolution, current: [('t2', '10s')]
dead ends: [('t1', '30s')]
flagged: billing depends on ['postgres', 'redis']

How do I record the files an agent produced?

Claims answer "what is true"; artifacts answer "what was produced" — same rules, append-only with provenance mandatory. put on the same name records a new version (with parents= and claim_ids= links) rather than replacing, and v1's bytes stay readable; versions, latest, read_text, media_type and sha256 round it out, and a source="" raises ValueError. name is metadata and is never used to build a path, so a traversal-looking name is stored under the hash of its content like everything else. Names are matched exactly, unlike a claim's subject — Report.md is not report.md, because a filename is not an entity and entity resolution on filenames would merge two real files. Content addressing means two artifacts with identical bytes share one blob while keeping their own rows and provenance: the same report produced by two runs is two events.

v1 = store.put("report.md", "# Findings\nauth-service times out after 30s\n",
               source="node:summarize", run_id="run-12", node="summarize")
v2 = store.put("report.md", "# Findings\nauth-service times out after 10s\n",
               source="node:summarize", run_id="run-37", node="summarize",
               parents=[v1.id], claim_ids=["t2"])

traversal = store.put("../../etc/passwd", b"not a path", source="node:evil")
print("name is metadata:", traversal.name, "| stored under:", traversal.blob_key)
print(render_artifacts(store))

versions: [('07ecc720d4b1', 'run-12'), ('decc2c843fc3', 'run-37')]
media_type: text/plain; charset=utf-8 | sha256: 00f97cfbcd1181fe | parents: ('07ecc720d4b1',)
Report.md is a different name: []
no provenance: artifact 'notes.txt' needs a source — an artifact with no provenance cannot be attributed after the fact
name is metadata: ../../etc/passwd | stored under: ('2c', '2c2cc34cd0084471e6f45f217a6d1716ae9e7178722ed56585f95a6bfc51e1e6')

Artifacts (id · name · provenance):
- fb86dfa2e567 ../../etc/passwd (application/octet-stream, 10 bytes) [source: node:evil]
- decc2c843fc3 report.md (text/plain; charset=utf-8, 44 bytes) [source: node:summarize, run: run-37, node: summarize]

How do I put claims and artifacts in one file?

Point SQLiteArtifactStore at the same path as SQLiteMemoryStore — two connections to one SQLite database is normal, both take the write lock and wait out contention (journal_mode: wal). claim_ids on the artifact is the edge between the two halves: this report was produced from that claim, by id, so nothing can drift out of sync, and a later process reopening the file finds both halves. Metadata is a SQLite row and content is a file under <db>.blobs/<first two hex>/<sha256>, so the database stays small enough to query while a 200 MB core dump is still storable. The blob is written and fsynced before the row is inserted: crash between the two and you have an unreferenced blob (garbage), never a row pointing at content that does not exist (a lie).

with tempfile.TemporaryDirectory() as tmp:
    db = Path(tmp) / "memory.sqlite"
    with SQLiteMemoryStore(db) as claims, SQLiteArtifactStore(db) as artifacts:
        claim = claims.add(Claim(subject="auth-service", predicate="times out after",
                                 object="10s", source="incident-4412", run_id="run-37"))
        art = artifacts.put("report.md", "auth-service times out after 10s\n",
                            source="node:summarize", run_id="run-37",
                            claim_ids=[claim.id])
        print("blob:", artifacts.blob_path(art).relative_to(tmp))
        print("journal_mode:", claims.journal_mode)

claim: c25bee060492 | artifact: 2a8b76fc4ad0 | links to: ('c25bee060492',)
blob: memory.sqlite.blobs/52/52bb15f4df1a5a838214381ba412ef6988ba5ccbbd52afd5b9bdb8db552131a3
journal_mode: wal
reopened claims:    ['10s']
reopened artifacts: ['report.md']
content: auth-service times out after 10s
files: ['memory.sqlite', 'memory.sqlite.blobs']

Putting it together

The pattern the two subsystems are meant to be used in, in four steps: recall before working (render_context(store, entities=[...]) into the node's prompt, so the run starts from what is known and its dead ends); verify before writing (verify_claim each extracted fact against its source, the anchor making fabricated citations free to reject); write with provenance (Claim(..., source=..., run_id=ctx.run_id), plus add_and_detect if something may already be believed); and correct, do not overwrite (supersede, so the next run sees the correction instead of re-deriving the old answer). grapharc/examples/stage5_verifier.py and grapharc/examples/stage6_memory.py are the smallest complete graphs doing exactly this.

grapharc demo stage5
grapharc demo stage6