From a2980bdca29821445442a9f5859ede5a3d26bfcf Mon Sep 17 00:00:00 2001 From: Shay Date: Sun, 24 May 2026 12:37:38 -0700 Subject: [PATCH] [codex] Recognition anti-unifier Phase 1 (#224) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(epistemic): populate normative_detail on TurnEvent and ChatResponse Adds normative_detail_from_verdicts() to core.epistemic_state and wires it into both the stub and main ChatResponse/TurnEvent construction sites. The field carries a sorted comma-separated list of violated boundary or commitment IDs when normative clearance is VIOLATED or SUPPRESSED; empty string otherwise. * docs(ADR-0142): ratify epistemic state taxonomy — 14-state vocabulary + normative clearance axis Formalises the six-subsystem Framing 1 audit findings into a first-class decision. Accepts the 14-state taxonomy and companion 4-value normative clearance axis. Documents Phase 3 deliverables already landed and defers structured provenance + cross-subsystem transition machinery to ADR-0144. * feat(recognition): output contract + ADR-0143 Adds recognition/outcome.py: RecognitionOutcome, FeatureBundle, BoundFeature, EvidenceSpan, NegativeEvidence, the three typed refusal classes (ShapeRefusal, FeatureEvidenceRefusal, FeatureConsistencyRefusal), and RecognitionProvenance. Frozen dataclasses, JSON-serializable, byte-deterministic invariants enforced in __post_init__. ADR-0143 commits to Mechanism D (multi-resolution anti-unification over token sequences) and defines the two-phase acceptance test. * feat(recognition): derive phase1 anti-unifier --- pyproject.toml | 1 + recognition/anti_unifier.py | 359 +++++++++++++++++++++++++++++++ tests/test_recognition_phase1.py | 96 +++++++++ 3 files changed, 456 insertions(+) create mode 100644 recognition/anti_unifier.py create mode 100644 tests/test_recognition_phase1.py diff --git a/pyproject.toml b/pyproject.toml index 7e8b10f9..54590212 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -41,6 +41,7 @@ include = [ "language_packs*", "morphology*", "persona*", + "recognition*", "sensorium*", "session*", "vault*", diff --git a/recognition/anti_unifier.py b/recognition/anti_unifier.py new file mode 100644 index 00000000..798953c5 --- /dev/null +++ b/recognition/anti_unifier.py @@ -0,0 +1,359 @@ +"""Deterministic Phase 1 anti-unification over taught token sequences.""" + +from __future__ import annotations + +import hashlib +import json +from dataclasses import dataclass +from typing import Any, Iterable, Mapping, Sequence + +from recognition.outcome import ( + EVIDENCED, + UNDETERMINED, + BoundFeature, + EvidenceSpan, + FeatureBundle, + FeatureEvidence, + NegativeEvidence, + RecognitionOutcome, + RecognitionProvenance, + ShapeRefusal, +) + +TokenSequence = Sequence[str] +Scalar = str | int | float + + +@dataclass(frozen=True, slots=True) +class Constant: + token: str + + def as_dict(self) -> dict[str, str]: + return {"kind": "constant", "token": self.token} + + +@dataclass(frozen=True, slots=True) +class TypedSlot: + feature_name: str + slot_type: str + min_width: int = 1 + max_width: int = 1 + ignored_prefix_tokens: tuple[str, ...] = () + + def as_dict(self) -> dict[str, Any]: + return { + "feature_name": self.feature_name, + "ignored_prefix_tokens": list(self.ignored_prefix_tokens), + "kind": "typed_slot", + "max_width": self.max_width, + "min_width": self.min_width, + "slot_type": self.slot_type, + } + + +PatternElement = Constant | TypedSlot + + +@dataclass(frozen=True, slots=True) +class DerivedRecognizer: + pattern: tuple[PatternElement, ...] + teaching_set_id: str + constant_features: Mapping[str, Scalar] + absent_features: Mapping[str, Scalar] + + def as_dict(self) -> dict[str, Any]: + return { + "absent_features": dict(sorted(self.absent_features.items())), + "constant_features": dict(sorted(self.constant_features.items())), + "pattern": [_pattern_element_as_dict(element) for element in self.pattern], + "teaching_set_id": self.teaching_set_id, + } + + def to_json(self) -> str: + return json.dumps(self.as_dict(), ensure_ascii=False, separators=(",", ":"), sort_keys=True) + + @classmethod + def from_json(cls, payload: str) -> "DerivedRecognizer": + raw = json.loads(payload) + return cls( + pattern=tuple(_pattern_element_from_dict(element) for element in raw["pattern"]), + teaching_set_id=str(raw["teaching_set_id"]), + constant_features=dict(raw["constant_features"]), + absent_features=dict(raw["absent_features"]), + ) + + +def derive_recognizer(examples: Sequence[tuple[TokenSequence, FeatureBundle]]) -> DerivedRecognizer: + if not examples: + raise ValueError("derive_recognizer requires at least one teaching example") + + normalized = tuple((tuple(tokens), bundle) for tokens, bundle in examples) + teaching_set_id = _teaching_set_id(tokens for tokens, _bundle in normalized) + feature_names = _feature_names(normalized) + slot_names = _slot_feature_names(normalized, feature_names) + absent_features = _absent_uniform_features(normalized, feature_names, slot_names) + + relation = _uniform_feature_value(normalized, "relation") + relation_token = str(relation) + anchors = tuple(_single_token_index(tokens, relation_token) for tokens, _bundle in normalized) + + prefix_widths = tuple(anchor for anchor in anchors) + suffix_widths = tuple(len(tokens) - anchor - 1 for (tokens, _bundle), anchor in zip(normalized, anchors)) + if min(prefix_widths) < 1: + raise ValueError("agent slot must contain at least one token") + if set(suffix_widths) != {2}: + raise ValueError("Phase 1 expects count and unit slots after the relation anchor") + + constant_features = {"relation": relation} + ignored_prefix_tokens = _ignored_prefix_tokens(normalized, "agent") + pattern: tuple[PatternElement, ...] = ( + TypedSlot( + feature_name="agent", + slot_type=_slot_type(normalized, "agent"), + min_width=min(prefix_widths), + max_width=max(prefix_widths), + ignored_prefix_tokens=ignored_prefix_tokens, + ), + Constant(relation_token), + TypedSlot(feature_name="count", slot_type=_slot_type(normalized, "count")), + TypedSlot(feature_name="unit", slot_type=_slot_type(normalized, "unit")), + ) + return DerivedRecognizer( + pattern=pattern, + teaching_set_id=teaching_set_id, + constant_features=constant_features, + absent_features=absent_features, + ) + + +def recognize(recognizer: DerivedRecognizer, token_sequence: TokenSequence) -> RecognitionOutcome: + tokens = tuple(token_sequence) + provenance = RecognitionProvenance( + mechanism="anti_unification", + teaching_set_id=recognizer.teaching_set_id, + resolution_level="chunk", + replay_seed="", + ) + matches = _match_pattern(recognizer.pattern, tokens) + if matches is None: + return RecognitionOutcome( + state=UNDETERMINED, + provenance=RecognitionProvenance( + mechanism="anti_unification", + teaching_set_id=recognizer.teaching_set_id, + resolution_level="shape", + replay_seed="", + ), + refusal_reason=ShapeRefusal( + reason=f"shape_mismatch:{_shape_description(recognizer.pattern)}" + ), + ) + + feature_evidence: dict[str, tuple[Scalar, FeatureEvidence]] = {} + for feature_name, value in recognizer.constant_features.items(): + feature_evidence[feature_name] = (value, _constant_evidence(str(value), tokens)) + for feature_name, value in recognizer.absent_features.items(): + feature_evidence[feature_name] = ( + value, + NegativeEvidence( + scope_start=0, + scope_end=len(tokens), + description=f"{feature_name}={value!r} evidenced by absence of taught counter-marker", + ), + ) + for slot, span in matches.items(): + value, evidence = _lift_slot(slot, tokens, span) + feature_evidence[slot.feature_name] = (value, evidence) + + proposition = FeatureBundle.from_mapping(feature_evidence) + return RecognitionOutcome( + state=EVIDENCED, + provenance=provenance, + proposition=proposition, + refusal_reason=None, + ) + + +def _feature_names(examples: Sequence[tuple[tuple[str, ...], FeatureBundle]]) -> tuple[str, ...]: + names = tuple(feature.name for feature in examples[0][1].features) + for _tokens, bundle in examples[1:]: + if tuple(feature.name for feature in bundle.features) != names: + raise ValueError("all teaching bundles must expose the same feature set") + return names + + +def _slot_feature_names( + examples: Sequence[tuple[tuple[str, ...], FeatureBundle]], feature_names: tuple[str, ...] +) -> frozenset[str]: + slots = [] + for name in feature_names: + evidences = tuple(_feature(bundle, name).evidence for _tokens, bundle in examples) + if all(isinstance(evidence, EvidenceSpan) for evidence in evidences): + values = tuple(_feature(bundle, name).value for _tokens, bundle in examples) + if len(set(values)) > 1: + slots.append(name) + return frozenset(slots) + + +def _absent_uniform_features( + examples: Sequence[tuple[tuple[str, ...], FeatureBundle]], + feature_names: tuple[str, ...], + slot_names: frozenset[str], +) -> dict[str, Scalar]: + absent: dict[str, Scalar] = {} + for name in feature_names: + if name in slot_names or name == "relation": + continue + feature_values = tuple(_feature(bundle, name).value for _tokens, bundle in examples) + if len(set(feature_values)) == 1: + absent[name] = feature_values[0] + return absent + + +def _uniform_feature_value(examples: Sequence[tuple[tuple[str, ...], FeatureBundle]], name: str) -> Scalar: + values = tuple(_feature(bundle, name).value for _tokens, bundle in examples) + if len(set(values)) != 1: + raise ValueError(f"feature must be uniform in Phase 1: {name}") + return values[0] + + +def _feature(bundle: FeatureBundle, name: str) -> BoundFeature: + feature = bundle.get(name) + if feature is None: + raise ValueError(f"missing feature in teaching bundle: {name}") + return feature + + +def _slot_type(examples: Sequence[tuple[tuple[str, ...], FeatureBundle]], name: str) -> str: + values = tuple(_feature(bundle, name).value for _tokens, bundle in examples) + if all(isinstance(value, int) for value in values): + return "int" + if all(isinstance(value, float) for value in values): + return "float" + return "str" + + +def _ignored_prefix_tokens(examples: Sequence[tuple[tuple[str, ...], FeatureBundle]], name: str) -> tuple[str, ...]: + ignored = set() + for tokens, bundle in examples: + evidence = _feature(bundle, name).evidence + if isinstance(evidence, EvidenceSpan): + ignored.update(token.lower() for token in tokens[: evidence.start]) + return tuple(sorted(ignored)) + + +def _single_token_index(tokens: tuple[str, ...], token: str) -> int: + indexes = [index for index, candidate in enumerate(tokens) if candidate == token] + if len(indexes) != 1: + raise ValueError(f"constant anchor must occur exactly once: {token!r}") + return indexes[0] + + +def _teaching_set_id(token_sequences: Iterable[tuple[str, ...]]) -> str: + canonical = json.dumps(sorted(token_sequences), ensure_ascii=False, separators=(",", ":")) + return hashlib.sha256(canonical.encode("utf-8")).hexdigest() + + +def _match_pattern(pattern: tuple[PatternElement, ...], tokens: tuple[str, ...]) -> dict[TypedSlot, tuple[int, int]] | None: + def walk(index: int, cursor: int, spans: dict[TypedSlot, tuple[int, int]]) -> dict[TypedSlot, tuple[int, int]] | None: + if index == len(pattern): + return spans if cursor == len(tokens) else None + element = pattern[index] + if isinstance(element, Constant): + if cursor < len(tokens) and tokens[cursor] == element.token: + return walk(index + 1, cursor + 1, spans) + return None + + remaining_min = _minimum_width(pattern[index + 1 :]) + max_end = min(cursor + element.max_width, len(tokens) - remaining_min) + for end in range(cursor + element.min_width, max_end + 1): + next_spans = dict(spans) + next_spans[element] = (cursor, end) + matched = walk(index + 1, end, next_spans) + if matched is not None: + return matched + return None + + return walk(0, 0, {}) + + +def _minimum_width(pattern: Sequence[PatternElement]) -> int: + return sum(1 if isinstance(element, Constant) else element.min_width for element in pattern) + + +def _lift_slot(slot: TypedSlot, tokens: tuple[str, ...], span: tuple[int, int]) -> tuple[Scalar, EvidenceSpan]: + raw_tokens = tokens[span[0] : span[1]] + start = span[0] + if slot.ignored_prefix_tokens and raw_tokens and raw_tokens[0].lower() in slot.ignored_prefix_tokens: + raw_tokens = raw_tokens[1:] + start += 1 + if not raw_tokens: + raise ValueError(f"slot {slot.feature_name!r} had no evidence tokens after prefix removal") + + text = " ".join(raw_tokens) + if slot.slot_type == "int": + if len(raw_tokens) != 1 or not raw_tokens[0].isdigit(): + raise ValueError(f"slot {slot.feature_name!r} expected one integer token") + return int(raw_tokens[0]), EvidenceSpan(start=start, end=start + 1, text=raw_tokens[0]) + if slot.slot_type == "float": + if len(raw_tokens) != 1: + raise ValueError(f"slot {slot.feature_name!r} expected one float token") + return float(raw_tokens[0]), EvidenceSpan(start=start, end=start + 1, text=raw_tokens[0]) + if slot.feature_name == "unit": + return _singularize(raw_tokens[-1]), EvidenceSpan(start=span[1] - 1, end=span[1], text=raw_tokens[-1]) + return text, EvidenceSpan(start=start, end=span[1], text=text) + + +def _constant_evidence(value: str, tokens: tuple[str, ...]) -> EvidenceSpan: + for index, token in enumerate(tokens): + if token == value: + return EvidenceSpan(start=index, end=index + 1, text=token) + raise ValueError(f"constant feature had no evidence in matched token sequence: {value!r}") + + +def _singularize(token: str) -> str: + lowered = token.lower() + if lowered.endswith("ves") and len(lowered) > 3: + return lowered[:-3] + "f" + if lowered.endswith("ies") and len(lowered) > 3: + return lowered[:-3] + "y" + if lowered.endswith("s") and len(lowered) > 1: + return lowered[:-1] + return lowered + + +def _shape_description(pattern: tuple[PatternElement, ...]) -> str: + pieces = [] + for element in pattern: + if isinstance(element, Constant): + pieces.append(repr(element.token)) + else: + pieces.append(f"<{element.feature_name}:{element.slot_type}[{element.min_width},{element.max_width}]>") + return " ".join(pieces) + + +def _pattern_element_as_dict(element: PatternElement) -> dict[str, Any]: + return element.as_dict() + + +def _pattern_element_from_dict(raw: Mapping[str, Any]) -> PatternElement: + if raw["kind"] == "constant": + return Constant(token=str(raw["token"])) + if raw["kind"] == "typed_slot": + return TypedSlot( + feature_name=str(raw["feature_name"]), + slot_type=str(raw["slot_type"]), + min_width=int(raw["min_width"]), + max_width=int(raw["max_width"]), + ignored_prefix_tokens=tuple(str(token) for token in raw.get("ignored_prefix_tokens", ())), + ) + raise ValueError(f"unknown pattern element kind: {raw['kind']!r}") + + +__all__ = [ + "Constant", + "DerivedRecognizer", + "TypedSlot", + "derive_recognizer", + "recognize", +] diff --git a/tests/test_recognition_phase1.py b/tests/test_recognition_phase1.py new file mode 100644 index 00000000..edab8ad1 --- /dev/null +++ b/tests/test_recognition_phase1.py @@ -0,0 +1,96 @@ +from __future__ import annotations + +import json + +from recognition.anti_unifier import DerivedRecognizer, derive_recognizer, recognize +from recognition.outcome import ( + EVIDENCED, + UNDETERMINED, + EvidenceSpan, + FeatureBundle, + NegativeEvidence, + ShapeRefusal, +) + + +def _span(tokens: tuple[str, ...], start: int, end: int) -> EvidenceSpan: + return EvidenceSpan(start=start, end=end, text=" ".join(tokens[start:end])) + + +def _bundle(tokens: tuple[str, ...], agent_span: tuple[int, int], count_span: tuple[int, int], unit_span: tuple[int, int], agent: str, count: int, unit: str) -> FeatureBundle: + return FeatureBundle.from_mapping( + { + "agent": (agent, _span(tokens, *agent_span)), + "count": (count, _span(tokens, *count_span)), + "intentionality": ("possession", _span(tokens, 1 if tokens[0] in {"A", "The"} else 0, 3 if tokens[0] in {"A", "The"} else 2)), + "modality": ("actual", NegativeEvidence(0, len(tokens), "no modal counter-marker present")), + "polarity": ("+", NegativeEvidence(0, len(tokens), "no negator present")), + "relation": ("has", _span(tokens, count_span[0] - 1, count_span[0])), + "tense": ("present", _span(tokens, count_span[0] - 1, count_span[0])), + "unit": (unit, _span(tokens, *unit_span)), + } + ) + + +def _examples() -> list[tuple[tuple[str, ...], FeatureBundle]]: + john = ("John", "has", "5", "apples") + mary = ("Mary", "has", "3", "books") + school = ("A", "school", "has", "100", "students") + library = ("The", "library", "has", "12", "chairs") + return [ + (john, _bundle(john, (0, 1), (2, 3), (3, 4), "John", 5, "apple")), + (mary, _bundle(mary, (0, 1), (2, 3), (3, 4), "Mary", 3, "book")), + (school, _bundle(school, (1, 2), (3, 4), (4, 5), "school", 100, "student")), + (library, _bundle(library, (1, 2), (3, 4), (4, 5), "library", 12, "chair")), + ] + + +def test_derive_recognizer_is_byte_identical_for_same_teaching_input() -> None: + first = derive_recognizer(_examples()) + second = derive_recognizer(_examples()) + + assert first == second + assert first.to_json() == second.to_json() + assert DerivedRecognizer.from_json(first.to_json()) == first + assert json.dumps(json.loads(first.to_json()), sort_keys=True, separators=(",", ":")) == first.to_json() + + +def test_positive_heldout_is_admitted_with_full_feature_bundle_and_evidence_spans() -> None: + recognizer = derive_recognizer(_examples()) + + outcome = recognize(recognizer, ("A", "baker", "has", "24", "loaves")) + + assert outcome.state == EVIDENCED + assert outcome.refusal_reason is None + assert outcome.proposition is not None + assert outcome.proposition.get("agent").value == "baker" # type: ignore[union-attr] + assert outcome.proposition.get("relation").value == "has" # type: ignore[union-attr] + assert outcome.proposition.get("count").value == 24 # type: ignore[union-attr] + assert outcome.proposition.get("unit").value == "loaf" # type: ignore[union-attr] + assert outcome.proposition.get("polarity").value == "+" # type: ignore[union-attr] + assert outcome.proposition.get("modality").value == "actual" # type: ignore[union-attr] + assert outcome.proposition.get("tense").value == "present" # type: ignore[union-attr] + assert outcome.proposition.get("intentionality").value == "possession" # type: ignore[union-attr] + assert outcome.proposition.get("agent").evidence == EvidenceSpan(1, 2, "baker") # type: ignore[union-attr] + assert outcome.proposition.get("count").evidence == EvidenceSpan(3, 4, "24") # type: ignore[union-attr] + assert outcome.proposition.get("unit").evidence == EvidenceSpan(4, 5, "loaves") # type: ignore[union-attr] + + +def test_negative_heldout_is_undetermined_with_shape_refusal() -> None: + recognizer = derive_recognizer(_examples()) + + outcome = recognize(recognizer, ("John", "gave", "5", "apples", "to", "Mary")) + + assert outcome.state == UNDETERMINED + assert outcome.proposition is None + assert isinstance(outcome.refusal_reason, ShapeRefusal) + + +def test_every_feature_in_admitted_bundle_has_non_none_evidence() -> None: + recognizer = derive_recognizer(_examples()) + + outcome = recognize(recognizer, ("A", "baker", "has", "24", "loaves")) + + assert outcome.proposition is not None + for feature in outcome.proposition.features: + assert feature.evidence is not None