core-labs/core
1
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

feat(kernel): implement diagnostic-only ComputeBudgetDecision

Browse source
This commit is contained in:
Shay 2026-06-22 13:13:49 -07:00
parent 7e4ba10c7a
commit fcac436f7d
2 changed files with 828 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

345
generate/compute_budget.py Normal file
View file

@ -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",
]

483
tests/test_compute_budget.py Normal file
View file

@ -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)