Harden Leo DB contract correction path

hermes-agent/leoclean-bin/kb_tool.py

@@ -583,6 +583,31 @@ def operational_contracts(
         }
     ]
 
+    apply_receipt_question = any(
+        term in lowered for term in ("canonical receipt", "postflight", "closure proof", "before/after")
+    ) or bool(
+        re.search(r"\bapplied_at\b\s+(?:readback|proof|receipt|postflight)\b", lowered)
+    )
+    if apply_receipt_question:
+        contracts.append(
+            {
+                "id": "apply_receipt_invariants",
+                "count_boundary": ("aggregate table counts need not change; in-place updates can preserve every count"),
+                "proposal_receipt_boundary": "applied_at is proposal-level proof, not a field on each created row",
+                "approve_claim_supported": ["claims", "sources", "evidence", "edges", "reasoning_tools"],
+                "unsupported_without_separate_reviewed_apply": [
+                    "belief updates",
+                    "behavioral_rules",
+                    "governance_gates",
+                    "existing-row updates",
+                ],
+                "proof_boundary": (
+                    "verify the expected row IDs and content hashes; reviewer approval does not prove direct "
+                    "applyability when a payload or apply surface is unsupported"
+                ),
+            }
+        )
+
     decision_matrix_question = "decision matrix" in lowered or "decision-matrix" in lowered
     source_link_audit_question = (
         any(term in lowered for term in ("proposal", "pending", "approved", "stuck", "backlog"))

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

@@ -316,6 +316,10 @@ def _format_context(
         "a staging or write operation unless the user separately authorizes the exact review/apply action. Use natural "
         "prose rather than a fixed benchmark template. When a contract covers runtime persistence, agent positions, "
         "or forecast resolution, preserve its exact boundary even when retrieved prose suggests another design.\n"
+        "For apply receipts, aggregate table counts need not change because updates can preserve counts. applied_at "
+        "is proposal-level proof, not a field on every created row. approve_claim supports claims, sources, evidence, "
+        "edges, and reasoning_tools only; belief updates, behavioral_rules, governance_gates, and existing-row updates "
+        "remain staged unless a separate reviewed apply capability exists.\n"
         "</leo_database_reasoning_protocol>"
     )
 
@@ -858,7 +862,7 @@ def _schema_contract_issues(response: str, contracts: list[dict[str, Any]]) -> l
             if not re.search(pattern, clause, re.I | re.S):
                 continue
             if not re.search(
-                r"\b(?:no|not|never|without|must not|do not|does not|cannot|isn't|aren't)\b",
+                r"\b(?:no|not|never|neither|without|must not|do not|does not|cannot|isn't|aren't)\b",
                 clause,
                 re.I,
             ):
@@ -866,9 +870,33 @@ def _schema_contract_issues(response: str, contracts: list[dict[str, Any]]) -> l
         return False
 
     if "runtime_persistence" in contract_ids:
+        process_local_contradiction = False
+        for match in re.finditer(
+            r"(?:state\.db|session\s+jsonl).{0,100}process[- ]local",
+            response,
+            re.I | re.S,
+        ):
+            target = re.search(r"process[- ]local", match.group(0), re.I)
+            before_target = match.group(0)[: target.start()] if target else match.group(0)
+            leading_context = response[max(0, match.start() - 40) : match.start()]
+            negation_context = f"{leading_context}{before_target}"
+            full_context = response[max(0, match.start() - 40) : min(len(response), match.end() + 30)]
+            locally_negated = bool(
+                re.search(r"\b(?:not|never)\b.{0,30}$", before_target, re.I | re.S)
+                or re.search(r"\b(?:false|untrue|not\s+true)\s+that\b", negation_context, re.I)
+                or (
+                    re.search(r"\bneither\b", negation_context, re.I)
+                    and re.search(r"\bnor\b", negation_context, re.I)
+                )
+                or re.search(r"\bneither\b.{0,100}process[- ]local\s+nor\b", full_context, re.I | re.S)
+            )
+            if not locally_negated:
+                process_local_contradiction = True
+                break
         direct_contradiction = affirmative(
             r"(?:state\.db|session\s+jsonl).{0,100}"
-            r"(?:in[- ]memory|ephemeral|disappears?|vanishes?|erased|discarded|gone|lost|deleted|absent|zeroed|empty)"
+            r"(?:in[- ]memory|ephemeral|disappears?|vanishes?|erased|discarded|gone|lost|"
+            r"deleted|absent|zeroed|empty)"
         )
         anaphoric_contradiction = bool(
             re.search(
@@ -880,7 +908,7 @@ def _schema_contract_issues(response: str, contracts: list[dict[str, Any]]) -> l
                 re.I | re.S,
             )
         )
-        if direct_contradiction or anaphoric_contradiction:
+        if process_local_contradiction or direct_contradiction or anaphoric_contradiction:
             issues.append("runtime_persistence_contradiction")
 
     if "shared_claims_agent_positions" in contract_ids:
@@ -896,6 +924,18 @@ def _schema_contract_issues(response: str, contracts: list[dict[str, Any]]) -> l
             issues.append("shared_position_edge_contradiction")
 
     if "forecast_resolution_schema" in contract_ids:
+        if re.search(
+            r"(?:missing|absent|omitted).{0,60}resolution (?:rule|criteria).{0,80}"
+            r"(?:is|are) not (?:a )?schema gap",
+            response,
+            re.I | re.S,
+        ):
+            issues.append("forecast_resolution_gap_denied")
+        if affirmative(
+            r"(?:new|distinct|observed).{0,80}claim.{0,100}(?:typed\s+)?(?:belief|observation|hypothesis)\b|"
+            r"(?:new|distinct)\s+(?:typed\s+)?(?:belief|observation|hypothesis)\s+claim\b"
+        ):
+            issues.append("forecast_unreviewed_claim_type")
         if affirmative(
             r"(?:use|repurpose|treat).{0,60}supersedes.{0,80}(?:resol|outcome)|"
             r"supersedes.{0,80}(?:record|represent|mark).{0,60}(?:resol|outcome)"
@@ -915,6 +955,105 @@ def _schema_contract_issues(response: str, contracts: list[dict[str, Any]]) -> l
     return sorted(set(issues))
 
 
+def _apply_receipt_issues(response: str) -> list[str]:
+    """Reject receipt rules that confuse count movement or approval with applyability."""
+
+    issues: list[str] = []
+    clauses = [
+        item.strip()
+        for item in re.split(
+            r"(?<=[.!?;])\s+|\n+|,\s+(?:but|and)\s+|\s+but\s+|"
+            r",\s+(?=(?:it|this|that|for this apply|database totals?|the current apply)\b)",
+            response,
+            flags=re.I,
+        )
+        if item.strip()
+    ]
+    unsupported_objects = r"(?:beliefs?|behavioral_rules|governance_gates|existing[- ]row updates?)"
+    for clause in clauses:
+        count_required = re.search(
+            r"\b(?:database\s+|aggregate\s+)?(?:counts?|totals?)\s+"
+            r"(?:must|need(?:s)? to|are required to)\s+(?:change|differ|move)\b|"
+            r"\b(?:count readback|counts?|totals?).{0,40}\bmust\b.{0,40}"
+            r"\b(?:counts?|totals?).{0,30}\b(?:change|differ|move)\b",
+            clause,
+            re.I | re.S,
+        )
+        if count_required:
+            issues.append("apply_receipt_count_movement_required")
+
+        unsupported_surface = re.search(
+            rf"(?:authorized|guarded|current|the)\s+apply.{{0,220}}{unsupported_objects}|"
+            rf"{unsupported_objects}.{{0,100}}(?:written|created|updated|applied).{{0,80}}"
+            rf"(?:authorized|guarded|current|the)\s+apply|"
+            rf"(?:approve_claim|it|this|(?:authorized|guarded|current|the)\s+apply).{{0,120}}"
+            rf"(?:writes?|creates?|updates?|includes?|supports?).{{0,80}}{unsupported_objects}",
+            clause,
+            re.I | re.S,
+        )
+        unsupported_boundary = re.search(
+            rf"not directly applyable|remain(?:s)? staged|separate reviewed apply|"
+            rf"unsupported apply (?:surface|payload)|"
+            rf"(?:approve_claim|apply).{{0,120}}(?:does not|do not|cannot|must not|will not|supports neither|only)"
+            rf".{{0,80}}{unsupported_objects}|"
+            rf"{unsupported_objects}.{{0,60}}(?:is|are) not (?:written|created|updated|applied)",
+            clause,
+            re.I | re.S,
+        )
+        if unsupported_surface and not unsupported_boundary:
+            issues.append("apply_receipt_unsupported_surface")
+    return sorted(set(issues))
+
+
+def _repair_apply_receipt_response(response: str) -> str:
+    correction = (
+        "Proof-boundary correction: aggregate counts need not change; verify expected row IDs, content hashes, and "
+        "proposal-level applied_at; approved content is not directly applyable when its payload or write surface is "
+        "unsupported; approve_claim supports claims, sources, evidence, edges, and reasoning_tools only; belief "
+        "updates, behavioral_rules, governance_gates, and existing-row updates remain staged for a separate reviewed "
+        "apply capability; this correction overrides any conflicting count or apply-surface wording below."
+    )
+    label_match = re.search(
+        r"\b(?P<kind>Label|Mnemonic)\s*:\s*(?P<token>[A-Za-z0-9][A-Za-z0-9._-]{2,100})",
+        response,
+        re.I,
+    )
+    if not label_match:
+        token_match = re.search(r"\b[A-Za-z][A-Za-z0-9]*-[A-Za-z0-9-]*-[A-Fa-f0-9]{8,}\b", response)
+        label = f"Label: {token_match.group(0)}." if token_match else ""
+    else:
+        label = f"{label_match.group('kind').title()}: {label_match.group('token')}."
+
+    blocker_match = re.search(r"\bBlocker\s*:\s*(?P<body>[^.!?\n]{1,500}[.!?]?)", response, re.I)
+    blocker = _truncate_to_word_budget(f"Blocker: {blocker_match.group('body').strip()}", 45) if blocker_match else ""
+    closure = (
+        "Closure proof: after an authorized supported apply, verify proposal-level applied_at and the expected "
+        "canonical row IDs plus content hashes; aggregate count movement is optional."
+    )
+
+    safe_segments: list[str] = []
+    for segment in re.split(
+        r"(?<=[.!?;])\s+|\n+|,\s+(?:but|and)\s+|\s+but\s+|"
+        r",\s+(?=(?:it|this|that|for this apply|database totals?|the current apply)\b)",
+        response,
+        flags=re.I,
+    ):
+        segment = segment.strip()
+        if not segment or _apply_receipt_issues(segment):
+            continue
+        if label_match and label_match.group(0).lower() in segment.lower():
+            continue
+        if blocker_match and blocker_match.group(0).lower() in segment.lower():
+            continue
+        safe_segments.append(segment)
+
+    parts = [correction]
+    parts.extend(item for item in (label, blocker, closure) if item)
+    if safe_segments:
+        parts.append("Additional context: " + " ".join(safe_segments))
+    return "\n".join(parts)
+
+
 def response_contract_issues(response: str, contracts: list[dict[str, Any]]) -> list[str]:
     by_id = _contract_map(contracts)
     contract_ids = set(by_id)
@@ -928,6 +1067,8 @@ def response_contract_issues(response: str, contracts: list[dict[str, Any]]) ->
         issues.extend(_source_intake_issues(response))
     if "claim_evidence_challenge" in contract_ids:
         issues.extend(_claim_evidence_challenge_issues(response, by_id["claim_evidence_challenge"]))
+    if "apply_receipt_invariants" in contract_ids:
+        issues.extend(_apply_receipt_issues(response))
     issues.extend(_live_readback_issues(response, contracts))
     issues.extend(_schema_contract_issues(response, contracts))
     return sorted(set(issues))
@@ -994,6 +1135,12 @@ def _post_llm_call(**kwargs: Any) -> dict[str, str] | None:
     contracts = list(snapshot.get("contracts") or [])
     contract_ids = {str(item.get("id") or "") for item in contracts}
     issues = response_contract_issues(response, contracts)
+    repaired_response = response
+    apply_receipt_repair_applied = False
+    if "apply_receipt_invariants" in contract_ids and any(issue.startswith("apply_receipt_") for issue in issues):
+        repaired_response = _repair_apply_receipt_response(response)
+        apply_receipt_repair_applied = True
+    post_repair_issues = response_contract_issues(repaired_response, contracts)
     compiled_response = snapshot.get("compiled_response")
     compiled_issues = (
         response_contract_issues(str(compiled_response), contracts) if isinstance(compiled_response, str) else []
@@ -1002,13 +1149,13 @@ def _post_llm_call(**kwargs: Any) -> dict[str, str] | None:
     compiled_valid = bool(isinstance(compiled_response, str) and compiled_response.strip() and not compiled_issues)
     fail_closed = bool(compiled_required and not compiled_valid)
     compiled_fallback_available = bool(compiled_required and compiled_valid)
-    compiled_fallback_required = bool(compiled_fallback_available and _requires_compiled_fallback(issues))
+    compiled_fallback_required = bool(compiled_fallback_available and _requires_compiled_fallback(post_repair_issues))
     if fail_closed:
         delivered = DATABASE_UNAVAILABLE_RESPONSE
     elif compiled_fallback_required:
         delivered = str(compiled_response)
     else:
-        delivered = response
+        delivered = repaired_response
     delivered, requested_subject_injected = _ensure_requested_subject(delivered, user_message)
     hard_max = _reply_hard_max(contracts)
     budget_compacted = bool(not fail_closed and hard_max is not None and _word_count(delivered) > hard_max)
@@ -1023,11 +1170,13 @@ def _post_llm_call(**kwargs: Any) -> dict[str, str] | None:
             "contract_satisfied": not delivered_issues,
             "contract_ids": sorted(contract_ids),
             "issues": issues,
+            "post_repair_issues": post_repair_issues,
             "delivered_issues": delivered_issues,
             "compiled_response_issues": compiled_issues,
             "transformed": transformed,
             "budget_compacted": budget_compacted,
             "compiled_response_enforced": compiled_fallback_required,
+            "apply_receipt_repair_applied": apply_receipt_repair_applied,
             "requested_subject_injected": requested_subject_injected,
             "fail_closed": fail_closed,
             "delivered_response_sha256": hashlib.sha256(delivered.encode("utf-8")).hexdigest(),

scripts/working_leo_m3taversal_oos_benchmark.py

@@ -391,6 +391,8 @@ CONCEPT_PATTERNS: dict[str, tuple[re.Pattern[str], ...]] = {
             r"unchanged.{0,120}(?:does not|doesn't|do not).{0,80}(?:behavior|answer)|"
             r"(?:unchanged|identical).{0,50}(?:counts|totals).{0,140}"
             r"(?:prove neither|say nothing about|do not prove|don't prove).{0,60}(?:behavior|answer)|"
+            r"(?:DB|database) totals.{0,250}(?:do not prove|don't prove).{0,100}"
+            r"(?:row contents?|answers?|answered|behavior)|"
             r"(?:does not|doesn't|do not) prove.{0,60}(?:behavior|answer)|"
             r"unchanged (?:database )?(?:totals|counts).{0,60}say nothing about.{0,30}(?:behavior|answer)|"
             r"(?:identical|unchanged).{0,40}(?:counts|totals).{0,60}(?:prove nothing about content|"
@@ -410,7 +412,7 @@ CONCEPT_PATTERNS: dict[str, tuple[re.Pattern[str], ...]] = {
         ),
     ),
     "row_content_proof": (
-        re.compile(r"(?:unchanged|identical) (?:database )?(?:counts?|totals?)", re.I),
+        re.compile(r"(?:unchanged|identical) (?:database )?(?:counts?|totals?)|(?:DB|database) totals", re.I),
         re.compile(r"does not prove|doesn't prove|do not prove|prove(?:s)? nothing|say nothing", re.I),
         re.compile(
             r"row (?:IDs?|hashes?)|fingerprints?|timestamps?|balanced (?:insert|write|change)|"
@@ -423,10 +425,15 @@ CONCEPT_PATTERNS: dict[str, tuple[re.Pattern[str], ...]] = {
         re.compile(
             r"proof tiers?|tier 1.{0,500}tier 2.{0,500}tier 3|"
             r"canonical rows.{0,600}deployed runtime inputs.{0,600}durable session state|"
-            r"content-level DB proof.{0,600}runtime configuration.{0,600}persisted conversation state",
+            r"content-level DB proof.{0,600}runtime configuration.{0,600}persisted conversation state|"
+            r"DB totals.{0,600}runtime configuration.{0,600}persisted conversation state.{0,600}"
+            r"Telegram-visible proof",
             re.I | re.S,
         ),
-        re.compile(r"canonical|public\.\*|DB mutation|database mutation|content-level DB proof", re.I),
+        re.compile(
+            r"canonical|public\.\*|DB mutation|database mutation|content-level DB proof|(?:DB|database) totals",
+            re.I,
+        ),
         re.compile(r"runtime|skills?|session|SOUL\.md", re.I),
     ),
     "shared_knowledge_commons": (
@@ -437,12 +444,14 @@ CONCEPT_PATTERNS: dict[str, tuple[re.Pattern[str], ...]] = {
             r"(?:shared source material|observation).{0,100}(?:one|single) claim row|"
             r"shared (?:source material|evidence).{0,100}one canonical claim|"
             r"one shared (?:structural |canonical )?claim row|"
+            r"share the fact.{0,100}one claim row|"
             r"fact shared.{0,80}(?:one|single).{0,40}(?:claim|public\.claims)",
             re.I | re.S,
         ),
         re.compile(r"source|evidence", re.I),
         re.compile(
-            r"do not duplicate|don'?t duplicate|do not fork|does not fork|one shared claim|single shared claim|"
+            r"do not duplicate|don'?t duplicate|duplicate nothing|do not fork|does not fork|"
+            r"one shared claim|single shared claim|"
             r"duplicating (?:(?:a|the) )?(?:factual )?claim|"
             r"one (?:canonical |structural )?claim.{0,100}shared (?:sources?|evidence|claim_evidence)|"
             r"one claim row.{0,100}shared (?:sources?|evidence|claim_evidence)",
@@ -629,12 +638,18 @@ CONCEPT_PATTERNS: dict[str, tuple[re.Pattern[str], ...]] = {
 INVALID_COUNT_INVARIANT_RE = re.compile(
     r"\ball five (?:canonical )?(?:numbers|counts) (?:must|need to) move\b|"
     r"\bneed all five (?:canonical )?(?:numbers|counts) to move\b|"
-    r"\ball five (?:canonical )?(?:numbers|counts).{0,40}\b(?:higher|increase|increased|rise|change|changed)\b",
+    r"\ball five (?:canonical )?(?:numbers|counts).{0,40}\b(?:higher|increase|increased|rise|change|changed)\b|"
-    re.I,
+    r"\b(?:database\s+|aggregate\s+)?(?:counts?|totals?)\s+"
+    r"(?:must|need(?:s)? to|are required to)\s+(?:change|differ|move)\b|"
+    r"\b(?:count readback|counts?|totals?).{0,40}\bmust\b.{0,40}"
+    r"\b(?:counts?|totals?).{0,30}\b(?:change|differ|move)\b",
+    re.I | re.S,
 )
 COUNT_INVARIANT_REJECTION_RE = re.compile(
     r"(?:all five (?:numbers|counts)|count invariant).{0,50}(?:wrong|false|invalid|not valid|too strong)|"
-    r"(?:wrong|false|invalid|not valid|too strong).{0,50}all five (?:numbers|counts)",
+    r"(?:wrong|false|invalid|not valid|too strong).{0,50}all five (?:numbers|counts)|"
+    r"(?:aggregate )?(?:counts?|totals?).{0,50}(?:need not|do not need to|are not required to).{0,30}"
+    r"(?:change|differ|move)",
     re.I | re.S,
 )
 
@@ -659,8 +674,13 @@ SCHEMA_CLAUSE_BOUNDARY_RE = re.compile(
 )
 CURRENT_SCHEMA_ASSERTION_PATTERNS: dict[str, re.Pattern[str]] = {
     "claims_unshipped_fields": re.compile(
-        r"(?:public\.)?claims?\b.{0,60}(?:stores?|has|have|with|column|field).{0,40}"
+        r"claims\s+(?:table|schema)\b.{0,80}\b(?:body|metadata|forecast[_ -]resolution|resolved_at|falsifier)\b|"
-        r"\b(?:body|metadata|forecast[_ -]resolution|resolved_at|falsifier)\b",
+        r"public\.claims\s+(?:stores?|has|contains?)\s+(?:an?\s+)?"
+        r"(?:body|metadata|forecast[_ -]resolution|resolved_at|falsifier)\b|"
+        r"public\.claims\b.{0,80}\b(?:column|field)\b.{0,30}"
+        r"\b(?:body|metadata|forecast[_ -]resolution|resolved_at|falsifier)\b|"
+        r"public\.claims\b.{0,80}\b(?:body|metadata|forecast[_ -]resolution|resolved_at|falsifier)\b.{0,30}"
+        r"\b(?:column|field)\b",
         re.I,
     ),
     "sources_unshipped_fields": re.compile(
@@ -682,7 +702,8 @@ CURRENT_SCHEMA_ASSERTION_PATTERNS: dict[str, re.Pattern[str]] = {
         re.I,
     ),
     "unreviewed_claim_type": re.compile(
-        r"(?:public\.)?claims?.{0,100}\b(?:type|typed)\s*[`'\":=]?\s*(?:observation|hypothesis|belief)\b",
+        r"(?:public\.)?claims?.{0,100}\b(?:type|typed)\s*[`'\":=]?\s*(?:observation|hypothesis|belief)\b|"
+        r"\b(?:new|distinct)\s+(?:observation|hypothesis|belief)\s+claim\b",
         re.I,
     ),
     "reasoning_tools_unshipped_fields": re.compile(
@@ -776,8 +797,9 @@ UNSUPPORTED_COMPOSITION_APPLY_RE = re.compile(
     re.I | re.S,
 )
 APPROVED_APPLY_ACTION_RE = re.compile(
-    r"approved.{0,200}(?:authorize|run|execute|perform|proceed).{0,60}(?:guarded )?apply|"
+    r"approved.{0,200}(?:authorize\b|run\b|execute\b|perform\b|proceed\b).{0,60}(?:guarded )?apply|"
-    r"(?:authorize(?:d)?(?: and)?|run|execute|perform|proceed with).{0,60}(?:guarded )?apply.{0,200}approved",
+    r"(?:authorize\b(?: and)?|run\b|execute\b|perform\b|proceed with\b).{0,60}"
+    r"(?:guarded )?apply.{0,200}approved",
     re.I | re.S,
 )
 APPLY_ACTION_DISCLAIMER_RE = re.compile(
@@ -797,6 +819,22 @@ APPLYABILITY_GAP_RE = re.compile(
     r"strict apply payload.{0,50}(?:built|build|reviewed|review)",
     re.I | re.S,
 )
+UNSUPPORTED_APPLY_OBJECTS = r"(?:beliefs?|behavioral_rules|governance_gates|existing[- ]row updates?)"
+UNSUPPORTED_APPLY_SURFACE_RE = re.compile(
+    rf"(?:authorized|guarded|current|the)\s+apply.{{0,220}}{UNSUPPORTED_APPLY_OBJECTS}|"
+    rf"{UNSUPPORTED_APPLY_OBJECTS}.{{0,100}}(?:written|created|updated|applied).{{0,80}}"
+    rf"(?:authorized|guarded|current|the)\s+apply|"
+    rf"(?:approve_claim|it|this|(?:authorized|guarded|current|the)\s+apply).{{0,120}}"
+    rf"(?:writes?|creates?|updates?|includes?|supports?).{{0,80}}{UNSUPPORTED_APPLY_OBJECTS}",
+    re.I | re.S,
+)
+UNSUPPORTED_APPLY_BOUNDARY_RE = re.compile(
+    rf"not directly applyable|remain(?:s)? staged|separate reviewed apply|unsupported apply (?:surface|payload)|"
+    rf"(?:approve_claim|apply).{{0,120}}(?:does not|do not|cannot|must not|will not|supports neither|only)"
+    rf".{{0,80}}{UNSUPPORTED_APPLY_OBJECTS}|"
+    rf"{UNSUPPORTED_APPLY_OBJECTS}.{{0,60}}(?:is|are) not (?:written|created|updated|applied)",
+    re.I | re.S,
+)
 CLAIM_EVIDENCE_CONFLATION_RE = re.compile(
     r"\bevidence(?: excerpt)?\s*:\s*(?:the\s+)?same as (?:the\s+)?claim\b|"
     r"\b(?:claim body|claim text)\s+(?:is|equals?)\s+(?:the\s+)?evidence\b",
@@ -907,7 +945,16 @@ def matched_concept(reply: str, concept: str) -> bool:
 
 
 def asserts_invalid_count_invariant(reply: str) -> bool:
-    return bool(INVALID_COUNT_INVARIANT_RE.search(reply) and not COUNT_INVARIANT_REJECTION_RE.search(reply))
+    clauses = re.split(
+        r"(?<=[.!?;])\s+|\n+|,\s+(?:but|and)\s+|\s+but\s+|"
+        r",\s+(?=(?:it|this|that|for this apply|database totals?|the current apply)\b)",
+        reply,
+        flags=re.I,
+    )
+    return any(
+        INVALID_COUNT_INVARIANT_RE.search(clause) and not COUNT_INVARIANT_REJECTION_RE.search(clause)
+        for clause in clauses
+    )
 
 
 def current_schema_overclaims(reply: str) -> list[str]:
@@ -1013,13 +1060,30 @@ def broad_semantic_issues(reply: str) -> list[str]:
 def proposal_readiness_issues(prompt_id: str, reply: str) -> list[str]:
     if prompt_id not in {"OOS-01", "OOS-04", "OOS-07", "OOS-08"}:
         return []
-    if (
+    issues: list[str] = []
-        APPROVED_APPLY_ACTION_RE.search(reply)
+    clauses = [
-        and not APPLYABILITY_GAP_RE.search(reply)
+        item.strip()
-        and not APPLY_ACTION_DISCLAIMER_RE.search(reply)
+        for item in re.split(
+            r"(?<=[.!?;])\s+|\n+|,\s+(?:but|and)\s+|\s+but\s+|"
+            r",\s+(?=(?:it|this|that|for this apply|database totals?|the current apply)\b)",
+            reply,
+            flags=re.I,
+        )
+        if item.strip()
+    ]
+    if any(
+        APPROVED_APPLY_ACTION_RE.search(clause)
+        and not APPLYABILITY_GAP_RE.search(clause)
+        and not APPLY_ACTION_DISCLAIMER_RE.search(clause)
+        for clause in clauses
     ):
-        return ["approved_proposal_applyability_overclaim"]
+        issues.append("approved_proposal_applyability_overclaim")
-    return []
+    if any(
+        UNSUPPORTED_APPLY_SURFACE_RE.search(clause) and not UNSUPPORTED_APPLY_BOUNDARY_RE.search(clause)
+        for clause in clauses
+    ):
+        issues.append("unsupported_apply_surface_overclaim")
+    return sorted(set(issues))
 
 
 def source_intake_issues(prompt_id: str, reply: str) -> list[str]:

scripts/working_leo_m3taversal_oos_protocol.py

@@ -29,7 +29,7 @@ PROTOCOL_SCHEMA = "livingip.leoM3taversalOosProtocol.v3"
 TRIAL_SCORE_SCHEMA = "livingip.leoM3taversalOosTrialScore.v3"
 AGGREGATE_SCHEMA = "livingip.leoM3taversalOosAggregate.v3"
 GENERATOR_VERSION = "blinded-family-generator-v5"
-SCORER_VERSION = "invariant-reasoning-live-receipts-and-factual-ablation-v8"
+SCORER_VERSION = "invariant-reasoning-live-receipts-and-factual-ablation-v9"
 BASELINE_VERSION = "live-current-build-db-tool-ablation-v3"
 DEFAULT_TRIAL_COUNT = 3
 MEMORY_SCORER_IDS = frozenset({"OOS-07", "OOS-08"})

tests/test_hermes_leoclean_db_context_plugin.py

@@ -71,9 +71,41 @@ def test_runtime_persistence_contract_rejects_lost_session_jsonl() -> None:
         "state.db and the session JSONL persist on disk and survive a service restart.",
         contracts,
     )
+    process_local = module.response_contract_issues(
+        "Skills survive on disk. State.db and session JSONL are process-local, so a service recycle flushes the "
+        "live session buffer.",
+        contracts,
+    )
+    process_local_with_later_negation = module.response_contract_issues(
+        "State.db and session JSONL are process-local: a recycle flushes the live session buffer, so any in-flight "
+        "conversation state not yet flushed to a durable store is gone.",
+        contracts,
+    )
+    process_local_denied = module.response_contract_issues(
+        "State.db and session JSONL are not process-local; both persist across a restart.",
+        contracts,
+    )
+    process_local_not_true = module.response_contract_issues(
+        "It is not true that state.db and session JSONL are process-local; both persist across a restart.",
+        contracts,
+    )
+    process_local_neither = module.response_contract_issues(
+        "Neither state.db nor session JSONL is process-local; both persist across a restart.",
+        contracts,
+    )
+    process_local_neither_order = module.response_contract_issues(
+        "State.db and session JSONL are neither process-local nor in-memory; both persist across a restart.",
+        contracts,
+    )
 
     assert "runtime_persistence_contradiction" in bad
     assert "runtime_persistence_contradiction" in absent
+    assert "runtime_persistence_contradiction" in process_local
+    assert "runtime_persistence_contradiction" in process_local_with_later_negation
+    assert "runtime_persistence_contradiction" not in process_local_denied
+    assert "runtime_persistence_contradiction" not in process_local_not_true
+    assert "runtime_persistence_contradiction" not in process_local_neither
+    assert "runtime_persistence_contradiction" not in process_local_neither_order
     assert "runtime_persistence_contradiction" not in good
 
 
@@ -106,9 +138,105 @@ def test_forecast_contract_requires_reviewed_apply_boundary() -> None:
         "or invent the new fields in the current schema.",
         contracts,
     )
+    denied_gap = module.response_contract_issues(
+        "The missing resolution rule is not a schema gap in claims. Stage a reviewed schema proposal before apply.",
+        contracts,
+    )
+    unreviewed_type = module.response_contract_issues(
+        "Preserve the original forecast. A new claim may record the result as a typed belief or structural claim. "
+        "Stage a reviewed schema proposal before apply.",
+        contracts,
+    )
+    reversed_unreviewed_type = module.response_contract_issues(
+        "Preserve the original forecast. Stage a new observation claim and a reviewed schema proposal before apply.",
+        contracts,
+    )
 
     assert "forecast_review_apply_incomplete" in incomplete
     assert "forecast_review_apply_incomplete" not in complete
+    assert "forecast_resolution_gap_denied" in denied_gap
+    assert "forecast_unreviewed_claim_type" in unreviewed_type
+    assert "forecast_unreviewed_claim_type" in reversed_unreviewed_type
+
+
+def test_apply_receipt_contract_rejects_disclaimer_masking() -> None:
+    module = load_plugin()
+    contracts = [{"id": "apply_receipt_invariants"}]
+    contradictory_counts = (
+        "Aggregate counts need not change for unrelated updates. For this apply, database totals must change."
+    )
+    same_clause_counts = "Although aggregate counts need not change in general, database totals must change for this apply."
+    contradictory_surface = (
+        "Beliefs and behavioral_rules are written by the authorized apply. "
+        "approve_claim supports claims only. The current apply also writes beliefs."
+    )
+    same_clause_surface = "Although the current apply does not write beliefs, it writes behavioral_rules."
+    safe = (
+        "Aggregate counts need not change. The current apply does not write beliefs; belief updates remain staged "
+        "for a separate reviewed apply."
+    )
+
+    assert "apply_receipt_count_movement_required" in module.response_contract_issues(
+        contradictory_counts, contracts
+    )
+    assert "apply_receipt_count_movement_required" in module.response_contract_issues(same_clause_counts, contracts)
+    assert "apply_receipt_unsupported_surface" in module.response_contract_issues(
+        contradictory_surface, contracts
+    )
+    assert "apply_receipt_unsupported_surface" in module.response_contract_issues(same_clause_surface, contracts)
+    assert module.response_contract_issues(safe, contracts) == []
+
+
+def test_apply_receipt_contract_repairs_count_and_applyability_overclaims(tmp_path: Path, monkeypatch) -> None:
+    module = load_plugin()
+    trace = tmp_path / "trace.jsonl"
+    monkeypatch.setenv("LEO_DB_CONTEXT_TRACE_PATH", str(trace))
+    query = "Recall the blocker and give the canonical receipt plus applied_at readback."
+    contracts = [
+        {"id": "reply_budget", "hard_max_words": 220},
+        {"id": "apply_receipt_invariants"},
+    ]
+    module._store_snapshot(
+        "apply-receipt-session",
+        {
+            "status": "ok",
+            "query_sha256": module.hashlib.sha256(query.encode("utf-8")).hexdigest(),
+            "contracts": contracts,
+            "compiled_response": None,
+            "requires_database_truth": True,
+            "context": "fixture",
+        },
+    )
+    draft = (
+        "This first sentence intentionally precedes the memory anchor. Label: blind-ledger-demo. "
+        "Blocker: approved proposals still lack canonical row readback. "
+        "After an authorized apply, each public row, including beliefs and "
+        "behavioral_rules, must exist. The aggregate count readback must show totals that differ by the exact delta. "
+        + "Additional proposal detail remains subject to row-level verification. " * 24
+    )
+
+    result = module._post_llm_call(
+        session_id="apply-receipt-session",
+        user_message=query,
+        assistant_response=draft,
+    )
+
+    assert result is not None
+    delivered = result["assistant_response"]
+    assert delivered.startswith("Proof-boundary correction:")
+    assert "aggregate counts need not change" in delivered
+    assert "not directly applyable" in delivered
+    assert "Label: blind-ledger-demo" in delivered
+    assert "Blocker: approved proposals still lack canonical row readback" in delivered
+    assert "must show totals that differ" not in delivered
+    assert "including beliefs and behavioral_rules, must exist" not in delivered
+    assert module._word_count(delivered) <= 220
+    assert module.response_contract_issues(delivered, contracts) == []
+    records = [json.loads(line) for line in trace.read_text(encoding="utf-8").splitlines()]
+    assert records[-1]["apply_receipt_repair_applied"] is True
+    assert records[-1]["budget_compacted"] is True
+    assert records[-1]["post_repair_issues"] == ["reply_budget_exceeded"]
+    assert records[-1]["delivered_issues"] == []
 
 
 def test_claim_challenge_contract_requires_exact_bound_claim_and_evidence() -> None:
@@ -249,6 +377,9 @@ def test_plugin_injects_bounded_retrieval_rows_and_writes_body_redacted_trace(tm
     assert "inspect the exact retrieved claim body, its evidence, and relevant edges" in context
     assert "Conversation statements may motivate a candidate but are not provenance" in context
     assert "Keep every candidate review-only" in context
+    assert "aggregate table counts need not change" in context
+    assert "applied_at is proposal-level proof" in context
+    assert "belief updates, behavioral_rules, governance_gates" in context
     assert "natural prose rather than a fixed benchmark template" in context
     assert "must not be injected" not in context
     assert "NEVER_INCLUDE_RAW_UNUSED_FIELD" not in context

tests/test_hermes_leoclean_kb_bridge_source.py

@@ -728,6 +728,28 @@ def test_vps_bridge_operational_contracts_are_query_specific_and_schema_exact()
     }
     assert {"reply_budget", "runtime_persistence"} <= runtime_ids
 
+    receipt_queries = [
+        "Recall the same blocker and give the canonical receipt plus applied_at postflight readback.",
+        "Return Closure proof for the same approved-versus-applied canonical gap.",
+        "Say which before/after canonical receipt and applied_at readback would resolve it.",
+    ]
+    for query in receipt_queries:
+        receipt_ids = {item["id"] for item in module.operational_contracts(query)}
+        assert "apply_receipt_invariants" in receipt_ids
+    receipt_contract = next(
+        item
+        for item in module.operational_contracts("Give the applied_at postflight proof.")
+        if item["id"] == "apply_receipt_invariants"
+    )
+    assert receipt_contract["approve_claim_supported"] == [
+        "claims",
+        "sources",
+        "evidence",
+        "edges",
+        "reasoning_tools",
+    ]
+    assert "aggregate table counts need not change" in receipt_contract["count_boundary"]
+
 
 def test_vps_bridge_binds_claim_challenge_to_exact_retrieved_rows() -> None:
     module = _load_module(BRIDGE_DIR / "kb_tool.py")
@@ -956,6 +978,16 @@ def test_vps_bridge_direct_intents_reject_unrelated_and_resolve_overlaps() -> No
         )
     }
     assert chat_memory_ids == {"reply_budget"}
+    memory_set_queries = [
+        "Remember the approved-versus-applied blocker as TEMP for the next turn only. Chat memory only.",
+        "Bind the approved-versus-applied canonical gap to TEMP until my next question; no staging or apply.",
+        "Name the blocker precisely enough to distinguish approval, applied_at, and canonical readback. It must not "
+        "become a source, memory record, or database write.",
+    ]
+    assert all(
+        "apply_receipt_invariants" not in {item["id"] for item in module.operational_contracts(query)}
+        for query in memory_set_queries
+    )
 
 
 def test_vps_bridge_named_document_question_matches_exact_proposal_and_compiles_receipt(monkeypatch) -> None:

tests/test_working_leo_m3taversal_oos_benchmark.py

@@ -510,8 +510,9 @@ def test_oos_direct_apply_case_rejects_approved_rows_without_readiness_gap() ->
     assert bad["pass"] is False
     assert bad["proposal_readiness_issues"] == ["approved_proposal_applyability_overclaim"]
 
-    repaired = bad_reply + (
+    repaired = (
-        " They are approved_needs_apply_payload, so normalize a strict payload and review it before any apply."
+        "No. Three proposals are approved with applied_at NULL. They are approved_needs_apply_payload, so normalize "
+        "a strict payload and review it before any apply."
     )
     assert benchmark.proposal_readiness_issues(prompt["id"], repaired) == []
 
@@ -841,6 +842,17 @@ def test_oos_runtime_reasoning_accepts_live_tiered_reply() -> None:
     assert benchmark.matched_concept(headed, "runtime_inputs") is True
     assert benchmark.matched_concept(headed, "proof_tiers") is True
 
+    four_layer = (
+        "DB totals prove no rows were added or deleted, but they do not prove row contents are unchanged or that "
+        "the agent answered consistently; compare row IDs and timestamps. Runtime configuration persists in skills, "
+        "SOUL.md, and Hermes config. Persisted conversation state lives in state.db and session JSONL; both survive "
+        "a restart. Telegram-visible proof requires a delivered message ID."
+    )
+    assert benchmark.matched_concept(four_layer, "runtime_inputs") is True
+    assert benchmark.matched_concept(four_layer, "durable_session_continuity") is True
+    assert benchmark.matched_concept(four_layer, "row_content_proof") is True
+    assert benchmark.matched_concept(four_layer, "proof_tiers") is True
+
 
 def test_oos_runtime_reasoning_rejects_durable_session_as_ephemeral() -> None:
     reply = "state.db survives, but the session JSONL is lost when the session closes."
@@ -904,6 +916,15 @@ def test_oos_agent_positions_accept_attribution_without_literal_column_name() ->
     assert benchmark.matched_concept(live_canonical, "shared_knowledge_commons") is True
     assert benchmark.matched_concept(live_canonical, "agent_specific_positions") is True
 
+    share_the_fact = (
+        "Share the fact. Duplicate nothing. One claim row holds the observation, and both agents use the same "
+        "source evidence. Each agent's position goes into a public.beliefs row with agent_id and confidence. "
+        "Beliefs has no claim-ID foreign key, so the disagreement is queryable by convention; claim_edges cannot "
+        "link a belief to a claim."
+    )
+    assert benchmark.matched_concept(share_the_fact, "shared_knowledge_commons") is True
+    assert benchmark.matched_concept(share_the_fact, "agent_specific_positions") is True
+
 
 def test_oos_memory_accepts_concrete_noncanonical_blocker_and_restatement() -> None:
     token = "demo-ledger-deadbeef"
@@ -932,6 +953,24 @@ def test_oos_memory_accepts_concrete_noncanonical_blocker_and_restatement() -> N
     assert benchmark.matched_concept("Blocker: something is wrong.", "blocker_definition") is False
 
 
+def test_oos_memory_recall_accepts_apply_receipt_correction_without_losing_the_token() -> None:
+    token = "demo-ledger-deadbeef"
+    prompt = next(item for item in benchmark.prompt_catalog(token) if item["id"] == "OOS-08")
+    reply = (
+        "Proof-boundary correction: aggregate counts need not change. Approved content is not directly applyable "
+        "when the payload or write surface is unsupported; belief updates and behavioral_rules remain staged for a "
+        "separate reviewed apply capability. "
+        f"Label: {token}. Blocker: approved proposals still have applied_at null and no canonical row readback. "
+        "After an authorized apply, verify proposal-level applied_at plus before/after row IDs and content hashes."
+    )
+
+    score = benchmark.score_reply(prompt, reply, memory_token=token)
+
+    assert score["pass"] is True
+    assert score["invalid_count_invariant_detected"] is False
+    assert score["proposal_readiness_issues"] == []
+
+
 def test_oos_forecast_accepts_natural_missing_resolution_language() -> None:
     reply = (
         "Preserve the original 60% forecast; it lacked success criteria. The schema has no outcome or resolved "
@@ -977,6 +1016,55 @@ def test_oos_schema_overclaim_accepts_anaphoric_semicolon_denial() -> None:
     assert benchmark.current_schema_overclaims(unsafe) == ["invalid_current_edge_type"]
 
 
+def test_oos_schema_overclaim_distinguishes_claim_body_readback_from_a_body_column() -> None:
+    readback = (
+        "A fresh query on the claim ID returns the expected row with matching body, confidence, and source links."
+    )
+    plural_readback = "The retrieved claims have matching body text and source links."
+    qualified_plural_readback = "Fresh public.claims rows have matching body text and source links."
+    field_claim = "The public.claims table has a body field."
+
+    assert benchmark.current_schema_overclaims(readback) == []
+    assert benchmark.current_schema_overclaims(plural_readback) == []
+    assert benchmark.current_schema_overclaims(qualified_plural_readback) == []
+    assert benchmark.current_schema_overclaims(field_claim) == ["claims_unshipped_fields"]
+
+
+def test_oos_count_invariant_rejects_a_later_contradiction_despite_a_disclaimer() -> None:
+    wrong = "The aggregate count readback must show totals that differ by the exact payload delta."
+    contradictory = (
+        "Aggregate counts need not change for unrelated updates. For this apply, database totals must change."
+    )
+    same_clause = "Although aggregate counts need not change in general, database totals must change for this apply."
+    safe = "Aggregate counts need not change because in-place updates preserve counts."
+
+    assert benchmark.asserts_invalid_count_invariant(wrong) is True
+    assert benchmark.asserts_invalid_count_invariant(contradictory) is True
+    assert benchmark.asserts_invalid_count_invariant(same_clause) is True
+    assert benchmark.asserts_invalid_count_invariant(safe) is False
+
+
+def test_oos_applyability_rejects_unsupported_surface_despite_a_disclaimer() -> None:
+    reply = (
+        "Beliefs and behavioral_rules are written by the authorized apply. "
+        "approve_claim supports claims only. The current apply also writes beliefs."
+    )
+    safe = "The current apply does not write beliefs; belief updates remain staged for a separate reviewed apply."
+    same_clause = "Although the current apply does not write beliefs, it writes behavioral_rules."
+
+    assert benchmark.proposal_readiness_issues("OOS-08", reply) == ["unsupported_apply_surface_overclaim"]
+    assert benchmark.proposal_readiness_issues("OOS-08", same_clause) == ["unsupported_apply_surface_overclaim"]
+    assert benchmark.proposal_readiness_issues("OOS-08", safe) == []
+
+
+def test_oos_schema_overclaim_rejects_reverse_unreviewed_claim_types() -> None:
+    observation = "Create a new observation claim in public.claims."
+    hypothesis = "Create a new hypothesis claim in public.claims."
+
+    assert benchmark.current_schema_overclaims(observation) == ["unreviewed_claim_type"]
+    assert benchmark.current_schema_overclaims(hypothesis) == ["unreviewed_claim_type"]
+
+
 def test_oos_challenge_accepts_plural_and_evidence_bounded_language() -> None:
     reply = (
         "Challenge: the sources and claim bodies do not establish causality; the conclusion is not paid for by "