"""Yardstick for CLOSE derived climb (PR-3 after #788+#789), hardened for full Claim B. Deterministic proof that: - is-a + relational-transitive derived facts enlarge the directly-answerable set monotonically to fixed point across idle_tick (with consolidate_determinations=True). - wrong_total == 0: excluded predicates (parent_of etc.) and member∨member canary never derive; negatives remain Undetermined. - proposal emission uses *real* ChatRuntime.idle_tick() + IdleTickResult.derived_close_proposals_emitted (gated by review_derived_close_proposals). - Positive growth scored via semantic determine() calls on materialized facts (rule='direct' post-FP). - All trajectories/replays stable at *content* level (closures + proposal bodies) + aggregates (no clock/LLM). Uses real comprehend/realize/idle_tick path. Small fixed scenarios for is-a, less_than, before_event, negatives. Proposal flag via lived idle_tick (not simulation). replay_checksum: aggregate sizes + wrong + flag (compatibility). content_replay_checksum: canonical closures (structure_key + Derivation/premises) + proposal bodies. """ from __future__ import annotations import hashlib import json from dataclasses import asdict, dataclass from pathlib import Path from typing import Any from chat.runtime import ChatRuntime from core.config import RuntimeConfig from dataclasses import replace from generate.determine import Determined, Undetermined, determine from generate.meaning_graph.reader import comprehend from generate.meaning_graph.relational import ( TRANSITIVE_PREDICATES, comprehend_relational, load_relational_pack_lemmas, ) from generate.realize import RealizedRecord, realize_comprehension, recall_realized from session.context import SessionContext _HIGH = 10**9 _PACK = load_relational_pack_lemmas() def _fresh_ctx( consolidate: bool = True, proposals: bool = False ) -> tuple[SessionContext, ChatRuntime]: from core.config import DEFAULT_CONFIG cfg = replace( DEFAULT_CONFIG, consolidate_determinations=consolidate, review_derived_close_proposals=proposals, ) rt = ChatRuntime(config=cfg, no_load_state=True) ctx = rt._context # ensure high reproject like other lanes if hasattr(ctx, "vault_reproject_interval"): ctx.vault_reproject_interval = _HIGH return ctx, rt def _tell(ctx: SessionContext, text: str) -> None: realize_comprehension(comprehend(text), ctx) def _tell_rel(ctx: SessionContext, text: str) -> None: realize_comprehension(comprehend_relational(text, _PACK), ctx) def _ask(ctx: SessionContext, text: str) -> Determined | Undetermined: return determine(comprehend(text), ctx) def _ask_rel(ctx: SessionContext, text: str) -> Determined | Undetermined: return determine(comprehend_relational(text, _PACK), ctx) def _count_answerable(ctx: SessionContext, asks: list[tuple[str, bool]]) -> int: """Count how many of the asked facts are directly realized in the vault (the 'direct answerable' from CLOSE).""" count = 0 for q, is_rel in asks: q = q.lower().strip('?').strip() if not q.startswith("is "): continue rest = q[3:].strip() parts = rest.split() if len(parts) < 2: continue subj = parts[0] if "less than" in rest: pred = "less_than" obj = parts[-1] elif "before" in rest: pred = "before_event" obj = parts[-1] else: # is-a "is a X" or "is an X" pred = "member" obj = parts[-1] # check if the specific (pred, subj, obj) is realized hits = [r for r in recall_realized(ctx, subject=subj, predicate=pred) if len(r.relation_arguments) > 1 and r.relation_arguments[1] == obj] if hits: count += 1 return count def _get_closure_content(ctx: SessionContext) -> list[dict[str, Any]]: """Project full closure for Claim B content checksum (structure_key + derivation with premises).""" data: list[dict[str, Any]] = [] for p in ["member", "subset"] + sorted(TRANSITIVE_PREDICATES): for r in recall_realized(ctx, predicate=p): der = asdict(r.derivation) if r.derivation is not None else None data.append( { "predicate": p, "args": list(r.relation_arguments), "structure_key": r.structure_key, "derived": r.derived, "derivation": der, "epistemic_status": getattr(r, "epistemic_status", None), } ) data.sort(key=lambda x: (x["predicate"], tuple(x["args"]))) return data # --- Scenarios --- def _is_a_climb() -> dict[str, Any]: """member/subset climb: rex dog -> mammal -> ... creature. Expect growth 1->4.""" ctx, rt = _fresh_ctx() _tell(ctx, "Rex is a dog.") _tell(ctx, "All dogs are mammals.") _tell(ctx, "All mammals are animals.") _tell(ctx, "All animals are creatures.") # Canary: member(dog, kingdom) + member(rex, dog) should not derive member(rex, kingdom) _tell(ctx, "Dog is a kingdom.") # Query set for answerable - used for live measurement at each stage queries = [ ("Is Rex a dog?", False), ("Is Rex a mammal?", False), ("Is Rex an animal?", False), ("Is Rex a creature?", False), ("Is Rex a kingdom?", False), # must stay refused ] before = _count_answerable(ctx, queries) res1 = rt.idle_tick() after1 = _count_answerable(ctx, queries) # continue ticking until fixed point (use facts_consolidated==0 as proxy, since IdleTickResult has no at_fixed_point) fp = res1 for _ in range(10): # safety bound fp = rt.idle_tick() if fp.facts_consolidated == 0: break after_fp = _count_answerable(ctx, queries) wrong_total = 0 # Check canary not derived (still valid for wrong=0) if "kingdom" in [r.relation_arguments[1] for r in recall_realized(ctx, subject="rex", predicate="member")]: wrong_total += 1 # Semantic Answerability (Claim B): explicitly call determine on positives post-fixed-point # and assert Determined(True) with rule='direct' (the materialized direct path). positive_queries = [q for q in queries if "kingdom" not in q[0].lower()] for q, is_rel in positive_queries: det = _ask(ctx, q) if not is_rel else _ask_rel(ctx, q) assert isinstance(det, Determined) and getattr(det, "answer", False) and getattr(det, "rule", None) == "direct", ( f"Claim B semantic answerability failed for {q}: got {det}" ) closure = _get_closure_content(ctx) return { "scenario": "is_a_climb", "before": before, "after_tick_1": after1, "after_fixed_point": after_fp, "facts_consolidated_tick1": res1.facts_consolidated if hasattr(res1, 'facts_consolidated') else 0, "at_fixed_point": (fp.facts_consolidated == 0), "wrong_total": wrong_total, "proposals_emitted": 0, # separate flag run below "closure": closure, "semantic_positives_determined_direct": len(positive_queries), } def _relational_climb_less_than() -> dict[str, Any]: ctx, rt = _fresh_ctx() _tell_rel(ctx, "A is less than B.") _tell_rel(ctx, "B is less than C.") _tell_rel(ctx, "C is less than D.") # Consistent query set for all stages queries = [ ("Is A less than C?", True), ("Is B less than D?", True), ("Is A less than D?", True), ] before = _count_answerable(ctx, queries) res1 = rt.idle_tick() after1 = _count_answerable(ctx, queries) res2 = rt.idle_tick() after_fp = _count_answerable(ctx, queries) # Negative: no cross (refused) neg = _ask_rel(ctx, "Is A less than E?") # no E, but to test refused for non-chain wrong = 0 if isinstance(neg, Determined): wrong += 1 # Semantic Answerability (Claim B) for q, is_rel in queries: det = _ask_rel(ctx, q) assert isinstance(det, Determined) and getattr(det, "answer", False) and getattr(det, "rule", None) == "direct", ( f"Claim B semantic answerability failed for {q}: got {det}" ) closure = _get_closure_content(ctx) return { "scenario": "less_than_climb", "before": before, "after_tick_1": after1, "after_fixed_point": after_fp, "wrong_total": wrong, "closure": closure, "semantic_positives_determined_direct": len(queries), } def _temporal_climb() -> dict[str, Any]: ctx, rt = _fresh_ctx() _tell_rel(ctx, "Dawn is before noon.") _tell_rel(ctx, "Noon is before dusk.") # Consistent query set (single but via helper for uniformity) queries = [ ("Is Dawn before dusk?", True), ] before = _count_answerable(ctx, queries) rt.idle_tick() after = _count_answerable(ctx, queries) # Semantic Answerability (Claim B) for q, is_rel in queries: det = _ask_rel(ctx, q) assert isinstance(det, Determined) and getattr(det, "answer", False) and getattr(det, "rule", None) == "direct", ( f"Claim B semantic answerability failed for {q}: got {det}" ) closure = _get_closure_content(ctx) return { "scenario": "before_event_climb", "before": before, "after_fixed_point": after, "wrong_total": 0, "closure": closure, "semantic_positives_determined_direct": len(queries), } def _negatives_refused() -> dict[str, Any]: ctx, rt = _fresh_ctx() _tell_rel(ctx, "X is parent of Y.") _tell_rel(ctx, "Y is parent of Z.") _tell_rel(ctx, "P is sibling of Q.") _tell_rel(ctx, "Q is sibling of R.") rt.idle_tick() rt.idle_tick() refused = 0 for q in ["Is X parent of Z?", "Is P sibling of R?"]: if isinstance(_ask_rel(ctx, q), Undetermined): refused += 1 return { "scenario": "negatives_refused", "parent_refused": refused >= 1, "sibling_refused": refused >= 2, "wrong_total": 0 if refused == 2 else 1, } def _proposal_flag_effect() -> dict[str, Any]: """Measure proposal bridge via *lived* runtime flag path. Uses real ChatRuntime.idle_tick() + IdleTickResult.derived_close_proposals_emitted (the runtime flag in idle_tick gates the call to the bridge).""" import generate.determine.derived_close_proposals as dcp import tempfile # "without" via real idle_tick with flag off (must emit 0) ctx_off, rt_off = _fresh_ctx(consolidate=True, proposals=False) _tell_rel(ctx_off, "A is less than B.") _tell_rel(ctx_off, "B is less than C.") res_off = rt_off.idle_tick() emitted_without = res_off.derived_close_proposals_emitted # "with" via real idle_tick with flag on (captures from IdleTickResult; temp sink for isolation + content) with tempfile.TemporaryDirectory() as tmp: sink = Path(tmp) orig_sink = dcp.DEFAULT_SINK dcp.DEFAULT_SINK = sink try: ctx_on, rt_on = _fresh_ctx(consolidate=True, proposals=True) _tell_rel(ctx_on, "A is less than B.") _tell_rel(ctx_on, "B is less than C.") res_on = rt_on.idle_tick() emitted_with = res_on.derived_close_proposals_emitted # capture full bodies for content checksum (lived emission path) proposals = [json.loads(p.read_text()) for p in sorted(sink.glob("*.json"))] finally: dcp.DEFAULT_SINK = orig_sink return { "scenario": "proposal_flag", "emitted_without_flag": emitted_without, "emitted_with_flag": emitted_with, "only_with_flag": emitted_with > 0 and emitted_without == 0, "proposals": proposals, } def run() -> dict[str, Any]: """Run the full yardstick. Returns report with climb metrics, wrong_total=0, lived proposal flag isolation (via IdleTickResult), semantic determine() on positives, replay_checksum (aggregates) + content_replay_checksum (closures + proposal bodies), and proposal_review_posture (additive visibility into the review/ratification side: emitted proposals are born proposal_only/SPECULATIVE/requires_review; the yardstick exercises no acceptance, rejection, or promotion paths).""" is_a = _is_a_climb() less = _relational_climb_less_than() temporal = _temporal_climb() neg = _negatives_refused() prop = _proposal_flag_effect() # Aggregate before_sets = [is_a["before"], less["before"], temporal["before"]] after1_sets = [is_a["after_tick_1"], less["after_tick_1"], temporal["after_fixed_point"]] fp_sets = [is_a["after_fixed_point"], less["after_fixed_point"], temporal["after_fixed_point"]] total_wrong = is_a["wrong_total"] + less["wrong_total"] + temporal.get("wrong_total", 0) + neg["wrong_total"] monotone_growth = all(a <= b for a, b in zip(before_sets, fp_sets)) strict_growth = sum(before_sets) < sum(fp_sets) report = { "is_a_climb": is_a, "less_than_climb": less, "before_event_climb": temporal, "negatives": neg, "proposal_flag": prop, "aggregate": { "direct_answerable_before": sum(before_sets), "direct_answerable_after_tick_1": sum(after1_sets), "direct_answerable_after_fixed_point": sum(fp_sets), "monotone_growth": monotone_growth, "strict_growth": strict_growth, "wrong_total": total_wrong, "proposals_only_with_flag": prop["only_with_flag"], }, } # Additive review/ratification-posture instrumentation for the CLOSE flywheel # (the previously weaker half). Computed purely from the proposal bodies # already emitted and captured for content_replay_checksum. No review logic # is invoked; the yardstick only observes the explicit review-gated posture # that every derived CLOSE proposal is born with (proposal_only / SPECULATIVE # / requires_review). This surfaces acceptance/rejection eligibility signals # and documents that the yardstick itself performs no ratification. _proposals = prop.get("proposals", []) or [] _all_status = [p.get("status") for p in _proposals] _all_epist = [p.get("epistemic_status") for p in _proposals] _all_req = [p.get("requires_review") for p in _proposals] report["proposal_review_posture"] = { "emitted_count": len(_proposals), "all_proposal_only": all(s == "proposal_only" for s in _all_status) if _proposals else True, "all_speculative": all(e == "speculative" for e in _all_epist) if _proposals else True, "all_requires_review": all(r is True for r in _all_req) if _proposals else True, "review_eligible": len(_proposals), "none_accepted_or_promoted": True, # yardstick is emission + semantic only; ratification is HITL / operator-gated outside this surface } # Replay checksum: stable sizes + wrongs + flag effect (kept for compatibility) checksum_input = json.dumps( { "sizes": [report["aggregate"][k] for k in ("direct_answerable_before", "direct_answerable_after_tick_1", "direct_answerable_after_fixed_point")], "wrong_total": total_wrong, "flag_isolation": prop["only_with_flag"], }, sort_keys=True, ).encode() report["replay_checksum"] = hashlib.sha256(checksum_input).hexdigest() # Parallel full-content checksum for Claim B fidelity (canonical closures + proposal bodies) # This aligns the yardstick with its documentation ("closure sets", "exact trajectories"). content = { "closures": { "is_a": is_a.get("closure", []), "less_than": less.get("closure", []), "before_event": temporal.get("closure", []), }, "proposals": prop.get("proposals", []), } content_input = json.dumps(content, sort_keys=True).encode() report["content_replay_checksum"] = hashlib.sha256(content_input).hexdigest() return report if __name__ == "__main__": import pprint pprint.pprint(run())