Harden Leo benchmark semantic invariants

scripts/working_leo_m3taversal_oos_benchmark.py

@@ -444,7 +444,11 @@ 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"(?:single|one) canonical assertion|"
+            r"deduplicate.{0,80}(?:single|one) canonical (?:assertion|record)|"
             r"share the fact.{0,100}one claim row|"
+            r"keep (?:the )?(?:fact|claim|proposition) (?:canonical )?once|"
+            r"one claim row.{0,80}(?:holds|contains|records).{0,60}(?:agreed|shared|factual) (?:fact|claim|proposition)|"
             r"fact shared.{0,80}(?:one|single).{0,40}(?:claim|public\.claims)",
             re.I | re.S,
         ),
@@ -454,7 +458,9 @@ CONCEPT_PATTERNS: dict[str, tuple[re.Pattern[str], ...]] = {
             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)",
+            r"one claim row.{0,180}(?:(?:both|each) agents?'|shared) (?:sources?|evidence|claim_evidence)|"
+            r"(?:both|each) (?:agents?|analysts?)' (?:sources?|source evidence|evidence)|"
+            r"both (?:agents?|analysts?)' evidence.{0,80}(?:one|single) (?:record|assertion|claim)",
             re.I | re.S,
         ),
     ),
@@ -495,11 +501,20 @@ CONCEPT_PATTERNS: dict[str, tuple[re.Pattern[str], ...]] = {
     ),
     "forecast_schema_gap": (
         re.compile(
-            r"current (?:v1|schema|columns?|fields?|edge types?)|public\.claims|the schema|schema has|claims table",
+            r"current (?:v1|schema|columns?|fields?|edge types?|(?:public\.)?claims schema|edge_type enum)|"
+            r"public\.claims|the schema|schema has|claims table",
             re.I,
         ),
         re.compile(
-            r"no.{0,50}(?:forecast[- ]resolution|resolution field|resolved_at)|does not have.{0,50}resolution|"
+            r"(?:there is |the schema has |(?:public\.)?claims has )?no "
+            r"(?:forecast[- ]resolution|resolution fields?|resolved_at)"
+            r"(?!\s+(?:migration|review|proposal|addition|change|work)\b)|"
+            r"does not have.{0,50}resolution|"
+            r"(?:schema|claims).{0,80}lacks?.{0,60}(?:resolution criteria|resolution_criteria|outcome|resolved_at)|"
+            r"(?:schema|claims).{0,80}(?:does not|doesn't|cannot) (?:model|express|represent|record|store)"
+            r".{0,60}(?:resolution criteria|resolution_criteria|outcome|resolved_at)|"
+            r"no (?:column|field) (?:on|in) .{0,100}(?:records?|stores?|contains?).{0,100}"
+            r"(?:resolution criteria|outcome|resolved_at)|"
             r"no.{0,80}(?:resolved|outcome|resolution_criteria).{0,30}field|"
             r"(?:forecast[- ]resolution|resolution field|resolved_at).{0,80}"
             r"(?:does not exist|is absent|is not present|isn't present)|"
@@ -511,7 +526,15 @@ CONCEPT_PATTERNS: dict[str, tuple[re.Pattern[str], ...]] = {
             re.I | re.S,
         ),
         re.compile(
-            r"resolves.{0,220}(?:not|isn'?t|does not|doesn't|absent)|no.{0,30}resolves",
+            r"resolves (?:edge|type).{0,100}(?:does not exist|is absent|is missing|is not present|isn't present|"
+            r"is not current|isn't current|is unsupported)|"
+            r"(?:there is|schema has|enum has|edge types? (?:has|have)|(?:current )?edge_type enum has) "
+            r"no `?resolves`?(?: (?:edge|type))?|"
+            r"no `?resolves`? (?:edge|type)(?: exists| is present)?|"
+            r"(?:edge_type|edge type|enum) lacks? resolves|"
+            r"(?:edge_type|edge type|enum) (?:cannot|does not|doesn't) (?:express|represent|support) resolves|"
+            r"(?:do not|don't|must not) invent (?:an? )?resolves (?:edge|type)|"
+            r"resolves (?:edge|type).{0,120}neither (?:exists|is present)",
             re.I | re.S,
         ),
     ),
@@ -655,15 +678,18 @@ 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|missing|"
-    r"has no|have no|there is no|no column|not an edge|would require|schema gap|must be added|"
+    r"has no|have no|lacks?|without|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|"
     r"schema extension|extension proposal|reviewed schema proposal|not (?:in|part of|among) the current)\b|"
+    r"\b(?:does not|doesn't|cannot)\s+"
+    r"(?:record|store|contain|include|have|expose|provide|support|model|express|represent)\b|"
     r"\b(?:requires?|needs?)\s+either\s+(?:an?\s+)?"
     r"(?:[a-z_]+\s+){0,3}(?:column|field|table|edge type)\b|"
     r"\b(?:would add|would introduce|would create)\s+(?:an?\s+)?"
     r"(?:[a-z_]+\s+){0,3}(?:column|field|table|edge type)\b|"
     r"\b(?:proposal|extension)\b.{0,100}\b(?:column|field|table|edge type)\b|"
-    r"\bno\s+(?:[a-z_]+\s+){0,3}(?:column|field|table|edge type)\b",
+    r"\bno\s+(?:[a-z_]+\s+){0,3}(?:column|field|table|edge type)\b"
+    r"(?!\s+(?:review|proposal|change|addition|work)\b)",
     re.I,
 )
 SCHEMA_CLAUSE_BOUNDARY_RE = re.compile(
@@ -674,12 +700,23 @@ SCHEMA_CLAUSE_BOUNDARY_RE = re.compile(
 )
 CURRENT_SCHEMA_ASSERTION_PATTERNS: dict[str, re.Pattern[str]] = {
     "claims_unshipped_fields": re.compile(
-        r"claims\s+(?:table|schema)\b.{0,80}\b(?:body|metadata|forecast[_ -]resolution|resolved_at|falsifier)\b|"
+        r"(?:the )?current schema\b.{0,120}\b(?:forecast[_ -]resolution|resolution[_ -]criteria|outcome|"
-        r"public\.claims\s+(?:stores?|has|contains?)\s+(?:an?\s+)?"
+        r"resolved_at|falsifier)\b|"
-        r"(?:body|metadata|forecast[_ -]resolution|resolved_at|falsifier)\b|"
+        r"public\.claims\b.{0,120}\b(?:forecast[_ -]resolution|resolution[_ -]criteria|outcome|resolved_at|"
+        r"falsifier)\b|"
+        r"(?:the )?current schema\s+(?:already\s+)?"
+        r"(?:stores?|has|contains?|records?|exposes?|provides?|supports?|includes?|models?|represents?)\s+.{0,80}"
+        r"\b(?:forecast[_ -]resolution|resolution[_ -]criteria|outcome|resolved_at|falsifier)\b|"
+        r"claims\s+(?:table|schema)\b.{0,100}\b(?:body|metadata|forecast[_ -]resolution|"
+        r"resolution[_ -]criteria|outcome|resolved_at|falsifier)\b|"
+        r"public\.claims\s+(?:(?:natively|already|currently)\s+)?"
+        r"(?:stores?|has|contains?|records?|exposes?|provides?|supports?|includes?|models?|represents?)\s+"
+        r"(?:an?\s+)?"
+        r"(?:body|metadata|forecast[_ -]resolution|resolution[_ -]criteria|outcome|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"\b(?:body|metadata|forecast[_ -]resolution|resolution[_ -]criteria|outcome|resolved_at|falsifier)\b|"
-        r"public\.claims\b.{0,80}\b(?:body|metadata|forecast[_ -]resolution|resolved_at|falsifier)\b.{0,30}"
+        r"public\.claims\b.{0,80}\b(?:body|metadata|forecast[_ -]resolution|resolution[_ -]criteria|"
+        r"outcome|resolved_at|falsifier)\b.{0,30}"
         r"\b(?:column|field)\b",
         re.I,
     ),
@@ -689,9 +726,15 @@ CURRENT_SCHEMA_ASSERTION_PATTERNS: dict[str, re.Pattern[str]] = {
         re.I,
     ),
     "invalid_current_edge_type": re.compile(
+        r"\b(?:current )?edge_type enum\b.{0,80}\bresolves\b|"
+        r"\bcurrent edge_type\b.{0,80}\bresolves\b|"
         r"(?:\b(?:claim_edges?|edge type|edge)\b\s*(?:is|=|:|named|called|of)?\s*[`'\"]?"
         r"(?:superseded_by|relates_to|resolves|derived_from)\b|"
-        r"\b(?:superseded_by|relates_to|resolves|derived_from)\b[`'\"]?\s+(?:claim_)?edge\b)",
+        r"\b(?:superseded_by|relates_to|resolves|derived_from)\b[`'\"]?\s+(?:claim_)?edge\b|"
+        r"\b(?:current )?edge_type enum\b.{0,50}\b(?:contains|has|includes|supports|enumerates)\b.{0,20}"
+        r"\bresolves\b|"
+        r"\b(?:current )?edge_type\b.{0,50}\b(?:contains|has|includes|supports|enumerates)\b.{0,20}"
+        r"\bresolves\b)",
         re.I,
     ),
     "unshipped_edge_rationale": re.compile(r"\bclaim_edges?\b.{0,100}\brationale\b", re.I),
@@ -721,6 +764,17 @@ CURRENT_SCHEMA_ASSERTION_PATTERNS: dict[str, re.Pattern[str]] = {
     ),
 }
 UNVERIFIED_M3TAVERSAL_ALIAS_RE = re.compile(r"\b(?:Cory|m3ta)\b", re.I)
+RESOLVES_PRESENCE_ASSERTION_RE = re.compile(
+    r"\bresolves (?:edge|type)\b.{0,120}\b(?:is not (?:absent|missing)|already (?:exists|supported|present)|"
+    r"is (?:already )?supported by (?:the )?(?:current )?edge_type)\b|"
+    r"\b(?:current )?edge_type\b.{0,60}\b(?:already )?(?:contains|has|includes|supports|enumerates)\b.{0,20}"
+    r"\bresolves\b",
+    re.I | re.S,
+)
+EDGE_TYPE_RESOLVES_DENIAL_RE = re.compile(
+    r"\b(?:current )?edge_type\b.{0,50}\b(?:does not|doesn't|cannot|lacks?|has no)\b.{0,40}\bresolves\b",
+    re.I | re.S,
+)
 
 SOURCE_EVIDENCE_CANONICAL_OBJECT_RE = re.compile(r"claim_evidence|public\.sources|source rows?", re.I)
 SOURCE_EVIDENCE_LOCATOR_GAP_RE = re.compile(
@@ -886,6 +940,41 @@ BLOCKER_DIAGNOSTIC_RE = re.compile(
     r"source|evidence|claim|proposal|runtime|restart|identity|tool|readback)\b",
     re.I,
 )
+BLOCKER_FAILURE_STATE_RE = re.compile(
+    r"\b(?:no|none|zero|missing|absent|null|unavailable|unsupported|unapplied|not|cannot|can't|blocked|"
+    r"fails?|stale|wrong|gap|lacks?|without)\b",
+    re.I,
+)
+BLOCKER_ABSENCE_RE = re.compile(
+    r"\b(?:there is |there's )?no (?:[a-z][a-z-]*\s+){0,4}blocker "
+    r"(?:exists|remains|is present)\b|\bnothing (?:is )?blocked\b",
+    re.I,
+)
+PER_AGENT_CLAIM_DUPLICATION_RE = re.compile(
+    r"\b(?:(?:one|separate|distinct|individual)\s+)?(?:public\.)?claims?(?:\s+row)?\s+per\s+agent\b|"
+    r"\b(?:claim|public\.claims)(?:\s+row)?(?:\s+once)?\s+(?:for|to)\s+each\s+agent\b|"
+    r"\beach\s+agent\s+(?:gets?|receives?|has|stores?|uses?)\s+"
+    r"(?:(?:its|their)\s+own|an?\s+separate|one)\s+(?:public\.)?claims?(?:\s+row)?\b|"
+    r"\b(?:give|assign)\s+each\s+agent\s+(?:(?:its|their)\s+own|an?\s+separate|one)\s+"
+    r"(?:public\.)?claims?(?:\s+row)?\b|"
+    r"\bone\s+(?:public\.)?claims?(?:\s+row)?\s+for\s+[A-Z][a-z0-9_-]+\s+and\s+one\s+"
+    r"(?:(?:public\.)?claims?(?:\s+row)?\s+)?for\s+[A-Z][a-z0-9_-]+\b|"
+    r"\b(?:fact|proposition|claim|assertion)\b.{0,80}\b(?:under|within|inside)\s+each\s+agent\b|"
+    r"\b(?:their|the agents?'?)\s+respective\s+(?:copies|claim rows?|records?)\b",
+    re.I,
+)
+PER_AGENT_DUPLICATION_PREFIX_DENIAL_RE = re.compile(
+    r"(?:\b(?:do not|don't|does not|doesn't|should not|shouldn't|must not|never|avoid)\b\s*"
+    r"(?:(?:create|store|write|give|assign|duplicate|fork|make|keep|use|add)(?:ing)?\s+)?"
+    r"(?:(?:the|an?|one)\s+)?(?:(?:factual|canonical|shared|separate|distinct)\s+)?|"
+    r"\b(?:instead of|rather than|not)\s+)$",
+    re.I,
+)
+PER_AGENT_DUPLICATION_SUFFIX_DENIAL_RE = re.compile(
+    r"^.{0,80}\b(?:is|would be|remains?)\s+(?:wrong|incorrect|invalid|duplicative)\b|"
+    r"^.{0,80}\b(?:creates?|causes?|would create)\s+(?:data\s+)?(?:divergence|fragmentation)\b",
+    re.I,
+)
 
 
 def prompt_catalog(memory_token: str) -> list[dict[str, Any]]:
@@ -902,6 +991,11 @@ def extract_blocker_clause(reply: str) -> str | None:
     for pattern in (
         re.compile(r"\bblocker(?:\s+restated)?\s*:\s*(?P<value>[^\n]+)", re.I),
         re.compile(r"\bblocker\s+(?:is|was)\s+(?P<value>[^\n]+)", re.I),
+        re.compile(
+            r"\b(?:mnemonic|label)(?:\s+(?:tag|retrieved))?\s*:?[^.\n]{0,140}?"
+            r"(?:[.]|\s+(?:-|\u2014)\s+)\s*(?P<value>[^\n]+)",
+            re.I,
+        ),
     ):
         match = pattern.search(reply)
         if match:
@@ -920,11 +1014,120 @@ def blocker_terms(value: str | None, *, memory_token: str) -> set[str]:
     }
 
 
+def blocker_signature(value: str | None) -> set[str]:
+    """Extract polarity-aware blocker categories for same-session recall."""
+
+    if not value:
+        return set()
+    signatures: set[str] = set()
+    clauses = [
+        clause.strip()
+        for clause in re.split(r"(?<=[.!?;])\s+|\n+|,\s+(?:but|and|so)\s+", value, flags=re.I)
+        if clause.strip()
+    ]
+    if re.search(r"\b(?:approved|reviewer approval|reviewer signature|reviewer sign[- ]off)\b", value, re.I):
+        signatures.add("reviewer_approval")
+    applied_missing = re.compile(
+        r"\bapproved[- ]but[- ]not[- ]applied\b|\b(?:not applied|unapplied)\b|"
+        r"\b(?:zero|no|missing|absent)\s+applied_at(?:\s+values?)?\b|"
+        r"\bapplied_at\b.{0,20}\b(?:unset|missing|absent|zero|none|null|not set)\b|"
+        r"\b(?:unset|missing|absent|zero|none|null)\b.{0,20}\bapplied_at\b",
+        re.I,
+    )
+    applied_present = re.compile(
+        r"\bapplied_at\b.{0,30}\b(?:non[- ]null|not null|exists|present|is set|has a value|"
+        r"missing (?:flag )?(?:is|=) false|is not missing)\b",
+        re.I,
+    )
+    if any(applied_missing.search(clause) and not applied_present.search(clause) for clause in clauses):
+        signatures.add("applied_missing")
+    canonical_missing = re.compile(
+        r"\bcanonical (?:gap|row|readback)\b.{0,30}\b(?:missing|absent|none|null|zero|not (?:live|written)|"
+        r"unavailable)\b|"
+        r"\b(?:no|zero|missing|absent)\b.{0,25}\b(?:canonical|public\.\*)\b.{0,25}\brows?\b|"
+        r"\b(?:no|zero|missing|absent)\s+public\.\*\s+row(?: ID)?\b|"
+        r"\bno rows?\b.{0,30}\b(?:landed|written|applied)\b.{0,20}\bpublic\.(?:claims|sources|\*)\b|"
+        r"\bapproved[- ]but[- ]not[- ]applied canonical gap\b|\bcanonical gap\b",
+        re.I,
+    )
+    canonical_present = re.compile(
+        r"\bcanonical (?:row|readback)\b.{0,20}\b(?:exists|present|current|complete|is live)\b|"
+        r"\bcanonical rows?\b.{0,20}\b(?:exist|are present|are current|are complete)\b|"
+        r"\bcanonical gap\b.{0,20}\b(?:is|was|has been|now)\s+(?:closed|resolved|cleared|fixed)\b",
+        re.I,
+    )
+    if any(canonical_missing.search(clause) and not canonical_present.search(clause) for clause in clauses):
+        signatures.add("canonical_missing")
+    telegram_missing = re.compile(
+        r"\b(?:telegram|delivery|visible message|message receipt)\b.{0,50}"
+        r"\b(?:no|missing|absent|blocked|unproven|not proven|unavailable)\b|"
+        r"\b(?:no|missing|absent|blocked|unproven|not proven|unavailable)\b.{0,50}"
+        r"\b(?:telegram|delivery|visible message|message receipt)\b",
+        re.I,
+    )
+    if any(telegram_missing.search(clause) for clause in clauses):
+        signatures.add("telegram_missing")
+    if re.search(
+        r"\b(?:beliefs?\b.{0,40})?(?:has|have|with)\s+no\s+claim[_ -]?id\s+foreign key\b|"
+        r"\b(?:missing|absent|no)\b.{0,50}\b(?:foreign key|belief[- ]to[- ]claim link)\b|"
+        r"\bschema gap\b",
+        value,
+        re.I | re.S,
+    ):
+        signatures.add("schema_link_missing")
+    if (
+        re.search(r"\bbeliefs?(?:\.claim[_ -]?id)?\b", value, re.I)
+        and re.search(r"\bclaim[_ -]?id\b", value, re.I)
+        and re.search(r"\b(?:foreign key|FK)\b", value, re.I)
+        and BLOCKER_FAILURE_STATE_RE.search(value)
+    ):
+        signatures.add("belief_claim_foreign_key_missing")
+    if (
+        re.search(r"\bbeliefs?(?:\.claim[_ -]?id)?\b", value, re.I)
+        and re.search(r"\bclaim[_ -]?id\b", value, re.I)
+        and re.search(r"\bindex\b", value, re.I)
+        and BLOCKER_FAILURE_STATE_RE.search(value)
+    ):
+        signatures.add("belief_claim_index_missing")
+    if re.search(
+        r"\bpublic\.sources\b.{0,80}\b(?:no|missing|absent|lacks?)\b.{0,40}\bforeign key\b|"
+        r"\b(?:no|missing|absent)\b.{0,40}\bforeign key\b.{0,80}\bpublic\.sources\b",
+        value,
+        re.I | re.S,
+    ):
+        signatures.add("source_foreign_key_missing")
+    return signatures
+
+
+def asserts_per_agent_claim_duplication(reply: str) -> bool:
+    """Reject affirmative per-agent duplication while allowing explicit denials."""
+
+    clauses = re.split(r"(?<=[.!?;])\s+|\n+|,\s+(?:but|and|so)\s+", reply, flags=re.I)
+    for clause in clauses:
+        for match in PER_AGENT_CLAIM_DUPLICATION_RE.finditer(clause):
+            prefix = clause[max(0, match.start() - 80) : match.start()]
+            suffix = clause[match.end() : match.end() + 100]
+            if PER_AGENT_DUPLICATION_PREFIX_DENIAL_RE.search(prefix):
+                continue
+            if PER_AGENT_DUPLICATION_SUFFIX_DENIAL_RE.search(suffix):
+                continue
+            return True
+    return False
+
+
 def matched_concept(reply: str, concept: str) -> bool:
     if concept == "blocker_definition":
         blocker = extract_blocker_clause(reply)
         terms = blocker_terms(blocker, memory_token="")
-        return bool(blocker and len(terms) >= 3 and BLOCKER_DIAGNOSTIC_RE.search(blocker))
+        return bool(
+            blocker
+            and len(terms) >= 3
+            and BLOCKER_DIAGNOSTIC_RE.search(blocker)
+            and BLOCKER_FAILURE_STATE_RE.search(blocker)
+            and not BLOCKER_ABSENCE_RE.search(blocker)
+        )
+    if concept == "shared_knowledge_commons" and asserts_per_agent_claim_duplication(reply):
+        return False
     if concept == "claim_body_evidence_distinction":
         if CLAIM_EVIDENCE_CONFLATION_RE.search(reply):
             return False
@@ -963,6 +1166,8 @@ def current_schema_overclaims(reply: str) -> list[str]:
     findings: list[str] = []
     sentence_segments = re.split(r"(?<=[.!?])\s+|\n+", reply)
     for sentence in sentence_segments:
+        if RESOLVES_PRESENCE_ASSERTION_RE.search(sentence) and not EDGE_TYPE_RESOLVES_DENIAL_RE.search(sentence):
+            findings.append("invalid_current_edge_type")
         # Negation and future-schema qualifiers apply to their clause, not to a
         # contradictory assertion in a later conjunction in the same sentence.
         clauses = SCHEMA_CLAUSE_BOUNDARY_RE.split(sentence)
@@ -1211,11 +1416,25 @@ def score_results(
     memory_source_reply = str((by_result.get(memory_source_id) or {}).get("reply") or "")
     memory_recall_reply = str((by_result.get(memory_recall_id) or {}).get("reply") or "")
     source_clause = extract_blocker_clause(memory_source_reply)
+    recall_clause = extract_blocker_clause(memory_recall_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)
+    recall_terms = blocker_terms(recall_clause, 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)
+    source_signature = blocker_signature(source_clause)
+    recall_signature = blocker_signature(recall_clause)
+    source_failure_signature = source_signature - {"reviewer_approval"}
+    recall_failure_signature = recall_signature - {"reviewer_approval"}
+    signature_overlap = source_failure_signature & recall_failure_signature
+    required_signature_overlap = min(2, len(source_failure_signature)) if source_failure_signature else 0
+    same_blocker_recalled = bool(
+        source_clause
+        and (
+            len(signature_overlap) >= required_signature_overlap and len(overlap_terms) >= required_overlap
+            if source_failure_signature
+            else len(overlap_terms) >= required_overlap
+        )
+    )
     for score in scores:
         if score["prompt_id"] != memory_recall_id:
             continue
@@ -1227,6 +1446,10 @@ def score_results(
         "source_blocker_terms": sorted(source_terms),
         "recall_overlap_terms": sorted(overlap_terms),
         "required_overlap": required_overlap,
+        "source_blocker_signature": sorted(source_signature),
+        "recall_blocker_signature": sorted(recall_signature),
+        "signature_overlap": sorted(signature_overlap),
+        "required_signature_overlap": required_signature_overlap,
         "same_blocker_recalled": same_blocker_recalled,
     }
     return {

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-v9"
+SCORER_VERSION = "invariant-reasoning-live-receipts-and-factual-ablation-v10"
 BASELINE_VERSION = "live-current-build-db-tool-ablation-v3"
 DEFAULT_TRIAL_COUNT = 3
 MEMORY_SCORER_IDS = frozenset({"OOS-07", "OOS-08"})
@@ -403,6 +403,14 @@ def harness_git_head() -> str:
     ).strip()
 
 
+def harness_worktree_clean() -> bool:
+    return not subprocess.check_output(
+        ["git", "status", "--porcelain", "--untracked-files=no"],
+        cwd=ROOT,
+        text=True,
+    ).strip()
+
+
 def instrument_db_context_plugin_source(source: str) -> str:
     marker = """    safe["receipt_sha256"] = hashlib.sha256(
         json.dumps(value, sort_keys=True, separators=(",", ":")).encode("utf-8")
@@ -562,6 +570,8 @@ def freeze_protocol(
 ) -> dict[str, Any]:
     if trial_count < 3:
         raise ValueError("at least three trials are required for clean/restart variance")
+    if not harness_worktree_clean():
+        raise RuntimeError("refusing to freeze a protocol from a dirty tracked worktree")
     modes = ["clean_session"] * (trial_count - 1) + ["post_restart_clean_session"]
     trials = [build_blinded_trial(seed, index, session_mode=modes[index]) for index in range(trial_count)]
     protocol: dict[str, Any] = {
@@ -610,12 +620,15 @@ def freeze_protocol(
             "no_supplied_row_ids": True,
             "prompt_variants_per_family": min(len(family["variants"]) for family in BLINDED_FAMILIES),
         },
+        "harness_worktree_clean_at_freeze": True,
         "harness_git_head": harness_git_head(),
         "source_hashes": {key: file_sha256(path) for key, path in source_paths().items()},
         "trials": trials,
     }
     protocol["protocol_hash_sha256"] = canonical_sha256(protocol)
-    validate_protocol(protocol, verify_source_hashes=True)
+    validation = validate_protocol(protocol, verify_source_hashes=True)
+    if not validation["pass"]:
+        raise RuntimeError("frozen protocol failed validation: " + ", ".join(validation["issues"]))
     return protocol
 
 
@@ -631,6 +644,8 @@ def validate_protocol(protocol: dict[str, Any], *, verify_source_hashes: bool) -
         issues.append("wrong_baseline_version")
     if not _valid_git_revision(protocol.get("harness_git_head")):
         issues.append("invalid_harness_git_head")
+    if protocol.get("harness_worktree_clean_at_freeze") is not True:
+        issues.append("harness_worktree_not_clean_at_freeze")
     supplied_hash = protocol.get("protocol_hash_sha256")
     unhashed = {key: value for key, value in protocol.items() if key != "protocol_hash_sha256"}
     if supplied_hash != canonical_sha256(unhashed):
@@ -700,6 +715,8 @@ def validate_protocol(protocol: dict[str, Any], *, verify_source_hashes: bool) -
         if len(seen) < min(3, len(trials)):
             issues.append(f"variant_repetition:{family_id}")
     if verify_source_hashes:
+        if not harness_worktree_clean():
+            issues.append("harness_worktree_dirty")
         if protocol.get("harness_git_head") != harness_git_head():
             issues.append("harness_git_head_changed_after_freeze")
         source_hashes = protocol.get("source_hashes") or {}

tests/test_run_leo_m3taversal_oos_handler_suite.py

@@ -14,6 +14,11 @@ import run_leo_m3taversal_oos_handler_suite as suite  # noqa: E402
 import working_leo_m3taversal_oos_protocol as protocol_lib  # noqa: E402
 
 
+@pytest.fixture(autouse=True)
+def clean_protocol_worktree(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(protocol_lib, "harness_worktree_clean", lambda: True)
+
+
 def protocol() -> dict:
     return protocol_lib.freeze_protocol(
         "runner-test-seed",

tests/test_working_leo_m3taversal_oos_benchmark.py

@@ -313,6 +313,11 @@ def test_oos_schema_guard_allows_explicit_future_schema_gap() -> None:
         "would require a schema gap proposal."
     )
     assert benchmark.current_schema_overclaims(reply) == []
+    assert benchmark.current_schema_overclaims(
+        "The current public.claims schema does not record outcome or resolution criteria."
+    ) == []
+    assert benchmark.current_schema_overclaims("The current edge_type enum does not contain resolves.") == []
+    assert benchmark.current_schema_overclaims("The current claims schema lacks an outcome field.") == []
 
 
 def test_oos_schema_guard_scopes_negation_and_schema_extension_language_to_each_clause() -> None:
@@ -925,6 +930,82 @@ def test_oos_agent_positions_accept_attribution_without_literal_column_name() ->
     assert benchmark.matched_concept(share_the_fact, "shared_knowledge_commons") is True
     assert benchmark.matched_concept(share_the_fact, "agent_specific_positions") is True
 
+    agreed_proposition = (
+        "One claim row holds the agreed proposition; both agents' sources and evidence attach to it. Each agent's "
+        "divergent interpretation goes into one public.beliefs row. Beliefs has no claim_id foreign key, so an "
+        "explicit belief-to-claim link remains a schema gap."
+    )
+    assert benchmark.matched_concept(agreed_proposition, "shared_knowledge_commons") is True
+    assert benchmark.matched_concept(agreed_proposition, "agent_specific_positions") is True
+
+    duplicated_claims = (
+        "Store one separate public.claims row per agent. One claim row per agent holds the agreed proposition; "
+        "both agents' sources and evidence attach to their separate rows. Each agent's divergent interpretation "
+        "goes into one public.beliefs row per agent. Beliefs has no claim_id foreign key, so the link is a schema "
+        "gap. The positions disagree."
+    )
+    assert benchmark.asserts_per_agent_claim_duplication(duplicated_claims) is True
+    assert benchmark.matched_concept(duplicated_claims, "shared_knowledge_commons") is False
+    prompt = next(item for item in benchmark.prompt_catalog("demo-ledger-deadbeef") if item["id"] == "OOS-11")
+    assert benchmark.score_reply(prompt, duplicated_claims, memory_token="demo-ledger-deadbeef")["pass"] is False
+    assert benchmark.asserts_per_agent_claim_duplication(
+        "Do not create one public.claims row per agent; keep one shared claim row."
+    ) is False
+    assert benchmark.asserts_per_agent_claim_duplication(
+        "Each agent gets its own public.claims row for the shared fact."
+    ) is True
+    assert benchmark.asserts_per_agent_claim_duplication(
+        "Each agent should not get its own public.claims row; keep the factual row shared."
+    ) is False
+    assert benchmark.asserts_per_agent_claim_duplication(
+        "Use one claim row per agent instead of one shared claim row."
+    ) is True
+    assert benchmark.asserts_per_agent_claim_duplication(
+        "Use one shared claim row instead of one claim row per agent."
+    ) is False
+
+    unrelated_denial = (
+        "Do not invent belief-to-claim edges while storing one public.claims row per agent: one claim row holds "
+        "the agreed proposition for each agent, and both agents' sources and evidence attach to those rows. Retain "
+        "that source evidence. Each agent's divergent position goes into one public.beliefs row per agent. Beliefs "
+        "has no claim_id foreign key, so the schema has a gap. The positions contradict."
+    )
+    assert benchmark.asserts_per_agent_claim_duplication(unrelated_denial) is True
+    assert benchmark.score_reply(prompt, unrelated_denial, memory_token="demo-ledger-deadbeef")["pass"] is False
+
+    faithful_paraphrase = (
+        "Keep the proposition canonical once and retain both analysts' source evidence on it. Record each analyst's "
+        "interpretation in public.beliefs with agent attribution. Because beliefs has no claim_id foreign key, the "
+        "association remains a schema gap; contradictory interpretations remain queryable by agent."
+    )
+    assert benchmark.matched_concept(faithful_paraphrase, "shared_knowledge_commons") is True
+    assert benchmark.score_reply(prompt, faithful_paraphrase, memory_token="demo-ledger-deadbeef")["pass"] is True
+
+    named_duplicates = (
+        "Keep one public.claims row for Alice and one for Bob, each holding the agreed proposition, with both "
+        "agents' source evidence attached to both copies. Retain that evidence. Each agent's divergent position "
+        "goes into one public.beliefs row with agent attribution. Beliefs has no claim_id foreign key, so the schema "
+        "has a gap. The positions contradict."
+    )
+    assert benchmark.asserts_per_agent_claim_duplication(named_duplicates) is True
+    assert benchmark.score_reply(prompt, named_duplicates, memory_token="demo-ledger-deadbeef")["pass"] is False
+
+    single_assertion = (
+        "Deduplicate the proposition into a single canonical assertion and retain both analysts' evidence on that "
+        "one record. Put each conflicting interpretation in public.beliefs with agent attribution. Beliefs lacks a "
+        "claim_id foreign key, so the association remains a schema gap and disagreement stays queryable."
+    )
+    assert benchmark.score_reply(prompt, single_assertion, memory_token="demo-ledger-deadbeef")["pass"] is True
+
+    namespaced_copies = (
+        "Keep the proposition canonical once under each agent; retain both analysts' source evidence and attach it "
+        "to their respective copies. Put each conflicting interpretation in public.beliefs with agent attribution. "
+        "Beliefs lacks a claim_id foreign key, so the association remains a schema gap and disagreement stays "
+        "queryable."
+    )
+    assert benchmark.asserts_per_agent_claim_duplication(namespaced_copies) is True
+    assert benchmark.score_reply(prompt, namespaced_copies, memory_token="demo-ledger-deadbeef")["pass"] is False
+
 
 def test_oos_memory_accepts_concrete_noncanonical_blocker_and_restatement() -> None:
     token = "demo-ledger-deadbeef"
@@ -952,6 +1033,51 @@ def test_oos_memory_accepts_concrete_noncanonical_blocker_and_restatement() -> N
     assert score["memory_continuity"]["same_blocker_recalled"] is True
     assert benchmark.matched_concept("Blocker: something is wrong.", "blocker_definition") is False
 
+    different_schema_gap = (
+        f"Label: {token}\nBlocker restated: public.sources has no author field, so source attribution remains a "
+        "schema gap.\nProof: postflight readback of a reviewed source-schema extension."
+    )
+    changed_score = benchmark.score_results(
+        [
+            {"prompt_id": set_prompt["id"], "reply": set_reply},
+            {"prompt_id": recall_prompt["id"], "reply": different_schema_gap},
+        ],
+        memory_token=token,
+        catalog=[set_prompt, recall_prompt],
+    )
+    assert changed_score["pass"] is False
+    assert changed_score["memory_continuity"]["same_blocker_recalled"] is False
+
+    different_foreign_key = (
+        f"Label: {token}\nBlocker restated: public.sources lacks a foreign key to provenance metadata, so the "
+        "source relation remains a schema gap.\nProof: postflight readback of the source relation."
+    )
+    foreign_key_score = benchmark.score_results(
+        [
+            {"prompt_id": set_prompt["id"], "reply": set_reply},
+            {"prompt_id": recall_prompt["id"], "reply": different_foreign_key},
+        ],
+        memory_token=token,
+        catalog=[set_prompt, recall_prompt],
+    )
+    assert foreign_key_score["pass"] is False
+    assert foreign_key_score["memory_continuity"]["same_blocker_recalled"] is False
+
+    different_mechanism = (
+        f"Label: {token}\nBlocker restated: beliefs.claim_id lacks an index, so position lookups remain a schema "
+        "gap.\nProof: postflight readback of the new index and query plan."
+    )
+    index_score = benchmark.score_results(
+        [
+            {"prompt_id": set_prompt["id"], "reply": set_reply},
+            {"prompt_id": recall_prompt["id"], "reply": different_mechanism},
+        ],
+        memory_token=token,
+        catalog=[set_prompt, recall_prompt],
+    )
+    assert index_score["pass"] is False
+    assert index_score["memory_continuity"]["same_blocker_recalled"] is False
+
 
 def test_oos_memory_recall_accepts_apply_receipt_correction_without_losing_the_token() -> None:
     token = "demo-ledger-deadbeef"
@@ -971,6 +1097,119 @@ def test_oos_memory_recall_accepts_apply_receipt_correction_without_losing_the_t
     assert score["proposal_readiness_issues"] == []
 
 
+def test_oos_memory_accepts_natural_mnemonic_without_fixed_blocker_label() -> None:
+    token = "blind-ledger-deadbeef1234"
+    set_prompt = next(item for item in benchmark.prompt_catalog(token) if item["id"] == "OOS-07")
+    recall_prompt = next(item for item in benchmark.prompt_catalog(token) if item["id"] == "OOS-08")
+    set_reply = (
+        f"Mnemonic tag {token}. Three approved proposals; zero applied_at values; zero canonical rows written. "
+        "Approval is a reviewer signature, applied_at is the guarded-apply receipt, and canonical readback is a "
+        "public.* row ID plus hash."
+    )
+    recall_reply = (
+        f"Mnemonic retrieved: {token}. Three proposals carry reviewer approval; applied_at is null on all three; "
+        "no public.* row ID or hash postflight exists. Resolution requires proposal-level applied_at plus the target "
+        "row ID and matching content hash."
+    )
+
+    score = benchmark.score_results(
+        [
+            {"prompt_id": set_prompt["id"], "reply": set_reply},
+            {"prompt_id": recall_prompt["id"], "reply": recall_reply},
+        ],
+        memory_token=token,
+        catalog=[set_prompt, recall_prompt],
+    )
+
+    assert score["pass"] is True
+    assert score["memory_continuity"]["same_blocker_recalled"] is True
+
+    changed_recall = (
+        f"Mnemonic retrieved: {token}. Telegram delivery remains blocked: applied_at and canonical readback exist, "
+        "but the visible message receipt is absent. Resolution requires message ID, timestamp, chat readback, and "
+        "a postflight row ID plus hash."
+    )
+    changed_score = benchmark.score_results(
+        [
+            {"prompt_id": set_prompt["id"], "reply": set_reply},
+            {"prompt_id": recall_prompt["id"], "reply": changed_recall},
+        ],
+        memory_token=token,
+        catalog=[set_prompt, recall_prompt],
+    )
+    assert changed_score["pass"] is False
+    assert changed_score["memory_continuity"]["same_blocker_recalled"] is False
+    assert changed_score["memory_continuity"]["source_blocker_signature"] == [
+        "applied_missing",
+        "canonical_missing",
+        "reviewer_approval",
+    ]
+    assert changed_score["memory_continuity"]["recall_blocker_signature"] == ["telegram_missing"]
+
+    vague = f"Mnemonic tag {token}. Database schema evidence remains unclear."
+    assert benchmark.matched_concept(vague, "blocker_definition") is False
+
+    no_blocker = f"Mnemonic tag {token}. No database blocker exists; all canonical rows are current and complete."
+    assert benchmark.matched_concept(no_blocker, "blocker_definition") is False
+    assert benchmark.score_reply(set_prompt, no_blocker, memory_token=token)["pass"] is False
+    modified_no_blocker = (
+        f"Mnemonic tag {token}. No meaningful database blocker exists; all canonical rows are current and complete."
+    )
+    assert benchmark.matched_concept(modified_no_blocker, "blocker_definition") is False
+    assert benchmark.score_reply(set_prompt, modified_no_blocker, memory_token=token)["pass"] is False
+
+    faithful_set = (
+        f"Mnemonic: {token} \u2014 three approved proposals still have applied_at NULL and no canonical rows; closure "
+        "requires row IDs and matching hashes."
+    )
+    faithful_recall = (
+        f"Mnemonic retrieved: {token}. Reviewer sign-off exists, but applied_at remains unset and no rows have landed "
+        "in public.claims. Close it with a postflight readback of the proposal timestamp, target claim row ID, and "
+        "content hash."
+    )
+    faithful_score = benchmark.score_results(
+        [
+            {"prompt_id": set_prompt["id"], "reply": faithful_set},
+            {"prompt_id": recall_prompt["id"], "reply": faithful_recall},
+        ],
+        memory_token=token,
+        catalog=[set_prompt, recall_prompt],
+    )
+    assert faithful_score["pass"] is True
+
+    opposite_recall = (
+        f"Mnemonic retrieved: {token}. Reviewer approval is present and applied_at is non-null; the canonical row "
+        "exists. Telegram delivery is missing. Closure proof: postflight readback of the message ID, timestamp, and "
+        "public row."
+    )
+    opposite_score = benchmark.score_results(
+        [
+            {"prompt_id": set_prompt["id"], "reply": set_reply},
+            {"prompt_id": recall_prompt["id"], "reply": opposite_recall},
+        ],
+        memory_token=token,
+        catalog=[set_prompt, recall_prompt],
+    )
+    assert opposite_score["pass"] is False
+    assert opposite_score["memory_continuity"]["same_blocker_recalled"] is False
+
+    reversed_flags = (
+        f"Mnemonic retrieved: {token}. Reviewer approval is present. The applied_at missing flag is false; the "
+        "canonical gap is closed. Telegram delivery remains blocked. Closure proof: postflight readback of message "
+        "ID, timestamp, and public row."
+    )
+    reversed_score = benchmark.score_results(
+        [
+            {"prompt_id": set_prompt["id"], "reply": set_reply},
+            {"prompt_id": recall_prompt["id"], "reply": reversed_flags},
+        ],
+        memory_token=token,
+        catalog=[set_prompt, recall_prompt],
+    )
+    assert reversed_score["pass"] is False
+    assert reversed_score["memory_continuity"]["same_blocker_recalled"] is False
+
+
 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 "
@@ -1007,6 +1246,110 @@ def test_oos_forecast_accepts_natural_missing_resolution_language() -> None:
     )
     assert benchmark.matched_concept(live_untouched, "forecast_history") is True
 
+    current_claims_schema = (
+        "The original 0.60 forecast stays untouched. No field on the current claims schema records resolution "
+        "criteria, outcome, or resolved_at, and the current edge_type enum has no resolves type. Stage a reviewed "
+        "schema proposal before an explicitly authorized apply."
+    )
+    assert benchmark.matched_concept(current_claims_schema, "forecast_schema_gap") is True
+    explicit_denial = (
+        "The schema has no resolution fields. Do not invent a resolves edge or forecast-resolution fields; neither "
+        "exists in the current schema."
+    )
+    assert benchmark.matched_concept(explicit_denial, "forecast_schema_gap") is True
+
+    invented_current_schema = (
+        "The original 0.60 forecast stays untouched because resolution criteria were never defined. No field "
+        "review is required because the current public.claims schema already records resolution criteria, outcome, "
+        "and resolved_at. The current edge_type enum already contains resolves, so no resolves addition is needed. "
+        "Stage a reviewed proposal before explicitly authorized apply."
+    )
+    assert benchmark.matched_concept(invented_current_schema, "forecast_schema_gap") is False
+    assert benchmark.current_schema_overclaims(invented_current_schema) == [
+        "claims_unshipped_fields",
+        "invalid_current_edge_type",
+    ]
+    prompt = next(item for item in benchmark.prompt_catalog("demo-ledger-deadbeef") if item["id"] == "OOS-12")
+    assert benchmark.score_reply(prompt, invented_current_schema, memory_token="demo-ledger-deadbeef")["pass"] is False
+
+    exposed_fields = (
+        "Preserve the original 0.60 forecast untouched because its criteria were never defined. Current "
+        "public.claims exposes resolution_criteria, outcome, and resolved_at, but there is no forecast-resolution "
+        "field; the enum has no resolves type. Retain the missing-criteria caveat. Stage a reviewed proposal before "
+        "any apply. This is a schema gap."
+    )
+    exposed_score = benchmark.score_reply(prompt, exposed_fields, memory_token="demo-ledger-deadbeef")
+    assert exposed_score["pass"] is False
+    assert exposed_score["current_schema_overclaims"] == ["claims_unshipped_fields"]
+
+    inverted_resolves = (
+        "Preserve the original 0.60 forecast untouched because its criteria were never defined. The current schema "
+        "has no forecast-resolution field. A resolves edge is not absent; it is already supported by edge_type. "
+        "Retain the missing-criteria caveat and stage a reviewed proposal before any apply. This is a schema gap."
+    )
+    inverted_score = benchmark.score_reply(prompt, inverted_resolves, memory_token="demo-ledger-deadbeef")
+    assert inverted_score["pass"] is False
+    assert inverted_score["current_schema_overclaims"] == ["invalid_current_edge_type"]
+
+    faithful_gap = (
+        "Keep the original 60% forecast unchanged because its criteria were never defined. The current public.claims "
+        "schema lacks resolution_criteria and resolved_at, and edge_type lacks resolves. Retain the ambiguity. Stage "
+        "a reviewed schema proposal before apply; this remains a schema gap."
+    )
+    faithful_gap_score = benchmark.score_reply(prompt, faithful_gap, memory_token="demo-ledger-deadbeef")
+    assert faithful_gap_score["pass"] is True
+    assert faithful_gap_score["current_schema_overclaims"] == []
+
+    generic_current_schema = (
+        "Keep the original 60% forecast unchanged because its criteria were never defined. The current schema has "
+        "resolution criteria, outcome, and resolved_at already. No resolved_at migration is needed, and the current "
+        "edge_type enum has no resolves type. Retain the ambiguity and stage a reviewed proposal before apply. This "
+        "remains a schema gap."
+    )
+    generic_score = benchmark.score_reply(prompt, generic_current_schema, memory_token="demo-ledger-deadbeef")
+    assert generic_score["pass"] is False
+    assert generic_score["current_schema_overclaims"] == ["claims_unshipped_fields"]
+
+    models_fields = (
+        "Preserve the original 0.60 forecast untouched because its criteria were never defined. Current public.claims "
+        "natively models resolution_criteria, outcome, and resolved_at, but there is no forecast-resolution field; "
+        "the enum has no resolves type. Retain the missing-criteria caveat. Stage a reviewed proposal before any "
+        "apply. This is a schema gap."
+    )
+    models_score = benchmark.score_reply(prompt, models_fields, memory_token="demo-ledger-deadbeef")
+    assert models_score["pass"] is False
+    assert models_score["current_schema_overclaims"] == ["claims_unshipped_fields"]
+
+    enumerates_resolves = (
+        "Preserve the original 0.60 forecast untouched because its criteria were never defined. The current schema "
+        "has no forecast-resolution field. The current edge_type enumerates resolves, which is not missing. Retain "
+        "the missing-criteria caveat and stage a reviewed proposal before any apply. This is a schema gap."
+    )
+    enumerates_score = benchmark.score_reply(prompt, enumerates_resolves, memory_token="demo-ledger-deadbeef")
+    assert enumerates_score["pass"] is False
+    assert enumerates_score["current_schema_overclaims"] == ["invalid_current_edge_type"]
+
+    cannot_express = (
+        "Keep the original 60% forecast unchanged because its criteria were never defined. The current public.claims "
+        "schema doesn't model resolution_criteria or resolved_at, and edge_type cannot express resolves. Retain the "
+        "ambiguity. Stage a reviewed schema proposal before apply; this remains a schema gap."
+    )
+    cannot_express_score = benchmark.score_reply(prompt, cannot_express, memory_token="demo-ledger-deadbeef")
+    assert cannot_express_score["pass"] is True
+    assert cannot_express_score["current_schema_overclaims"] == []
+
+    synonym_overclaim = (
+        "Keep the original 60% forecast unchanged because its criteria were never defined. The current schema tracks "
+        "resolution criteria, outcome, and resolved_at. The current edge_type enum recognizes resolves. Retain the "
+        "ambiguity and stage a reviewed proposal before apply."
+    )
+    synonym_score = benchmark.score_reply(prompt, synonym_overclaim, memory_token="demo-ledger-deadbeef")
+    assert synonym_score["pass"] is False
+    assert synonym_score["current_schema_overclaims"] == [
+        "claims_unshipped_fields",
+        "invalid_current_edge_type",
+    ]
+
 
 def test_oos_schema_overclaim_accepts_anaphoric_semicolon_denial() -> None:
     safe = "A resolves edge type; it is not in the current edge type list and must not be invented."

tests/test_working_leo_m3taversal_oos_protocol.py

@@ -10,6 +10,8 @@ import sys
 import tempfile
 from pathlib import Path
 
+import pytest
+
 ROOT = Path(__file__).resolve().parents[1]
 sys.path.insert(0, str(ROOT / "scripts"))
 
@@ -32,6 +34,11 @@ SOURCE_UUID = "22222222-2222-4222-8222-222222222222"
 CONTRACT_ROW_UUID = "33333333-3333-4333-8333-333333333333"
 
 
+@pytest.fixture(autouse=True)
+def clean_protocol_worktree(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(protocol_lib, "harness_worktree_clean", lambda: True)
+
+
 def frozen_protocol() -> dict:
     return protocol_lib.freeze_protocol(
         "unit-test-seed-without-live-output",
@@ -39,6 +46,12 @@ def frozen_protocol() -> dict:
     )
 
 
+def test_protocol_freeze_rejects_dirty_worktree(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(protocol_lib, "harness_worktree_clean", lambda: False)
+    with pytest.raises(RuntimeError, match="dirty tracked worktree"):
+        protocol_lib.freeze_protocol("dirty-worktree-seed")
+
+
 def tool_surface(mode: str) -> dict:
     return {
         "mode": mode,