From 1d9e40602c7049f5d704142245858cf5100449a6 Mon Sep 17 00:00:00 2001 From: Shay Date: Mon, 22 Jun 2026 19:27:41 -0700 Subject: [PATCH] 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. --- generate/candidate_operator.py | 946 ++++++++++++++++++++++++ tests/test_candidate_operator.py | 1191 ++++++++++++++++++++++++++++++ 2 files changed, 2137 insertions(+) create mode 100644 generate/candidate_operator.py create mode 100644 tests/test_candidate_operator.py diff --git a/generate/candidate_operator.py b/generate/candidate_operator.py new file mode 100644 index 00000000..8cd42dc3 --- /dev/null +++ b/generate/candidate_operator.py @@ -0,0 +1,946 @@ +"""Inert, diagnostic-only missing-role candidate operator shell.""" + +from __future__ import annotations + +import hashlib +import json +from dataclasses import dataclass +from enum import Enum, unique + +from generate.compute_budget import ComputeBudgetDecision, ComputeBudgetStatus +from generate.geometric_search_run import ( + BudgetCharge, + CandidateAttempt, + CandidateReplayStatus, + GeometricSearchRun, +) +from generate.kernel_facts import SourceSpan +from generate.search_gate import SearchGateDecision, SearchGateStatus + +CANDIDATE_OPERATOR_POLICY_VERSION = "candidate_operator.v1" +CANDIDATE_OPERATOR_SET_VERSION = "candidate_operators.v1" + +MISSING_ROLE_CANDIDATE_OPERATOR_NAME = "missing_role_candidate" +MISSING_ROLE_CANDIDATE_OPERATOR_VERSION = "missing_role_candidate.v1" + +MISSING_ROLE_RESIDUAL_KIND = "missing_role" +MISSING_ROLE_RESIDUAL_CODE = "direction_unbound" +MISSING_ROLE_CANDIDATE_ORGAN = "unary_delta_transition" + +_OPERATOR_FAMILY = "residual_missing_role" +_SUPPORTED_DIRECTIONS = frozenset({"increase", "decrease"}) +_FIXED_BUDGET_CHARGE = BudgetCharge(candidates=1, steps=1) +_OPERATOR_DEPTH = 1 +_OPERATOR_MAX_PARALLELISM = 1 +_DETERMINISM_REQUIREMENTS = ( + "canonical_json_sha256", + "static_operator_table", + "no_source_reread", + "no_stochastic_generation", +) +_FORBIDDEN_AUTHORITY_PATHS = ( + "answer_production", + "contract_replay", + "proof_replay", + "sealed_trace", + "serving", + "mutation", + "promotion", + "workbench", + "repair", + "open_ended_search", +) + + +@unique +class CandidateOperatorRefusalReason(str, Enum): + UNSUPPORTED_OPERATOR_POLICY = "unsupported_operator_policy" + UNSUPPORTED_OPERATOR = "unsupported_operator" + UNSUPPORTED_RESIDUAL_KIND = "unsupported_residual_kind" + UNSUPPORTED_RESIDUAL_CODE = "unsupported_residual_code" + UNSUPPORTED_CANDIDATE_ORGAN = "unsupported_candidate_organ" + INELIGIBLE_SEARCH_GATE = "ineligible_search_gate" + NON_ALLOWED_BUDGET = "non_allowed_budget" + GATE_BUDGET_MISMATCH = "gate_budget_mismatch" + RUN_GATE_MISMATCH = "run_gate_mismatch" + RUN_BUDGET_MISMATCH = "run_budget_mismatch" + OPERATOR_SET_MISMATCH = "operator_set_mismatch" + ATTEMPT_INDEX_EXCEEDS_BUDGET = "attempt_index_exceeds_budget" + ATTEMPT_INDEX_EXCEEDS_OPERATOR_POLICY = "attempt_index_exceeds_operator_policy" + BUDGET_CHARGE_EXCEEDS_BUDGET = "budget_charge_exceeds_budget" + NON_SERIAL_BUDGET = "non_serial_budget" + MISSING_TYPED_CUE = "missing_typed_cue" + AMBIGUOUS_TYPED_CUE = "ambiguous_typed_cue" + UNSUPPORTED_TYPED_CUE = "unsupported_typed_cue" + MALFORMED_EVIDENCE_SPAN = "malformed_evidence_span" + INVALID_OPERATOR_INPUT = "invalid_operator_input" + UNSUPPORTED_SCHEMA_VERSION = "unsupported_schema_version" + + +@dataclass(frozen=True, slots=True) +class CandidateOperatorPolicy: + operator_policy_version: str + operator_family: str + operator_name: str + operator_version: str + allowed_residual_kinds: tuple[str, ...] + allowed_residual_codes: tuple[str, ...] + allowed_candidate_organs: tuple[str, ...] + max_attempts_per_run: int + budget_charge: BudgetCharge + depth: int + max_parallelism: int + determinism_requirements: tuple[str, ...] + forbidden_authority_paths: tuple[str, ...] + + +MISSING_ROLE_OPERATOR_POLICY = CandidateOperatorPolicy( + operator_policy_version=CANDIDATE_OPERATOR_POLICY_VERSION, + operator_family=_OPERATOR_FAMILY, + 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=_FIXED_BUDGET_CHARGE, + depth=_OPERATOR_DEPTH, + max_parallelism=_OPERATOR_MAX_PARALLELISM, + determinism_requirements=_DETERMINISM_REQUIREMENTS, + forbidden_authority_paths=_FORBIDDEN_AUTHORITY_PATHS, +) + + +@dataclass(frozen=True, slots=True) +class GroundedUnaryDeltaCue: + direction: str + evidence_spans: tuple[SourceSpan, ...] + + +@dataclass(frozen=True, slots=True) +class CandidateOperatorInput: + input_digest: str + operator_policy_version: str + operator_name: str + operator_version: str + problem_frame_digest: str + original_contract_assessment_id: str + residual_id: str + residual_kind: str + residual_code: str + candidate_organ: str + search_gate_decision_id: str + compute_budget_id: str + geometric_search_run_id: str + operator_set_id: str + operator_set_version: str + attempt_index: int + schema_versions: tuple[tuple[str, str], ...] + policy_versions: tuple[tuple[str, str], ...] + + +@dataclass(frozen=True, slots=True) +class CandidateReconstruction: + candidate_digest: str + candidate_reconstruction_digest: str + candidate_organ: str + candidate_payload: tuple[tuple[str, str], ...] + evidence_spans: tuple[SourceSpan, ...] + operator_name: str + operator_version: str + operator_provenance: tuple[tuple[str, str], ...] + source_residual_id: str + problem_frame_digest: str + original_contract_assessment_id: str + + +@dataclass(frozen=True, slots=True) +class CandidateOperatorResult: + operator_result_id: str + operator_policy_version: str + input_digest: str + geometric_search_run_id: str + attempt_id: str + attempt_index: int + candidate_digest: str + candidate_reconstruction_digest: str + candidate_organ: str + operator_name: str + operator_version: str + candidate_attempt: CandidateAttempt + candidate_reconstruction: CandidateReconstruction + reason_codes: tuple[str, ...] + evidence_spans: tuple[SourceSpan, ...] + explanation: str + + +@dataclass(frozen=True, slots=True) +class CandidateOperatorRefusal: + operator_refusal_id: str + operator_policy_version: str + input_digest: str | None + geometric_search_run_id: str | None + residual_id: str | None + operator_name: str + reason_codes: tuple[str, ...] + explanation: str + + +CandidateOperatorOutcome = CandidateOperatorResult | CandidateOperatorRefusal + + +def _canonical_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 _safe_getattr(value: object, name: str) -> object: + try: + return getattr(value, name, None) + except Exception: + return None + + +def _nonempty_text(value: object) -> bool: + return isinstance(value, str) and bool(value.strip()) + + +def _sha256_text(value: object) -> bool: + return ( + isinstance(value, str) + and len(value) == 64 + and value == value.lower() + and all(character in "0123456789abcdef" for character in value) + ) + + +def _text_or_none(value: object) -> str | None: + return value if isinstance(value, str) and value else None + + +def _enum_value(value: object) -> object: + return getattr(value, "value", value) + + +def _valid_span(span: object) -> bool: + return ( + isinstance(span, SourceSpan) + and isinstance(span.text, str) + and type(span.start) is int + and type(span.end) is int + and span.start >= 0 + and span.end >= span.start + and (span.sentence_index is None or type(span.sentence_index) is int) + ) + + +def _valid_spans(value: object) -> bool: + return isinstance(value, tuple) and all(_valid_span(span) for span in 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 _version_pairs_payload( + versions: tuple[tuple[str, str], ...], +) -> list[list[str]]: + return [[name, version] for name, version in versions] + + +def _valid_version_pairs(value: object) -> bool: + if not isinstance(value, tuple): + return False + if not value: + return True + names: list[str] = [] + for entry in value: + if not isinstance(entry, tuple) or len(entry) != 2: + return False + name, version = entry + if not _nonempty_text(name) or not _nonempty_text(version): + return False + names.append(name) + if len(names) != len(set(names)): + return False + return names == sorted(names) + + +def _operator_set_table_payload() -> dict[str, object]: + return { + "operator_set_version": CANDIDATE_OPERATOR_SET_VERSION, + "operators": [ + { + "operator_family": MISSING_ROLE_OPERATOR_POLICY.operator_family, + "operator_name": MISSING_ROLE_OPERATOR_POLICY.operator_name, + "operator_version": MISSING_ROLE_OPERATOR_POLICY.operator_version, + "allowed_residual_kinds": list( + MISSING_ROLE_OPERATOR_POLICY.allowed_residual_kinds + ), + "allowed_residual_codes": list( + MISSING_ROLE_OPERATOR_POLICY.allowed_residual_codes + ), + "allowed_candidate_organs": list( + MISSING_ROLE_OPERATOR_POLICY.allowed_candidate_organs + ), + "max_attempts_per_run": MISSING_ROLE_OPERATOR_POLICY.max_attempts_per_run, + "budget_charge": { + "candidates": MISSING_ROLE_OPERATOR_POLICY.budget_charge.candidates, + "steps": MISSING_ROLE_OPERATOR_POLICY.budget_charge.steps, + }, + "depth": MISSING_ROLE_OPERATOR_POLICY.depth, + "max_parallelism": MISSING_ROLE_OPERATOR_POLICY.max_parallelism, + } + ], + "schema_versions": [], + "policy_versions": [], + } + + +def candidate_operator_set_id() -> str: + """Return the canonical digest of the closed one-row operator table.""" + return _canonical_digest(_operator_set_table_payload()) + + +def _candidate_payload(direction: str) -> tuple[tuple[str, 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 _candidate_payload_dict( + payload: tuple[tuple[str, str], ...], +) -> list[list[str]]: + return [[key, value] for key, value in payload] + + +def _candidate_digest_payload( + *, + problem_frame_digest: str, + candidate_organ: str, + candidate_payload: tuple[tuple[str, str], ...], + evidence_spans: tuple[SourceSpan, ...], +) -> dict[str, object]: + return { + "problem_frame_digest": problem_frame_digest, + "candidate_organ": candidate_organ, + "candidate_payload": _candidate_payload_dict(candidate_payload), + "evidence_spans": [_span_payload(span) for span in evidence_spans], + } + + +def _input_digest_payload( + *, + operator_policy_version: str, + operator_name: str, + operator_version: str, + problem_frame_digest: str, + original_contract_assessment_id: str, + residual_id: str, + residual_kind: str, + residual_code: str, + candidate_organ: str, + search_gate_decision_id: str, + compute_budget_id: str, + geometric_search_run_id: str, + operator_set_id: str, + operator_set_version: str, + attempt_index: int, + schema_versions: tuple[tuple[str, str], ...], + policy_versions: tuple[tuple[str, str], ...], +) -> dict[str, object]: + return { + "operator_policy_version": operator_policy_version, + "operator_name": operator_name, + "operator_version": operator_version, + "problem_frame_digest": problem_frame_digest, + "original_contract_assessment_id": original_contract_assessment_id, + "residual_id": residual_id, + "residual_kind": residual_kind, + "residual_code": residual_code, + "candidate_organ": 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": operator_set_id, + "operator_set_version": operator_set_version, + "attempt_index": attempt_index, + "schema_versions": _version_pairs_payload(schema_versions), + "policy_versions": _version_pairs_payload(policy_versions), + } + + +def _attempt_id_payload( + *, + attempt_index: int, + parent_attempt_id: str | None, + operator_id: str, + operator_version: str, + input_digest: str, + candidate_digest: str, + budget_charge: BudgetCharge, + depth: int, + step_index: int, + replay_status: CandidateReplayStatus, + replay_blockers: tuple[str, ...], + evidence_spans: tuple[SourceSpan, ...], +) -> dict[str, object]: + return { + "attempt_id": "", + "attempt_index": attempt_index, + "parent_attempt_id": parent_attempt_id, + "operator_id": operator_id, + "operator_version": operator_version, + "input_digest": input_digest, + "candidate_digest": candidate_digest, + "budget_charge": { + "candidates": budget_charge.candidates, + "steps": budget_charge.steps, + }, + "depth": depth, + "step_index": step_index, + "replay_status": replay_status.value, + "replay_blockers": list(replay_blockers), + "evidence_spans": [_span_payload(span) for span in evidence_spans], + } + + +def _reconstruction_digest_payload( + *, + geometric_search_run_id: str, + attempt_id: str, + attempt_index: int, + operator_set_id: str, + operator_set_version: str, + operator_name: str, + operator_version: str, + residual_id: str, + residual_kind: str, + residual_code: str, + candidate_digest: str, + evidence_spans: tuple[SourceSpan, ...], + schema_versions: tuple[tuple[str, str], ...], + policy_versions: tuple[tuple[str, str], ...], +) -> dict[str, object]: + return { + "candidate_reconstruction_digest": "", + "geometric_search_run_id": geometric_search_run_id, + "attempt_id": attempt_id, + "attempt_index": attempt_index, + "operator_set_id": operator_set_id, + "operator_set_version": operator_set_version, + "operator_name": operator_name, + "operator_version": operator_version, + "residual_id": residual_id, + "residual_kind": residual_kind, + "residual_code": residual_code, + "candidate_digest": candidate_digest, + "evidence_spans": [_span_payload(span) for span in evidence_spans], + "schema_versions": _version_pairs_payload(schema_versions), + "policy_versions": _version_pairs_payload(policy_versions), + } + + +def _operator_result_id_payload( + *, + operator_policy_version: str, + input_digest: str, + geometric_search_run_id: str, + attempt_id: str, + attempt_index: int, + candidate_digest: str, + candidate_reconstruction_digest: str, + candidate_organ: str, + operator_name: str, + operator_version: str, + reason_codes: tuple[str, ...], + evidence_spans: tuple[SourceSpan, ...], +) -> dict[str, object]: + return { + "operator_result_id": "", + "operator_policy_version": 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": candidate_organ, + "operator_name": operator_name, + "operator_version": operator_version, + "reason_codes": list(reason_codes), + "evidence_spans": [_span_payload(span) for span in evidence_spans], + } + + +def _operator_refusal_id_payload( + *, + operator_policy_version: str, + input_digest: str | None, + geometric_search_run_id: str | None, + residual_id: str | None, + operator_name: str, + reason_codes: tuple[str, ...], +) -> dict[str, object]: + return { + "operator_refusal_id": "", + "operator_policy_version": operator_policy_version, + "input_digest": input_digest, + "geometric_search_run_id": geometric_search_run_id, + "residual_id": residual_id, + "operator_name": operator_name, + "reason_codes": list(reason_codes), + } + + +def _refusal( + *, + input_digest: str | None, + geometric_search_run_id: str | None, + residual_id: str | None, + reason_codes: tuple[str, ...], +) -> CandidateOperatorRefusal: + operator_refusal_id = _canonical_digest( + _operator_refusal_id_payload( + 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=reason_codes, + ) + ) + return CandidateOperatorRefusal( + operator_refusal_id=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=reason_codes, + explanation="Candidate operator refused: " + ", ".join(reason_codes) + ".", + ) + + +def _spans_grounded_in_residual( + cue_spans: tuple[SourceSpan, ...], + residual_spans: tuple[SourceSpan, ...], +) -> bool: + if not cue_spans: + return False + for cue_span in cue_spans: + if cue_span not in residual_spans: + return False + return True + + +def _valid_residual_record(residual: object) -> bool: + required = ( + "residual_id", + "candidate_organ", + "residual_kind", + "residual_code", + "evidence_spans", + ) + return all(_safe_getattr(residual, name) is not None for name in required) + + +def build_missing_role_candidate( + *, + residual: object, + search_gate: SearchGateDecision, + compute_budget: ComputeBudgetDecision, + run: GeometricSearchRun, + problem_frame_digest: str, + original_contract_assessment_id: str, + grounded_unary_delta_cues: tuple[GroundedUnaryDeltaCue, ...], + attempt_index: int = 0, + schema_versions: tuple[tuple[str, str], ...] = (), + policy_versions: tuple[tuple[str, str], ...] = (), + explanation: str = "", +) -> CandidateOperatorOutcome: + """Construct one missing-role candidate or return a typed refusal.""" + + operator_set_id = candidate_operator_set_id() + reasons: list[str] = [] + + residual_id = _safe_getattr(residual, "residual_id") + residual_kind = _enum_value(_safe_getattr(residual, "residual_kind")) + residual_code = _safe_getattr(residual, "residual_code") + candidate_organ = _safe_getattr(residual, "candidate_organ") + residual_spans = _safe_getattr(residual, "evidence_spans") + + gate_id = _safe_getattr(search_gate, "decision_id") + gate_status = _safe_getattr(search_gate, "status") + gate_reason = _safe_getattr(search_gate, "reason_code") + gate_residual_ids = _safe_getattr(search_gate, "residual_ids") + gate_candidate_organ = _safe_getattr(search_gate, "candidate_organ") + + budget_id = _safe_getattr(compute_budget, "budget_id") + budget_status = _safe_getattr(compute_budget, "status") + budget_reason = _safe_getattr(compute_budget, "reason_code") + budget_gate_id = _safe_getattr(compute_budget, "gate_decision_id") + max_candidates = _safe_getattr(compute_budget, "max_candidates") + max_depth = _safe_getattr(compute_budget, "max_depth") + max_steps = _safe_getattr(compute_budget, "max_steps") + max_parallelism = _safe_getattr(compute_budget, "max_parallelism") + + run_id = _safe_getattr(run, "run_id") + run_gate_id = _safe_getattr(run, "gate_decision_id") + run_budget_id = _safe_getattr(run, "budget_id") + run_operator_set_id = _safe_getattr(run, "operator_set_id") + run_operator_set_version = _safe_getattr(run, "operator_set_version") + run_problem_frame_digest = _safe_getattr(run, "problem_frame_digest") + run_assessment_id = _safe_getattr(run, "contract_assessment_id") + run_residual_ids = _safe_getattr(run, "residual_ids") + + if not _valid_residual_record(residual): + reasons.append(CandidateOperatorRefusalReason.INVALID_OPERATOR_INPUT.value) + if not isinstance(search_gate, SearchGateDecision): + reasons.append(CandidateOperatorRefusalReason.INVALID_OPERATOR_INPUT.value) + if not isinstance(compute_budget, ComputeBudgetDecision): + reasons.append(CandidateOperatorRefusalReason.INVALID_OPERATOR_INPUT.value) + if not isinstance(run, GeometricSearchRun): + reasons.append(CandidateOperatorRefusalReason.INVALID_OPERATOR_INPUT.value) + + if not _nonempty_text(residual_id): + reasons.append(CandidateOperatorRefusalReason.INVALID_OPERATOR_INPUT.value) + if residual_kind != MISSING_ROLE_RESIDUAL_KIND: + reasons.append(CandidateOperatorRefusalReason.UNSUPPORTED_RESIDUAL_KIND.value) + if residual_code != MISSING_ROLE_RESIDUAL_CODE: + reasons.append(CandidateOperatorRefusalReason.UNSUPPORTED_RESIDUAL_CODE.value) + if candidate_organ != MISSING_ROLE_CANDIDATE_ORGAN: + reasons.append(CandidateOperatorRefusalReason.UNSUPPORTED_CANDIDATE_ORGAN.value) + if not _valid_spans(residual_spans): + reasons.append(CandidateOperatorRefusalReason.MALFORMED_EVIDENCE_SPAN.value) + + if gate_status is not SearchGateStatus.ELIGIBLE: + reasons.append(CandidateOperatorRefusalReason.INELIGIBLE_SEARCH_GATE.value) + if gate_reason != "eligible_missing_role": + reasons.append(CandidateOperatorRefusalReason.INELIGIBLE_SEARCH_GATE.value) + if gate_candidate_organ != MISSING_ROLE_CANDIDATE_ORGAN: + reasons.append(CandidateOperatorRefusalReason.UNSUPPORTED_CANDIDATE_ORGAN.value) + + if budget_status is not ComputeBudgetStatus.BUDGET_ALLOWED: + reasons.append(CandidateOperatorRefusalReason.NON_ALLOWED_BUDGET.value) + if budget_reason != "budget_allowed_missing_role": + reasons.append(CandidateOperatorRefusalReason.NON_ALLOWED_BUDGET.value) + + if not _nonempty_text(budget_gate_id) or not _nonempty_text(gate_id): + reasons.append(CandidateOperatorRefusalReason.GATE_BUDGET_MISMATCH.value) + elif budget_gate_id != gate_id: + reasons.append(CandidateOperatorRefusalReason.GATE_BUDGET_MISMATCH.value) + + if not _nonempty_text(run_gate_id) or not _nonempty_text(gate_id): + reasons.append(CandidateOperatorRefusalReason.RUN_GATE_MISMATCH.value) + elif run_gate_id != gate_id: + reasons.append(CandidateOperatorRefusalReason.RUN_GATE_MISMATCH.value) + + if not _nonempty_text(run_budget_id) or not _nonempty_text(budget_id): + reasons.append(CandidateOperatorRefusalReason.RUN_BUDGET_MISMATCH.value) + elif run_budget_id != budget_id: + reasons.append(CandidateOperatorRefusalReason.RUN_BUDGET_MISMATCH.value) + + if run_operator_set_id != operator_set_id: + reasons.append(CandidateOperatorRefusalReason.OPERATOR_SET_MISMATCH.value) + if run_operator_set_version != CANDIDATE_OPERATOR_SET_VERSION: + reasons.append(CandidateOperatorRefusalReason.OPERATOR_SET_MISMATCH.value) + + if type(attempt_index) is not int or attempt_index < 0: + reasons.append(CandidateOperatorRefusalReason.INVALID_OPERATOR_INPUT.value) + if type(max_candidates) is not int: + reasons.append(CandidateOperatorRefusalReason.INVALID_OPERATOR_INPUT.value) + elif attempt_index >= max_candidates: + reasons.append( + CandidateOperatorRefusalReason.ATTEMPT_INDEX_EXCEEDS_BUDGET.value + ) + if attempt_index >= MISSING_ROLE_OPERATOR_POLICY.max_attempts_per_run: + reasons.append( + CandidateOperatorRefusalReason.ATTEMPT_INDEX_EXCEEDS_OPERATOR_POLICY.value + ) + + if type(max_depth) is not int or type(max_steps) is not int: + reasons.append(CandidateOperatorRefusalReason.INVALID_OPERATOR_INPUT.value) + else: + charge = MISSING_ROLE_OPERATOR_POLICY.budget_charge + if charge.candidates > max_candidates: + reasons.append( + CandidateOperatorRefusalReason.BUDGET_CHARGE_EXCEEDS_BUDGET.value + ) + if charge.steps > max_steps: + reasons.append( + CandidateOperatorRefusalReason.BUDGET_CHARGE_EXCEEDS_BUDGET.value + ) + if MISSING_ROLE_OPERATOR_POLICY.depth > max_depth: + reasons.append( + CandidateOperatorRefusalReason.BUDGET_CHARGE_EXCEEDS_BUDGET.value + ) + + if type(max_parallelism) is not int or max_parallelism != 1: + reasons.append(CandidateOperatorRefusalReason.NON_SERIAL_BUDGET.value) + + if not _valid_version_pairs(schema_versions): + reasons.append(CandidateOperatorRefusalReason.UNSUPPORTED_SCHEMA_VERSION.value) + if not _valid_version_pairs(policy_versions): + reasons.append(CandidateOperatorRefusalReason.UNSUPPORTED_SCHEMA_VERSION.value) + + if not _sha256_text(problem_frame_digest): + reasons.append(CandidateOperatorRefusalReason.INVALID_OPERATOR_INPUT.value) + if not _nonempty_text(original_contract_assessment_id): + reasons.append(CandidateOperatorRefusalReason.INVALID_OPERATOR_INPUT.value) + if not _sha256_text(run_id): + reasons.append(CandidateOperatorRefusalReason.INVALID_OPERATOR_INPUT.value) + + if ( + isinstance(residual_id, str) + and isinstance(gate_residual_ids, tuple) + and residual_id not in gate_residual_ids + ): + reasons.append(CandidateOperatorRefusalReason.INVALID_OPERATOR_INPUT.value) + if ( + isinstance(residual_id, str) + and isinstance(run_residual_ids, tuple) + and residual_id not in run_residual_ids + ): + reasons.append(CandidateOperatorRefusalReason.INVALID_OPERATOR_INPUT.value) + if ( + isinstance(problem_frame_digest, str) + and isinstance(run_problem_frame_digest, str) + and problem_frame_digest != run_problem_frame_digest + ): + reasons.append(CandidateOperatorRefusalReason.INVALID_OPERATOR_INPUT.value) + if ( + isinstance(original_contract_assessment_id, str) + and isinstance(run_assessment_id, str) + and original_contract_assessment_id != run_assessment_id + ): + reasons.append(CandidateOperatorRefusalReason.INVALID_OPERATOR_INPUT.value) + + if len(grounded_unary_delta_cues) == 0: + reasons.append(CandidateOperatorRefusalReason.MISSING_TYPED_CUE.value) + elif len(grounded_unary_delta_cues) > 1: + reasons.append(CandidateOperatorRefusalReason.AMBIGUOUS_TYPED_CUE.value) + + cue: GroundedUnaryDeltaCue | None = None + if len(grounded_unary_delta_cues) == 1: + cue = grounded_unary_delta_cues[0] + if not _nonempty_text(cue.direction): + reasons.append(CandidateOperatorRefusalReason.MISSING_TYPED_CUE.value) + elif cue.direction not in _SUPPORTED_DIRECTIONS: + reasons.append(CandidateOperatorRefusalReason.UNSUPPORTED_TYPED_CUE.value) + if not _valid_spans(cue.evidence_spans): + reasons.append(CandidateOperatorRefusalReason.MALFORMED_EVIDENCE_SPAN.value) + elif isinstance(residual_spans, tuple) and not _spans_grounded_in_residual( + cue.evidence_spans, residual_spans + ): + reasons.append(CandidateOperatorRefusalReason.MALFORMED_EVIDENCE_SPAN.value) + + input_digest: str | None = None + if ( + not reasons + and isinstance(residual_id, str) + and isinstance(run_id, str) + and isinstance(gate_id, str) + and isinstance(budget_id, str) + and type(attempt_index) is int + ): + input_digest = _canonical_digest( + _input_digest_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=gate_id, + compute_budget_id=budget_id, + geometric_search_run_id=run_id, + operator_set_id=operator_set_id, + operator_set_version=CANDIDATE_OPERATOR_SET_VERSION, + attempt_index=attempt_index, + schema_versions=schema_versions, + policy_versions=policy_versions, + ) + ) + + if reasons: + return _refusal( + input_digest=input_digest, + geometric_search_run_id=_text_or_none(run_id), + residual_id=_text_or_none(residual_id), + reason_codes=tuple(dict.fromkeys(reasons)), + ) + + assert cue is not None + assert input_digest is not None + assert isinstance(residual_id, str) + assert isinstance(run_id, str) + + evidence_spans = cue.evidence_spans + payload = _candidate_payload(cue.direction) + candidate_digest = _canonical_digest( + _candidate_digest_payload( + problem_frame_digest=problem_frame_digest, + candidate_organ=MISSING_ROLE_CANDIDATE_ORGAN, + candidate_payload=payload, + evidence_spans=evidence_spans, + ) + ) + + attempt_id = _canonical_digest( + _attempt_id_payload( + 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=_FIXED_BUDGET_CHARGE, + depth=_OPERATOR_DEPTH, + step_index=attempt_index, + replay_status=CandidateReplayStatus.REPLAY_PENDING, + replay_blockers=(), + evidence_spans=evidence_spans, + ) + ) + + candidate_reconstruction_digest = _canonical_digest( + _reconstruction_digest_payload( + geometric_search_run_id=run_id, + attempt_id=attempt_id, + attempt_index=attempt_index, + operator_set_id=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=evidence_spans, + schema_versions=schema_versions, + policy_versions=policy_versions, + ) + ) + + operator_provenance = ( + ("operator_family", _OPERATOR_FAMILY), + ("operator_name", MISSING_ROLE_CANDIDATE_OPERATOR_NAME), + ("operator_version", MISSING_ROLE_CANDIDATE_OPERATOR_VERSION), + ("operator_set_id", operator_set_id), + ("operator_set_version", CANDIDATE_OPERATOR_SET_VERSION), + ) + + candidate_reconstruction = CandidateReconstruction( + candidate_digest=candidate_digest, + candidate_reconstruction_digest=candidate_reconstruction_digest, + candidate_organ=MISSING_ROLE_CANDIDATE_ORGAN, + candidate_payload=payload, + evidence_spans=evidence_spans, + operator_name=MISSING_ROLE_CANDIDATE_OPERATOR_NAME, + operator_version=MISSING_ROLE_CANDIDATE_OPERATOR_VERSION, + operator_provenance=operator_provenance, + source_residual_id=residual_id, + problem_frame_digest=problem_frame_digest, + original_contract_assessment_id=original_contract_assessment_id, + ) + + candidate_attempt = CandidateAttempt( + attempt_id=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=_FIXED_BUDGET_CHARGE, + depth=_OPERATOR_DEPTH, + step_index=attempt_index, + replay_status=CandidateReplayStatus.REPLAY_PENDING, + replay_blockers=(), + evidence_spans=evidence_spans, + explanation="", + ) + + reason_codes: tuple[str, ...] = () + operator_result_id = _canonical_digest( + _operator_result_id_payload( + operator_policy_version=CANDIDATE_OPERATOR_POLICY_VERSION, + input_digest=input_digest, + geometric_search_run_id=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=reason_codes, + evidence_spans=evidence_spans, + ) + ) + + safe_explanation = explanation or ( + "Missing-role candidate constructed for residual " + + residual_id + + " with direction " + + cue.direction + + "." + ) + return CandidateOperatorResult( + operator_result_id=operator_result_id, + operator_policy_version=CANDIDATE_OPERATOR_POLICY_VERSION, + input_digest=input_digest, + geometric_search_run_id=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, + candidate_attempt=candidate_attempt, + candidate_reconstruction=candidate_reconstruction, + reason_codes=reason_codes, + evidence_spans=evidence_spans, + explanation=safe_explanation, + ) + + +__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", +] \ No newline at end of file diff --git a/tests/test_candidate_operator.py b/tests/test_candidate_operator.py new file mode 100644 index 00000000..db8a186e --- /dev/null +++ b/tests/test_candidate_operator.py @@ -0,0 +1,1191 @@ +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" \ No newline at end of file