From fcac436f7d42b39b569a8ef0e020b60578888c52 Mon Sep 17 00:00:00 2001 From: Shay Date: Mon, 22 Jun 2026 13:13:49 -0700 Subject: [PATCH] feat(kernel): implement diagnostic-only ComputeBudgetDecision --- generate/compute_budget.py | 345 +++++++++++++++++++++++++ tests/test_compute_budget.py | 483 +++++++++++++++++++++++++++++++++++ 2 files changed, 828 insertions(+) create mode 100644 generate/compute_budget.py create mode 100644 tests/test_compute_budget.py diff --git a/generate/compute_budget.py b/generate/compute_budget.py new file mode 100644 index 00000000..77e1e941 --- /dev/null +++ b/generate/compute_budget.py @@ -0,0 +1,345 @@ +"""Diagnostic-only compute budget projection over search-gate decisions.""" + +from __future__ import annotations + +import hashlib +import json +from dataclasses import dataclass +from enum import Enum, unique + +from generate.kernel_facts import SourceSpan +from generate.search_gate import SearchGateDecision, SearchGateStatus + +COMPUTE_BUDGET_POLICY_VERSION = "compute_budget.v1" + + +@unique +class ComputeBudgetStatus(str, Enum): + BUDGET_ALLOWED = "budget_allowed" + BUDGET_BLOCKED = "budget_blocked" + BUDGET_ZERO = "budget_zero" + BUDGET_UNASSESSABLE = "budget_unassessable" + + +@dataclass(frozen=True, slots=True) +class ComputeBudgetDecision: + budget_id: str + policy_version: str + gate_decision_id: str + gate_policy_version: str + gate_input_digest: str + status: ComputeBudgetStatus + reason_code: str + max_candidates: int + max_depth: int + max_steps: int + max_wallclock_ms: int | None + max_parallelism: int + evidence_spans: tuple[SourceSpan, ...] + explanation: str + + +@dataclass(frozen=True, slots=True) +class _BudgetPolicyRow: + reason_code: str + max_candidates: int + max_depth: int + max_steps: int + max_parallelism: int + + +_ELIGIBLE_POLICY: dict[str, _BudgetPolicyRow] = { + "eligible_missing_role": _BudgetPolicyRow( + reason_code="budget_allowed_missing_role", + max_candidates=5, + max_depth=2, + max_steps=10, + max_parallelism=1, + ), + "eligible_missing_relation": _BudgetPolicyRow( + reason_code="budget_allowed_missing_relation", + max_candidates=5, + max_depth=2, + max_steps=10, + max_parallelism=1, + ), + "eligible_missing_proposal": _BudgetPolicyRow( + reason_code="budget_allowed_missing_proposal", + max_candidates=3, + max_depth=1, + max_steps=5, + max_parallelism=1, + ), + "eligible_target_unbound": _BudgetPolicyRow( + reason_code="budget_allowed_target_unbound", + max_candidates=5, + max_depth=2, + max_steps=10, + max_parallelism=1, + ), +} + +_ZERO_ROW = _BudgetPolicyRow( + reason_code="", + max_candidates=0, + max_depth=0, + max_steps=0, + max_parallelism=0, +) + + +def _safe_getattr(value: object, name: str) -> object: + try: + return getattr(value, name, None) + except Exception: + return None + + +def _valid_nonempty_text(value: object) -> bool: + if not isinstance(value, str): + return False + try: + return bool(value.strip()) + except Exception: + return False + + +def _text_or_empty(value: object) -> str: + return value if isinstance(value, str) else "" + + +def _valid_span(span: object) -> bool: + if not isinstance(span, SourceSpan): + return False + try: + if not isinstance(span.text, str): + return False + if type(span.start) is not int or type(span.end) is not int: + return False + if span.start < 0 or span.end < span.start: + return False + return span.sentence_index is None or type(span.sentence_index) is int + except Exception: + return False + + +def _evidence_spans(value: object) -> tuple[SourceSpan, ...] | None: + if not isinstance(value, tuple): + return None + if not all(_valid_span(span) for span in value): + return None + return value + + +def _span_payload(span: SourceSpan) -> dict[str, object]: + return { + "text": span.text, + "start": span.start, + "end": span.end, + "sentence_index": span.sentence_index, + } + + +def _budget_id( + *, + policy_version: str, + gate_decision_id: str, + gate_policy_version: str, + gate_input_digest: str, + status: ComputeBudgetStatus, + reason_code: str, + row: _BudgetPolicyRow, + evidence_spans: tuple[SourceSpan, ...], +) -> str: + payload = { + "policy_version": policy_version, + "gate_decision_id": gate_decision_id, + "gate_policy_version": gate_policy_version, + "gate_input_digest": gate_input_digest, + "status": status.value, + "reason_code": reason_code, + "max_candidates": row.max_candidates, + "max_depth": row.max_depth, + "max_steps": row.max_steps, + "max_parallelism": row.max_parallelism, + "evidence_spans": [_span_payload(span) for span in evidence_spans], + } + encoded = json.dumps( + payload, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ).encode("utf-8") + return hashlib.sha256(encoded).hexdigest() + + +def _make_decision( + *, + gate_decision_id: str, + gate_policy_version: str, + gate_input_digest: str, + status: ComputeBudgetStatus, + reason_code: str, + row: _BudgetPolicyRow, + evidence_spans: tuple[SourceSpan, ...], + explanation: str, +) -> ComputeBudgetDecision: + budget_id = _budget_id( + policy_version=COMPUTE_BUDGET_POLICY_VERSION, + gate_decision_id=gate_decision_id, + gate_policy_version=gate_policy_version, + gate_input_digest=gate_input_digest, + status=status, + reason_code=reason_code, + row=row, + evidence_spans=evidence_spans, + ) + return ComputeBudgetDecision( + budget_id=budget_id, + policy_version=COMPUTE_BUDGET_POLICY_VERSION, + gate_decision_id=gate_decision_id, + gate_policy_version=gate_policy_version, + gate_input_digest=gate_input_digest, + status=status, + reason_code=reason_code, + max_candidates=row.max_candidates, + max_depth=row.max_depth, + max_steps=row.max_steps, + max_wallclock_ms=None, + max_parallelism=row.max_parallelism, + evidence_spans=evidence_spans, + explanation=explanation, + ) + + +def _unassessable_decision( + *, + gate_decision_id: str, + gate_policy_version: str, + gate_input_digest: str, + evidence_spans: tuple[SourceSpan, ...], + reason_code: str, +) -> ComputeBudgetDecision: + return _make_decision( + gate_decision_id=gate_decision_id, + gate_policy_version=gate_policy_version, + gate_input_digest=gate_input_digest, + status=ComputeBudgetStatus.BUDGET_UNASSESSABLE, + reason_code=reason_code, + row=_ZERO_ROW, + evidence_spans=evidence_spans, + explanation=f"Compute budget input is unassessable: {reason_code}.", + ) + + +def decide_compute_budget_for_gate( + gate_decision: SearchGateDecision, +) -> ComputeBudgetDecision: + raw_decision_id = _safe_getattr(gate_decision, "decision_id") + raw_policy_version = _safe_getattr(gate_decision, "policy_version") + raw_input_digest = _safe_getattr(gate_decision, "input_digest") + raw_status = _safe_getattr(gate_decision, "status") + raw_reason_code = _safe_getattr(gate_decision, "reason_code") + raw_evidence_spans = _safe_getattr(gate_decision, "evidence_spans") + + gate_decision_id = _text_or_empty(raw_decision_id) + gate_policy_version = _text_or_empty(raw_policy_version) + gate_input_digest = _text_or_empty(raw_input_digest) + validated_spans = _evidence_spans(raw_evidence_spans) + evidence_spans = validated_spans if validated_spans is not None else () + + if not _valid_nonempty_text(raw_decision_id): + return _unassessable_decision( + gate_decision_id=gate_decision_id, + gate_policy_version=gate_policy_version, + gate_input_digest=gate_input_digest, + evidence_spans=evidence_spans, + reason_code="budget_unassessable_missing_gate_decision_id", + ) + if not _valid_nonempty_text(raw_policy_version): + return _unassessable_decision( + gate_decision_id=gate_decision_id, + gate_policy_version=gate_policy_version, + gate_input_digest=gate_input_digest, + evidence_spans=evidence_spans, + reason_code="budget_unassessable_missing_gate_policy_version", + ) + if not _valid_nonempty_text(raw_input_digest): + return _unassessable_decision( + gate_decision_id=gate_decision_id, + gate_policy_version=gate_policy_version, + gate_input_digest=gate_input_digest, + evidence_spans=evidence_spans, + reason_code="budget_unassessable_missing_gate_input_digest", + ) + if not any(raw_status is status for status in SearchGateStatus): + return _unassessable_decision( + gate_decision_id=gate_decision_id, + gate_policy_version=gate_policy_version, + gate_input_digest=gate_input_digest, + evidence_spans=evidence_spans, + reason_code="budget_unassessable_unknown_gate_status", + ) + if validated_spans is None: + return _unassessable_decision( + gate_decision_id=gate_decision_id, + gate_policy_version=gate_policy_version, + gate_input_digest=gate_input_digest, + evidence_spans=(), + reason_code="budget_unassessable_malformed_evidence_spans", + ) + + if raw_status is not SearchGateStatus.ELIGIBLE: + return _make_decision( + gate_decision_id=gate_decision_id, + gate_policy_version=gate_policy_version, + gate_input_digest=gate_input_digest, + status=ComputeBudgetStatus.BUDGET_BLOCKED, + reason_code="budget_blocked_gate_not_eligible", + row=_ZERO_ROW, + evidence_spans=evidence_spans, + explanation="Compute budget blocked because the gate is not eligible.", + ) + + policy_row = ( + _ELIGIBLE_POLICY.get(raw_reason_code) + if isinstance(raw_reason_code, str) + else None + ) + if policy_row is None: + return _unassessable_decision( + gate_decision_id=gate_decision_id, + gate_policy_version=gate_policy_version, + gate_input_digest=gate_input_digest, + evidence_spans=evidence_spans, + reason_code="budget_unassessable_unknown_gate_reason", + ) + + return _make_decision( + gate_decision_id=gate_decision_id, + gate_policy_version=gate_policy_version, + gate_input_digest=gate_input_digest, + status=ComputeBudgetStatus.BUDGET_ALLOWED, + reason_code=policy_row.reason_code, + row=policy_row, + evidence_spans=evidence_spans, + explanation="Compute budget allowed by the closed v1 policy.", + ) + + +def decide_compute_budget( + gate_decisions: tuple[SearchGateDecision, ...], +) -> tuple[ComputeBudgetDecision, ...]: + return tuple( + decide_compute_budget_for_gate(gate_decision) + for gate_decision in gate_decisions + ) + + +__all__ = [ + "COMPUTE_BUDGET_POLICY_VERSION", + "ComputeBudgetStatus", + "ComputeBudgetDecision", + "decide_compute_budget", + "decide_compute_budget_for_gate", +] diff --git a/tests/test_compute_budget.py b/tests/test_compute_budget.py new file mode 100644 index 00000000..ea3c1471 --- /dev/null +++ b/tests/test_compute_budget.py @@ -0,0 +1,483 @@ +from __future__ import annotations + +import ast +import dataclasses +import hashlib +import json +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from generate.compute_budget import ( + COMPUTE_BUDGET_POLICY_VERSION, + ComputeBudgetDecision, + ComputeBudgetStatus, + decide_compute_budget, + decide_compute_budget_for_gate, +) +from generate.kernel_facts import SourceSpan +from generate.search_gate import SearchGateDecision, SearchGateStatus + + +def _gate( + *, + decision_id: str = "gate-a", + policy_version: str = "search_gate.v1", + input_digest: str = "a" * 64, + status: SearchGateStatus = SearchGateStatus.ELIGIBLE, + reason_code: str = "eligible_missing_role", + evidence_spans: tuple[SourceSpan, ...] = (), + explanation: str = "gate explanation", +) -> SearchGateDecision: + return SearchGateDecision( + decision_id=decision_id, + policy_version=policy_version, + input_digest=input_digest, + residual_ids=("residual-a",), + candidate_organ="unary_delta_transition", + status=status, + reason_code=reason_code, + evidence_spans=evidence_spans, + explanation=explanation, + ) + + +def _span_payload(span: SourceSpan) -> dict[str, object]: + return { + "text": span.text, + "start": span.start, + "end": span.end, + "sentence_index": span.sentence_index, + } + + +def _canonical_budget_id(decision: ComputeBudgetDecision) -> str: + payload = { + "policy_version": decision.policy_version, + "gate_decision_id": decision.gate_decision_id, + "gate_policy_version": decision.gate_policy_version, + "gate_input_digest": decision.gate_input_digest, + "status": decision.status.value, + "reason_code": decision.reason_code, + "max_candidates": decision.max_candidates, + "max_depth": decision.max_depth, + "max_steps": decision.max_steps, + "max_parallelism": decision.max_parallelism, + "evidence_spans": [ + _span_payload(span) for span in decision.evidence_spans + ], + } + encoded = json.dumps( + payload, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ).encode("utf-8") + return hashlib.sha256(encoded).hexdigest() + + +def _assert_zero_limits(decision: ComputeBudgetDecision) -> None: + assert decision.max_candidates == 0 + assert decision.max_depth == 0 + assert decision.max_steps == 0 + assert decision.max_parallelism == 0 + assert decision.max_wallclock_ms is None + + +def test_public_api_exports_are_exact() -> None: + import generate.compute_budget as compute_budget + + assert COMPUTE_BUDGET_POLICY_VERSION == "compute_budget.v1" + assert tuple(compute_budget.__all__) == ( + "COMPUTE_BUDGET_POLICY_VERSION", + "ComputeBudgetStatus", + "ComputeBudgetDecision", + "decide_compute_budget", + "decide_compute_budget_for_gate", + ) + + +def test_batch_is_one_to_one_ordered_and_does_not_aggregate() -> None: + first_gate = _gate(decision_id="gate-first", input_digest="1" * 64) + second_gate = _gate( + decision_id="gate-second", + input_digest="2" * 64, + reason_code="eligible_missing_proposal", + ) + + decisions = decide_compute_budget((first_gate, second_gate)) + + assert len(decisions) == 2 + assert tuple(decision.gate_decision_id for decision in decisions) == ( + "gate-first", + "gate-second", + ) + assert decisions[0].max_candidates == 5 + assert decisions[1].max_candidates == 3 + assert decide_compute_budget(()) == () + + +@pytest.mark.parametrize( + "gate_status", + ( + SearchGateStatus.BLOCKED, + SearchGateStatus.INELIGIBLE, + SearchGateStatus.UNASSESSABLE, + ), +) +def test_valid_noneligible_gates_receive_blocked_zero_budget( + gate_status: SearchGateStatus, +) -> None: + decision = decide_compute_budget_for_gate( + _gate(status=gate_status, reason_code="any_gate_reason") + ) + + assert decision.status is ComputeBudgetStatus.BUDGET_BLOCKED + assert decision.reason_code == "budget_blocked_gate_not_eligible" + _assert_zero_limits(decision) + + +@pytest.mark.parametrize( + "malformed_gate", + ( + SimpleNamespace( + decision_id="gate-a", + policy_version="search_gate.v1", + input_digest="a" * 64, + status="eligible", + reason_code="eligible_missing_role", + evidence_spans=(), + ), + SimpleNamespace( + decision_id="gate-a", + policy_version="search_gate.v1", + input_digest="a" * 64, + reason_code="eligible_missing_role", + evidence_spans=(), + ), + ), +) +def test_unknown_or_missing_status_fails_closed(malformed_gate: object) -> None: + decision = decide_compute_budget_for_gate(malformed_gate) # type: ignore[arg-type] + + assert decision.status is ComputeBudgetStatus.BUDGET_UNASSESSABLE + assert decision.reason_code == "budget_unassessable_unknown_gate_status" + _assert_zero_limits(decision) + + +@pytest.mark.parametrize( + ("reason_code", "expected_reason", "expected_limits"), + ( + ( + "eligible_missing_role", + "budget_allowed_missing_role", + (5, 2, 10, 1), + ), + ( + "eligible_missing_relation", + "budget_allowed_missing_relation", + (5, 2, 10, 1), + ), + ( + "eligible_missing_proposal", + "budget_allowed_missing_proposal", + (3, 1, 5, 1), + ), + ( + "eligible_target_unbound", + "budget_allowed_target_unbound", + (5, 2, 10, 1), + ), + ), +) +def test_eligible_reason_code_table_is_closed( + reason_code: str, + expected_reason: str, + expected_limits: tuple[int, int, int, int], +) -> None: + decision = decide_compute_budget_for_gate(_gate(reason_code=reason_code)) + + assert decision.status is ComputeBudgetStatus.BUDGET_ALLOWED + assert decision.reason_code == expected_reason + assert ( + decision.max_candidates, + decision.max_depth, + decision.max_steps, + decision.max_parallelism, + ) == expected_limits + assert decision.max_wallclock_ms is None + + +@pytest.mark.parametrize("reason_code", ("eligible_future_reason", [], None)) +def test_unknown_eligible_reason_fails_closed(reason_code: object) -> None: + gate = SimpleNamespace( + decision_id="gate-a", + policy_version="search_gate.v1", + input_digest="a" * 64, + status=SearchGateStatus.ELIGIBLE, + reason_code=reason_code, + evidence_spans=(), + ) + decision = decide_compute_budget_for_gate( + gate # type: ignore[arg-type] + ) + + assert decision.status is ComputeBudgetStatus.BUDGET_UNASSESSABLE + assert decision.reason_code == "budget_unassessable_unknown_gate_reason" + _assert_zero_limits(decision) + + +def test_budget_zero_is_reserved_and_unreachable_in_v1() -> None: + assert ComputeBudgetStatus.BUDGET_ZERO.value == "budget_zero" + + gates = tuple( + _gate(status=status, reason_code=reason_code) + for status, reason_code in ( + (SearchGateStatus.ELIGIBLE, "eligible_missing_role"), + (SearchGateStatus.ELIGIBLE, "eligible_future_reason"), + (SearchGateStatus.BLOCKED, "blocked_hazard"), + (SearchGateStatus.INELIGIBLE, "ineligible_reason"), + (SearchGateStatus.UNASSESSABLE, "unassessable_reason"), + ) + ) + + assert all( + decision.status is not ComputeBudgetStatus.BUDGET_ZERO + for decision in decide_compute_budget(gates) + ) + + +def test_budget_id_is_canonical_deterministic_and_structurally_sensitive() -> None: + span = SourceSpan("gained", 4, 10, 0) + gate = _gate(evidence_spans=(span,)) + first = decide_compute_budget_for_gate(gate) + second = decide_compute_budget_for_gate(gate) + changed_digest = decide_compute_budget_for_gate( + dataclasses.replace(gate, input_digest="b" * 64) + ) + changed_policy_row = decide_compute_budget_for_gate( + dataclasses.replace(gate, reason_code="eligible_missing_proposal") + ) + + assert first.budget_id == second.budget_id + assert first.budget_id == _canonical_budget_id(first) + assert len(first.budget_id) == 64 + assert first.budget_id == first.budget_id.lower() + assert changed_digest.budget_id != first.budget_id + assert changed_policy_row.budget_id != first.budget_id + + +def test_gate_explanation_and_budget_prose_do_not_participate_in_identity() -> None: + first = decide_compute_budget_for_gate(_gate(explanation="first prose")) + second = decide_compute_budget_for_gate(_gate(explanation="second prose")) + + assert first.budget_id == second.budget_id + assert dataclasses.replace( + first, + explanation="localized explanation", + max_wallclock_ms=999, + ).budget_id == first.budget_id + assert first.budget_id == _canonical_budget_id(first) + + +def test_evidence_spans_preserve_order_duplicates_and_identity() -> None: + first_span = SourceSpan("same", 0, 4, 0) + second_span = SourceSpan("other", 5, 10, 0) + decision = decide_compute_budget_for_gate( + _gate(evidence_spans=(first_span, second_span, first_span)) + ) + reversed_decision = decide_compute_budget_for_gate( + _gate(evidence_spans=(second_span, first_span, first_span)) + ) + + assert decision.evidence_spans == (first_span, second_span, first_span) + assert len(decision.evidence_spans) == 3 + assert decision.budget_id != reversed_decision.budget_id + + +@pytest.mark.parametrize( + "evidence_spans", + ( + None, + [], + (object(),), + (SimpleNamespace(text="x", start=0, end=1, sentence_index=0),), + ), +) +def test_malformed_evidence_spans_fail_closed(evidence_spans: object) -> None: + malformed_gate = SimpleNamespace( + decision_id="gate-a", + policy_version="search_gate.v1", + input_digest="a" * 64, + status=SearchGateStatus.ELIGIBLE, + reason_code="eligible_missing_role", + evidence_spans=evidence_spans, + ) + + decision = decide_compute_budget_for_gate(malformed_gate) # type: ignore[arg-type] + + assert decision.status is ComputeBudgetStatus.BUDGET_UNASSESSABLE + assert decision.reason_code == "budget_unassessable_malformed_evidence_spans" + assert decision.evidence_spans == () + _assert_zero_limits(decision) + + +@pytest.mark.parametrize( + ("field_name", "field_value", "expected_reason"), + ( + ( + "decision_id", + "", + "budget_unassessable_missing_gate_decision_id", + ), + ( + "policy_version", + " ", + "budget_unassessable_missing_gate_policy_version", + ), + ( + "input_digest", + None, + "budget_unassessable_missing_gate_input_digest", + ), + ), +) +def test_missing_gate_identity_fields_fail_closed( + field_name: str, + field_value: object, + expected_reason: str, +) -> None: + values = { + "decision_id": "gate-a", + "policy_version": "search_gate.v1", + "input_digest": "a" * 64, + "status": SearchGateStatus.ELIGIBLE, + "reason_code": "eligible_missing_role", + "evidence_spans": (), + } + values[field_name] = field_value + + decision = decide_compute_budget_for_gate( # type: ignore[arg-type] + SimpleNamespace(**values) + ) + + assert decision.status is ComputeBudgetStatus.BUDGET_UNASSESSABLE + assert decision.reason_code == expected_reason + _assert_zero_limits(decision) + + +def test_compute_budget_decision_contains_no_authority_fields() -> None: + forbidden_fields = { + "candidate", + "proposal", + "proof", + "answer", + "verdict", + "repair", + "action", + "search", + "search_run", + "runtime", + "serving_allowed", + "mutation", + "promotion", + "rank", + "priority", + "score", + "confidence", + } + + assert forbidden_fields.isdisjoint( + {field.name for field in dataclasses.fields(ComputeBudgetDecision)} + ) + + +def test_module_coupling_and_side_effect_guards() -> None: + path = Path("generate/compute_budget.py") + source = path.read_text("utf-8") + tree = ast.parse(source) + imports: set[str] = set() + imported_names: set[str] = set() + calls: set[str] = set() + + for node in ast.walk(tree): + if isinstance(node, ast.Import): + imports.update(alias.name for alias in node.names) + elif isinstance(node, ast.ImportFrom): + imports.add(node.module or "") + imported_names.update(alias.name for alias in node.names) + elif isinstance(node, ast.Call): + if isinstance(node.func, ast.Name): + calls.add(node.func.id) + elif isinstance(node.func, ast.Attribute): + calls.add(node.func.attr) + + assert imports <= { + "__future__", + "dataclasses", + "enum", + "hashlib", + "json", + "generate.kernel_facts", + "generate.search_gate", + } + assert { + "decide_search_gate", + "assess_contracts", + "project_contract_residuals", + "determine", + }.isdisjoint(imported_names) + assert { + "decide_search_gate", + "assess_contracts", + "project_contract_residuals", + "determine", + "repair", + "serve", + "store", + "write", + "open", + "write_text", + "write_bytes", + "request", + "urlopen", + "sleep", + "time", + "uuid4", + }.isdisjoint(calls) + assert "generate.search_gate" in source + assert "compute_budget" not in Path("generate/search_gate.py").read_text() + assert "compute_budget" not in Path( + "generate/contract_residuals.py" + ).read_text() + + +def test_no_filesystem_network_clock_random_or_environment_identity() -> None: + source = Path("generate/compute_budget.py").read_text("utf-8") + forbidden_fragments = ( + "import os", + "from os", + "import pathlib", + "from pathlib", + "import random", + "from random", + "import time", + "from time", + "import datetime", + "from datetime", + "import uuid", + "from uuid", + "import socket", + "from socket", + "import subprocess", + "from subprocess", + "import requests", + "from requests", + "os.environ", + "getenv(", + "gethostname(", + "Path(", + ) + + assert not any(fragment in source for fragment in forbidden_fragments)