diff --git a/Autotests/run_mandatory b/Autotests/run_mandatory
index cc815905..fdb4f6b4 100644
--- a/Autotests/run_mandatory
+++ b/Autotests/run_mandatory
@@ -33,4 +33,5 @@ mock/test_transition_episodes_after_eviction_mock.py
 mock/test_transition_metta_to_remember_mock.py
 mock/test_transition_pin_to_remember_mock.py
 mock_websocket/test_wschat_unit.py
+test_memory_schema.py
 test_websearch_smoke.py
diff --git a/Autotests/test_memory_schema.py b/Autotests/test_memory_schema.py
new file mode 100644
index 00000000..155b795c
--- /dev/null
+++ b/Autotests/test_memory_schema.py
@@ -0,0 +1,249 @@
+"""Unit tests for the provenance-aware memory schema (Issue #5).
+
+Pure schema/validation/filter tests run everywhere (no chromadb needed). The
+chroma-backed store tests use an in-memory EphemeralClient and skip on hosts
+without chromadb (they run in-container). Runs under pytest and standalone
+(`python3 Autotests/test_memory_schema.py`).
+"""
+import os
+import sys
+
+_REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+_SRC = os.path.join(_REPO_ROOT, "src")
+for _p in (_SRC, _REPO_ROOT):
+    if _p not in sys.path:
+        sys.path.insert(0, _p)
+
+import importlib.util  # noqa: E402
+import json  # noqa: E402
+import memory_schema as ms  # noqa: E402
+
+_HAS_CHROMA = importlib.util.find_spec("chromadb") is not None
+
+
+class _ChromaSkip(Exception):
+    """Raised to skip a chroma-backed test when chromadb is unavailable (standalone)."""
+
+
+# --- pure: build / validate / defaults -----------------------------------
+
+def test_default_confidence_game_state_vs_llm():
+    assert ms.build_metadata("c", "s", "game_state")["confidence"] == 1.0
+    assert ms.build_metadata("c", "s", "llm")["confidence"] == 0.55
+    assert ms.default_confidence("knowledge_prior") == 0.7
+
+
+def test_build_metadata_shape_and_atoms_json():
+    m = ms.build_metadata("City A low food", "freeciv.turn_42", "game_state",
+                          session_id="game-123", turn_id=42, atoms=["(Inheritance CityA LowFood)"])
+    assert m["claim"] and m["source"] == "freeciv.turn_42" and m["source_type"] == "game_state"
+    assert m["session_id"] == "game-123" and m["turn_id"] == 42
+    assert isinstance(m["atoms_json"], str)
+    assert json.loads(m["atoms_json"]) == ["(Inheritance CityA LowFood)"]
+    assert m["created_at"] and m["supersedes"] == ""
+    # all metadata values are chroma-safe scalars
+    for v in m.values():
+        assert isinstance(v, (str, int, float, bool)), v
+
+
+def test_explicit_confidence_overrides_default():
+    assert ms.build_metadata("c", "s", "llm", confidence=0.9)["confidence"] == 0.9
+
+
+def test_validate_rejects_bad_records():
+    assert ms.validate_metadata(ms.build_metadata("c", "s", "user")) is None
+    assert ms.validate_metadata({"claim": "c", "source": "s", "source_type": "bogus", "confidence": 1.0})
+    assert ms.validate_metadata({"claim": "c", "source": "s", "source_type": "user", "confidence": 2.0})
+    assert ms.validate_metadata({"claim": "", "source": "s", "source_type": "user", "confidence": 1.0})
+    assert ms.validate_metadata({"claim": "c", "source": "", "source_type": "user", "confidence": 1.0})
+
+
+def test_claim_id_stable_and_deterministic():
+    a = ms.build_metadata("same claim", "src.1", "game_state", turn_id=1)
+    b = ms.build_metadata("same claim", "src.1", "game_state", turn_id=1)
+    assert ms.claim_id(a) == ms.claim_id(b)
+    c = ms.build_metadata("different", "src.1", "game_state", turn_id=1)
+    assert ms.claim_id(c) != ms.claim_id(a)
+
+
+# --- pure: filters --------------------------------------------------------
+
+def test_build_where_defaults_scope_provenance_and_supersession():
+    # By default: only provenance source_types, and exclude superseded.
+    d = ms.build_where(None)
+    assert {"source_type": {"$in": list(ms.SOURCE_TYPES)}} in d["$and"]
+    assert {"superseded": {"$ne": True}} in d["$and"]
+    # explicit widening yields no clause
+    assert ms.build_where({"any_source": True, "include_superseded": True}) is None
+
+
+def test_build_where_variants():
+    w = ms.build_where({"source_type": "user", "min_confidence": 0.8})
+    assert {"source_type": {"$eq": "user"}} in w["$and"] and {"confidence": {"$gte": 0.8}} in w["$and"]
+    # superseded exclusion is added unless opted out
+    assert {"superseded": {"$ne": True}} in w["$and"]
+
+
+def test_matches_filters_parity():
+    llm = ms.build_metadata("maybe", "model", "llm")          # 0.55
+    gs = ms.build_metadata("fact", "freeciv", "game_state")    # 1.0
+    assert ms.matches_filters(llm, {"min_confidence": 0.5})
+    assert not ms.matches_filters(llm, {"min_confidence": 0.7})
+    assert ms.matches_filters(gs, {"source_type": "game_state"})
+    assert not ms.matches_filters(gs, {"source_type": "llm"})
+    assert ms.matches_filters(gs, {"min_confidence": 0.9, "source_type": "game_state"})
+
+
+def test_matches_filters_default_scoping():
+    # default scoping excludes non-provenance records (no/unknown source_type)
+    assert not ms.matches_filters({"source_type": "hash"})
+    assert not ms.matches_filters({})
+    # and excludes superseded unless asked
+    sup = ms.build_metadata("old", "s", "game_state"); sup["superseded"] = True
+    assert not ms.matches_filters(sup)
+    assert ms.matches_filters(sup, {"include_superseded": True})
+
+
+def test_rag_style_metadata_conforms():
+    # RAG chunks (issue blocking-2) now build via build_metadata -> full schema.
+    meta = {**ms.build_metadata(claim="Intro > Setup", source="readme.md",
+                                source_type="knowledge_prior", confidence=ms.KNOWLEDGE_PRIOR_CONFIDENCE),
+            "breadcrumb": "Intro > Setup", "type": "chunk", "time": "knowledge_prior"}
+    assert ms.validate_metadata(meta) is None
+    for k in ("claim", "created_at", "atoms_json", "superseded", "source_type", "confidence"):
+        assert k in meta
+
+
+# NOTE: the action-protocol round-trip test for the remember-claim / query-claims tools lives
+# with the action_protocol module (not part of this memory-provenance change).
+
+# --- chroma-backed store (in-memory; skip without chromadb) --------------
+
+_coll_counter = [0]
+
+
+def _ephemeral_collection():
+    # Unique collection name per call: EphemeralClient instances can share in-process
+    # state, so reusing one name would leak documents across tests.
+    import chromadb
+    _coll_counter[0] += 1
+    client = chromadb.EphemeralClient()
+    return client.get_or_create_collection(name=f"test_claims_{_coll_counter[0]}", embedding_function=None)
+
+
+def _with_ephemeral(fn):
+    """Run fn() with memory_schema._collection pointed at an in-memory collection.
+
+    Gated on chromadb presence only (no pytest dependency), so it RUNS standalone
+    in-container (chromadb present) and skips on hosts without it.
+    """
+    if not _HAS_CHROMA:
+        try:
+            import pytest
+            pytest.skip("chromadb not installed")
+        except ImportError:
+            raise _ChromaSkip("chromadb not installed")
+    coll = _ephemeral_collection()
+    saved = ms._collection
+    ms._collection = lambda: coll
+    try:
+        return fn(coll)
+    finally:
+        ms._collection = saved
+
+
+def test_remember_claim_and_query_returns_provenance():
+    def body(coll):
+        cid = ms.remember_claim("City A has low food", [1.0, 0.0, 0.0], "game_state",
+                                source="freeciv.turn_42", turn_id=42, atoms=["(Inheritance CityA LowFood)"])
+        assert cid.startswith("claim_")
+        res = ms.query_claims([1.0, 0.0, 0.0], n=5)
+        assert res and res[0]["document"] == "City A has low food"
+        meta = res[0]["metadata"]
+        assert meta["source_type"] == "game_state" and meta["confidence"] == 1.0
+        assert meta["source"] == "freeciv.turn_42" and meta["turn_id"] == 42
+    _with_ephemeral(body)
+
+
+def test_min_confidence_filter_excludes_low_confidence():
+    def body(coll):
+        ms.remember_claim("trusted game fact", [1.0, 0.0, 0.0], "game_state", source="g")
+        ms.remember_claim("shaky llm guess", [0.0, 1.0, 0.0], "llm", source="m")  # 0.55
+        hi = ms.query_claims([0.0, 1.0, 0.0], n=5, filters={"min_confidence": 0.8})
+        docs = [r["document"] for r in hi]
+        assert "shaky llm guess" not in docs       # filtered out by confidence
+        assert "trusted game fact" in docs
+    _with_ephemeral(body)
+
+
+def test_supersession_excludes_old_record():
+    def body(coll):
+        old = ms.remember_claim("City A food unknown (pending)", [1.0, 0.0, 0.0], "game_state",
+                                source="freeciv.turn_41", turn_id=41)
+        # a later claim supersedes the old one
+        ms.remember_claim("City A has low food", [1.0, 0.0, 0.0], "game_state",
+                          source="freeciv.turn_42", turn_id=42, supersedes=old)
+        docs = [r["document"] for r in ms.query_claims([1.0, 0.0, 0.0], n=5)]
+        assert "City A has low food" in docs
+        assert "City A food unknown (pending)" not in docs        # superseded -> excluded by default
+        # opt back in
+        docs_all = [r["document"] for r in ms.query_claims([1.0, 0.0, 0.0], n=5, filters={"include_superseded": True})]
+        assert "City A food unknown (pending)" in docs_all
+    _with_ephemeral(body)
+
+
+def test_query_claims_excludes_non_provenance_records():
+    def body(coll):
+        # a bare doc lacking provenance (like a legacy memory / hash sentinel)
+        coll.upsert(ids=["bare1"], embeddings=[[1.0, 0.0, 0.0]], documents=["legacy memory"],
+                    metadatas=[{"type": "hash", "source": "x"}])
+        ms.remember_claim("a real claim", [1.0, 0.0, 0.0], "game_state", source="g")
+        docs = [r["document"] for r in ms.query_claims([1.0, 0.0, 0.0], n=5)]
+        assert "a real claim" in docs
+        assert "legacy memory" not in docs                        # no source_type -> excluded
+    _with_ephemeral(body)
+
+
+def test_source_type_filter():
+    def body(coll):
+        ms.remember_claim("a game fact", [1.0, 0.0, 0.0], "game_state", source="g")
+        ms.remember_claim("a user fact", [0.0, 1.0, 0.0], "user", source="u")
+        only_user = ms.query_claims([1.0, 0.0, 0.0], n=5, filters={"source_type": "user"})
+        assert [r["metadata"]["source_type"] for r in only_user] == ["user"] or all(
+            r["metadata"]["source_type"] == "user" for r in only_user)
+    _with_ephemeral(body)
+
+
+def test_query_claims_text_formats_provenance():
+    def body(coll):
+        ms.remember_claim("formatted fact", [1.0, 0.0, 0.0], "game_state", source="g")
+        text = ms.query_claims_text([1.0, 0.0, 0.0], n=5)
+        assert "formatted fact" in text and "source_type=game_state" in text and "confidence=1.0" in text
+    _with_ephemeral(body)
+
+
+def _run_standalone():
+    failures = 0
+    for name, fn in sorted(globals().items()):
+        if name.startswith("test_") and callable(fn):
+            try:
+                fn()
+                print(f"PASS {name}")
+            except BaseException as exc:  # pytest Skipped subclasses BaseException
+                if exc.__class__.__name__ in ("Skipped", "_ChromaSkip") or isinstance(exc, (ImportError, _ChromaSkip)) or "chromadb" in str(exc):
+                    print(f"SKIP {name} (no chromadb)")
+                    continue
+                if isinstance(exc, AssertionError):
+                    failures += 1
+                    print(f"FAIL {name}: {exc}")
+                else:
+                    failures += 1
+                    print(f"ERROR {name}: {exc!r}")
+    if failures:
+        print(f"\n{failures} test(s) failed")
+        sys.exit(1)
+    print("\nmemory_schema tests passed (chroma-backed skipped without chromadb)")
+
+
+if __name__ == "__main__":
+    _run_standalone()
diff --git a/lib_omegaclaw.metta b/lib_omegaclaw.metta
index 883f7e59..c69b4f40 100644
--- a/lib_omegaclaw.metta
+++ b/lib_omegaclaw.metta
@@ -10,6 +10,7 @@
 !(import! &self (library OmegaClaw-Core ./src/utils))
 !(import! &self (library OmegaClaw-Core ./src/plugin))
 !(import! &self (library OmegaClaw-Core ./src/helper.py))
+!(import! &self (library OmegaClaw-Core ./src/memory_schema.py))
 !(import! &self (library OmegaClaw-Core ./src/agentverse.py))
 !(import! &self (library OmegaClaw-Core ./src/channels))
 !(import! &self (library OmegaClaw-Core ./profile/policy))
diff --git a/src/memory.metta b/src/memory.metta
index 89007afd..1a0b5bf1 100644
--- a/src/memory.metta
+++ b/src/memory.metta
@@ -63,5 +63,15 @@
 (= (query $str)
    (py-call (lib_chromadb.query (embed $str) (maxRecallItems))))
 
+;Provenance-aware memory (Issue #5). remember/query above are unchanged.
+;remember-claim records a provenance-tagged claim; emitted by the agent it is an
+;`llm` source (confidence 0.55). query-claims recalls claims with inline provenance.
+(= (remember-claim $claim)
+   (progn (py-call (memory_schema.remember_claim_llm $claim (embed $claim)))
+          REMEMBER-CLAIM-SUCCESS))
+
+(= (query-claims $str)
+   (py-call (memory_schema.query_claims_text (embed $str) (maxRecallItems))))
+
 (= (episodes $time)
    (py-call (helper.around_time $time (maxEpisodeRecallLines))))
diff --git a/src/memory_schema.py b/src/memory_schema.py
new file mode 100644
index 00000000..44694cbf
--- /dev/null
+++ b/src/memory_schema.py
@@ -0,0 +1,309 @@
+"""Provenance-aware memory schema (Issue #5).
+
+A common metadata shape for long-term memories and knowledge-prior chunks so
+retrieved facts carry *source*, *source_type*, *confidence*, *timestamp*,
+*session/turn*, and (optionally) related symbolic *atoms* — for auditability and
+benchmark reliability.
+
+Split into:
+
+* **Pure helpers** (no chromadb dependency, host-testable): build/validate the
+  metadata, default confidences per source type, and build/evaluate retrieval
+  filters.
+* **Chroma-backed store** (reuses ``rag._get_collection()``): ``remember_claim`` /
+  ``query_claims``. Imported lazily so the pure helpers work on hosts without
+  chromadb installed.
+
+The existing ``remember``/``query`` skills (backed by the external
+``petta_lib_chromadb`` library) are unchanged; this adds a parallel, provenance-
+aware path writing to the same ``memories`` collection, so structured claims are
+still recalled by the normal similarity ``query`` while ``query_claims`` adds
+metadata filtering.
+
+Note: Chroma metadata values must be scalars (str/int/float/bool), so ``atoms``
+(a list) is stored as a JSON string in ``atoms_json``.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import re
+import time
+
+SOURCE_TYPES = ("game_state", "user", "llm", "knowledge_prior", "tool_result")
+
+# Deterministic game-state facts are fully trusted; LLM guesses are discounted.
+DEFAULT_CONFIDENCE = {
+    "game_state": 1.0,
+    "tool_result": 0.9,
+    "user": 0.8,
+    "knowledge_prior": 0.7,
+    "llm": 0.55,
+}
+
+# Knowledge-prior chunks indexed by rag.py use this confidence.
+KNOWLEDGE_PRIOR_CONFIDENCE = DEFAULT_CONFIDENCE["knowledge_prior"]
+
+_PROVENANCE_FIELDS = ("source", "source_type", "confidence", "created_at")
+
+
+def _now_iso():
+    return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
+
+
+def default_confidence(source_type):
+    return DEFAULT_CONFIDENCE.get(source_type, 0.55)
+
+
+def build_metadata(claim, source, source_type, confidence=None, session_id="",
+                   turn_id=None, atoms=None, supersedes=None, created_at=None):
+    """Build a chroma-safe (scalar-valued) provenance metadata dict.
+
+    Confidence defaults by ``source_type`` (game_state→1.0, llm→0.55, …). ``atoms``
+    (a list of symbolic atom strings) is serialized to ``atoms_json``.
+    """
+    if confidence is None:
+        confidence = default_confidence(source_type)
+    meta = {
+        "claim": claim,
+        "source": source,
+        "source_type": source_type,
+        "confidence": float(confidence),
+        "created_at": created_at or _now_iso(),
+        "session_id": session_id or "",
+        "atoms_json": json.dumps(list(atoms) if atoms else [], ensure_ascii=False),
+        "supersedes": supersedes or "",
+        # Set True on a record when a later claim supersedes it; excluded from
+        # query_claims by default (see build_where).
+        "superseded": False,
+    }
+    if turn_id is not None:
+        meta["turn_id"] = int(turn_id)
+    return meta
+
+
+def validate_metadata(meta):
+    """Return an error string if ``meta`` is not a valid provenance record, else None."""
+    if not isinstance(meta, dict):
+        return "metadata is not a mapping"
+    if not meta.get("claim"):
+        return "missing 'claim'"
+    if not meta.get("source"):
+        return "missing 'source'"
+    st = meta.get("source_type")
+    if st not in SOURCE_TYPES:
+        return f"unknown source_type {st!r} (allowed: {list(SOURCE_TYPES)})"
+    conf = meta.get("confidence")
+    if not isinstance(conf, (int, float)) or not (0.0 <= float(conf) <= 1.0):
+        return f"confidence {conf!r} must be a number in [0, 1]"
+    return None
+
+
+def claim_id(meta):
+    """Deterministic, stable id so re-remembering an identical claim is idempotent."""
+    digest = hashlib.sha1((meta.get("claim") or "").encode("utf-8")).hexdigest()[:10]
+    turn = meta.get("turn_id", "")
+    src = re.sub(r"[^A-Za-z0-9_.-]", "_", str(meta.get("source") or "src"))
+    return f"claim_{src}_{turn}_{digest}"
+
+
+def build_where(filters=None):
+    """Build a Chroma ``where`` clause from a filter dict.
+
+    Supported filters: ``source_type``, ``min_confidence``, ``session_id``,
+    ``turn_id``, ``since``/``until`` (ISO ``created_at`` bounds).
+
+    Two scoping clauses are applied **by default** (the reviewer-flagged behavior):
+
+    * **provenance scoping** — only records whose ``source_type`` is one of the known
+      provenance types are matched, so legacy plain ``remember`` memories and hash
+      sentinels (which lack ``source_type``) are excluded. Pass ``any_source=True`` to
+      disable.
+    * **supersession exclusion** — records marked ``superseded`` are excluded. Pass
+      ``include_superseded=True`` to disable.
+
+    Always returns a clause (the defaults are non-empty).
+    """
+    filters = filters or {}
+    clauses = []
+
+    if filters.get("source_type"):
+        clauses.append({"source_type": {"$eq": filters["source_type"]}})
+    elif not filters.get("any_source"):
+        clauses.append({"source_type": {"$in": list(SOURCE_TYPES)}})
+
+    if not filters.get("include_superseded"):
+        clauses.append({"superseded": {"$ne": True}})
+
+    if filters.get("min_confidence") is not None:
+        clauses.append({"confidence": {"$gte": float(filters["min_confidence"])}})
+    if filters.get("session_id"):
+        clauses.append({"session_id": {"$eq": filters["session_id"]}})
+    if filters.get("turn_id") is not None:
+        clauses.append({"turn_id": {"$eq": int(filters["turn_id"])}})
+    if filters.get("since"):
+        clauses.append({"created_at": {"$gte": filters["since"]}})
+    if filters.get("until"):
+        clauses.append({"created_at": {"$lte": filters["until"]}})
+
+    if not clauses:
+        return None
+    if len(clauses) == 1:
+        return clauses[0]
+    return {"$and": clauses}
+
+
+def matches_filters(meta, filters=None):
+    """Host-side mirror of :func:`build_where` for a single metadata record."""
+    filters = filters or {}
+    if filters.get("source_type"):
+        if meta.get("source_type") != filters["source_type"]:
+            return False
+    elif not filters.get("any_source"):
+        if meta.get("source_type") not in SOURCE_TYPES:
+            return False
+    if not filters.get("include_superseded") and meta.get("superseded") is True:
+        return False
+    if filters.get("min_confidence") is not None and float(meta.get("confidence", 0)) < float(filters["min_confidence"]):
+        return False
+    if filters.get("session_id") and meta.get("session_id") != filters["session_id"]:
+        return False
+    if filters.get("turn_id") is not None and meta.get("turn_id") != int(filters["turn_id"]):
+        return False
+    if filters.get("since") and str(meta.get("created_at", "")) < filters["since"]:
+        return False
+    if filters.get("until") and str(meta.get("created_at", "")) > filters["until"]:
+        return False
+    return True
+
+
+# --- chroma-backed store (reuses rag.py's client) -------------------------
+
+def _collection():
+    try:
+        from rag import _get_collection
+    except ImportError:  # pragma: no cover - alternate import path
+        from src.rag import _get_collection
+    return _get_collection()
+
+
+def remember_claim(claim, embedding, source_type, source=None, confidence=None,
+                   session_id="", turn_id=None, atoms=None, supersedes=None):
+    """Write a structured, provenance-tagged claim to the shared memories collection.
+
+    ``embedding`` is computed MeTTa-side (``(embed $claim)``), mirroring ``remember``.
+    Returns the stable claim id.
+    """
+    meta = build_metadata(
+        claim, source or f"{source_type}:adhoc", source_type, confidence=confidence,
+        session_id=session_id, turn_id=turn_id, atoms=atoms, supersedes=supersedes,
+    )
+    err = validate_metadata(meta)
+    if err:
+        raise ValueError(f"invalid claim metadata: {err}")
+    cid = claim_id(meta)
+    coll = _collection()
+    coll.upsert(ids=[cid], embeddings=[embedding], documents=[claim], metadatas=[meta])
+    # If this claim supersedes an earlier record, mark that record superseded so it is
+    # excluded from default recall. Best-effort: ignore if the id is absent.
+    if supersedes:
+        try:
+            coll.update(ids=[supersedes], metadatas=[{"superseded": True}])
+        except Exception as exc:  # pragma: no cover - defensive
+            print(f"[memory_schema] WARNING could not mark superseded id={supersedes}: {exc}", flush=True)
+    print(f"[memory_schema] REMEMBER_CLAIM id={cid} source_type={source_type} confidence={meta['confidence']}", flush=True)
+    return cid
+
+
+def query_claims(embedding, n=5, filters=None):
+    """Similarity query returning provenance metadata.
+
+    Returns a list of ``{document, metadata, distance}``. By default only
+    provenance-bearing, non-superseded records are returned (see :func:`build_where`),
+    so legacy memories / hash sentinels and superseded claims are excluded. Pass
+    ``filters={"any_source": True}`` / ``{"include_superseded": True}`` to widen.
+    """
+    where = build_where(filters)
+    kwargs = {"query_embeddings": [embedding], "n_results": int(n)}
+    if where is not None:
+        kwargs["where"] = where
+    res = _collection().query(**kwargs)
+    docs = (res.get("documents") or [[]])[0]
+    metas = (res.get("metadatas") or [[]])[0]
+    dists = (res.get("distances") or [[]])[0]
+    out = []
+    for i, doc in enumerate(docs):
+        out.append({
+            "document": doc,
+            "metadata": metas[i] if i < len(metas) else {},
+            "distance": dists[i] if i < len(dists) else None,
+        })
+    return out
+
+
+def query_claims_text(embedding, n=5):
+    """MeTTa-facing recall: a readable string with inline provenance per result."""
+    results = query_claims(embedding, n=n, filters=None)
+    if not results:
+        return "NO_CLAIMS_FOUND"
+    lines = []
+    for r in results:
+        m = r.get("metadata") or {}
+        lines.append(
+            f"- {r.get('document')} "
+            f"[source_type={m.get('source_type', '?')} confidence={m.get('confidence', '?')} "
+            f"source={m.get('source', '?')}]"
+        )
+    return "\n".join(lines)
+
+
+def remember_claim_llm(claim, embedding):
+    """MeTTa-facing wrapper: a claim the agent itself asserts is an ``llm`` source
+    (discounted confidence 0.55). Programmatic producers call ``remember_claim`` with
+    a higher-trust source_type (game_state/tool_result/...)."""
+    return remember_claim(claim, embedding, "llm")
+
+
+def _selftest():
+    # game_state -> 1.0, llm -> 0.55
+    m = build_metadata("City A low food", "freeciv.turn_42", "game_state")
+    assert m["confidence"] == 1.0 and m["source_type"] == "game_state"
+    assert validate_metadata(m) is None
+    ml = build_metadata("maybe X", "model", "llm")
+    assert ml["confidence"] == 0.55
+
+    # atoms serialized to JSON string (chroma scalar constraint)
+    ma = build_metadata("c", "s", "user", atoms=["(Inheritance A B)", "(stv 1 0.9)"])
+    assert isinstance(ma["atoms_json"], str) and json.loads(ma["atoms_json"])[0] == "(Inheritance A B)"
+
+    # validation failures
+    assert validate_metadata({"claim": "c", "source": "s", "source_type": "bogus", "confidence": 1.0})
+    assert validate_metadata({"claim": "c", "source": "s", "source_type": "user", "confidence": 2.0})
+    assert validate_metadata({"claim": "", "source": "s", "source_type": "user", "confidence": 1.0})
+
+    # stable id is deterministic
+    assert claim_id(m) == claim_id(build_metadata("City A low food", "freeciv.turn_42", "game_state"))
+
+    # where builder: defaults scope to provenance + non-superseded
+    d = build_where(None)
+    assert {"source_type": {"$in": list(SOURCE_TYPES)}} in d["$and"]
+    assert {"superseded": {"$ne": True}} in d["$and"]
+    w = build_where({"source_type": "user", "min_confidence": 0.8})
+    assert {"source_type": {"$eq": "user"}} in w["$and"] and {"confidence": {"$gte": 0.8}} in w["$and"]
+    # explicit widening
+    assert build_where({"any_source": True, "include_superseded": True}) is None
+
+    # matches_filters parity (default provenance + supersession scoping)
+    assert matches_filters(m)                                  # provenance-bearing, not superseded
+    assert not matches_filters({"source_type": "hash"})        # not a provenance source -> excluded
+    superseded = build_metadata("old", "s", "game_state"); superseded["superseded"] = True
+    assert not matches_filters(superseded) and matches_filters(superseded, {"include_superseded": True})
+    assert matches_filters(ml, {"min_confidence": 0.5}) and not matches_filters(ml, {"min_confidence": 0.7})
+    assert matches_filters(m, {"source_type": "game_state"}) and not matches_filters(m, {"source_type": "llm"})
+
+    print("memory_schema self-tests passed")
+
+
+if __name__ == "__main__":
+    _selftest()
diff --git a/src/rag.py b/src/rag.py
index 76da6974..39faed8a 100644
--- a/src/rag.py
+++ b/src/rag.py
@@ -5,6 +5,10 @@
 import chromadb
 import openai
 from   lib_llm_ext import initLocalEmbedding, useLocalEmbedding
+try:  # provenance schema (Issue #5) — robust under flat or repo-root package import
+    from memory_schema import KNOWLEDGE_PRIOR_CONFIDENCE, build_metadata
+except ImportError:  # pragma: no cover - package-style import path
+    from src.memory_schema import KNOWLEDGE_PRIOR_CONFIDENCE, build_metadata
 from src.logger import get_logger
 
 logger = get_logger(__name__)
@@ -262,12 +266,18 @@ def init_knowledge(embedding_selection):
 
             # Store chunks
             ids = [f"{filename}_chunk_{i}" for i in range(len(chunks))]
+            # Provenance schema (Issue #5): each chunk carries the full common schema
+            # (claim/source/source_type/confidence/created_at/atoms_json/supersedes/superseded)
+            # via build_metadata, plus the legacy breadcrumb/type/time keys for back-compat.
             metadatas = [
                 {
-                    "source": filename,
+                    **build_metadata(
+                        claim=c["breadcrumb"], source=filename, source_type="knowledge_prior",
+                        confidence=KNOWLEDGE_PRIOR_CONFIDENCE,
+                    ),
                     "breadcrumb": c["breadcrumb"],
                     "type": "chunk",
-                    "time": "knowledge_prior"
+                    "time": "knowledge_prior",
                 }
                 for c in chunks
             ]