core-labs/core
1
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

Merge pull request #603 from AssetOverflow/feat/loop-learns-determined

Browse source 
feat(determine): idle deductive consolidation — the loop learns from determined facts (Step D)
This commit is contained in:
Shay 2026-06-06 12:36:45 -07:00 committed by GitHub
commit 8320c2cbb4
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
14 changed files with 1252 additions and 54 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

128
chat/runtime.py
View file

@ -523,11 +523,20 @@ class ChatResponse:
@dataclass(frozen=True, slots=True)
class IdleTickResult:
"""Outcome of one ``idle_tick`` — proposal-only learning, never ratification."""
"""Outcome of one ``idle_tick``.
The proposal pass is PROPOSAL-ONLY learning, never ratification (HITL ratifies). The
Step D consolidation pass is SESSION-memory learning: ``facts_consolidated`` derived
facts were written back into the held self so the next ``determine`` reaches them
directly — still never corpus mutation, never a proposal.
"""
candidates_contemplated: int
proposals_created: int
pending_proposals: int
#: Step D — derived facts consolidated into the held self this tick (0 unless
#: ``config.consolidate_determinations`` and the closure had a new layer to add).
facts_consolidated: int = 0
class ChatRuntime:
@ -834,63 +843,86 @@ class ChatRuntime:
)
def idle_tick(self) -> "IdleTickResult":
"""Advance the reviewed-learning flywheel during idle (NO user turn).
"""Advance learning during idle (NO user turn). Two disjoint passes:
This is how the engine "learns while it lives": between turns it turns its
lived experience (the pending discovery backlog) into reviewable teaching
proposals. It contemplates each pending candidate (enrichment) and runs
the replay-gated ``propose_from_candidate``, which leaves a PROPOSAL-ONLY
``pending`` entry in the persistent proposal log.
1. PROPOSAL pass (the reviewed-learning flywheel): turn the pending discovery
backlog into reviewable teaching proposals. Contemplate each pending
candidate (enrichment) and run the replay-gated ``propose_from_candidate``,
which leaves a PROPOSAL-ONLY ``pending`` entry in the persistent proposal
log. An idle tick NEVER ratifies — ratification (appending to the corpus)
stays HITL via ``teaching/review`` (CLAUDE.md teaching safety). The tick only
*proposes*; the reviewed loop is not bypassed or duplicated.
Teaching safety (CLAUDE.md): an idle tick NEVER ratifies. Ratification —
moving a proposal to ``accepted`` and appending to the corpus — stays
HITL via ``teaching/review``. The tick only *proposes*; the reviewed loop
is not bypassed or duplicated.
2. CONSOLIDATION pass (Step D — CLOSE): the loop learns from *determined* facts.
Run one semi-naive layer of the member/subset deductive closure over the held
self (``generate.determine.consolidate_once``); each soundly-derived,
proof_chain-verified fact is written back as a SPECULATIVE realized record so
the next ``determine`` reaches it directly. SESSION memory (immediate,
allowed) — NOT corpus mutation, NOT a proposal; the HITL path is untouched.
Gated by ``config.consolidate_determinations``. The closure converges (a
saturated tick consolidates nothing — ``at_fixed_point``).
The proposal log and the candidate backlog both live in the engine-state
dir, so this learning progress persists across reboot (CL-2).
The proposal log, the candidate backlog, and (with ``persist_session_state``)
the consolidated facts all live in the engine-state dir, so this learning
progress persists across reboot (CL-2).
"""
if self._proposal_log is None or not self._pending_candidates:
return IdleTickResult(0, 0, self._count_pending_proposals())
from teaching.contemplation import contemplate
from teaching.proposals import (
ProposalError,
TeachingChainProposal,
_current_revision,
propose_from_candidate,
)
from teaching.source import ProposalSource
vault_probe = (
_vault_probe_for_context(self._context) if self._context else None
)
contemplated = [
contemplate(candidate, vault_probe=vault_probe)
for candidate in self._pending_candidates
]
contemplated_count = 0
created = 0
for candidate in contemplated:
source = ProposalSource(
kind="contemplation",
source_id=candidate.candidate_id,
emitted_at_revision=_current_revision(),
facts_consolidated = 0
did_work = False
# 1. Proposal pass — unchanged behavior, runs only with a log + backlog.
if self._proposal_log is not None and self._pending_candidates:
from teaching.contemplation import contemplate
from teaching.proposals import (
ProposalError,
TeachingChainProposal,
_current_revision,
propose_from_candidate,
)
try:
result = propose_from_candidate(
candidate, log=self._proposal_log, source=source
from teaching.source import ProposalSource
vault_probe = (
_vault_probe_for_context(self._context) if self._context else None
)
contemplated = [
contemplate(candidate, vault_probe=vault_probe)
for candidate in self._pending_candidates
]
contemplated_count = len(contemplated)
for candidate in contemplated:
source = ProposalSource(
kind="contemplation",
source_id=candidate.candidate_id,
emitted_at_revision=_current_revision(),
)
except ProposalError:
continue
if isinstance(result, TeachingChainProposal):
created += 1
# Persist the advanced backlog (candidates + lineage); the proposal log
# is already file-backed.
self.checkpoint_engine_state()
try:
result = propose_from_candidate(
candidate, log=self._proposal_log, source=source
)
except ProposalError:
continue
if isinstance(result, TeachingChainProposal):
created += 1
did_work = True
# 2. Consolidation pass (Step D) — runs independently of the backlog.
if self.config.consolidate_determinations and self._context is not None:
from generate.determine import consolidate_once
facts_consolidated = consolidate_once(self._context).consolidated
did_work = True
# Persist the advanced state once (backlog + lineage +, with
# persist_session_state, the consolidated facts). Skipped on a no-op tick so an
# idle engine with nothing to learn does not churn the checkpoint.
if did_work:
self.checkpoint_engine_state()
return IdleTickResult(
candidates_contemplated=len(contemplated),
candidates_contemplated=contemplated_count,
proposals_created=created,
pending_proposals=self._count_pending_proposals(),
facts_consolidated=facts_consolidated,
)
def record_recognition_example(

13
core/config.py
View file

@ -303,6 +303,19 @@ class RuntimeConfig:
# teaching/review HITL path is untouched.
accrue_realized_knowledge: bool = False
# Step D (CLOSE) — when on, idle_tick consolidates soundly-derived determinations
# back into the held self: one semi-naive layer of the member/subset deductive
# closure (member∘subset, subset∘subset — NEVER member∘member) per tick, each
# derived edge proof_chain-verified, written as a SPECULATIVE realized record with
# derived-provenance. This is how the loop "learns from determined facts": derive
# once, remember, reach one hop further next tick — the directly answerable set
# climbs monotonically to the deductive-closure fixed point. OFF by default; the
# production L10 process enables it alongside accrue_realized_knowledge +
# persist_session_state. SESSION memory (immediate), NOT reviewed/corpus learning —
# no proposal coupling, the teaching/review HITL path is untouched. Bounded by the
# same _SUBSUMPTION_SUBSET_FACT_BUDGET; converges (a saturated tick is a no-op).
consolidate_determinations: bool = False
DEFAULT_IDENTITY_PACK: str = "default_general_v1"
DEFAULT_ETHICS_PACK: str = "default_general_ethics_v1"

153
docs/analysis/D-close-consolidation-design-2026-06-06.md Normal file
View file

@ -0,0 +1,153 @@
# Step D — CLOSE: idle deductive consolidation of soundly-derived facts
**Date:** 2026-06-06
**Branch:** `feat/loop-learns-determined`
**Sequence:** A INSTRUMENT (#598/#599) → B WIRE (#600/#601) → C DEEPEN (#602) → **D CLOSE** → E ESTIMATION (last)
**Telos:** [[project-core-is-one-continuous-life]] — the loop *learns from determined facts*, not just the partial discovery chains it emits today.
## The scoped seam (read, not assumed)
The autonomous loop today is `extract_discovery_candidates → _pending_candidates →
idle_tick(contemplate → propose_from_candidate) → ProposalLog (pending, HITL)`. Two
findings from reading the source fix what D actually is:
1. **Today's loop is a within-corpus generalizer.** `extract_discovery_candidates`
emits an *intentionally partial* chain `{subject, intent, connective:None,
object:None}`; `contemplate` completes it *only* by enumerating objects the
reviewed corpus already used, and `check_eligibility` **Gate 2** requires a
`source="corpus"` evidence pointer. So it can only re-propose a shape the corpus
already contains. It cannot learn a fact that arose in lived conversation.
2. **The HITL ProposalLog never reaches serving.** D's falsification — "the capability
index climbs across loop iterations" — therefore *cannot* be met by emitting
proposals (the index is a static breadth×accuracy eval, hard-gated to 0 on any
wrong; proposals are HITL-gated and never feed serving). For the index to climb,
a *determined fact must feed back into what the engine can answer.*
So the inevitable D is **session-memory consolidation**, not proposal emission:
> When idle, the engine consolidates each soundly-derived determination
> (`member∘subset`, `subset∘subset` — the transitive is-a reasoning C built) back
> into the realized-knowledge vault as a new realized record, so the *next*
> `determine()` reaches it directly and can chain one hop further. The directly
> answerable set climbs monotonically across idle ticks to the deductive-closure
> fixed point.
Emitting determined facts as HITL teaching-chain proposals is a *distinct* capability
that collides with Gate 2 (the corpus-evidence floor, a teaching-safety / wrong=0
surface). It is **deliberately deferred** to its own evidence-floor-touching PR rather
than bolted onto D — the falsifiable core stays clean.
## Mechanism (semi-naive deductive closure, one layer per tick)
`idle_tick` (gated by `config.consolidate_determinations`, default OFF) runs
`consolidate_once(ctx)`:
1. Recall realized `member` + `subset` facts.
2. Compute every **one-hop** extension under the two SOUND is-a rules:
- `member(s,b) ∧ subset(b,t) → member(s,t)`
- `subset(a,b) ∧ subset(b,t) → subset(a,t)`
- **`member ∘ member` is never an edge** — instance-of is not transitive
("Socrates is a man" + "man is a species" ⊬ "Socrates is a species"). The
reader's member/subset split is exactly what makes the included rules sound.
3. For each derived edge not already realized, **verify with the sound+complete
proof_chain ROBDD** (reusing C's single verifier `_verify_subsumption` — no
duplicate proof logic), then write it via `realize_derived`.
Each tick adds exactly one hop-layer; the closure saturates after `diameter` rounds;
re-running at the fixed point is a no-op. This is textbook semi-naive evaluation, and
it is precisely what produces the monotone "climbs across iterations" signal.
## Invariants held (the wrong=0 / honesty boundary)
- **Soundness (wrong=0).** Every consolidated fact is the conclusion of a sound rule
over realized premises, *confirmed by the sound+complete decider* — defense in depth
beyond the one-hop rule. The `member∘member` fallacy is structurally unreachable (no
member→member edge). A consolidated `member(s,t)` can only be extended by a *subset*
edge, never a member edge, so the fallacy stays unreachable across iterations.
- **Epistemic honesty.** A fact derived from SPECULATIVE premises is SPECULATIVE,
`basis="as_told"`. The soundness of the *inference* never upgrades the *standing* of
the premises. **COHERENT is never minted.**
- **Session memory, not reviewed learning.** Consolidation is the immediate session
tier (allowed), an extension of the `generate.realize` path — **not** corpus
mutation and **not** coupled to proposals. The teaching/review HITL path is
untouched; no parallel learning path is created.
- **Sanctioned write path.** Writes reuse `_realize_structured` (the INV-21 allowed
vault writer). No new normalization; no closure/repair — `algebra/versor.py` keeps
closure. The derived record reuses the subject's `probe_ingest` placement, identical
to a told fact about the same subject.
- **Idempotency / determinism.** Dedup on the span-free `structure_key` (identical to
a told fact's, so a later told duplicate collapses). Deterministic order; no clock,
no LLM, no metric call. Bounded by the existing `_SUBSUMPTION_SUBSET_FACT_BUDGET`.
## Provenance — the replayable proof obligation (Fork 2)
`Determined` already carries `grounds: tuple[RealizedRecord, ...]` — the premise
records that entailed it. The consolidated record records, as derived-provenance, the
premise `structure_key`s + the `rule` + the `verdict` (always `entailed`). This makes
the soundness claim **meaningfully fail** (per the Schema-Defined Proof Obligations
rule): a replay re-fetches the premises by `structure_key` and re-runs the rule +
proof_chain — if a consolidated fact were ever unsound, re-verification fails loudly.
## Falsification — `evals/determination_closure/`
A frozen replay seeds a deep is-a chain (`member(x,c0)`, `subset(c0,c1)…subset(cₙ₋₁,cₙ)`)
and runs idle consolidation ticks. Asserts:
- **Monotone climb:** the directly-realized `member(x, ·)` closure grows by exactly one
per tick (each layer), strictly increasing until the fixed point.
- **Convergence:** after `n` ticks the closure is saturated; a further tick is a
**no-op** (`at_fixed_point=True`, 0 consolidated).
- **wrong=0:** no `member(x, y)` is ever consolidated for `y` outside the chain's
reachable set (no fabricated fact); the `member∘member` canary derives nothing.
- **Provenance replay:** every consolidated record's recorded premises re-verify as
`ENTAILED`.
## Adversarial verification (5 independent skeptics, refute-the-claim)
A panel re-read the shipped source under five distinct lenses, each tasked to *refute*.
- **wrong=0 / soundness** — held. `member ∘ member` is structurally unreachable (member
facts are only ever extended by subset edges); every write is proof_chain-`ENTAILED`;
no cross-subject leakage. **Acted on its note:** `_verify_subsumption` now *refuses* a
mislabeled/wrong-arity path (a member fact smuggled into `subset_path`), converting
soundness-by-caller-discipline into soundness-by-construction now that consolidation is
a second caller.
- **epistemic honesty** — held. `realize_derived` writes SPECULATIVE unconditionally;
`_basis` returns `as_told` for SPECULATIVE grounds; `promote_eligible_entries`
(SPECULATIVE→COHERENT) requires energy metadata derived facts never carry and is
disjoint from `idle_tick`.
- **teaching-safety boundary** — held. Single write path (`ctx.vault.store`, `tier=
"session"`); zero `teaching/`/proposal/pack/identity coupling; the two `idle_tick`
passes are orthogonal (the proposal pass's vault probe filters to COHERENT, excluding
these SPECULATIVE facts).
- **determinism / replay / persistence** — held. Sorted write order; derived
`structure_key` identical to a told fact's; `Derivation` round-trips through the
snapshot; the lane's `reverify_derived` meaningfully fails on a non-entailing record.
- **normalization / closure invariant** — flagged `high`, assessed a **misattribution**.
The cited `vault.store → reproject → null_project` is pre-existing (`aadaf116`,
2026-05-13, ~3 weeks before D), triggered *identically* by the already-merged
`realize_comprehension` path, operates on vault **content** null-vectors (which
CLAUDE.md says to preserve as null vectors — sanctioned), and is a different object
from the runtime field `versor_condition(F)<1e-6` invariant the claim referenced. The
lane's high `vault_reproject_interval` is the established determine/realize test idiom,
not a D-specific sidestep. D adds zero normalization code and reuses the INV-21-allowed
writer; the claim ("consolidation *adds* no forbidden normalization") stands.
## Out of scope (follow-ups)
- **Promotion firewall (defensive).** If a future change ever called
`promote_eligible_entries` inside `idle_tick` or attached energy metadata to derived
facts, SPECULATIVE derived facts could promote to COHERENT and bootstrap standing. No
live path (the separation is architectural); a structural `derived ⇒ never-promote`
marker would harden it.
- **Runtime vs. lane proof obligation.** The provenance replay (`reverify_derived`) runs
in the eval lane, not per-write at runtime — matching the repo's "wrong=0 proven by
lanes, not runtime asserts" pattern (consolidation already verifies each write *before*
writing). Noted, not a gap.
- Determined-fact → HITL teaching proposal (touches Gate 2 / evidence floor — its own PR).
- Incremental frontier (semi-naive with a delta set) instead of recompute-and-dedup per
tick — an O() optimization once the closure substrate is proven.
- Order/containment transitivity (`less_than`, `before_event`, `inside_of`) — the C-2
predicate widening; consolidation generalizes to them once C-2 lands.

36
docs/runtime_contracts.md
View file

@ -76,6 +76,37 @@ selection only: it adds no field op, no normalization, and proposes no learning
(`accrue_in_turn` writes SESSION memory through the INV-21 vault writer; the HITL
teaching path is untouched).
### Idle consolidation (Step D — CLOSE)
When `consolidate_determinations` is enabled, `ChatRuntime.idle_tick` runs a
**consolidation pass** (`generate.determine.consolidate_once`): one semi-naive layer
of the member/subset deductive closure over the held self. For every sound one-hop
inference — `member(s,b) ∧ subset(b,t) → member(s,t)` and `subset(a,b) ∧ subset(b,t)
→ subset(a,t)`; **never `member ∘ member`** (instance-of is not transitive) — whose
conclusion is not yet realized, the hop is **verified by the sound+complete
proof_chain ROBDD** (reusing DETERMINE's single verifier) and written back as a
realized record via `generate.realize.realize_derived`, so the next `determine`
reaches it directly. Across idle ticks the directly-answerable set climbs
monotonically to the deductive-closure fixed point; a saturated tick consolidates
nothing (`IdleTickResult.facts_consolidated == 0`).
Contract:
- **SPECULATIVE / as-told.** A fact derived from SPECULATIVE premises stays
SPECULATIVE — a sound inference never upgrades the *standing* of its premises;
COHERENT is never minted here.
- **SESSION memory, not reviewed learning.** Consolidation is an extension of the
`generate.realize` session path — **not** corpus mutation and **not** coupled to
proposals. The teaching/review HITL path is untouched (no parallel learning path).
- **wrong=0 by proof-gating.** Only proof_chain-`ENTAILED` conclusions are written;
the `member ∘ member` fallacy is structurally unreachable.
- **Replayable provenance.** Each derived record carries a `Derivation` (premise
`structure_key`s + rule + the `entailed` verdict), so a replay re-derives and
re-verifies — the soundness claim can meaningfully fail.
- **No new normalization, no closure/repair.** Writes reuse the INV-21 vault writer;
`algebra/versor.py` keeps closure. Off by default; the falsification lane is
`evals.determination_closure`.
### Refusal contract (ADR-0024 Phase 2)
When the inner-loop admissibility check leaves no admissible destination
@ -246,7 +277,10 @@ it.
## Memory and teaching contract
Session memory can be immediate and local to the running context.
Session memory can be immediate and local to the running context. This includes
idle consolidation of soundly-derived (proof_chain-verified) facts back into the
held self (Step D — see *Idle consolidation* above): it is the immediate session
tier, not reviewed memory, and proposes nothing.
Reviewed memory must be explicit: user corrections or teaching examples become
reviewed memory only through the reviewed teaching loop.

17
evals/determination_closure/__init__.py Normal file
View file

@ -0,0 +1,17 @@
"""Determination-closure lane (Step D — CLOSE of the refined sequencing).
The falsification for "the loop learns from determined facts": idle consolidation
(``generate.determine.consolidate_once``) makes the engine's directly-answerable set
climb monotonically across idle ticks to the deductive-closure fixed point — with
wrong=0 (the member ∘ member fallacy is never derived), honesty (derived facts stay
SPECULATIVE), and a replayable provenance proof obligation.
"""
from evals.determination_closure.runner import (
TickRecord,
reverify_derived,
run,
seed_chain,
)
__all__ = ["TickRecord", "reverify_derived", "run", "seed_chain"]

22
evals/determination_closure/__main__.py Normal file
View file

@ -0,0 +1,22 @@
"""CLI: print the determination-closure falsification report.
python -m evals.determination_closure [depth]
"""
from __future__ import annotations
import json
import sys
from evals.determination_closure.runner import run
def main() -> int:
depth = int(sys.argv[1]) if len(sys.argv) > 1 else 9
report = run(depth)
print(json.dumps(report, indent=2))
return 0 if report["falsification_met"] else 1
if __name__ == "__main__":
raise SystemExit(main())

253
evals/determination_closure/runner.py Normal file
View file

@ -0,0 +1,253 @@
"""Determination-closure lane — the falsification for Step D (CLOSE).
Proves — deterministically, not by assertion — that idle consolidation makes the engine
LEARN from its determined facts: the set it can answer DIRECTLY climbs monotonically
across idle ticks to the deductive-closure fixed point, and a further tick is a no-op.
The soak seeds a deep is-a chain (``member(rex, c0)`` + ``subset(c0, c1) … subset(cₙ₋₁,
cₙ)``) through the REAL comprehend → realize path, then runs ``consolidate_once``
repeatedly. After each tick it records the directly-realized closure size. The lane also
carries the wrong=0 canary (a ``member ∘ member`` trap that must NEVER be derived) and
the provenance replay obligation (every derived record must re-verify ENTAILED from its
recorded premises — so the soundness claim can MEANINGFULLY FAIL).
Determinism: no clock, no LLM, no metric call; a pure function of the seeded chain.
"""
from __future__ import annotations
from dataclasses import asdict, dataclass
from typing import Any
from chat.runtime import ChatRuntime
from generate.determine.consolidate import consolidate_once
from generate.determine.determine import Determined, _verify_subsumption
from generate.meaning_graph.reader import comprehend
from generate.realize import RealizedRecord, realize_comprehension, recall_realized
from session.context import SessionContext
_HIGH_REPROJECT = 10**9
#: A deep is-a chain whose every ``All <plural[i]> are <plural[i+1]>`` parses as a
#: ``subset`` edge (singular lemmas: dog → mammal → … → item). The seed subject ``rex``
#: is a member of ``dog``; the chain gives 9 subset edges, so the closure reaches 10
#: classes from the one told membership.
_CHAIN_PLURALS = (
"dogs",
"mammals",
"animals",
"creatures",
"beings",
"mortals",
"things",
"entities",
"objects",
"items",
)
#: Singular lemmas the reader produces for the chain (what realized facts are keyed on).
_CHAIN_LEMMAS = (
"dog",
"mammal",
"animal",
"creature",
"being",
"mortal",
"thing",
"entity",
"object",
"item",
)
_SEED_SUBJECT = "rex"
#: The wrong=0 canary: ``member(dog, kingdom)`` is a membership ABOUT the class ``dog``.
#: With ``member(rex, dog)`` also held, an unsound ``member ∘ member`` rule would derive
#: ``member(rex, kingdom)``. The sound closure must NEVER produce it.
_CANARY_CLASS_MEMBERSHIP = ("Dog", "kingdom") # → member(dog, kingdom)
_CANARY_FORBIDDEN = (_SEED_SUBJECT, "kingdom") # member(rex, kingdom) must never appear
@dataclass(frozen=True, slots=True)
class TickRecord:
"""One idle consolidation tick's outcome."""
tick: int
member_closure_size: int # directly-realized member(rex, ·) facts
considered: int
consolidated: int
at_fixed_point: bool
def _fresh_context() -> SessionContext:
rt = ChatRuntime(no_load_state=True)
return SessionContext(
vocab=rt._context.vocab,
persona=rt._context.persona,
vault_reproject_interval=_HIGH_REPROJECT,
)
def _tell(ctx: SessionContext, text: str) -> None:
realize_comprehension(comprehend(text), ctx)
def seed_chain(ctx: SessionContext, depth: int) -> None:
"""Seed ``member(rex, c0)`` + ``depth`` subset edges + the canary, via the REAL
comprehend → realize path. ``depth`` is clamped to the available chain length."""
depth = max(1, min(depth, len(_CHAIN_PLURALS) - 1))
_tell(ctx, f"{_SEED_SUBJECT.capitalize()} is a {_CHAIN_LEMMAS[0]}.")
for i in range(depth):
_tell(ctx, f"All {_CHAIN_PLURALS[i]} are {_CHAIN_PLURALS[i + 1]}.")
# Canary membership about the class (member ∘ member trap).
_tell(ctx, f"{_CANARY_CLASS_MEMBERSHIP[0]} is a {_CANARY_CLASS_MEMBERSHIP[1]}.")
def _member_closure(ctx: SessionContext) -> tuple[str, ...]:
"""The objects ``rex`` is DIRECTLY realized to be a member of (sorted)."""
return tuple(
sorted(
f.relation_arguments[1]
for f in recall_realized(ctx, subject=_SEED_SUBJECT, predicate="member")
)
)
def _order_subset_path(
start: str, premises: tuple[RealizedRecord, ...]
) -> tuple[RealizedRecord, ...]:
"""Order subset premise records into a path from ``start`` (forward walk)."""
by_src: dict[str, RealizedRecord] = {
p.relation_arguments[0]: p for p in premises if len(p.relation_arguments) == 2
}
path: list[RealizedRecord] = []
cur = start
while cur in by_src:
edge = by_src.pop(cur)
path.append(edge)
cur = edge.relation_arguments[1]
return tuple(path)
def reverify_derived(ctx: SessionContext, record: RealizedRecord) -> bool:
"""Re-verify a derived record ENTAILED from its RECORDED premises (the provenance
replay obligation). Re-fetches each premise by ``structure_key`` and re-runs the
SAME sound+complete decider — so a record whose stored premises do not actually
entail it fails loudly."""
if not record.derived or record.derivation is None:
return False
premises: list[RealizedRecord] = []
for key in record.derivation.premise_structure_keys:
hits = recall_realized(ctx, structure_key=key)
if len(hits) != 1:
return False # a premise vanished or is ambiguous → cannot stand
premises.append(hits[0])
subject, target = record.relation_arguments
member_premises = tuple(p for p in premises if p.relation_predicate == "member")
subset_premises = tuple(p for p in premises if p.relation_predicate == "subset")
if record.relation_predicate == "member":
if len(member_premises) != 1:
return False
member_fact = member_premises[0]
path = _order_subset_path(member_fact.relation_arguments[1], subset_premises)
verdict = _verify_subsumption(
"member", subject, target, member_fact=member_fact, subset_path=path
)
else:
path = _order_subset_path(subject, subset_premises)
verdict = _verify_subsumption(
"subset", subject, target, member_fact=None, subset_path=path
)
return isinstance(verdict, Determined)
def run(depth: int = 9) -> dict[str, Any]:
"""Soak the closure and return the falsification report.
Runs to the fixed point (bounded by ``depth + 2`` ticks — a safety backstop, never
reached by a converging closure), then computes the falsification verdicts.
"""
ctx = _fresh_context()
seed_chain(ctx, depth)
effective_depth = max(1, min(depth, len(_CHAIN_PLURALS) - 1))
trajectory: list[TickRecord] = []
pre = len(_member_closure(ctx))
trajectory.append(TickRecord(0, pre, 0, 0, False))
max_ticks = effective_depth + 2
tick = 0
while tick < max_ticks:
tick += 1
result = consolidate_once(ctx)
trajectory.append(
TickRecord(
tick=tick,
member_closure_size=len(_member_closure(ctx)),
considered=result.considered,
consolidated=result.consolidated,
at_fixed_point=result.at_fixed_point,
)
)
if result.at_fixed_point:
break
sizes = [t.member_closure_size for t in trajectory]
# Monotone non-decreasing across all ticks; strictly increasing on every tick that
# consolidated something (a real layer must enlarge the directly-answerable set).
monotone = all(b >= a for a, b in zip(sizes, sizes[1:]))
strict_on_work = all(
t.member_closure_size > trajectory[i].member_closure_size
for i, t in enumerate(trajectory[1:])
if t.consolidated > 0
)
converged = trajectory[-1].at_fixed_point
# rex reaches every class on the chain it was seeded into (c0 told + c1..c_depth).
final_closure = list(_member_closure(ctx))
expected_closure = sorted(_CHAIN_LEMMAS[: effective_depth + 1])
closure_complete = final_closure == expected_closure
# wrong=0: the member ∘ member canary is never derived, and no class outside the
# chain ever appears in rex's closure.
forbidden_present = _CANARY_FORBIDDEN[1] in final_closure
chain_set = set(_CHAIN_LEMMAS[: effective_depth + 1])
no_fabricated = set(final_closure) <= chain_set
# Provenance replay: every derived record re-verifies ENTAILED from its premises.
derived_records = [
r
for r in recall_realized(ctx)
if r.derived and r.relation_predicate in ("member", "subset")
]
provenance_replay_ok = all(reverify_derived(ctx, r) for r in derived_records)
all_derived_speculative = all(
r.epistemic_status == "speculative" for r in derived_records
)
falsification_met = (
monotone
and strict_on_work
and converged
and closure_complete
and not forbidden_present
and no_fabricated
and provenance_replay_ok
and all_derived_speculative
)
return {
"depth": effective_depth,
"trajectory": [asdict(t) for t in trajectory],
"member_closure_sizes": sizes,
"final_member_closure": final_closure,
"expected_member_closure": expected_closure,
"derived_record_count": len(derived_records),
"verdicts": {
"monotone": monotone,
"strict_increase_on_consolidating_tick": strict_on_work,
"converged_to_fixed_point": converged,
"closure_complete": closure_complete,
"canary_member_member_never_derived": not forbidden_present,
"no_fabricated_membership": no_fabricated,
"provenance_replay_ok": provenance_replay_ok,
"all_derived_speculative": all_derived_speculative,
},
"falsification_met": falsification_met,
}

16
generate/determine/__init__.py
View file

@ -1,6 +1,18 @@
"""DETERMINE — reason over realized structure → assert (as-told) / refuse (roadmap Step 4)."""
"""DETERMINE — reason over realized structure → assert (as-told) / refuse (roadmap Step 4).
Step D (CLOSE) adds ``consolidate_once``: idle deductive consolidation of soundly-derived
facts back into the held self, so the loop learns from determined facts.
"""
from generate.determine.consolidate import ConsolidationResult, consolidate_once
from generate.determine.determine import Determined, Undetermined, determine
from generate.determine.render import render_determination
__all__ = ["Determined", "Undetermined", "determine", "render_determination"]
__all__ = [
"ConsolidationResult",
"Determined",
"Undetermined",
"consolidate_once",
"determine",
"render_determination",
]

189
generate/determine/consolidate.py Normal file
View file

@ -0,0 +1,189 @@
"""D — CLOSE (roadmap Step 5): idle deductive consolidation of derived facts.
The loop "learns from determined facts." Between turns, ``consolidate_once`` runs ONE
semi-naive layer of the member/subset deductive closure over the held self: for every
sound one-hop is-a inference whose conclusion is not yet realized, it VERIFIES the hop
with the sound+complete proof_chain ROBDD (reusing C's single verifier — no duplicate
proof logic) and writes the conclusion as a SPECULATIVE realized record carrying
derived-provenance (``generate.realize.realize_derived``). The next ``determine``
reaches the conclusion directly, and a later hop can chain off it — so the directly
answerable set climbs monotonically across idle ticks to the deductive-closure fixed
point, where a further tick is a no-op.
The only sound is-a rules are ``member ∘ subset → member`` and ``subset ∘ subset →
subset`` (Description-Logic subsumption). ``member ∘ member`` is NEVER an edge —
instance-of is not transitive ("Socrates is a man" + "man is a species" ⊬ "Socrates is
a species"). A consolidated ``member(s, t)`` is a member fact, extendable ONLY by a
subset edge, so the fallacy stays structurally unreachable across iterations.
wrong=0: every consolidated fact is the conclusion of a sound rule over realized
premises AND confirmed by the sound+complete decider (defence in depth). Honesty: a
fact derived from SPECULATIVE premises stays SPECULATIVE / as-told — the soundness of
the inference never upgrades the standing of the premises (COHERENT is never minted).
Boundary: SESSION memory (immediate, allowed), an extension of the ``generate.realize``
path — NOT reviewed/corpus learning and NOT coupled to proposals; the teaching/review
HITL path is untouched (no parallel learning path).
Determinism: a pure function of the current realized set. Candidates are consolidated
in sorted ``(predicate, subject, object)`` order, so the vault write order — and thus
the replayed structure — is independent of recall order. No clock, no LLM, no metric
call. Bounded by the same ``_SUBSUMPTION_SUBSET_FACT_BUDGET`` as the transitive search.
"""
from __future__ import annotations
from dataclasses import dataclass
from generate.determine.determine import (
_SUBSUMPTION_SUBSET_FACT_BUDGET,
Determined,
_verify_subsumption,
)
from generate.realize import RealizedRecord, Realized, realize_derived, recall_realized
from session.context import SessionContext
#: Rule label stamped on the derived record's provenance, keyed by the conclusion's
#: predicate. ``member`` conclusions come from ``member ∘ subset``; ``subset``
#: conclusions from ``subset ∘ subset``. (``member ∘ member`` is not a rule.)
_RULE_FOR_PREDICATE = {"member": "member_subset", "subset": "subset_subset"}
@dataclass(frozen=True, slots=True)
class ConsolidationResult:
"""Outcome of one idle consolidation layer.
``considered`` — one-hop conclusions not already realized (this tick's frontier).
``consolidated`` — conclusions newly written as derived realized records.
``redundant`` — verified conclusions that dedup-hit an existing record (defensive;
the frontier already excludes present facts).
``at_fixed_point`` — True iff nothing new was consolidated (the closure is
saturated; re-running is a no-op).
``budget_exceeded`` — True iff the realized subset-fact count exceeded the bound, so
the layer safely declined (a coverage backstop, never an unsound write).
"""
considered: int
consolidated: int
redundant: int
at_fixed_point: bool
budget_exceeded: bool = False
def _one_hop_candidates(
member_facts: tuple[RealizedRecord, ...],
subset_facts: tuple[RealizedRecord, ...],
) -> list[tuple[str, str, str, RealizedRecord | None, tuple[RealizedRecord, ...]]]:
"""Enumerate the sound one-hop conclusions NOT already realized.
Returns ``(predicate, subject, object, member_fact, subset_path)`` tuples ready for
``_verify_subsumption``. A ``member`` candidate carries its seed member fact and a
one-edge subset path; a ``subset`` candidate carries a two-edge subset path and no
member fact. Reflexive conclusions (subject == object) are excluded — they are not
real subsumption claims.
"""
existing: set[tuple[str, str, str]] = set()
for f in member_facts:
if len(f.relation_arguments) == 2:
existing.add(("member", f.relation_arguments[0], f.relation_arguments[1]))
for f in subset_facts:
if len(f.relation_arguments) == 2:
existing.add(("subset", f.relation_arguments[0], f.relation_arguments[1]))
# subset adjacency: class → [(superclass, fact)] — the subclass-of edges.
supers: dict[str, list[tuple[str, RealizedRecord]]] = {}
for f in subset_facts:
if len(f.relation_arguments) == 2:
supers.setdefault(f.relation_arguments[0], []).append(
(f.relation_arguments[1], f)
)
out: list[
tuple[str, str, str, RealizedRecord | None, tuple[RealizedRecord, ...]]
] = []
# member ∘ subset → member: member(s, b) + subset(b, t) ⊢ member(s, t).
for m in member_facts:
if len(m.relation_arguments) != 2:
continue
s, b = m.relation_arguments
for t, sub_fact in supers.get(b, ()):
if s == t:
continue
if ("member", s, t) not in existing:
out.append(("member", s, t, m, (sub_fact,)))
# subset ∘ subset → subset: subset(a, b) + subset(b, t) ⊢ subset(a, t).
for f in subset_facts:
if len(f.relation_arguments) != 2:
continue
a, b = f.relation_arguments
for t, sub_fact in supers.get(b, ()):
if a == t:
continue
if ("subset", a, t) not in existing:
out.append(("subset", a, t, None, (f, sub_fact)))
return out
def consolidate_once(
ctx: SessionContext, *, fact_budget: int = _SUBSUMPTION_SUBSET_FACT_BUDGET
) -> ConsolidationResult:
"""Consolidate one semi-naive layer of the member/subset deductive closure.
Reads the realized member + subset facts present NOW, derives every sound one-hop
conclusion not yet realized, proof_chain-verifies each, and writes the verified ones
as derived realized records. Facts written this tick are NOT re-read this tick — the
next layer derives off them next tick (semi-naive evaluation → monotone climb).
"""
member_facts = recall_realized(ctx, predicate="member")
subset_facts = recall_realized(ctx, predicate="subset")
if len(subset_facts) > fact_budget:
# Bounded — a safe coverage decline, never an unsound write.
return ConsolidationResult(
considered=0,
consolidated=0,
redundant=0,
at_fixed_point=False,
budget_exceeded=True,
)
candidates = _one_hop_candidates(member_facts, subset_facts)
# Deterministic write order, independent of recall order.
candidates.sort(key=lambda c: (c[0], c[1], c[2]))
consolidated = 0
redundant = 0
for predicate, subject, obj, member_fact, subset_path in candidates:
verdict = _verify_subsumption(
predicate, subject, obj, member_fact=member_fact, subset_path=subset_path
)
if not isinstance(verdict, Determined):
# The sound+complete decider did not confirm the hop — never write
# (defence in depth: a candidate-construction bug cannot consolidate a
# wrong fact).
continue
premise_keys = tuple(g.structure_key for g in verdict.grounds)
outcome = realize_derived(
ctx,
predicate=predicate,
subject=subject,
obj=obj,
rule=_RULE_FOR_PREDICATE[predicate],
premise_structure_keys=premise_keys,
)
if isinstance(outcome, Realized):
if outcome.created:
consolidated += 1
else:
redundant += 1
return ConsolidationResult(
considered=len(candidates),
consolidated=consolidated,
redundant=redundant,
at_fixed_point=consolidated == 0,
)
__all__ = ["ConsolidationResult", "consolidate_once"]

18
generate/determine/determine.py
View file

@ -241,6 +241,24 @@ def _verify_subsumption(
true atoms + the sound rule instantiated at each hop), so it scales — unlike the full
closure grounding. Returns ``None`` if the decider does not confirm entailment
(defence in depth: a path-construction bug cannot produce a wrong assertion)."""
# Soundness-by-construction (belt-and-suspenders): the propositional theory below
# labels every ``subset_path`` fact ``S`` and the ``member_fact`` ``M``. Both callers
# (``_determine_subsumption`` and ``consolidate``) build these from predicate-filtered
# recalls, so the labels match — but the theory would otherwise verify a smuggled
# member fact in ``subset_path`` as a (potentially unsound) subset edge. Refuse a
# mislabeled or wrong-arity chain here rather than trust two callers' discipline, so
# ``member ∘ member`` cannot be laundered through a corrupted path.
if member_fact is not None and (
member_fact.relation_predicate != "member"
or len(member_fact.relation_arguments) != 2
):
return None
if any(
f.relation_predicate != "subset" or len(f.relation_arguments) != 2
for f in subset_path
):
return None
atoms: dict[tuple[str, str, str], str] = {}
def atom(p: str, a: str, b: str) -> str:

4
generate/realize/__init__.py
View file

@ -2,18 +2,22 @@
from generate.realize.quantitative import realize_quantitative
from generate.realize.realize import (
Derivation,
NotRealized,
Realized,
RealizedRecord,
realize_comprehension,
realize_derived,
)
from generate.realize.recall import recall_realized
__all__ = [
"Derivation",
"NotRealized",
"Realized",
"RealizedRecord",
"realize_comprehension",
"realize_derived",
"realize_quantitative",
"recall_realized",
]

153
generate/realize/realize.py
View file

@ -29,6 +29,7 @@ NOTHING — it is returned as ``NotRealized(reason)``, never coerced into a vaul
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
import numpy as np
@ -47,6 +48,41 @@ _GROUNDING_FAILURES = (RuntimeError, KeyError, IndexError, ValueError)
_STRUCTURE_KIND_MEANING_GRAPH = "meaning_graph"
@dataclass(frozen=True, slots=True)
class Derivation:
"""Provenance of a DERIVED realized fact (Step D consolidation).
A derived fact is the conclusion of a SOUND is-a rule (``member_subset`` or
``subset_subset``) over premises that were themselves realized (told or previously
consolidated), confirmed by the sound+complete proof_chain ROBDD. ``rule`` names
the inference; ``premise_structure_keys`` are the span-free identities of the
premise records (so a replay re-fetches them and re-verifies the chain);
``verdict`` is the decider's outcome — always ``"entailed"`` (a non-entailed
candidate is never consolidated). This makes the soundness claim MEANINGFULLY FAIL
on replay (Schema-Defined Proof Obligations): re-deriving from the recorded
premises must still entail the fact.
"""
rule: str
premise_structure_keys: tuple[str, ...]
verdict: str
def as_dict(self) -> dict[str, Any]:
return {
"rule": self.rule,
"premise_structure_keys": list(self.premise_structure_keys),
"verdict": self.verdict,
}
@classmethod
def from_dict(cls, payload: dict[str, Any]) -> "Derivation":
return cls(
rule=payload["rule"],
premise_structure_keys=tuple(payload.get("premise_structure_keys", [])),
verdict=payload.get("verdict", ""),
)
@dataclass(frozen=True, slots=True)
class RealizedRecord:
"""The realized-knowledge record (mirrors the stored vault metadata)."""
@ -72,6 +108,12 @@ class RealizedRecord:
#: LIVE deque position at recall/write time — authoritative in the unbounded
#: session tier; provenance-only under bounded-vault eviction.
vault_index: int
#: Step D — a TOLD fact is ``derived=False`` (``derivation=None``). A consolidated
#: fact (the conclusion of a sound is-a rule, proof_chain-verified) is
#: ``derived=True`` with a :class:`Derivation`. Defaults preserve the told-record
#: shape (and bit-exact persistence) for every pre-D caller.
derived: bool = False
derivation: "Derivation | None" = None
@dataclass(frozen=True, slots=True)
@ -212,22 +254,25 @@ def _realize_structured(
replay_hash: str,
versor: np.ndarray,
status: EpistemicStatus,
derivation: "Derivation | None" = None,
) -> Realized:
"""Dedup-or-store a realized record — the single wrong=0 write path shared by
every substrate (meaning_graph relations, binding_graph quantities).
every substrate (meaning_graph relations, binding_graph quantities, and Step D
derived facts).
Idempotency dedups exactly on the span-free ``structure_key`` (callers refuse
ambiguous identity before getting here, so a hit is always the genuinely-same
proposition, never a distinct one). The ``vault.store`` call is the only
mutation; ``iter_metadata`` is the public read-only accessor and ``idx`` is the
LIVE deque position. The epistemic status is whatever the caller declared —
SPECULATIVE, never COHERENT by default (ADR-0021).
SPECULATIVE, never COHERENT by default (ADR-0021). A ``derivation`` marks a
consolidated (derived) fact; ``None`` is a told fact (the historical shape).
"""
for idx, meta in ctx.vault.iter_metadata():
if meta.get("kind") == "realized" and meta.get("structure_key") == structure_key:
return Realized(record=_record_from_metadata(meta, idx), created=False)
metadata = {
metadata: dict[str, Any] = {
"kind": "realized",
"structure_kind": structure_kind,
"structure_canonical": structure_canonical,
@ -240,6 +285,11 @@ def _realize_structured(
"replay_hash": replay_hash,
"tier": "session",
}
if derivation is not None:
# Derived-fact provenance is additive — a told fact omits both keys, so its
# on-disk metadata stays byte-identical to the pre-D encoding.
metadata["derived"] = True
metadata["derivation"] = derivation.as_dict()
vault_index = ctx.vault.store(versor, metadata, epistemic_status=status)
return Realized(
record=RealizedRecord(
@ -255,13 +305,104 @@ def _realize_structured(
epistemic_status=status.value,
tier="session",
vault_index=vault_index,
derived=derivation is not None,
derivation=derivation,
),
created=True,
)
def realize_derived(
ctx: SessionContext,
*,
predicate: str,
subject: str,
obj: str,
rule: str,
premise_structure_keys: tuple[str, ...],
) -> Realized | NotRealized:
"""Consolidate a SOUNDLY-DERIVED fact into the held self (Step D — CLOSE).
The caller (``generate.determine.consolidate``) has already found and
proof_chain-VERIFIED the is-a chain that entails ``predicate(subject, obj)`` — this
only writes the conclusion as a realized record so the next ``determine`` reaches it
directly. The record is:
- SPECULATIVE / ``as_told`` — a sound INFERENCE never upgrades the STANDING of its
(SPECULATIVE) premises; COHERENT is never minted here (ADR-0021 honesty).
- SESSION memory (immediate) — NOT reviewed/corpus learning; nothing is proposed,
the teaching/review HITL path is untouched (no parallel learning path).
- keyed by the SAME span-free ``structure_key`` a told ``predicate(subject, obj)``
would carry, so it dedups against / is found by the told path identically, and
is recalled by ``recall_realized`` like any other realized fact.
- provenance-rich (``Derivation``): the premise ``structure_key``s + rule + the
ENTAILED verdict, so a replay re-derives and re-verifies (the soundness claim
can MEANINGFULLY FAIL).
Idempotent (dedups on ``structure_key``); returns ``NotRealized`` on any ineligible
input or grounding failure — never a partial / coerced write (wrong=0).
"""
subject = (subject or "").strip()
obj = (obj or "").strip()
predicate = (predicate or "").strip()
if not subject or not obj or not predicate:
return NotRealized("incomplete_derived_fact")
# The derived fact embeds its subject identically to a told fact about that subject
# (same probe_ingest placement). The versor is not the recall key (structural
# metadata is), so a degenerate placement is a clean no-op, never a wrong write.
try:
field_state = ctx.probe_ingest([subject])
except _GROUNDING_FAILURES:
return NotRealized("grounding_failed")
versor = np.asarray(field_state.F, dtype=np.float32)
relation_arguments = (subject, obj)
structure_canonical = f"derived:{predicate}({subject},{obj})"
source_span = f"derived:{rule}"
content_hash = sha256_of(structure_canonical) # span-inclusive provenance
structure_key = sha256_of(
{
"structure_kind": _STRUCTURE_KIND_MEANING_GRAPH,
"predicate": predicate,
"negated": False,
"arguments": list(relation_arguments),
}
)
status = EpistemicStatus.SPECULATIVE # derived from SPECULATIVE premises — as-told
replay_hash = sha256_of(
{
"content_hash": content_hash,
"source_span": source_span,
"epistemic_status": status.value,
}
)
entity_names = tuple(sorted({subject, obj}))
derivation = Derivation(
rule=rule,
premise_structure_keys=tuple(premise_structure_keys),
verdict="entailed",
)
return _realize_structured(
ctx,
structure_kind=_STRUCTURE_KIND_MEANING_GRAPH,
structure_canonical=structure_canonical,
relation_predicate=predicate,
relation_arguments=relation_arguments,
entity_names=entity_names,
source_span=source_span,
content_hash=content_hash,
structure_key=structure_key,
replay_hash=replay_hash,
versor=versor,
status=status,
derivation=derivation,
)
def _record_from_metadata(meta: dict, idx: int) -> RealizedRecord:
"""Reconstruct a RealizedRecord from a stored vault metadata dict."""
derivation_payload = meta.get("derivation")
return RealizedRecord(
structure_kind=meta.get("structure_kind", _STRUCTURE_KIND_MEANING_GRAPH),
structure_canonical=meta.get("structure_canonical", ""),
@ -275,4 +416,10 @@ def _record_from_metadata(meta: dict, idx: int) -> RealizedRecord:
epistemic_status=meta.get("epistemic_status", "speculative"),
tier=meta.get("tier", "session"),
vault_index=idx,
derived=bool(meta.get("derived", False)),
derivation=(
Derivation.from_dict(derivation_payload)
if isinstance(derivation_payload, dict)
else None
),
)

55
tests/test_determination_closure_lane.py Normal file
View file

@ -0,0 +1,55 @@
"""Step D falsification lane — ``evals.determination_closure``.
The lane is the deterministic, falsifiable proof that idle consolidation makes the
engine learn from its determined facts: the directly-answerable set climbs monotonically
across idle ticks to the deductive-closure fixed point, with wrong=0 (the member ∘
member fallacy is never derived), honesty (derived facts stay SPECULATIVE), and a
provenance replay obligation (every derived record re-verifies ENTAILED).
"""
from __future__ import annotations
from evals.determination_closure import run
def test_falsification_met() -> None:
report = run(depth=9)
assert report["falsification_met"] is True, report["verdicts"]
def test_each_verdict_holds() -> None:
v = run(depth=9)["verdicts"]
assert v["monotone"]
assert v["strict_increase_on_consolidating_tick"]
assert v["converged_to_fixed_point"]
assert v["closure_complete"]
assert v["canary_member_member_never_derived"]
assert v["no_fabricated_membership"]
assert v["provenance_replay_ok"]
assert v["all_derived_speculative"]
def test_closure_climbs_strictly_then_plateaus() -> None:
sizes = run(depth=9)["member_closure_sizes"]
# Non-decreasing throughout.
assert all(b >= a for a, b in zip(sizes, sizes[1:]))
# Starts at 1 (the single told membership) and ends at the full chain (10 classes).
assert sizes[0] == 1
assert sizes[-1] == 10
# The final transition is a no-op (the converged fixed point repeats the size).
assert sizes[-1] == sizes[-2]
def test_run_is_deterministic() -> None:
a = run(depth=7)
b = run(depth=7)
assert a["member_closure_sizes"] == b["member_closure_sizes"]
assert a["final_member_closure"] == b["final_member_closure"]
assert a["derived_record_count"] == b["derived_record_count"]
def test_depth_is_clamped_to_chain_length() -> None:
# Asking for more depth than the chain provides clamps, never errors.
report = run(depth=999)
assert report["depth"] == 9
assert report["falsification_met"] is True

249
tests/test_determination_consolidation.py Normal file
View file

@ -0,0 +1,249 @@
"""Step D — CLOSE: idle deductive consolidation of soundly-derived facts.
The loop learns from *determined* facts: between turns, ``consolidate_once`` writes each
soundly-derived (proof_chain-verified) member/subset conclusion back into the held self
as a SPECULATIVE realized record, so the next ``determine`` reaches it directly and the
directly-answerable set climbs across idle ticks.
The load-bearing wrong=0 bite: a consolidated fact must be SOUND (the member ∘ member
fallacy must never be consolidated) and HONEST (a fact derived from SPECULATIVE premises
stays SPECULATIVE — a sound inference never mints COHERENT).
"""
from __future__ import annotations
from dataclasses import replace
from pathlib import Path
import pytest
from chat.runtime import ChatRuntime
from core.config import DEFAULT_CONFIG
from generate.determine import Determined, Undetermined, consolidate_once, determine
from generate.determine.determine import _verify_subsumption
from generate.meaning_graph.reader import comprehend
from generate.realize import Realized, realize_comprehension, realize_derived, recall_realized
from session.context import SessionContext
_HIGH = 10**9
@pytest.fixture(scope="module")
def vocab_persona():
rt = ChatRuntime(no_load_state=True)
return rt._context.vocab, rt._context.persona
def _ctx(vocab_persona) -> SessionContext:
vocab, persona = vocab_persona
return SessionContext(vocab=vocab, persona=persona, vault_reproject_interval=_HIGH)
def _tell(text: str, ctx: SessionContext):
return realize_comprehension(comprehend(text), ctx)
def _ask(text: str, ctx: SessionContext):
return determine(comprehend(text), ctx)
def _members(ctx: SessionContext, subject: str) -> set[str]:
return {f.relation_arguments[1] for f in recall_realized(ctx, subject=subject, predicate="member")}
# --------------------------------------------------------------------------- #
# realize_derived — the consolidated record
# --------------------------------------------------------------------------- #
def test_realize_derived_writes_speculative_provenance_record(vocab_persona) -> None:
ctx = _ctx(vocab_persona)
_tell("Socrates is a man.", ctx)
_tell("All men are mortals.", ctx)
out = realize_derived(
ctx,
predicate="member",
subject="socrates",
obj="mortal",
rule="member_subset",
premise_structure_keys=("k1", "k2"),
)
assert isinstance(out, Realized) and out.created is True
rec = out.record
assert rec.relation_predicate == "member"
assert rec.relation_arguments == ("socrates", "mortal")
assert rec.derived is True
assert rec.derivation is not None
assert rec.derivation.rule == "member_subset"
assert rec.derivation.verdict == "entailed"
assert rec.derivation.premise_structure_keys == ("k1", "k2")
# A sound INFERENCE never upgrades the STANDING of its (SPECULATIVE) premises.
assert rec.epistemic_status == "speculative"
def test_realize_derived_is_idempotent(vocab_persona) -> None:
ctx = _ctx(vocab_persona)
_tell("Socrates is a man.", ctx)
a = realize_derived(ctx, predicate="member", subject="socrates", obj="mortal", rule="member_subset", premise_structure_keys=())
b = realize_derived(ctx, predicate="member", subject="socrates", obj="mortal", rule="member_subset", premise_structure_keys=())
assert a.created is True and b.created is False
assert a.record.structure_key == b.record.structure_key
def test_derived_structure_key_matches_told(vocab_persona) -> None:
# A derived member(s,t) carries the SAME span-free structure_key a TOLD
# "s is a t" would — so the told path finds / dedups against it identically.
ctx_a = _ctx(vocab_persona)
told = _tell("Socrates is a mortal.", ctx_a)
ctx_b = _ctx(vocab_persona)
derived = realize_derived(ctx_b, predicate="member", subject="socrates", obj="mortal", rule="member_subset", premise_structure_keys=())
assert told.record.structure_key == derived.record.structure_key
# --------------------------------------------------------------------------- #
# consolidate_once — one deductive-closure layer
# --------------------------------------------------------------------------- #
def test_consolidate_derives_then_is_recalled_directly(vocab_persona) -> None:
ctx = _ctx(vocab_persona)
_tell("Socrates is a man.", ctx)
_tell("All men are mortals.", ctx)
assert _members(ctx, "socrates") == {"man"} # only the told fact
result = consolidate_once(ctx)
assert result.consolidated >= 1
assert "mortal" in _members(ctx, "socrates") # now directly realized
# And DETERMINE answers it directly (a told fact path, not a re-derivation).
res = _ask("Is Socrates a mortal?", ctx)
assert isinstance(res, Determined) and res.answer is True
def test_consolidation_climbs_monotonically_to_fixed_point(vocab_persona) -> None:
ctx = _ctx(vocab_persona)
_tell("Rex is a dog.", ctx)
_tell("All dogs are mammals.", ctx)
_tell("All mammals are animals.", ctx)
_tell("All animals are creatures.", ctx)
sizes = [len(_members(ctx, "rex"))]
for _ in range(6):
r = consolidate_once(ctx)
sizes.append(len(_members(ctx, "rex")))
if r.at_fixed_point:
break
assert all(b >= a for a, b in zip(sizes, sizes[1:])) # monotone
assert sizes[-1] == 4 # dog, mammal, animal, creature
# A further tick is a no-op (converged).
assert consolidate_once(ctx).at_fixed_point is True
def test_subset_transitivity_is_consolidated(vocab_persona) -> None:
ctx = _ctx(vocab_persona)
_tell("All dogs are mammals.", ctx)
_tell("All mammals are animals.", ctx)
consolidate_once(ctx)
subs = {f.relation_arguments[1] for f in recall_realized(ctx, subject="dog", predicate="subset")}
assert "animal" in subs # subset ∘ subset → subset(dog, animal)
# --------------------------------------------------------------------------- #
# wrong=0 — the member ∘ member fallacy stays unreachable
# --------------------------------------------------------------------------- #
def test_member_member_fallacy_never_consolidated(vocab_persona) -> None:
ctx = _ctx(vocab_persona)
_tell("Socrates is a man.", ctx)
_tell("Man is a species.", ctx) # member(man, species) — NOT subset
for _ in range(4):
if consolidate_once(ctx).at_fixed_point:
break
# "Socrates is a species" must NEVER be derived (instance-of is not transitive).
assert "species" not in _members(ctx, "socrates")
assert isinstance(_ask("Is Socrates a species?", ctx), Undetermined)
def test_verify_subsumption_refuses_mislabeled_path(vocab_persona) -> None:
# Belt-and-suspenders: the verifier labels subset_path facts "S". If a MEMBER fact
# is smuggled into subset_path (member ∘ member laundered as a subset edge), the
# guard must refuse rather than verify an unsound chain.
ctx = _ctx(vocab_persona)
_tell("Socrates is a man.", ctx)
_tell("Man is a species.", ctx) # member(man, species)
member_socrates_man = recall_realized(ctx, subject="socrates", predicate="member")[0]
member_man_species = recall_realized(ctx, subject="man", predicate="member")[0]
# Feed the member(man, species) fact where a subset edge is expected.
out = _verify_subsumption(
"member",
"socrates",
"species",
member_fact=member_socrates_man,
subset_path=(member_man_species,),
)
assert out is None # refused — the mislabeled path cannot launder the fallacy
def test_no_membership_fabricated_outside_chain(vocab_persona) -> None:
ctx = _ctx(vocab_persona)
_tell("Rex is a dog.", ctx)
_tell("All dogs are mammals.", ctx)
_tell("Whales are mammals.", ctx) # member(whale, mammal) — unrelated subject
for _ in range(4):
if consolidate_once(ctx).at_fixed_point:
break
# rex's closure is exactly its reachable classes; nothing leaks from other subjects.
assert _members(ctx, "rex") == {"dog", "mammal"}
# --------------------------------------------------------------------------- #
# idle_tick wiring + reboot persistence
# --------------------------------------------------------------------------- #
def test_idle_tick_consolidates_when_flagged(tmp_path: Path) -> None:
cfg = replace(
DEFAULT_CONFIG,
consolidate_determinations=True,
accrue_realized_knowledge=True,
persist_session_state=True,
)
rt = ChatRuntime(config=cfg, engine_state_path=tmp_path)
ctx = rt._context
realize_comprehension(comprehend("Socrates is a man."), ctx)
realize_comprehension(comprehend("All men are mortals."), ctx)
result = rt.idle_tick()
assert result.facts_consolidated >= 1
assert "mortal" in _members(ctx, "socrates")
def test_idle_tick_noop_when_flag_off(tmp_path: Path) -> None:
cfg = replace(DEFAULT_CONFIG, consolidate_determinations=False, accrue_realized_knowledge=True)
rt = ChatRuntime(config=cfg, engine_state_path=tmp_path)
realize_comprehension(comprehend("Socrates is a man."), rt._context)
realize_comprehension(comprehend("All men are mortals."), rt._context)
assert rt.idle_tick().facts_consolidated == 0
assert _members(rt._context, "socrates") == {"man"}
def test_consolidated_facts_persist_across_reboot(tmp_path: Path) -> None:
cfg = replace(
DEFAULT_CONFIG,
consolidate_determinations=True,
accrue_realized_knowledge=True,
persist_session_state=True,
)
rt = ChatRuntime(config=cfg, engine_state_path=tmp_path)
realize_comprehension(comprehend("Socrates is a man."), rt._context)
realize_comprehension(comprehend("All men are mortals."), rt._context)
for _ in range(3):
if rt.idle_tick().facts_consolidated == 0:
break
before = _members(rt._context, "socrates")
assert "mortal" in before
# Reboot: a new runtime over the same state dir resumes the SAME consolidated life.
rt2 = ChatRuntime(config=cfg, engine_state_path=tmp_path)
assert _members(rt2._context, "socrates") == before
derived = [
r for r in recall_realized(rt2._context, subject="socrates", predicate="member") if r.derived
]
assert derived and all(r.derivation is not None and r.derivation.verdict == "entailed" for r in derived)