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

Merge pull request #879 from AssetOverflow/feat/replay-bound-attempt-input

Browse source
This commit is contained in:
Shay 2026-06-22 23:49:33 -07:00 committed by GitHub
commit 5b30ec877f
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 754 additions and 39 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

386
generate/replay_adapter.py
View file

@ -7,8 +7,10 @@ import json
from dataclasses import dataclass
from enum import Enum, unique
from generate.candidate_operator import CandidateOperatorResult
from generate.geometric_search_run import CandidateAttempt, GeometricSearchRun
from generate.kernel_facts import SourceSpan
from generate.run_attempt_binding import CandidateAttemptRunBinding
CONTRACT_PROOF_REPLAY_POLICY_VERSION = "contract_proof_replay.v1"
REPLAY_ADAPTER_POLICY_VERSION = CONTRACT_PROOF_REPLAY_POLICY_VERSION
@ -299,6 +301,63 @@ def _refusal(
)
def _replay_input(
*,
replay_policy_version: str,
run: GeometricSearchRun,
attempt: CandidateAttempt,
candidate_digest: str,
candidate_reconstruction_digest: str,
candidate_organ: str,
contract_replay_target: str,
proof_obligation_refs: tuple[str, ...],
schema_versions: tuple[tuple[str, str], ...],
) -> ReplayAdapterInput:
input_digest = _canonical_digest(
_input_payload(
replay_policy_version=replay_policy_version,
run_id=run.run_id,
run_policy_version=run.run_policy_version,
attempt_id=attempt.attempt_id,
attempt_index=attempt.attempt_index,
candidate_digest=candidate_digest,
candidate_reconstruction_digest=candidate_reconstruction_digest,
problem_frame_digest=run.problem_frame_digest,
original_contract_assessment_id=run.contract_assessment_id,
candidate_organ=candidate_organ,
residual_ids=run.residual_ids,
gate_decision_id=run.gate_decision_id,
budget_id=run.budget_id,
operator_set_id=run.operator_set_id,
operator_set_version=run.operator_set_version,
contract_replay_target=contract_replay_target,
proof_obligation_refs=proof_obligation_refs,
schema_versions=schema_versions,
)
)
return ReplayAdapterInput(
input_digest=input_digest,
replay_policy_version=replay_policy_version,
run_id=run.run_id,
run_policy_version=run.run_policy_version,
attempt_id=attempt.attempt_id,
attempt_index=attempt.attempt_index,
candidate_digest=candidate_digest,
candidate_reconstruction_digest=candidate_reconstruction_digest,
problem_frame_digest=run.problem_frame_digest,
original_contract_assessment_id=run.contract_assessment_id,
candidate_organ=candidate_organ,
residual_ids=run.residual_ids,
gate_decision_id=run.gate_decision_id,
budget_id=run.budget_id,
operator_set_id=run.operator_set_id,
operator_set_version=run.operator_set_version,
contract_replay_target=contract_replay_target,
proof_obligation_refs=proof_obligation_refs,
schema_versions=schema_versions,
)
def build_replay_adapter_input(
*,
run: GeometricSearchRun,
@ -436,45 +495,296 @@ def build_replay_adapter_input(
assert isinstance(attempt_index, int)
assert contract_replay_target is not None
input_digest = _canonical_digest(
_input_payload(
replay_policy_version=replay_policy_version,
run_id=run.run_id,
run_policy_version=run.run_policy_version,
attempt_id=attempt.attempt_id,
attempt_index=attempt_index,
candidate_digest=candidate_digest,
candidate_reconstruction_digest=candidate_reconstruction_digest,
problem_frame_digest=problem_frame_digest,
original_contract_assessment_id=original_contract_assessment_id,
candidate_organ=candidate_organ,
residual_ids=run.residual_ids,
gate_decision_id=run.gate_decision_id,
budget_id=run.budget_id,
operator_set_id=run.operator_set_id,
operator_set_version=run.operator_set_version,
contract_replay_target=contract_replay_target,
proof_obligation_refs=proof_obligation_refs,
schema_versions=schema_versions,
)
)
return ReplayAdapterInput(
input_digest=input_digest,
return _replay_input(
replay_policy_version=replay_policy_version,
run_id=run.run_id,
run_policy_version=run.run_policy_version,
attempt_id=attempt.attempt_id,
attempt_index=attempt_index,
run=run,
attempt=attempt,
candidate_digest=candidate_digest,
candidate_reconstruction_digest=candidate_reconstruction_digest,
problem_frame_digest=problem_frame_digest,
original_contract_assessment_id=original_contract_assessment_id,
candidate_organ=candidate_organ,
residual_ids=run.residual_ids,
gate_decision_id=run.gate_decision_id,
budget_id=run.budget_id,
operator_set_id=run.operator_set_id,
operator_set_version=run.operator_set_version,
contract_replay_target=contract_replay_target,
proof_obligation_refs=proof_obligation_refs,
schema_versions=schema_versions,
)
def _bound_refusal(
*,
run: GeometricSearchRun | None,
binding: CandidateAttemptRunBinding | None,
result: CandidateOperatorResult | None,
replay_disposition: ReplayRefusalReason,
reason_code: str,
replay_policy_version: str = CONTRACT_PROOF_REPLAY_POLICY_VERSION,
) -> ReplayAdapterRefusal:
return _refusal(
replay_policy_version=replay_policy_version,
input_digest=None,
run_id=run.run_id if isinstance(run, GeometricSearchRun) else None,
attempt_id=binding.candidate_attempt_id
if isinstance(binding, CandidateAttemptRunBinding)
else _text_or_none(_safe_getattr(result, "attempt_id")),
candidate_digest=binding.candidate_digest
if isinstance(binding, CandidateAttemptRunBinding)
else _text_or_none(_safe_getattr(result, "candidate_digest")),
replay_disposition=replay_disposition,
reason_codes=(reason_code,),
)
def build_replay_adapter_input_from_binding(
*,
run: GeometricSearchRun,
binding: CandidateAttemptRunBinding,
candidate_operator_result: CandidateOperatorResult,
proof_obligation_refs: tuple[str, ...] = (),
schema_versions: tuple[tuple[str, str], ...] = (),
replay_policy_version: str = CONTRACT_PROOF_REPLAY_POLICY_VERSION,
) -> ReplayAdapterInput | ReplayAdapterRefusal:
"""Build replay input from external CandidateAttemptRunBinding evidence."""
if not isinstance(run, GeometricSearchRun):
return _bound_refusal(
run=None,
binding=None,
result=None,
replay_disposition=ReplayRefusalReason.INVALID_REPLAY_INPUT,
reason_code="invalid_run_type",
replay_policy_version=replay_policy_version,
)
if not isinstance(binding, CandidateAttemptRunBinding):
return _bound_refusal(
run=run,
binding=None,
result=None,
replay_disposition=ReplayRefusalReason.INVALID_REPLAY_INPUT,
reason_code="invalid_binding_type",
replay_policy_version=replay_policy_version,
)
if not isinstance(candidate_operator_result, CandidateOperatorResult):
return _bound_refusal(
run=run,
binding=binding,
result=None,
replay_disposition=ReplayRefusalReason.INVALID_REPLAY_INPUT,
reason_code="invalid_operator_result_type",
replay_policy_version=replay_policy_version,
)
if replay_policy_version != CONTRACT_PROOF_REPLAY_POLICY_VERSION:
return _bound_refusal(
run=run,
binding=binding,
result=candidate_operator_result,
replay_disposition=ReplayRefusalReason.UNSUPPORTED_REPLAY_POLICY,
reason_code="unsupported_replay_policy_version",
replay_policy_version=replay_policy_version,
)
if proof_obligation_refs and not _valid_string_tuple(proof_obligation_refs):
return _bound_refusal(
run=run,
binding=binding,
result=candidate_operator_result,
replay_disposition=ReplayRefusalReason.INVALID_REPLAY_INPUT,
reason_code="invalid_proof_obligation_refs",
)
if not _valid_schema_versions(schema_versions):
return _bound_refusal(
run=run,
binding=binding,
result=candidate_operator_result,
replay_disposition=ReplayRefusalReason.UNSUPPORTED_SCHEMA_VERSION,
reason_code="unsupported_schema_version",
)
attempt = candidate_operator_result.candidate_attempt
reconstruction = candidate_operator_result.candidate_reconstruction
if binding.original_run_id != run.run_id:
return _bound_refusal(
run=run,
binding=binding,
result=candidate_operator_result,
replay_disposition=ReplayRefusalReason.CANDIDATE_IDENTITY_MISMATCH,
reason_code="binding_run_mismatch",
)
if binding.operator_result_id != candidate_operator_result.operator_result_id:
return _bound_refusal(
run=run,
binding=binding,
result=candidate_operator_result,
replay_disposition=ReplayRefusalReason.CANDIDATE_IDENTITY_MISMATCH,
reason_code="binding_result_mismatch",
)
if binding.run_attempt_membership != "structurally_bound":
return _bound_refusal(
run=run,
binding=binding,
result=candidate_operator_result,
replay_disposition=ReplayRefusalReason.CANDIDATE_IDENTITY_MISMATCH,
reason_code="binding_not_structurally_bound",
)
if binding.reason_codes != ():
return _bound_refusal(
run=run,
binding=binding,
result=candidate_operator_result,
replay_disposition=ReplayRefusalReason.CANDIDATE_IDENTITY_MISMATCH,
reason_code="binding_not_successful",
)
if not _nonempty_text(binding.candidate_attempt_ref):
return _bound_refusal(
run=run,
binding=binding,
result=candidate_operator_result,
replay_disposition=ReplayRefusalReason.CANDIDATE_IDENTITY_MISMATCH,
reason_code="binding_missing_attempt_ref",
)
if candidate_operator_result.geometric_search_run_id != run.run_id:
return _bound_refusal(
run=run,
binding=binding,
result=candidate_operator_result,
replay_disposition=ReplayRefusalReason.CANDIDATE_IDENTITY_MISMATCH,
reason_code="result_run_mismatch",
)
if binding.candidate_attempt_id != candidate_operator_result.attempt_id:
return _bound_refusal(
run=run,
binding=binding,
result=candidate_operator_result,
replay_disposition=ReplayRefusalReason.CANDIDATE_IDENTITY_MISMATCH,
reason_code="binding_result_mismatch",
)
if binding.candidate_attempt_id != attempt.attempt_id:
return _bound_refusal(
run=run,
binding=binding,
result=candidate_operator_result,
replay_disposition=ReplayRefusalReason.CANDIDATE_IDENTITY_MISMATCH,
reason_code="binding_attempt_mismatch",
)
if binding.attempt_index != candidate_operator_result.attempt_index:
return _bound_refusal(
run=run,
binding=binding,
result=candidate_operator_result,
replay_disposition=ReplayRefusalReason.CANDIDATE_IDENTITY_MISMATCH,
reason_code="binding_result_mismatch",
)
if binding.attempt_index != attempt.attempt_index:
return _bound_refusal(
run=run,
binding=binding,
result=candidate_operator_result,
replay_disposition=ReplayRefusalReason.CANDIDATE_IDENTITY_MISMATCH,
reason_code="binding_attempt_mismatch",
)
if binding.candidate_digest != candidate_operator_result.candidate_digest:
return _bound_refusal(
run=run,
binding=binding,
result=candidate_operator_result,
replay_disposition=ReplayRefusalReason.CANDIDATE_IDENTITY_MISMATCH,
reason_code="binding_result_mismatch",
)
if binding.candidate_digest != attempt.candidate_digest:
return _bound_refusal(
run=run,
binding=binding,
result=candidate_operator_result,
replay_disposition=ReplayRefusalReason.CANDIDATE_IDENTITY_MISMATCH,
reason_code="binding_attempt_mismatch",
)
if binding.candidate_reconstruction_digest != candidate_operator_result.candidate_reconstruction_digest:
return _bound_refusal(
run=run,
binding=binding,
result=candidate_operator_result,
replay_disposition=ReplayRefusalReason.CANDIDATE_IDENTITY_MISMATCH,
reason_code="binding_result_mismatch",
)
if binding.candidate_reconstruction_digest != reconstruction.candidate_reconstruction_digest:
return _bound_refusal(
run=run,
binding=binding,
result=candidate_operator_result,
replay_disposition=ReplayRefusalReason.CANDIDATE_IDENTITY_MISMATCH,
reason_code="binding_reconstruction_mismatch",
)
if attempt.input_digest != candidate_operator_result.input_digest:
return _bound_refusal(
run=run,
binding=binding,
result=candidate_operator_result,
replay_disposition=ReplayRefusalReason.CANDIDATE_IDENTITY_MISMATCH,
reason_code="attempt_input_digest_mismatch",
)
if binding.evidence_spans != candidate_operator_result.evidence_spans:
return _bound_refusal(
run=run,
binding=binding,
result=candidate_operator_result,
replay_disposition=ReplayRefusalReason.CANDIDATE_IDENTITY_MISMATCH,
reason_code="binding_evidence_mismatch",
)
if binding.evidence_spans != attempt.evidence_spans:
return _bound_refusal(
run=run,
binding=binding,
result=candidate_operator_result,
replay_disposition=ReplayRefusalReason.CANDIDATE_IDENTITY_MISMATCH,
reason_code="attempt_evidence_mismatch",
)
if binding.evidence_spans != reconstruction.evidence_spans:
return _bound_refusal(
run=run,
binding=binding,
result=candidate_operator_result,
replay_disposition=ReplayRefusalReason.CANDIDATE_IDENTITY_MISMATCH,
reason_code="reconstruction_evidence_mismatch",
)
if reconstruction.problem_frame_digest != run.problem_frame_digest:
return _bound_refusal(
run=run,
binding=binding,
result=candidate_operator_result,
replay_disposition=ReplayRefusalReason.CANDIDATE_IDENTITY_MISMATCH,
reason_code="reconstruction_problem_frame_mismatch",
)
if reconstruction.original_contract_assessment_id != run.contract_assessment_id:
return _bound_refusal(
run=run,
binding=binding,
result=candidate_operator_result,
replay_disposition=ReplayRefusalReason.CANDIDATE_IDENTITY_MISMATCH,
reason_code="reconstruction_assessment_mismatch",
)
if reconstruction.source_residual_id not in run.residual_ids:
return _bound_refusal(
run=run,
binding=binding,
result=candidate_operator_result,
replay_disposition=ReplayRefusalReason.CANDIDATE_IDENTITY_MISMATCH,
reason_code="reconstruction_residual_mismatch",
)
contract_replay_target = _contract_replay_target(candidate_operator_result.candidate_organ)
if contract_replay_target is None:
return _bound_refusal(
run=run,
binding=binding,
result=candidate_operator_result,
replay_disposition=ReplayRefusalReason.INVALID_REPLAY_INPUT,
reason_code="unsupported_candidate_organ",
)
return _replay_input(
replay_policy_version=replay_policy_version,
run=run,
attempt=attempt,
candidate_digest=binding.candidate_digest,
candidate_reconstruction_digest=binding.candidate_reconstruction_digest,
candidate_organ=candidate_operator_result.candidate_organ,
contract_replay_target=contract_replay_target,
proof_obligation_refs=proof_obligation_refs,
schema_versions=schema_versions,
@ -677,9 +987,7 @@ def _result(
}
replay_result_id = _canonical_digest(result_payload)
safe_explanation = explanation or (
"Replay adapter classified candidate as "
+ replay_disposition.value
+ "."
"Replay adapter classified candidate as " + replay_disposition.value + "."
)
return ReplayAdapterResult(
replay_result_id=replay_result_id,
@ -710,4 +1018,4 @@ __all__ = [
"ReplayAdapterRefusal",
"build_replay_adapter_input",
"classify_replay_result",
]
]

407
tests/test_replay_adapter_bound_attempt.py Normal file
View file

@ -0,0 +1,407 @@
from __future__ import annotations
import hashlib
import json
from dataclasses import replace
from types import SimpleNamespace
from generate.candidate_operator import (
GroundedUnaryDeltaCue,
build_missing_role_candidate,
candidate_operator_set_id,
)
from generate.compute_budget import ComputeBudgetDecision, ComputeBudgetStatus
from generate.contract_residuals import ContractResidual, ResidualKind, ResidualSourceAxis
from generate.geometric_search_run import GeometricSearchRun, initialize_geometric_search_run
from generate.kernel_facts import SourceSpan
from generate.replay_adapter import (
CONTRACT_PROOF_REPLAY_POLICY_VERSION,
ReplayAdapterInput,
ReplayAdapterRefusal,
ReplayRefusalReason,
build_replay_adapter_input_from_binding,
)
from generate.run_attempt_binding import (
CandidateAttemptRunBinding,
bind_candidate_attempt_to_run,
)
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 _residual() -> ContractResidual:
spans = (SourceSpan(text="gained", start=10, end=16, sentence_index=0),)
residual_id = _digest(
{
"candidate_organ": "unary_delta_transition",
"residual_kind": ResidualKind.MISSING_ROLE.value,
"residual_code": "direction_unbound",
"evidence_spans": [_span_payload(span) for span in spans],
}
)
return ContractResidual(
residual_id=residual_id,
candidate_organ="unary_delta_transition",
family_id="state_change.unary_delta",
residual_kind=ResidualKind.MISSING_ROLE,
residual_code="direction_unbound",
source_axis=ResidualSourceAxis.ROLE,
evidence_spans=spans,
explanation="residual prose",
)
def _gate(residual: ContractResidual) -> SearchGateDecision:
payload = {
"policy_version": "search_gate.v1",
"input_digest": "a" * 64,
"residual_ids": [residual.residual_id],
"candidate_organ": residual.candidate_organ,
"status": SearchGateStatus.ELIGIBLE.value,
"reason_code": "eligible_missing_role",
"evidence_spans": [_span_payload(span) for span in residual.evidence_spans],
}
return SearchGateDecision(
decision_id=_digest(payload),
policy_version="search_gate.v1",
input_digest="a" * 64,
residual_ids=(residual.residual_id,),
candidate_organ=residual.candidate_organ,
status=SearchGateStatus.ELIGIBLE,
reason_code="eligible_missing_role",
evidence_spans=residual.evidence_spans,
explanation="gate prose",
)
def _budget(gate: SearchGateDecision) -> ComputeBudgetDecision:
payload = {
"policy_version": "compute_budget.v1",
"gate_decision_id": gate.decision_id,
"gate_policy_version": gate.policy_version,
"gate_input_digest": gate.input_digest,
"status": ComputeBudgetStatus.BUDGET_ALLOWED.value,
"reason_code": "budget_allowed_missing_role",
"max_candidates": 5,
"max_depth": 2,
"max_steps": 10,
"max_parallelism": 1,
"evidence_spans": [_span_payload(span) for span in gate.evidence_spans],
}
return ComputeBudgetDecision(
budget_id=_digest(payload),
policy_version="compute_budget.v1",
gate_decision_id=gate.decision_id,
gate_policy_version=gate.policy_version,
gate_input_digest=gate.input_digest,
status=ComputeBudgetStatus.BUDGET_ALLOWED,
reason_code="budget_allowed_missing_role",
max_candidates=5,
max_depth=2,
max_steps=10,
max_wallclock_ms=None,
max_parallelism=1,
evidence_spans=gate.evidence_spans,
explanation="budget prose",
)
def _chain() -> tuple[GeometricSearchRun, object, CandidateAttemptRunBinding]:
residual = _residual()
gate = _gate(residual)
budget = _budget(gate)
run = initialize_geometric_search_run(
problem_frame_digest="f" * 64,
contract_assessment_id="assessment-a",
residual_ids=(residual.residual_id,),
gate_decision=gate,
compute_budget=budget,
operator_set_id=candidate_operator_set_id(),
operator_set_version="candidate_operators.v1",
)
assert isinstance(run, GeometricSearchRun)
result = build_missing_role_candidate(
residual=residual,
search_gate=gate,
compute_budget=budget,
run=run,
problem_frame_digest=run.problem_frame_digest,
original_contract_assessment_id=run.contract_assessment_id,
grounded_unary_delta_cues=(
GroundedUnaryDeltaCue(
direction="increase",
evidence_spans=residual.evidence_spans,
),
),
)
assert hasattr(result, "candidate_attempt")
binding = bind_candidate_attempt_to_run(
original_run=run,
candidate_operator_result=result,
)
assert isinstance(binding, CandidateAttemptRunBinding)
return run, result, binding
def _bound_input(
run: GeometricSearchRun,
result: object,
binding: CandidateAttemptRunBinding,
**kwargs: object,
) -> ReplayAdapterInput | ReplayAdapterRefusal:
return build_replay_adapter_input_from_binding(
run=run,
binding=binding,
candidate_operator_result=result, # type: ignore[arg-type]
**kwargs,
)
def _refused(
expected_reason: ReplayRefusalReason,
expected_code: str,
*,
run: GeometricSearchRun,
result: object,
binding: object,
**kwargs: object,
) -> None:
outcome = build_replay_adapter_input_from_binding(
run=run,
binding=binding, # type: ignore[arg-type]
candidate_operator_result=result, # type: ignore[arg-type]
**kwargs,
)
assert isinstance(outcome, ReplayAdapterRefusal)
assert outcome.replay_disposition is expected_reason
assert outcome.reason_codes == (expected_code,)
def test_valid_bound_attempt_builds_replay_input_without_mutating_run() -> None:
run, result, binding = _chain()
before = run.candidate_attempts
outcome = _bound_input(run, result, binding)
assert isinstance(outcome, ReplayAdapterInput)
assert outcome.run_id == run.run_id
assert outcome.run_policy_version == run.run_policy_version
assert outcome.attempt_id == binding.candidate_attempt_id
assert outcome.attempt_index == binding.attempt_index
assert outcome.candidate_digest == binding.candidate_digest
assert outcome.candidate_reconstruction_digest == binding.candidate_reconstruction_digest
assert outcome.problem_frame_digest == run.problem_frame_digest
assert outcome.original_contract_assessment_id == run.contract_assessment_id
assert outcome.candidate_organ == result.candidate_organ
assert outcome.residual_ids == run.residual_ids
assert outcome.gate_decision_id == run.gate_decision_id
assert outcome.budget_id == run.budget_id
assert outcome.operator_set_id == run.operator_set_id
assert outcome.operator_set_version == run.operator_set_version
assert outcome.contract_replay_target == "problem_frame_contracts.unary_delta"
assert run.candidate_attempts == before == ()
def test_bound_input_digest_is_deterministic_and_binding_explanation_free() -> None:
run, result, binding = _chain()
first = _bound_input(run, result, binding)
changed_binding = replace(binding, explanation="different prose")
second = _bound_input(run, result, changed_binding)
assert isinstance(first, ReplayAdapterInput)
assert isinstance(second, ReplayAdapterInput)
assert first.input_digest == second.input_digest
def test_invalid_bound_input_types_refuse() -> None:
run, result, binding = _chain()
outcome = build_replay_adapter_input_from_binding(
run=SimpleNamespace(), # type: ignore[arg-type]
binding=binding,
candidate_operator_result=result,
)
assert isinstance(outcome, ReplayAdapterRefusal)
assert outcome.reason_codes == ("invalid_run_type",)
_refused(
ReplayRefusalReason.INVALID_REPLAY_INPUT,
"invalid_binding_type",
run=run,
result=result,
binding=SimpleNamespace(),
)
_refused(
ReplayRefusalReason.INVALID_REPLAY_INPUT,
"invalid_operator_result_type",
run=run,
result=SimpleNamespace(),
binding=binding,
)
def test_binding_identity_mismatches_refuse() -> None:
run, result, binding = _chain()
_refused(
ReplayRefusalReason.CANDIDATE_IDENTITY_MISMATCH,
"binding_run_mismatch",
run=run,
result=result,
binding=replace(binding, original_run_id="wrong"),
)
_refused(
ReplayRefusalReason.CANDIDATE_IDENTITY_MISMATCH,
"binding_result_mismatch",
run=run,
result=result,
binding=replace(binding, operator_result_id="wrong"),
)
_refused(
ReplayRefusalReason.CANDIDATE_IDENTITY_MISMATCH,
"binding_not_structurally_bound",
run=run,
result=result,
binding=replace(binding, run_attempt_membership="other"),
)
_refused(
ReplayRefusalReason.CANDIDATE_IDENTITY_MISMATCH,
"binding_not_successful",
run=run,
result=result,
binding=replace(binding, reason_codes=("blocked",)),
)
def test_result_and_attempt_mismatches_refuse() -> None:
run, result, binding = _chain()
_refused(
ReplayRefusalReason.CANDIDATE_IDENTITY_MISMATCH,
"result_run_mismatch",
run=run,
result=replace(result, geometric_search_run_id="wrong"),
binding=binding,
)
_refused(
ReplayRefusalReason.CANDIDATE_IDENTITY_MISMATCH,
"binding_result_mismatch",
run=run,
result=replace(result, attempt_id="wrong"),
binding=binding,
)
bad_attempt = replace(result.candidate_attempt, attempt_id="wrong")
_refused(
ReplayRefusalReason.CANDIDATE_IDENTITY_MISMATCH,
"binding_attempt_mismatch",
run=run,
result=replace(result, candidate_attempt=bad_attempt),
binding=binding,
)
bad_attempt = replace(result.candidate_attempt, input_digest="wrong")
_refused(
ReplayRefusalReason.CANDIDATE_IDENTITY_MISMATCH,
"attempt_input_digest_mismatch",
run=run,
result=replace(result, candidate_attempt=bad_attempt),
binding=binding,
)
def test_reconstruction_mismatches_refuse() -> None:
run, result, binding = _chain()
bad = replace(result.candidate_reconstruction, candidate_reconstruction_digest="wrong")
_refused(
ReplayRefusalReason.CANDIDATE_IDENTITY_MISMATCH,
"binding_reconstruction_mismatch",
run=run,
result=replace(result, candidate_reconstruction=bad),
binding=binding,
)
bad = replace(result.candidate_reconstruction, problem_frame_digest="wrong")
_refused(
ReplayRefusalReason.CANDIDATE_IDENTITY_MISMATCH,
"reconstruction_problem_frame_mismatch",
run=run,
result=replace(result, candidate_reconstruction=bad),
binding=binding,
)
bad = replace(result.candidate_reconstruction, original_contract_assessment_id="wrong")
_refused(
ReplayRefusalReason.CANDIDATE_IDENTITY_MISMATCH,
"reconstruction_assessment_mismatch",
run=run,
result=replace(result, candidate_reconstruction=bad),
binding=binding,
)
bad = replace(result.candidate_reconstruction, source_residual_id="missing")
_refused(
ReplayRefusalReason.CANDIDATE_IDENTITY_MISMATCH,
"reconstruction_residual_mismatch",
run=run,
result=replace(result, candidate_reconstruction=bad),
binding=binding,
)
def test_evidence_and_policy_refusals() -> None:
run, result, binding = _chain()
other_span = (SourceSpan(text="other", start=20, end=25, sentence_index=0),)
_refused(
ReplayRefusalReason.CANDIDATE_IDENTITY_MISMATCH,
"binding_evidence_mismatch",
run=run,
result=result,
binding=replace(binding, evidence_spans=other_span),
)
_refused(
ReplayRefusalReason.INVALID_REPLAY_INPUT,
"invalid_proof_obligation_refs",
run=run,
result=result,
binding=binding,
proof_obligation_refs=(object(),),
)
_refused(
ReplayRefusalReason.UNSUPPORTED_SCHEMA_VERSION,
"unsupported_schema_version",
run=run,
result=result,
binding=binding,
schema_versions=(("b", "1"), ("a", "1")),
)
_refused(
ReplayRefusalReason.UNSUPPORTED_REPLAY_POLICY,
"unsupported_replay_policy_version",
run=run,
result=result,
binding=binding,
replay_policy_version="contract_proof_replay.future",
)
def test_unsupported_candidate_organ_refuses() -> None:
run, result, binding = _chain()
bad_result = replace(result, candidate_organ="unsupported_candidate_organ")
_refused(
ReplayRefusalReason.INVALID_REPLAY_INPUT,
"unsupported_candidate_organ",
run=run,
result=bad_result,
binding=binding,
)
def test_contract_policy_constant_is_preserved() -> None:
assert CONTRACT_PROOF_REPLAY_POLICY_VERSION == "contract_proof_replay.v1"