core-labs/core
1
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions
core/tests/test_candidate_operator.py
Shay 1d9e40602c feat(kernel): implement missing-role candidate operator shell 
Implements ADR-0231's inert missing_role_candidate.v1 operator shell for
direction_unbound residuals under unary_delta_transition. Adds deterministic
operator-set identity, candidate reconstruction with split digests,
CandidateAttempt-compatible output, typed refusals, and isolation tests.
2026-06-22 19:27:41 -07:00

1191 lines
No EOL
42 KiB
Python
Raw Blame History

from __future__ import annotations
import ast
import dataclasses
import hashlib
import json
from pathlib import Path
from types import SimpleNamespace
import pytest
from generate.candidate_operator import (
CANDIDATE_OPERATOR_POLICY_VERSION,
CANDIDATE_OPERATOR_SET_VERSION,
MISSING_ROLE_CANDIDATE_OPERATOR_NAME,
MISSING_ROLE_CANDIDATE_OPERATOR_VERSION,
MISSING_ROLE_CANDIDATE_ORGAN,
MISSING_ROLE_RESIDUAL_CODE,
MISSING_ROLE_RESIDUAL_KIND,
CandidateOperatorPolicy,
CandidateOperatorRefusal,
CandidateOperatorRefusalReason,
CandidateOperatorResult,
CandidateReconstruction,
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.search_gate import SearchGateDecision, SearchGateStatus
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(
*,
residual_id: str | None = None,
evidence_spans: tuple[SourceSpan, ...] | None = None,
residual_kind: ResidualKind = ResidualKind.MISSING_ROLE,
residual_code: str = MISSING_ROLE_RESIDUAL_CODE,
candidate_organ: str = MISSING_ROLE_CANDIDATE_ORGAN,
) -> ContractResidual:
spans = evidence_spans if evidence_spans is not None else (_span(),)
if residual_id is None:
payload = {
"candidate_organ": candidate_organ,
"residual_kind": residual_kind.value,
"residual_code": residual_code,
"evidence_spans": [_span_payload(span) for span in spans],
}
residual_id = _digest(payload)
return ContractResidual(
residual_id=residual_id,
candidate_organ=candidate_organ,
family_id="state_change.unary_delta",
residual_kind=residual_kind,
residual_code=residual_code,
source_axis=ResidualSourceAxis.ROLE,
evidence_spans=spans,
explanation="residual prose",
)
def _gate(
residual: ContractResidual,
*,
status: SearchGateStatus = SearchGateStatus.ELIGIBLE,
reason_code: str = "eligible_missing_role",
input_digest: str = "a" * 64,
) -> SearchGateDecision:
payload = {
"policy_version": "search_gate.v1",
"input_digest": input_digest,
"residual_ids": [residual.residual_id],
"candidate_organ": residual.candidate_organ,
"status": status.value,
"reason_code": reason_code,
"evidence_spans": [_span_payload(span) for span in residual.evidence_spans],
}
return SearchGateDecision(
decision_id=_digest(payload),
policy_version="search_gate.v1",
input_digest=input_digest,
residual_ids=(residual.residual_id,),
candidate_organ=residual.candidate_organ,
status=status,
reason_code=reason_code,
evidence_spans=residual.evidence_spans,
explanation="gate prose",
)
def _budget(
gate: SearchGateDecision,
*,
status: ComputeBudgetStatus = ComputeBudgetStatus.BUDGET_ALLOWED,
reason_code: str = "budget_allowed_missing_role",
max_candidates: int = 5,
max_depth: int = 2,
max_steps: int = 10,
max_parallelism: int = 1,
) -> 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": status.value,
"reason_code": reason_code,
"max_candidates": max_candidates,
"max_depth": max_depth,
"max_steps": max_steps,
"max_parallelism": max_parallelism,
"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=status,
reason_code=reason_code,
max_candidates=max_candidates,
max_depth=max_depth,
max_steps=max_steps,
max_wallclock_ms=None,
max_parallelism=max_parallelism,
evidence_spans=gate.evidence_spans,
explanation="budget prose",
)
def _run(
*,
residual: ContractResidual,
gate: SearchGateDecision,
budget: ComputeBudgetDecision,
problem_frame_digest: str = "f" * 64,
assessment_id: str = "assessment-a",
) -> GeometricSearchRun:
outcome = initialize_geometric_search_run(
problem_frame_digest=problem_frame_digest,
contract_assessment_id=assessment_id,
residual_ids=(residual.residual_id,),
gate_decision=gate,
compute_budget=budget,
operator_set_id=candidate_operator_set_id(),
operator_set_version=CANDIDATE_OPERATOR_SET_VERSION,
)
assert isinstance(outcome, GeometricSearchRun)
return outcome
def _cue(
*,
direction: str = "increase",
evidence_spans: tuple[SourceSpan, ...] | None = None,
) -> GroundedUnaryDeltaCue:
return GroundedUnaryDeltaCue(
direction=direction,
evidence_spans=evidence_spans if evidence_spans is not None else (_span(),),
)
def _chain() -> tuple[
ContractResidual,
SearchGateDecision,
ComputeBudgetDecision,
GeometricSearchRun,
GroundedUnaryDeltaCue,
]:
residual = _residual()
gate = _gate(residual)
budget = _budget(gate)
run = _run(residual=residual, gate=gate, budget=budget)
cue = _cue(evidence_spans=residual.evidence_spans)
return residual, gate, budget, run, cue
def _build(**changes: object) -> CandidateOperatorResult | CandidateOperatorRefusal:
residual, gate, budget, run, cue = _chain()
values: dict[str, object] = {
"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,
"grounded_unary_delta_cues": (cue,),
}
values.update(changes)
return build_missing_role_candidate(**values) # type: ignore[arg-type]
def _expected_operator_set_id() -> str:
payload = {
"operator_set_version": CANDIDATE_OPERATOR_SET_VERSION,
"operators": [
{
"operator_family": "residual_missing_role",
"operator_name": MISSING_ROLE_CANDIDATE_OPERATOR_NAME,
"operator_version": MISSING_ROLE_CANDIDATE_OPERATOR_VERSION,
"allowed_residual_kinds": [MISSING_ROLE_RESIDUAL_KIND],
"allowed_residual_codes": [MISSING_ROLE_RESIDUAL_CODE],
"allowed_candidate_organs": [MISSING_ROLE_CANDIDATE_ORGAN],
"max_attempts_per_run": 1,
"budget_charge": {"candidates": 1, "steps": 1},
"depth": 1,
"max_parallelism": 1,
}
],
"schema_versions": [],
"policy_versions": [],
}
return _digest(payload)
def _expected_input_digest(
*,
problem_frame_digest: str,
original_contract_assessment_id: str,
residual_id: str,
search_gate_decision_id: str,
compute_budget_id: str,
geometric_search_run_id: str,
attempt_index: int = 0,
schema_versions: tuple[tuple[str, str], ...] = (),
policy_versions: tuple[tuple[str, str], ...] = (),
) -> str:
payload = {
"operator_policy_version": CANDIDATE_OPERATOR_POLICY_VERSION,
"operator_name": MISSING_ROLE_CANDIDATE_OPERATOR_NAME,
"operator_version": MISSING_ROLE_CANDIDATE_OPERATOR_VERSION,
"problem_frame_digest": problem_frame_digest,
"original_contract_assessment_id": original_contract_assessment_id,
"residual_id": residual_id,
"residual_kind": MISSING_ROLE_RESIDUAL_KIND,
"residual_code": MISSING_ROLE_RESIDUAL_CODE,
"candidate_organ": MISSING_ROLE_CANDIDATE_ORGAN,
"search_gate_decision_id": search_gate_decision_id,
"compute_budget_id": compute_budget_id,
"geometric_search_run_id": geometric_search_run_id,
"operator_set_id": _expected_operator_set_id(),
"operator_set_version": CANDIDATE_OPERATOR_SET_VERSION,
"attempt_index": attempt_index,
"schema_versions": [[name, version] for name, version in schema_versions],
"policy_versions": [[name, version] for name, version in policy_versions],
}
return _digest(payload)
def _expected_candidate_payload(direction: str) -> list[list[str]]:
return [
["candidate_organ", MISSING_ROLE_CANDIDATE_ORGAN],
["direction", direction],
["kind", "role_binding_delta"],
["relation_type", "state_change.unary_delta"],
["role", "direction"],
["source", "GroundedUnaryDeltaCue.direction"],
]
def _expected_candidate_digest(
*,
problem_frame_digest: str,
direction: str,
evidence_spans: tuple[SourceSpan, ...],
) -> str:
payload = {
"problem_frame_digest": problem_frame_digest,
"candidate_organ": MISSING_ROLE_CANDIDATE_ORGAN,
"candidate_payload": _expected_candidate_payload(direction),
"evidence_spans": [_span_payload(span) for span in evidence_spans],
}
return _digest(payload)
def _expected_attempt_id(
*,
attempt_index: int,
input_digest: str,
candidate_digest: str,
evidence_spans: tuple[SourceSpan, ...],
) -> str:
payload = {
"attempt_id": "",
"attempt_index": attempt_index,
"parent_attempt_id": None,
"operator_id": MISSING_ROLE_CANDIDATE_OPERATOR_NAME,
"operator_version": MISSING_ROLE_CANDIDATE_OPERATOR_VERSION,
"input_digest": input_digest,
"candidate_digest": candidate_digest,
"budget_charge": {"candidates": 1, "steps": 1},
"depth": 1,
"step_index": attempt_index,
"replay_status": CandidateReplayStatus.REPLAY_PENDING.value,
"replay_blockers": [],
"evidence_spans": [_span_payload(span) for span in evidence_spans],
}
return _digest(payload)
def _expected_reconstruction_digest(
*,
geometric_search_run_id: str,
attempt_id: str,
attempt_index: int,
residual_id: str,
candidate_digest: str,
evidence_spans: tuple[SourceSpan, ...],
schema_versions: tuple[tuple[str, str], ...] = (),
policy_versions: tuple[tuple[str, str], ...] = (),
) -> str:
payload = {
"candidate_reconstruction_digest": "",
"geometric_search_run_id": geometric_search_run_id,
"attempt_id": attempt_id,
"attempt_index": attempt_index,
"operator_set_id": _expected_operator_set_id(),
"operator_set_version": CANDIDATE_OPERATOR_SET_VERSION,
"operator_name": MISSING_ROLE_CANDIDATE_OPERATOR_NAME,
"operator_version": MISSING_ROLE_CANDIDATE_OPERATOR_VERSION,
"residual_id": residual_id,
"residual_kind": MISSING_ROLE_RESIDUAL_KIND,
"residual_code": MISSING_ROLE_RESIDUAL_CODE,
"candidate_digest": candidate_digest,
"evidence_spans": [_span_payload(span) for span in evidence_spans],
"schema_versions": [[name, version] for name, version in schema_versions],
"policy_versions": [[name, version] for name, version in policy_versions],
}
return _digest(payload)
def _expected_result_id(
*,
input_digest: str,
geometric_search_run_id: str,
attempt_id: str,
attempt_index: int,
candidate_digest: str,
candidate_reconstruction_digest: str,
evidence_spans: tuple[SourceSpan, ...],
) -> str:
payload = {
"operator_result_id": "",
"operator_policy_version": CANDIDATE_OPERATOR_POLICY_VERSION,
"input_digest": input_digest,
"geometric_search_run_id": geometric_search_run_id,
"attempt_id": attempt_id,
"attempt_index": attempt_index,
"candidate_digest": candidate_digest,
"candidate_reconstruction_digest": candidate_reconstruction_digest,
"candidate_organ": MISSING_ROLE_CANDIDATE_ORGAN,
"operator_name": MISSING_ROLE_CANDIDATE_OPERATOR_NAME,
"operator_version": MISSING_ROLE_CANDIDATE_OPERATOR_VERSION,
"reason_codes": [],
"evidence_spans": [_span_payload(span) for span in evidence_spans],
}
return _digest(payload)
def _expected_refusal_id(
*,
input_digest: str | None,
geometric_search_run_id: str | None,
residual_id: str | None,
reason_codes: tuple[str, ...],
) -> str:
payload = {
"operator_refusal_id": "",
"operator_policy_version": CANDIDATE_OPERATOR_POLICY_VERSION,
"input_digest": input_digest,
"geometric_search_run_id": geometric_search_run_id,
"residual_id": residual_id,
"operator_name": MISSING_ROLE_CANDIDATE_OPERATOR_NAME,
"reason_codes": list(reason_codes),
}
return _digest(payload)
def test_public_api_exports_are_exact() -> None:
import generate.candidate_operator as candidate_operator
assert CANDIDATE_OPERATOR_POLICY_VERSION == "candidate_operator.v1"
assert CANDIDATE_OPERATOR_SET_VERSION == "candidate_operators.v1"
assert MISSING_ROLE_CANDIDATE_OPERATOR_NAME == "missing_role_candidate"
assert MISSING_ROLE_CANDIDATE_OPERATOR_VERSION == "missing_role_candidate.v1"
assert MISSING_ROLE_RESIDUAL_KIND == "missing_role"
assert MISSING_ROLE_RESIDUAL_CODE == "direction_unbound"
assert MISSING_ROLE_CANDIDATE_ORGAN == "unary_delta_transition"
assert tuple(candidate_operator.__all__) == (
"CANDIDATE_OPERATOR_POLICY_VERSION",
"CANDIDATE_OPERATOR_SET_VERSION",
"MISSING_ROLE_CANDIDATE_OPERATOR_NAME",
"MISSING_ROLE_CANDIDATE_OPERATOR_VERSION",
"MISSING_ROLE_RESIDUAL_KIND",
"MISSING_ROLE_RESIDUAL_CODE",
"MISSING_ROLE_CANDIDATE_ORGAN",
"CandidateOperatorRefusalReason",
"CandidateOperatorPolicy",
"GroundedUnaryDeltaCue",
"CandidateOperatorInput",
"CandidateReconstruction",
"CandidateOperatorResult",
"CandidateOperatorRefusal",
"CandidateOperatorOutcome",
"candidate_operator_set_id",
"build_missing_role_candidate",
)
def test_operator_set_identity_is_deterministic() -> None:
first = candidate_operator_set_id()
second = candidate_operator_set_id()
assert first == second == _expected_operator_set_id()
assert len(first) == 64
def test_valid_chain_produces_candidate_operator_result() -> None:
outcome = _build()
assert isinstance(outcome, CandidateOperatorResult)
assert outcome.operator_name == MISSING_ROLE_CANDIDATE_OPERATOR_NAME
assert outcome.operator_version == MISSING_ROLE_CANDIDATE_OPERATOR_VERSION
assert outcome.candidate_organ == MISSING_ROLE_CANDIDATE_ORGAN
assert outcome.reason_codes == ()
def test_candidate_digest_differs_from_reconstruction_digest() -> None:
outcome = _build()
assert isinstance(outcome, CandidateOperatorResult)
assert outcome.candidate_digest != outcome.candidate_reconstruction_digest
def test_candidate_reconstruction_binds_upstream_identity() -> None:
residual, gate, budget, run, cue = _chain()
outcome = _build()
assert isinstance(outcome, CandidateOperatorResult)
reconstruction = outcome.candidate_reconstruction
assert reconstruction.source_residual_id == residual.residual_id
assert reconstruction.problem_frame_digest == run.problem_frame_digest
assert reconstruction.original_contract_assessment_id == run.contract_assessment_id
assert reconstruction.operator_name == MISSING_ROLE_CANDIDATE_OPERATOR_NAME
assert reconstruction.operator_version == MISSING_ROLE_CANDIDATE_OPERATOR_VERSION
assert reconstruction.candidate_digest == outcome.candidate_digest
assert ("operator_set_id", candidate_operator_set_id()) in reconstruction.operator_provenance
assert outcome.geometric_search_run_id == run.run_id
assert outcome.attempt_id == outcome.candidate_attempt.attempt_id
assert gate.decision_id in (run.gate_decision_id, gate.decision_id)
assert budget.budget_id == run.budget_id
def test_candidate_attempt_fields_are_replay_pending() -> None:
residual, _, _, _, cue = _chain()
outcome = _build()
assert isinstance(outcome, CandidateOperatorResult)
attempt = outcome.candidate_attempt
assert attempt.attempt_index == 0
assert attempt.operator_id == MISSING_ROLE_CANDIDATE_OPERATOR_NAME
assert attempt.operator_version == MISSING_ROLE_CANDIDATE_OPERATOR_VERSION
assert attempt.replay_status is CandidateReplayStatus.REPLAY_PENDING
assert attempt.replay_blockers == ()
assert attempt.budget_charge == BudgetCharge(candidates=1, steps=1)
assert attempt.depth == 1
assert attempt.candidate_digest == outcome.candidate_digest
assert attempt.evidence_spans == cue.evidence_spans
def test_unsupported_residual_kind_refuses() -> None:
residual = _residual(residual_kind=ResidualKind.MISSING_RELATION)
gate = _gate(residual, reason_code="eligible_missing_relation")
budget = _budget(gate, reason_code="budget_allowed_missing_relation")
run = _run(residual=residual, gate=gate, budget=budget)
outcome = 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,
grounded_unary_delta_cues=(_cue(evidence_spans=residual.evidence_spans),),
)
assert isinstance(outcome, CandidateOperatorRefusal)
assert CandidateOperatorRefusalReason.UNSUPPORTED_RESIDUAL_KIND.value in outcome.reason_codes
def test_unsupported_residual_code_refuses() -> None:
residual = _residual(residual_code="quantity_unbound")
gate = _gate(residual)
budget = _budget(gate)
run = _run(residual=residual, gate=gate, budget=budget)
outcome = 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,
grounded_unary_delta_cues=(_cue(evidence_spans=residual.evidence_spans),),
)
assert isinstance(outcome, CandidateOperatorRefusal)
assert CandidateOperatorRefusalReason.UNSUPPORTED_RESIDUAL_CODE.value in outcome.reason_codes
def test_unsupported_candidate_organ_refuses() -> None:
residual = _residual(candidate_organ="fraction_decrease")
gate = _gate(residual)
budget = _budget(gate)
run = _run(residual=residual, gate=gate, budget=budget)
outcome = 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,
grounded_unary_delta_cues=(_cue(evidence_spans=residual.evidence_spans),),
)
assert isinstance(outcome, CandidateOperatorRefusal)
assert CandidateOperatorRefusalReason.UNSUPPORTED_CANDIDATE_ORGAN.value in outcome.reason_codes
@pytest.mark.parametrize(
"status",
(
SearchGateStatus.BLOCKED,
SearchGateStatus.INELIGIBLE,
SearchGateStatus.UNASSESSABLE,
),
)
def test_ineligible_search_gate_refuses(status: SearchGateStatus) -> None:
residual, valid_gate, valid_budget, run, cue = _chain()
gate = _gate(residual, status=status, reason_code=f"{status.value}_reason")
outcome = build_missing_role_candidate(
residual=residual,
search_gate=gate,
compute_budget=valid_budget,
run=run,
problem_frame_digest=run.problem_frame_digest,
original_contract_assessment_id=run.contract_assessment_id,
grounded_unary_delta_cues=(cue,),
)
assert isinstance(outcome, CandidateOperatorRefusal)
assert CandidateOperatorRefusalReason.INELIGIBLE_SEARCH_GATE.value in outcome.reason_codes
@pytest.mark.parametrize(
"status",
(
ComputeBudgetStatus.BUDGET_BLOCKED,
ComputeBudgetStatus.BUDGET_ZERO,
ComputeBudgetStatus.BUDGET_UNASSESSABLE,
),
)
def test_non_allowed_budget_refuses(status: ComputeBudgetStatus) -> None:
residual, gate, _, run, cue = _chain()
budget = _budget(
gate,
status=status,
reason_code=f"{status.value}_reason",
max_candidates=0,
max_depth=0,
max_steps=0,
max_parallelism=0,
)
outcome = 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,
grounded_unary_delta_cues=(cue,),
)
assert isinstance(outcome, CandidateOperatorRefusal)
assert CandidateOperatorRefusalReason.NON_ALLOWED_BUDGET.value in outcome.reason_codes
def test_gate_budget_mismatch_refuses() -> None:
residual, gate, _, run, cue = _chain()
budget = dataclasses.replace(_budget(gate), gate_decision_id="mismatch" * 8)
outcome = 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,
grounded_unary_delta_cues=(cue,),
)
assert isinstance(outcome, CandidateOperatorRefusal)
assert CandidateOperatorRefusalReason.GATE_BUDGET_MISMATCH.value in outcome.reason_codes
def test_run_gate_mismatch_refuses() -> None:
residual, gate, budget, run, cue = _chain()
run = dataclasses.replace(run, gate_decision_id="mismatch" * 8)
outcome = 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,
grounded_unary_delta_cues=(cue,),
)
assert isinstance(outcome, CandidateOperatorRefusal)
assert CandidateOperatorRefusalReason.RUN_GATE_MISMATCH.value in outcome.reason_codes
def test_run_budget_mismatch_refuses() -> None:
residual, gate, budget, run, cue = _chain()
run = dataclasses.replace(run, budget_id="mismatch" * 8)
outcome = 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,
grounded_unary_delta_cues=(cue,),
)
assert isinstance(outcome, CandidateOperatorRefusal)
assert CandidateOperatorRefusalReason.RUN_BUDGET_MISMATCH.value in outcome.reason_codes
def test_operator_set_mismatch_refuses() -> None:
residual, gate, budget, run, cue = _chain()
run = dataclasses.replace(run, operator_set_id="wrong" * 8)
outcome = 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,
grounded_unary_delta_cues=(cue,),
)
assert isinstance(outcome, CandidateOperatorRefusal)
assert CandidateOperatorRefusalReason.OPERATOR_SET_MISMATCH.value in outcome.reason_codes
def test_attempt_index_beyond_max_candidates_refuses() -> None:
residual = _residual()
gate = _gate(residual)
budget = _budget(gate, max_candidates=1)
run = _run(residual=residual, gate=gate, budget=budget)
outcome = 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,
grounded_unary_delta_cues=(_cue(evidence_spans=residual.evidence_spans),),
attempt_index=1,
)
assert isinstance(outcome, CandidateOperatorRefusal)
assert (
CandidateOperatorRefusalReason.ATTEMPT_INDEX_EXCEEDS_BUDGET.value
in outcome.reason_codes
)
def test_attempt_index_beyond_operator_policy_refuses() -> None:
outcome = _build(attempt_index=1)
assert isinstance(outcome, CandidateOperatorRefusal)
assert (
CandidateOperatorRefusalReason.ATTEMPT_INDEX_EXCEEDS_OPERATOR_POLICY.value
in outcome.reason_codes
)
def test_budget_charge_exceeds_structural_budget_refuses() -> None:
residual = _residual()
gate = _gate(residual)
budget = _budget(gate, max_depth=0)
run = _run(residual=residual, gate=gate, budget=budget)
outcome = 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,
grounded_unary_delta_cues=(_cue(evidence_spans=residual.evidence_spans),),
)
assert isinstance(outcome, CandidateOperatorRefusal)
assert (
CandidateOperatorRefusalReason.BUDGET_CHARGE_EXCEEDS_BUDGET.value
in outcome.reason_codes
)
def test_non_serial_budget_refuses() -> None:
residual, gate, _, run, cue = _chain()
budget = _budget(gate, max_parallelism=2)
outcome = 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,
grounded_unary_delta_cues=(cue,),
)
assert isinstance(outcome, CandidateOperatorRefusal)
assert CandidateOperatorRefusalReason.NON_SERIAL_BUDGET.value in outcome.reason_codes
def test_missing_typed_cue_refuses() -> None:
outcome = _build(grounded_unary_delta_cues=())
assert isinstance(outcome, CandidateOperatorRefusal)
assert CandidateOperatorRefusalReason.MISSING_TYPED_CUE.value in outcome.reason_codes
def test_ambiguous_multiple_typed_cues_refuse() -> None:
residual, gate, budget, run, cue = _chain()
second = _cue(direction="decrease", evidence_spans=residual.evidence_spans)
outcome = 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,
grounded_unary_delta_cues=(cue, second),
)
assert isinstance(outcome, CandidateOperatorRefusal)
assert CandidateOperatorRefusalReason.AMBIGUOUS_TYPED_CUE.value in outcome.reason_codes
def test_empty_cue_direction_refuses() -> None:
residual, gate, budget, run, _ = _chain()
cue = GroundedUnaryDeltaCue(direction=" ", evidence_spans=residual.evidence_spans)
outcome = 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,
grounded_unary_delta_cues=(cue,),
)
assert isinstance(outcome, CandidateOperatorRefusal)
assert CandidateOperatorRefusalReason.MISSING_TYPED_CUE.value in outcome.reason_codes
def test_unsupported_typed_cue_direction_refuses() -> None:
residual, gate, budget, run, _ = _chain()
cue = GroundedUnaryDeltaCue(direction="sideways", evidence_spans=residual.evidence_spans)
outcome = 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,
grounded_unary_delta_cues=(cue,),
)
assert isinstance(outcome, CandidateOperatorRefusal)
assert CandidateOperatorRefusalReason.UNSUPPORTED_TYPED_CUE.value in outcome.reason_codes
def test_malformed_evidence_span_refuses() -> None:
residual, gate, budget, run, _ = _chain()
bad_span = SimpleNamespace(text="bad", start=-1, end=3, sentence_index=0)
cue = GroundedUnaryDeltaCue(direction="increase", evidence_spans=(bad_span,)) # type: ignore[arg-type]
outcome = 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,
grounded_unary_delta_cues=(cue,),
)
assert isinstance(outcome, CandidateOperatorRefusal)
assert CandidateOperatorRefusalReason.MALFORMED_EVIDENCE_SPAN.value in outcome.reason_codes
def test_ungrounded_cue_span_refuses() -> None:
residual, gate, budget, run, _ = _chain()
cue = _cue(evidence_spans=(_span(text="other", start=99, end=104),))
outcome = 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,
grounded_unary_delta_cues=(cue,),
)
assert isinstance(outcome, CandidateOperatorRefusal)
assert CandidateOperatorRefusalReason.MALFORMED_EVIDENCE_SPAN.value in outcome.reason_codes
def test_duplicate_evidence_spans_are_preserved() -> None:
span = _span()
residual = _residual(evidence_spans=(span, span))
gate = _gate(residual)
budget = _budget(gate)
run = _run(residual=residual, gate=gate, budget=budget)
cue = GroundedUnaryDeltaCue(direction="increase", evidence_spans=(span, span))
outcome = 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,
grounded_unary_delta_cues=(cue,),
)
assert isinstance(outcome, CandidateOperatorResult)
assert outcome.evidence_spans == (span, span)
assert outcome.candidate_attempt.evidence_spans == (span, span)
def test_evidence_span_reorder_changes_candidate_identity() -> None:
first_span = _span(text="one", start=0, end=3)
second_span = _span(text="two", start=4, end=7)
residual = _residual(evidence_spans=(first_span, second_span))
gate = _gate(residual)
budget = _budget(gate)
run = _run(residual=residual, gate=gate, budget=budget)
first = 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,
grounded_unary_delta_cues=(
GroundedUnaryDeltaCue(direction="increase", evidence_spans=(first_span, second_span)),
),
)
second = 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,
grounded_unary_delta_cues=(
GroundedUnaryDeltaCue(direction="increase", evidence_spans=(second_span, first_span)),
),
)
assert isinstance(first, CandidateOperatorResult)
assert isinstance(second, CandidateOperatorResult)
assert first.candidate_digest != second.candidate_digest
assert first.attempt_id != second.attempt_id
assert first.operator_result_id != second.operator_result_id
def test_explanation_changes_do_not_affect_ids() -> None:
first = _build(explanation="first prose")
second = _build(explanation="second prose")
assert isinstance(first, CandidateOperatorResult)
assert isinstance(second, CandidateOperatorResult)
assert first.input_digest == second.input_digest
assert first.candidate_digest == second.candidate_digest
assert first.candidate_reconstruction_digest == second.candidate_reconstruction_digest
assert first.attempt_id == second.attempt_id
assert first.operator_result_id == second.operator_result_id
def test_refusal_is_not_partial_candidate() -> None:
outcome = _build(grounded_unary_delta_cues=())
assert isinstance(outcome, CandidateOperatorRefusal)
assert not hasattr(outcome, "candidate_attempt")
assert not hasattr(outcome, "candidate_reconstruction")
assert not hasattr(outcome, "candidate_digest")
def test_public_dataclasses_expose_no_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",
"best",
"selected_candidate",
"best_candidate",
"serving_allowed",
"runnable",
}
from generate.candidate_operator import CandidateOperatorInput
for record_type in (
CandidateOperatorPolicy,
GroundedUnaryDeltaCue,
CandidateOperatorInput,
CandidateReconstruction,
CandidateOperatorResult,
CandidateOperatorRefusal,
):
assert forbidden.isdisjoint({field.name for field in dataclasses.fields(record_type)})
def test_candidate_result_cannot_represent_replay_closure() -> None:
outcome = _build()
assert isinstance(outcome, CandidateOperatorResult)
assert not hasattr(outcome, "replay_disposition")
assert not hasattr(outcome, "contract_replay_assessment_id")
assert not hasattr(outcome, "trace_id")
assert outcome.candidate_attempt.replay_status is CandidateReplayStatus.REPLAY_PENDING
def test_canonical_ids_match_independent_recomputation() -> None:
residual, gate, budget, run, cue = _chain()
outcome = 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,
grounded_unary_delta_cues=(cue,),
)
assert isinstance(outcome, CandidateOperatorResult)
input_digest = _expected_input_digest(
problem_frame_digest=run.problem_frame_digest,
original_contract_assessment_id=run.contract_assessment_id,
residual_id=residual.residual_id,
search_gate_decision_id=gate.decision_id,
compute_budget_id=budget.budget_id,
geometric_search_run_id=run.run_id,
)
candidate_digest = _expected_candidate_digest(
problem_frame_digest=run.problem_frame_digest,
direction="increase",
evidence_spans=cue.evidence_spans,
)
attempt_id = _expected_attempt_id(
attempt_index=0,
input_digest=input_digest,
candidate_digest=candidate_digest,
evidence_spans=cue.evidence_spans,
)
reconstruction_digest = _expected_reconstruction_digest(
geometric_search_run_id=run.run_id,
attempt_id=attempt_id,
attempt_index=0,
residual_id=residual.residual_id,
candidate_digest=candidate_digest,
evidence_spans=cue.evidence_spans,
)
result_id = _expected_result_id(
input_digest=input_digest,
geometric_search_run_id=run.run_id,
attempt_id=attempt_id,
attempt_index=0,
candidate_digest=candidate_digest,
candidate_reconstruction_digest=reconstruction_digest,
evidence_spans=cue.evidence_spans,
)
assert outcome.input_digest == input_digest
assert outcome.candidate_digest == candidate_digest
assert outcome.candidate_reconstruction_digest == reconstruction_digest
assert outcome.attempt_id == attempt_id
assert outcome.operator_result_id == result_id
def test_refusal_id_matches_canonical_payload() -> None:
outcome = _build(grounded_unary_delta_cues=())
assert isinstance(outcome, CandidateOperatorRefusal)
assert outcome.operator_refusal_id == _expected_refusal_id(
input_digest=outcome.input_digest,
geometric_search_run_id=outcome.geometric_search_run_id,
residual_id=outcome.residual_id,
reason_codes=outcome.reason_codes,
)
def test_public_api_has_no_source_text_parameter() -> None:
import inspect
signature = inspect.signature(build_missing_role_candidate)
assert "problem_text" not in signature.parameters
assert "source_text" not in signature.parameters
assert "raw_text" not in signature.parameters
def test_module_has_no_regex_or_template_matching() -> None:
source = Path("generate/candidate_operator.py").read_text("utf-8")
forbidden_fragments = (
"import re",
"from re",
"re.search",
"re.match",
"re.findall",
"regex",
"template",
)
lowered = source.lower()
assert not any(fragment in lowered for fragment in forbidden_fragments)
def test_module_coupling_and_side_effect_guards() -> None:
path = Path("generate/candidate_operator.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.compute_budget",
"generate.geometric_search_run",
"generate.kernel_facts",
"generate.search_gate",
}
assert {
"initialize_geometric_search_run",
"decide_search_gate",
"decide_compute_budget",
"project_contract_residuals",
"assess_contracts",
"build_replay_adapter_input",
"classify_replay_result",
"build_practice_trace_input",
"seal_practice_trace",
"determine",
"build_problem_frame",
}.isdisjoint(imported_names)
assert {
"initialize_geometric_search_run",
"decide_search_gate",
"decide_compute_budget",
"project_contract_residuals",
"assess_contracts",
"build_replay_adapter_input",
"classify_replay_result",
"build_practice_trace_input",
"seal_practice_trace",
"determine",
"repair",
"serve",
"store",
"write",
"open",
"write_text",
"write_bytes",
"request",
"urlopen",
"sleep",
"time",
"uuid4",
}.isdisjoint(calls)
for upstream in (
"generate/geometric_search_run.py",
"generate/replay_adapter.py",
"generate/sealed_practice_trace.py",
"generate/search_gate.py",
"generate/compute_budget.py",
"generate/contract_residuals.py",
):
assert "candidate_operator" not in Path(upstream).read_text("utf-8")
def test_no_filesystem_network_clock_random_or_environment_identity() -> None:
source = Path("generate/candidate_operator.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)
def test_malformed_objects_fail_closed_without_throwing() -> None:
outcome = build_missing_role_candidate(
residual=SimpleNamespace(), # type: ignore[arg-type]
search_gate=SimpleNamespace(), # type: ignore[arg-type]
compute_budget=SimpleNamespace(), # type: ignore[arg-type]
run=SimpleNamespace(), # type: ignore[arg-type]
problem_frame_digest="f" * 64,
original_contract_assessment_id="assessment-a",
grounded_unary_delta_cues=(),
)
assert isinstance(outcome, CandidateOperatorRefusal)
def test_only_one_operator_row_in_static_set() -> None:
payload = json.loads(
json.dumps(
{
"operator_set_version": CANDIDATE_OPERATOR_SET_VERSION,
"operators": [
{
"operator_family": "residual_missing_role",
"operator_name": MISSING_ROLE_CANDIDATE_OPERATOR_NAME,
"operator_version": MISSING_ROLE_CANDIDATE_OPERATOR_VERSION,
"allowed_residual_kinds": [MISSING_ROLE_RESIDUAL_KIND],
"allowed_residual_codes": [MISSING_ROLE_RESIDUAL_CODE],
"allowed_candidate_organs": [MISSING_ROLE_CANDIDATE_ORGAN],
"max_attempts_per_run": 1,
"budget_charge": {"candidates": 1, "steps": 1},
"depth": 1,
"max_parallelism": 1,
}
],
"schema_versions": [],
"policy_versions": [],
},
sort_keys=True,
separators=(",", ":"),
)
)
assert len(payload["operators"]) == 1
assert payload["operators"][0]["operator_name"] == "missing_role_candidate"
Reference in a new issue View git blame Copy permalink