From 2a44ad938ca5158a9bd9e820dd3f8be731fb474e Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Sat, 31 Oct 2026 17:38:30 +0100 Subject: [PATCH] feat(detection-core): derive bounded verdict outputs --- .kiro/specs/detection-core/tasks.md | 2 +- src/openultrasast/core/verdict_v2.py | 271 +++++++++++++++++++++++++++ tests/test_core_verdict.py | 231 +++++++++++++++++++++++ 3 files changed, 503 insertions(+), 1 deletion(-) create mode 100644 src/openultrasast/core/verdict_v2.py create mode 100644 tests/test_core_verdict.py diff --git a/.kiro/specs/detection-core/tasks.md b/.kiro/specs/detection-core/tasks.md index 6f05abe..d7741a2 100644 --- a/.kiro/specs/detection-core/tasks.md +++ b/.kiro/specs/detection-core/tasks.md @@ -193,7 +193,7 @@ - Refuse constructible “validated” markers or model programs outside the context. - Done when admission and consumer tests exercise the same validator and neither can bypass it with a raw record. - _Requirements: 9.2, 9.3, 9.4, 9.5, 10.3, 10.4, 12.6, 14.1_ -- [ ] 7.3 (P) Derive defect claims, bounded satisfaction, and bounded silence +- [x] 7.3 (P) Derive defect claims, bounded satisfaction, and bounded silence - Require exact change-tied locations and complete evidence for defect claims. - Carry completed question identities and exact scope for satisfaction. - Preserve named coverage reasons and record identities for every undecided question. diff --git a/src/openultrasast/core/verdict_v2.py b/src/openultrasast/core/verdict_v2.py new file mode 100644 index 0000000..1db97fb --- /dev/null +++ b/src/openultrasast/core/verdict_v2.py @@ -0,0 +1,271 @@ +"""Bounded v2 verdict outputs derived from complete evidence records.""" + +from __future__ import annotations + +import re +from collections.abc import Sequence +from dataclasses import dataclass + +from openultrasast.core.evidence import ( + AnalyzedScope, + EvidenceBasis, + EvidenceRecord, + EvidenceValidationError, + NeutralOutcome, + ScopeValidationError, + SourceRegion, + derive_record_id, + validate_scope, +) +from openultrasast.core.family import FamilyContract, FamilyContractError, ResidualQuestion, validate_contract, validate_record +from openultrasast.core.semantics import Observation, RelationStatus, SemanticRelation, SourceLocation + +_DIGEST = re.compile(r"[0-9a-f]{64}") + + +class VerdictContractError(ValueError): + """Validated evidence cannot be represented as a bounded verdict.""" + + +@dataclass(frozen=True) +class DefectClaim: + record: EvidenceRecord + basis: EvidenceBasis + location: SourceLocation + change_anchor: SourceRegion + + def __post_init__(self) -> None: + if not isinstance(self.record, EvidenceRecord) or self.record.decision.outcome is not NeutralOutcome.VIOLATION: + raise VerdictContractError("defect claim requires a violation evidence record") + try: + self.record.__post_init__() + validate_scope(self.record.packet, self.record.packet.scope) + except (TypeError, ValueError) as exc: + raise VerdictContractError(f"defect claim evidence record is invalid: {exc}") from exc + if not isinstance(self.basis, EvidenceBasis) or self.basis is not self.record.decision.basis: + raise VerdictContractError("defect claim basis must equal its evidence decision basis") + if not isinstance(self.location, SourceLocation) or not isinstance(self.change_anchor, SourceRegion): + raise VerdictContractError("defect claim requires an exact location and change anchor") + members: tuple[SemanticRelation | Observation, ...] = ( + *self.record.packet.transitions, + *self.record.packet.observations, + ) + if self.record.packet.completeness_reasons or any(item.status is not RelationStatus.COMPLETE for item in members): + raise VerdictContractError("defect claim requires complete evidence") + if self.location not in tuple(item.location for item in members): + raise VerdictContractError("defect location must name an exact evidence location") + if self.change_anchor not in self.record.decision.bounded_scope.regions: + raise VerdictContractError("defect change anchor must belong to the bounded decision scope") + if not _region_contains(self.change_anchor, self.location): + raise VerdictContractError("defect location is outside its change anchor") + + +@dataclass(frozen=True) +class BoundedSatisfaction: + question_ids: tuple[str, ...] + records: tuple[EvidenceRecord, ...] + scope: AnalyzedScope + + def __post_init__(self) -> None: + _text_tuple(self.question_ids, "satisfaction question") + if ( + not isinstance(self.records, tuple) + or not self.records + or not all(isinstance(record, EvidenceRecord) for record in self.records) + ): + raise VerdictContractError("bounded satisfaction requires evidence records") + if not isinstance(self.scope, AnalyzedScope): + raise VerdictContractError("bounded satisfaction requires an AnalyzedScope") + if any(record.decision.outcome is not NeutralOutcome.SATISFIED for record in self.records): + raise VerdictContractError("bounded satisfaction accepts only satisfied evidence records") + if any(record.decision.bounded_scope != self.scope for record in self.records): + raise VerdictContractError("satisfaction scope must equal every supporting record scope") + + +@dataclass(frozen=True) +class BoundedSilence: + question_ids: tuple[str, ...] + reasons: tuple[str, ...] + evidence_record_digests: tuple[str, ...] + coverage_digest: str + + def __post_init__(self) -> None: + _text_tuple(self.question_ids, "silence question") + _text_tuple(self.reasons, "silence reason") + _digest_tuple(self.evidence_record_digests, "silence evidence record") + _digest(self.coverage_digest, "coverage") + + +@dataclass(frozen=True) +class VerdictOutputs: + defects: tuple[DefectClaim, ...] + satisfactions: tuple[BoundedSatisfaction, ...] + silence: tuple[BoundedSilence, ...] + + def __post_init__(self) -> None: + values = ( + (self.defects, DefectClaim, "defects"), + (self.satisfactions, BoundedSatisfaction, "satisfactions"), + (self.silence, BoundedSilence, "silence"), + ) + for items, item_type, field in values: + if not isinstance(items, tuple) or not all(isinstance(item, item_type) for item in items): + raise VerdictContractError(f"{field} must be a tuple of {item_type.__name__}") + + +def _digest(value: object, field: str) -> None: + if not isinstance(value, str) or _DIGEST.fullmatch(value) is None: + raise VerdictContractError(f"{field} must be a full lowercase SHA-256 digest") + + +def _text_tuple(value: object, field: str) -> None: + if not isinstance(value, tuple) or not value or not all(isinstance(item, str) and item for item in value): + raise VerdictContractError(f"{field} identities must be a nonempty tuple") + if len(set(value)) != len(value): + raise VerdictContractError(f"duplicate {field} identity") + + +def _digest_tuple(value: object, field: str) -> None: + _text_tuple(value, field) + assert isinstance(value, tuple) + for item in value: + _digest(item, field) + + +def _region_contains(region: SourceRegion, location: SourceLocation) -> bool: + return location.path == region.path and region.start_line <= location.line <= region.end_line + + +def _scope_contains(scope: AnalyzedScope, location: SourceLocation) -> bool: + return any(_region_contains(region, location) for region in scope.regions) + + +def _members(record: EvidenceRecord, role: str) -> tuple[SemanticRelation | Observation, ...]: + transitions = tuple(item for item in record.packet.transitions if item.role == role) + observations = tuple(item for item in record.packet.observations if item.role == role) + return (*transitions, *observations) + + +def _question_state(record: EvidenceRecord, question: ResidualQuestion) -> tuple[bool, tuple[str, ...]]: + reasons: set[str] = set() + if record.decision.family_code not in question.allowed_decisions: + reasons.add(f"decision_not_declared_for_question:{record.decision.family_code}") + for role in question.required_roles: + members = _members(record, role) + if not members: + reasons.add(f"missing_required_role:{role}") + elif not any( + item.status is RelationStatus.COMPLETE and _scope_contains(record.decision.bounded_scope, item.location) for item in members + ): + relation_reasons = {reason for item in members if isinstance(item, SemanticRelation) for reason in item.reasons} + complete_outside = any(item.status is RelationStatus.COMPLETE for item in members) + fallback = f"evidence_outside_bounded_scope:{role}" if complete_outside else f"incomplete_required_role:{role}" + reasons.update(relation_reasons or record.packet.completeness_reasons or (fallback,)) + if record.decision.outcome is NeutralOutcome.INDETERMINATE: + reasons.update(record.decision.reasons) + complete = not reasons and record.decision.outcome is not NeutralOutcome.INDETERMINATE + return complete, tuple(sorted(reasons)) + + +def _location(record: EvidenceRecord, question: ResidualQuestion) -> tuple[SourceLocation, SourceRegion]: + candidates = tuple( + item.location for role in question.required_roles for item in _members(record, role) if item.status is RelationStatus.COMPLETE + ) + ordered = sorted(candidates, key=lambda item: (item.path, item.line, -1 if item.column is None else item.column, item.symbol or "")) + for location in ordered: + anchors = tuple(region for region in record.decision.bounded_scope.regions if _region_contains(region, location)) + if anchors: + anchor = min(anchors, key=lambda item: (item.end_line - item.start_line, item.path, item.start_line)) + return location, anchor + raise VerdictContractError("complete defect evidence has no exact location inside its bounded change scope") + + +def _validate_inputs(records: Sequence[EvidenceRecord], contract: FamilyContract, coverage_digest: str) -> tuple[EvidenceRecord, ...]: + _digest(coverage_digest, "coverage") + try: + validate_contract(contract) + except (TypeError, ValueError) as exc: + raise VerdictContractError(f"family contract is invalid: {exc}") from exc + values = tuple(records) + if not values or not all(isinstance(record, EvidenceRecord) for record in values): + raise VerdictContractError("records must be a nonempty sequence of EvidenceRecord values") + identities: set[str] = set() + try: + for record in values: + record.__post_init__() + if record.record_id != derive_record_id(record.packet, record.decision): + raise EvidenceValidationError("record identity cannot be re-derived") + if record.record_id in identities: + raise EvidenceValidationError("duplicate evidence record identity") + identities.add(record.record_id) + validate_scope(record.packet, record.packet.scope) + validate_record(contract, record) + except (TypeError, ValueError, EvidenceValidationError, ScopeValidationError, FamilyContractError) as exc: + raise VerdictContractError(f"evidence record validation failed: {exc}") from exc + return values + + +def _silence( + question: ResidualQuestion, + reasons: set[str], + record_ids: set[str], + coverage_digest: str, +) -> BoundedSilence | None: + if not reasons: + return None + return BoundedSilence( + (question.question_id,), + tuple(sorted(reasons)), + tuple(sorted(record_ids)), + coverage_digest, + ) + + +def derive_verdict_outputs( + records: Sequence[EvidenceRecord], + contract: FamilyContract, + coverage_digest: str, +) -> VerdictOutputs: + """Derive bounded outputs without converting incomplete evidence to safety.""" + validated = _validate_inputs(records, contract, coverage_digest) + defects: dict[str, DefectClaim] = {} + satisfactions: dict[str, set[str]] = {} + silence: list[BoundedSilence] = [] + for question in contract.residual_questions: + reasons: set[str] = set() + undecided: set[str] = set() + for record in validated: + complete, record_reasons = _question_state(record, question) + if not complete: + reasons.update(record_reasons) + undecided.add(record.record_id) + continue + if record.decision.outcome is NeutralOutcome.VIOLATION: + if record.record_id not in defects: + location, anchor = _location(record, question) + defects[record.record_id] = DefectClaim(record, record.decision.basis, location, anchor) + elif record.decision.outcome is NeutralOutcome.SATISFIED: + satisfactions.setdefault(record.record_id, set()).add(question.question_id) + bounded = _silence(question, reasons, undecided, coverage_digest) + if bounded is not None: + silence.append(bounded) + records_by_id = {record.record_id: record for record in validated} + satisfied = tuple( + BoundedSatisfaction(tuple(sorted(question_ids)), (records_by_id[record_id],), records_by_id[record_id].decision.bounded_scope) + for record_id, question_ids in sorted(satisfactions.items()) + ) + return VerdictOutputs( + tuple(defects[identity] for identity in sorted(defects)), + satisfied, + tuple(sorted(silence, key=lambda item: item.question_ids)), + ) + + +__all__ = [ + "BoundedSatisfaction", + "BoundedSilence", + "DefectClaim", + "VerdictContractError", + "VerdictOutputs", + "derive_verdict_outputs", +] diff --git a/tests/test_core_verdict.py b/tests/test_core_verdict.py new file mode 100644 index 0000000..a5da69e --- /dev/null +++ b/tests/test_core_verdict.py @@ -0,0 +1,231 @@ +"""Bounded defect, satisfaction, and silence derivation.""" + +from __future__ import annotations + +import copy +from dataclasses import replace + +import pytest + +from openultrasast.core import canonical, provenance +from openultrasast.core.evidence import ( + EVIDENCE_PACKET_SCHEMA, + AnalyzedScope, + Decision, + EvidenceBasis, + EvidencePacket, + EvidenceRecord, + NeutralOutcome, + SourceRegion, + derive_packet_digest, +) +from openultrasast.core.family import ( + DecisionRule, + FamilyContract, + FieldSpec, + ObjectSpec, + RelationOriginPolicy, + ResidualQuestion, + RoleSpec, + derive_security_object_id, + family_contract_digest, +) +from openultrasast.core.semantics import ( + EvidenceProvenance, + FieldKind, + FieldValue, + ProvenanceKind, + RelationKind, + RelationStatus, + SemanticNode, + SemanticRelation, + SourceLocation, + make_relation, +) +from openultrasast.core.verdict_v2 import DefectClaim, VerdictContractError, derive_verdict_outputs + + +def _digest(value: str) -> str: + return canonical.content_digest(value) + + +def _contract(*, extra_question: bool = False) -> FamilyContract: + questions = [ResidualQuestion("flow_decision", ("flow",), ("bad", "good", "unknown"))] + if extra_question: + questions.append(ResidualQuestion("missing_guard", ("guard",), ("bad", "good", "unknown"))) + return FamilyContract( + "reference-flow", + "1", + EVIDENCE_PACKET_SCHEMA, + ("python",), + ObjectSpec("operation", (FieldSpec("name", FieldKind.TEXT, True),), ("name",)), + ( + RoleSpec("flow", (), (RelationKind.CALL,), RelationOriginPolicy.PRODUCT_OR_REVIEWED_FACT), + RoleSpec("guard", (), (RelationKind.CONSTRAINT_EFFECT,), RelationOriginPolicy.PRODUCT_OR_REVIEWED_FACT), + ), + ( + DecisionRule("bad", NeutralOutcome.VIOLATION, ("flow",), (RelationStatus.COMPLETE,)), + DecisionRule("good", NeutralOutcome.SATISFIED, ("flow",), (RelationStatus.COMPLETE,)), + DecisionRule( + "unknown", + NeutralOutcome.INDETERMINATE, + ("flow",), + (RelationStatus.INCOMPLETE, RelationStatus.UNRESOLVED), + ), + ), + tuple(questions), + ("missing_flow",), + ) + + +def _stamp(contract: FamilyContract) -> provenance.ProvenanceStamp: + return provenance.ProvenanceStamp( + _digest("analyzer"), + family_contract_digest(contract), + _digest("facts"), + _digest("acquisition"), + _digest("configuration"), + _digest("population"), + _digest("change"), + ) + + +def _relation(status: RelationStatus, *, line: int = 12) -> SemanticRelation: + source = SourceLocation("src/app.py", 10, 2, "source") + target = SourceLocation("src/app.py", 11, 2, "target") + return make_relation( + kind=RelationKind.CALL, + role="flow", + source=SemanticNode("source", "source", source), + target=SemanticNode("target", "target", target), + location=SourceLocation("src/app.py", line, 4, "call"), + status=status, + conditions=(), + reasons=() if status is RelationStatus.COMPLETE else ("missing_flow",), + provenance=EvidenceProvenance(ProvenanceKind.PRODUCT_CODE, _digest("source"), None, None, None), + ) + + +def _record(contract: FamilyContract, code: str, status: RelationStatus, *, line: int = 12) -> EvidenceRecord: + relation = _relation(status, line=line) + fields = (FieldValue("name", FieldKind.TEXT, "read"),) + scope = AnalyzedScope("base", "head", _digest("change"), (SourceRegion("src/app.py", 1, 20),)) + reasons = () if status is RelationStatus.COMPLETE else ("missing_flow",) + packet = EvidencePacket( + EVIDENCE_PACKET_SCHEMA, + contract.family_id, + family_contract_digest(contract), + derive_security_object_id(contract.security_object, fields), + "python", + scope, + (relation,), + (), + fields, + reasons, + _stamp(contract), + ) + outcomes = {"bad": NeutralOutcome.VIOLATION, "good": NeutralOutcome.SATISFIED, "unknown": NeutralOutcome.INDETERMINATE} + decision = Decision( + code, + outcomes[code], + EvidenceBasis.DETERMINISTIC, + scope, + reasons, + derive_packet_digest(packet), + provenance.DecisionProvenance(None, None, None, None, None, None), + ) + return EvidenceRecord(packet, decision, "") + + +def test_complete_violation_derives_exact_change_tied_claim() -> None: + contract = _contract() + record = _record(contract, "bad", RelationStatus.COMPLETE) + result = derive_verdict_outputs((record,), contract, _digest("coverage")) + assert result.satisfactions == () and result.silence == () + assert len(result.defects) == 1 + claim = result.defects[0] + assert claim.record is record + assert claim.basis is record.decision.basis + assert claim.location == record.packet.transitions[0].location + assert claim.change_anchor == record.packet.scope.regions[0] + + +def test_complete_satisfaction_carries_question_and_exact_scope() -> None: + contract = _contract() + record = _record(contract, "good", RelationStatus.COMPLETE) + result = derive_verdict_outputs((record,), contract, _digest("coverage")) + assert result.defects == () and result.silence == () + assert result.satisfactions[0].question_ids == ("flow_decision",) + assert result.satisfactions[0].records == (record,) + assert result.satisfactions[0].scope is record.decision.bounded_scope + + +def test_satisfaction_requires_support_inside_its_exact_bounded_scope() -> None: + contract = _contract() + record = _record(contract, "good", RelationStatus.COMPLETE) + narrow = AnalyzedScope("base", "head", _digest("change"), (SourceRegion("src/app.py", 1, 5),)) + changed = replace(record, decision=replace(record.decision, bounded_scope=narrow), record_id="") + result = derive_verdict_outputs((changed,), contract, _digest("coverage")) + assert result.satisfactions == () + assert result.silence[0].reasons == ("evidence_outside_bounded_scope:flow",) + assert result.silence[0].evidence_record_digests == (changed.record_id,) + + +def test_indeterminate_record_preserves_reasons_identity_and_coverage() -> None: + contract = _contract() + record = _record(contract, "unknown", RelationStatus.INCOMPLETE) + coverage = _digest("coverage") + result = derive_verdict_outputs((record,), contract, coverage) + assert result.defects == () and result.satisfactions == () + silence = result.silence[0] + assert silence.question_ids == ("flow_decision",) + assert silence.reasons == ("missing_flow",) + assert silence.evidence_record_digests == (record.record_id,) + assert silence.coverage_digest == coverage + + +def test_mixed_complete_and_incomplete_paths_keep_claim_and_silence() -> None: + contract = _contract() + complete = _record(contract, "bad", RelationStatus.COMPLETE) + incomplete = _record(contract, "unknown", RelationStatus.INCOMPLETE) + result = derive_verdict_outputs((incomplete, complete), contract, _digest("coverage")) + assert tuple(item.record.record_id for item in result.defects) == (complete.record_id,) + assert result.satisfactions == () + assert result.silence[0].evidence_record_digests == (incomplete.record_id,) + + +def test_unanswered_question_is_bounded_silence_not_repository_clean() -> None: + contract = _contract(extra_question=True) + record = _record(contract, "good", RelationStatus.COMPLETE) + result = derive_verdict_outputs((record,), contract, _digest("coverage")) + assert result.satisfactions[0].question_ids == ("flow_decision",) + silence = next(item for item in result.silence if item.question_ids == ("missing_guard",)) + assert silence.reasons == ("missing_required_role:guard",) + assert silence.evidence_record_digests == (record.record_id,) + + +def test_claim_location_outside_change_scope_is_refused() -> None: + contract = _contract() + record = _record(contract, "bad", RelationStatus.COMPLETE, line=99) + with pytest.raises(VerdictContractError, match="scope"): + derive_verdict_outputs((record,), contract, _digest("coverage")) + + +def test_claim_contract_refuses_redirected_location_and_anchor() -> None: + contract = _contract() + record = _record(contract, "bad", RelationStatus.COMPLETE) + with pytest.raises(VerdictContractError, match="evidence location"): + DefectClaim(record, record.decision.basis, SourceLocation("src/app.py", 15, 0, "redirected"), record.packet.scope.regions[0]) + with pytest.raises(VerdictContractError, match="bounded decision scope"): + DefectClaim(record, record.decision.basis, record.packet.transitions[0].location, SourceRegion("src/app.py", 2, 20)) + + +def test_invalid_coverage_identity_and_forged_record_are_refused() -> None: + contract = _contract() + record = _record(contract, "bad", RelationStatus.COMPLETE) + with pytest.raises(VerdictContractError, match="coverage"): + derive_verdict_outputs((record,), contract, "not-a-digest") + forged = copy.copy(record) + object.__setattr__(forged, "record_id", _digest("forged")) + with pytest.raises(VerdictContractError, match="record"): + derive_verdict_outputs((forged,), contract, _digest("coverage"))