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

feat: scaffold core_ingest/ governance layer (ADR-0012)

Browse source
This commit is contained in:
Shay 2026-05-13 11:35:00 -07:00
parent 6010924405
commit 7737f8ca66
7 changed files with 1455 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

58
core_ingest/__init__.py Normal file
View file

@ -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",
]

303
core_ingest/compiler.py Normal file
View file

@ -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

98
core_ingest/manifold.py Normal file
View file

@ -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

115
core_ingest/pressure.py Normal file
View file

@ -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"))

228
core_ingest/segmenter.py Normal file
View file

@ -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<hashes>#{1,6})\s+(?P<text>.+)$|^(?P<text2>.+)\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:<ref>'.
"""
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<fence>`{3,}|~{3,})(?P<lang>[^\n]*)\n(?P<body>.*?)^(?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"),
)

311
core_ingest/types.py Normal file
View file

@ -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:<n>_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

342
tests/test_core_ingest.py Normal file
View file

@ -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") == []