From ec6c8f7f3ecaabbc6f97d15956dcbaa544e31bdc Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Sun, 1 Nov 2026 17:42:05 +0100 Subject: [PATCH] feat(access-control-prover): match reviewed entry facts --- .kiro/specs/access-control-prover/tasks.md | 2 +- .../mapping/entry_fact_matcher.py | 206 ++++++ .../mapping/entry_fact_structure.py | 391 ++++++++++ tests/test_entry_fact_matcher.py | 686 ++++++++++++++++++ 4 files changed, 1284 insertions(+), 1 deletion(-) create mode 100644 src/openultrasast/mapping/entry_fact_matcher.py create mode 100644 src/openultrasast/mapping/entry_fact_structure.py create mode 100644 tests/test_entry_fact_matcher.py diff --git a/.kiro/specs/access-control-prover/tasks.md b/.kiro/specs/access-control-prover/tasks.md index 3421688..4e398b8 100644 --- a/.kiro/specs/access-control-prover/tasks.md +++ b/.kiro/specs/access-control-prover/tasks.md @@ -373,7 +373,7 @@ its final resolution, dominance, fact-acquisition and measurement semantics with - Done when stored observation and match bytes resolve canonically and byte or reference substitution is refused deterministically. - _Requirements: 13.5_ - _Boundary: Fact-Match Artifact Writer_ -- [ ] 14.7 Implement the authority-bearing structural matcher +- [x] 14.7 Implement the authority-bearing structural matcher - Build typed observations from repository structure for the requirements declared by the selected reviewed fact; do not infer missing structure from a regex or name hit. - Use task 14.6 to persist every observation and completed `FactMatchEvidence`, then pass the immutable match reference with family contract, policies, lineages, custody and read-only resolver to detection-core. - Bind matcher behavior and fact lineage to current custody and the evaluation population so same-population label influence is incomplete or refused. diff --git a/src/openultrasast/mapping/entry_fact_matcher.py b/src/openultrasast/mapping/entry_fact_matcher.py new file mode 100644 index 0000000..0890dc9 --- /dev/null +++ b/src/openultrasast/mapping/entry_fact_matcher.py @@ -0,0 +1,206 @@ +"""Authority-bearing application of reviewed Express structural facts.""" + +from __future__ import annotations + +from dataclasses import dataclass, replace +from pathlib import Path + +from openultrasast.core import canonical +from openultrasast.core.factapply import apply_fact +from openultrasast.core.factmatch import ( + FactMatchDecision, + FactMatchEvidence, + StructuralObservation, + derive_fact_match_digest, + validate_fact_match, +) +from openultrasast.core.factpack import FactPack, SemanticFact, StructuralRequirement, StructuralRequirementKind +from openultrasast.core.semantics import SemanticRelation, SourceLocation +from openultrasast.mapping.entry_fact_structure import ( + CandidateStructure, + EntryStructureError, + ResolvedHandler, + analyze_candidate, + candidates, + read_source_structure, +) +from openultrasast.mapping.fact_authority import EntryFactAuthorityContext +from openultrasast.mapping.fact_match_store import FactMatchArtifactWriter + + +class EntryFactMatcherError(ValueError): + """Reviewed structural evidence could not be safely produced.""" + + +@dataclass(frozen=True) +class _JoinedResolver: + primary: canonical.ContentResolver + fallback: canonical.ContentResolver + + def resolve(self, reference: canonical.ContentRef) -> canonical.ResolvedContent: + try: + return self.primary.resolve(reference) + except canonical.ArtifactResolutionError: + return self.fallback.resolve(reference) + + +@dataclass(frozen=True) +class EntryFactMatchResult: + """Persisted match, lexical evidence, and exact core application result.""" + + pack: FactPack + match: FactMatchEvidence + match_reference: canonical.ContentRef + decision: FactMatchDecision + relations: tuple[SemanticRelation, ...] + resolver: canonical.ContentResolver + resolved_handlers: tuple[ResolvedHandler, ...] + structural_reasons: tuple[str, ...] + + +def _selected(pack: FactPack, fact_id: str) -> SemanticFact: + facts = tuple(fact for fact in pack.facts if fact.fact_id == fact_id) + if len(facts) != 1: + raise EntryFactMatcherError("reviewed fact is missing or ambiguous") + return facts[0] + + +def _pack(authority: EntryFactAuthorityContext, language: str, fact_id: str) -> tuple[FactPack, SemanticFact]: + if not isinstance(authority, EntryFactAuthorityContext): + raise TypeError("authority must be EntryFactAuthorityContext") + packs = tuple(pack for pack in authority.bundle.packs if pack.language == language) + if len(packs) != 1: + raise EntryFactMatcherError("reviewed language pack is missing or ambiguous") + return packs[0], _selected(packs[0], fact_id) + + +def _qualified_target(fact: SemanticFact) -> str: + targets = tuple( + item.expected + for item in fact.structural_requirements + if item.kind is StructuralRequirementKind.QUALIFIED_NAME and isinstance(item.expected, str) + ) + if len(targets) != 1: + raise EntryFactMatcherError("reviewed fact lacks one textual qualified target") + return targets[0] + + +def _observed(requirement: StructuralRequirement, structure: CandidateStructure) -> tuple[canonical.JsonValue, ...]: + if requirement.kind is StructuralRequirementKind.QUALIFIED_NAME: + return (structure.candidate.target,) + if requirement.kind is StructuralRequirementKind.AST_SHAPE: + return (structure.candidate.shape,) + if requirement.kind is StructuralRequirementKind.RECEIVER_TYPE: + return structure.receiver_types + if requirement.kind is StructuralRequirementKind.ARGUMENT_SHAPE and requirement.field == "arguments": + return structure.argument_shapes + if requirement.kind is StructuralRequirementKind.ARGUMENT_SHAPE and requirement.field == "binding": + return structure.binding_shapes + return () + + +def _location(path: str, structure: CandidateStructure) -> SourceLocation: + node = structure.candidate.node + symbol = structure.candidate.target + if len(structure.resolved_handlers) == 1: + symbol = f"{symbol}->{structure.resolved_handlers[0].identity}" + return SourceLocation(path, node.start_point.row + 1, node.start_point.column, symbol) + + +def _persist_observations( + fact: SemanticFact, + structure: CandidateStructure, + location: SourceLocation, + writer: FactMatchArtifactWriter, +) -> tuple[StructuralObservation, ...]: + values: list[StructuralObservation] = [] + pending = canonical.ContentRef("pending", "0" * 64, "application/json", 0, "0" * 64, "0" * 64) + for requirement in fact.structural_requirements: + for observed in _observed(requirement, structure): + provisional = StructuralObservation(requirement.requirement_id, location, observed, pending) + values.append(replace(provisional, evidence=writer.persist_observation(provisional))) + return tuple(values) + + +def _match( + pack: FactPack, + fact: SemanticFact, + location: SourceLocation, + observations: tuple[StructuralObservation, ...], + authority: EntryFactAuthorityContext, +) -> FactMatchEvidence: + bundle = authority.bundle + provisional = FactMatchEvidence( + pack.normalized_content_digest, + fact.fact_id, + location, + observations, + bundle.matcher_revision, + bundle.matcher_behavior_digest, + bundle.matcher_lineage_manifest_digest, + "0" * 64, + ) + return replace(provisional, match_digest=derive_fact_match_digest(provisional)) + + +def _apply( + pack: FactPack, + fact: SemanticFact, + match: FactMatchEvidence, + structure: CandidateStructure, + authority: EntryFactAuthorityContext, + writer: FactMatchArtifactWriter, + evaluation_population_digest: str, +) -> EntryFactMatchResult: + reference = writer.persist_match(match) + resolver = _JoinedResolver(writer.resolver, authority.resolver) + bundle = authority.bundle + core_tail = ( + evaluation_population_digest, + bundle.lineages, + authority.custody, + bundle.custody_policy, + bundle.review_policy, + resolver, + ) + decision = validate_fact_match(pack, fact.fact_id, match, *core_tail) + relations = apply_fact(pack, fact.fact_id, match, reference, bundle.contract, *core_tail) + return EntryFactMatchResult( + pack, + match, + reference, + decision, + relations, + resolver, + structure.resolved_handlers, + structure.reasons, + ) + + +def match_entry_fact( + *, + repository_root: Path, + relative_path: str, + language: str, + fact_id: str, + authority: EntryFactAuthorityContext, + writer: FactMatchArtifactWriter, + evaluation_population_digest: str, +) -> tuple[EntryFactMatchResult, ...]: + """Resolve lexical structure, persist it, and apply one reviewed fact.""" + try: + source = read_source_structure(repository_root, relative_path, language) + except EntryStructureError as exc: + raise EntryFactMatcherError(str(exc)) from exc + pack, fact = _pack(authority, language, fact_id) + results: list[EntryFactMatchResult] = [] + for candidate in candidates(source, _qualified_target(fact)): + structure = analyze_candidate(source, candidate) + location = _location(source.path, structure) + observations = _persist_observations(fact, structure, location, writer) + match = _match(pack, fact, location, observations, authority) + results.append(_apply(pack, fact, match, structure, authority, writer, evaluation_population_digest)) + return tuple(results) + + +__all__ = ["EntryFactMatchResult", "EntryFactMatcherError", "ResolvedHandler", "match_entry_fact"] diff --git a/src/openultrasast/mapping/entry_fact_structure.py b/src/openultrasast/mapping/entry_fact_structure.py new file mode 100644 index 0000000..696ebed --- /dev/null +++ b/src/openultrasast/mapping/entry_fact_structure.py @@ -0,0 +1,391 @@ +"""Lexical JavaScript/TypeScript structure for reviewed Express facts.""" + +from __future__ import annotations + +from collections.abc import Iterator +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from openultrasast.core.semantics import SourceLocation +from openultrasast.semantic.extra import grammar_for + +_LANGUAGES = frozenset({"javascript", "typescript"}) +_SCOPES = frozenset({"program", "statement_block", "function_declaration", "function_expression", "arrow_function", "method_definition"}) +_FUNCTION_VALUES = frozenset({"function_expression", "arrow_function"}) +_HOISTED = frozenset({"express_import", "function", "parameter"}) +_STRINGS = frozenset({"string", "template_string"}) + + +class EntryStructureError(ValueError): + """Repository structure was unavailable or unsafe to interpret.""" + + +@dataclass(frozen=True) +class ResolvedHandler: + identity: str + symbol: str + location: SourceLocation + declaration_kind: str + + +@dataclass(frozen=True) +class _Declaration: + name: str + node: Any + scope: Any + kind: str + value: Any | None + + +@dataclass(frozen=True) +class _Resolution: + declarations: tuple[_Declaration, ...] + reasons: tuple[str, ...] + + +@dataclass(frozen=True) +class StructuralCandidate: + node: Any + target: str + shape: str + receiver: str | None + arguments: Any + + +@dataclass(frozen=True) +class CandidateStructure: + candidate: StructuralCandidate + receiver_types: tuple[str, ...] + argument_shapes: tuple[str, ...] + binding_shapes: tuple[str, ...] + resolved_handlers: tuple[ResolvedHandler, ...] + reasons: tuple[str, ...] + + +@dataclass(frozen=True) +class SourceStructure: + path: str + data: bytes + root: Any + declarations: tuple[_Declaration, ...] + writes: tuple[Any, ...] + + +def _walk(node: Any) -> Iterator[Any]: + yield node + for child in node.named_children: + yield from _walk(child) + + +def _text(node: Any, data: bytes) -> str: + return data[node.start_byte : node.end_byte].decode("utf-8") + + +def _string_value(node: Any, data: bytes) -> str | None: + if node is None or node.type != "string": + return None + raw = _text(node, data) + if len(raw) < 2 or raw[0] not in {'"', "'"} or raw[-1] != raw[0]: + return None + return raw[1:-1] + + +def _member_target(node: Any, data: bytes) -> tuple[str, str] | None: + if node is None or node.type != "member_expression": + return None + owner = node.child_by_field_name("object") + member = node.child_by_field_name("property") + if owner is None or member is None or owner.type != "identifier": + return None + return _text(owner, data), _text(member, data) + + +def _call_target(call: Any, data: bytes) -> str | None: + function = call.child_by_field_name("function") + if function is None: + return None + if function.type == "identifier": + return _text(function, data) + member = _member_target(function, data) + return ".".join(member) if member is not None else None + + +def _scope(node: Any) -> Any: + parent = node.parent + while parent is not None: + if parent.type in _SCOPES: + return parent + parent = parent.parent + raise EntryStructureError("repository source declaration has no lexical scope") + + +def _import_declarations(root: Any, data: bytes) -> list[_Declaration]: + values: list[_Declaration] = [] + for node in root.named_children: + if node.type != "import_statement" or _string_value(node.child_by_field_name("source"), data) != "express": + continue + clause = next((item for item in node.named_children if item.type == "import_clause"), None) + if clause is None: + continue + for name in clause.named_children: + if name.type == "identifier": + values.append(_Declaration(_text(name, data), name, root, "express_import", None)) + return values + + +def _is_express_require(value: Any, data: bytes) -> bool: + if value is None or value.type != "call_expression" or _call_target(value, data) != "require": + return False + arguments = value.child_by_field_name("arguments") + return bool(arguments and len(arguments.named_children) == 1 and _string_value(arguments.named_children[0], data) == "express") + + +def _variable_declaration(node: Any, data: bytes) -> _Declaration | None: + name = node.child_by_field_name("name") + if name is None or name.type != "identifier": + return None + value = node.child_by_field_name("value") + kind = "express_require" if _is_express_require(value, data) else "variable" + if getattr(value, "type", None) in _FUNCTION_VALUES: + kind = "handler_variable" + return _Declaration(_text(name, data), name, _scope(node), kind, value) + + +def _parameter_declarations(node: Any, data: bytes) -> list[_Declaration]: + parameters = node.child_by_field_name("parameters") + if parameters is None: + return [] + values: list[_Declaration] = [] + for item in parameters.named_children: + name = item if item.type == "identifier" else item.child_by_field_name("pattern") + if name is not None and name.type == "identifier": + values.append(_Declaration(_text(name, data), name, node, "parameter", None)) + return values + + +def _declarations(root: Any, data: bytes) -> tuple[_Declaration, ...]: + values = _import_declarations(root, data) + for node in _walk(root): + if node.type == "variable_declarator": + declaration = _variable_declaration(node, data) + if declaration is not None: + values.append(declaration) + if node.type == "function_declaration": + name = node.child_by_field_name("name") + if name is not None: + values.append(_Declaration(_text(name, data), name, _scope(node), "function", node)) + if node.type in {"function_declaration", "function_expression", "arrow_function", "method_definition"}: + values.extend(_parameter_declarations(node, data)) + return tuple(values) + + +def _writes(root: Any) -> tuple[Any, ...]: + kinds = {"assignment_expression", "augmented_assignment_expression", "update_expression"} + return tuple(node for node in _walk(root) if node.type in kinds) + + +def read_source_structure(repository_root: Path, relative_path: str, language: str) -> SourceStructure: + if language not in _LANGUAGES: + raise EntryStructureError("repository source language is unsupported") + root_path = Path(repository_root).resolve() + source_path = (root_path / relative_path).resolve() + try: + normalized = source_path.relative_to(root_path).as_posix() + data = source_path.read_bytes() + except (OSError, ValueError) as exc: + raise EntryStructureError("repository source is unreadable or outside its root") from exc + if not data: + raise EntryStructureError("repository source is empty") + grammar = grammar_for(language) + if grammar is None: + raise EntryStructureError("repository source parser is unavailable") + try: + from tree_sitter import Parser + + tree = Parser(grammar).parse(data) + except Exception as exc: + raise EntryStructureError("repository source parse failed") from exc + if tree.root_node.has_error: + raise EntryStructureError("repository source parse failed") + return SourceStructure(normalized, data, tree.root_node, _declarations(tree.root_node, data), _writes(tree.root_node)) + + +def candidates(source: SourceStructure, target: str) -> tuple[StructuralCandidate, ...]: + values: list[StructuralCandidate] = [] + for node in _walk(source.root): + if node.type != "call_expression" or _call_target(node, source.data) != target: + continue + function = node.child_by_field_name("function") + member = _member_target(function, source.data) + values.append( + StructuralCandidate( + node, + target, + "member_call" if member is not None else "factory_call", + member[0] if member is not None else None, + node.child_by_field_name("arguments"), + ) + ) + return tuple(values) + + +def _visible_scopes(node: Any) -> tuple[Any, ...]: + scopes: list[Any] = [] + current = node.parent + while current is not None: + if current.type in _SCOPES: + scopes.append(current) + current = current.parent + return tuple(scopes) + + +def _raw_resolution(source: SourceStructure, use: Any, name: str) -> tuple[_Declaration, ...]: + for scope in _visible_scopes(use): + matches = tuple(item for item in source.declarations if item.name == name and item.scope == scope) + if matches: + live = tuple(item for item in matches if item.kind in _HOISTED or item.node.start_byte < use.start_byte) + return live or matches + return () + + +def _write_name(write: Any, data: bytes) -> str | None: + target = write.child_by_field_name("left") or write.child_by_field_name("argument") + return _text(target, data) if target is not None and target.type == "identifier" else None + + +def _is_reassigned(source: SourceStructure, use: Any, declaration: _Declaration) -> bool: + for write in source.writes: + if write.start_byte >= use.start_byte or _write_name(write, source.data) != declaration.name: + continue + resolved = _raw_resolution(source, write, declaration.name) + if len(resolved) == 1 and resolved[0] == declaration: + return True + return False + + +def _resolve(source: SourceStructure, use: Any, name: str, role: str) -> _Resolution: + nearest = _raw_resolution(source, use, name) + reasons: list[str] = [] + if not nearest: + reasons.append(f"{role}_binding_missing:{name}") + if len(nearest) > 1: + reasons.append(f"{role}_binding_ambiguous:{name}") + if len(nearest) == 1 and nearest[0].kind not in _HOISTED and nearest[0].node.start_byte > use.start_byte: + reasons.append(f"{role}_binding_not_live:{name}") + if len(nearest) == 1 and _is_reassigned(source, use, nearest[0]): + reasons.append(f"{role}_binding_reassigned:{name}") + return _Resolution(nearest, tuple(sorted(set(reasons)))) + + +def _express_binding(source: SourceStructure, use: Any, name: str) -> tuple[bool, tuple[str, ...]]: + resolved = _resolve(source, use, name, "express") + reasons = list(resolved.reasons) + authoritative = len(resolved.declarations) == 1 and resolved.declarations[0].kind in {"express_import", "express_require"} + if resolved.declarations and not authoritative: + reasons.append(f"express_binding_not_authoritative:{name}") + return authoritative and not reasons, tuple(sorted(set(reasons))) + + +def _receiver_kind(source: SourceStructure, declaration: _Declaration) -> tuple[str | None, tuple[str, ...]]: + value = declaration.value + if value is None or value.type != "call_expression": + return None, (f"receiver_factory_not_authoritative:{declaration.name}",) + arguments = value.child_by_field_name("arguments") + if arguments is None or arguments.named_children: + return None, (f"receiver_factory_not_authoritative:{declaration.name}",) + target = _call_target(value, source.data) + if target is None: + return None, (f"receiver_factory_not_authoritative:{declaration.name}",) + express_name = target.removesuffix(".Router") if target.endswith(".Router") else target + authoritative, reasons = _express_binding(source, value, express_name) + if not authoritative: + return None, reasons + return ("express_router" if target.endswith(".Router") else "express_application"), () + + +def _receiver(source: SourceStructure, candidate: StructuralCandidate) -> tuple[tuple[str, ...], tuple[str, ...]]: + if candidate.receiver is None: + return (), () + resolved = _resolve(source, candidate.node, candidate.receiver, "receiver") + if any("reassigned" in item or "not_live" in item for item in resolved.reasons): + return (), resolved.reasons + kinds: list[str] = [] + reasons = list(resolved.reasons) + for declaration in resolved.declarations: + kind, extra = _receiver_kind(source, declaration) + if kind is not None: + kinds.append(kind) + reasons.extend(extra) + normalized = tuple(sorted(set(reasons))) + return (tuple(kinds) if len(kinds) == len(resolved.declarations) else ()), normalized + + +def _handler(source: SourceStructure, node: Any) -> tuple[tuple[ResolvedHandler, ...], tuple[str, ...]]: + if node.type in _FUNCTION_VALUES: + location = SourceLocation(source.path, node.start_point.row + 1, node.start_point.column, "") + identity = f"@{source.path}:{location.line}:{location.column}" + return (ResolvedHandler(identity, "", location, node.type),), () + if node.type != "identifier": + return (), ("handler_shape_unsupported",) + name = _text(node, source.data) + resolved = _resolve(source, node, name, "handler") + if any("reassigned" in item or "not_live" in item for item in resolved.reasons): + return (), resolved.reasons + handlers: list[ResolvedHandler] = [] + reasons = list(resolved.reasons) + for declaration in resolved.declarations: + if declaration.kind not in {"function", "handler_variable"}: + reasons.append(f"handler_binding_not_repository_defined:{name}") + continue + location = SourceLocation( + source.path, + declaration.node.start_point.row + 1, + declaration.node.start_point.column, + name, + ) + identity = f"{name}@{source.path}:{location.line}:{location.column}" + handlers.append(ResolvedHandler(identity, name, location, declaration.kind)) + normalized = tuple(sorted(set(reasons))) + return (tuple(handlers) if len(handlers) == len(resolved.declarations) else ()), normalized + + +def _arguments( + source: SourceStructure, candidate: StructuralCandidate +) -> tuple[tuple[str, ...], tuple[ResolvedHandler, ...], tuple[str, ...]]: + arguments = candidate.arguments + if arguments is None or len(arguments.named_children) < 2: + return (), (), ("route_arguments_incomplete",) + if arguments.named_children[0].type not in _STRINGS: + return (), (), ("route_path_not_literal",) + handlers, reasons = _handler(source, arguments.named_children[-1]) + return tuple("route_path_and_handler" for _ in handlers), handlers, reasons + + +def _binding(source: SourceStructure, candidate: StructuralCandidate) -> tuple[tuple[str, ...], tuple[str, ...]]: + parent = candidate.node.parent + if parent is None or parent.type != "variable_declarator" or parent.child_by_field_name("value") != candidate.node: + return (), ("application_binding_missing",) + if candidate.arguments is None or candidate.arguments.named_children: + return (), ("application_factory_arguments_unsupported",) + authoritative, reasons = _express_binding(source, candidate.node, candidate.target) + return (("application_instance_binding",) if authoritative else ()), reasons + + +def analyze_candidate(source: SourceStructure, candidate: StructuralCandidate) -> CandidateStructure: + receiver_types, receiver_reasons = _receiver(source, candidate) + argument_shapes, handlers, argument_reasons = _arguments(source, candidate) if candidate.receiver else ((), (), ()) + binding_shapes, binding_reasons = _binding(source, candidate) if candidate.receiver is None else ((), ()) + reasons = tuple(sorted(set((*receiver_reasons, *argument_reasons, *binding_reasons)))) + return CandidateStructure(candidate, receiver_types, argument_shapes, binding_shapes, handlers, reasons) + + +__all__ = [ + "CandidateStructure", + "EntryStructureError", + "ResolvedHandler", + "SourceStructure", + "analyze_candidate", + "candidates", + "read_source_structure", +] diff --git a/tests/test_entry_fact_matcher.py b/tests/test_entry_fact_matcher.py new file mode 100644 index 0000000..de1ee2e --- /dev/null +++ b/tests/test_entry_fact_matcher.py @@ -0,0 +1,686 @@ +"""Authority-bearing structural matching for reviewed Express entry facts.""" + +from __future__ import annotations + +from dataclasses import replace +from pathlib import Path + +import pytest + +from openultrasast.core import canonical, provenance +from openultrasast.core.factapply import FactApplicationError, apply_fact +from openultrasast.core.factpack import FactPackValidationError +from openultrasast.core.semantics import RelationKind, RelationStatus +from openultrasast.mapping.entry_fact_matcher import match_entry_fact +from openultrasast.mapping.entry_packs import load_installed_entry_fact_authority +from openultrasast.mapping.fact_authority import EntryFactAuthorityContext +from openultrasast.mapping.fact_match_store import FactMatchArtifactWriter + + +class _UnusedCustody: + def current_log(self, population_digest: str) -> provenance.PopulationCustodyLog: + raise AssertionError(f"authority load read custody: {population_digest}") + + +class _LogsCustody: + def __init__( + self, + logs: tuple[provenance.PopulationCustodyLog, ...], + policy: provenance.PopulationCustodyPolicy, + ) -> None: + self._logs = {log.population_digest: log for log in logs} + self._policy = policy + + def current_log(self, population_digest: str) -> provenance.PopulationCustodyLog: + try: + return self._logs[population_digest] + except KeyError as exc: + raise provenance.PopulationCustodyError("population custody log not found") from exc + + def append(self, request: provenance.PopulationUseRequest) -> provenance.PopulationUseReceipt: + prior = self.current_log(request.population_digest) + custodian = self._policy.reviewed_custodian_digests[0] + receipt = provenance.record_population_use(request, prior, custodian, self._policy) + self._logs[request.population_digest] = provenance.append_population_use(prior, receipt, self._policy) + return receipt + + +def _authority() -> tuple[EntryFactAuthorityContext, str, _LogsCustody]: + loaded = load_installed_entry_fact_authority(_UnusedCustody()) + bundle = loaded.bundle + custodian = bundle.custody_policy.reviewed_custodian_digests[0] + logs: dict[str, provenance.PopulationCustodyLog] = {} + for lineage in bundle.lineages: + for use in lineage.uses: + prior = logs.setdefault( + use.source_population_digest, + provenance.empty_population_custody_log(use.source_population_digest, custodian), + ) + request = provenance.PopulationUseRequest( + use.source_population_digest, + lineage.arm_id, + use.purpose, + use.labels_accessed, + use.influenced_artifact_digest, + prior.log_digest, + ) + receipt = provenance.record_population_use(request, prior, custodian, bundle.custody_policy) + assert receipt.receipt_digest == use.population_use_receipt_digest + logs[use.source_population_digest] = provenance.append_population_use(prior, receipt, bundle.custody_policy) + evaluation = canonical.content_digest("untouched-entry-matcher-evaluation") + logs[evaluation] = provenance.empty_population_custody_log(evaluation, custodian) + custody = _LogsCustody(tuple(logs.values()), bundle.custody_policy) + return replace(loaded, custody=custody), evaluation, custody + + +def _writer(tmp_path: Path, context: EntryFactAuthorityContext) -> FactMatchArtifactWriter: + return FactMatchArtifactWriter( + tmp_path / "matches", + context.bundle.review_policy.trusted_store_identities[0], + canonical.content_digest("retain-entry-fact-match-tests"), + ) + + +def _source(tmp_path: Path, name: str, content: str) -> tuple[Path, str]: + root = tmp_path / "repository" + path = root / name + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(content, encoding="utf-8") + assert path.stat().st_size > 0 # Prove the matcher instrument receives readable input. + return root, name + + +@pytest.mark.parametrize("language,suffix", (("javascript", "js"), ("typescript", "ts"))) +def test_complete_route_structure_persists_and_applies_declared_relation( + tmp_path: Path, + language: str, + suffix: str, +) -> None: + context, evaluation, _ = _authority() + root, relative = _source( + tmp_path, + f"src/routes.{suffix}", + 'import express from "express";\n' + "const app = express();\n" + 'app.get("/admin", adminHandler);\n' + "function adminHandler(req, res) { res.sendStatus(204); }\n", + ) + writer = _writer(tmp_path, context) + + results = match_entry_fact( + repository_root=root, + relative_path=relative, + language=language, + fact_id="entry:express:app-route:get", + authority=context, + writer=writer, + evaluation_population_digest=evaluation, + ) + + assert len(results) == 1 + result = results[0] + assert result.decision.status is RelationStatus.COMPLETE + assert tuple(observation.requirement_id for observation in result.match.observations) == ( + "qualified_target", + "call_shape", + "receiver", + "arguments", + ) + assert tuple((relation.kind, relation.role, relation.status) for relation in result.relations) == ( + (RelationKind.REGISTRATION, "entry_registration", RelationStatus.COMPLETE), + ) + assert writer.resolver.resolve(result.match_reference).content + assert all(writer.resolver.resolve(item.evidence).content for item in result.match.observations) + + +def test_factory_binding_applies_only_from_verified_express_import(tmp_path: Path) -> None: + context, evaluation, _ = _authority() + root, relative = _source( + tmp_path, + "src/app.js", + 'import express from "express";\nconst app = express();\n', + ) + + result = match_entry_fact( + repository_root=root, + relative_path=relative, + language="javascript", + fact_id="entry:express:express-app", + authority=context, + writer=_writer(tmp_path, context), + evaluation_population_digest=evaluation, + )[0] + + assert result.decision.status is RelationStatus.COMPLETE + assert tuple(item.observed for item in result.match.observations) == ( + "express", + "factory_call", + "application_instance_binding", + ) + + untrusted_root, untrusted_relative = _source( + tmp_path, + "untrusted/local.js", + "function express() {}\nconst app = express();\n", + ) + untrusted = match_entry_fact( + repository_root=untrusted_root, + relative_path=untrusted_relative, + language="javascript", + fact_id="entry:express:express-app", + authority=context, + writer=_writer(tmp_path / "untrusted", context), + evaluation_population_digest=evaluation, + )[0] + assert untrusted.decision.status is RelationStatus.INCOMPLETE + assert "missing_observation:binding" in untrusted.decision.reasons + assert "express_binding_not_authoritative:express" in untrusted.structural_reasons + + +@pytest.mark.parametrize( + "source,missing_reason,structural_reason", + ( + ( + 'app.get("/admin", adminHandler);\nfunction adminHandler(req, res) {}\n', + "missing_observation:receiver", + "receiver_binding_missing:app", + ), + ( + 'import express from "express";\nconst app = express();\napp.get("/admin");\n', + "missing_observation:arguments", + "route_arguments_incomplete", + ), + ( + 'import express from "express";\nconst app = express();\napp.get("/admin", adminHandler);\n', + "missing_observation:arguments", + "handler_binding_missing:adminHandler", + ), + ), +) +def test_name_only_and_partial_structure_cannot_complete( + tmp_path: Path, + source: str, + missing_reason: str, + structural_reason: str, +) -> None: + context, evaluation, _ = _authority() + root, relative = _source(tmp_path, "src/routes.js", source) + + result = match_entry_fact( + repository_root=root, + relative_path=relative, + language="javascript", + fact_id="entry:express:app-route:get", + authority=context, + writer=_writer(tmp_path, context), + evaluation_population_digest=evaluation, + )[0] + + assert result.decision.status is RelationStatus.INCOMPLETE + assert missing_reason in result.decision.reasons + assert structural_reason in result.structural_reasons + assert all(relation.status is not RelationStatus.COMPLETE for relation in result.relations) + + +def test_ambiguous_receiver_bindings_are_unresolved(tmp_path: Path) -> None: + context, evaluation, _ = _authority() + root, relative = _source( + tmp_path, + "src/routes.js", + 'import express from "express";\n' + "const app = express();\n" + "const app = express.Router();\n" + 'app.get("/admin", adminHandler);\n' + "function adminHandler(req, res) {}\n", + ) + + result = match_entry_fact( + repository_root=root, + relative_path=relative, + language="javascript", + fact_id="entry:express:app-route:get", + authority=context, + writer=_writer(tmp_path, context), + evaluation_population_digest=evaluation, + )[0] + + assert result.decision.status is RelationStatus.UNRESOLVED + assert "duplicate_observation:receiver" in result.decision.reasons + assert "conflicting_observation:receiver" in result.decision.reasons + assert "receiver_binding_ambiguous:app" in result.structural_reasons + + +def test_substituted_match_reference_is_refused_by_core(tmp_path: Path) -> None: + context, evaluation, _ = _authority() + root, relative = _source( + tmp_path, + "src/routes.js", + 'import express from "express";\nconst app = express();\napp.get("/admin", adminHandler);\nfunction adminHandler(req, res) {}\n', + ) + result = match_entry_fact( + repository_root=root, + relative_path=relative, + language="javascript", + fact_id="entry:express:app-route:get", + authority=context, + writer=_writer(tmp_path, context), + evaluation_population_digest=evaluation, + )[0] + + substituted = replace(result.match_reference, storage_receipt_digest="f" * 64) + with pytest.raises(FactApplicationError, match="match reference cannot be validated"): + apply_fact( + result.pack, + result.match.fact_id, + result.match, + substituted, + context.bundle.contract, + evaluation, + context.bundle.lineages, + context.custody, + context.bundle.custody_policy, + context.bundle.review_policy, + result.resolver, + ) + + +def test_current_same_population_label_influence_cannot_complete(tmp_path: Path) -> None: + context, _, custody = _authority() + matcher_lineage = next( + item for item in context.bundle.lineages if item.manifest_digest == context.bundle.matcher_lineage_manifest_digest + ) + population = matcher_lineage.uses[0].source_population_digest + current = custody.current_log(population) + custody.append( + provenance.PopulationUseRequest( + population, + matcher_lineage.arm_id, + provenance.LineagePurpose.MATCHING_BEHAVIOR, + True, + context.bundle.matcher_behavior_digest, + current.log_digest, + ) + ) + root, relative = _source( + tmp_path, + "src/routes.js", + 'import express from "express";\nconst app = express();\napp.get("/admin", adminHandler);\nfunction adminHandler(req, res) {}\n', + ) + + result = match_entry_fact( + repository_root=root, + relative_path=relative, + language="javascript", + fact_id="entry:express:app-route:get", + authority=context, + writer=_writer(tmp_path, context), + evaluation_population_digest=population, + )[0] + + assert result.decision.status is RelationStatus.UNRESOLVED + assert f"protected_population_labels:{population}" in result.decision.reasons + assert all(relation.status is not RelationStatus.COMPLETE for relation in result.relations) + + +def test_current_same_population_fact_label_influence_is_refused(tmp_path: Path) -> None: + context, _, custody = _authority() + pack = next(item for item in context.bundle.packs if item.language == "javascript") + fact_lineage = next(item for item in context.bundle.lineages if item.subject_digest == pack.normalized_content_digest) + population = fact_lineage.uses[0].source_population_digest + current = custody.current_log(population) + custody.append( + provenance.PopulationUseRequest( + population, + fact_lineage.arm_id, + provenance.LineagePurpose.SEMANTIC_FACT, + True, + pack.normalized_content_digest, + current.log_digest, + ) + ) + root, relative = _source( + tmp_path, + "src/routes.js", + 'import express from "express";\nconst app = express();\napp.get("/admin", adminHandler);\nfunction adminHandler(req, res) {}\n', + ) + + with pytest.raises(FactPackValidationError, match="semantic fact lineage is contaminated.*protected_population_labels"): + match_entry_fact( + repository_root=root, + relative_path=relative, + language="javascript", + fact_id="entry:express:app-route:get", + authority=context, + writer=_writer(tmp_path, context), + evaluation_population_digest=population, + ) + + +def test_matcher_refuses_empty_unreadable_or_outside_source(tmp_path: Path) -> None: + context, evaluation, _ = _authority() + root = tmp_path / "repository" + root.mkdir() + (root / "empty.js").touch() + + for relative in ("empty.js", "missing.js", "../outside.js"): + with pytest.raises(ValueError, match="repository source"): + match_entry_fact( + repository_root=root, + relative_path=relative, + language="javascript", + fact_id="entry:express:app-route:get", + authority=context, + writer=_writer(tmp_path, context), + evaluation_population_digest=evaluation, + ) + + +@pytest.mark.parametrize("language,suffix", (("javascript", "js"), ("typescript", "ts"))) +def test_function_local_require_does_not_authorize_top_level_local_express( + tmp_path: Path, + language: str, + suffix: str, +) -> None: + context, evaluation, _ = _authority() + root, relative = _source( + tmp_path, + f"src/local.{suffix}", + "function unrelated() { const express = require('express'); return express; }\n" + "function express() {}\n" + "const app = express();\n" + "function handler(req, res) {}\n" + "app.get('/x', handler);\n", + ) + + result = match_entry_fact( + repository_root=root, + relative_path=relative, + language=language, + fact_id="entry:express:app-route:get", + authority=context, + writer=_writer(tmp_path, context), + evaluation_population_digest=evaluation, + )[0] + + assert result.decision.status is RelationStatus.INCOMPLETE + assert "express_binding_not_authoritative:express" in result.structural_reasons + + +@pytest.mark.parametrize("language,suffix", (("javascript", "js"), ("typescript", "ts"))) +def test_receiver_reassignment_cannot_complete(tmp_path: Path, language: str, suffix: str) -> None: + context, evaluation, _ = _authority() + root, relative = _source( + tmp_path, + f"src/reassigned.{suffix}", + "import express from 'express';\n" + "let app = express();\n" + "app = express.Router();\n" + "function handler(req, res) {}\n" + "app.get('/x', handler);\n", + ) + + result = match_entry_fact( + repository_root=root, + relative_path=relative, + language=language, + fact_id="entry:express:app-route:get", + authority=context, + writer=_writer(tmp_path, context), + evaluation_population_digest=evaluation, + )[0] + + assert result.decision.status is RelationStatus.INCOMPLETE + assert "receiver_binding_reassigned:app" in result.structural_reasons + + +@pytest.mark.parametrize("language,suffix", (("javascript", "js"), ("typescript", "ts"))) +def test_handler_reassignment_cannot_complete(tmp_path: Path, language: str, suffix: str) -> None: + context, evaluation, _ = _authority() + root, relative = _source( + tmp_path, + f"src/handler-reassigned.{suffix}", + "import express from 'express';\n" + "const app = express();\n" + "let handler = (req, res) => {};\n" + "handler = otherHandler;\n" + "app.get('/x', handler);\n", + ) + + result = match_entry_fact( + repository_root=root, + relative_path=relative, + language=language, + fact_id="entry:express:app-route:get", + authority=context, + writer=_writer(tmp_path, context), + evaluation_population_digest=evaluation, + )[0] + + assert result.decision.status is RelationStatus.INCOMPLETE + assert "handler_binding_reassigned:handler" in result.structural_reasons + assert result.resolved_handlers == () + + +@pytest.mark.parametrize("language,suffix", (("javascript", "js"), ("typescript", "ts"))) +def test_duplicate_handler_declarations_are_unresolved(tmp_path: Path, language: str, suffix: str) -> None: + context, evaluation, _ = _authority() + root, relative = _source( + tmp_path, + f"src/duplicate-handler.{suffix}", + "import express from 'express';\n" + "const app = express();\n" + "function handler(req, res) {}\n" + "function handler(req, res) {}\n" + "app.get('/x', handler);\n", + ) + + result = match_entry_fact( + repository_root=root, + relative_path=relative, + language=language, + fact_id="entry:express:app-route:get", + authority=context, + writer=_writer(tmp_path, context), + evaluation_population_digest=evaluation, + )[0] + + assert result.decision.status is RelationStatus.UNRESOLVED + assert "handler_binding_ambiguous:handler" in result.structural_reasons + assert len(result.resolved_handlers) == 2 + + +@pytest.mark.parametrize("language,suffix", (("javascript", "js"), ("typescript", "ts"))) +def test_valid_nested_registration_resolves_exact_handler_declaration( + tmp_path: Path, + language: str, + suffix: str, +) -> None: + context, evaluation, _ = _authority() + root, relative = _source( + tmp_path, + f"src/nested.{suffix}", + "import express from 'express';\n" + "function register() {\n" + " const app = express();\n" + " const nestedHandler = (req, res) => {};\n" + " app.get('/x', nestedHandler);\n" + "}\n", + ) + + result = match_entry_fact( + repository_root=root, + relative_path=relative, + language=language, + fact_id="entry:express:app-route:get", + authority=context, + writer=_writer(tmp_path, context), + evaluation_population_digest=evaluation, + )[0] + + assert result.decision.status is RelationStatus.COMPLETE + assert result.structural_reasons == () + assert len(result.resolved_handlers) == 1 + assert result.resolved_handlers[0].symbol == "nestedHandler" + assert result.resolved_handlers[0].identity == f"nestedHandler@src/nested.{suffix}:4:8" + assert result.resolved_handlers[0].location.path == f"src/nested.{suffix}" + assert result.resolved_handlers[0].location.line == 4 + assert result.match.product_location.symbol == f"app.get->nestedHandler@src/nested.{suffix}:4:8" + + +@pytest.mark.parametrize("language,suffix", (("javascript", "js"), ("typescript", "ts"))) +def test_mixed_receiver_declarations_suppress_authoritative_observation( + tmp_path: Path, + language: str, + suffix: str, +) -> None: + context, evaluation, _ = _authority() + root, relative = _source( + tmp_path, + f"src/mixed-receiver.{suffix}", + "import express from 'express';\nvar app = express();\nvar app = fake;\nfunction handler(req, res) {}\napp.get('/x', handler);\n", + ) + + result = match_entry_fact( + repository_root=root, + relative_path=relative, + language=language, + fact_id="entry:express:app-route:get", + authority=context, + writer=_writer(tmp_path, context), + evaluation_population_digest=evaluation, + )[0] + + assert result.decision.status is RelationStatus.INCOMPLETE + assert "receiver_binding_ambiguous:app" in result.structural_reasons + assert "receiver_factory_not_authoritative:app" in result.structural_reasons + assert not any(item.requirement_id == "receiver" for item in result.match.observations) + + +@pytest.mark.parametrize("language,suffix", (("javascript", "js"), ("typescript", "ts"))) +def test_mixed_handler_declarations_suppress_argument_observation( + tmp_path: Path, + language: str, + suffix: str, +) -> None: + context, evaluation, _ = _authority() + root, relative = _source( + tmp_path, + f"src/mixed-handler.{suffix}", + "import express from 'express';\n" + "const app = express();\n" + "function handler(req, res) {}\n" + "var handler = externalHandler;\n" + "app.get('/x', handler);\n", + ) + + result = match_entry_fact( + repository_root=root, + relative_path=relative, + language=language, + fact_id="entry:express:app-route:get", + authority=context, + writer=_writer(tmp_path, context), + evaluation_population_digest=evaluation, + )[0] + + assert result.decision.status is RelationStatus.INCOMPLETE + assert "handler_binding_ambiguous:handler" in result.structural_reasons + assert "handler_binding_not_repository_defined:handler" in result.structural_reasons + assert result.resolved_handlers == () + assert not any(item.requirement_id == "arguments" for item in result.match.observations) + + +@pytest.mark.parametrize("language,suffix", (("javascript", "js"), ("typescript", "ts"))) +def test_inner_lexical_bindings_cleanly_shadow_outer_bindings(tmp_path: Path, language: str, suffix: str) -> None: + context, evaluation, _ = _authority() + root, relative = _source( + tmp_path, + f"src/shadowed.{suffix}", + "import express from 'express';\n" + "const app = express();\n" + "function handler(req, res) {}\n" + "function register() {\n" + " const express = require('express');\n" + " const app = express();\n" + " const handler = (req, res) => {};\n" + " app.get('/x', handler);\n" + "}\n", + ) + + result = match_entry_fact( + repository_root=root, + relative_path=relative, + language=language, + fact_id="entry:express:app-route:get", + authority=context, + writer=_writer(tmp_path, context), + evaluation_population_digest=evaluation, + )[0] + + assert result.decision.status is RelationStatus.COMPLETE + assert result.structural_reasons == () + assert result.resolved_handlers[0].identity == f"handler@src/shadowed.{suffix}:7:8" + + +@pytest.mark.parametrize("language,suffix", (("javascript", "js"), ("typescript", "ts"))) +def test_post_registration_reassignment_does_not_contaminate_match( + tmp_path: Path, + language: str, + suffix: str, +) -> None: + context, evaluation, _ = _authority() + root, relative = _source( + tmp_path, + f"src/post-write.{suffix}", + "import express from 'express';\n" + "let app = express();\n" + "let handler = (req, res) => {};\n" + "app.get('/x', handler);\n" + "app = express.Router();\n" + "handler = externalHandler;\n", + ) + + result = match_entry_fact( + repository_root=root, + relative_path=relative, + language=language, + fact_id="entry:express:app-route:get", + authority=context, + writer=_writer(tmp_path, context), + evaluation_population_digest=evaluation, + )[0] + + assert result.decision.status is RelationStatus.COMPLETE + assert result.structural_reasons == () + + +@pytest.mark.parametrize("language,suffix", (("javascript", "js"), ("typescript", "ts"))) +def test_post_registration_declaration_initializers_do_not_contaminate_match( + tmp_path: Path, + language: str, + suffix: str, +) -> None: + context, evaluation, _ = _authority() + root, relative = _source( + tmp_path, + f"src/post-initializer.{suffix}", + "import express from 'express';\n" + "var app = express();\n" + "function handler(req, res) {}\n" + "app.get('/x', handler);\n" + "var app = fake;\n" + "var handler = externalHandler;\n", + ) + + result = match_entry_fact( + repository_root=root, + relative_path=relative, + language=language, + fact_id="entry:express:app-route:get", + authority=context, + writer=_writer(tmp_path, context), + evaluation_population_digest=evaluation, + )[0] + + assert result.decision.status is RelationStatus.COMPLETE + assert result.structural_reasons == ()