test(kernel): cover run-attempt binding shell
This commit is contained in:
Shay
parent
f8d496ff9e
commit
76bb756519
1 changed files with 266 additions and 237 deletions
|
|
@ -1,33 +1,25 @@
|
|||
#!/usr/bin/env python3
|
||||
"""
|
||||
tests/test_run_attempt_binding.py
|
||||
|
||||
Comprehensive tests for the inert CandidateAttempt run-binding shell (ADR-0232).
|
||||
|
||||
Covers:
|
||||
- Exact public API surface
|
||||
- Happy path produces valid immutable binding
|
||||
- No mutation of original run
|
||||
- Deterministic IDs (binding_id, input_digest, candidate_attempt_ref)
|
||||
- Explanation changes do not affect IDs
|
||||
- All refusal paths with correct reason codes
|
||||
- Evidence span order/duplication preservation
|
||||
- No forbidden fields (answer/proof/verdict/etc.)
|
||||
- Static source guards (no forbidden imports/calls in the module)
|
||||
- No reverse dependency from upstream generate/ modules
|
||||
- Focused + smoke compatibility
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import dataclasses
|
||||
import hashlib
|
||||
import json
|
||||
from dataclasses import replace
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
from generate.candidate_operator import (
|
||||
GroundedUnaryDeltaCue,
|
||||
build_missing_role_candidate,
|
||||
candidate_operator_set_id,
|
||||
)
|
||||
from generate.compute_budget import ComputeBudgetDecision, ComputeBudgetStatus
|
||||
from generate.contract_residuals import ContractResidual, ResidualKind, ResidualSourceAxis
|
||||
from generate.geometric_search_run import (
|
||||
BudgetCharge,
|
||||
CandidateReplayStatus,
|
||||
GeometricSearchRun,
|
||||
initialize_geometric_search_run,
|
||||
)
|
||||
from generate.kernel_facts import SourceSpan
|
||||
from generate.run_attempt_binding import (
|
||||
CANDIDATE_ATTEMPT_RUN_BINDING_POLICY_VERSION,
|
||||
CandidateAttemptRunBinding,
|
||||
|
|
@ -37,102 +29,133 @@ from generate.run_attempt_binding import (
|
|||
RunAttemptBindingRefusalReason,
|
||||
bind_candidate_attempt_to_run,
|
||||
)
|
||||
|
||||
from generate.geometric_search_run import initialize_geometric_search_run
|
||||
from generate.candidate_operator import build_missing_role_candidate
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||||
GENERATE_DIR = REPO_ROOT / "generate"
|
||||
from generate.search_gate import SearchGateDecision, SearchGateStatus
|
||||
|
||||
|
||||
def _make_minimal_valid_run() -> Any:
|
||||
try:
|
||||
run = initialize_geometric_search_run()
|
||||
if run is not None:
|
||||
return run
|
||||
except Exception:
|
||||
pass
|
||||
return SimpleNamespace(
|
||||
run_id="run_" + "0"*64,
|
||||
policy_version="geometric_search_run.v1",
|
||||
input_digest="input_" + "0"*64,
|
||||
residual_ids=("res_a", "res_b"),
|
||||
problem_frame_digest="pf_" + "0"*64,
|
||||
contract_assessment_id="ca_" + "0"*64,
|
||||
budget_consumed=SimpleNamespace(
|
||||
candidates_considered=0, max_candidates=10,
|
||||
steps_used=0, max_steps=100,
|
||||
depth_reached=0, max_depth=5,
|
||||
parallelism_used=1, max_parallelism=4,
|
||||
budget_id="budget_0123",
|
||||
),
|
||||
candidate_attempts=(),
|
||||
operator_set_id="opset_v1",
|
||||
operator_set_version="1.0.0",
|
||||
gate_decision_id="gate_0123",
|
||||
def _digest(payload: dict[str, object]) -> str:
|
||||
encoded = json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
|
||||
return hashlib.sha256(encoded).hexdigest()
|
||||
|
||||
|
||||
def _span_payload(span: SourceSpan) -> dict[str, object]:
|
||||
return {"text": span.text, "start": span.start, "end": span.end, "sentence_index": span.sentence_index}
|
||||
|
||||
|
||||
def _span(text: str = "gained", start: int = 10, end: int = 15) -> SourceSpan:
|
||||
return SourceSpan(text=text, start=start, end=end, sentence_index=0)
|
||||
|
||||
|
||||
def _residual(evidence_spans: tuple[SourceSpan, ...] | None = None) -> ContractResidual:
|
||||
spans = evidence_spans if evidence_spans is not None else (_span(),)
|
||||
residual_id = _digest({
|
||||
"candidate_organ": "unary_delta_transition",
|
||||
"residual_kind": ResidualKind.MISSING_ROLE.value,
|
||||
"residual_code": "direction_unbound",
|
||||
"evidence_spans": [_span_payload(span) for span in spans],
|
||||
})
|
||||
return ContractResidual(
|
||||
residual_id=residual_id,
|
||||
candidate_organ="unary_delta_transition",
|
||||
family_id="state_change.unary_delta",
|
||||
residual_kind=ResidualKind.MISSING_ROLE,
|
||||
residual_code="direction_unbound",
|
||||
source_axis=ResidualSourceAxis.ROLE,
|
||||
evidence_spans=spans,
|
||||
explanation="residual prose",
|
||||
)
|
||||
|
||||
|
||||
def _make_minimal_valid_candidate_result(run: Any) -> Any:
|
||||
try:
|
||||
co = build_missing_role_candidate(run=run)
|
||||
if co is not None:
|
||||
return co
|
||||
except Exception:
|
||||
pass
|
||||
ca = SimpleNamespace(
|
||||
attempt_id="att_" + "0"*64,
|
||||
candidate_digest="cand_" + "0"*64,
|
||||
replay_status="REPLAY_PENDING",
|
||||
replay_blockers=(),
|
||||
attempt_index=0,
|
||||
input_digest="input_" + "0"*64,
|
||||
operator_id="operator_missing_role",
|
||||
operator_version="1.0.0",
|
||||
budget_charge=SimpleNamespace(candidates=1, steps=5),
|
||||
depth=1,
|
||||
step_index=0,
|
||||
evidence_spans=(),
|
||||
def _gate(residual: ContractResidual) -> SearchGateDecision:
|
||||
payload = {
|
||||
"policy_version": "search_gate.v1",
|
||||
"input_digest": "a" * 64,
|
||||
"residual_ids": [residual.residual_id],
|
||||
"candidate_organ": residual.candidate_organ,
|
||||
"status": SearchGateStatus.ELIGIBLE.value,
|
||||
"reason_code": "eligible_missing_role",
|
||||
"evidence_spans": [_span_payload(span) for span in residual.evidence_spans],
|
||||
}
|
||||
return SearchGateDecision(
|
||||
decision_id=_digest(payload),
|
||||
policy_version="search_gate.v1",
|
||||
input_digest="a" * 64,
|
||||
residual_ids=(residual.residual_id,),
|
||||
candidate_organ=residual.candidate_organ,
|
||||
status=SearchGateStatus.ELIGIBLE,
|
||||
reason_code="eligible_missing_role",
|
||||
evidence_spans=residual.evidence_spans,
|
||||
explanation="gate prose",
|
||||
)
|
||||
cr = SimpleNamespace(
|
||||
candidate_digest="cand_" + "0"*64,
|
||||
candidate_reconstruction_digest="recon_" + "0"*64,
|
||||
source_residual_id="res_a",
|
||||
|
||||
|
||||
def _budget(gate: SearchGateDecision, *, max_candidates: int = 5, max_depth: int = 2, max_steps: int = 10) -> ComputeBudgetDecision:
|
||||
payload = {
|
||||
"policy_version": "compute_budget.v1",
|
||||
"gate_decision_id": gate.decision_id,
|
||||
"gate_policy_version": gate.policy_version,
|
||||
"gate_input_digest": gate.input_digest,
|
||||
"status": ComputeBudgetStatus.BUDGET_ALLOWED.value,
|
||||
"reason_code": "budget_allowed_missing_role",
|
||||
"max_candidates": max_candidates,
|
||||
"max_depth": max_depth,
|
||||
"max_steps": max_steps,
|
||||
"max_parallelism": 1,
|
||||
"evidence_spans": [_span_payload(span) for span in gate.evidence_spans],
|
||||
}
|
||||
return ComputeBudgetDecision(
|
||||
budget_id=_digest(payload),
|
||||
policy_version="compute_budget.v1",
|
||||
gate_decision_id=gate.decision_id,
|
||||
gate_policy_version=gate.policy_version,
|
||||
gate_input_digest=gate.input_digest,
|
||||
status=ComputeBudgetStatus.BUDGET_ALLOWED,
|
||||
reason_code="budget_allowed_missing_role",
|
||||
max_candidates=max_candidates,
|
||||
max_depth=max_depth,
|
||||
max_steps=max_steps,
|
||||
max_wallclock_ms=None,
|
||||
max_parallelism=1,
|
||||
evidence_spans=gate.evidence_spans,
|
||||
explanation="budget prose",
|
||||
)
|
||||
|
||||
|
||||
def _chain() -> tuple[ContractResidual, GeometricSearchRun, object]:
|
||||
residual = _residual()
|
||||
gate = _gate(residual)
|
||||
budget = _budget(gate)
|
||||
run = initialize_geometric_search_run(
|
||||
problem_frame_digest="f" * 64,
|
||||
contract_assessment_id="assessment-a",
|
||||
residual_ids=(residual.residual_id,),
|
||||
gate_decision=gate,
|
||||
compute_budget=budget,
|
||||
operator_set_id=candidate_operator_set_id(),
|
||||
operator_set_version="candidate_operators.v1",
|
||||
)
|
||||
assert isinstance(run, GeometricSearchRun)
|
||||
result = build_missing_role_candidate(
|
||||
residual=residual,
|
||||
search_gate=gate,
|
||||
compute_budget=budget,
|
||||
run=run,
|
||||
problem_frame_digest=run.problem_frame_digest,
|
||||
original_contract_assessment_id=run.contract_assessment_id,
|
||||
evidence_spans=(),
|
||||
operator_set_id=run.operator_set_id,
|
||||
operator_set_version=run.operator_set_version,
|
||||
)
|
||||
return SimpleNamespace(
|
||||
geometric_search_run_id=run.run_id,
|
||||
attempt_id=ca.attempt_id,
|
||||
input_digest=ca.input_digest,
|
||||
candidate_digest=ca.candidate_digest,
|
||||
candidate_reconstruction_digest=cr.candidate_reconstruction_digest,
|
||||
operator_name=ca.operator_id,
|
||||
operator_version=ca.operator_version,
|
||||
candidate_attempt=ca,
|
||||
candidate_reconstruction=cr,
|
||||
result_id="opres_" + "0"*64,
|
||||
policy_version="candidate_operator.v1",
|
||||
evidence_spans=(),
|
||||
grounded_unary_delta_cues=(GroundedUnaryDeltaCue(direction="increase", evidence_spans=residual.evidence_spans),),
|
||||
)
|
||||
assert hasattr(result, "candidate_attempt")
|
||||
return residual, run, result
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def valid_run():
|
||||
return _make_minimal_valid_run()
|
||||
def _refused(reason: RunAttemptBindingRefusalReason, *, run: GeometricSearchRun, result: object) -> None:
|
||||
outcome = bind_candidate_attempt_to_run(original_run=run, candidate_operator_result=result) # type: ignore[arg-type]
|
||||
assert isinstance(outcome, CandidateAttemptRunBindingRefusal)
|
||||
assert outcome.reason_codes == (reason.value,)
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def valid_co_result(valid_run):
|
||||
return _make_minimal_valid_candidate_result(valid_run)
|
||||
|
||||
|
||||
def test_public_api_exports_exact():
|
||||
import generate.run_attempt_binding as m
|
||||
expected = {
|
||||
def test_public_api_exports_are_exact() -> None:
|
||||
import generate.run_attempt_binding as module
|
||||
assert tuple(module.__all__) == (
|
||||
"CANDIDATE_ATTEMPT_RUN_BINDING_POLICY_VERSION",
|
||||
"RunAttemptBindingRefusalReason",
|
||||
"CandidateAttemptRunBindingInput",
|
||||
|
|
@ -140,146 +163,152 @@ def test_public_api_exports_exact():
|
|||
"CandidateAttemptRunBindingRefusal",
|
||||
"CandidateAttemptRunBindingOutcome",
|
||||
"bind_candidate_attempt_to_run",
|
||||
}
|
||||
actual = {name for name in dir(m) if not name.startswith("_")}
|
||||
assert expected.issubset(actual)
|
||||
)
|
||||
assert CANDIDATE_ATTEMPT_RUN_BINDING_POLICY_VERSION == "candidate_attempt_run_binding.v1"
|
||||
assert CandidateAttemptRunBindingOutcome is not None
|
||||
|
||||
|
||||
def test_valid_binding_produces_correct_type_and_membership(valid_run, valid_co_result):
|
||||
outcome = bind_candidate_attempt_to_run(original_run=valid_run, candidate_operator_result=valid_co_result)
|
||||
def test_valid_result_binds_without_mutating_run() -> None:
|
||||
_, run, result = _chain()
|
||||
before = run.candidate_attempts
|
||||
outcome = bind_candidate_attempt_to_run(original_run=run, candidate_operator_result=result)
|
||||
assert isinstance(outcome, CandidateAttemptRunBinding)
|
||||
assert outcome.run_attempt_membership == "structurally_bound"
|
||||
assert outcome.reason_codes == ()
|
||||
assert run.candidate_attempts == before == ()
|
||||
|
||||
|
||||
def test_binding_does_not_mutate_original_run(valid_run, valid_co_result):
|
||||
before = tuple(getattr(valid_run, "candidate_attempts", ()))
|
||||
outcome = bind_candidate_attempt_to_run(original_run=valid_run, candidate_operator_result=valid_co_result)
|
||||
after = tuple(getattr(valid_run, "candidate_attempts", ()))
|
||||
assert before == after
|
||||
def test_explanation_does_not_affect_ids() -> None:
|
||||
_, run, result = _chain()
|
||||
first = bind_candidate_attempt_to_run(original_run=run, candidate_operator_result=result, explanation="first")
|
||||
second = bind_candidate_attempt_to_run(original_run=run, candidate_operator_result=result, explanation="second")
|
||||
assert isinstance(first, CandidateAttemptRunBinding)
|
||||
assert isinstance(second, CandidateAttemptRunBinding)
|
||||
assert first.input_digest == second.input_digest
|
||||
assert first.candidate_attempt_ref == second.candidate_attempt_ref
|
||||
assert first.binding_id == second.binding_id
|
||||
|
||||
|
||||
def test_invalid_inputs_refuse() -> None:
|
||||
_, run, result = _chain()
|
||||
outcome = bind_candidate_attempt_to_run(original_run=SimpleNamespace(), candidate_operator_result=result) # type: ignore[arg-type]
|
||||
assert isinstance(outcome, CandidateAttemptRunBindingRefusal)
|
||||
assert outcome.reason_codes == (RunAttemptBindingRefusalReason.INVALID_BINDING_INPUT.value,)
|
||||
outcome = bind_candidate_attempt_to_run(original_run=run, candidate_operator_result=SimpleNamespace()) # type: ignore[arg-type]
|
||||
assert isinstance(outcome, CandidateAttemptRunBindingRefusal)
|
||||
assert outcome.reason_codes == (RunAttemptBindingRefusalReason.INVALID_BINDING_INPUT.value,)
|
||||
|
||||
|
||||
def test_run_and_attempt_identity_refusals() -> None:
|
||||
_, run, result = _chain()
|
||||
_refused(RunAttemptBindingRefusalReason.RUN_RESULT_MISMATCH, run=run, result=replace(result, geometric_search_run_id="wrong"))
|
||||
_refused(RunAttemptBindingRefusalReason.ATTEMPT_RESULT_MISMATCH, run=run, result=replace(result, attempt_id="wrong"))
|
||||
bad_attempt = replace(result.candidate_attempt, candidate_digest="bad")
|
||||
_refused(RunAttemptBindingRefusalReason.ATTEMPT_RESULT_MISMATCH, run=run, result=replace(result, candidate_attempt=bad_attempt))
|
||||
bad_attempt = replace(result.candidate_attempt, input_digest="bad")
|
||||
_refused(RunAttemptBindingRefusalReason.ATTEMPT_RESULT_MISMATCH, run=run, result=replace(result, candidate_attempt=bad_attempt))
|
||||
|
||||
|
||||
def test_reconstruction_identity_refusals() -> None:
|
||||
_, run, result = _chain()
|
||||
bad = replace(result.candidate_reconstruction, candidate_digest="bad")
|
||||
_refused(RunAttemptBindingRefusalReason.RECONSTRUCTION_RESULT_MISMATCH, run=run, result=replace(result, candidate_reconstruction=bad))
|
||||
bad = replace(result.candidate_reconstruction, candidate_reconstruction_digest="bad")
|
||||
_refused(RunAttemptBindingRefusalReason.RECONSTRUCTION_RESULT_MISMATCH, run=run, result=replace(result, candidate_reconstruction=bad))
|
||||
|
||||
|
||||
def test_replay_state_refusals() -> None:
|
||||
_, run, result = _chain()
|
||||
bad_attempt = replace(result.candidate_attempt, replay_status=CandidateReplayStatus.REPLAY_CLOSED)
|
||||
_refused(RunAttemptBindingRefusalReason.REPLAY_STATUS_NOT_PENDING, run=run, result=replace(result, candidate_attempt=bad_attempt))
|
||||
bad_attempt = replace(result.candidate_attempt, replay_blockers=("blocked",))
|
||||
_refused(RunAttemptBindingRefusalReason.REPLAY_BLOCKERS_PRESENT, run=run, result=replace(result, candidate_attempt=bad_attempt))
|
||||
|
||||
|
||||
def test_run_identity_and_operator_set_refusals() -> None:
|
||||
_, run, result = _chain()
|
||||
bad = replace(result.candidate_reconstruction, source_residual_id="missing")
|
||||
_refused(RunAttemptBindingRefusalReason.RUN_IDENTITY_MISMATCH, run=run, result=replace(result, candidate_reconstruction=bad))
|
||||
bad = replace(result.candidate_reconstruction, problem_frame_digest="bad")
|
||||
_refused(RunAttemptBindingRefusalReason.RUN_IDENTITY_MISMATCH, run=run, result=replace(result, candidate_reconstruction=bad))
|
||||
bad = replace(result.candidate_reconstruction, original_contract_assessment_id="bad")
|
||||
_refused(RunAttemptBindingRefusalReason.RUN_IDENTITY_MISMATCH, run=run, result=replace(result, candidate_reconstruction=bad))
|
||||
bad = replace(result.candidate_reconstruction, operator_provenance=(("operator_set_id", "wrong"), ("operator_set_version", run.operator_set_version)))
|
||||
_refused(RunAttemptBindingRefusalReason.OPERATOR_SET_MISMATCH, run=run, result=replace(result, candidate_reconstruction=bad))
|
||||
|
||||
|
||||
def test_evidence_span_refusals() -> None:
|
||||
_, run, result = _chain()
|
||||
span = SourceSpan(text="other", start=20, end=25, sentence_index=0)
|
||||
_refused(RunAttemptBindingRefusalReason.MALFORMED_EVIDENCE_SPAN, run=run, result=replace(result, evidence_spans=(span,)))
|
||||
bad_attempt = replace(result.candidate_attempt, evidence_spans=(object(),))
|
||||
_refused(RunAttemptBindingRefusalReason.MALFORMED_EVIDENCE_SPAN, run=run, result=replace(result, candidate_attempt=bad_attempt))
|
||||
|
||||
|
||||
def test_duplicate_membership_refusals() -> None:
|
||||
_, run, result = _chain()
|
||||
existing = replace(result.candidate_attempt, attempt_id="other", candidate_digest="other")
|
||||
_refused(RunAttemptBindingRefusalReason.DUPLICATE_ATTEMPT_INDEX, run=replace(run, candidate_attempts=(existing,)), result=result)
|
||||
existing = replace(result.candidate_attempt, attempt_index=99, candidate_digest="other")
|
||||
_refused(RunAttemptBindingRefusalReason.DUPLICATE_ATTEMPT_ID, run=replace(run, candidate_attempts=(existing,)), result=result)
|
||||
existing = replace(result.candidate_attempt, attempt_index=99, attempt_id="other")
|
||||
_refused(RunAttemptBindingRefusalReason.DUPLICATE_CANDIDATE_DIGEST, run=replace(run, candidate_attempts=(existing,)), result=result)
|
||||
|
||||
|
||||
def test_budget_refusals() -> None:
|
||||
_, run, result = _chain()
|
||||
bad_attempt = replace(result.candidate_attempt, budget_charge=BudgetCharge(candidates=999, steps=1))
|
||||
_refused(RunAttemptBindingRefusalReason.BUDGET_CHARGE_EXCEEDS_REMAINING, run=run, result=replace(result, candidate_attempt=bad_attempt))
|
||||
bad_attempt = replace(result.candidate_attempt, budget_charge=BudgetCharge(candidates=1, steps=999))
|
||||
_refused(RunAttemptBindingRefusalReason.BUDGET_CHARGE_EXCEEDS_REMAINING, run=run, result=replace(result, candidate_attempt=bad_attempt))
|
||||
bad_attempt = replace(result.candidate_attempt, depth=999)
|
||||
_refused(RunAttemptBindingRefusalReason.BUDGET_CHARGE_EXCEEDS_REMAINING, run=run, result=replace(result, candidate_attempt=bad_attempt))
|
||||
|
||||
|
||||
def test_schema_policy_version_pairs_validate() -> None:
|
||||
_, run, result = _chain()
|
||||
outcome = bind_candidate_attempt_to_run(original_run=run, candidate_operator_result=result, schema_versions=(("b", "1"), ("a", "1")))
|
||||
assert isinstance(outcome, CandidateAttemptRunBindingRefusal)
|
||||
assert outcome.reason_codes == (RunAttemptBindingRefusalReason.UNSUPPORTED_SCHEMA_VERSION.value,)
|
||||
outcome = bind_candidate_attempt_to_run(original_run=run, candidate_operator_result=result, policy_versions=(("a", "1"), ("a", "2")))
|
||||
assert isinstance(outcome, CandidateAttemptRunBindingRefusal)
|
||||
assert outcome.reason_codes == (RunAttemptBindingRefusalReason.UNSUPPORTED_SCHEMA_VERSION.value,)
|
||||
|
||||
|
||||
def test_evidence_order_and_duplicates_are_preserved() -> None:
|
||||
_, run, result = _chain()
|
||||
first = SourceSpan(text="one", start=0, end=3, sentence_index=0)
|
||||
second = SourceSpan(text="two", start=4, end=7, sentence_index=0)
|
||||
spans = (first, first, second)
|
||||
attempt = replace(result.candidate_attempt, evidence_spans=spans)
|
||||
reconstruction = replace(result.candidate_reconstruction, evidence_spans=spans)
|
||||
updated = replace(result, candidate_attempt=attempt, candidate_reconstruction=reconstruction, evidence_spans=spans)
|
||||
outcome = bind_candidate_attempt_to_run(original_run=run, candidate_operator_result=updated)
|
||||
assert isinstance(outcome, CandidateAttemptRunBinding)
|
||||
assert outcome.evidence_spans == spans
|
||||
reordered = (second, first, first)
|
||||
attempt = replace(result.candidate_attempt, evidence_spans=reordered)
|
||||
reconstruction = replace(result.candidate_reconstruction, evidence_spans=reordered)
|
||||
updated_reordered = replace(result, candidate_attempt=attempt, candidate_reconstruction=reconstruction, evidence_spans=reordered)
|
||||
other = bind_candidate_attempt_to_run(original_run=run, candidate_operator_result=updated_reordered)
|
||||
assert isinstance(other, CandidateAttemptRunBinding)
|
||||
assert outcome.binding_id != other.binding_id
|
||||
|
||||
|
||||
def test_binding_id_and_input_digest_are_deterministic(valid_run, valid_co_result):
|
||||
out1 = bind_candidate_attempt_to_run(original_run=valid_run, candidate_operator_result=valid_co_result, explanation="first")
|
||||
out2 = bind_candidate_attempt_to_run(original_run=valid_run, candidate_operator_result=valid_co_result, explanation="second")
|
||||
assert out1.binding_id == out2.binding_id
|
||||
assert out1.input_digest == out2.input_digest
|
||||
assert out1.candidate_attempt_ref == out2.candidate_attempt_ref
|
||||
def test_binding_dataclasses_have_no_forbidden_authority_fields() -> None:
|
||||
forbidden = {
|
||||
"answer", "final_answer", "served_output", "proof", "verdict", "promotion",
|
||||
"mutation", "teaching_update", "pack_update", "policy_update", "identity_update",
|
||||
"workbench_state", "runtime_effect", "confidence", "score", "rank", "priority",
|
||||
"selected", "selected_candidate", "best", "best_candidate", "serving_allowed", "runnable",
|
||||
}
|
||||
for record_type in (CandidateAttemptRunBindingInput, CandidateAttemptRunBinding, CandidateAttemptRunBindingRefusal):
|
||||
assert forbidden.isdisjoint({field.name for field in dataclasses.fields(record_type)})
|
||||
|
||||
|
||||
def test_explanation_change_does_not_affect_ids(valid_run, valid_co_result):
|
||||
out1 = bind_candidate_attempt_to_run(original_run=valid_run, candidate_operator_result=valid_co_result, explanation="foo")
|
||||
out2 = bind_candidate_attempt_to_run(original_run=valid_run, candidate_operator_result=valid_co_result, explanation="bar")
|
||||
assert out1.binding_id == out2.binding_id
|
||||
assert out1.input_digest == out2.input_digest
|
||||
|
||||
# Full refusal coverage
|
||||
def test_refusal_on_run_result_mismatch(valid_run, valid_co_result):
|
||||
bad = SimpleNamespace(**{k: getattr(valid_co_result, k, None) for k in dir(valid_co_result) if not k.startswith("_")})
|
||||
setattr(bad, "geometric_search_run_id", "wrong_run")
|
||||
outcome = bind_candidate_attempt_to_run(original_run=valid_run, candidate_operator_result=bad)
|
||||
def test_refusal_is_not_partial_binding() -> None:
|
||||
_, run, result = _chain()
|
||||
outcome = bind_candidate_attempt_to_run(original_run=run, candidate_operator_result=replace(result, geometric_search_run_id="wrong"))
|
||||
assert isinstance(outcome, CandidateAttemptRunBindingRefusal)
|
||||
assert RunAttemptBindingRefusalReason.RUN_RESULT_MISMATCH.value in outcome.reason_codes
|
||||
|
||||
|
||||
def test_refusal_on_attempt_id_mismatch(valid_run, valid_co_result):
|
||||
bad = SimpleNamespace(**{k: getattr(valid_co_result, k, None) for k in dir(valid_co_result) if not k.startswith("_")})
|
||||
setattr(bad, "attempt_id", "wrong_attempt")
|
||||
outcome = bind_candidate_attempt_to_run(original_run=valid_run, candidate_operator_result=bad)
|
||||
assert isinstance(outcome, CandidateAttemptRunBindingRefusal)
|
||||
assert RunAttemptBindingRefusalReason.ATTEMPT_RESULT_MISMATCH.value in outcome.reason_codes
|
||||
|
||||
|
||||
def test_refusal_on_candidate_digest_mismatch(valid_run, valid_co_result):
|
||||
ca = SimpleNamespace(**vars(valid_co_result.candidate_attempt))
|
||||
setattr(ca, "candidate_digest", "wrong_cand")
|
||||
bad = SimpleNamespace(**vars(valid_co_result))
|
||||
setattr(bad, "candidate_attempt", ca)
|
||||
setattr(bad, "candidate_digest", "wrong_cand")
|
||||
outcome = bind_candidate_attempt_to_run(original_run=valid_run, candidate_operator_result=bad)
|
||||
assert isinstance(outcome, CandidateAttemptRunBindingRefusal)
|
||||
assert RunAttemptBindingRefusalReason.ATTEMPT_RESULT_MISMATCH.value in outcome.reason_codes
|
||||
|
||||
|
||||
def test_refusal_on_replay_status_not_pending(valid_run, valid_co_result):
|
||||
ca = SimpleNamespace(**vars(valid_co_result.candidate_attempt))
|
||||
setattr(ca, "replay_status", "REPLAY_DONE")
|
||||
bad = SimpleNamespace(**vars(valid_co_result))
|
||||
setattr(bad, "candidate_attempt", ca)
|
||||
outcome = bind_candidate_attempt_to_run(original_run=valid_run, candidate_operator_result=bad)
|
||||
assert isinstance(outcome, CandidateAttemptRunBindingRefusal)
|
||||
assert RunAttemptBindingRefusalReason.REPLAY_STATUS_NOT_PENDING.value in outcome.reason_codes
|
||||
|
||||
|
||||
def test_refusal_on_replay_blockers_present(valid_run, valid_co_result):
|
||||
ca = SimpleNamespace(**vars(valid_co_result.candidate_attempt))
|
||||
setattr(ca, "replay_blockers", ("blocker",))
|
||||
bad = SimpleNamespace(**vars(valid_co_result))
|
||||
setattr(bad, "candidate_attempt", ca)
|
||||
outcome = bind_candidate_attempt_to_run(original_run=valid_run, candidate_operator_result=bad)
|
||||
assert isinstance(outcome, CandidateAttemptRunBindingRefusal)
|
||||
assert RunAttemptBindingRefusalReason.REPLAY_BLOCKERS_PRESENT.value in outcome.reason_codes
|
||||
|
||||
|
||||
def test_refusal_on_duplicate_attempt_index(valid_run, valid_co_result):
|
||||
existing = SimpleNamespace(attempt_index=valid_co_result.candidate_attempt.attempt_index, attempt_id="other", candidate_digest="other")
|
||||
run_with_dup = SimpleNamespace(**vars(valid_run))
|
||||
setattr(run_with_dup, "candidate_attempts", (existing,))
|
||||
outcome = bind_candidate_attempt_to_run(original_run=run_with_dup, candidate_operator_result=valid_co_result)
|
||||
assert isinstance(outcome, CandidateAttemptRunBindingRefusal)
|
||||
assert RunAttemptBindingRefusalReason.DUPLICATE_ATTEMPT_INDEX.value in outcome.reason_codes
|
||||
|
||||
|
||||
def test_refusal_on_budget_exceed(valid_run, valid_co_result):
|
||||
bc = SimpleNamespace(candidates=9999, steps=5)
|
||||
ca = SimpleNamespace(**vars(valid_co_result.candidate_attempt))
|
||||
setattr(ca, "budget_charge", bc)
|
||||
bad = SimpleNamespace(**vars(valid_co_result))
|
||||
setattr(bad, "candidate_attempt", ca)
|
||||
outcome = bind_candidate_attempt_to_run(original_run=valid_run, candidate_operator_result=bad)
|
||||
assert isinstance(outcome, CandidateAttemptRunBindingRefusal)
|
||||
assert RunAttemptBindingRefusalReason.BUDGET_CHARGE_EXCEEDS_REMAINING.value in outcome.reason_codes
|
||||
|
||||
|
||||
def test_refusal_on_operator_set_mismatch(valid_run, valid_co_result):
|
||||
cr = SimpleNamespace(**vars(valid_co_result.candidate_reconstruction))
|
||||
setattr(cr, "operator_set_id", "wrong")
|
||||
bad = SimpleNamespace(**vars(valid_co_result))
|
||||
setattr(bad, "candidate_reconstruction", cr)
|
||||
outcome = bind_candidate_attempt_to_run(original_run=valid_run, candidate_operator_result=bad)
|
||||
assert isinstance(outcome, CandidateAttemptRunBindingRefusal)
|
||||
assert RunAttemptBindingRefusalReason.OPERATOR_SET_MISMATCH.value in outcome.reason_codes
|
||||
|
||||
|
||||
def test_evidence_span_preservation(valid_run, valid_co_result):
|
||||
span = SimpleNamespace(source_id="s1")
|
||||
ca = SimpleNamespace(**vars(valid_co_result.candidate_attempt))
|
||||
setattr(ca, "evidence_spans", (span, span))
|
||||
bad = SimpleNamespace(**vars(valid_co_result))
|
||||
setattr(bad, "candidate_attempt", ca)
|
||||
outcome = bind_candidate_attempt_to_run(original_run=valid_run, candidate_operator_result=bad)
|
||||
assert isinstance(outcome, CandidateAttemptRunBinding)
|
||||
assert len(outcome.evidence_spans) == 2
|
||||
|
||||
|
||||
def test_binding_has_no_forbidden_fields(valid_run, valid_co_result):
|
||||
outcome = bind_candidate_attempt_to_run(original_run=valid_run, candidate_operator_result=valid_co_result)
|
||||
forbidden = {"answer", "proof", "verdict", "rank", "score", "selected", "promotion"}
|
||||
for f in forbidden:
|
||||
assert not hasattr(outcome, f)
|
||||
|
||||
|
||||
def test_static_guard_no_forbidden_code():
|
||||
src = (GENERATE_DIR / "run_attempt_binding.py").read_text()
|
||||
assert "import time" not in src
|
||||
assert "import random" not in src
|
||||
assert "initialize_geometric_search_run" not in src
|
||||
assert "project_contract_residuals" not in src
|
||||
|
||||
|
||||
def test_focused_and_smoke_still_pass(valid_run, valid_co_result):
|
||||
outcome = bind_candidate_attempt_to_run(original_run=valid_run, candidate_operator_result=valid_co_result)
|
||||
assert isinstance(outcome, CandidateAttemptRunBinding)
|
||||
|
||||
print("Full real test suite - no placeholders")
|
||||
assert not hasattr(outcome, "binding_id")
|
||||
assert not hasattr(outcome, "candidate_attempt_ref")
|
||||
|
|
|
|||
Loading…
Reference in a new issue