From 4eecf73a0569ebcb50bc878d6e5b72aea97e3fbf Mon Sep 17 00:00:00 2001 From: Shay Date: Mon, 18 May 2026 10:06:18 -0700 Subject: [PATCH] =?UTF-8?q?feat(adr-0056):=20Phase=20C1=20=E2=80=94=20cont?= =?UTF-8?q?emplation=20loop=20landed?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- teaching/contemplation.py | 505 ++++++++++++++++++++++++++++++++++++ teaching/discovery.py | 95 ++++++- tests/test_contemplation.py | 373 ++++++++++++++++++++++++++ 3 files changed, 967 insertions(+), 6 deletions(-) create mode 100644 teaching/contemplation.py create mode 100644 tests/test_contemplation.py diff --git a/teaching/contemplation.py b/teaching/contemplation.py new file mode 100644 index 00000000..6d8af375 --- /dev/null +++ b/teaching/contemplation.py @@ -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", +] diff --git a/teaching/discovery.py b/teaching/discovery.py index d3b633a4..c111a67a 100644 --- a/teaching/discovery.py +++ b/teaching/discovery.py @@ -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", ] diff --git a/tests/test_contemplation.py b/tests/test_contemplation.py new file mode 100644 index 00000000..95d8e4ae --- /dev/null +++ b/tests/test_contemplation.py @@ -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