feat(adr-0056): Phase C1 — contemplation loop landed

Implements ADR-0056's cognitive surface: takes a Phase B
DiscoveryCandidate and returns an enriched candidate with composed
polarity, classified claim_domain, evidence pointers, and recursive
sub-questions.  No corpus mutation; no async; no LLM step.

Changes
- teaching/discovery.py: DiscoveryCandidate gains six C1 fields
  with defaults that preserve Phase B JSONL byte-equality.  Adds
  EvidencePointer, SubQuestion, ClaimDomain types.
- teaching/contemplation.py (new): contemplate(candidate) +
  canonical probe order (vault → pack → corpus), deterministic
  decomposition over corpus-known intent objects, composition
  rules from ADR-0056 §Composition, bounded-depth failsafe with
  recursion_overflow audit signal.  Vault probe is injectable;
  None means no vault contribution this pass.
- tests/test_contemplation.py (16 tests): determinism (byte-
  identical JSONL), no input/corpus mutation, empty pack+corpus
  termination with gap-recorded sub-question, factual affirming
  composition, direct same-pack contradiction → falsifies, mixed
  evidence → undetermined + domain upgrade, recursion overflow,
  frame-dependent connective → relational, Phase B byte-equality
  preserved on uncontemplated candidates, sub_id stability,
  evidence pointer admissibility, vault probe injection +
  exception isolation.

Invariants preserved
- versor_condition(F) < 1e-6 — C1 touches no algebra path.
- No corpus / pack / runtime mutation — trust boundary intact.
- No clock-time, no LLM, no stochastic sampling, no async.

Lanes
- smoke 67, cognition 121, runtime 19, teaching 17, contemplation 16.
- core eval cognition: intent 100% / surface 100% /
  term_capture 91.7% / versor 100% — unchanged.

Open questions stay open: frame-dependent connective table
authorship (v1 lives as a small constant in contemplation.py
pending pack-data migration), person-axis intent classification
for auto-evaluative, recursion-overflow telemetry shape, sub-
question deduplication.  None block C1.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
Shay 2026-05-18 10:06:18 -07:00
parent f1121b5822
commit 4eecf73a05
3 changed files with 967 additions and 6 deletions

505
teaching/contemplation.py Normal file
View file

@ -0,0 +1,505 @@
"""ADR-0056 Phase C1 — Contemplation loop.
``contemplate(candidate)`` takes a Phase B ``DiscoveryCandidate``
(a *posed question*: "would a chain of shape (subject, intent) have
grounded this turn?") and returns an *enriched* candidate with:
- ``polarity {affirms, falsifies, undetermined}`` what
composed reviewed evidence says about the proposed relation.
- ``claim_domain {factual, relational, evaluative}`` the
epistemic register the claim sits in. Determines the evidence
threshold the future C2 review gate will demand.
- ``evidence`` tuple of ``EvidencePointer`` from the canonical
probe order (vault pack corpus).
- ``sub_questions`` decomposed sub-questions and their outcomes
(``grounded``, ``gap_recorded``, ``depth_failsafe``).
- ``contemplation_depth`` recursion depth reached.
- ``recursion_overflow`` True iff the bounded-depth failsafe
fired. Hitting the ceiling is itself an audit event;
contemplation never silently truncates.
The loop is a pure function of the candidate, the reviewed teaching
corpus, the ratified cognition pack, and an optional vault probe
hook. No clock-time, no LLM, no stochastic sampling, no concurrency
ADR-0056 Call 4 (sync, not async).
Trust boundary: this module reads ``_pack_index()`` and
``_corpus_index()`` only. It NEVER writes to the corpus, the pack,
or runtime state. Output enriched candidates flow back through the
same Phase B sink as JSONL lines.
"""
from __future__ import annotations
import hashlib
from dataclasses import replace
from typing import Any, Callable, Literal
from chat.pack_grounding import _pack_index
from chat.teaching_grounding import _corpus_index
from teaching.discovery import (
ClaimDomain,
DiscoveryCandidate,
EvidencePointer,
SubQuestion,
)
# Frame-dependent connectives (open question §1 in ADR-0056). v1
# list lives here as a small reviewed constant; the long-term home
# is versioned pack data so that refining the taxonomy doesn't
# require a code change. Adding/removing entries here is a reviewed
# code change, same as any other reviewed surface.
_FRAME_DEPENDENT_CONNECTIVES: frozenset[str] = frozenset({
"orders",
"grounds",
"informs",
"constrains",
})
_VaultProbe = Callable[[str, str], tuple[EvidencePointer, ...]]
"""Optional injectable vault probe.
Signature: ``probe(subject_lemma, object_lemma) -> tuple[EvidencePointer, ...]``.
Implementations MUST return only ``vault_coherent`` pointers
(``EpistemicStatus.COHERENT``); SPECULATIVE / CONTESTED / FALSIFIED
vault entries are filtered out by the implementation, not by the
loop. ``None`` means "no vault probe in this contemplation pass."
"""
_DEFAULT_MAX_DEPTH: int = 8
# ---------------------------------------------------------------------------
# Sub-question id derivation
# ---------------------------------------------------------------------------
def _sub_id(parent_candidate_id: str, index: int, payload: dict[str, Any]) -> str:
"""Deterministic sub-question id.
SHA-256 over ``(parent_id, index, sorted_payload_json)`` keeps the
id stable across runs and ties the sub-question's identity to
both its parent and its content.
"""
import json as _json
blob = _json.dumps(
{"parent": parent_candidate_id, "index": index, "payload": payload},
sort_keys=True,
separators=(",", ":"),
)
return hashlib.sha256(blob.encode("utf-8")).hexdigest()[:32]
# ---------------------------------------------------------------------------
# Probing — vault → pack → corpus
# ---------------------------------------------------------------------------
def _probe_corpus_direct(
subject: str, intent: str, connective: str | None, obj: str | None
) -> tuple[EvidencePointer, ...]:
"""Look in the active reviewed corpus for affirming/falsifying chains.
- Exact match on ``(subject, intent, connective, object)`` is
affirming evidence (the proposed chain already exists).
- Same ``(subject, intent, object)`` but different connective is
a same-pack contradiction falsifying evidence.
- ``(subject, intent)`` match with no object filter and any
connective is weak affirming evidence (the *shape* exists in
reviewed memory).
"""
out: list[EvidencePointer] = []
corpus = _corpus_index()
chain = corpus.get((subject, intent))
if chain is None:
return ()
if obj is None and connective is None:
# Phase B shape: shape evidence only. The exact (subject,
# intent) cell is in the corpus — affirming.
out.append(EvidencePointer(
source="corpus",
ref=chain.chain_id,
polarity="affirms",
epistemic_status="coherent",
))
return tuple(out)
if obj is not None and chain.object == obj:
if connective is None or chain.connective == connective:
out.append(EvidencePointer(
source="corpus",
ref=chain.chain_id,
polarity="affirms",
epistemic_status="coherent",
))
else:
# Same subject + intent + object, different connective.
# Direct same-pack contradiction.
out.append(EvidencePointer(
source="corpus",
ref=chain.chain_id,
polarity="falsifies",
epistemic_status="coherent",
))
return tuple(out)
def _probe_pack(subject: str, obj: str | None) -> tuple[EvidencePointer, ...]:
"""Pack lemma residency is shape-level affirming evidence.
A pack-resident subject means the subject is grounded; if both
subject and object are pack-resident, the relation has both
endpoints anchored in ratified memory. Pack residency cannot
falsify (pack ``semantic_domains`` don't express negation —
Call 2 of ADR-0056).
"""
pack = _pack_index()
out: list[EvidencePointer] = []
if subject in pack:
out.append(EvidencePointer(
source="pack",
ref=subject,
polarity="affirms",
epistemic_status="coherent",
))
if obj is not None and obj in pack:
out.append(EvidencePointer(
source="pack",
ref=obj,
polarity="affirms",
epistemic_status="coherent",
))
return tuple(out)
def _probe_vault(
subject: str, obj: str | None, vault_probe: _VaultProbe | None
) -> tuple[EvidencePointer, ...]:
if vault_probe is None or obj is None:
return ()
try:
return tuple(vault_probe(subject, obj))
except Exception: # pragma: no cover — defensive: vault probe must not poison loop
return ()
# ---------------------------------------------------------------------------
# Decomposition
# ---------------------------------------------------------------------------
def _decompose(
candidate: DiscoveryCandidate,
) -> tuple[dict[str, Any], ...]:
"""Return decomposed sub-question payloads.
For a Phase B partial chain ``(subject, intent, None, None)``,
enumerate every reviewed object the corpus has used with the
same ``intent`` and treat each as a candidate match for
``subject``. This is the deterministic, pack-grounded analogue
of "what could this relation be about?"
Returns an empty tuple when no decomposition is possible the
parent records the gap (Call 1 of ADR-0056) and stops.
"""
intent = str(candidate.proposed_chain.get("intent") or "")
if not intent:
return ()
obj = candidate.proposed_chain.get("object")
if obj is not None:
# Already has a concrete object — no further decomposition.
return ()
corpus = _corpus_index()
# Deterministic order: sort by object lemma.
seen_objects: list[tuple[str, str]] = []
for key, chain in corpus.items():
if key[1] != intent:
continue
seen_objects.append((chain.object, chain.connective))
if not seen_objects:
return ()
seen_objects.sort()
subject = str(candidate.proposed_chain.get("subject") or "")
out: list[dict[str, Any]] = []
for cand_obj, cand_conn in seen_objects:
out.append({
"subject": subject,
"intent": intent,
"connective": cand_conn,
"object": cand_obj,
})
return tuple(out)
# ---------------------------------------------------------------------------
# Classification + composition
# ---------------------------------------------------------------------------
def _classify_claim_domain(chain: dict[str, Any]) -> ClaimDomain:
"""Deterministic claim-domain classification.
- ``relational`` if the connective is in the reviewed
frame-dependent set (e.g. ``orders``, ``grounds``).
- ``factual`` otherwise (the default for pack-resident
cognition lemmas).
- ``evaluative`` is NOT auto-assigned in C1 open question §2
in ADR-0056. Operator-assignable only.
"""
connective = str(chain.get("connective") or "").strip().lower()
if connective and connective in _FRAME_DEPENDENT_CONNECTIVES:
return "relational"
return "factual"
_DOMAIN_TIER: dict[ClaimDomain, int] = {
"factual": 0,
"relational": 1,
"evaluative": 2,
}
_DOMAIN_BY_TIER: dict[int, ClaimDomain] = {
0: "factual",
1: "relational",
2: "evaluative",
}
def _upgrade_domain(domain: ClaimDomain) -> ClaimDomain:
tier = _DOMAIN_TIER[domain]
return _DOMAIN_BY_TIER[min(tier + 1, 2)]
def _compose_polarity(
direct_evidence: tuple[EvidencePointer, ...],
sub_questions: tuple[SubQuestion, ...],
) -> Literal["affirms", "falsifies", "undetermined"]:
"""Reduce evidence + sub-question outcomes to one polarity verdict.
Rules (Call 1 + Call 2 of ADR-0056):
- Any direct ``falsifies`` evidence on the parent ``falsifies``.
A same-pack contradiction overrides supporting sub-evidence
because reviewed contradiction is the strongest signal.
- All admissible evidence ``affirms`` and at least one direct
reviewed pointer (corpus or vault_coherent) ``affirms``.
- Mixed (some affirm, some falsify, but no direct parent-level
falsification) ``undetermined``.
- No admissible evidence at all ``undetermined``.
"""
# Direct same-pack contradiction is dispositive — but ONLY when
# the falsifying pointer comes from the reviewed teaching corpus
# (Call 2 of ADR-0056: reviewed evidence in the same pack family).
# Vault and pack pointers cannot dispositively falsify; they
# contest but compose into the mixed-evidence path below.
if any(
e.polarity == "falsifies" and e.source == "corpus"
for e in direct_evidence
):
return "falsifies"
# Gather all evidence pointers (direct + sub-question contributions).
all_evidence: list[EvidencePointer] = list(direct_evidence)
for sq in sub_questions:
all_evidence.extend(sq.evidence)
if not all_evidence:
return "undetermined"
has_falsifies = any(e.polarity == "falsifies" for e in all_evidence)
has_affirms = any(e.polarity == "affirms" for e in all_evidence)
if has_falsifies and has_affirms:
return "undetermined"
if has_falsifies:
return "falsifies"
# Require at least one *reviewed* affirming pointer (corpus or
# vault_coherent) before promoting to ``affirms`` — pack
# residency alone is shape evidence, not relation evidence.
has_reviewed_affirm = any(
e.polarity == "affirms" and e.source in ("corpus", "vault_coherent")
for e in all_evidence
)
if has_reviewed_affirm:
return "affirms"
return "undetermined"
# ---------------------------------------------------------------------------
# The loop itself
# ---------------------------------------------------------------------------
def _materialise_sub_candidate(
parent: DiscoveryCandidate, sub_payload: dict[str, Any], index: int
) -> DiscoveryCandidate:
"""Build a sub-candidate from a decomposed payload.
Sub-candidates inherit ``trigger`` and ``source_turn_trace`` from
the parent. The ``candidate_id`` is derived deterministically
from parent + index + payload same as ``_sub_id``.
"""
sub_id = _sub_id(parent.candidate_id, index, sub_payload)
return replace(
parent,
candidate_id=sub_id,
proposed_chain=dict(sub_payload),
contemplation_depth=parent.contemplation_depth + 1,
evidence=(),
sub_questions=(),
polarity="undetermined",
claim_domain="factual",
recursion_overflow=False,
)
def _probe(
chain: dict[str, Any], vault_probe: _VaultProbe | None
) -> tuple[EvidencePointer, ...]:
"""Canonical probe order: vault → pack → corpus.
The first source that grounds wins for *that* axis, but all
admissible pointers contribute composition reduces them.
"""
subject = str(chain.get("subject") or "").strip().lower()
intent = str(chain.get("intent") or "").strip().lower()
connective_raw = chain.get("connective")
connective = str(connective_raw).strip().lower() if connective_raw else None
obj_raw = chain.get("object")
obj = str(obj_raw).strip().lower() if obj_raw else None
out: list[EvidencePointer] = []
out.extend(_probe_vault(subject, obj, vault_probe))
out.extend(_probe_pack(subject, obj))
out.extend(_probe_corpus_direct(subject, intent, connective, obj))
return tuple(out)
def _gap_subquestion(parent: DiscoveryCandidate) -> SubQuestion:
subject = str(parent.proposed_chain.get("subject") or "")
intent = str(parent.proposed_chain.get("intent") or "")
payload = {"subject": subject, "intent": intent, "outcome": "gap_recorded"}
return SubQuestion(
sub_id=_sub_id(parent.candidate_id, -1, payload),
proposed_subject=subject,
proposed_intent=intent,
outcome="gap_recorded",
evidence=(),
)
def _depth_failsafe_subquestion(parent: DiscoveryCandidate) -> SubQuestion:
subject = str(parent.proposed_chain.get("subject") or "")
intent = str(parent.proposed_chain.get("intent") or "")
payload = {"subject": subject, "intent": intent, "outcome": "depth_failsafe"}
return SubQuestion(
sub_id=_sub_id(parent.candidate_id, -2, payload),
proposed_subject=subject,
proposed_intent=intent,
outcome="depth_failsafe",
evidence=(),
)
def contemplate(
candidate: DiscoveryCandidate,
*,
max_depth: int = _DEFAULT_MAX_DEPTH,
vault_probe: _VaultProbe | None = None,
) -> DiscoveryCandidate:
"""Run the contemplation loop on a single candidate.
Returns an *enriched* candidate (same id, populated C1 fields).
Never mutates the corpus, the pack, or the input candidate
(``DiscoveryCandidate`` is frozen).
"""
# Failsafe (Call 1 of ADR-0056): bounded depth ceiling whose hit
# is itself an audit event, not a silent truncation.
if candidate.contemplation_depth >= max_depth:
return replace(
candidate,
recursion_overflow=True,
sub_questions=(_depth_failsafe_subquestion(candidate),),
)
# Direct probe on the parent chain.
direct_evidence = _probe(candidate.proposed_chain, vault_probe)
# Decompose into sub-questions.
sub_payloads = _decompose(candidate)
if not sub_payloads:
# Terminal: cannot decompose further. Record the gap.
# Direct evidence (if any) still composes — a parent may be
# directly groundable without sub-decomposition.
if direct_evidence:
polarity = _compose_polarity(direct_evidence, ())
domain = _classify_claim_domain(candidate.proposed_chain)
if polarity == "undetermined":
has_aff = any(p.polarity == "affirms" for p in direct_evidence)
has_fal = any(p.polarity == "falsifies" for p in direct_evidence)
if has_aff and has_fal:
domain = _upgrade_domain(domain)
return replace(
candidate,
polarity=polarity,
claim_domain=domain,
evidence=direct_evidence,
sub_questions=(),
)
# No evidence and no decomposition → gap recorded.
return replace(
candidate,
polarity="undetermined",
claim_domain=_classify_claim_domain(candidate.proposed_chain),
evidence=(),
sub_questions=(_gap_subquestion(candidate),),
)
sub_results: list[SubQuestion] = []
for index, payload in enumerate(sub_payloads):
sub_candidate = _materialise_sub_candidate(candidate, payload, index)
recursed = contemplate(
sub_candidate, max_depth=max_depth, vault_probe=vault_probe
)
outcome: Literal["grounded", "gap_recorded", "depth_failsafe"]
if recursed.recursion_overflow:
outcome = "depth_failsafe"
elif recursed.evidence and recursed.polarity != "undetermined":
outcome = "grounded"
elif recursed.evidence:
# Has evidence but composed to undetermined: treat as
# grounded (evidence exists) — the parent's compose step
# will see the pointers and may still go undetermined.
outcome = "grounded"
else:
outcome = "gap_recorded"
sub_results.append(SubQuestion(
sub_id=_sub_id(candidate.candidate_id, index, payload),
proposed_subject=str(payload.get("subject") or ""),
proposed_intent=str(payload.get("intent") or ""),
outcome=outcome,
evidence=recursed.evidence,
))
sub_tuple = tuple(sub_results)
polarity = _compose_polarity(direct_evidence, sub_tuple)
domain = _classify_claim_domain(candidate.proposed_chain)
# Composition rule from ADR-0056: mixed evidence ⇒
# ``undetermined`` AND claim_domain upgrades one tier.
if polarity == "undetermined":
all_ptrs = list(direct_evidence) + [p for sq in sub_tuple for p in sq.evidence]
has_aff = any(p.polarity == "affirms" for p in all_ptrs)
has_fal = any(p.polarity == "falsifies" for p in all_ptrs)
if has_aff and has_fal:
domain = _upgrade_domain(domain)
return replace(
candidate,
polarity=polarity,
claim_domain=domain,
evidence=direct_evidence,
sub_questions=sub_tuple,
)
__all__ = [
"contemplate",
]

View file

@ -64,15 +64,68 @@ DiscoveryTrigger = Literal[
]
# ADR-0056 Phase C1: typed claim domain for the contemplation loop.
ClaimDomain = Literal["factual", "relational", "evaluative"]
@dataclass(frozen=True, slots=True)
class EvidencePointer:
"""One unit of admissible evidence used by the contemplation loop.
Only three source families admit a pointer: reviewed teaching
corpus chains, ratified pack atoms, and vault entries stamped
``EpistemicStatus.COHERENT``. SPECULATIVE / CONTESTED / FALSIFIED
vault entries contest but do not contribute as evidence.
"""
source: Literal["corpus", "pack", "vault_coherent"]
ref: str
polarity: Literal["affirms", "falsifies"]
epistemic_status: str
def as_dict(self) -> dict[str, Any]:
return {
"source": self.source,
"ref": self.ref,
"polarity": self.polarity,
"epistemic_status": self.epistemic_status,
}
@dataclass(frozen=True, slots=True)
class SubQuestion:
"""One decomposed sub-question + its outcome (ADR-0056 §SubQuestion).
``outcome="gap_recorded"`` is the load-bearing case from Call 1
in ADR-0056: the sub-question could not be decomposed further so
the system records the gap and stops.
"""
sub_id: str
proposed_subject: str
proposed_intent: str
outcome: Literal["grounded", "gap_recorded", "depth_failsafe"]
evidence: tuple[EvidencePointer, ...] = ()
def as_dict(self) -> dict[str, Any]:
return {
"sub_id": self.sub_id,
"proposed_subject": self.proposed_subject,
"proposed_intent": self.proposed_intent,
"outcome": self.outcome,
"evidence": [e.as_dict() for e in self.evidence],
}
@dataclass(frozen=True, slots=True)
class DiscoveryCandidate:
"""Structured evidence that a reviewed chain would have helped.
``proposed_chain`` is *partial* by design: Phase B can only see
that a chain of a given ``(subject, intent)`` would have grounded
the turn it cannot infer the connective or object. Phase C's
``TeachingChainProposal`` is the place where a complete proposed
entry is constructed and gated through review + replay.
Phase B emits the Phase-B fields only. ADR-0056 Phase C1 adds
typed contemplation fields (``polarity``, ``claim_domain``,
``evidence``, ``sub_questions``, ``contemplation_depth``,
``recursion_overflow``). Defaults make a freshly-emitted Phase B
candidate a trivially-valid un-contemplated C1 candidate.
"""
candidate_id: str
@ -82,9 +135,17 @@ class DiscoveryCandidate:
pack_consistent: bool
boundary_clean: bool
review_state: Literal["unreviewed"] = "unreviewed"
# Phase C1 fields. Defaults preserve byte-equality with Phase B
# ``as_dict`` output when the candidate has not been contemplated.
polarity: Literal["affirms", "falsifies", "undetermined"] = "undetermined"
claim_domain: ClaimDomain = "factual"
evidence: tuple[EvidencePointer, ...] = ()
sub_questions: tuple[SubQuestion, ...] = ()
contemplation_depth: int = 0
recursion_overflow: bool = False
def as_dict(self) -> dict[str, Any]:
return {
out: dict[str, Any] = {
"candidate_id": self.candidate_id,
"proposed_chain": self.proposed_chain,
"trigger": self.trigger,
@ -93,6 +154,25 @@ class DiscoveryCandidate:
"boundary_clean": self.boundary_clean,
"review_state": self.review_state,
}
# Phase C1 fields are emitted only when contemplation has
# produced non-default content. This keeps a freshly-emitted
# Phase B candidate's JSONL line byte-identical to the pre-C1
# encoding.
if (
self.polarity != "undetermined"
or self.claim_domain != "factual"
or self.evidence
or self.sub_questions
or self.contemplation_depth != 0
or self.recursion_overflow
):
out["polarity"] = self.polarity
out["claim_domain"] = self.claim_domain
out["evidence"] = [e.as_dict() for e in self.evidence]
out["sub_questions"] = [s.as_dict() for s in self.sub_questions]
out["contemplation_depth"] = self.contemplation_depth
out["recursion_overflow"] = self.recursion_overflow
return out
_TEACHING_INTENT_NAME: dict[IntentTag, str] = {
@ -225,8 +305,11 @@ def format_candidate_jsonl(candidate: DiscoveryCandidate) -> str:
__all__ = [
"ClaimDomain",
"DiscoveryCandidate",
"DiscoveryTrigger",
"EvidencePointer",
"SubQuestion",
"extract_discovery_candidates",
"format_candidate_jsonl",
]

373
tests/test_contemplation.py Normal file
View file

@ -0,0 +1,373 @@
"""ADR-0056 Phase C1 — contemplation loop tests.
Verification matrix mirrors the acceptance criteria in
``docs/decisions/ADR-0056-contemplation-loop-c1.md``:
- Determinism across runs (byte-identical JSONL).
- Empty corpus + empty pack terminates with gap recorded.
- Factual candidate with one reviewed line polarity=affirms,
claim_domain=factual.
- Direct same-pack contradiction polarity=falsifies.
- Mixed evidence polarity=undetermined + claim_domain upgraded.
- Recursion overflow flips flag + emits subquestion outcome.
- No corpus mutation (byte-identical before/after).
- DiscoveryCandidate Phase B as_dict() unchanged when C1 fields
are at default.
"""
from __future__ import annotations
import hashlib
import json
from chat.teaching_grounding import _CORPUS_PATH
from teaching.contemplation import contemplate
from teaching.discovery import (
DiscoveryCandidate,
EvidencePointer,
format_candidate_jsonl,
)
CORPUS_BYTES_BEFORE = _CORPUS_PATH.read_bytes() if _CORPUS_PATH.exists() else b""
def _phase_b_candidate(
*, subject: str = "wisdom", intent: str = "cause",
candidate_id: str = "cand_abc", trace: str = "trace_xyz",
) -> DiscoveryCandidate:
return DiscoveryCandidate(
candidate_id=candidate_id,
proposed_chain={
"subject": subject,
"intent": intent,
"connective": None,
"object": None,
},
trigger="would_have_grounded",
source_turn_trace=trace,
pack_consistent=True,
boundary_clean=True,
)
# ---------------------------------------------------------------------------
# Determinism
# ---------------------------------------------------------------------------
def test_contemplate_is_deterministic_across_runs():
"""Same candidate input ⇒ byte-identical JSONL across runs."""
cand = _phase_b_candidate()
a = format_candidate_jsonl(contemplate(cand))
b = format_candidate_jsonl(contemplate(cand))
assert a == b
# Hash equality, not just string equality.
assert hashlib.sha256(a.encode()).digest() == hashlib.sha256(b.encode()).digest()
def test_contemplate_does_not_mutate_input():
cand = _phase_b_candidate()
before_chain = dict(cand.proposed_chain)
_ = contemplate(cand)
assert cand.proposed_chain == before_chain
assert cand.polarity == "undetermined"
assert cand.evidence == ()
assert cand.sub_questions == ()
def test_contemplate_does_not_mutate_corpus_on_disk():
"""Trust boundary: contemplation NEVER writes to the corpus."""
cand = _phase_b_candidate()
_ = contemplate(cand)
after = _CORPUS_PATH.read_bytes() if _CORPUS_PATH.exists() else b""
assert after == CORPUS_BYTES_BEFORE
# ---------------------------------------------------------------------------
# Empty / cold-start
# ---------------------------------------------------------------------------
def test_empty_pack_and_corpus_terminates_with_gap(monkeypatch):
"""No pack, no corpus ⇒ every probe fails, parent gap-records."""
from teaching import contemplation as contemp_mod
monkeypatch.setattr(contemp_mod, "_pack_index", lambda: {})
monkeypatch.setattr(contemp_mod, "_corpus_index", lambda: {})
cand = _phase_b_candidate()
out = contemplate(cand)
assert out.polarity == "undetermined"
assert out.evidence == ()
assert out.sub_questions # gap-recorded
assert all(sq.outcome == "gap_recorded" for sq in out.sub_questions)
assert out.recursion_overflow is False
# ---------------------------------------------------------------------------
# Factual affirming evidence
# ---------------------------------------------------------------------------
def test_factual_candidate_with_one_reviewed_line_affirms():
"""Concrete chain matching a reviewed corpus entry → affirms/factual."""
# ``light reveals truth`` is in the production corpus (ADR-0052).
cand = DiscoveryCandidate(
candidate_id="cand_factual",
proposed_chain={
"subject": "light",
"intent": "cause",
"connective": "reveals",
"object": "truth",
},
trigger="would_have_grounded",
source_turn_trace="t1",
pack_consistent=True,
boundary_clean=True,
)
out = contemplate(cand)
assert out.polarity == "affirms"
assert out.claim_domain == "factual"
assert any(
e.source == "corpus" and e.polarity == "affirms" for e in out.evidence
)
# ---------------------------------------------------------------------------
# Falsification: same-pack direct contradiction
# ---------------------------------------------------------------------------
def test_direct_same_pack_contradiction_falsifies():
"""Same subject+intent+object, different connective → falsifies."""
# Corpus has ``light reveals truth``; propose ``light obscures truth``.
cand = DiscoveryCandidate(
candidate_id="cand_contradiction",
proposed_chain={
"subject": "light",
"intent": "cause",
"connective": "obscures",
"object": "truth",
},
trigger="would_have_grounded",
source_turn_trace="t2",
pack_consistent=True,
boundary_clean=True,
)
out = contemplate(cand)
assert out.polarity == "falsifies"
assert any(
e.source == "corpus" and e.polarity == "falsifies" for e in out.evidence
)
# ---------------------------------------------------------------------------
# Mixed evidence → undetermined + claim_domain upgrade
# ---------------------------------------------------------------------------
def test_mixed_evidence_upgrades_claim_domain(monkeypatch):
"""Mixed affirm + falsify ⇒ undetermined AND domain upgrades one tier."""
from teaching import contemplation as contemp_mod
def fake_corpus_probe(subject, intent, connective, obj):
return (
EvidencePointer(
source="corpus", ref="chain_aff", polarity="affirms",
epistemic_status="coherent",
),
)
def fake_vault(subject, obj):
return (
EvidencePointer(
source="vault_coherent", ref="vault_42",
polarity="falsifies", epistemic_status="coherent",
),
)
monkeypatch.setattr(contemp_mod, "_probe_corpus_direct", fake_corpus_probe)
monkeypatch.setattr(contemp_mod, "_decompose", lambda _c: ())
cand = DiscoveryCandidate(
candidate_id="cand_mixed",
proposed_chain={
"subject": "wisdom", "intent": "cause",
"connective": "informs", "object": "judgment",
},
trigger="would_have_grounded",
source_turn_trace="t3",
pack_consistent=True,
boundary_clean=True,
)
out = contemplate(cand, vault_probe=fake_vault)
assert out.polarity == "undetermined"
# ``informs`` is in _FRAME_DEPENDENT_CONNECTIVES → start at relational.
# Mixed evidence upgrades by one tier → evaluative.
assert out.claim_domain == "evaluative"
# ---------------------------------------------------------------------------
# Recursion overflow
# ---------------------------------------------------------------------------
def test_recursion_overflow_sets_flag_and_emits_subquestion():
cand = _phase_b_candidate()
out = contemplate(cand, max_depth=0) # depth 0 ⇒ immediate failsafe
assert out.recursion_overflow is True
assert out.sub_questions
assert any(sq.outcome == "depth_failsafe" for sq in out.sub_questions)
def test_max_depth_one_terminates_without_overflow_at_root():
"""Depth 1 should let the root execute once; sub-candidates fire failsafe."""
cand = _phase_b_candidate(subject="memory", intent="verification")
out = contemplate(cand, max_depth=1)
# Root processed; sub-candidates (depth=1) hit failsafe immediately.
assert out.recursion_overflow is False
# The sub-question outcomes will reflect depth_failsafe propagation.
assert all(
sq.outcome in ("grounded", "gap_recorded", "depth_failsafe")
for sq in out.sub_questions
)
# ---------------------------------------------------------------------------
# Frame-dependent classification
# ---------------------------------------------------------------------------
def test_frame_dependent_connective_classifies_as_relational():
cand = DiscoveryCandidate(
candidate_id="cand_relational",
proposed_chain={
"subject": "wisdom", "intent": "cause",
"connective": "orders", "object": "judgment",
},
trigger="would_have_grounded",
source_turn_trace="t4",
pack_consistent=True,
boundary_clean=True,
)
out = contemplate(cand)
assert out.claim_domain == "relational"
# ---------------------------------------------------------------------------
# Phase B byte-equality preservation
# ---------------------------------------------------------------------------
def test_uncontemplated_candidate_jsonl_unchanged():
"""A Phase B candidate (defaults only) must serialise byte-identical
to its pre-C1 encoding no new keys leak into the line."""
cand = _phase_b_candidate()
line = format_candidate_jsonl(cand)
parsed = json.loads(line)
assert set(parsed.keys()) == {
"candidate_id",
"proposed_chain",
"trigger",
"source_turn_trace",
"pack_consistent",
"boundary_clean",
"review_state",
}
def test_contemplated_candidate_jsonl_carries_c1_fields():
"""An enriched candidate's JSONL line must include the C1 fields."""
cand = DiscoveryCandidate(
candidate_id="cand_enriched",
proposed_chain={
"subject": "light", "intent": "cause",
"connective": "reveals", "object": "truth",
},
trigger="would_have_grounded",
source_turn_trace="t5",
pack_consistent=True,
boundary_clean=True,
)
out = contemplate(cand)
parsed = json.loads(format_candidate_jsonl(out))
assert "polarity" in parsed
assert "claim_domain" in parsed
assert "evidence" in parsed
assert "sub_questions" in parsed
assert "contemplation_depth" in parsed
assert "recursion_overflow" in parsed
# ---------------------------------------------------------------------------
# Determinism of sub_id derivation
# ---------------------------------------------------------------------------
def test_subquestion_ids_stable_across_runs():
cand = _phase_b_candidate()
a = contemplate(cand)
b = contemplate(cand)
assert [sq.sub_id for sq in a.sub_questions] == [
sq.sub_id for sq in b.sub_questions
]
# ---------------------------------------------------------------------------
# Evidence pointer admissibility
# ---------------------------------------------------------------------------
def test_evidence_pointers_only_admit_three_sources():
"""No emitted pointer escapes the {corpus, pack, vault_coherent} set."""
cand = _phase_b_candidate(subject="memory", intent="verification")
out = contemplate(cand)
all_ptrs = list(out.evidence) + [
p for sq in out.sub_questions for p in sq.evidence
]
for p in all_ptrs:
assert p.source in ("corpus", "pack", "vault_coherent")
assert p.polarity in ("affirms", "falsifies")
# ---------------------------------------------------------------------------
# Vault probe injection
# ---------------------------------------------------------------------------
def test_vault_probe_injection_contributes_evidence():
cand = DiscoveryCandidate(
candidate_id="cand_vault",
proposed_chain={
"subject": "memory", "intent": "verification",
"connective": "requires", "object": "recall",
},
trigger="would_have_grounded",
source_turn_trace="t6",
pack_consistent=True,
boundary_clean=True,
)
def probe(subj, obj):
return (
EvidencePointer(
source="vault_coherent", ref="v_1",
polarity="affirms", epistemic_status="coherent",
),
)
out = contemplate(cand, vault_probe=probe)
assert any(e.source == "vault_coherent" for e in out.evidence)
def test_vault_probe_failure_does_not_poison_loop():
cand = _phase_b_candidate()
def bad_probe(subj, obj):
raise RuntimeError("vault unreachable")
# Loop must still terminate cleanly.
out = contemplate(cand, vault_probe=bad_probe)
assert out is not None