From 0d606031e6d5e21872d971db8dc1f0ea65d5bb13 Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Sat, 31 Oct 2026 13:55:12 +0100 Subject: [PATCH] feat(detection-core): represent family-neutral semantic evidence --- .kiro/specs/detection-core/tasks.md | 2 +- src/openultrasast/core/semantics.py | 425 ++++++++++++++++++++++++++++ tests/test_core_semantics.py | 400 ++++++++++++++++++++++++++ 3 files changed, 826 insertions(+), 1 deletion(-) create mode 100644 src/openultrasast/core/semantics.py create mode 100644 tests/test_core_semantics.py diff --git a/.kiro/specs/detection-core/tasks.md b/.kiro/specs/detection-core/tasks.md index b46398c..869b3a1 100644 --- a/.kiro/specs/detection-core/tasks.md +++ b/.kiro/specs/detection-core/tasks.md @@ -34,7 +34,7 @@ - _Requirements: 1.4, 10.4_ - [ ] 2. Implement canonical semantic, provenance, and evidence contracts -- [ ] 2.1 (P) Represent family-neutral semantic relations and observations +- [x] 2.1 (P) Represent family-neutral semantic relations and observations - Cover registration, invocation, call, return, value transformation, operation effect, and constraint effect. - Preserve role, location, conditions, provenance, and independent complete, incomplete, or unresolved status. - Refuse conflicting complete candidates by producing explicit ambiguity rather than selecting one. diff --git a/src/openultrasast/core/semantics.py b/src/openultrasast/core/semantics.py new file mode 100644 index 0000000..adcfe4a --- /dev/null +++ b/src/openultrasast/core/semantics.py @@ -0,0 +1,425 @@ +"""Family-neutral semantic evidence contracts for detection core v2.""" + +import math +import re +from collections.abc import Mapping, Sequence +from dataclasses import dataclass +from enum import Enum +from types import MappingProxyType + +from openultrasast.core.canonical import ContentRef, JsonValue, content_digest, require_unique_identities + +SEMANTIC_RELATION_SCHEMA = "openultrasast.detection-core.semantic-relation.v2" +OBSERVATION_SCHEMA = "openultrasast.detection-core.observation.v2" +_DIGEST = re.compile(r"[0-9a-f]{64}") + + +class SemanticContractError(ValueError): + pass + + +class FieldKind(Enum): + TEXT = "text" + INTEGER = "integer" + NUMBER = "number" + BOOLEAN = "boolean" + TEXT_TUPLE = "text_tuple" + SOURCE_LOCATION = "source_location" + + +class RelationKind(Enum): + REGISTRATION = "registration" + INVOCATION = "invocation" + CALL = "call" + RETURN = "return" + VALUE_TRANSFORMATION = "value_transformation" + OPERATION_EFFECT = "operation_effect" + CONSTRAINT_EFFECT = "constraint_effect" + + +class RelationStatus(Enum): + COMPLETE = "complete" + INCOMPLETE = "incomplete" + UNRESOLVED = "unresolved" + + +class ProvenanceKind(Enum): + PRODUCT_CODE = "product_code" + SEMANTIC_FACT = "semantic_fact" + + +def _require_text(value: object, field: str) -> None: + if not isinstance(value, str) or not value: + raise SemanticContractError(f"{field} must be nonempty text") + + +def _require_digest(value: object, field: str) -> None: + if not isinstance(value, str) or _DIGEST.fullmatch(value) is None: + raise SemanticContractError(f"{field} must be a full lowercase SHA-256 digest") + + +def _require_enum(value: object, enum_type: type[Enum], field: str) -> None: + if not isinstance(value, enum_type): + raise SemanticContractError(f"{field} must be a {enum_type.__name__}") + + +def _require_text_tuple(values: object, field: str) -> None: + if not isinstance(values, tuple) or not all(isinstance(value, str) and value for value in values): + raise SemanticContractError(f"{field} must be a tuple of nonempty text") + try: + require_unique_identities(values, kind=field) + except ValueError as exc: + raise SemanticContractError(str(exc)) from exc + + +@dataclass(frozen=True) +class SourceLocation: + path: str + line: int + column: int | None + symbol: str | None + + def __post_init__(self) -> None: + _require_text(self.path, "path") + if isinstance(self.line, bool) or not isinstance(self.line, int) or self.line < 1: + raise SemanticContractError("line must be a positive integer") + if self.column is not None and (isinstance(self.column, bool) or not isinstance(self.column, int) or self.column < 0): + raise SemanticContractError("column must be a non-negative integer or None") + if self.symbol is not None: + _require_text(self.symbol, "symbol") + + +@dataclass(frozen=True) +class SemanticNode: + node_id: str + role: str + location: SourceLocation + + def __post_init__(self) -> None: + _require_text(self.node_id, "node_id") + _require_text(self.role, "role") + if not isinstance(self.location, SourceLocation): + raise SemanticContractError("location must be a SourceLocation") + + +def _location_value(location: SourceLocation) -> Mapping[str, JsonValue]: + return {"path": location.path, "line": location.line, "column": location.column, "symbol": location.symbol} + + +def _node_value(node: SemanticNode) -> Mapping[str, JsonValue]: + return {"node_id": node.node_id, "role": node.role, "location": _location_value(node.location)} + + +def _content_ref_value(reference: ContentRef) -> Mapping[str, JsonValue]: + return { + "artifact_id": reference.artifact_id, + "content_digest": reference.content_digest, + "media_type": reference.media_type, + "byte_length": reference.byte_length, + "retention_policy_digest": reference.retention_policy_digest, + "storage_receipt_digest": reference.storage_receipt_digest, + } + + +@dataclass(frozen=True) +class EvidenceProvenance: + kind: ProvenanceKind + source_content_digest: str + semantic_pack_digest: str | None + semantic_fact_id: str | None + fact_match: ContentRef | None + + def __post_init__(self) -> None: + _require_enum(self.kind, ProvenanceKind, "kind") + _require_digest(self.source_content_digest, "source_content_digest") + if self.kind is ProvenanceKind.PRODUCT_CODE: + if any(value is not None for value in (self.semantic_pack_digest, self.semantic_fact_id, self.fact_match)): + raise SemanticContractError("product-code provenance forbids semantic fact fields") + return + if self.semantic_pack_digest is None: + raise SemanticContractError("semantic_pack_digest is required") + _require_digest(self.semantic_pack_digest, "semantic_pack_digest") + if self.semantic_fact_id is None: + raise SemanticContractError("semantic_fact_id is required") + _require_text(self.semantic_fact_id, "semantic_fact_id") + if not isinstance(self.fact_match, ContentRef): + raise SemanticContractError("fact_match is required and must be a ContentRef") + + +def _provenance_value(provenance: EvidenceProvenance) -> Mapping[str, JsonValue]: + fact_match: JsonValue = None if provenance.fact_match is None else _content_ref_value(provenance.fact_match) + return { + "kind": provenance.kind.value, + "source_content_digest": provenance.source_content_digest, + "semantic_pack_digest": provenance.semantic_pack_digest, + "semantic_fact_id": provenance.semantic_fact_id, + "fact_match": fact_match, + } + + +def _source_location_field(value: object) -> bool: + if not isinstance(value, Mapping) or set(value) != {"path", "line", "column", "symbol"}: + return False + path, line, column, symbol = value["path"], value["line"], value["column"], value["symbol"] + return ( + isinstance(path, str) + and bool(path) + and isinstance(line, int) + and not isinstance(line, bool) + and line >= 1 + and (column is None or (isinstance(column, int) and not isinstance(column, bool) and column >= 0)) + and (symbol is None or (isinstance(symbol, str) and bool(symbol))) + ) + + +def _valid_field_value(kind: FieldKind, value: object) -> bool: + if kind is FieldKind.TEXT: + return isinstance(value, str) + if kind is FieldKind.INTEGER: + return isinstance(value, int) and not isinstance(value, bool) + if kind is FieldKind.NUMBER: + return isinstance(value, (int, float)) and not isinstance(value, bool) and math.isfinite(value) + if kind is FieldKind.BOOLEAN: + return isinstance(value, bool) + if kind is FieldKind.TEXT_TUPLE: + return isinstance(value, tuple) and all(isinstance(item, str) for item in value) + return _source_location_field(value) + + +def _freeze_json(value: JsonValue) -> JsonValue: + if isinstance(value, Mapping): + return MappingProxyType({key: _freeze_json(item) for key, item in value.items()}) + if isinstance(value, tuple): + return tuple(_freeze_json(item) for item in value) + return value + + +@dataclass(frozen=True) +class FieldValue: + name: str + kind: FieldKind + value: JsonValue + + def __post_init__(self) -> None: + _require_text(self.name, "name") + _require_enum(self.kind, FieldKind, "kind") + frozen = _freeze_json(self.value) + if not _valid_field_value(self.kind, frozen): + raise SemanticContractError(f"value does not match FieldKind.{self.kind.name}") + object.__setattr__(self, "value", frozen) + + +def _field_value(field: FieldValue) -> Mapping[str, JsonValue]: + return {"name": field.name, "kind": field.kind.value, "value": field.value} + + +def _relation_payload(relation: "SemanticRelation") -> Mapping[str, JsonValue]: + return { + "schema_version": SEMANTIC_RELATION_SCHEMA, + "kind": relation.kind.value, + "role": relation.role, + "source": _node_value(relation.source), + "target": _node_value(relation.target), + "location": _location_value(relation.location), + "status": relation.status.value, + "conditions": relation.conditions, + "reasons": relation.reasons, + "provenance": _provenance_value(relation.provenance), + } + + +@dataclass(frozen=True) +class SemanticRelation: + relation_id: str + kind: RelationKind + role: str + source: SemanticNode + target: SemanticNode + location: SourceLocation + status: RelationStatus + conditions: tuple[str, ...] + reasons: tuple[str, ...] + provenance: EvidenceProvenance + + def __post_init__(self) -> None: + _validate_relation_fields(self, allow_empty_id=True) + expected = content_digest(_relation_payload(self)) + if not self.relation_id: + object.__setattr__(self, "relation_id", expected) + elif self.relation_id != expected: + raise SemanticContractError("relation_id does not match canonical relation fields") + + +def _validate_relation_fields(relation: SemanticRelation, *, allow_empty_id: bool = False) -> None: + if relation.relation_id or not allow_empty_id: + _require_digest(relation.relation_id, "relation_id") + _require_enum(relation.kind, RelationKind, "kind") + _require_text(relation.role, "role") + if not isinstance(relation.source, SemanticNode) or not isinstance(relation.target, SemanticNode): + raise SemanticContractError("source and target must be SemanticNode values") + if not isinstance(relation.location, SourceLocation): + raise SemanticContractError("location must be a SourceLocation") + _require_enum(relation.status, RelationStatus, "status") + _require_text_tuple(relation.conditions, "condition") + _require_text_tuple(relation.reasons, "reason") + if relation.status is RelationStatus.COMPLETE and relation.reasons: + raise SemanticContractError("complete relation must not carry incompleteness reasons") + if relation.status is not RelationStatus.COMPLETE and not relation.reasons: + raise SemanticContractError("incomplete or unresolved relation requires a reason") + if not isinstance(relation.provenance, EvidenceProvenance): + raise SemanticContractError("provenance must be EvidenceProvenance") + + +def make_relation( + *, + kind: RelationKind, + role: str, + source: SemanticNode, + target: SemanticNode, + location: SourceLocation, + status: RelationStatus, + conditions: tuple[str, ...], + reasons: tuple[str, ...], + provenance: EvidenceProvenance, +) -> SemanticRelation: + return SemanticRelation("", kind, role, source, target, location, status, conditions, reasons, provenance) + + +def relation_from_conditions( + *, + kind: RelationKind, + role: str, + source: SemanticNode, + target: SemanticNode, + location: SourceLocation, + declared_conditions: tuple[str, ...], + unmet_conditions: tuple[str, ...], + provenance: EvidenceProvenance, +) -> SemanticRelation: + _require_text_tuple(declared_conditions, "declared condition") + _require_text_tuple(unmet_conditions, "unmet condition") + if not set(unmet_conditions).issubset(declared_conditions): + raise SemanticContractError("unmet conditions must be declared") + status = RelationStatus.INCOMPLETE if unmet_conditions else RelationStatus.COMPLETE + reasons = tuple(f"unmet_condition:{condition}" for condition in unmet_conditions) + return make_relation( + kind=kind, + role=role, + source=source, + target=target, + location=location, + status=status, + conditions=declared_conditions, + reasons=reasons, + provenance=provenance, + ) + + +def _same_slot(candidate: SemanticRelation, first: SemanticRelation) -> bool: + return all(getattr(candidate, field) == getattr(first, field) for field in ("kind", "role", "source", "target", "location")) + + +def _candidate_provenance_reason(candidate: SemanticRelation) -> str: + provenance = candidate.provenance + match = "-" if provenance.fact_match is None else provenance.fact_match.content_digest + return ":".join( + ( + "conflicting_candidate_provenance", + content_digest(_provenance_value(provenance)), + provenance.kind.value, + provenance.semantic_pack_digest or "-", + provenance.semantic_fact_id or "-", + match, + ) + ) + + +def resolve_relation_candidates(candidates: Sequence[SemanticRelation]) -> SemanticRelation: + if not candidates: + raise SemanticContractError("at least one relation candidate is required") + if not all(isinstance(candidate, SemanticRelation) for candidate in candidates): + raise SemanticContractError("every candidate must be a SemanticRelation") + first = candidates[0] + if not all(_same_slot(candidate, first) for candidate in candidates): + raise SemanticContractError("relation candidates must describe the same semantic slot") + try: + require_unique_identities(tuple(candidate.relation_id for candidate in candidates), kind="candidate") + except ValueError as exc: + raise SemanticContractError(str(exc)) from exc + complete = [candidate for candidate in candidates if candidate.status is RelationStatus.COMPLETE] + if len(complete) == 1: + return complete[0] + if len(candidates) == 1: + return first + ordered = sorted(candidates, key=lambda candidate: candidate.relation_id) + anchor = ordered[0] + conditions = tuple(sorted({condition for candidate in ordered for condition in candidate.conditions})) + reasons = ( + tuple(f"conflicting_complete_candidate:{candidate.relation_id}" for candidate in ordered) + + tuple(sorted({_candidate_provenance_reason(candidate) for candidate in ordered})) + + (f"ambiguity_provenance_anchor:{anchor.relation_id}",) + ) + return make_relation( + kind=anchor.kind, + role=anchor.role, + source=anchor.source, + target=anchor.target, + location=anchor.location, + status=RelationStatus.UNRESOLVED, + conditions=conditions, + reasons=reasons, + provenance=anchor.provenance, + ) + + +def _observation_payload(observation: "Observation") -> Mapping[str, JsonValue]: + return { + "schema_version": OBSERVATION_SCHEMA, + "role": observation.role, + "location": _location_value(observation.location), + "fields": tuple(_field_value(field) for field in observation.fields), + "status": observation.status.value, + "provenance": _provenance_value(observation.provenance), + } + + +@dataclass(frozen=True) +class Observation: + observation_id: str + role: str + location: SourceLocation + fields: tuple[FieldValue, ...] + status: RelationStatus + provenance: EvidenceProvenance + + def __post_init__(self) -> None: + if self.observation_id: + _require_digest(self.observation_id, "observation_id") + _require_text(self.role, "role") + if not isinstance(self.location, SourceLocation): + raise SemanticContractError("location must be a SourceLocation") + if not isinstance(self.fields, tuple) or not all(isinstance(field, FieldValue) for field in self.fields): + raise SemanticContractError("fields must be a tuple of FieldValue values") + try: + require_unique_identities(tuple(field.name for field in self.fields), kind="field") + except ValueError as exc: + raise SemanticContractError(str(exc)) from exc + _require_enum(self.status, RelationStatus, "status") + if not isinstance(self.provenance, EvidenceProvenance): + raise SemanticContractError("provenance must be EvidenceProvenance") + expected = content_digest(_observation_payload(self)) + if not self.observation_id: + object.__setattr__(self, "observation_id", expected) + elif self.observation_id != expected: + raise SemanticContractError("observation_id does not match canonical observation fields") + + +def make_observation( + *, + role: str, + location: SourceLocation, + fields: tuple[FieldValue, ...], + status: RelationStatus, + provenance: EvidenceProvenance, +) -> Observation: + return Observation("", role, location, fields, status, provenance) diff --git a/tests/test_core_semantics.py b/tests/test_core_semantics.py new file mode 100644 index 0000000..f469eea --- /dev/null +++ b/tests/test_core_semantics.py @@ -0,0 +1,400 @@ +"""Family-neutral semantic evidence contracts (Req 1.2, 4.1-4.3, 5.2).""" + +from __future__ import annotations + +from collections.abc import Mapping, MutableMapping +from dataclasses import FrozenInstanceError, replace +from enum import Enum +from typing import cast + +import pytest +from detection_core_fixtures import stored_artifact + +import openultrasast.core.canonical as canonical +import openultrasast.core.semantics as semantics + + +def _location(line: int = 7) -> semantics.SourceLocation: + return semantics.SourceLocation(path="src/app.py", line=line, column=3, symbol="handle") + + +def _node(node_id: str, role: str) -> semantics.SemanticNode: + return semantics.SemanticNode(node_id=node_id, role=role, location=_location()) + + +def _product_provenance() -> semantics.EvidenceProvenance: + return semantics.EvidenceProvenance( + kind=semantics.ProvenanceKind.PRODUCT_CODE, + source_content_digest="1" * 64, + semantic_pack_digest=None, + semantic_fact_id=None, + fact_match=None, + ) + + +def _fact_provenance() -> semantics.EvidenceProvenance: + reference, _ = stored_artifact(b'{"matched":true}') + return semantics.EvidenceProvenance( + kind=semantics.ProvenanceKind.SEMANTIC_FACT, + source_content_digest="2" * 64, + semantic_pack_digest="3" * 64, + semantic_fact_id="express.route.registration", + fact_match=reference, + ) + + +def _named_fact_provenance(pack: str, fact: str) -> semantics.EvidenceProvenance: + reference, _ = stored_artifact(fact.encode()) + return semantics.EvidenceProvenance( + kind=semantics.ProvenanceKind.SEMANTIC_FACT, + source_content_digest="4" * 64, + semantic_pack_digest=pack, + semantic_fact_id=fact, + fact_match=reference, + ) + + +def _relation( + *, + kind_name: str = "CALL", + status_name: str = "COMPLETE", + target_id: str = "target", + conditions: tuple[str, ...] = (), + reasons: tuple[str, ...] = (), + provenance: semantics.EvidenceProvenance | None = None, +) -> semantics.SemanticRelation: + return semantics.make_relation( + kind=getattr(semantics.RelationKind, kind_name), + role="framework_transition", + source=_node("source", "caller"), + target=_node(target_id, "callee"), + location=_location(), + status=getattr(semantics.RelationStatus, status_name), + conditions=conditions, + reasons=reasons, + provenance=_product_provenance() if provenance is None else provenance, + ) + + +@pytest.mark.parametrize( + "kind_name", + [ + "REGISTRATION", + "INVOCATION", + "CALL", + "RETURN", + "VALUE_TRANSFORMATION", + "OPERATION_EFFECT", + "CONSTRAINT_EFFECT", + ], +) +def test_every_relation_kind_has_a_canonical_identity(kind_name: str) -> None: + relation = _relation(kind_name=kind_name) + same = _relation(kind_name=kind_name) + + assert relation.kind.value == kind_name.lower() + assert relation.relation_id == same.relation_id + assert len(relation.relation_id) == 64 + assert relation.role == "framework_transition" + assert relation.location == _location() + assert relation.provenance == _product_provenance() + + +@pytest.mark.parametrize( + ("status_name", "reasons"), + [ + ("COMPLETE", ()), + ("INCOMPLETE", ("unmet_condition:authenticated",)), + ("UNRESOLVED", ("ambiguous_framework_match",)), + ], +) +def test_relation_status_is_independent_and_canonical(status_name: str, reasons: tuple[str, ...]) -> None: + relation = _relation(status_name=status_name, reasons=reasons) + + assert relation.status.value == status_name.lower() + assert relation.reasons == reasons + assert relation.relation_id == _relation(status_name=status_name, reasons=reasons).relation_id + + +def test_relation_identity_preserves_condition_order_and_rejects_tampering() -> None: + left = _relation(conditions=("registered", "authenticated")) + right = _relation(conditions=("authenticated", "registered")) + + assert left.relation_id != right.relation_id + with pytest.raises(semantics.SemanticContractError, match="relation_id"): + replace(left, relation_id="0" * 64) + + +def test_unmet_declared_conditions_make_relation_incomplete() -> None: + relation = semantics.relation_from_conditions( + kind=semantics.RelationKind.CONSTRAINT_EFFECT, + role="authorization_constraint", + source=_node("request", "request"), + target=_node("resource", "resource"), + location=_location(), + declared_conditions=("authenticated", "owns_resource"), + unmet_conditions=("owns_resource",), + provenance=_product_provenance(), + ) + + assert relation.status is semantics.RelationStatus.INCOMPLETE + assert relation.conditions == ("authenticated", "owns_resource") + assert relation.reasons == ("unmet_condition:owns_resource",) + + +def test_conflicting_complete_candidates_produce_stable_explicit_ambiguity() -> None: + first = _relation(conditions=("framework:v1",)) + second = _relation(conditions=("framework:v2",)) + + forward = semantics.resolve_relation_candidates((first, second)) + reverse = semantics.resolve_relation_candidates((second, first)) + + assert forward.status is semantics.RelationStatus.UNRESOLVED + assert forward.relation_id == reverse.relation_id + assert forward.conditions == ("framework:v1", "framework:v2") + for identity in (first.relation_id, second.relation_id): + assert f"conflicting_complete_candidate:{identity}" in forward.reasons + assert any(reason.startswith("ambiguity_provenance_anchor:") for reason in forward.reasons) + assert forward not in (first, second) + + +def test_distinct_reviewed_fact_candidates_produce_order_independent_ambiguity() -> None: + first_provenance = _named_fact_provenance("5" * 64, "express.route.v1") + second_provenance = _named_fact_provenance("6" * 64, "express.route.v2") + first = _relation(conditions=("framework:v1",), provenance=first_provenance) + second = _relation(conditions=("framework:v2",), provenance=second_provenance) + + forward = semantics.resolve_relation_candidates((first, second)) + reverse = semantics.resolve_relation_candidates((second, first)) + + assert forward.status is semantics.RelationStatus.UNRESOLVED + assert forward.relation_id == reverse.relation_id + assert forward.provenance == reverse.provenance + assert any(reason.startswith("ambiguity_provenance_anchor:") for reason in forward.reasons) + for candidate in (first, second): + assert f"conflicting_complete_candidate:{candidate.relation_id}" in forward.reasons + provenance = candidate.provenance + assert provenance.semantic_pack_digest is not None + assert provenance.semantic_fact_id is not None + assert provenance.fact_match is not None + assert any( + provenance.semantic_pack_digest in reason + and provenance.semantic_fact_id in reason + and provenance.fact_match.content_digest in reason + for reason in forward.reasons + ) + + +def test_candidate_resolution_returns_the_only_complete_candidate() -> None: + complete = _relation() + incomplete = _relation(status_name="INCOMPLETE", reasons=("missing_fact",)) + + assert semantics.resolve_relation_candidates((incomplete, complete)) is complete + + +def test_candidate_resolution_refuses_unrelated_relation_slots() -> None: + with pytest.raises(semantics.SemanticContractError, match="same semantic slot"): + semantics.resolve_relation_candidates((_relation(), _relation(target_id="other"))) + + +def test_product_provenance_forbids_semantic_fact_fields() -> None: + reference, _ = stored_artifact(b"match") + with pytest.raises(semantics.SemanticContractError, match="forbids semantic fact"): + semantics.EvidenceProvenance( + kind=semantics.ProvenanceKind.PRODUCT_CODE, + source_content_digest="1" * 64, + semantic_pack_digest="2" * 64, + semantic_fact_id="fact", + fact_match=reference, + ) + + +@pytest.mark.parametrize("missing", ["semantic_pack_digest", "semantic_fact_id", "fact_match"]) +def test_semantic_fact_provenance_requires_all_fact_identity_fields(missing: str) -> None: + reference, _ = stored_artifact(b"match") + values: dict[str, object] = { + "semantic_pack_digest": "2" * 64, + "semantic_fact_id": "fact", + "fact_match": reference, + } + values[missing] = None + + with pytest.raises(semantics.SemanticContractError, match=missing): + semantics.EvidenceProvenance( + kind=semantics.ProvenanceKind.SEMANTIC_FACT, + source_content_digest="1" * 64, + semantic_pack_digest=cast(str | None, values["semantic_pack_digest"]), + semantic_fact_id=cast(str | None, values["semantic_fact_id"]), + fact_match=cast(canonical.ContentRef | None, values["fact_match"]), + ) + + +def test_semantic_fact_identity_changes_relation_identity() -> None: + product = _relation() + fact = semantics.make_relation( + kind=product.kind, + role=product.role, + source=product.source, + target=product.target, + location=product.location, + status=product.status, + conditions=product.conditions, + reasons=product.reasons, + provenance=_fact_provenance(), + ) + + assert fact.relation_id != product.relation_id + + +@pytest.mark.parametrize( + ("kind_name", "value"), + [ + ("TEXT", "query"), + ("INTEGER", 7), + ("NUMBER", 2.5), + ("BOOLEAN", True), + ("TEXT_TUPLE", ("read", "write")), + ("SOURCE_LOCATION", {"path": "src/app.py", "line": 7, "column": 3, "symbol": "handle"}), + ], +) +def test_observation_fields_have_closed_typed_values(kind_name: str, value: object) -> None: + field = semantics.FieldValue(name="observed", kind=getattr(semantics.FieldKind, kind_name), value=cast(canonical.JsonValue, value)) + observation = semantics.make_observation( + role="operation", + location=_location(), + fields=(field,), + status=semantics.RelationStatus.COMPLETE, + provenance=_product_provenance(), + ) + + assert observation.fields == (field,) + assert ( + observation.observation_id + == semantics.make_observation( + role="operation", + location=_location(), + fields=(field,), + status=semantics.RelationStatus.COMPLETE, + provenance=_product_provenance(), + ).observation_id + ) + + +def test_field_value_defensively_freezes_caller_mapping() -> None: + caller_value: dict[str, canonical.JsonValue] = { + "path": "src/original.py", + "line": 3, + "column": None, + "symbol": "original", + } + field = semantics.FieldValue("location", semantics.FieldKind.SOURCE_LOCATION, caller_value) + observation = semantics.make_observation( + role="operation", + location=_location(), + fields=(field,), + status=semantics.RelationStatus.COMPLETE, + provenance=_product_provenance(), + ) + + caller_value["path"] = "src/mutated.py" + exposed = cast(MutableMapping[str, canonical.JsonValue], field.value) + with pytest.raises(TypeError): + exposed["path"] = "src/also-mutated.py" + + frozen_value = cast(Mapping[str, canonical.JsonValue], field.value) + assert frozen_value["path"] == "src/original.py" + assert ( + observation.observation_id + == semantics.make_observation( + role="operation", + location=_location(), + fields=(field,), + status=semantics.RelationStatus.COMPLETE, + provenance=_product_provenance(), + ).observation_id + ) + + +def test_observation_preserves_field_order_and_all_statuses() -> None: + first = semantics.FieldValue("first", semantics.FieldKind.TEXT, "one") + second = semantics.FieldValue("second", semantics.FieldKind.INTEGER, 2) + identities = [] + for status in semantics.RelationStatus: + observation = semantics.make_observation( + role="operation", + location=_location(), + fields=(first, second), + status=status, + provenance=_product_provenance(), + ) + reversed_fields = semantics.make_observation( + role="operation", + location=_location(), + fields=(second, first), + status=status, + provenance=_product_provenance(), + ) + assert observation.observation_id != reversed_fields.observation_id + identities.append(observation.observation_id) + assert len(set(identities)) == 3 + + +def test_observation_refuses_duplicate_field_names() -> None: + duplicate = semantics.FieldValue("same", semantics.FieldKind.TEXT, "value") + with pytest.raises(semantics.SemanticContractError, match="duplicate field"): + semantics.make_observation( + role="operation", + location=_location(), + fields=(duplicate, duplicate), + status=semantics.RelationStatus.COMPLETE, + provenance=_product_provenance(), + ) + + +@pytest.mark.parametrize( + ("kind_name", "value"), + [ + ("TEXT", 1), + ("INTEGER", True), + ("NUMBER", float("nan")), + ("BOOLEAN", "true"), + ("TEXT_TUPLE", ("valid", 2)), + ("SOURCE_LOCATION", {"path": "src/app.py"}), + ], +) +def test_field_kind_refuses_wrong_runtime_shape(kind_name: str, value: object) -> None: + with pytest.raises(semantics.SemanticContractError, match="value"): + semantics.FieldValue("observed", getattr(semantics.FieldKind, kind_name), cast(canonical.JsonValue, value)) + + +@pytest.mark.parametrize( + ("path", "line", "column", "symbol"), + [("", 1, None, None), ("src/app.py", 0, None, None), ("src/app.py", 1, -1, None), ("src/app.py", 1, None, "")], +) +def test_source_location_refuses_invalid_runtime_values(path: str, line: int, column: int | None, symbol: str | None) -> None: + with pytest.raises(semantics.SemanticContractError): + semantics.SourceLocation(path=path, line=line, column=column, symbol=symbol) + + +def test_unknown_enum_values_are_refused_at_runtime() -> None: + class Unknown(Enum): + VALUE = "call" + + with pytest.raises(semantics.SemanticContractError, match="RelationKind"): + semantics.make_relation( + kind=cast(semantics.RelationKind, Unknown.VALUE), + role="role", + source=_node("source", "source"), + target=_node("target", "target"), + location=_location(), + status=semantics.RelationStatus.COMPLETE, + conditions=(), + reasons=(), + provenance=_product_provenance(), + ) + + +def test_semantic_contracts_are_frozen() -> None: + relation = _relation() + with pytest.raises(FrozenInstanceError): + relation.__setattr__("role", "mutated")