7a20356ab6
- Semantic: determine() asserts (rule=direct) on positives post-FP in climbs - Lived flag: _proposal_flag_effect now uses real idle_tick + IdleTickResult.derived_close_proposals_emitted (temp sink isolation) - Checksum: added content_replay_checksum (canonical closures + proposal bodies) - Docs: contract.md + module docstrings updated for accuracy Ratification artifact created before impl (see docs/analysis/...-ratification-...md). All original metrics, wrong_total=0, invariants preserved. Closes the 3 gaps from post-merge audit for Claim B.
396 lines
No EOL
14 KiB
Python
396 lines
No EOL
14 KiB
Python
"""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)."""
|
||
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"],
|
||
},
|
||
}
|
||
|
||
# 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()) |