Add declarative harness prototype
This commit is contained in:
395
src/aituner/declarative_harness.py
Normal file
395
src/aituner/declarative_harness.py
Normal file
@@ -0,0 +1,395 @@
|
||||
from __future__ import annotations
|
||||
|
||||
"""Experimental declarative harness substrate.
|
||||
|
||||
This module intentionally stays separate from the production harness while the
|
||||
coverage-relative design is being validated. It models a small, typed subset of
|
||||
the proposed intervention grammar: axes, generic operators, complete candidate
|
||||
sets, failure regions, and stop reports.
|
||||
"""
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
from dataclasses import dataclass
|
||||
from typing import Any, Literal, Mapping, Sequence
|
||||
|
||||
|
||||
AxisKind = Literal["ordered_lattice", "bounded_numeric"]
|
||||
OperatorKind = Literal["bracket", "step_up", "step_down", "jump_to_floor", "local_climb"]
|
||||
RegionRelation = Literal["eq", "ge", "le"]
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class AxisSpec:
|
||||
name: str
|
||||
kind: AxisKind
|
||||
values: tuple[Any, ...] = ()
|
||||
floor: float | None = None
|
||||
ceiling: float | None = None
|
||||
step: float | None = None
|
||||
|
||||
def validate(self) -> None:
|
||||
if not self.name:
|
||||
raise ValueError("axis name must be non-empty")
|
||||
if self.kind == "ordered_lattice":
|
||||
if not self.values:
|
||||
raise ValueError(f"ordered lattice axis {self.name!r} needs values")
|
||||
if len(set(_stable_token(value) for value in self.values)) != len(self.values):
|
||||
raise ValueError(f"ordered lattice axis {self.name!r} has duplicate values")
|
||||
return
|
||||
if self.floor is None or self.ceiling is None:
|
||||
raise ValueError(f"bounded numeric axis {self.name!r} needs floor and ceiling")
|
||||
if self.floor > self.ceiling:
|
||||
raise ValueError(f"bounded numeric axis {self.name!r} has floor above ceiling")
|
||||
if self.step is None or self.step <= 0:
|
||||
raise ValueError(f"bounded numeric axis {self.name!r} needs a positive step")
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class OperatorSpec:
|
||||
name: str
|
||||
axis: str
|
||||
kind: OperatorKind
|
||||
harness_priority: float = 0.0
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CoverageUnit:
|
||||
axis: str
|
||||
operator: str
|
||||
target: Any
|
||||
|
||||
@property
|
||||
def unit_id(self) -> str:
|
||||
return coverage_unit_id(self.axis, self.operator, self.target)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CandidateAction:
|
||||
action_id: str
|
||||
operator: str
|
||||
axis: str
|
||||
patch: Mapping[str, Any]
|
||||
harness_priority: float
|
||||
planner_score: float | None = None
|
||||
backend_score: float | None = None
|
||||
coverage_units: tuple[CoverageUnit, ...] = ()
|
||||
source_value: Any = None
|
||||
target_value: Any = None
|
||||
|
||||
@property
|
||||
def signature(self) -> str:
|
||||
return config_signature(self.patch)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class BlockedCandidate:
|
||||
candidate: CandidateAction
|
||||
reason: str
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class FailureRegion:
|
||||
axis: str
|
||||
relation: RegionRelation
|
||||
value: Any
|
||||
reason: str = "prior_failure"
|
||||
|
||||
def contains(self, candidate: CandidateAction) -> bool:
|
||||
if candidate.axis != self.axis:
|
||||
return False
|
||||
target = candidate.target_value
|
||||
if self.relation == "eq":
|
||||
return target == self.value
|
||||
if self.relation == "ge":
|
||||
return target >= self.value
|
||||
if self.relation == "le":
|
||||
return target <= self.value
|
||||
raise ValueError(f"unknown region relation {self.relation!r}")
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CoverageState:
|
||||
tested_signatures: frozenset[str] = frozenset()
|
||||
covered_unit_ids: frozenset[str] = frozenset()
|
||||
failed_regions: tuple[FailureRegion, ...] = ()
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class HarnessPolicy:
|
||||
operators: tuple[OperatorSpec, ...]
|
||||
no_repeat: bool = True
|
||||
required_coverage_unit_ids: frozenset[str] = frozenset()
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CandidateSet:
|
||||
eligible: tuple[CandidateAction, ...]
|
||||
blocked: tuple[BlockedCandidate, ...]
|
||||
candidate_set_hash: str
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class StopReport:
|
||||
should_stop: bool
|
||||
reason: str
|
||||
candidate_set_hash: str
|
||||
uncovered_unit_ids: tuple[str, ...] = ()
|
||||
eligible_count: int = 0
|
||||
blocked_count: int = 0
|
||||
|
||||
|
||||
def config_signature(patch: Mapping[str, Any]) -> str:
|
||||
return json.dumps(dict(patch), sort_keys=True, separators=(",", ":"), ensure_ascii=False)
|
||||
|
||||
|
||||
def coverage_unit_id(axis: str, operator: str, target: Any) -> str:
|
||||
target_text = json.dumps(target, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
|
||||
return f"{axis}:{operator}:{target_text}"
|
||||
|
||||
|
||||
def ordered_lattice_failure_region(
|
||||
axis: AxisSpec,
|
||||
failed_value: Any,
|
||||
*,
|
||||
direction: Literal["up", "down", "exact"],
|
||||
reason: str = "prior_failure",
|
||||
) -> FailureRegion:
|
||||
axis.validate()
|
||||
if axis.kind != "ordered_lattice":
|
||||
raise ValueError("ordered_lattice_failure_region requires an ordered lattice axis")
|
||||
if failed_value not in axis.values:
|
||||
raise ValueError(f"{failed_value!r} is not in lattice axis {axis.name!r}")
|
||||
if direction == "up":
|
||||
return FailureRegion(axis=axis.name, relation="ge", value=failed_value, reason=reason)
|
||||
if direction == "down":
|
||||
return FailureRegion(axis=axis.name, relation="le", value=failed_value, reason=reason)
|
||||
return FailureRegion(axis=axis.name, relation="eq", value=failed_value, reason=reason)
|
||||
|
||||
|
||||
def enumerate_candidate_set(
|
||||
state: Mapping[str, Any],
|
||||
axes: Sequence[AxisSpec],
|
||||
policy: HarnessPolicy,
|
||||
coverage_state: CoverageState | None = None,
|
||||
) -> CandidateSet:
|
||||
coverage_state = coverage_state or CoverageState()
|
||||
axis_by_name = {axis.name: axis for axis in axes}
|
||||
for axis in axes:
|
||||
axis.validate()
|
||||
|
||||
eligible: list[CandidateAction] = []
|
||||
blocked: list[BlockedCandidate] = []
|
||||
for operator in sorted(
|
||||
policy.operators,
|
||||
key=lambda item: (item.axis, item.name, item.kind),
|
||||
):
|
||||
axis = axis_by_name.get(operator.axis)
|
||||
if axis is None:
|
||||
raise ValueError(f"operator {operator.name!r} references unknown axis {operator.axis!r}")
|
||||
generated, generated_blocked = _generate_operator_actions(state, axis, operator)
|
||||
blocked.extend(generated_blocked)
|
||||
for candidate in generated:
|
||||
reason = _blocking_reason(candidate, policy, coverage_state)
|
||||
if reason is None:
|
||||
eligible.append(candidate)
|
||||
else:
|
||||
blocked.append(BlockedCandidate(candidate=candidate, reason=reason))
|
||||
|
||||
eligible_tuple = tuple(sorted(eligible, key=_candidate_sort_key))
|
||||
blocked_tuple = tuple(
|
||||
sorted(blocked, key=lambda item: (_candidate_sort_key(item.candidate), item.reason))
|
||||
)
|
||||
return CandidateSet(
|
||||
eligible=eligible_tuple,
|
||||
blocked=blocked_tuple,
|
||||
candidate_set_hash=_candidate_set_hash(eligible_tuple, blocked_tuple),
|
||||
)
|
||||
|
||||
|
||||
def validate_coverage_stop(
|
||||
candidate_set: CandidateSet,
|
||||
policy: HarnessPolicy,
|
||||
coverage_state: CoverageState,
|
||||
) -> StopReport:
|
||||
uncovered = tuple(sorted(policy.required_coverage_unit_ids - coverage_state.covered_unit_ids))
|
||||
if uncovered:
|
||||
return StopReport(
|
||||
should_stop=False,
|
||||
reason="coverage_units_missing",
|
||||
candidate_set_hash=candidate_set.candidate_set_hash,
|
||||
uncovered_unit_ids=uncovered,
|
||||
eligible_count=len(candidate_set.eligible),
|
||||
blocked_count=len(candidate_set.blocked),
|
||||
)
|
||||
if candidate_set.eligible:
|
||||
return StopReport(
|
||||
should_stop=False,
|
||||
reason="eligible_candidates_remain",
|
||||
candidate_set_hash=candidate_set.candidate_set_hash,
|
||||
eligible_count=len(candidate_set.eligible),
|
||||
blocked_count=len(candidate_set.blocked),
|
||||
)
|
||||
return StopReport(
|
||||
should_stop=True,
|
||||
reason="coverage_complete_no_eligible_candidates",
|
||||
candidate_set_hash=candidate_set.candidate_set_hash,
|
||||
eligible_count=0,
|
||||
blocked_count=len(candidate_set.blocked),
|
||||
)
|
||||
|
||||
|
||||
def _generate_operator_actions(
|
||||
state: Mapping[str, Any],
|
||||
axis: AxisSpec,
|
||||
operator: OperatorSpec,
|
||||
) -> tuple[list[CandidateAction], list[BlockedCandidate]]:
|
||||
if axis.kind == "ordered_lattice":
|
||||
return _ordered_lattice_actions(state, axis, operator)
|
||||
return _bounded_numeric_actions(state, axis, operator)
|
||||
|
||||
|
||||
def _ordered_lattice_actions(
|
||||
state: Mapping[str, Any],
|
||||
axis: AxisSpec,
|
||||
operator: OperatorSpec,
|
||||
) -> tuple[list[CandidateAction], list[BlockedCandidate]]:
|
||||
if operator.kind not in {"bracket", "step_up", "step_down"}:
|
||||
raise ValueError(
|
||||
f"operator {operator.name!r} is not valid for ordered lattice axis {axis.name!r}"
|
||||
)
|
||||
current = state.get(axis.name)
|
||||
if current not in axis.values:
|
||||
raise ValueError(f"state value {current!r} is not in lattice axis {axis.name!r}")
|
||||
index = axis.values.index(current)
|
||||
if operator.kind == "bracket":
|
||||
targets = [value for value in axis.values if value != current]
|
||||
return ([_candidate(axis, operator, current, target) for target in targets], [])
|
||||
if operator.kind == "step_up":
|
||||
if index == len(axis.values) - 1:
|
||||
return (
|
||||
[],
|
||||
[_boundary_block(axis, operator, current, "ordered_lattice_upper_boundary")],
|
||||
)
|
||||
return ([_candidate(axis, operator, current, axis.values[index + 1])], [])
|
||||
if index == 0:
|
||||
return (
|
||||
[],
|
||||
[_boundary_block(axis, operator, current, "ordered_lattice_lower_boundary")],
|
||||
)
|
||||
return ([_candidate(axis, operator, current, axis.values[index - 1])], [])
|
||||
|
||||
|
||||
def _bounded_numeric_actions(
|
||||
state: Mapping[str, Any],
|
||||
axis: AxisSpec,
|
||||
operator: OperatorSpec,
|
||||
) -> tuple[list[CandidateAction], list[BlockedCandidate]]:
|
||||
if operator.kind not in {"jump_to_floor", "local_climb"}:
|
||||
raise ValueError(
|
||||
f"operator {operator.name!r} is not valid for bounded numeric axis {axis.name!r}"
|
||||
)
|
||||
current = _as_float(state.get(axis.name), axis=axis.name)
|
||||
assert axis.floor is not None
|
||||
assert axis.ceiling is not None
|
||||
assert axis.step is not None
|
||||
if operator.kind == "jump_to_floor":
|
||||
if current < axis.floor:
|
||||
return ([_candidate(axis, operator, current, axis.floor)], [])
|
||||
return ([], [_boundary_block(axis, operator, current, "numeric_at_or_above_floor")])
|
||||
if current < axis.floor:
|
||||
return ([], [_boundary_block(axis, operator, current, "numeric_below_floor")])
|
||||
if current >= axis.ceiling:
|
||||
return ([], [_boundary_block(axis, operator, current, "numeric_upper_boundary")])
|
||||
target = min(axis.ceiling, current + axis.step)
|
||||
return ([_candidate(axis, operator, current, target)], [])
|
||||
|
||||
|
||||
def _candidate(axis: AxisSpec, operator: OperatorSpec, source: Any, target: Any) -> CandidateAction:
|
||||
coverage = CoverageUnit(axis=axis.name, operator=operator.kind, target=target)
|
||||
return CandidateAction(
|
||||
action_id=f"{operator.name}:{axis.name}:{_stable_token(source)}->{_stable_token(target)}",
|
||||
operator=operator.name,
|
||||
axis=axis.name,
|
||||
patch={axis.name: target},
|
||||
harness_priority=operator.harness_priority,
|
||||
coverage_units=(coverage,),
|
||||
source_value=source,
|
||||
target_value=target,
|
||||
)
|
||||
|
||||
|
||||
def _boundary_block(axis: AxisSpec, operator: OperatorSpec, current: Any, reason: str) -> BlockedCandidate:
|
||||
candidate = CandidateAction(
|
||||
action_id=f"{operator.name}:{axis.name}:{_stable_token(current)}->boundary",
|
||||
operator=operator.name,
|
||||
axis=axis.name,
|
||||
patch={axis.name: current},
|
||||
harness_priority=operator.harness_priority,
|
||||
coverage_units=(),
|
||||
source_value=current,
|
||||
target_value=current,
|
||||
)
|
||||
return BlockedCandidate(candidate=candidate, reason=reason)
|
||||
|
||||
|
||||
def _blocking_reason(
|
||||
candidate: CandidateAction,
|
||||
policy: HarnessPolicy,
|
||||
coverage_state: CoverageState,
|
||||
) -> str | None:
|
||||
if policy.no_repeat and candidate.signature in coverage_state.tested_signatures:
|
||||
return "no_repeat: signature already tested"
|
||||
for region in coverage_state.failed_regions:
|
||||
if region.contains(candidate):
|
||||
return f"failure_region:{region.axis}:{region.relation}:{_stable_token(region.value)}:{region.reason}"
|
||||
return None
|
||||
|
||||
|
||||
def _candidate_set_hash(
|
||||
eligible: tuple[CandidateAction, ...],
|
||||
blocked: tuple[BlockedCandidate, ...],
|
||||
) -> str:
|
||||
payload = {
|
||||
"eligible": [_candidate_payload(candidate) for candidate in eligible],
|
||||
"blocked": [
|
||||
{"candidate": _candidate_payload(item.candidate), "reason": item.reason}
|
||||
for item in blocked
|
||||
],
|
||||
}
|
||||
encoded = json.dumps(
|
||||
payload,
|
||||
sort_keys=True,
|
||||
separators=(",", ":"),
|
||||
ensure_ascii=False,
|
||||
).encode("utf-8")
|
||||
return hashlib.sha256(encoded).hexdigest()
|
||||
|
||||
|
||||
def _candidate_payload(candidate: CandidateAction) -> dict[str, Any]:
|
||||
return {
|
||||
"action_id": candidate.action_id,
|
||||
"axis": candidate.axis,
|
||||
"operator": candidate.operator,
|
||||
"patch": dict(candidate.patch),
|
||||
"harness_priority": candidate.harness_priority,
|
||||
"planner_score": candidate.planner_score,
|
||||
"backend_score": candidate.backend_score,
|
||||
"coverage_unit_ids": [unit.unit_id for unit in candidate.coverage_units],
|
||||
"source_value": candidate.source_value,
|
||||
"target_value": candidate.target_value,
|
||||
}
|
||||
|
||||
|
||||
def _candidate_sort_key(candidate: CandidateAction) -> tuple[float, str, str]:
|
||||
return (-candidate.harness_priority, candidate.axis, candidate.action_id)
|
||||
|
||||
|
||||
def _stable_token(value: Any) -> str:
|
||||
return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
|
||||
|
||||
|
||||
def _as_float(value: Any, *, axis: str) -> float:
|
||||
if isinstance(value, bool) or not isinstance(value, (int, float)):
|
||||
raise ValueError(f"state value for numeric axis {axis!r} must be numeric")
|
||||
return float(value)
|
||||
Reference in New Issue
Block a user