diff --git a/.kiro/specs/detection-core/tasks.md b/.kiro/specs/detection-core/tasks.md index 4b800c2..6bec39a 100644 --- a/.kiro/specs/detection-core/tasks.md +++ b/.kiro/specs/detection-core/tasks.md @@ -209,7 +209,7 @@ - Reject opaque, self-asserted, inconclusive, same-side, or mismatched proof records. - Done when only the valid differential fixture yields execution-backed basis and no attachment changes rollout policy. - _Requirements: 2.5, 9.6, 14.2_ -- [ ] 7.5 Integrate validated records, proof, scope, and budget at the core consumer edge +- [x] 7.5 Integrate validated records, proof, scope, and budget at the core consumer edge - Accept raw validation envelopes and call the shared validator before creating any output. - Preserve every bounded-silence reason and reject records or anchors outside the supplied change scope. - Apply optional proof attachments through the proof validator and include all processing in total elapsed time. diff --git a/src/openultrasast/core/consumer_v2.py b/src/openultrasast/core/consumer_v2.py new file mode 100644 index 0000000..ae7be08 --- /dev/null +++ b/src/openultrasast/core/consumer_v2.py @@ -0,0 +1,341 @@ +"""Closed v2 push-consumer edge for validated bounded verdicts.""" + +from __future__ import annotations + +import time +from collections.abc import Sequence +from dataclasses import dataclass, replace + +from openultrasast.core import canonical +from openultrasast.core.budget import BudgetDerivation, BudgetGateFailure, ChangeComplexity, check_budget +from openultrasast.core.evidence import AnalyzedScope, EvidenceRecord, SourceRegion, derive_packet_digest, validate_scope +from openultrasast.core.family import FamilyContract, family_contract_digest +from openultrasast.core.family_catalog import resolve_contract +from openultrasast.core.proof import ProofAttachment, ProofTrustPolicy, attach_differential_proof +from openultrasast.core.provenance import PopulationCustodyPort +from openultrasast.core.semantics import SourceLocation +from openultrasast.core.validation import EvidenceValidationContext, EvidenceValidationInput, validate_evidence_input +from openultrasast.core.verdict_v2 import ( + BoundedSatisfaction, + BoundedSilence, + DefectClaim, + VerdictOutputs, + derive_verdict_outputs, +) + +PUSH_REQUEST_SCHEMA = "openultrasast.detection-core.push-request.v2" +CONSUMER_COVERAGE_SCHEMA = "openultrasast.detection-core.consumer-coverage.v2" +_monotonic_ns = time.monotonic_ns + + +class ConsumerContractError(ValueError): + """A raw push request cannot produce a bounded v2 verdict.""" + + +@dataclass(frozen=True) +class PushRequest: + request_id: str + scope: AnalyzedScope + workload_class: str + complexity: ChangeComplexity + budget: BudgetDerivation + + +@dataclass(frozen=True) +class PushVerdict: + request_digest: str + defects: tuple[DefectClaim, ...] + satisfactions: tuple[BoundedSatisfaction, ...] + silence: tuple[BoundedSilence, ...] + input_record_digests: tuple[str, ...] + complexity: ChangeComplexity + total_reply_ms: int + budget_derivation_digest: str + + +def _content_ref_value(value: canonical.ContentRef) -> canonical.JsonValue: + return { + "artifact_id": value.artifact_id, + "content_digest": value.content_digest, + "media_type": value.media_type, + "byte_length": value.byte_length, + "retention_policy_digest": value.retention_policy_digest, + "storage_receipt_digest": value.storage_receipt_digest, + } + + +def _scope_value(value: AnalyzedScope) -> canonical.JsonValue: + return { + "base_revision": value.base_revision, + "head_revision": value.head_revision, + "change_digest": value.change_digest, + "regions": tuple({"path": region.path, "start_line": region.start_line, "end_line": region.end_line} for region in value.regions), + } + + +def _request_value(request: PushRequest) -> canonical.JsonValue: + complexity = request.complexity + return { + "schema_version": PUSH_REQUEST_SCHEMA, + "scope": _scope_value(request.scope), + "workload_class": request.workload_class, + "complexity": { + "change_id": complexity.change_id, + "workload_class": complexity.workload_class, + "changed_regions": complexity.changed_regions, + "transition_count": complexity.transition_count, + "changed_bytes": complexity.changed_bytes, + "measurement": _content_ref_value(complexity.measurement), + }, + "budget_derivation_digest": request.budget.derivation_digest, + } + + +def derive_push_request_id(request: PushRequest) -> str: + if not isinstance(request, PushRequest): + raise ConsumerContractError("request must be a PushRequest") + return canonical.content_digest(_request_value(request)) + + +def make_push_request( + *, + scope: AnalyzedScope, + workload_class: str, + complexity: ChangeComplexity, + budget: BudgetDerivation, +) -> PushRequest: + provisional = PushRequest("0" * 64, scope, workload_class, complexity, budget) + return PushRequest(derive_push_request_id(provisional), scope, workload_class, complexity, budget) + + +def _validate_request(request: PushRequest) -> None: + if not isinstance(request, PushRequest): + raise ConsumerContractError("request must be a PushRequest") + try: + request.scope.__post_init__() + request.complexity.__post_init__() + request.budget.__post_init__() + except (AttributeError, TypeError, ValueError) as exc: + raise ConsumerContractError(f"push request is invalid: {exc}") from exc + if request.request_id != derive_push_request_id(request): + raise ConsumerContractError("push request identity cannot be re-derived") + if request.workload_class != request.complexity.workload_class or request.workload_class != request.budget.workload_class: + raise ConsumerContractError("push workload class does not match complexity and budget") + if request.complexity.change_id != request.scope.change_digest: + raise ConsumerContractError("complexity change identity does not match request scope") + if request.complexity.changed_regions != len(request.scope.regions): + raise ConsumerContractError("complexity changed-region count does not match request scope") + + +def _validated_records( + values: Sequence[EvidenceValidationInput], + context: EvidenceValidationContext, + custody: PopulationCustodyPort, + resolver: canonical.ContentResolver, +) -> tuple[EvidenceRecord, ...]: + selected = tuple(values) + if not selected: + raise ConsumerContractError("push inputs must be nonempty") + if not all(isinstance(value, EvidenceValidationInput) for value in selected): + raise ConsumerContractError("every push input must be an EvidenceValidationInput") + input_ids = tuple(value.input_digest for value in selected) + if len(set(input_ids)) != len(input_ids): + raise ConsumerContractError("duplicate validation input identity") + try: + return tuple(validate_evidence_input(value, context, custody, resolver) for value in selected) + except (AttributeError, LookupError, TypeError, ValueError) as exc: + raise ConsumerContractError(f"evidence validation failed: {exc}") from exc + + +def _contract(context: EvidenceValidationContext, record: EvidenceRecord) -> FamilyContract: + packet = record.packet + try: + return resolve_contract(context.family_catalog, packet.family_id, packet.family_contract_digest, packet.ecosystem) + except (LookupError, TypeError, ValueError) as exc: + raise ConsumerContractError(f"record family cannot be resolved: {exc}") from exc + + +def _coverage_digest(request: PushRequest, contract: FamilyContract, records: tuple[EvidenceRecord, ...]) -> str: + return canonical.content_digest( + { + "schema_version": CONSUMER_COVERAGE_SCHEMA, + "request_id": request.request_id, + "family_contract_digest": family_contract_digest(contract), + "record_ids": tuple(sorted(record.record_id for record in records)), + } + ) + + +def _derive_outputs( + request: PushRequest, + records: tuple[EvidenceRecord, ...], + context: EvidenceValidationContext, +) -> VerdictOutputs: + grouped: dict[str, tuple[FamilyContract, list[EvidenceRecord]]] = {} + for record in records: + try: + validate_scope(record.packet, request.scope) + except (TypeError, ValueError) as exc: + raise ConsumerContractError(f"record scope is outside the push request: {exc}") from exc + contract = _contract(context, record) + digest = record.packet.family_contract_digest + if digest not in grouped: + grouped[digest] = (contract, []) + grouped[digest][1].append(record) + defects: list[DefectClaim] = [] + satisfactions: list[BoundedSatisfaction] = [] + silence: list[BoundedSilence] = [] + try: + for digest in sorted(grouped): + contract, members = grouped[digest] + selected = tuple(members) + outputs = derive_verdict_outputs(selected, contract, _coverage_digest(request, contract, selected)) + defects.extend(outputs.defects) + satisfactions.extend(outputs.satisfactions) + silence.extend(outputs.silence) + except (TypeError, ValueError) as exc: + raise ConsumerContractError(f"bounded verdict derivation failed: {exc}") from exc + return VerdictOutputs(tuple(defects), tuple(satisfactions), tuple(silence)) + + +def _region_within(allowed: AnalyzedScope, candidate: SourceRegion) -> bool: + return any( + candidate.path == region.path and candidate.start_line >= region.start_line and candidate.end_line <= region.end_line + for region in allowed.regions + ) + + +def _proof_target_index(claims: tuple[DefectClaim, ...]) -> dict[tuple[str, str, SourceLocation], int]: + keys: dict[tuple[str, str, SourceLocation], int] = {} + for index, claim in enumerate(claims): + key = (claim.record.packet.family_id, derive_packet_digest(claim.record.packet), claim.location) + if key in keys: + raise ConsumerContractError("proof target is ambiguous across defect claims") + keys[key] = index + return keys + + +def _apply_attachment( + request: PushRequest, + result: list[DefectClaim], + keys: dict[tuple[str, str, SourceLocation], int], + consumed: set[int], + attachment: ProofAttachment, + trust: ProofTrustPolicy, + resolver: canonical.ContentResolver, +) -> None: + if not isinstance(attachment, ProofAttachment): + raise ConsumerContractError("every proof attachment must use the v2 proof contract") + key = (attachment.family_id, attachment.packet_digest, attachment.location) + if key not in keys: + raise ConsumerContractError("proof attachment does not match one exact defect claim") + index = keys[key] + if index in consumed: + raise ConsumerContractError("multiple proof attachments target one defect claim") + if not _region_within(request.scope, result[index].change_anchor): + raise ConsumerContractError("proof claim anchor is outside the push request scope") + result[index] = attach_differential_proof(result[index], attachment, trust, resolver) + consumed.add(index) + + +def _apply_proofs( + request: PushRequest, + claims: tuple[DefectClaim, ...], + attachments: Sequence[ProofAttachment], + trust: ProofTrustPolicy | None, + resolver: canonical.ContentResolver, +) -> tuple[DefectClaim, ...]: + selected = tuple(attachments) + if not selected: + return claims + if not isinstance(trust, ProofTrustPolicy): + raise ConsumerContractError("nonempty proof attachments require a proof trust policy") + keys = _proof_target_index(claims) + result = list(claims) + consumed: set[int] = set() + try: + for attachment in selected: + _apply_attachment(request, result, keys, consumed, attachment, trust, resolver) + except ConsumerContractError: + raise + except (AttributeError, LookupError, TypeError, ValueError) as exc: + raise ConsumerContractError(f"proof validation failed: {exc}") from exc + return tuple(result) + + +def _total_reply_ms(elapsed_before_entry_ms: int, started_ns: int) -> int: + if isinstance(elapsed_before_entry_ms, bool) or not isinstance(elapsed_before_entry_ms, int) or elapsed_before_entry_ms < 0: + raise ConsumerContractError("elapsed-before-entry measurement must be a non-negative integer") + finished_ns = _monotonic_ns() + ticks = (started_ns, finished_ns) + if any(isinstance(tick, bool) or not isinstance(tick, int) or tick < 0 for tick in ticks): + raise ConsumerContractError("monotonic core timing ticks must be non-negative integers") + if finished_ns < started_ns: + raise ConsumerContractError("monotonic core timing measurement moved backwards") + core_elapsed_ms = (finished_ns - started_ns + 999_999) // 1_000_000 + total_reply_ms = elapsed_before_entry_ms + core_elapsed_ms + if total_reply_ms <= 0: + raise ConsumerContractError("combined elapsed measurement must be a strictly positive integer") + return total_reply_ms + + +def _validate_budget_artifacts(request: PushRequest, resolver: canonical.ContentResolver) -> None: + try: + check_budget(request.complexity, 1, request.budget, resolver) + except (BudgetGateFailure, AttributeError, LookupError, TypeError, ValueError) as exc: + raise ConsumerContractError(f"measured budget validation failed: {exc}") from exc + + +def _enforce_validated_deadline(request: PushRequest, total_reply_ms: int) -> None: + if total_reply_ms > request.budget.budget_ms: + raise ConsumerContractError( + f"measured budget gate failed: budget:{request.complexity.change_id}:" + f"{total_reply_ms}ms>{request.budget.budget_ms}ms " + f"class={request.budget.workload_class} derivation={request.budget.derivation_digest}" + ) + + +def consume_records( + request: PushRequest, + inputs: Sequence[EvidenceValidationInput], + validation_context: EvidenceValidationContext, + custody: PopulationCustodyPort, + elapsed_before_entry_ms: int, + resolver: canonical.ContentResolver, + attachments: Sequence[ProofAttachment] = (), + proof_trust: ProofTrustPolicy | None = None, +) -> PushVerdict: + started_ns = _monotonic_ns() + _validate_request(request) + records = _validated_records(inputs, validation_context, custody, resolver) + transition_count = sum(len(record.packet.transitions) for record in records) + if request.complexity.transition_count != transition_count: + raise ConsumerContractError("complexity transition count does not match validated records") + outputs = _derive_outputs(request, records, validation_context) + defects = _apply_proofs(request, outputs.defects, attachments, proof_trust, resolver) + if not defects and not outputs.satisfactions and not outputs.silence: + raise ConsumerContractError("push verdict cannot be an unbounded clean result") + _validate_budget_artifacts(request, resolver) + provisional = PushVerdict( + request.request_id, + defects, + outputs.satisfactions, + outputs.silence, + tuple(sorted(record.record_id for record in records)), + request.complexity, + 0, + request.budget.derivation_digest, + ) + total_reply_ms = _total_reply_ms(elapsed_before_entry_ms, started_ns) + _enforce_validated_deadline(request, total_reply_ms) + return replace(provisional, total_reply_ms=total_reply_ms) + + +__all__ = [ + "ConsumerContractError", + "PushRequest", + "PushVerdict", + "consume_records", + "derive_push_request_id", + "make_push_request", +] diff --git a/tests/test_core_consumer_v2.py b/tests/test_core_consumer_v2.py new file mode 100644 index 0000000..8ac4f0d --- /dev/null +++ b/tests/test_core_consumer_v2.py @@ -0,0 +1,374 @@ +"""Closed v2 consumer integration for validation, proof, scope, and budget.""" + +from __future__ import annotations + +from dataclasses import replace + +import pytest +from detection_core_fixtures import InMemoryContentResolver, InMemoryPopulationCustody, custody_contract, stored_artifact +from test_core_budget_v2 import _hardware, _reference_workload +from test_core_family import _contract as family_contract +from test_core_family import _record as family_record +from test_core_family import _relation as family_relation +from test_core_validation_v2 import ( + _catalog, + _deterministic_case, + _named_inputs, + _review_policy, + _trust_policy, + _with_fact_set, +) + +import openultrasast.core.consumer_v2 as consumer_v2 +from openultrasast.core import budget, canonical, evidence, validation +from openultrasast.core._proof_attestation import proof_attestation_bytes +from openultrasast.core.consumer_v2 import ConsumerContractError, PushRequest, consume_records, make_push_request +from openultrasast.core.family import FamilyContract +from openultrasast.core.proof import ( + DIFFERENTIAL_PROOF_SCHEMA, + ControlBinding, + DifferentialProofRecord, + ProofAttachment, + ProofOutcome, + ProofProducerIdentity, + ProofSideResult, + ProofTrustPolicy, + ProofVerificationReceipt, + derive_producer_identity_digest, +) +from openultrasast.core.semantics import RelationStatus + + +def _measurement(changed_bytes: int = 120) -> tuple[canonical.ContentRef, InMemoryContentResolver]: + content = canonical.canonical_bytes({"schema_version": budget.CHANGED_BYTES_MEASUREMENT_SCHEMA, "changed_bytes": changed_bytes}) + return stored_artifact(content, artifact_id="consumer-changed-bytes") + + +def _resolver(*values: InMemoryContentResolver) -> InMemoryContentResolver: + artifacts: dict[str, canonical.ResolvedContent] = {} + for value in values: + artifacts.update(value.artifacts) + return InMemoryContentResolver(artifacts) + + +def _request( + value: validation.EvidenceValidationInput, + resolver: InMemoryContentResolver, + *, + scope: evidence.AnalyzedScope | None = None, + budget_ms: int = 1_000, + transition_count: int | None = None, +) -> tuple[PushRequest, InMemoryContentResolver]: + selected_scope = value.record.packet.scope if scope is None else scope + measurement, measurement_resolver = _measurement() + complexity = budget.ChangeComplexity( + selected_scope.change_digest, + "small", + len(selected_scope.regions), + len(value.record.packet.transitions) if transition_count is None else transition_count, + 120, + measurement, + ) + derivation = budget.BudgetDerivation("small", 5, 0.8, _reference_workload(), _hardware(), budget_ms, "") + request = make_push_request(scope=selected_scope, workload_class="small", complexity=complexity, budget=derivation) + return request, _resolver(resolver, measurement_resolver) + + +def _incomplete_case() -> tuple[ + validation.EvidenceValidationInput, + validation.EvidenceValidationContext, + InMemoryPopulationCustody, + InMemoryContentResolver, +]: + contract = family_contract() + record = family_record(contract, relation=family_relation(status=RelationStatus.INCOMPLETE), code="unknown") + return _record_case(record, contract) + + +def _record_case( + record: evidence.EvidenceRecord, + contract: FamilyContract, +) -> tuple[ + validation.EvidenceValidationInput, + validation.EvidenceValidationContext, + InMemoryPopulationCustody, + InMemoryContentResolver, +]: + record = _with_fact_set(record, ()) + named, resolver = _named_inputs(record, contract, ()) + initial, custody_policy, _ = custody_contract() + context = validation.make_evidence_validation_context( + family_catalog=_catalog(contract), + fact_packs=(), + model_programs=(), + lineages=(), + fact_review_policy=_review_policy(), + provenance_trust_policy=_trust_policy(), + custody_policy=custody_policy, + ) + value = validation.make_evidence_validation_input(record=record, named_inputs=named) + return value, context, InMemoryPopulationCustody(initial, custody_policy), resolver + + +def _proof( + record: evidence.EvidenceRecord, +) -> tuple[ProofAttachment, ProofTrustPolicy, InMemoryContentResolver]: + reviewer = canonical.content_digest("consumer-control-reviewer") + control_ref, control_resolver = stored_artifact(b'{"control":"safe"}', artifact_id="consumer-control") + witness_ref, witness_resolver = stored_artifact(b'{"affected":true,"safe":false}', artifact_id="consumer-witness") + producer = ProofProducerIdentity( + "consumer-search-proof", + "verifier-v1", + canonical.content_digest("consumer-sandbox"), + canonical.content_digest("consumer-oracle"), + canonical.content_digest("consumer-producer-review"), + ) + affected = ProofSideResult( + record.packet.provenance.change_input_digest, + True, + canonical.content_digest("consumer-affected"), + 2, + "", + ) + safe = ProofSideResult( + canonical.content_digest("consumer-safe-input"), + False, + canonical.content_digest("consumer-safe"), + 2, + "", + ) + control = ControlBinding(affected.input_digest, safe.input_digest, control_ref, reviewer, "") + proof = DifferentialProofRecord( + DIFFERENTIAL_PROOF_SCHEMA, + producer, + affected, + safe, + control, + witness_ref, + ProofOutcome.DEMONSTRATED, + "", + ) + producer_digest = derive_producer_identity_digest(producer) + attestation_ref, attestation_resolver = stored_artifact( + proof_attestation_bytes( + producer_digest, + proof.record_digest, + affected.result_digest, + safe.result_digest, + witness_ref.content_digest, + ), + artifact_id="consumer-attestation", + ) + verification = ProofVerificationReceipt( + producer_digest, + proof.record_digest, + affected.result_digest, + safe.result_digest, + witness_ref.content_digest, + attestation_ref, + "", + ) + attachment = ProofAttachment( + "", + record.packet.family_id, + evidence.derive_packet_digest(record.packet), + record.packet.transitions[0].location, + proof, + verification, + ) + policy = ProofTrustPolicy((producer_digest,), (reviewer,), ("trusted-store",), 4096, "") + return attachment, policy, _resolver(control_resolver, witness_resolver, attestation_resolver) + + +def test_synthetic_push_returns_bounded_defect_and_input_budget_identities() -> None: + value, context, custody, resolver = _deterministic_case() + request, combined = _request(value, resolver) + + result = consume_records(request, (value,), context, custody, 100, combined) + + assert result.request_digest == request.request_id + assert len(result.defects) == 1 and result.satisfactions == () and result.silence == () + assert result.input_record_digests == (value.record.record_id,) + assert result.complexity is request.complexity + assert result.total_reply_ms >= 100 + assert result.budget_derivation_digest == request.budget.derivation_digest + + +def test_incomplete_push_preserves_exact_named_silence_reasons() -> None: + value, context, custody, resolver = _incomplete_case() + request, combined = _request(value, resolver) + + result = consume_records(request, (value,), context, custody, 100, combined) + + assert result.defects == () and result.satisfactions == () + assert len(result.silence) == 1 + assert result.silence[0].reasons == ("missing_reviewed_dependency_relation",) + assert result.silence[0].evidence_record_digests == (value.record.record_id,) + + +def test_satisfied_push_returns_completed_questions_and_exact_record_scope() -> None: + contract = family_contract() + base = family_record(contract) + decision = replace(base.decision, family_code="safe", outcome=evidence.NeutralOutcome.SATISFIED) + value, context, custody, resolver = _record_case(evidence.EvidenceRecord(base.packet, decision, ""), contract) + request, combined = _request(value, resolver) + + result = consume_records(request, (value,), context, custody, 100, combined) + + assert result.defects == () and result.silence == () + assert result.satisfactions[0].question_ids == ("resolve_constraint",) + assert result.satisfactions[0].scope == value.record.decision.bounded_scope + + +def test_push_request_identity_is_rederived_before_record_validation() -> None: + value, context, custody, resolver = _deterministic_case() + request, combined = _request(value, resolver) + + with pytest.raises(ConsumerContractError, match="request identity"): + consume_records(replace(request, request_id="0" * 64), (value,), context, custody, 100, combined) + + +def test_raw_forged_input_is_revalidated_before_any_output() -> None: + value, context, custody, resolver = _deterministic_case() + request, combined = _request(value, resolver) + forged = replace(value, input_digest="0" * 64) + + with pytest.raises(ConsumerContractError, match="validation.*input digest"): + consume_records(request, (forged,), context, custody, 100, combined) + + +def test_request_scope_and_complexity_must_match_validated_records() -> None: + value, context, custody, resolver = _deterministic_case() + narrow = evidence.AnalyzedScope( + value.record.packet.scope.base_revision, + value.record.packet.scope.head_revision, + value.record.packet.scope.change_digest, + (evidence.SourceRegion("src/app.py", 1, 5),), + ) + request, combined = _request(value, resolver, scope=narrow) + with pytest.raises(ConsumerContractError, match="scope"): + consume_records(request, (value,), context, custody, 100, combined) + + wrong_count, combined = _request(value, resolver, transition_count=2) + with pytest.raises(ConsumerContractError, match="transition count"): + consume_records(wrong_count, (value,), context, custody, 100, combined) + + +def test_optional_proof_is_applied_through_trust_validator() -> None: + value, context, custody, resolver = _deterministic_case() + attachment, trust, proof_resolver = _proof(value.record) + request, combined = _request(value, _resolver(resolver, proof_resolver)) + combined = _resolver(combined, proof_resolver) + + result = consume_records( + request, + (value,), + context, + custody, + 100, + combined, + attachments=(attachment,), + proof_trust=trust, + ) + + assert result.defects[0].basis is evidence.EvidenceBasis.EXECUTION_BACKED + assert result.defects[0].record.decision.provenance.proof_record_digest == attachment.proof.record_digest + with pytest.raises(ConsumerContractError, match="proof trust"): + consume_records(request, (value,), context, custody, 100, combined, attachments=(attachment,)) + + +def test_measured_deadline_is_checked_after_processing() -> None: + value, context, custody, resolver = _deterministic_case() + request, combined = _request(value, resolver, budget_ms=10) + + with pytest.raises(ConsumerContractError, match=r"budget:.*ms>10ms"): + consume_records(request, (value,), context, custody, 11, combined) + + +def test_measured_deadline_adds_actual_core_processing_time(monkeypatch: pytest.MonkeyPatch) -> None: + value, context, custody, resolver = _deterministic_case() + request, combined = _request(value, resolver, budget_ms=10) + ticks = iter((1_000_000, 12_000_001)) + monkeypatch.setattr(consumer_v2, "_monotonic_ns", lambda: next(ticks), raising=False) + + with pytest.raises(ConsumerContractError, match=r"budget:.*13ms>10ms"): + consume_records(request, (value,), context, custody, 1, combined) + + +def test_verdict_reports_elapsed_before_entry_plus_measured_core_time(monkeypatch: pytest.MonkeyPatch) -> None: + value, context, custody, resolver = _deterministic_case() + request, combined = _request(value, resolver) + ticks = iter((1_000_000, 3_500_001)) + monkeypatch.setattr(consumer_v2, "_monotonic_ns", lambda: next(ticks)) + + result = consume_records(request, (value,), context, custody, 5, combined) + + assert result.total_reply_ms == 8 + + +def test_budget_measurement_resolution_is_inside_measured_core_time(monkeypatch: pytest.MonkeyPatch) -> None: + value, context, custody, resolver = _deterministic_case() + request, combined = _request(value, resolver, budget_ms=10) + clock = [0] + + class DelayedMeasurementResolver: + def resolve(self, reference: canonical.ContentRef) -> canonical.ResolvedContent: + if reference.artifact_id == request.complexity.measurement.artifact_id: + clock[0] += 20_000_000 + return combined.resolve(reference) + + monkeypatch.setattr(consumer_v2, "_monotonic_ns", lambda: clock[0]) + + with pytest.raises(ConsumerContractError, match=r"budget:.*21ms>10ms"): + consume_records(request, (value,), context, custody, 1, DelayedMeasurementResolver()) + + +@pytest.mark.parametrize( + ("elapsed_before", "started", "finished", "message"), + ( + (0, 0, 0, "strictly positive"), + (1, 2, 1, "monotonic"), + (1, True, 1, "monotonic"), + (1, -1, 0, "monotonic"), + (1, 0, True, "monotonic"), + (1, 0, -1, "monotonic"), + ), +) +def test_elapsed_measurement_refuses_invalid_clock_matrix( + monkeypatch: pytest.MonkeyPatch, + elapsed_before: int, + started: int, + finished: int, + message: str, +) -> None: + value, context, custody, resolver = _deterministic_case() + request, combined = _request(value, resolver) + ticks = iter((started, finished)) + monkeypatch.setattr(consumer_v2, "_monotonic_ns", lambda: next(ticks)) + + with pytest.raises(ConsumerContractError, match=message): + consume_records(request, (value,), context, custody, elapsed_before, combined) + + +@pytest.mark.parametrize( + ("elapsed_before", "started", "finished", "expected"), + ( + (0, 5, 6, 1), + (2, 5, 5, 2), + ), +) +def test_elapsed_measurement_rounds_nanoseconds_and_allows_preentry_time( + monkeypatch: pytest.MonkeyPatch, + elapsed_before: int, + started: int, + finished: int, + expected: int, +) -> None: + value, context, custody, resolver = _deterministic_case() + request, combined = _request(value, resolver) + ticks = iter((started, finished)) + monkeypatch.setattr(consumer_v2, "_monotonic_ns", lambda: next(ticks)) + + result = consume_records(request, (value,), context, custody, elapsed_before, combined) + + assert result.total_reply_ms == expected