feat: record compiler version provenance

server/services/compilers/heuristic.py

@@ -21,6 +21,19 @@
 from server.services.memory_ttl import compute_valid_to
 
 
+_HEURISTIC_COMPILER_REVISION = "1"
+
+
+def _heuristic_compiler_metadata(extra: Mapping[str, Any] | None = None) -> dict[str, Any]:
+    """Provenance for memories produced by the heuristic compiler."""
+    metadata: dict[str, Any] = {
+        "compiler": "heuristic",
+        "compiler_version": f"heuristic:{_HEURISTIC_COMPILER_REVISION}",
+    }
+    if extra:
+        metadata.update(extra)
+    return metadata
+
 class HeuristicCompiler:
     """Pattern-based memory compiler. Implements BaseCompiler protocol."""
 
@@ -70,7 +83,7 @@ def _compile_episode(
                 valid_from=ep_valid_from,
                 valid_to=compute_valid_to("episode_summary", ep_valid_from, ttl),
                 source_episode_ids=[ep.id],
-                metadata_={},
+                metadata_=_heuristic_compiler_metadata(),
                 status="active",
             )
         )
@@ -90,7 +103,7 @@ def _compile_episode(
                     valid_from=ep_valid_from,
                     valid_to=compute_valid_to("profile_fact", ep_valid_from, ttl),
                     source_episode_ids=[ep.id],
-                    metadata_=claim_md or {},
+                    metadata_=_heuristic_compiler_metadata(claim_md),
                     status="active",
                 )
             )

server/services/compilers/llm.py

@@ -45,6 +45,7 @@
 from server.services.llm import LLMResponseError
 from server.services.compilers.heuristic import episode_valid_from, extract_payload_text
 from server.services.memory_ttl import compute_valid_to
+from server.services.compilers.versioning import stable_version
 
 logger = structlog.stdlib.get_logger()
 
@@ -296,6 +297,46 @@ def _system_prompt() -> str:
     return _SYSTEM_PROMPT_COMPOSE if settings.compile_compose_unit else _SYSTEM_PROMPT
 
 
+def _prompt_version(*prompts: str) -> str:
+    """Stable fingerprint of the prompt templates that produced a memory."""
+    return stable_version(*prompts)
+
+
+def _llm_compiler_metadata(model: str, *, recall_sweep: bool = False) -> dict[str, str]:
+    """Provenance for the LLM compiler path that produced a memory.
+
+    First-pass memories depend on the active extraction prompt. Recall-sweep
+    memories additionally depend on the sweep prompt and on the first-pass
+    output supplied to that sweep, so their version covers both templates.
+    """
+    main_prompt = _system_prompt()
+
+    if recall_sweep:
+        prompt_version = _prompt_version(main_prompt, _RECALL_SWEEP_PROMPT)
+        pass_name = "recall_sweep"
+    else:
+        prompt_version = _prompt_version(main_prompt)
+        pass_name = "extract"
+
+    compiler_version = stable_version(
+        "llm",
+        model,
+        pass_name,
+        prompt_version,
+    )
+
+    metadata = {
+        "compiler": "llm",
+        "model": model,
+        "prompt_version": prompt_version,
+        "compiler_version": compiler_version,
+    }
+    if recall_sweep:
+        metadata["pass"] = "recall_sweep"
+
+    return metadata
+
+
 def _window_text(
     text: str, max_total: int, window: int, overlap: int
 ) -> list[str]:
@@ -747,6 +788,11 @@ async def _process_batch(
                 except Exception:  # pragma: no cover - defensive; never break compile
                     logger.warning("recall_sweep_failed", exc_info=True)
 
+            # The first-pass prompt determines every regular extraction. Recall
+            # memories additionally depend on the sweep prompt and on the
+            # first-pass output shown to that sweep, so its version covers both
+            # prompt templates.
+
             # Map memories back to their source episodes
             results: list[MemoryRow] = []
             for mem in raw_memories:
@@ -785,17 +831,22 @@ async def _process_batch(
                     summary = str(raw_summary)[:200] if raw_summary else content[:200]
 
                 ep_valid_from = episode_valid_from(source_ep)
+
+                is_recall_sweep = bool(mem.get("_recall_sweep"))
+                metadata: dict[str, Any] = _llm_compiler_metadata(
+                    self._model,
+                    recall_sweep=is_recall_sweep,
+                )
+
                 # Optional, validated claim. A malformed/unknown proposal is
                 # dropped and never fails the rest of the response.
-                metadata: dict[str, Any] = {"compiler": "llm", "model": self._model}
-                if mem.get("_recall_sweep"):
-                    metadata["pass"] = "recall_sweep"
                 try:
                     claim_md = _llm_claim_metadata(mem, default_valid_from=ep_valid_from)
                 except Exception:  # pragma: no cover - defensive; never break compile
                     claim_md = None
                 if claim_md:
                     metadata.update(claim_md)
+
                 results.append(
                     MemoryRow(
                         id=uuid.uuid4(),
@@ -812,7 +863,11 @@ async def _process_batch(
                     )
                 )
 
-            logger.info("llm_batch_done", episodes=len(batch), memories_extracted=len(results))
+            logger.info(
+                "llm_batch_done",
+                episodes=len(batch),
+                memories_extracted=len(results),
+            )
             return results
 
     async def _call_llm_async(self, text: str, episode_count: int) -> list[dict[str, Any]]:

server/services/compilers/versioning.py

@@ -0,0 +1,19 @@
+"""Compiler-version helpers shared by memory compilers."""
+
+from __future__ import annotations
+
+import hashlib
+
+
+def stable_version(*parts: str) -> str:
+    """Return a deterministic fingerprint for compiler-version components.
+
+    Each part is length-prefixed before hashing so different component
+    boundaries cannot produce the same payload.
+    """
+    digest = hashlib.sha256()
+    for part in parts:
+        encoded = part.encode("utf-8")
+        digest.update(len(encoded).to_bytes(8, byteorder="big"))
+        digest.update(encoded)
+    return f"sha256:{digest.hexdigest()}"

server/services/structured.py

@@ -52,6 +52,7 @@
 
 CANDIDATE_CONTAINER_KEY = "statewave"
 CANDIDATE_LIST_KEY = "memory_candidates"
+_STRUCTURED_COMPILER_VERSION = "structured:1"
 
 # Candidate kind -> stored MemoryKind. No enum change: "domain_fact" maps to the
 # existing ``profile_fact`` kind. Unknown kinds make the container non-acceptable
@@ -165,7 +166,15 @@ def compile_candidates(
     for c in cands:
         kind = _KIND_MAP[c["kind"].strip().lower()]
         text = c["text"]
+
         metadata: dict[str, Any] = dict(c.get("metadata") or {})
+        metadata.update(
+            {
+            "compiler": "structured",
+            "compiler_version": _STRUCTURED_COMPILER_VERSION,
+            }
+        )
+
         claim_md = _candidate_claim_metadata(
             c.get("claim"), default_valid_from=vf, claim_keys=claim_keys
         )

tests/test_compiler_claims.py

@@ -95,10 +95,15 @@ def test_heuristic_claim_anchor_is_the_episode_time_not_invented_content():
     ],
 )
 def test_heuristic_emits_fact_without_claim(text):
-    # The memory is still extracted (behavior unchanged); it just carries no claim.
+    # The memory is still extracted; it carries compiler provenance but no claim.
     facts = _facts(text)
     assert facts, "expected the fact to still be emitted"
-    assert all(f.metadata_ == {} for f in facts)
+    assert all("claim" not in f.metadata_ for f in facts)
+    assert all(f.metadata_.get("compiler") == "heuristic" for f in facts)
+    assert all(
+        f.metadata_.get("compiler_version") == "heuristic:1"
+        for f in facts
+    )
 
 
 @pytest.mark.parametrize(
@@ -120,9 +125,11 @@ def test_heuristic_generic_stripe_is_not_a_billing_claim():
 
 
 def test_heuristic_preserves_unrelated_nothing_lost():
-    # No claim case keeps metadata exactly {} (no spurious keys).
+    # No claim is invented; compiler provenance is still recorded.
     facts = _facts("I use Stripe")
-    assert facts and all(f.metadata_ == {} for f in facts)
+    assert facts
+    assert all("claim" not in f.metadata_ for f in facts)
+    assert all(f.metadata_.get("compiler") == "heuristic" for f in facts)
 
 
 # --- LLM claim validation (registry is authoritative, model is untrusted) -----