995f4fc3b8
- Physics dual operators set discovery_eligible (γ=0.35); is_discovery_eligible pure predicate; no teaching/vault imports in core.physics.surprise. - teaching.discovery: trigger high_surprise; candidate_from_surprise_dual + emit_surprise_discovery (opt-in sink, proposal-only unreviewed, domain=math). - Boundary tests: threshold gates, determinism, no VaultStore, no teaching import in physics. Ledger §6 notes wiring surface (issue #30). Does not self-install; contemplation consumes via existing DiscoveryCandidateSink.
507 lines
18 KiB
Python
507 lines
18 KiB
Python
"""ADR-0055 Phase B — DiscoveryCandidate emission from the turn loop.
|
||
|
||
A ``DiscoveryCandidate`` is **structured evidence** — never a
|
||
mutation. When a turn's audit trail satisfies a deterministic
|
||
predicate, an entry is emitted to the discovery candidate stream.
|
||
Candidates **never** load into the active teaching corpus; the only
|
||
path to corpus extension is the review-gated
|
||
``TeachingChainProposal`` (Phase C, not yet built).
|
||
|
||
Trigger set (Phase B lands the first; the others are stubbed in the
|
||
``Literal`` so the structure is stable when later phases add them):
|
||
|
||
- ``would_have_grounded`` — the turn fell through to the universal
|
||
"insufficient grounding" disclosure, the classified intent was
|
||
``CAUSE`` or ``VERIFICATION``, the subject lemma is in the
|
||
ratified cognition pack, and no active chain matched
|
||
``(subject, intent)``. A reviewed chain of that subject/intent
|
||
would have grounded the turn.
|
||
|
||
- ``successful_comparison`` — open question §5 in ADR-0055; not
|
||
fired in Phase B.
|
||
|
||
- ``hedge_acknowledged`` — open question §5 in ADR-0055; not
|
||
fired in Phase B.
|
||
|
||
- ``oov_resolved_via_decomp`` — not fired in Phase B.
|
||
|
||
Determinism contract:
|
||
|
||
- ``extract_discovery_candidates`` is a pure function of its
|
||
inputs.
|
||
- ``candidate_id`` is a SHA-256 hash of a canonical JSON encoding
|
||
of the candidate's load-bearing fields; identical inputs always
|
||
produce the identical id.
|
||
- No LLM, no stochastic sampling, no clock-time read.
|
||
|
||
Trust boundary:
|
||
|
||
- This module reads pack + corpus indices and a ``TurnEvent``.
|
||
It never writes to the corpus, the pack, or runtime state.
|
||
- The ``source_turn_trace`` is the upstream ``TurnEvent.trace_hash``
|
||
when present; absent that, the empty string. Tying every
|
||
candidate to a replayable turn is the load-bearing audit
|
||
property.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import json
|
||
import math
|
||
from dataclasses import dataclass
|
||
from typing import Any, Literal
|
||
|
||
from generate.intent import IntentTag
|
||
|
||
# ``chat.pack_grounding`` and ``chat.teaching_grounding`` are
|
||
# imported lazily inside ``extract_discovery_candidates`` to break a
|
||
# circular import chain when an entry-point (e.g. the CLI) imports
|
||
# ``teaching.proposals`` → ``teaching.discovery`` before ``chat``
|
||
# has been fully initialized.
|
||
|
||
|
||
DiscoveryTrigger = Literal[
|
||
"would_have_grounded",
|
||
"successful_comparison",
|
||
"hedge_acknowledged",
|
||
"oov_resolved_via_decomp",
|
||
# Third-Door Super §3.2 / fidelity #20: high geometric surprise on the
|
||
# dual Procrustes–Surprise operator (epistemic frontier signal).
|
||
"high_surprise",
|
||
]
|
||
|
||
|
||
# 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,
|
||
}
|
||
|
||
@classmethod
|
||
def from_dict(cls, payload: dict[str, Any]) -> "EvidencePointer":
|
||
return cls(
|
||
source=payload["source"],
|
||
ref=payload["ref"],
|
||
polarity=payload["polarity"],
|
||
epistemic_status=payload["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],
|
||
}
|
||
|
||
@classmethod
|
||
def from_dict(cls, payload: dict[str, Any]) -> "SubQuestion":
|
||
return cls(
|
||
sub_id=payload["sub_id"],
|
||
proposed_subject=payload["proposed_subject"],
|
||
proposed_intent=payload["proposed_intent"],
|
||
outcome=payload["outcome"],
|
||
evidence=tuple(
|
||
EvidencePointer.from_dict(e) for e in payload.get("evidence", [])
|
||
),
|
||
)
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class DiscoveryCandidate:
|
||
"""Structured evidence that a reviewed chain would have helped.
|
||
|
||
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
|
||
proposed_chain: dict[str, Any]
|
||
trigger: DiscoveryTrigger
|
||
source_turn_trace: str
|
||
pack_consistent: bool
|
||
boundary_clean: bool
|
||
review_state: Literal["unreviewed"] = "unreviewed"
|
||
domain: Literal["cognition", "math"] = "cognition"
|
||
# 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]:
|
||
out: dict[str, Any] = {
|
||
"candidate_id": self.candidate_id,
|
||
"proposed_chain": self.proposed_chain,
|
||
"trigger": self.trigger,
|
||
"source_turn_trace": self.source_turn_trace,
|
||
"pack_consistent": self.pack_consistent,
|
||
"boundary_clean": self.boundary_clean,
|
||
"review_state": self.review_state,
|
||
}
|
||
if self.domain != "cognition":
|
||
out["domain"] = self.domain
|
||
# 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
|
||
|
||
@classmethod
|
||
def from_dict(cls, payload: dict[str, Any]) -> "DiscoveryCandidate":
|
||
return cls(
|
||
candidate_id=payload["candidate_id"],
|
||
proposed_chain=payload["proposed_chain"],
|
||
trigger=payload["trigger"],
|
||
source_turn_trace=payload["source_turn_trace"],
|
||
pack_consistent=payload["pack_consistent"],
|
||
boundary_clean=payload["boundary_clean"],
|
||
review_state=payload.get("review_state", "unreviewed"),
|
||
domain=payload.get("domain", "cognition"),
|
||
polarity=payload.get("polarity", "undetermined"),
|
||
claim_domain=payload.get("claim_domain", "factual"),
|
||
evidence=tuple(
|
||
EvidencePointer.from_dict(e) for e in payload.get("evidence", [])
|
||
),
|
||
sub_questions=tuple(
|
||
SubQuestion.from_dict(s) for s in payload.get("sub_questions", [])
|
||
),
|
||
contemplation_depth=payload.get("contemplation_depth", 0),
|
||
recursion_overflow=payload.get("recursion_overflow", False),
|
||
)
|
||
|
||
|
||
_TEACHING_INTENT_NAME: dict[IntentTag, str] = {
|
||
IntentTag.CAUSE: "cause",
|
||
IntentTag.VERIFICATION: "verification",
|
||
}
|
||
|
||
|
||
def _hash_candidate_id(payload: dict[str, Any]) -> str:
|
||
"""Deterministic SHA-256 over a canonical JSON encoding.
|
||
|
||
Sorted keys + tight separators keep the hash stable across
|
||
Python runtimes and dict-insertion order. This is the
|
||
``candidate_id`` — used both as the on-disk JSONL line key and
|
||
by Phase C to look up the originating candidate.
|
||
"""
|
||
blob = json.dumps(payload, sort_keys=True, separators=(",", ":"))
|
||
return hashlib.sha256(blob.encode("utf-8")).hexdigest()
|
||
|
||
|
||
def _boundary_clean(turn_event: Any) -> bool:
|
||
"""Return True iff the source turn produced no safety/ethics
|
||
refusal and no hedge injection.
|
||
|
||
Tolerates events that pre-date the bundled-verdicts era (ADR-0039
|
||
onward) by reading the canonical fields directly.
|
||
"""
|
||
refusal_emitted = bool(getattr(turn_event, "refusal_emitted", False) or False)
|
||
hedge_injected = bool(getattr(turn_event, "hedge_injected", False) or False)
|
||
if refusal_emitted or hedge_injected:
|
||
return False
|
||
verdicts = getattr(turn_event, "verdicts", None)
|
||
if verdicts is not None:
|
||
if bool(getattr(verdicts, "refusal_emitted", False) or False):
|
||
return False
|
||
if bool(getattr(verdicts, "hedge_injected", False) or False):
|
||
return False
|
||
return True
|
||
|
||
|
||
def _trace_hash(turn_event: Any) -> str:
|
||
value = getattr(turn_event, "trace_hash", "") or ""
|
||
return str(value)
|
||
|
||
|
||
def extract_discovery_candidates(
|
||
turn_event: Any,
|
||
intent_tag: IntentTag | None,
|
||
intent_subject_lemma: str | None,
|
||
*,
|
||
grounding_source: str | None = None,
|
||
) -> tuple[DiscoveryCandidate, ...]:
|
||
"""Return zero or more DiscoveryCandidates for a single turn.
|
||
|
||
Phase B only fires the ``would_have_grounded`` trigger. All
|
||
other triggers in the ``DiscoveryTrigger`` Literal are reserved
|
||
for later phases.
|
||
|
||
Fires when **every** condition holds (deterministic predicate):
|
||
|
||
1. ``grounding_source`` is ``"none"`` or absent — the turn
|
||
fell through to the universal disclosure.
|
||
2. ``intent_tag`` is ``CAUSE`` or ``VERIFICATION`` — the
|
||
intent set the teaching-grounded surface answers.
|
||
3. ``intent_subject_lemma`` is a non-empty pack lemma in the
|
||
ratified cognition pack.
|
||
4. ``(subject_lemma, intent_name)`` is **not** in the active
|
||
corpus — a chain of that shape would have grounded the
|
||
turn but does not exist.
|
||
|
||
Order of conditions matters for tests: short-circuit on the
|
||
cheapest predicate first.
|
||
"""
|
||
source = (grounding_source or getattr(turn_event, "grounding_source", "none") or "none").lower()
|
||
if source != "none":
|
||
return ()
|
||
if intent_tag is None or intent_tag not in _TEACHING_INTENT_NAME:
|
||
return ()
|
||
if not intent_subject_lemma or not isinstance(intent_subject_lemma, str):
|
||
return ()
|
||
lemma = intent_subject_lemma.strip().lower()
|
||
if not lemma:
|
||
return ()
|
||
|
||
from chat.pack_resolver import is_resolvable
|
||
from chat.teaching_grounding import _all_chains_index
|
||
|
||
# ADR-0064 — discovery gate uses cross-pack residency (any mounted
|
||
# lexicon pack) AND cross-corpus chain lookup (any registered
|
||
# teaching corpus). A kinship CAUSE prompt whose subject is in
|
||
# the relations pack but has no relations-chain in the active
|
||
# corpus is now also a discovery signal.
|
||
if not is_resolvable(lemma):
|
||
return ()
|
||
|
||
intent_name = _TEACHING_INTENT_NAME[intent_tag]
|
||
if (lemma, intent_name) in _all_chains_index():
|
||
return ()
|
||
|
||
# The candidate's proposed_chain is intentionally partial: Phase B
|
||
# can only assert that a chain of this (subject, intent) would
|
||
# have helped. Connective and object remain null; Phase C is
|
||
# where a complete proposed entry is constructed and review-gated.
|
||
proposed_chain = {
|
||
"subject": lemma,
|
||
"intent": intent_name,
|
||
"connective": None,
|
||
"object": None,
|
||
}
|
||
|
||
trace_hash = _trace_hash(turn_event)
|
||
boundary_clean = _boundary_clean(turn_event)
|
||
trigger: DiscoveryTrigger = "would_have_grounded"
|
||
|
||
hash_payload = {
|
||
"proposed_chain": proposed_chain,
|
||
"trigger": trigger,
|
||
"source_turn_trace": trace_hash,
|
||
}
|
||
candidate_id = _hash_candidate_id(hash_payload)
|
||
|
||
candidate = DiscoveryCandidate(
|
||
candidate_id=candidate_id,
|
||
proposed_chain=proposed_chain,
|
||
trigger=trigger,
|
||
source_turn_trace=trace_hash,
|
||
pack_consistent=True, # subject is in pack; object is null pending Phase C
|
||
boundary_clean=boundary_clean,
|
||
review_state="unreviewed",
|
||
)
|
||
return (candidate,)
|
||
|
||
|
||
def format_candidate_jsonl(candidate: DiscoveryCandidate) -> str:
|
||
"""Serialise to one JSONL line (sorted keys, no trailing newline)."""
|
||
return json.dumps(candidate.as_dict(), sort_keys=True, separators=(",", ":"))
|
||
|
||
|
||
def candidate_from_surprise_dual(
|
||
dual: dict[str, Any],
|
||
*,
|
||
source_turn_trace: str = "",
|
||
discovery_gamma: float | None = None,
|
||
) -> DiscoveryCandidate | None:
|
||
"""Build a proposal-only ``DiscoveryCandidate`` from a dual-operator audit dict.
|
||
|
||
Pure function of its inputs (plus the dual dict). Fires only when the dual
|
||
result is discovery-eligible: measured surprise above γ, not a productive
|
||
transfer, and no metric refusal. Never writes vault / packs / corpus.
|
||
|
||
``dual`` is the audit dict from
|
||
:func:`core.physics.surprise.dual_operator` or
|
||
:func:`core.physics.surprise.dual_procrustes_surprise` (must carry
|
||
``surprise_norm``; preferably also ``discovery_eligible``).
|
||
|
||
Returns ``None`` when the signal is not a high-surprise discovery.
|
||
"""
|
||
from core.physics.surprise import (
|
||
DEFAULT_DISCOVERY_GAMMA,
|
||
is_discovery_eligible,
|
||
)
|
||
|
||
if not isinstance(dual, dict):
|
||
return None
|
||
|
||
gamma = float(
|
||
discovery_gamma
|
||
if discovery_gamma is not None
|
||
else dual.get("discovery_gamma", DEFAULT_DISCOVERY_GAMMA)
|
||
)
|
||
sur_norm = float(dual.get("surprise_norm", float("nan")))
|
||
refused = dual.get("surprise_refused")
|
||
productive = bool(
|
||
dual.get("productive", False) or dual.get("transfer_accepted", False)
|
||
)
|
||
|
||
# Prefer the physics flag when present; recompute otherwise so older dual
|
||
# dicts without the field still route correctly.
|
||
if "discovery_eligible" in dual:
|
||
eligible = bool(dual["discovery_eligible"])
|
||
else:
|
||
eligible = is_discovery_eligible(
|
||
surprise_norm=sur_norm,
|
||
productive_or_transfer=productive,
|
||
surprise_refused=refused if isinstance(refused, str) else None,
|
||
discovery_gamma=gamma,
|
||
)
|
||
if not eligible:
|
||
return None
|
||
|
||
proc_r = dual.get("procrustes_residual")
|
||
try:
|
||
proc_f = float(proc_r) if proc_r is not None and math.isfinite(float(proc_r)) else None
|
||
except (TypeError, ValueError):
|
||
proc_f = None
|
||
analog_id = dual.get("selected_analog_id")
|
||
|
||
# Partial proposed chain — geometric frontier evidence, not a ready teaching
|
||
# chain. Review / Phase C remains the only path to corpus extension.
|
||
proposed_chain: dict[str, Any] = {
|
||
"subject": "geometric_frontier",
|
||
"intent": "discovery",
|
||
"connective": None,
|
||
"object": None,
|
||
"kind": "high_surprise",
|
||
"surprise_norm": sur_norm,
|
||
"discovery_gamma": gamma,
|
||
}
|
||
if proc_f is not None:
|
||
proposed_chain["procrustes_residual"] = proc_f
|
||
if analog_id is not None:
|
||
proposed_chain["selected_analog_id"] = str(analog_id)
|
||
|
||
trace = str(source_turn_trace or "")
|
||
trigger: DiscoveryTrigger = "high_surprise"
|
||
hash_payload = {
|
||
"proposed_chain": proposed_chain,
|
||
"trigger": trigger,
|
||
"source_turn_trace": trace,
|
||
}
|
||
candidate_id = _hash_candidate_id(hash_payload)
|
||
return DiscoveryCandidate(
|
||
candidate_id=candidate_id,
|
||
proposed_chain=proposed_chain,
|
||
trigger=trigger,
|
||
source_turn_trace=trace,
|
||
pack_consistent=True, # geometric signal; pack residency N/A
|
||
boundary_clean=True,
|
||
review_state="unreviewed",
|
||
domain="math",
|
||
)
|
||
|
||
|
||
def emit_surprise_discovery(
|
||
dual: dict[str, Any],
|
||
sink: Any | None = None,
|
||
*,
|
||
source_turn_trace: str = "",
|
||
discovery_gamma: float | None = None,
|
||
) -> DiscoveryCandidate | None:
|
||
"""Opt-in sink emission of a high-surprise ``DiscoveryCandidate``.
|
||
|
||
Pure construction via :func:`candidate_from_surprise_dual`. When ``sink`` is
|
||
provided and a candidate is produced, emits one JSONL line through the
|
||
existing :class:`~teaching.discovery_sink.DiscoveryCandidateSink` protocol
|
||
(same stream contemplation already consumes). Without a sink this is a pure
|
||
factory. Never mutates vault, packs, or the teaching corpus.
|
||
"""
|
||
candidate = candidate_from_surprise_dual(
|
||
dual,
|
||
source_turn_trace=source_turn_trace,
|
||
discovery_gamma=discovery_gamma,
|
||
)
|
||
if candidate is None:
|
||
return None
|
||
if sink is not None:
|
||
emit = getattr(sink, "emit", None)
|
||
if emit is None or not callable(emit):
|
||
raise TypeError(
|
||
"emit_surprise_discovery: sink must provide emit(line: str) "
|
||
"(DiscoveryCandidateSink protocol)"
|
||
)
|
||
emit(format_candidate_jsonl(candidate))
|
||
return candidate
|
||
|
||
|
||
__all__ = [
|
||
"ClaimDomain",
|
||
"DiscoveryCandidate",
|
||
"DiscoveryTrigger",
|
||
"EvidencePointer",
|
||
"SubQuestion",
|
||
"candidate_from_surprise_dual",
|
||
"emit_surprise_discovery",
|
||
"extract_discovery_candidates",
|
||
"format_candidate_jsonl",
|
||
]
|