Make broad Leo answers database-composed (#120)

hermes-agent/leoclean-bin/kb_tool.py

@@ -32,6 +32,7 @@ DIRECT_READBACK_CONTRACTS = frozenset(
         "identity_canonicality_readback",
     }
 )
+SCHEMA_COMPILED_CONTRACTS = frozenset({"forecast_resolution_schema", "claim_supersession_schema"})
 DECISION_MATRIX_TABLES = tuple(
     f"{schema}.{table}"
     for schema in ("public", "kb_stage")
@@ -344,6 +345,16 @@ def with_proposal_readiness(proposal: dict[str, Any]) -> dict[str, Any]:
     return {**proposal, "readiness": classify_proposal_readiness(proposal)}
 
 
+def _has_unnegated_action(query: str, actions: tuple[str, ...]) -> bool:
+    for segment in re.split(r"(?<=[.!?])\s+|\n+", query.lower()):
+        if not any(re.search(rf"\b{re.escape(action)}\b", segment) for action in actions):
+            continue
+        if "read-only" in segment or re.search(r"\b(?:do not|don't|dont|must not|without)\b", segment):
+            continue
+        return True
+    return False
+
+
 def operational_contracts(
     query: str,
     *,
@@ -372,11 +383,29 @@ def operational_contracts(
         any(term in lowered for term in ("proposal", "pending", "approved", "stuck", "backlog"))
         and any(
             term in lowered
-            for term in ("document", "source row", "source_ref", "source ref", "pointer", "pointed", "evidence link")
+            for term in (
+                "document",
+                "attachment",
+                "extracted text",
+                "source row",
+                "source_ref",
+                "source ref",
+                "pointer",
+                "pointed",
+                "evidence link",
             )
-        and any(term in lowered for term in ("because", "wrong", "missing", "stuck", "link", "point", "cause"))
         )
-    kb_scope = "knowledge base" in lowered or bool(re.search(r"\bkb\b", lowered)) or "canonical knowledge" in lowered
+        and any(
+            term in lowered
+            for term in ("because", "wrong", "missing", "stuck", "link", "point", "cause", "canonical evidence", "audit")
+        )
+    )
+    kb_scope = (
+        "knowledge base" in lowered
+        or "database" in lowered
+        or bool(re.search(r"\bkb\b", lowered))
+        or "canonical knowledge" in lowered
+    )
     kb_implementation_question = bool(
         re.search(
             r"\b(?:scorer|benchmark|test harness|plugin|parser|implementation|search|dashboard|api|script|"
@@ -384,16 +413,19 @@ def operational_contracts(
             lowered,
         )
     )
+    forecast_resolution_question = "forecast" in lowered and any(
+        term in lowered for term in ("resolution", "criteria", "event is now over", "event is over", "60%")
+    )
     proposal_lifecycle_question = any(
-        term in lowered for term in ("proposal", "proposed", "pending", "approved", "applied", "staged")
+        term in lowered
+        for term in ("proposal", "proposed", "pending", "approved", "applied", "staged", "reviewer approval")
     )
-    broad_kb_change_question = any(
+    broad_kb_change_question = _has_unnegated_action(
-        term in lowered for term in ("changed", "change", "updated", "update", "landed", "live")
+        query, ("change", "changed", "update", "updated", "landed")
     )
-    proposal_state_question = (
+    proposal_state_question = kb_scope and not forecast_resolution_question and (
-        "demo" not in lowered
+        proposal_lifecycle_question
-        and kb_scope
+        or ("demo" not in lowered and broad_kb_change_question and not kb_implementation_question)
-        and (proposal_lifecycle_question or (broad_kb_change_question and not kb_implementation_question))
     )
     named_packet_question = "helmer" in lowered and any(
         term in lowered for term in ("7 powers", "seven powers", "powers framework", "powers packet")
@@ -414,8 +446,11 @@ def operational_contracts(
             )
         )
         or (
-            (kb_scope or "database" in lowered)
+            kb_scope
-            and any(term in lowered for term in ("write", "apply", "mutate", "stage"))
+            and (
+                any(term in lowered for term in ("learned", "updated", "update", "live", "current", "working"))
+                or _has_unnegated_action(query, ("write", "apply", "mutate", "stage"))
+            )
         )
     )
     identity_question = any(term in lowered for term in ("soul.md", "soul file")) and any(
@@ -428,9 +463,9 @@ def operational_contracts(
                 ("decision_matrix_readback", decision_matrix_question),
                 ("source_link_audit", source_link_audit_question),
                 ("named_packet_readback", named_packet_question),
-                ("demo_capability_readback", demo_question),
                 ("identity_canonicality_readback", identity_question),
                 ("proposal_state_readback", proposal_state_question),
+                ("demo_capability_readback", demo_question),
             )
             if enabled
         ),
@@ -483,6 +518,7 @@ def operational_contracts(
         contracts.append(
             {
                 "id": "proposal_state_readback",
+                "approval_only_claim": "reviewer approval" in lowered and broad_kb_change_question,
                 "state_boundary": (
                     "proposal status, reviewer approval, and canonical apply are separate states; applied rows still "
                     "need proposal-level applied_at plus row-level postflight for a specific change claim"
@@ -517,6 +553,9 @@ def operational_contracts(
         contracts.append(
             {
                 "id": "demo_capability_readback",
+                "temporal_comparison_requested": any(
+                    term in lowered for term in ("since yesterday", "since last night", "since earlier")
+                ),
                 "tiers": [
                     "read existing canonical and proposal receipts",
                     "stage a safe pending_review proposal",
@@ -621,6 +660,53 @@ def operational_contracts(
             }
         )
 
+    if forecast_resolution_question:
+        contracts.append(
+            {
+                "id": "forecast_resolution_schema",
+                "current_columns": {
+                    name: list(schema.get(name, ())) for name in ("claims", "claim_edges")
+                },
+                "history_boundary": "preserve the original forecast and its missing resolution criteria",
+                "schema_boundary": (
+                    "do not invent forecast-resolution fields or a resolves edge; stage a reviewed schema proposal"
+                ),
+            }
+        )
+
+    claim_supersession_question = (
+        "claim" in lowered
+        and any(term in lowered for term in ("replacement", "wrong", "retired", "supersed"))
+        and any(term in lowered for term in ("approve_claim", "current v1", "edge field", "old claim"))
+    )
+    if claim_supersession_question:
+        contracts.append(
+            {
+                "id": "claim_supersession_schema",
+                "current_columns": {
+                    name: list(schema.get(name, ())) for name in ("claims", "claim_edges")
+                },
+                "insert_boundary": "approve_claim may insert a replacement claim plus a claim-to-claim edge",
+                "update_boundary": (
+                    "existing claim status and superseded_by updates require a separate reviewed apply capability"
+                ),
+            }
+        )
+
+    telegram_delivery_question = (
+        "telegram" in lowered
+        and any(term in lowered for term in ("gatewayrunner", "temporary-profile", "temporary profile", "posted nothing"))
+        and any(term in lowered for term in ("proven", "delivery", "visible"))
+    )
+    if telegram_delivery_question:
+        contracts.append(
+            {
+                "id": "telegram_delivery_proof",
+                "handler_boundary": "a non-posting GatewayRunner reply proves handler behavior, not Telegram delivery",
+                "receipt": "visible reply in the intended chat with message ID, timestamp, and chat/sender readback",
+            }
+        )
+
     if any(term in lowered for term in ("restart", "prior session", "erased", "answer behavior", "database totals")):
         contracts.append(
             {
@@ -726,13 +812,14 @@ def compile_operational_response(contracts: list[dict[str, Any]]) -> str | None:
         retain_step = safe_steps[0] if safe_steps else "retain the artifact plus its locator and content hash"
         stage_step = safe_steps[1] if len(safe_steps) > 1 else "stage candidates in a pending_review proposal"
         return (
-            f"- {source.get('required_lead')} Immediately {retain_step}. A filename, chat label, attachment reference, or "
+            f"- {source.get('required_lead')} Immediately {retain_step}; retain provenance for each PDF, Telegram "
-            "source_ref is not source identity.\n\n"
+            "message, and tweet. A filename, chat label, attachment reference, or source_ref is not source identity.\n\n"
-            f"- {stage_step}. Deduplicate candidate claims and retain evidence excerpts and contradictions. Do not "
+            f"- {stage_step}. Deduplicate overlapping candidate claims, preserve evidence excerpts, source-specific "
-            "create public.sources or other canonical rows during capture or staging.\n\n"
+            "caveats, uncertainty, and contradictions. Candidates are not canonical until reviewed and applied: create "
-            f"- Explicit approval begins {source.get('approval_begins')}. The receipt should include the retained "
+            "no public.sources or other public.* rows during capture or staging.\n\n"
-            "locator and hash, proposal ID and status, reviewed payload, proposal-level applied_at, created row IDs, "
+            f"- Explicit approval begins {source.get('approval_begins')}. Receipt: locator and content hash, proposal ID/"
-            "and postflight readback. Until apply succeeds, the material is staged rather than live canonical knowledge."
+            "status, reviewed payload, applied_at, created row IDs, and postflight readback. Next proof-changing "
+            "follow-up: run the retained artifact through the staging compiler, then review its typed candidates."
         )
 
     named_packet = by_id.get("named_packet_readback")
@@ -821,14 +908,17 @@ def compile_operational_response(contracts: list[dict[str, Any]]) -> str | None:
     if source_audit:
         audit = source_audit.get("live_link_audit") or {}
         return (
-            f"Not proven.\n{format_database_count_readback(source_audit.get('database_status'))}\nThe live row-link "
+            f"No. The attachment does not itself create canonical evidence; neither does extracted text on disk.\n"
+            f"{format_database_count_readback(source_audit.get('database_status'))}\nThe live row-link "
             f"audit found {audit.get('pending_or_approved', 'unknown')} pending/approved proposals: "
             f"{audit.get('with_source_ref', 'unknown')} have source_ref and "
             f"{audit.get('exact_public_source_matches', 'unknown')} exactly match a public.sources ID, URL, or "
             "storage_path. Pending status therefore is not by itself proof of a bad document pointer.\n\n"
             "Audit telegram_file_refs/raw files -> document_evaluations -> kb_proposals.source_ref -> public.sources "
-            "-> claim_evidence. Next proof-changing follow-up: list each proposal UUID/status/source_ref and exact "
+            "-> claim_evidence. Canonical linkage and provenance quality are separate: a canonical link can still point "
-            "source match, then stage only the missing guarded source/evidence apply contract."
+            "to a weak, untraceable source with no URL/storage_path or raw artifact. Before/after receipt: proposal UUID/"
+            "status/source_ref, source and claim_evidence row IDs, locator/hash, applied_at, and postflight. Next proof-"
+            "changing follow-up: list each broken layer, then stage only the missing guarded source/evidence contract."
         )
 
     identity = by_id.get("identity_canonicality_readback")
@@ -836,10 +926,11 @@ def compile_operational_response(contracts: list[dict[str, Any]]) -> str | None:
         return (
             f"No.\n{format_database_count_readback(identity.get('database_status'))}\nA direct SOUL.md edit is a "
             "runtime/profile change; it does not change canonical Postgres rows in personas, strategies, beliefs, "
-            "strategy_nodes, or strategy_node_anchors. No active general DB-to-SOUL.md renderer automation is "
+            "strategy_nodes, or strategy_node_anchors. A chat correction is also not canonical until it is staged, "
-            "currently proven, so neither a future render nor persistence of the edit should be assumed.\n\n"
+            "reviewed, and applied as exact identity rows. No active general DB-to-SOUL.md renderer automation is "
+            "currently proven, so neither surface should be assumed to survive as identity after restart.\n\n"
             "Next proof-changing follow-up: audit the intended canonical identity rows and proposal UUID/status/"
-            "applied_at, stage or apply the reviewed DB change with authorization, run or verify render/sync, then "
+            "applied_at, stage the exact change, review it, apply it with authorization, run or verify render/sync, then "
             "compare SOUL.md hash/content and restart readback."
         )
 
@@ -848,15 +939,35 @@ def compile_operational_response(contracts: list[dict[str, Any]]) -> str | None:
         status_data = demo.get("database_status") or {}
         states = status_data.get("proposal_status_counts") or {}
         applied_rows = list(demo.get("applied_proposals") or [])
+        approved_rows = [
+            item for item in demo.get("applied_and_approved_proposals") or [] if item.get("status") == "approved"
+        ]
+        approved_needs_payload = sum(
+            (item.get("readiness") or {}).get("review_state") == "approved_needs_apply_payload"
+            for item in approved_rows
+        )
         applied_types = human_join(
             sorted({str(item.get("proposal_type")) for item in applied_rows if item.get("proposal_type")})
         )
+        if demo.get("temporal_comparison_requested"):
             return (
-            f"Partly.\n{format_database_count_readback(status_data)}\nProposal states are applied: "
+                f"Not proven.\nFresh live readback.\n{format_database_count_readback(status_data)}\nProposal states are "
+                f"applied: {states.get('applied', 0)}, approved: {states.get('approved', 0)}, pending_review: "
+                f"{states.get('pending_review', 0)}. This current snapshot does not establish what changed since "
+                f"yesterday. Existing receipts identify applied types {applied_types or 'none'}, but not a time-window "
+                f"delta. Approved is not applied; {approved_needs_payload}/{len(approved_rows)} approved rows are "
+                "approved_needs_apply_payload and not directly applyable.\n\n"
+                "Next proof-changing follow-up: build and review a strict apply_payload for the highest-impact approved "
+                "proposal, then seek explicit apply authorization. Until its applied_at and public.* before/after row "
+                "IDs or hashes exist, do not claim the canonical database changed."
+            )
+        return (
+            f"Partly.\nFresh live readback.\n{format_database_count_readback(status_data)}\nProposal states are applied: "
             f"{states.get('applied', 0)}, approved: {states.get('approved', 0)}, pending_review: "
             f"{states.get('pending_review', 0)}. Existing receipts show applied proposal types: "
             f"{applied_types or 'none identified'}; that receipt artifact does not prove every rich packet or chat "
-            "apply path is live. "
+            f"apply path is live. Approved is not applied. {approved_needs_payload}/{len(approved_rows)} approved rows "
+            "are approved_needs_apply_payload and not directly applyable. "
             "You can safely demo canonical readback and a pending_review staging write. A new canonical apply requires "
             "explicit authorization, a supported strict payload, and postflight row IDs.\n\n"
             "Next proof-changing follow-up: choose the demo tier - existing applied receipt, safe staging canary, or "
@@ -868,16 +979,99 @@ def compile_operational_response(contracts: list[dict[str, Any]]) -> str | None:
     if proposal_state:
         status_data = proposal_state.get("database_status") or {}
         states = status_data.get("proposal_status_counts") or {}
+        approved_rows = [
+            item
+            for item in proposal_state.get("applied_and_approved_proposals") or []
+            if item.get("status") == "approved"
+        ]
+        approved_needs_payload = sum(
+            (item.get("readiness") or {}).get("review_state") == "approved_needs_apply_payload"
+            for item in approved_rows
+        )
+        lead = "No." if proposal_state.get("approval_only_claim") else "Partly."
         return (
-            f"Partly.\n{format_database_count_readback(status_data)}\nProposal states are applied: "
+            f"{lead}\nFresh live readback.\n{format_database_count_readback(status_data)}\nProposal states are applied: "
             f"{states.get('applied', 0)}, approved: {states.get('approved', 0)}, pending_review: "
             f"{states.get('pending_review', 0)}, canceled: {states.get('canceled', 0)}. Approved is not applied, and "
-            "an applied proposal receipt alone does not identify the exact public.* rows changed.\n\n"
+            f"{approved_needs_payload}/{len(approved_rows)} approved rows are approved_needs_apply_payload and not "
+            "directly applyable. An applied proposal receipt alone does not identify the exact public.* rows changed.\n\n"
             "Next proof-changing follow-up: choose the proposal or time window, return its UUID/status/applied_at, "
             "compare the affected public.* row IDs or before/after hashes, and for approved rows normalize and review "
             "the strict payload before requesting explicit guarded-apply authorization."
         )
 
+    runtime = by_id.get("runtime_persistence")
+    if runtime:
+        return (
+            "No. Unchanged database totals do not prove unchanged rows or unchanged answer behavior, and a restart "
+            "does not necessarily erase prior-session facts. Compare row IDs, timestamps, and hashes, not counts alone; "
+            "balanced inserts, updates, or deletes can leave totals unchanged.\n\n"
+            "Persisted or deployed answer inputs include Postgres rows, skills, runtime config, SOUL.md, state.db, and "
+            "session JSONL. state.db/session JSONL can preserve durable session continuity across a restart, while a "
+            "changed skill or runtime artifact can change the answer with identical database totals.\n\n"
+            "Proof tiers: 1) canonical: public.* row/hash readback; 2) handler: the deployed runtime answers the same "
+            "prompt with retained trace; 3) Telegram: a visible reply with message ID, timestamp, and chat/sender "
+            "readback. Only a reviewed apply plus row-level postflight proves a database mutation."
+        )
+
+    shared = by_id.get("shared_claims_agent_positions")
+    if shared:
+        return (
+            "Store one factual claim once as shared knowledge with shared sources/evidence; do not duplicate the claim "
+            "per agent. Record each agent-specific position in public.beliefs with agent_id, statement, confidence, "
+            "falsifier, and status.\n\n"
+            "The current-schema gap matters: public.beliefs has no claim-ID foreign key and there is no belief-edge "
+            "table. claim_edges connects claims only, so do not invent belief-to-claim or belief-to-belief edges. "
+            "Disagreement remains queryable by retrieving the shared factual claim/evidence and semantically matching "
+            "each agent's belief rows; a future reviewed schema proposal can add an explicit link if needed."
+        )
+
+    forecast = by_id.get("forecast_resolution_schema")
+    if forecast:
+        columns = forecast.get("current_columns") or {}
+        claim_columns = human_join([str(item) for item in columns.get("claims") or []])
+        edge_columns = human_join([str(item) for item in columns.get("claim_edges") or []])
+        edge_types = [str(item) for item in forecast.get("current_edge_types") or []]
+        resolves_state = "`resolves` is present" if "resolves" in edge_types else "there is no `resolves` edge type"
+        return (
+            "Preserve the original 60% forecast; do not overwrite it. Because it defined no resolution criteria, its "
+            "outcome remains ambiguous. Leo may add a sourced explanatory claim about the observed event, but must not "
+            "retroactively label the forecast resolved.\n\n"
+            f"Current public.claims columns are {claim_columns}; there is no forecast-resolution or resolved_at field. "
+            f"Current public.claim_edges columns are {edge_columns}; {resolves_state}. Stage a reviewed schema proposal "
+            "for explicit criteria, outcome, resolved_at, resolver/provenance, and a valid link contract. Do not apply or "
+            "invent those fields in v1."
+        )
+
+    delivery = by_id.get("telegram_delivery_proof")
+    if delivery:
+        return (
+            "No. A temporary-profile GatewayRunner response proves handler execution, model/plugin integration, and "
+            "reply construction only. Because it posted nothing, it does not prove Telegram delivery or a Telegram-"
+            "visible reply.\n\n"
+            "The smallest closing test is one explicitly authorized test message in the intended chat, followed by "
+            "Leo's visible reply. Retain a delivery receipt with the outgoing and reply message IDs, timestamps, chat "
+            "ID/title, sender handle, reply linkage, and a post-send readback. Then remove the temporary profile and "
+            "verify no orphan process remains."
+        )
+
+    supersession = by_id.get("claim_supersession_schema")
+    if supersession:
+        columns = supersession.get("current_columns") or {}
+        claim_columns = human_join([str(item) for item in columns.get("claims") or []])
+        edge_columns = human_join([str(item) for item in columns.get("claim_edges") or []])
+        edge_types = [str(item) for item in supersession.get("current_edge_types") or []]
+        supersedes_state = "`supersedes` is valid" if "supersedes" in edge_types else "`supersedes` is not valid"
+        return (
+            f"Current public.claims fields: {claim_columns}. Current claim_edges fields: {edge_columns}; the edge has no "
+            f"rationale field, and {supersedes_state} in the live edge_type enum.\n\n"
+            "approve_claim can insert the replacement claim and a claim-to-claim supersedes edge using from_claim, "
+            "to_claim, edge_type, weight, and created_by. Put the explanation in sourced evidence or an explanatory "
+            "claim. Updating the existing old claim's status or superseded_by to visibly retire it requires a separate "
+            "reviewed apply capability; approve_claim does not update those fields. Receipt: proposal applied_at, new "
+            "claim/edge IDs, old-claim before/after, and postflight readback."
+        )
+
     return None
 
 
@@ -917,6 +1111,25 @@ def load_current_public_constraints(args: argparse.Namespace, table_names: set[s
     return {row["table_name"]: row["constraints"] for row in psql_json(args, sql)}
 
 
+def load_current_enum_values(args: argparse.Namespace, type_names: set[str]) -> dict[str, list[str]]:
+    if not type_names:
+        return {}
+    sql = f"""
+    select jsonb_build_object(
+             'type_name', t.typname,
+             'values', jsonb_agg(e.enumlabel order by e.enumsortorder)
+           )::text
+      from pg_type t
+      join pg_enum e on e.enumtypid = t.oid
+      join pg_namespace n on n.oid = t.typnamespace
+     where n.nspname = 'public'
+       and t.typname = any({sql_array(sorted(type_names))})
+     group by t.typname
+     order by t.typname;
+    """
+    return {row["type_name"]: row["values"] for row in psql_json(args, sql)}
+
+
 def compact_proposal(proposal: dict[str, Any]) -> dict[str, Any]:
     return {
         key: proposal.get(key)
@@ -1120,6 +1333,9 @@ def context_operational_contracts(args: argparse.Namespace, query: str) -> list[
         table_names.update(
             {"claims", "claim_edges", "reasoning_tools", "behavioral_rules", "governance_gates", "beliefs"}
         )
+    if contract_ids & SCHEMA_COMPILED_CONTRACTS:
+        table_names.update({"claims", "claim_edges"})
+    enum_values = load_current_enum_values(args, {"edge_type"}) if contract_ids & SCHEMA_COMPILED_CONTRACTS else {}
     contracts = operational_contracts(
         query,
         current_schema=load_current_public_schema(args, table_names) if table_names else None,
@@ -1152,6 +1368,8 @@ def context_operational_contracts(args: argparse.Namespace, query: str) -> list[
             contract["live_link_audit"] = direct_snapshot.get("source_link_audit") or {}
         elif contract_id == "demo_capability_readback":
             contract["applied_proposals"] = [item for item in proposal_rows if item.get("status") == "applied"]
+        elif contract_id in SCHEMA_COMPILED_CONTRACTS:
+            contract["current_edge_types"] = enum_values.get("edge_type", [])
     return contracts
 
 

hermes-agent/leoclean-plugins/vps/leo-db-context/__init__.py

@@ -28,6 +28,11 @@ COMPILED_CONTRACT_IDS = frozenset(
         "source_link_audit",
         "demo_capability_readback",
         "identity_canonicality_readback",
+        "runtime_persistence",
+        "shared_claims_agent_positions",
+        "forecast_resolution_schema",
+        "telegram_delivery_proof",
+        "claim_supersession_schema",
     }
 )
 DECISION_MATRIX_TABLES = tuple(
@@ -79,9 +84,11 @@ def _requires_database_truth(query: str) -> bool:
     )
     return bool(
         named_packet_question
+        or ("telegram" in lowered and "gatewayrunner" in lowered)
         or re.search(
             r"\b(?:databases?|db|knowledge bases?|kb|proposals?|canonical|source_ref|source rows?|"
-            r"decision[- ]matri(?:x|ces)|soul\.md|soul file|claim_evidence|claim_edges|reasoning_tools)\b|"
+            r"decision[- ]matri(?:x|ces)|soul\.md|soul file|claim_evidence|claim_edges|reasoning_tools|"
+            r"approve_claim|superseded_by|current[- ]schema)\b|"
             r"public\.|kb_stage",
             lowered,
         )

scripts/working_leo_m3taversal_oos_benchmark.py

@@ -10,6 +10,7 @@ or Telegram-delivery harness.
 from __future__ import annotations
 
 import argparse
+import hashlib
 import json
 import re
 from datetime import datetime, timezone
@@ -242,7 +243,7 @@ CONCEPT_PATTERNS: dict[str, tuple[re.Pattern[str], ...]] = {
     ),
     "state_boundary": (
         re.compile(
-            r"approved is not the same as applied|applied_at:\s*(?:none|null)|not canonical|"
+            r"approved is not (?:the same as )?applied|applied_at:\s*(?:none|null)|not canonical|"
             r"proposal does not commit|no receipt,? no durable knowledge|"
             r"approval.{0,60}not (?:a )?(?:DB|database|canonical) write|"
             r"(?:approval|sign-off) is intent,? not (?:apply|a canonical write)|none.{0,30}applied",
@@ -258,7 +259,9 @@ CONCEPT_PATTERNS: dict[str, tuple[re.Pattern[str], ...]] = {
             re.I,
         ),
     ),
-    "provenance": (re.compile(r"provenance|stable (?:reference|ref)|source row|author/channel|file hash", re.I),),
+    "provenance": (
+        re.compile(r"provenance|stable (?:reference|ref)|source row|author/channel|(?:file|content) hash", re.I),
+    ),
     "deduplication": (re.compile(r"deduplic|duplicate|existing claim|overlap", re.I),),
     "contradiction": (re.compile(r"contradict|divergence|competing interpretation|disagree", re.I),),
     "staged_review_apply": (
@@ -472,9 +475,9 @@ COUNT_INVARIANT_REJECTION_RE = re.compile(
 
 SCHEMA_GAP_QUALIFIER_RE = re.compile(
     r"\b(?:proposed|future|not current|not shipped|does not exist|doesn't exist|absent|"
-    r"has no|have no|no column|not an edge|would require|schema gap|must be added|"
+    r"has no|have no|there is no|no column|not an edge|would require|schema gap|must be added|"
     r"does not support|doesn't support|supports neither|not supported|must not|do not invent)\b|"
-    r"\bno\s+(?:[a-z_]+\s+){0,3}(?:column|field|table)\b",
+    r"\bno\s+(?:[a-z_]+\s+){0,3}(?:column|field|table|edge type)\b",
     re.I,
 )
 CURRENT_SCHEMA_ASSERTION_PATTERNS: dict[str, re.Pattern[str]] = {
@@ -605,6 +608,20 @@ APPLYABILITY_GAP_RE = re.compile(
 
 DEFAULT_MAX_RESPONSE_WORDS = 220
 MAX_RESPONSE_WORDS = {"OOS-09": 100, "OOS-10": 180}
+BLOCKER_STOPWORDS = {
+    "blocker",
+    "chat",
+    "demo",
+    "label",
+    "only",
+    "remember",
+    "single",
+    "that",
+    "this",
+    "under",
+    "will",
+    "with",
+}
 
 
 def prompt_catalog(memory_token: str) -> list[dict[str, Any]]:
@@ -617,6 +634,28 @@ def prompt_catalog(memory_token: str) -> list[dict[str, Any]]:
     ]
 
 
+def extract_blocker_clause(reply: str) -> str | None:
+    for pattern in (
+        re.compile(r"\bblocker\s*:\s*(?P<value>[^\n]+)", re.I),
+        re.compile(r"\bblocker\s+(?:is|was)\s+(?P<value>[^\n]+)", re.I),
+    ):
+        match = pattern.search(reply)
+        if match:
+            return match.group("value").strip()
+    return None
+
+
+def blocker_terms(value: str | None, *, memory_token: str) -> set[str]:
+    if not value:
+        return set()
+    token_terms = set(re.findall(r"[a-z0-9_]+", memory_token.lower()))
+    return {
+        term
+        for term in re.findall(r"[a-z0-9_]+", value.lower())
+        if len(term) >= 4 and term not in BLOCKER_STOPWORDS and term not in token_terms
+    }
+
+
 def matched_concept(reply: str, concept: str) -> bool:
     return all(pattern.search(reply) for pattern in CONCEPT_PATTERNS[concept])
 
@@ -803,6 +842,27 @@ def score_results(results: list[dict[str, Any]], *, memory_token: str) -> dict[s
         for prompt_id in expected_ids
         if prompt_id in by_result
     ]
+    memory_source_reply = str((by_result.get("OOS-07") or {}).get("reply") or "")
+    memory_recall_reply = str((by_result.get("OOS-08") or {}).get("reply") or "")
+    source_clause = extract_blocker_clause(memory_source_reply)
+    source_terms = blocker_terms(source_clause, memory_token=memory_token)
+    recall_terms = blocker_terms(extract_blocker_clause(memory_recall_reply), memory_token=memory_token)
+    overlap_terms = source_terms & recall_terms
+    required_overlap = min(3, max(1, (len(source_terms) + 2) // 3)) if source_terms else 1
+    same_blocker_recalled = bool(source_clause and len(overlap_terms) >= required_overlap)
+    for score in scores:
+        if score["prompt_id"] != "OOS-08":
+            continue
+        score["custom_signals"]["same_blocker_recalled"] = same_blocker_recalled
+        score["pass"] = bool(score["pass"] and same_blocker_recalled)
+    memory_continuity = {
+        "source_reply_sha256": hashlib.sha256(memory_source_reply.encode("utf-8")).hexdigest(),
+        "source_blocker_clause": source_clause,
+        "source_blocker_terms": sorted(source_terms),
+        "recall_overlap_terms": sorted(overlap_terms),
+        "required_overlap": required_overlap,
+        "same_blocker_recalled": same_blocker_recalled,
+    }
     return {
         "expected_prompt_count": len(expected_ids),
         "expected_prompt_ids": expected_ids,
@@ -812,6 +872,7 @@ def score_results(results: list[dict[str, Any]], *, memory_token: str) -> dict[s
         "passes": sum(1 for score in scores if score["pass"]),
         "failures": [score for score in scores if not score["pass"]],
         "scores": scores,
+        "memory_continuity": memory_continuity,
         "pass": not missing
         and not unexpected
         and len(scores) == len(expected_ids)

tests/test_hermes_leoclean_db_context_plugin.py

@@ -279,6 +279,8 @@ def test_database_truth_fallback_covers_every_direct_readback_question() -> None
         "Are the pending proposals stuck because the documents are not pointed at the right source rows?",
         "Can I demo that Leo changes the KB?",
         "If we changed SOUL.md, did we change Leo's canonical identity?",
+        "In the current-schema, can approve_claim set superseded_by?",
+        "A temporary-profile GatewayRunner posted nothing to Telegram. Is delivery proven?",
     )
 
     assert all(module._requires_database_truth(question) for question in direct_questions)
@@ -286,6 +288,39 @@ def test_database_truth_fallback_covers_every_direct_readback_question() -> None
     assert module._requires_database_truth("How are you?") is False
 
 
+def test_plugin_enforces_every_database_backed_oos_compiler() -> None:
+    module = load_plugin()
+    expected_ids = {
+        "runtime_persistence",
+        "shared_claims_agent_positions",
+        "forecast_resolution_schema",
+        "telegram_delivery_proof",
+        "claim_supersession_schema",
+    }
+    assert expected_ids <= module.COMPILED_CONTRACT_IDS
+
+    for index, contract_id in enumerate(sorted(expected_ids)):
+        query = f"database-backed query {index}"
+        query_sha256 = module.hashlib.sha256(query.encode("utf-8")).hexdigest()
+        compiled = f"Database-composed response for {contract_id}."
+        contracts = [{"id": "reply_budget", "hard_max_words": 200}, {"id": contract_id}]
+        module._store_snapshot(
+            f"compiled-{index}",
+            {
+                "status": "ok",
+                "query_sha256": query_sha256,
+                "contracts": contracts,
+                "compiled_response": compiled,
+                "requires_database_truth": True,
+                "context": "fixture",
+            },
+        )
+
+        assert module._post_llm_call(
+            session_id=f"compiled-{index}", user_message=query, assistant_response="Unconstrained draft."
+        ) == {"assistant_response": compiled}
+
+
 def test_plugin_fails_closed_on_missing_snapshot_or_invalid_compiler() -> None:
     module = load_plugin()
     query = "Did an approved proposal change the knowledge base?"

tests/test_hermes_leoclean_kb_bridge_source.py

@@ -293,6 +293,43 @@ def test_vps_bridge_direct_intents_reject_unrelated_and_resolve_overlaps() -> No
     assert substring_ids & direct_ids == {"proposal_state_readback"}
 
 
+def test_vps_bridge_oos_intents_preserve_specific_contracts_and_negated_actions() -> None:
+    module = _load_module(BRIDGE_DIR / "kb_tool.py")
+    cases = {
+        (
+            "I corrected Leo's worldview in chat and someone also edited SOUL.md. Will that correction be part of "
+            "Leo's canonical identity tomorrow after a restart? Give me the exact truth test from database row to "
+            "runtime artifact. Do not change either surface."
+        ): {"identity_canonicality_readback", "runtime_persistence"},
+        (
+            "The five database totals are unchanged after a restart. Does that prove Leo's answer behavior is "
+            "unchanged and that every fact from the prior session was erased?"
+        ): {"runtime_persistence"},
+        (
+            "An old claim recorded a 60% forecast but never defined resolution criteria. The event is now over. "
+            "What needs a schema proposal?"
+        ): {"forecast_resolution_schema"},
+        (
+            "A canonical claim is wrong. I want the replacement and the old claim visibly retired. In current v1, "
+            "which writes fit approve_claim and which require a separate reviewed apply capability?"
+        ): {"claim_supersession_schema"},
+        (
+            "A temporary-profile GatewayRunner answered, but posted nothing to Telegram. Is delivery proven live?"
+        ): {"telegram_delivery_proof"},
+    }
+
+    for query, expected_ids in cases.items():
+        contract_ids = {item["id"] for item in module.operational_contracts(query)} - {"reply_budget"}
+        assert contract_ids == expected_ids, query
+
+    memory_query = (
+        "Remember this blocker under __MEMORY_TOKEN__. This is chat memory only; do not write it to the KB."
+    )
+    memory_ids = {item["id"] for item in module.operational_contracts(memory_query)}
+    assert not memory_ids & module.DIRECT_READBACK_CONTRACTS
+    assert memory_ids == {"reply_budget"}
+
+
 def _direct_contract_fixtures() -> tuple[dict, dict, dict, dict]:
     database_status = {
         "high_signal_rows": {
@@ -581,6 +618,40 @@ def test_vps_bridge_mixed_contract_reads_live_postgres_constraints(monkeypatch)
     assert "proposal-level applied_at" in mixed["review_apply_sequence"][-1]
 
 
+def test_vps_bridge_schema_contract_reads_live_columns_and_edge_enum(monkeypatch) -> None:
+    module = _load_module(BRIDGE_DIR / "kb_tool.py")
+    captured_sql: list[str] = []
+
+    def fake_psql_json(_args, sql):
+        captured_sql.append(sql)
+        if "pg_enum" in sql:
+            return [{"type_name": "edge_type", "values": ["supports", "supersedes"]}]
+        if "information_schema.columns" in sql:
+            return [
+                {"table_name": "claims", "columns": ["id", "text", "superseded_by"]},
+                {"table_name": "claim_edges", "columns": ["id", "from_claim", "to_claim", "edge_type"]},
+            ]
+        if "pg_constraint" in sql:
+            return []
+        raise AssertionError(sql)
+
+    monkeypatch.setattr(module, "psql_json", fake_psql_json)
+    contracts = {
+        item["id"]: item
+        for item in module.context_operational_contracts(
+            SimpleNamespace(),
+            "A canonical claim is wrong. In current v1, can approve_claim insert its replacement and supersedes edge?",
+        )
+    }
+
+    supersession = contracts["claim_supersession_schema"]
+    assert supersession["current_columns"]["claims"] == ["id", "text", "superseded_by"]
+    assert supersession["current_columns"]["claim_edges"] == ["id", "from_claim", "to_claim", "edge_type"]
+    assert supersession["current_edge_types"] == ["supports", "supersedes"]
+    assert any("pg_enum" in sql and "t.typname" in sql for sql in captured_sql)
+    assert all(not re.search(r"\b(?:insert|update|delete|alter|drop|create)\b", sql, re.I) for sql in captured_sql)
+
+
 def test_vps_bridge_decision_matrix_status_checks_schema_tables(monkeypatch) -> None:
     module = _load_module(BRIDGE_DIR / "kb_tool.py")
     captured_sql: list[str] = []

tests/test_working_leo_m3taversal_oos_benchmark.py

@@ -154,6 +154,10 @@ def test_oos_score_passes_complete_behavior_and_memory_pair() -> None:
     score = benchmark.score_results(results, memory_token=token)
     assert score["pass"] is True
     assert score["passes"] == 15
+    assert score["memory_continuity"]["same_blocker_recalled"] is True
+    assert {"approved", "applied", "canonical"} <= set(
+        score["memory_continuity"]["recall_overlap_terms"]
+    )
 
 
 def test_oos_live_check_accepts_live_readback_wording() -> None:
@@ -174,6 +178,26 @@ def test_oos_score_fails_when_memory_token_is_not_recalled() -> None:
     assert score["failures"][0]["custom_signals"]["memory_token"] is False
 
 
+def test_oos_score_fails_when_recalled_label_has_a_different_blocker() -> None:
+    token = "demo-ledger-deadbeef"
+    results = [
+        {"prompt_id": prompt["id"], "reply": good_reply(prompt["id"], token)}
+        for prompt in benchmark.prompt_catalog(token)
+    ]
+    recall = next(result for result in results if result["prompt_id"] == "OOS-08")
+    recall["reply"] = (
+        f"Label: {token}. Blocker: Telegram delivery has no visible message receipt. "
+        "Proof: send one authorized canary and retain its message ID, timestamp, and chat readback."
+    )
+
+    score = benchmark.score_results(results, memory_token=token)
+
+    assert score["pass"] is False
+    recall_score = next(item for item in score["scores"] if item["prompt_id"] == "OOS-08")
+    assert recall_score["custom_signals"]["memory_token"] is True
+    assert recall_score["custom_signals"]["same_blocker_recalled"] is False
+
+
 def test_oos_identity_case_requires_exact_visible_handle_and_no_alias() -> None:
     token = "demo-ledger-deadbeef"
     prompt = benchmark.prompt_catalog(token)[8]
@@ -508,6 +532,88 @@ def test_db_compiled_responses_pass_composition_and_source_intake_benchmarks() -
         assert benchmark.score_reply(prompt, response, memory_token=token)["pass"] is True
 
 
+def test_db_compiled_responses_pass_all_database_backed_oos_cases() -> None:
+    kb_tool = load_kb_tool()
+    token = "demo-ledger-deadbeef"
+    database_status = {
+        "high_signal_rows": {
+            "claims": 1837,
+            "sources": 4145,
+            "claim_edges": 4916,
+            "claim_evidence": 4670,
+            "kb_proposals": 26,
+        },
+        "proposal_status_counts": {"applied": 2, "approved": 3, "pending_review": 14, "canceled": 7},
+    }
+    approved = {
+        "id": "a64df080-8502-42e2-98f4-9bbdecb8da73",
+        "proposal_type": "approve_claim",
+        "status": "approved",
+        "applied_at": None,
+        "readiness": {"review_state": "approved_needs_apply_payload"},
+    }
+    applied = {
+        **approved,
+        "id": "11111111-1111-4111-8111-111111111111",
+        "status": "applied",
+        "applied_at": "2026-07-01T00:00:00+00:00",
+        "readiness": {"review_state": "applied"},
+    }
+    source_audit = {
+        "pending_or_approved": 17,
+        "with_source_ref": 16,
+        "without_source_ref": 1,
+        "exact_public_source_matches": 2,
+        "samples": [],
+    }
+    compiled_prompt_ids = {
+        "OOS-01",
+        "OOS-02",
+        "OOS-03",
+        "OOS-04",
+        "OOS-05",
+        "OOS-06",
+        "OOS-10",
+        "OOS-11",
+        "OOS-12",
+        "OOS-13",
+        "OOS-14",
+        "OOS-15",
+    }
+
+    for prompt in benchmark.prompt_catalog(token):
+        if prompt["id"] not in compiled_prompt_ids:
+            continue
+        contracts = kb_tool.operational_contracts(prompt["message"])
+        for contract in contracts:
+            contract_id = contract["id"]
+            if contract_id in kb_tool.DIRECT_READBACK_CONTRACTS:
+                contract["database_status"] = database_status
+            if contract_id == "demo_capability_readback":
+                contract["applied_proposals"] = [applied]
+                contract["applied_and_approved_proposals"] = [approved, applied]
+            elif contract_id == "proposal_state_readback":
+                contract["applied_and_approved_proposals"] = [approved, applied]
+            elif contract_id == "source_link_audit":
+                contract["live_link_audit"] = source_audit
+            elif contract_id in kb_tool.SCHEMA_COMPILED_CONTRACTS:
+                contract["current_edge_types"] = ["supports", "supersedes"]
+
+        response = kb_tool.compile_operational_response(contracts)
+        assert response is not None, prompt["id"]
+        score = benchmark.score_reply(prompt, response, memory_token=token)
+        assert score["pass"] is True, score
+
+
+def test_oos_state_boundary_accepts_approved_is_not_applied() -> None:
+    assert benchmark.matched_concept("Approved is not applied.", "state_boundary") is True
+
+
+def test_oos_schema_gap_accepts_explicitly_absent_edge_type() -> None:
+    reply = "Current claim_edges has edge_type; there is no `resolves` edge type."
+    assert benchmark.current_schema_overclaims(reply) == []
+
+
 def test_oos_source_intake_accepts_live_not_shipped_wording() -> None:
     patterns = benchmark.CONCEPT_PATTERNS["bounded_intake_tier"]
     reply = "The local compiler is proven; live VPS/chat intake is not shipped."