diff --git a/core_ingest/__init__.py b/core_ingest/__init__.py new file mode 100644 index 00000000..f01fe659 --- /dev/null +++ b/core_ingest/__init__.py @@ -0,0 +1,58 @@ +""" +core_ingest — Universal input-to-pressure boundary. + +The sole job of this package is converting external information from any +source into CORE-native pressure: typed, content-addressed, provenance- +anchored candidate packets that can be validated and, if accepted, exported +as LearningArtifact objects for the train/ layer. + +Two paths: + runtime — transient working pressure for active cognition + (short-circuit: bypasses governance, feeds ingest/gate.py directly) + durable — validated candidate pressure for learning and update paths + (full three-gate IngestCompiler pipeline) + +The gate (ingest/gate.py) is NEVER imported or modified from this package. +This package is upstream of the gate, not a replacement for it. +""" + +from core_ingest.types import ( + Modality, + DeterminismClass, + ReviewLevel, + SourceSpan, + FrontendTrace, + CandidateGeometricPressure, + ValidationResult, + ValidationReport, + LearningArtifact, + ReviewDecision, +) +from core_ingest.pressure import make_pressure_id, make_semantic_key +from core_ingest.compiler import IngestCompiler +from core_ingest.segmenter import StructuralSegmenter, SegmentKind +from core_ingest.manifold import SegmentManifold + +__all__ = [ + # types + "Modality", + "DeterminismClass", + "ReviewLevel", + "SourceSpan", + "FrontendTrace", + "CandidateGeometricPressure", + "ValidationResult", + "ValidationReport", + "LearningArtifact", + "ReviewDecision", + # pressure addressing + "make_pressure_id", + "make_semantic_key", + # compiler + "IngestCompiler", + # segmenter + "StructuralSegmenter", + "SegmentKind", + # manifold index + "SegmentManifold", +] diff --git a/core_ingest/compiler.py b/core_ingest/compiler.py new file mode 100644 index 00000000..295f614f --- /dev/null +++ b/core_ingest/compiler.py @@ -0,0 +1,303 @@ +""" +IngestCompiler — deterministic three-gate validation pipeline. + +Gate order (sequential; a packet failing any gate does not proceed): + 1. ProvenanceGate — SourceSpan integrity: non-empty text, valid SHA-256, + byte offsets non-negative and ordered + 2. SemanticGate — span completeness: non-empty payload, balanced + delimiters (for code/math), minimum length + 3. GovernanceGate — ReviewLevel vs DeterminismClass consistency; + ReviewDecision override check + +The compiler: + - performs structural deduplication by pressure_id (exact structural identity) + - tracks semantic convergence by semantic_key and annotates packets that + assert the same claim as prior packets with a warning + - applies ReviewDecision overrides without mutating the original packet + - produces a ValidationReport alongside the original immutable packets + - exports accepted packets as LearningArtifact objects + +Ingest/gate.py is NOT imported or called here. +""" + +from __future__ import annotations + +from collections import defaultdict +from typing import Sequence + +from core_ingest.types import ( + CandidateGeometricPressure, + DeterminismClass, + GateDisposition, + LearningArtifact, + ReviewDecision, + ReviewLevel, + ValidationReport, + ValidationResult, +) + + +# --------------------------------------------------------------------------- +# Individual gates +# --------------------------------------------------------------------------- + +class ProvenanceGate: + """ + Gate 1: SourceSpan integrity. + + Checks: + - At least one SourceSpan (enforced at packet construction; double-checked here) + - source_sha256 is 64 hex characters + - byte_start >= 0 + - byte_end > byte_start + """ + + def check(self, packet: CandidateGeometricPressure) -> str | None: + """Return a failure reason string, or None if the packet passes.""" + if not packet.provenance: + return "provenance is empty" + for span in packet.provenance: + if span.byte_start < 0: + return f"negative byte_start ({span.byte_start}) in SourceSpan" + if span.byte_end <= span.byte_start: + return ( + f"byte_end ({span.byte_end}) <= byte_start ({span.byte_start})" + ) + if len(span.source_sha256) != 64: + return ( + f"source_sha256 is {len(span.source_sha256)} chars, expected 64" + ) + return None + + +class SemanticGate: + """ + Gate 2: Span completeness. + + Checks: + - payload_json is non-empty (not just '{}') + - lemma is non-empty for TEXT and SCRIPTURE modalities + - For CODE modality: no unbalanced braces/brackets/parens in lemma + - Minimum combined span length of 1 byte + """ + _PAIRS = {"{": "}", "[": "]", "(": ")", "<": ">"} + + def check(self, packet: CandidateGeometricPressure) -> str | None: + import json + from core_ingest.types import Modality + + if packet.payload_json in ("{}", ""): + return "payload_json is empty — no content to ingest" + + # Require non-empty lemma for text / scripture + if packet.modality in (Modality.TEXT, Modality.SCRIPTURE) and not packet.lemma: + return ( + f"modality {packet.modality.value} requires a non-empty lemma" + ) + + # CODE: balanced delimiter check on lemma + if packet.modality == Modality.CODE and packet.lemma: + stack: list[str] = [] + for ch in packet.lemma: + if ch in self._PAIRS: + stack.append(self._PAIRS[ch]) + elif ch in self._PAIRS.values(): + if not stack or stack[-1] != ch: + return f"unbalanced delimiter '{ch}' in code lemma" + stack.pop() + if stack: + return f"unclosed delimiter(s) in code lemma: {''.join(stack)}" + + # Minimum span length + total_bytes = sum(s.byte_end - s.byte_start for s in packet.provenance) + if total_bytes < 1: + return "total span length is 0 bytes" + + return None + + +class GovernanceGate: + """ + Gate 3: ReviewLevel and DeterminismClass consistency. + + Checks: + - AUTO_ACCEPT_ELIGIBLE is only claimed by D0/D1 frontends + (enforced at packet construction; double-checked here) + - AUTO_REJECT packets are always rejected + - D2–D4 packets not covered by a ReviewDecision land in review_required + """ + + def check( + self, + packet: CandidateGeometricPressure, + authorized_ids: frozenset[str], + ) -> GateDisposition: + if packet.review_level == ReviewLevel.AUTO_REJECT: + return GateDisposition.REJECTED_GOVERNANCE + + # Override: an authorized ReviewDecision accepts regardless of level + if packet.pressure_id in authorized_ids: + return GateDisposition.OVERRIDE_ACCEPTED + + if packet.review_level == ReviewLevel.AUTO_ACCEPT_ELIGIBLE: + # Invariant already enforced at construction — D0/D1 only + return GateDisposition.ACCEPTED + + # OPERATOR_REVIEW_REQUIRED or ARCHITECT_REVIEW_REQUIRED without override + return GateDisposition.REVIEW_REQUIRED + + +# --------------------------------------------------------------------------- +# Compiler +# --------------------------------------------------------------------------- + +class IngestCompiler: + """ + Processes batches of CandidateGeometricPressure packets through the + three-gate pipeline and produces a ValidationReport. + + Usage + ----- + compiler = IngestCompiler() + report, artifacts = compiler.compile(packets, review_decision=decision) + """ + + def __init__(self) -> None: + self._provenance_gate = ProvenanceGate() + self._semantic_gate = SemanticGate() + self._governance_gate = GovernanceGate() + + def compile( + self, + packets: Sequence[CandidateGeometricPressure], + review_decision: ReviewDecision | None = None, + ) -> tuple[ValidationReport, list[LearningArtifact]]: + """ + Validate a batch of candidate packets. + + Parameters + ---------- + packets : The batch to validate (any sequence). + review_decision : Optional operator/architect authorization. Packets + whose pressure_id appears in + review_decision.authorized_ids are accepted + regardless of their ReviewLevel. + + Returns + ------- + (ValidationReport, list[LearningArtifact]) + """ + authorized_ids: frozenset[str] = ( + review_decision.authorized_ids + if review_decision is not None + else frozenset() + ) + + # Deduplication: track seen pressure_ids (structural) and semantic_keys + seen_pressure_ids: set[str] = set() + semantic_counts: dict[str, int] = defaultdict(int) + + results: list[ValidationResult] = [] + artifacts: list[LearningArtifact] = [] + + accepted_ids: set[str] = set() + rejected_ids: set[str] = set() + review_ids: set[str] = set() + + for packet in packets: + pid = packet.pressure_id + sk = packet.semantic_key + + # Structural deduplication + if pid in seen_pressure_ids: + results.append(ValidationResult( + pressure_id=pid, + semantic_key=sk, + disposition=GateDisposition.REJECTED_PROVENANCE, + gate_failed="provenance", + failure_reason="duplicate pressure_id — structural duplicate", + )) + rejected_ids.add(pid) + continue + seen_pressure_ids.add(pid) + + # Convergent-evidence tracking + prior_count = semantic_counts[sk] + semantic_counts[sk] += 1 + convergence_warning: tuple[str, ...] = () + if prior_count > 0: + convergence_warning = ( + f"semantic_convergence:{prior_count}_prior_sources", + ) + + # Gate 1: Provenance + prov_failure = self._provenance_gate.check(packet) + if prov_failure is not None: + results.append(ValidationResult( + pressure_id=pid, + semantic_key=sk, + disposition=GateDisposition.REJECTED_PROVENANCE, + gate_failed="provenance", + failure_reason=prov_failure, + warnings=convergence_warning, + )) + rejected_ids.add(pid) + continue + + # Gate 2: Semantic + sem_failure = self._semantic_gate.check(packet) + if sem_failure is not None: + results.append(ValidationResult( + pressure_id=pid, + semantic_key=sk, + disposition=GateDisposition.REJECTED_SEMANTIC, + gate_failed="semantic", + failure_reason=sem_failure, + warnings=convergence_warning, + )) + rejected_ids.add(pid) + continue + + # Gate 3: Governance + disposition = self._governance_gate.check(packet, authorized_ids) + + if disposition == GateDisposition.REVIEW_REQUIRED: + results.append(ValidationResult( + pressure_id=pid, + semantic_key=sk, + disposition=disposition, + warnings=convergence_warning, + )) + review_ids.add(pid) + continue + + if disposition == GateDisposition.REJECTED_GOVERNANCE: + results.append(ValidationResult( + pressure_id=pid, + semantic_key=sk, + disposition=disposition, + gate_failed="governance", + failure_reason="AUTO_REJECT review level", + warnings=convergence_warning, + )) + rejected_ids.add(pid) + continue + + # ACCEPTED or OVERRIDE_ACCEPTED + result = ValidationResult( + pressure_id=pid, + semantic_key=sk, + disposition=disposition, + warnings=convergence_warning, + ) + results.append(result) + accepted_ids.add(pid) + artifacts.append(LearningArtifact(packet=packet, result=result)) + + report = ValidationReport( + results=tuple(results), + accepted_ids=frozenset(accepted_ids), + rejected_ids=frozenset(rejected_ids), + review_ids=frozenset(review_ids), + ) + return report, artifacts diff --git a/core_ingest/manifold.py b/core_ingest/manifold.py new file mode 100644 index 00000000..064a3339 --- /dev/null +++ b/core_ingest/manifold.py @@ -0,0 +1,98 @@ +""" +SegmentManifold — reconstruction index for pre-injection provenance. + +Maps semantic_key -> list of SourceSpan positions in source documents. + +Purpose: given a vault recall hit on a semantic_key, recover the exact +provenance spans in the original source documents. This extends +Reconstruction-over-Storage to the pre-injection layer — we do not store +full document copies; we store enough structured provenance to reconstruct +the relevant span on demand. + +The manifold is append-only. Entries are never modified or deleted. +""" + +from __future__ import annotations + +from collections import defaultdict +from dataclasses import dataclass +from typing import Sequence + +from core_ingest.types import CandidateGeometricPressure, SourceSpan + + +@dataclass(frozen=True, slots=True) +class ManifoldEntry: + """ + A single provenance record in the SegmentManifold. + + semantic_key — SHA-256 over semantic fields; shared by packets asserting + the same claim + pressure_id — SHA-256 over the full packet; unique per structural variant + spans — provenance spans from the packet + instrument_id — identity of the proposing instrument + """ + semantic_key: str + pressure_id: str + spans: tuple[SourceSpan, ...] + instrument_id: str + + +class SegmentManifold: + """ + Append-only index: semantic_key -> list[ManifoldEntry]. + + Usage + ----- + manifold = SegmentManifold() + manifold.register(packets) # register a batch after compilation + entries = manifold.lookup(sk) # retrieve all entries for a semantic_key + spans = manifold.spans_for(sk) # retrieve all SourceSpans for a key + """ + + def __init__(self) -> None: + self._index: dict[str, list[ManifoldEntry]] = defaultdict(list) + + def register( + self, + packets: Sequence[CandidateGeometricPressure], + ) -> None: + """ + Register a batch of packets into the manifold. + + Typically called with the accepted packets after IngestCompiler.compile(). + Can also be called with the full batch to index all attempted ingest; + the caller decides the registration policy. + """ + for packet in packets: + entry = ManifoldEntry( + semantic_key=packet.semantic_key, + pressure_id=packet.pressure_id, + spans=packet.provenance, + instrument_id=packet.frontend.instrument_id, + ) + self._index[packet.semantic_key].append(entry) + + def lookup(self, semantic_key: str) -> list[ManifoldEntry]: + """ + Return all ManifoldEntry records for a given semantic_key. + Returns an empty list if the key is not indexed. + """ + return list(self._index.get(semantic_key, [])) + + def spans_for(self, semantic_key: str) -> list[SourceSpan]: + """ + Return all SourceSpan records for a given semantic_key, + flattened across all ManifoldEntry records. + """ + spans: list[SourceSpan] = [] + for entry in self._index.get(semantic_key, []): + spans.extend(entry.spans) + return spans + + def __len__(self) -> int: + """Return the number of distinct semantic_keys indexed.""" + return len(self._index) + + def __contains__(self, semantic_key: str) -> bool: + return semantic_key in self._index diff --git a/core_ingest/pressure.py b/core_ingest/pressure.py new file mode 100644 index 00000000..2125ddc8 --- /dev/null +++ b/core_ingest/pressure.py @@ -0,0 +1,115 @@ +""" +Content-addressed identifier computation for CandidateGeometricPressure. + +Two functions: + make_pressure_id — SHA-256 over the full canonical packet (structural identity) + make_semantic_key — SHA-256 over semantic fields only (claim-level identity) + +This module transparently attempts to import core_ingest_rs (the PyO3/Rust +backend) for these operations. If the Rust extension is not built, pure-Python +fallbacks handle every case with identical behavior. + +The Rust path is a compilation target chosen after the data model was locked — +Axiom 6, Compilation-Last. +""" + +from __future__ import annotations + +import hashlib +import json +from typing import Any + +try: + from core_ingest_rs import sha256_hex, canonical_json # type: ignore[import] + _RUST = True +except ImportError: + _RUST = False + + +# --------------------------------------------------------------------------- +# Pure-Python fallbacks +# --------------------------------------------------------------------------- + +def _sha256_hex(data: bytes) -> str: + return hashlib.sha256(data).hexdigest() + + +def _canonical_json(obj: Any) -> str: + return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False) + + +# --------------------------------------------------------------------------- +# Public API +# --------------------------------------------------------------------------- + +def make_pressure_id( + *, + kind: str, + modality: str, + provenance: list[tuple[int, int, str]], # [(byte_start, byte_end, sha256), ...] + frontend_id: str, + determinism: str, + review_level: str, + confidence: float, + uncertainty: float, + lemma: str, + subject: str, + verb: str, + object_: str, + payload_json: str, +) -> str: + """ + SHA-256 over all canonical packet fields. + + Two packets are structurally identical (same pressure_id) iff every field + is identical, including provenance. Two packets asserting the same claim + from different sources will have different pressure_ids but the same + semantic_key. + """ + payload = _canonical_json({ + "kind": kind, + "modality": modality, + "provenance": sorted(provenance), + "frontend_id": frontend_id, + "determinism": determinism, + "review_level": review_level, + "confidence": confidence, + "uncertainty": uncertainty, + "lemma": lemma, + "subject": subject, + "verb": verb, + "object": object_, + "payload_json": payload_json, + }) + fn = sha256_hex if _RUST else _sha256_hex + return fn(payload.encode("utf-8")) + + +def make_semantic_key( + *, + kind: str, + modality: str, + lemma: str, + subject: str, + verb: str, + object_: str, + payload_json: str, +) -> str: + """ + SHA-256 over semantic fields only. + + Two packets with the same semantic_key assert the same claim regardless + of where or by whom they were proposed. The IngestCompiler uses this for + convergent-evidence detection. + """ + payload = _canonical_json({ + "kind": kind, + "modality": modality, + "lemma": lemma, + "subject": subject, + "verb": verb, + "object": object_, + "payload_json": payload_json, + }) + fn = sha256_hex if _RUST else _sha256_hex + return fn(payload.encode("utf-8")) diff --git a/core_ingest/segmenter.py b/core_ingest/segmenter.py new file mode 100644 index 00000000..d5d6db34 --- /dev/null +++ b/core_ingest/segmenter.py @@ -0,0 +1,228 @@ +""" +StructuralSegmenter — D0 form-boundary segmentation. + +Carves source documents at deterministic structural signals: + prose — blank-line paragraph breaks, ATX/Setext headings + scripture — verse markers (canonical chapter:verse format) + code — fenced code blocks (``` or ~~~) + math — LaTeX display environments (\\[...\\], $$...$$, \\begin{...}) + +The segmenter operates on the *form* of the source, not its content. +It does NOT interpret meaning. Meaning stays inside the versor field. + +All segmenters produce SourceSpan objects with a determinism class of D0: +fully deterministic given the same source bytes, no external state. +""" + +from __future__ import annotations + +import hashlib +import re +from dataclasses import dataclass +from enum import Enum, auto +from typing import Iterator + +from core_ingest.types import SourceSpan + + +class SegmentKind(str, Enum): + """Structural role of a segment within its source document.""" + HEADING = "heading" + BODY = "body" + VERSE = "verse" + CODE = "code" + MATH = "math" + UNKNOWN = "unknown" + + +@dataclass(frozen=True, slots=True) +class Segment: + """ + A single structural unit extracted from a source document. + + span — SourceSpan with byte offsets and source SHA-256 + kind — structural role of this segment + text — decoded surface text of the segment + """ + span: SourceSpan + kind: SegmentKind + text: str + + +class StructuralSegmenter: + """ + Entry point for D0 structural segmentation. + + Usage + ----- + segmenter = StructuralSegmenter() + for segment in segmenter.segment(source_bytes, modality_hint="prose"): + ... + + modality_hint selects the sub-segmenter: + "prose" — blank-line paragraph + heading detection + "scripture" — verse-marker based (chapter:verse or book.chapter.verse) + "code" — fenced code block extraction + "math" — LaTeX display math extraction + """ + + _SEGMENTERS = { + "prose": "_segment_prose", + "scripture": "_segment_scripture", + "code": "_segment_code", + "math": "_segment_math", + } + + def segment( + self, + source: bytes, + modality_hint: str = "prose", + ) -> list[Segment]: + """ + Segment `source` according to `modality_hint`. + + Parameters + ---------- + source : Raw source bytes (UTF-8 expected). + modality_hint : One of 'prose', 'scripture', 'code', 'math'. + + Returns + ------- + List of Segment objects in document order. + """ + source_sha = hashlib.sha256(source).hexdigest() + method_name = self._SEGMENTERS.get(modality_hint, "_segment_prose") + method = getattr(self, method_name) + return list(method(source, source_sha)) + + # ------------------------------------------------------------------ + # Prose segmenter + # ------------------------------------------------------------------ + + _HEADING_RE = re.compile( + rb"^(?P#{1,6})\s+(?P.+)$|^(?P.+)\n[=\-]{2,}$", + re.MULTILINE, + ) + + def _segment_prose(self, source: bytes, source_sha: str) -> Iterator[Segment]: + """ + Split on blank lines. Classify leading ATX/Setext headings. + Each non-empty block becomes one Segment. + """ + # Split on one or more blank lines + blocks = re.split(rb"\n{2,}", source) + offset = 0 + for block in blocks: + block = block.strip() + if not block: + offset += len(block) + 2 # account for separator + continue + start = source.find(block, offset) + end = start + len(block) + text = block.decode("utf-8", errors="replace") + + # Heading detection + kind = SegmentKind.BODY + if self._HEADING_RE.match(block): + kind = SegmentKind.HEADING + + yield Segment( + span=SourceSpan( + byte_start=start, + byte_end=end, + source_sha256=source_sha, + region=kind.value, + ), + kind=kind, + text=text, + ) + offset = end + + # ------------------------------------------------------------------ + # Scripture segmenter + # ------------------------------------------------------------------ + + # Matches: GEN.1.1, Gen 1:1, 1 Cor 13:4, etc. + _VERSE_RE = re.compile( + rb"(?:(?:[1-3]\s)?[A-Za-z]+\.?\s*\d{1,3}[:.\s]\d{1,3})", + re.MULTILINE, + ) + + def _segment_scripture( + self, source: bytes, source_sha: str + ) -> Iterator[Segment]: + """ + Split scripture source at verse boundaries. + Each verse becomes one Segment with region='verse:'. + """ + matches = list(self._VERSE_RE.finditer(source)) + for i, match in enumerate(matches): + start = match.start() + end = matches[i + 1].start() if i + 1 < len(matches) else len(source) + block = source[start:end].strip() + if not block: + continue + ref = match.group(0).decode("utf-8", errors="replace").strip() + yield Segment( + span=SourceSpan( + byte_start=start, + byte_end=start + len(block), + source_sha256=source_sha, + region=f"verse:{ref}", + ), + kind=SegmentKind.VERSE, + text=block.decode("utf-8", errors="replace"), + ) + + # ------------------------------------------------------------------ + # Code segmenter + # ------------------------------------------------------------------ + + _FENCE_RE = re.compile( + rb"^(?P`{3,}|~{3,})(?P[^\n]*)\n(?P.*?)^(?P=fence)\s*$", + re.MULTILINE | re.DOTALL, + ) + + def _segment_code(self, source: bytes, source_sha: str) -> Iterator[Segment]: + """Extract fenced code blocks (``` or ~~~).""" + for match in self._FENCE_RE.finditer(source): + start = match.start() + end = match.end() + body = match.group("body") + lang = match.group("lang").decode("utf-8", errors="replace").strip() + region = f"code_block:{lang}" if lang else "code_block" + yield Segment( + span=SourceSpan( + byte_start=start, + byte_end=end, + source_sha256=source_sha, + region=region, + ), + kind=SegmentKind.CODE, + text=body.decode("utf-8", errors="replace"), + ) + + # ------------------------------------------------------------------ + # Math segmenter + # ------------------------------------------------------------------ + + _MATH_RE = re.compile( + rb"\\\[.*?\\\]|\$\$.*?\$\$|\\begin\{[^}]+\}.*?\\end\{[^}]+\}", + re.DOTALL, + ) + + def _segment_math(self, source: bytes, source_sha: str) -> Iterator[Segment]: + """Extract LaTeX display math environments.""" + for match in self._MATH_RE.finditer(source): + start = match.start() + end = match.end() + yield Segment( + span=SourceSpan( + byte_start=start, + byte_end=end, + source_sha256=source_sha, + region="math_env", + ), + kind=SegmentKind.MATH, + text=match.group(0).decode("utf-8", errors="replace"), + ) diff --git a/core_ingest/types.py b/core_ingest/types.py new file mode 100644 index 00000000..8deff5bb --- /dev/null +++ b/core_ingest/types.py @@ -0,0 +1,311 @@ +""" +Canonical type definitions for the core_ingest governance layer. + +All dataclasses are frozen and immutable. Packets are proposed, validated, +and exported — never modified in place. +""" + +from __future__ import annotations + +import json +from dataclasses import dataclass, field +from enum import Enum, auto +from typing import Any + +from core_ingest.pressure import make_pressure_id, make_semantic_key + + +# --------------------------------------------------------------------------- +# Enumerations +# --------------------------------------------------------------------------- + +class Modality(str, Enum): + """Source medium of the incoming signal.""" + TEXT = "text" + CODE = "code" + MATH = "math" + SCRIPTURE = "scripture" # Hebrew / Koine Greek canonical texts + VISION = "vision" + AUDIO = "audio" + MOTOR = "motor" + + +class DeterminismClass(str, Enum): + """ + Reliability class of the proposing instrument. + + D0 Fully deterministic: pinned inputs, pinned code, identical output + on every run. Auto-accept eligible. + D1 Deterministic with a pinned external model artifact (e.g. a frozen + embedding table at a fixed SHA). Auto-accept eligible. + D2 Nondeterministic but replay-captured (output was logged and the log + is being replayed). Requires review. + D3 External unpinned model or API (live LLM call, external service). + Requires review. + D4 Human / operator proposal. Requires review. + """ + D0 = "D0" + D1 = "D1" + D2 = "D2" + D3 = "D3" + D4 = "D4" + + @property + def auto_accept_eligible(self) -> bool: + return self in (DeterminismClass.D0, DeterminismClass.D1) + + +class ReviewLevel(str, Enum): + AUTO_REJECT = "auto_reject" + AUTO_ACCEPT_ELIGIBLE = "auto_accept_eligible" + OPERATOR_REVIEW_REQUIRED = "operator_review_required" + ARCHITECT_REVIEW_REQUIRED = "architect_review_required" + + +# --------------------------------------------------------------------------- +# Provenance +# --------------------------------------------------------------------------- + +@dataclass(frozen=True, slots=True) +class SourceSpan: + """ + Exact location of a candidate within its source document. + + byte_start / byte_end — byte offsets in the source bytestring + page — page number (1-indexed), None for non-paginated + region — structural region label (e.g. "heading", "body", + "verse:GEN.1.1", "code_block", "math_env") + source_sha256 — SHA-256 hex digest of the full source document + at ingest time, for provenance anchoring + """ + byte_start: int + byte_end: int + source_sha256: str + page: int | None = None + region: str | None = None + + def __post_init__(self) -> None: + if self.byte_end <= self.byte_start: + raise ValueError( + f"SourceSpan byte_end ({self.byte_end}) must be > " + f"byte_start ({self.byte_start})" + ) + if len(self.source_sha256) != 64: + raise ValueError( + "source_sha256 must be a 64-character hex SHA-256 digest" + ) + + +@dataclass(frozen=True, slots=True) +class FrontendTrace: + """ + Identity and determinism class of the instrument that proposed a packet. + + instrument_id — stable, unique identifier for the proposing instrument + (e.g. "StructuralSegmenter/prose/v1", + "StructuralSegmenter/scripture-he/v1") + determinism — DeterminismClass of this instrument + version — semantic version string of the instrument + """ + instrument_id: str + determinism: DeterminismClass + version: str + + +# --------------------------------------------------------------------------- +# Candidate packet +# --------------------------------------------------------------------------- + +@dataclass(frozen=True, slots=True) +class CandidateGeometricPressure: + """ + A single proposed evidence packet at the input-to-pressure boundary. + + Every piece of incoming information — text, code, scripture, math — + is lifted into this envelope before any validation or gate interaction. + + Fields + ------ + kind — claim type label (free string, e.g. "assertion", + "definition", "verse", "theorem") + modality — source medium (Modality enum) + provenance — tuple of SourceSpan records (at least one required) + frontend — identity + determinism class of the proposing instrument + review_level — governance disposition for this packet + confidence — probability in [0.0, 1.0]: how confident the instrument is + uncertainty — probability in [0.0, 1.0]: explicit epistemic uncertainty + lemma — canonical surface form of the primary term (for semantic + key computation; empty string if not applicable) + subject — SVO subject (empty string if not applicable) + verb — SVO verb (empty string if not applicable) + object_ — SVO object (empty string if not applicable) + payload_json — structured claim content as a JSON string, normalized to + canonical (sorted-keys, no-whitespace) form on construction + + Computed in __post_init__ + ------------------------- + pressure_id — SHA-256 over the full canonical packet (structural identity) + semantic_key — SHA-256 over semantic fields only (claim-level identity; + two packets with the same semantic_key assert the same claim + regardless of provenance or instrument) + + Invariants (enforced at construction time) + ------------------------------------------ + - A D2–D4 frontend is forbidden from claiming AUTO_ACCEPT_ELIGIBLE. + - confidence and uncertainty must be in [0.0, 1.0]. + - provenance must be non-empty. + - payload_json must be valid JSON. + """ + kind: str + modality: Modality + provenance: tuple[SourceSpan, ...] + frontend: FrontendTrace + review_level: ReviewLevel + confidence: float + uncertainty: float + lemma: str = "" + subject: str = "" + verb: str = "" + object_: str = "" + payload_json: str = "{}" + + # Computed — set by __post_init__, declared as class-level defaults + # to satisfy the frozen dataclass protocol. + pressure_id: str = field(default="", init=False, compare=False, hash=False) + semantic_key: str = field(default="", init=False, compare=False, hash=False) + + def __post_init__(self) -> None: + # Confidence bounds + if not (0.0 <= self.confidence <= 1.0): + raise ValueError(f"confidence must be in [0.0, 1.0], got {self.confidence}") + if not (0.0 <= self.uncertainty <= 1.0): + raise ValueError(f"uncertainty must be in [0.0, 1.0], got {self.uncertainty}") + + # Provenance non-empty + if not self.provenance: + raise ValueError("provenance must contain at least one SourceSpan") + + # payload_json must be valid JSON; normalize to canonical form + try: + parsed = json.loads(self.payload_json) + canonical = json.dumps(parsed, sort_keys=True, separators=(",", ":")) + except json.JSONDecodeError as exc: + raise ValueError(f"payload_json is not valid JSON: {exc}") from exc + # Bypass frozen to set the normalized payload and computed fields + object.__setattr__(self, "payload_json", canonical) + + # Governance invariant: D2-D4 cannot claim AUTO_ACCEPT_ELIGIBLE + if ( + not self.frontend.determinism.auto_accept_eligible + and self.review_level == ReviewLevel.AUTO_ACCEPT_ELIGIBLE + ): + raise ValueError( + f"A {self.frontend.determinism} frontend cannot claim " + "AUTO_ACCEPT_ELIGIBLE. Assign OPERATOR_REVIEW_REQUIRED or " + "ARCHITECT_REVIEW_REQUIRED instead." + ) + + # Compute content-addressed identifiers + pid = make_pressure_id( + kind=self.kind, + modality=self.modality.value, + provenance=[(s.byte_start, s.byte_end, s.source_sha256) for s in self.provenance], + frontend_id=self.frontend.instrument_id, + determinism=self.frontend.determinism.value, + review_level=self.review_level.value, + confidence=self.confidence, + uncertainty=self.uncertainty, + lemma=self.lemma, + subject=self.subject, + verb=self.verb, + object_=self.object_, + payload_json=self.payload_json, + ) + sk = make_semantic_key( + kind=self.kind, + modality=self.modality.value, + lemma=self.lemma, + subject=self.subject, + verb=self.verb, + object_=self.object_, + payload_json=self.payload_json, + ) + object.__setattr__(self, "pressure_id", pid) + object.__setattr__(self, "semantic_key", sk) + + +# --------------------------------------------------------------------------- +# Validation outputs +# --------------------------------------------------------------------------- + +class GateDisposition(str, Enum): + ACCEPTED = "accepted" + REJECTED_PROVENANCE = "rejected_provenance" + REJECTED_SEMANTIC = "rejected_semantic" + REJECTED_GOVERNANCE = "rejected_governance" + REVIEW_REQUIRED = "review_required" + OVERRIDE_ACCEPTED = "override_accepted" # ReviewDecision authorized + + +@dataclass(frozen=True, slots=True) +class ValidationResult: + """ + Per-packet disposition from the IngestCompiler. + + The original packet is referenced by pressure_id, never mutated. + warnings may contain 'semantic_convergence:_prior_sources' when + multiple independent packets share the same semantic_key. + """ + pressure_id: str + semantic_key: str + disposition: GateDisposition + gate_failed: str | None = None # "provenance", "semantic", "governance" + failure_reason: str | None = None + warnings: tuple[str, ...] = () + + +@dataclass(frozen=True, slots=True) +class ValidationReport: + """ + Full output of one IngestCompiler.compile() call. + + results — one ValidationResult per input packet, in input order + accepted_ids — pressure_ids of packets that passed all three gates + rejected_ids — pressure_ids of rejected packets + review_ids — pressure_ids that require human/operator review + """ + results: tuple[ValidationResult, ...] + accepted_ids: frozenset[str] + rejected_ids: frozenset[str] + review_ids: frozenset[str] + + @property + def acceptance_rate(self) -> float: + total = len(self.results) + return len(self.accepted_ids) / total if total else 0.0 + + +@dataclass(frozen=True, slots=True) +class LearningArtifact: + """ + A governance-cleared packet exported to the train/ layer. + + Carries the original immutable packet plus the validation result that + authorized its export. The train/ layer must not modify either field. + """ + packet: CandidateGeometricPressure + result: ValidationResult + + +@dataclass(frozen=True, slots=True) +class ReviewDecision: + """ + An operator or architect decision to authorize a packet that would + otherwise be blocked by a review_required disposition. + + Does not mutate the original packet. The IngestCompiler checks the + authorized_ids set before applying the GovernanceGate rejection. + """ + authorized_ids: frozenset[str] # pressure_ids authorized for acceptance + authorized_by: str # operator or architect identifier + reason: str diff --git a/tests/test_core_ingest.py b/tests/test_core_ingest.py new file mode 100644 index 00000000..3df5ae09 --- /dev/null +++ b/tests/test_core_ingest.py @@ -0,0 +1,342 @@ +""" +Tests for core_ingest/ governance layer. + +Covers: + - CandidateGeometricPressure construction and invariants + - content-addressing (pressure_id, semantic_key) + - IngestCompiler three-gate pipeline + - structural deduplication and convergent-evidence detection + - ReviewDecision override path + - StructuralSegmenter (prose, scripture, code, math) + - SegmentManifold registration and lookup +""" + +from __future__ import annotations + +import hashlib +import json +import pytest + +from core_ingest.types import ( + CandidateGeometricPressure, + DeterminismClass, + FrontendTrace, + GateDisposition, + LearningArtifact, + Modality, + ReviewDecision, + ReviewLevel, + SourceSpan, +) +from core_ingest.compiler import IngestCompiler +from core_ingest.segmenter import SegmentKind, StructuralSegmenter +from core_ingest.manifold import SegmentManifold + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + +def _sha256(data: bytes) -> str: + return hashlib.sha256(data).hexdigest() + + +SOURCE_BYTES = b"In the beginning God created the heavens and the earth." +SOURCE_SHA = _sha256(SOURCE_BYTES) + + +def make_span( + start: int = 0, + end: int = 10, + sha: str = SOURCE_SHA, + region: str | None = "body", +) -> SourceSpan: + return SourceSpan( + byte_start=start, + byte_end=end, + source_sha256=sha, + region=region, + ) + + +def make_frontend( + det: DeterminismClass = DeterminismClass.D0, + iid: str = "StructuralSegmenter/prose/v1", +) -> FrontendTrace: + return FrontendTrace( + instrument_id=iid, + determinism=det, + version="1.0.0", + ) + + +def make_packet( + review_level: ReviewLevel = ReviewLevel.AUTO_ACCEPT_ELIGIBLE, + det: DeterminismClass = DeterminismClass.D0, + lemma: str = "beginning", + payload: dict | None = None, + kind: str = "assertion", + modality: Modality = Modality.TEXT, + span_start: int = 0, + span_end: int = 10, +) -> CandidateGeometricPressure: + return CandidateGeometricPressure( + kind=kind, + modality=modality, + provenance=(make_span(span_start, span_end),), + frontend=make_frontend(det), + review_level=review_level, + confidence=0.95, + uncertainty=0.05, + lemma=lemma, + payload_json=json.dumps(payload or {"text": "In the beginning"}), + ) + + +# --------------------------------------------------------------------------- +# CandidateGeometricPressure +# --------------------------------------------------------------------------- + +class TestCandidatePacket: + + def test_construction_d0(self): + p = make_packet() + assert p.pressure_id != "" + assert p.semantic_key != "" + assert len(p.pressure_id) == 64 + assert len(p.semantic_key) == 64 + + def test_payload_normalized(self): + # Unsorted keys should be canonicalized + p = make_packet(payload={"z": 1, "a": 2}) + parsed = json.loads(p.payload_json) + assert list(parsed.keys()) == sorted(parsed.keys()) + + def test_same_claim_same_semantic_key(self): + # Two packets with same semantic fields but different provenance + p1 = make_packet(span_start=0, span_end=10) + p2 = make_packet(span_start=20, span_end=30) + assert p1.semantic_key == p2.semantic_key + assert p1.pressure_id != p2.pressure_id + + def test_governance_invariant_d3_cannot_claim_auto_accept(self): + with pytest.raises(ValueError, match="AUTO_ACCEPT_ELIGIBLE"): + make_packet( + det=DeterminismClass.D3, + review_level=ReviewLevel.AUTO_ACCEPT_ELIGIBLE, + ) + + def test_confidence_out_of_range(self): + with pytest.raises(ValueError, match="confidence"): + CandidateGeometricPressure( + kind="assertion", + modality=Modality.TEXT, + provenance=(make_span(),), + frontend=make_frontend(), + review_level=ReviewLevel.AUTO_ACCEPT_ELIGIBLE, + confidence=1.5, + uncertainty=0.0, + lemma="test", + payload_json='{"text": "test"}', + ) + + def test_empty_provenance_rejected(self): + with pytest.raises(ValueError, match="provenance"): + CandidateGeometricPressure( + kind="assertion", + modality=Modality.TEXT, + provenance=(), + frontend=make_frontend(), + review_level=ReviewLevel.AUTO_ACCEPT_ELIGIBLE, + confidence=0.9, + uncertainty=0.1, + lemma="test", + payload_json='{"text": "test"}', + ) + + def test_invalid_payload_json_rejected(self): + with pytest.raises(ValueError, match="payload_json"): + CandidateGeometricPressure( + kind="assertion", + modality=Modality.TEXT, + provenance=(make_span(),), + frontend=make_frontend(), + review_level=ReviewLevel.AUTO_ACCEPT_ELIGIBLE, + confidence=0.9, + uncertainty=0.1, + lemma="test", + payload_json="not valid json{", + ) + + +# --------------------------------------------------------------------------- +# IngestCompiler +# --------------------------------------------------------------------------- + +class TestIngestCompiler: + + def setup_method(self): + self.compiler = IngestCompiler() + + def test_d0_packet_accepted(self): + p = make_packet() + report, artifacts = self.compiler.compile([p]) + assert p.pressure_id in report.accepted_ids + assert len(artifacts) == 1 + assert isinstance(artifacts[0], LearningArtifact) + + def test_d3_operator_review_required(self): + p = make_packet( + det=DeterminismClass.D3, + review_level=ReviewLevel.OPERATOR_REVIEW_REQUIRED, + ) + report, artifacts = self.compiler.compile([p]) + assert p.pressure_id in report.review_ids + assert len(artifacts) == 0 + + def test_auto_reject_always_rejected(self): + p = make_packet( + det=DeterminismClass.D0, + review_level=ReviewLevel.AUTO_REJECT, + ) + report, artifacts = self.compiler.compile([p]) + assert p.pressure_id in report.rejected_ids + + def test_structural_deduplication(self): + p = make_packet() + report, artifacts = self.compiler.compile([p, p]) + # First accepted, second rejected as duplicate + assert len(report.accepted_ids) == 1 + assert len(report.rejected_ids) == 1 + + def test_convergent_evidence_warning(self): + # Two packets same semantic claim, different provenance + p1 = make_packet(span_start=0, span_end=10) + p2 = make_packet(span_start=20, span_end=30) + assert p1.semantic_key == p2.semantic_key + report, _ = self.compiler.compile([p1, p2]) + # Second packet should carry a convergence warning + result_p2 = next(r for r in report.results if r.pressure_id == p2.pressure_id) + assert any("semantic_convergence" in w for w in result_p2.warnings) + + def test_review_decision_override(self): + p = make_packet( + det=DeterminismClass.D4, + review_level=ReviewLevel.ARCHITECT_REVIEW_REQUIRED, + ) + decision = ReviewDecision( + authorized_ids=frozenset({p.pressure_id}), + authorized_by="joshua.shay", + reason="Manually reviewed and approved", + ) + report, artifacts = self.compiler.compile([p], review_decision=decision) + assert p.pressure_id in report.accepted_ids + assert len(artifacts) == 1 + + def test_acceptance_rate(self): + p1 = make_packet(span_start=0, span_end=10) + p2 = make_packet( + det=DeterminismClass.D4, + review_level=ReviewLevel.OPERATOR_REVIEW_REQUIRED, + span_start=20, span_end=30, + ) + report, _ = self.compiler.compile([p1, p2]) + assert report.acceptance_rate == 0.5 + + +# --------------------------------------------------------------------------- +# StructuralSegmenter +# --------------------------------------------------------------------------- + +class TestStructuralSegmenter: + + def setup_method(self): + self.seg = StructuralSegmenter() + + def test_prose_segments_non_empty(self): + source = b"# Heading\n\nFirst paragraph.\n\nSecond paragraph." + segments = self.seg.segment(source, modality_hint="prose") + assert len(segments) >= 2 + + def test_prose_heading_detected(self): + source = b"# In the Beginning\n\nGod created the heavens." + segments = self.seg.segment(source, modality_hint="prose") + kinds = [s.kind for s in segments] + assert SegmentKind.HEADING in kinds + + def test_scripture_verse_detected(self): + source = b"Gen 1:1 In the beginning God created.\nGen 1:2 The earth was formless." + segments = self.seg.segment(source, modality_hint="scripture") + assert len(segments) >= 1 + for seg in segments: + assert seg.kind == SegmentKind.VERSE + assert "verse:" in seg.span.region + + def test_code_block_extracted(self): + source = b"Some prose.\n\n```python\nprint('logos')\n```\n\nMore prose." + segments = self.seg.segment(source, modality_hint="code") + assert len(segments) == 1 + assert segments[0].kind == SegmentKind.CODE + + def test_math_env_extracted(self): + source = rb"Let \[E = mc^2\] be the energy equation." + segments = self.seg.segment(source, modality_hint="math") + assert len(segments) == 1 + assert segments[0].kind == SegmentKind.MATH + + def test_span_byte_offsets_valid(self): + source = b"# Title\n\nBody text here." + for seg in self.seg.segment(source, modality_hint="prose"): + assert seg.span.byte_start >= 0 + assert seg.span.byte_end > seg.span.byte_start + assert seg.span.byte_end <= len(source) + + def test_source_sha256_consistent(self): + source = b"Consistent source." + expected_sha = _sha256(source) + for seg in self.seg.segment(source, modality_hint="prose"): + assert seg.span.source_sha256 == expected_sha + + +# --------------------------------------------------------------------------- +# SegmentManifold +# --------------------------------------------------------------------------- + +class TestSegmentManifold: + + def test_register_and_lookup(self): + manifold = SegmentManifold() + p = make_packet() + manifold.register([p]) + entries = manifold.lookup(p.semantic_key) + assert len(entries) == 1 + assert entries[0].pressure_id == p.pressure_id + + def test_spans_for_returns_all_spans(self): + manifold = SegmentManifold() + p1 = make_packet(span_start=0, span_end=10) + p2 = make_packet(span_start=20, span_end=30) + # Same semantic_key — different structural identity + assert p1.semantic_key == p2.semantic_key + manifold.register([p1, p2]) + spans = manifold.spans_for(p1.semantic_key) + assert len(spans) == 2 + + def test_len(self): + manifold = SegmentManifold() + p1 = make_packet(span_start=0, span_end=10) + p2 = make_packet(span_start=20, span_end=30, lemma="earth") + manifold.register([p1, p2]) + # p1 and p2 have different semantic keys (different lemma) + assert len(manifold) == 2 + + def test_contains(self): + manifold = SegmentManifold() + p = make_packet() + assert p.semantic_key not in manifold + manifold.register([p]) + assert p.semantic_key in manifold + + def test_lookup_missing_key(self): + manifold = SegmentManifold() + assert manifold.lookup("nonexistent_key") == []