1081 lines
No EOL
37 KiB
Python
1081 lines
No EOL
37 KiB
Python
from __future__ import annotations
|
|
|
|
import ast
|
|
import dataclasses
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from generate.compute_budget import ComputeBudgetDecision, ComputeBudgetStatus
|
|
from generate.geometric_search_run import (
|
|
GEOMETRIC_SEARCH_RUN_POLICY_VERSION,
|
|
GEOMETRIC_SEARCH_RUN_SCHEMA_VERSION,
|
|
BudgetCharge,
|
|
BudgetConsumed,
|
|
CandidateAttempt,
|
|
CandidateReplayStatus,
|
|
GeometricSearchRun,
|
|
SearchRunDisposition,
|
|
initialize_geometric_search_run,
|
|
)
|
|
from generate.kernel_facts import SourceSpan
|
|
from generate.replay_adapter import (
|
|
CONTRACT_PROOF_REPLAY_POLICY_VERSION,
|
|
REPLAY_ADAPTER_POLICY_VERSION,
|
|
VACUOUS_PROOF_DECLARATION,
|
|
ProofReplayRef,
|
|
ReplayAdapterInput,
|
|
ReplayAdapterRefusal,
|
|
ReplayAdapterResult,
|
|
ReplayDisposition,
|
|
ReplayRefusalReason,
|
|
build_replay_adapter_input,
|
|
classify_replay_result,
|
|
)
|
|
from generate.search_gate import SearchGateDecision, SearchGateStatus
|
|
|
|
DEFAULT_CANDIDATE_ORGAN = "unary_delta_transition"
|
|
DEFAULT_CONTRACT_TARGET = "problem_frame_contracts.unary_delta"
|
|
|
|
|
|
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 _proof_ref_payload(ref: ProofReplayRef) -> dict[str, object]:
|
|
return {
|
|
"proof_obligation_id": ref.proof_obligation_id,
|
|
"proof_obligation_version": ref.proof_obligation_version,
|
|
"proof_replay_id": ref.proof_replay_id,
|
|
"closed": ref.closed,
|
|
"reason_code": ref.reason_code,
|
|
}
|
|
|
|
|
|
def _gate(
|
|
*,
|
|
input_digest: str = "a" * 64,
|
|
residual_ids: tuple[str, ...] = ("residual-a",),
|
|
) -> SearchGateDecision:
|
|
payload = {
|
|
"policy_version": "search_gate.v1",
|
|
"input_digest": input_digest,
|
|
"residual_ids": list(residual_ids),
|
|
"candidate_organ": "unary_delta_transition",
|
|
"status": SearchGateStatus.ELIGIBLE.value,
|
|
"reason_code": "eligible_missing_role",
|
|
"evidence_spans": [],
|
|
}
|
|
return SearchGateDecision(
|
|
decision_id=_digest(payload),
|
|
policy_version="search_gate.v1",
|
|
input_digest=input_digest,
|
|
residual_ids=residual_ids,
|
|
candidate_organ="unary_delta_transition",
|
|
status=SearchGateStatus.ELIGIBLE,
|
|
reason_code="eligible_missing_role",
|
|
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": [],
|
|
}
|
|
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=(),
|
|
explanation="budget prose",
|
|
)
|
|
|
|
|
|
def _attempt(
|
|
*,
|
|
attempt_id: str = "attempt-a",
|
|
attempt_index: int = 0,
|
|
input_digest: str = "d" * 64,
|
|
candidate_digest: str = "b" * 64,
|
|
evidence_spans: tuple[SourceSpan, ...] = (SourceSpan("x", 0, 1, 0),),
|
|
) -> CandidateAttempt:
|
|
return CandidateAttempt(
|
|
attempt_id=attempt_id,
|
|
attempt_index=attempt_index,
|
|
parent_attempt_id=None,
|
|
operator_id="operator-a",
|
|
operator_version="operator.v1",
|
|
input_digest=input_digest,
|
|
candidate_digest=candidate_digest,
|
|
budget_charge=BudgetCharge(candidates=1, steps=1),
|
|
depth=1,
|
|
step_index=0,
|
|
replay_status=CandidateReplayStatus.REPLAY_PENDING,
|
|
replay_blockers=("replay_not_authorized",),
|
|
evidence_spans=evidence_spans,
|
|
explanation="attempt prose",
|
|
)
|
|
|
|
|
|
def _run_with_attempt(
|
|
attempt: CandidateAttempt | None = None,
|
|
*,
|
|
input_digest: str = "d" * 64,
|
|
) -> GeometricSearchRun:
|
|
gate = _gate()
|
|
budget = _budget(gate)
|
|
base = initialize_geometric_search_run(
|
|
problem_frame_digest="f" * 64,
|
|
contract_assessment_id="assessment-a",
|
|
residual_ids=gate.residual_ids,
|
|
gate_decision=gate,
|
|
compute_budget=budget,
|
|
operator_set_id="operator-set-empty",
|
|
operator_set_version="operators.v1",
|
|
)
|
|
assert isinstance(base, GeometricSearchRun)
|
|
selected = attempt or _attempt(input_digest=input_digest)
|
|
consumed = BudgetConsumed(
|
|
candidates_considered=1,
|
|
max_candidates=5,
|
|
depth_reached=1,
|
|
max_depth=2,
|
|
steps_used=1,
|
|
max_steps=10,
|
|
parallelism_used=1,
|
|
max_parallelism=1,
|
|
exhausted=False,
|
|
)
|
|
run_payload = {
|
|
"run_id": "",
|
|
"run_policy_version": base.run_policy_version,
|
|
"schema_version": base.schema_version,
|
|
"problem_frame_digest": base.problem_frame_digest,
|
|
"contract_assessment_id": base.contract_assessment_id,
|
|
"residual_ids": list(base.residual_ids),
|
|
"gate_decision_id": base.gate_decision_id,
|
|
"budget_id": base.budget_id,
|
|
"operator_set_id": base.operator_set_id,
|
|
"operator_set_version": base.operator_set_version,
|
|
"input_digest": input_digest,
|
|
"candidate_attempts": [
|
|
{
|
|
"attempt_id": selected.attempt_id,
|
|
"attempt_index": selected.attempt_index,
|
|
"parent_attempt_id": selected.parent_attempt_id,
|
|
"operator_id": selected.operator_id,
|
|
"operator_version": selected.operator_version,
|
|
"input_digest": selected.input_digest,
|
|
"candidate_digest": selected.candidate_digest,
|
|
"budget_charge": {
|
|
"candidates": selected.budget_charge.candidates,
|
|
"steps": selected.budget_charge.steps,
|
|
},
|
|
"depth": selected.depth,
|
|
"step_index": selected.step_index,
|
|
"replay_status": selected.replay_status.value,
|
|
"replay_blockers": list(selected.replay_blockers),
|
|
"evidence_spans": [
|
|
_span_payload(span) for span in selected.evidence_spans
|
|
],
|
|
}
|
|
],
|
|
"budget_consumed": {
|
|
"candidates_considered": consumed.candidates_considered,
|
|
"max_candidates": consumed.max_candidates,
|
|
"depth_reached": consumed.depth_reached,
|
|
"max_depth": consumed.max_depth,
|
|
"steps_used": consumed.steps_used,
|
|
"max_steps": consumed.max_steps,
|
|
"parallelism_used": consumed.parallelism_used,
|
|
"max_parallelism": consumed.max_parallelism,
|
|
"exhausted": consumed.exhausted,
|
|
},
|
|
"run_disposition": SearchRunDisposition.CANDIDATE_REPLAY_PENDING.value,
|
|
"exhaustion_code": None,
|
|
}
|
|
return GeometricSearchRun(
|
|
run_id=_digest(run_payload),
|
|
run_policy_version=base.run_policy_version,
|
|
schema_version=base.schema_version,
|
|
problem_frame_digest=base.problem_frame_digest,
|
|
contract_assessment_id=base.contract_assessment_id,
|
|
residual_ids=base.residual_ids,
|
|
gate_decision_id=base.gate_decision_id,
|
|
budget_id=base.budget_id,
|
|
operator_set_id=base.operator_set_id,
|
|
operator_set_version=base.operator_set_version,
|
|
input_digest=input_digest,
|
|
candidate_attempts=(selected,),
|
|
budget_consumed=consumed,
|
|
run_disposition=SearchRunDisposition.CANDIDATE_REPLAY_PENDING,
|
|
exhaustion_code=None,
|
|
explanation="diagnostic run with one attempt",
|
|
)
|
|
|
|
|
|
def _build_input(
|
|
*,
|
|
run: GeometricSearchRun | None = None,
|
|
attempt: CandidateAttempt | None = None,
|
|
candidate_digest: str = "b" * 64,
|
|
reconstruction_digest: str = "c" * 64,
|
|
**changes: object,
|
|
) -> ReplayAdapterInput | ReplayAdapterRefusal:
|
|
selected_run = run or _run_with_attempt()
|
|
selected_attempt = attempt or selected_run.candidate_attempts[0]
|
|
values: dict[str, object] = {
|
|
"run": selected_run,
|
|
"attempt": selected_attempt,
|
|
"candidate_digest": candidate_digest,
|
|
"candidate_reconstruction_digest": reconstruction_digest,
|
|
"problem_frame_digest": selected_run.problem_frame_digest,
|
|
"original_contract_assessment_id": selected_run.contract_assessment_id,
|
|
"candidate_organ": DEFAULT_CANDIDATE_ORGAN,
|
|
}
|
|
values.update(changes)
|
|
return build_replay_adapter_input(**values) # type: ignore[arg-type]
|
|
|
|
|
|
def _expected_input_digest(
|
|
*,
|
|
run: GeometricSearchRun,
|
|
attempt: CandidateAttempt,
|
|
candidate_digest: str,
|
|
candidate_reconstruction_digest: str,
|
|
problem_frame_digest: str,
|
|
original_contract_assessment_id: str,
|
|
candidate_organ: str = DEFAULT_CANDIDATE_ORGAN,
|
|
contract_replay_target: str = DEFAULT_CONTRACT_TARGET,
|
|
proof_obligation_refs: tuple[str, ...] = (),
|
|
schema_versions: tuple[tuple[str, str], ...] = (),
|
|
) -> str:
|
|
return _digest(
|
|
{
|
|
"replay_policy_version": CONTRACT_PROOF_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": problem_frame_digest,
|
|
"original_contract_assessment_id": original_contract_assessment_id,
|
|
"candidate_organ": candidate_organ,
|
|
"residual_ids": list(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": list(proof_obligation_refs),
|
|
"schema_versions": [[name, version] for name, version in schema_versions],
|
|
}
|
|
)
|
|
|
|
|
|
def _expected_result_id(
|
|
*,
|
|
replay_input: ReplayAdapterInput,
|
|
contract_replay_assessment_id: str,
|
|
proof_replay_refs: tuple[ProofReplayRef, ...],
|
|
replay_disposition: ReplayDisposition,
|
|
reason_codes: tuple[str, ...],
|
|
evidence_spans: tuple[SourceSpan, ...],
|
|
) -> str:
|
|
return _digest(
|
|
{
|
|
"replay_result_id": "",
|
|
"replay_policy_version": replay_input.replay_policy_version,
|
|
"input_digest": replay_input.input_digest,
|
|
"run_id": replay_input.run_id,
|
|
"attempt_id": replay_input.attempt_id,
|
|
"candidate_digest": replay_input.candidate_digest,
|
|
"contract_replay_assessment_id": contract_replay_assessment_id,
|
|
"proof_obligation_refs": list(replay_input.proof_obligation_refs),
|
|
"proof_replay_refs": [_proof_ref_payload(ref) for ref in proof_replay_refs],
|
|
"replay_disposition": replay_disposition.value,
|
|
"reason_codes": list(reason_codes),
|
|
"evidence_spans": [_span_payload(span) for span in evidence_spans],
|
|
}
|
|
)
|
|
|
|
|
|
def _expected_refusal_id(
|
|
*,
|
|
input_digest: str | None,
|
|
run_id: str | None,
|
|
attempt_id: str | None,
|
|
candidate_digest: str | None,
|
|
replay_refusal_reason: ReplayRefusalReason,
|
|
reason_codes: tuple[str, ...],
|
|
) -> str:
|
|
return _digest(
|
|
{
|
|
"replay_refusal_id": "",
|
|
"replay_policy_version": REPLAY_ADAPTER_POLICY_VERSION,
|
|
"input_digest": input_digest,
|
|
"run_id": run_id,
|
|
"attempt_id": attempt_id,
|
|
"candidate_digest": candidate_digest,
|
|
"replay_disposition": replay_refusal_reason.value,
|
|
"reason_codes": list(reason_codes),
|
|
}
|
|
)
|
|
|
|
|
|
def test_public_api_exports_are_exact() -> None:
|
|
import generate.replay_adapter as replay_adapter
|
|
|
|
assert CONTRACT_PROOF_REPLAY_POLICY_VERSION == "contract_proof_replay.v1"
|
|
assert REPLAY_ADAPTER_POLICY_VERSION == CONTRACT_PROOF_REPLAY_POLICY_VERSION
|
|
assert tuple(replay_adapter.__all__) == (
|
|
"CONTRACT_PROOF_REPLAY_POLICY_VERSION",
|
|
"REPLAY_ADAPTER_POLICY_VERSION",
|
|
"VACUOUS_PROOF_DECLARATION",
|
|
"ReplayDisposition",
|
|
"ReplayRefusalReason",
|
|
"ProofReplayRef",
|
|
"ReplayAdapterInput",
|
|
"ReplayAdapterResult",
|
|
"ReplayAdapterRefusal",
|
|
"build_replay_adapter_input",
|
|
"classify_replay_result",
|
|
)
|
|
|
|
|
|
def test_input_construction_success_and_deterministic_digest() -> None:
|
|
run = _run_with_attempt()
|
|
attempt = run.candidate_attempts[0]
|
|
outcome = _build_input(run=run, attempt=attempt)
|
|
|
|
assert isinstance(outcome, ReplayAdapterInput)
|
|
assert outcome.run_id == run.run_id
|
|
assert outcome.attempt_id == attempt.attempt_id
|
|
assert outcome.candidate_digest == attempt.candidate_digest
|
|
assert outcome.candidate_reconstruction_digest == "c" * 64
|
|
assert outcome.problem_frame_digest == run.problem_frame_digest
|
|
assert outcome.original_contract_assessment_id == run.contract_assessment_id
|
|
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.candidate_organ == DEFAULT_CANDIDATE_ORGAN
|
|
assert outcome.contract_replay_target == DEFAULT_CONTRACT_TARGET
|
|
assert outcome.input_digest == _expected_input_digest(
|
|
run=run,
|
|
attempt=attempt,
|
|
candidate_digest=attempt.candidate_digest,
|
|
candidate_reconstruction_digest="c" * 64,
|
|
problem_frame_digest=run.problem_frame_digest,
|
|
original_contract_assessment_id=run.contract_assessment_id,
|
|
)
|
|
|
|
|
|
def test_adapter_failure_is_distinct_from_contract_refused() -> None:
|
|
refusal = _build_input(candidate_digest="z" * 64)
|
|
assert isinstance(refusal, ReplayAdapterRefusal)
|
|
assert refusal.replay_disposition is ReplayRefusalReason.CANDIDATE_IDENTITY_MISMATCH
|
|
|
|
replay_input = _build_input()
|
|
assert isinstance(replay_input, ReplayAdapterInput)
|
|
result = classify_replay_result(
|
|
replay_input,
|
|
contract_replay_assessment_id="e" * 64,
|
|
contract_closed=False,
|
|
)
|
|
assert isinstance(result, ReplayAdapterResult)
|
|
assert result.replay_disposition is ReplayDisposition.CONTRACT_REFUSED
|
|
assert type(refusal) is not type(result)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("factory", "expected_reason"),
|
|
(
|
|
(
|
|
lambda: _build_input(
|
|
attempt=_attempt(attempt_id="missing"),
|
|
),
|
|
ReplayRefusalReason.CANDIDATE_IDENTITY_MISMATCH,
|
|
),
|
|
(
|
|
lambda: _build_input(
|
|
attempt=_attempt(attempt_index=9),
|
|
),
|
|
ReplayRefusalReason.CANDIDATE_IDENTITY_MISMATCH,
|
|
),
|
|
(
|
|
lambda: _build_input(candidate_digest=""),
|
|
ReplayRefusalReason.INVALID_REPLAY_INPUT,
|
|
),
|
|
(
|
|
lambda: _build_input(candidate_reconstruction_digest=""),
|
|
ReplayRefusalReason.INVALID_REPLAY_INPUT,
|
|
),
|
|
(
|
|
lambda: _build_input(problem_frame_digest=""),
|
|
ReplayRefusalReason.INVALID_REPLAY_INPUT,
|
|
),
|
|
(
|
|
lambda: _build_input(original_contract_assessment_id=""),
|
|
ReplayRefusalReason.INVALID_REPLAY_INPUT,
|
|
),
|
|
),
|
|
)
|
|
def test_identity_mismatches_fail_closed(
|
|
factory: object, expected_reason: ReplayRefusalReason
|
|
) -> None:
|
|
outcome = factory() # type: ignore[operator]
|
|
assert isinstance(outcome, ReplayAdapterRefusal)
|
|
assert outcome.replay_disposition is expected_reason
|
|
|
|
|
|
def test_duplicate_attempt_ids_fail_closed() -> None:
|
|
first = _attempt(attempt_id="dup", attempt_index=0)
|
|
second = _attempt(attempt_id="dup", attempt_index=1)
|
|
run = GeometricSearchRun(
|
|
run_id="r" * 64,
|
|
run_policy_version=GEOMETRIC_SEARCH_RUN_POLICY_VERSION,
|
|
schema_version=GEOMETRIC_SEARCH_RUN_SCHEMA_VERSION,
|
|
problem_frame_digest="f" * 64,
|
|
contract_assessment_id="assessment-a",
|
|
residual_ids=("residual-a",),
|
|
gate_decision_id="g" * 64,
|
|
budget_id="h" * 64,
|
|
operator_set_id="operator-set-empty",
|
|
operator_set_version="operators.v1",
|
|
input_digest="d" * 64,
|
|
candidate_attempts=(first, second),
|
|
budget_consumed=BudgetConsumed(
|
|
candidates_considered=2,
|
|
max_candidates=5,
|
|
depth_reached=1,
|
|
max_depth=2,
|
|
steps_used=2,
|
|
max_steps=10,
|
|
parallelism_used=1,
|
|
max_parallelism=1,
|
|
exhausted=False,
|
|
),
|
|
run_disposition=SearchRunDisposition.CANDIDATE_REPLAY_PENDING,
|
|
exhaustion_code=None,
|
|
explanation="duplicate attempt ids",
|
|
)
|
|
outcome = _build_input(run=run, attempt=first)
|
|
assert isinstance(outcome, ReplayAdapterRefusal)
|
|
assert "ambiguous_or_missing_attempt_id" in outcome.reason_codes
|
|
|
|
|
|
def test_unsupported_replay_policy_fails_closed() -> None:
|
|
run = _run_with_attempt()
|
|
attempt = run.candidate_attempts[0]
|
|
outcome = build_replay_adapter_input(
|
|
run=run,
|
|
attempt=attempt,
|
|
candidate_digest=attempt.candidate_digest,
|
|
candidate_reconstruction_digest="c" * 64,
|
|
problem_frame_digest=run.problem_frame_digest,
|
|
original_contract_assessment_id=run.contract_assessment_id,
|
|
candidate_organ=DEFAULT_CANDIDATE_ORGAN,
|
|
replay_policy_version="replay_adapter.future",
|
|
)
|
|
assert isinstance(outcome, ReplayAdapterRefusal)
|
|
assert outcome.replay_disposition is ReplayRefusalReason.UNSUPPORTED_REPLAY_POLICY
|
|
|
|
|
|
def test_contract_replay_unavailable_fails_closed() -> None:
|
|
replay_input = _build_input()
|
|
assert isinstance(replay_input, ReplayAdapterInput)
|
|
|
|
for kwargs in (
|
|
{"contract_replay_assessment_id": None, "contract_closed": True},
|
|
{"contract_replay_assessment_id": "e" * 64, "contract_closed": None},
|
|
):
|
|
outcome = classify_replay_result(replay_input, **kwargs) # type: ignore[arg-type]
|
|
assert isinstance(outcome, ReplayAdapterRefusal)
|
|
assert outcome.replay_disposition is ReplayRefusalReason.CONTRACT_REPLAY_UNAVAILABLE
|
|
|
|
|
|
def test_contract_refused_skips_proof_when_obligations_present() -> None:
|
|
run = _run_with_attempt()
|
|
attempt = run.candidate_attempts[0]
|
|
replay_input = build_replay_adapter_input(
|
|
run=run,
|
|
attempt=attempt,
|
|
candidate_digest=attempt.candidate_digest,
|
|
candidate_reconstruction_digest="c" * 64,
|
|
problem_frame_digest=run.problem_frame_digest,
|
|
original_contract_assessment_id=run.contract_assessment_id,
|
|
candidate_organ=DEFAULT_CANDIDATE_ORGAN,
|
|
proof_obligation_refs=("obligation-z", "obligation-a"),
|
|
)
|
|
assert isinstance(replay_input, ReplayAdapterInput)
|
|
result = classify_replay_result(
|
|
replay_input,
|
|
contract_replay_assessment_id="e" * 64,
|
|
contract_closed=False,
|
|
)
|
|
assert isinstance(result, ReplayAdapterResult)
|
|
assert result.replay_disposition is ReplayDisposition.CONTRACT_REFUSED
|
|
assert result.proof_replay_refs == ()
|
|
|
|
|
|
def test_contract_refused_replay_emits_result_not_refusal() -> None:
|
|
replay_input = _build_input()
|
|
assert isinstance(replay_input, ReplayAdapterInput)
|
|
result = classify_replay_result(
|
|
replay_input,
|
|
contract_replay_assessment_id="e" * 64,
|
|
contract_closed=False,
|
|
reason_codes=("missing_binding",),
|
|
)
|
|
assert isinstance(result, ReplayAdapterResult)
|
|
assert result.replay_disposition is ReplayDisposition.CONTRACT_REFUSED
|
|
assert result.proof_replay_refs == ()
|
|
|
|
|
|
def test_contract_closed_proof_refused_cases() -> None:
|
|
run = _run_with_attempt()
|
|
attempt = run.candidate_attempts[0]
|
|
replay_input = build_replay_adapter_input(
|
|
run=run,
|
|
attempt=attempt,
|
|
candidate_digest=attempt.candidate_digest,
|
|
candidate_reconstruction_digest="c" * 64,
|
|
problem_frame_digest=run.problem_frame_digest,
|
|
original_contract_assessment_id=run.contract_assessment_id,
|
|
candidate_organ=DEFAULT_CANDIDATE_ORGAN,
|
|
proof_obligation_refs=("obligation-a",),
|
|
)
|
|
assert isinstance(replay_input, ReplayAdapterInput)
|
|
|
|
missing_proof = classify_replay_result(
|
|
replay_input,
|
|
contract_replay_assessment_id="e" * 64,
|
|
contract_closed=True,
|
|
)
|
|
assert isinstance(missing_proof, ReplayAdapterResult)
|
|
assert (
|
|
missing_proof.replay_disposition
|
|
is ReplayDisposition.CONTRACT_CLOSED_BUT_PROOF_REFUSED
|
|
)
|
|
|
|
open_proof = classify_replay_result(
|
|
replay_input,
|
|
contract_replay_assessment_id="e" * 64,
|
|
contract_closed=True,
|
|
proof_replay_refs=(
|
|
ProofReplayRef(
|
|
proof_obligation_id="obligation-a",
|
|
proof_obligation_version="proof.v1",
|
|
proof_replay_id="p" * 64,
|
|
closed=False,
|
|
reason_code="proof_refused",
|
|
),
|
|
),
|
|
)
|
|
assert isinstance(open_proof, ReplayAdapterResult)
|
|
assert (
|
|
open_proof.replay_disposition
|
|
is ReplayDisposition.CONTRACT_CLOSED_BUT_PROOF_REFUSED
|
|
)
|
|
|
|
|
|
def test_contract_closed_without_obligations_fails_without_declaration() -> None:
|
|
replay_input = _build_input()
|
|
assert isinstance(replay_input, ReplayAdapterInput)
|
|
outcome = classify_replay_result(
|
|
replay_input,
|
|
contract_replay_assessment_id="e" * 64,
|
|
contract_closed=True,
|
|
)
|
|
assert isinstance(outcome, ReplayAdapterRefusal)
|
|
assert outcome.replay_disposition is ReplayRefusalReason.PROOF_REPLAY_UNAVAILABLE
|
|
|
|
|
|
def test_contract_and_proof_closed_with_vacuous_proof_declaration() -> None:
|
|
replay_input = _build_input(
|
|
schema_versions=(VACUOUS_PROOF_DECLARATION,),
|
|
)
|
|
assert isinstance(replay_input, ReplayAdapterInput)
|
|
result = classify_replay_result(
|
|
replay_input,
|
|
contract_replay_assessment_id="e" * 64,
|
|
contract_closed=True,
|
|
)
|
|
assert isinstance(result, ReplayAdapterResult)
|
|
assert result.replay_disposition is ReplayDisposition.CONTRACT_AND_PROOF_CLOSED
|
|
assert result.proof_obligation_refs == ()
|
|
|
|
|
|
def test_contract_and_proof_closed_with_all_obligations_closed() -> None:
|
|
run = _run_with_attempt()
|
|
attempt = run.candidate_attempts[0]
|
|
replay_input = build_replay_adapter_input(
|
|
run=run,
|
|
attempt=attempt,
|
|
candidate_digest=attempt.candidate_digest,
|
|
candidate_reconstruction_digest="c" * 64,
|
|
problem_frame_digest=run.problem_frame_digest,
|
|
original_contract_assessment_id=run.contract_assessment_id,
|
|
candidate_organ=DEFAULT_CANDIDATE_ORGAN,
|
|
proof_obligation_refs=("obligation-a", "obligation-b"),
|
|
)
|
|
assert isinstance(replay_input, ReplayAdapterInput)
|
|
refs = (
|
|
ProofReplayRef("obligation-a", "proof.v1", "p1" * 32, True, "closed"),
|
|
ProofReplayRef("obligation-b", "proof.v1", "p2" * 32, True, "closed"),
|
|
)
|
|
result = classify_replay_result(
|
|
replay_input,
|
|
contract_replay_assessment_id="e" * 64,
|
|
contract_closed=True,
|
|
proof_replay_refs=refs,
|
|
)
|
|
assert isinstance(result, ReplayAdapterResult)
|
|
assert result.replay_disposition is ReplayDisposition.CONTRACT_AND_PROOF_CLOSED
|
|
assert result.proof_replay_refs == refs
|
|
|
|
|
|
def test_multiple_candidates_are_not_selected() -> None:
|
|
first_input = _build_input(schema_versions=(VACUOUS_PROOF_DECLARATION,))
|
|
second_run = _run_with_attempt(
|
|
_attempt(candidate_digest="9" * 64),
|
|
input_digest="d" * 64,
|
|
)
|
|
second_attempt = second_run.candidate_attempts[0]
|
|
second_input = _build_input(
|
|
run=second_run,
|
|
attempt=second_attempt,
|
|
candidate_digest=second_attempt.candidate_digest,
|
|
schema_versions=(VACUOUS_PROOF_DECLARATION,),
|
|
)
|
|
assert isinstance(first_input, ReplayAdapterInput)
|
|
assert isinstance(second_input, ReplayAdapterInput)
|
|
|
|
first_result = classify_replay_result(
|
|
first_input,
|
|
contract_replay_assessment_id="e" * 64,
|
|
contract_closed=True,
|
|
)
|
|
second_result = classify_replay_result(
|
|
second_input,
|
|
contract_replay_assessment_id="f" * 64,
|
|
contract_closed=True,
|
|
)
|
|
assert isinstance(first_result, ReplayAdapterResult)
|
|
assert isinstance(second_result, ReplayAdapterResult)
|
|
assert first_result.replay_result_id != second_result.replay_result_id
|
|
forbidden = {"rank", "score", "priority", "selected", "best", "answer"}
|
|
for record in (first_result, second_result):
|
|
assert forbidden.isdisjoint({field.name for field in dataclasses.fields(record)})
|
|
|
|
|
|
def test_deterministic_ids_and_explanation_exclusion() -> None:
|
|
replay_input = _build_input()
|
|
assert isinstance(replay_input, ReplayAdapterInput)
|
|
spans = (SourceSpan("a", 0, 1, 0), SourceSpan("a", 0, 1, 0))
|
|
|
|
first = classify_replay_result(
|
|
replay_input,
|
|
contract_replay_assessment_id="e" * 64,
|
|
contract_closed=False,
|
|
evidence_spans=spans,
|
|
explanation="first prose",
|
|
)
|
|
second = classify_replay_result(
|
|
replay_input,
|
|
contract_replay_assessment_id="e" * 64,
|
|
contract_closed=False,
|
|
evidence_spans=spans,
|
|
explanation="second prose",
|
|
)
|
|
assert isinstance(first, ReplayAdapterResult)
|
|
assert isinstance(second, ReplayAdapterResult)
|
|
assert first.replay_result_id == second.replay_result_id
|
|
assert first.replay_result_id == _expected_result_id(
|
|
replay_input=replay_input,
|
|
contract_replay_assessment_id="e" * 64,
|
|
proof_replay_refs=(),
|
|
replay_disposition=ReplayDisposition.CONTRACT_REFUSED,
|
|
reason_codes=(),
|
|
evidence_spans=spans,
|
|
)
|
|
|
|
changed_assessment = classify_replay_result(
|
|
replay_input,
|
|
contract_replay_assessment_id="f" * 64,
|
|
contract_closed=False,
|
|
evidence_spans=spans,
|
|
)
|
|
assert isinstance(changed_assessment, ReplayAdapterResult)
|
|
assert changed_assessment.replay_result_id != first.replay_result_id
|
|
|
|
closed_input = _build_input(schema_versions=(VACUOUS_PROOF_DECLARATION,))
|
|
assert isinstance(closed_input, ReplayAdapterInput)
|
|
changed_disposition = classify_replay_result(
|
|
closed_input,
|
|
contract_replay_assessment_id="e" * 64,
|
|
contract_closed=True,
|
|
evidence_spans=spans,
|
|
)
|
|
assert isinstance(changed_disposition, ReplayAdapterResult)
|
|
assert changed_disposition.replay_result_id != first.replay_result_id
|
|
|
|
changed_run = _run_with_attempt(
|
|
_attempt(candidate_digest="1" * 64),
|
|
input_digest="d" * 64,
|
|
)
|
|
changed_attempt = changed_run.candidate_attempts[0]
|
|
changed_input = _build_input(
|
|
run=changed_run,
|
|
attempt=changed_attempt,
|
|
candidate_digest=changed_attempt.candidate_digest,
|
|
)
|
|
assert isinstance(changed_input, ReplayAdapterInput)
|
|
assert changed_input.input_digest != replay_input.input_digest
|
|
|
|
|
|
def test_evidence_spans_preserve_order_and_duplicates() -> None:
|
|
replay_input = _build_input()
|
|
assert isinstance(replay_input, ReplayAdapterInput)
|
|
ordered = (
|
|
SourceSpan("first", 0, 5, 0),
|
|
SourceSpan("second", 6, 11, 1),
|
|
SourceSpan("first", 0, 5, 0),
|
|
)
|
|
result = classify_replay_result(
|
|
replay_input,
|
|
contract_replay_assessment_id="e" * 64,
|
|
contract_closed=False,
|
|
evidence_spans=ordered,
|
|
)
|
|
assert isinstance(result, ReplayAdapterResult)
|
|
assert result.evidence_spans == ordered
|
|
|
|
reordered = (
|
|
SourceSpan("second", 6, 11, 1),
|
|
SourceSpan("first", 0, 5, 0),
|
|
SourceSpan("first", 0, 5, 0),
|
|
)
|
|
reordered_result = classify_replay_result(
|
|
replay_input,
|
|
contract_replay_assessment_id="e" * 64,
|
|
contract_closed=False,
|
|
evidence_spans=reordered,
|
|
)
|
|
assert isinstance(reordered_result, ReplayAdapterResult)
|
|
assert reordered_result.replay_result_id != result.replay_result_id
|
|
|
|
|
|
def test_malformed_evidence_spans_fail_closed() -> None:
|
|
replay_input = _build_input()
|
|
assert isinstance(replay_input, ReplayAdapterInput)
|
|
bad_span = SimpleNamespace(text="bad", start=3, end=1, sentence_index=0)
|
|
outcome = classify_replay_result(
|
|
replay_input,
|
|
contract_replay_assessment_id="e" * 64,
|
|
contract_closed=False,
|
|
evidence_spans=(bad_span,), # type: ignore[arg-type]
|
|
)
|
|
assert isinstance(outcome, ReplayAdapterRefusal)
|
|
assert outcome.replay_disposition is ReplayRefusalReason.INVALID_REPLAY_INPUT
|
|
|
|
|
|
def test_proof_replay_refs_preserve_order_and_validate() -> None:
|
|
run = _run_with_attempt()
|
|
attempt = run.candidate_attempts[0]
|
|
replay_input = build_replay_adapter_input(
|
|
run=run,
|
|
attempt=attempt,
|
|
candidate_digest=attempt.candidate_digest,
|
|
candidate_reconstruction_digest="c" * 64,
|
|
problem_frame_digest=run.problem_frame_digest,
|
|
original_contract_assessment_id=run.contract_assessment_id,
|
|
candidate_organ=DEFAULT_CANDIDATE_ORGAN,
|
|
proof_obligation_refs=("obligation-a", "obligation-b"),
|
|
)
|
|
assert isinstance(replay_input, ReplayAdapterInput)
|
|
refs = (
|
|
ProofReplayRef("obligation-a", "proof.v1", "p1" * 32, True, "closed"),
|
|
ProofReplayRef("obligation-b", "proof.v1", "p2" * 32, True, "closed"),
|
|
)
|
|
result = classify_replay_result(
|
|
replay_input,
|
|
contract_replay_assessment_id="e" * 64,
|
|
contract_closed=True,
|
|
proof_replay_refs=refs,
|
|
)
|
|
assert isinstance(result, ReplayAdapterResult)
|
|
assert result.proof_replay_refs == refs
|
|
|
|
malformed = classify_replay_result(
|
|
replay_input,
|
|
contract_replay_assessment_id="e" * 64,
|
|
contract_closed=True,
|
|
proof_replay_refs=(
|
|
ProofReplayRef("", "proof.v1", "p1" * 32, True, "closed"),
|
|
),
|
|
)
|
|
assert isinstance(malformed, ReplayAdapterRefusal)
|
|
|
|
|
|
def test_unsupported_proof_obligation_fails_closed() -> None:
|
|
run = _run_with_attempt()
|
|
attempt = run.candidate_attempts[0]
|
|
replay_input = build_replay_adapter_input(
|
|
run=run,
|
|
attempt=attempt,
|
|
candidate_digest=attempt.candidate_digest,
|
|
candidate_reconstruction_digest="c" * 64,
|
|
problem_frame_digest=run.problem_frame_digest,
|
|
original_contract_assessment_id=run.contract_assessment_id,
|
|
candidate_organ=DEFAULT_CANDIDATE_ORGAN,
|
|
proof_obligation_refs=("obligation-a",),
|
|
)
|
|
assert isinstance(replay_input, ReplayAdapterInput)
|
|
outcome = classify_replay_result(
|
|
replay_input,
|
|
contract_replay_assessment_id="e" * 64,
|
|
contract_closed=True,
|
|
proof_replay_refs=(
|
|
ProofReplayRef("other", "proof.v1", "p1" * 32, True, "closed"),
|
|
),
|
|
)
|
|
assert isinstance(outcome, ReplayAdapterRefusal)
|
|
assert outcome.replay_disposition is ReplayRefusalReason.UNSUPPORTED_PROOF_OBLIGATION
|
|
|
|
|
|
def test_public_dataclasses_expose_no_authority_fields() -> None:
|
|
forbidden = {
|
|
"answer",
|
|
"final_answer",
|
|
"served_output",
|
|
"promotion",
|
|
"mutation",
|
|
"teaching_update",
|
|
"pack_update",
|
|
"policy_update",
|
|
"identity_update",
|
|
"workbench_state",
|
|
"runtime_effect",
|
|
"confidence",
|
|
"score",
|
|
"rank",
|
|
"priority",
|
|
"selected",
|
|
"best",
|
|
"serving_allowed",
|
|
"runnable",
|
|
"tool_call",
|
|
}
|
|
for record_type in (
|
|
ProofReplayRef,
|
|
ReplayAdapterInput,
|
|
ReplayAdapterResult,
|
|
ReplayAdapterRefusal,
|
|
):
|
|
assert forbidden.isdisjoint(
|
|
{field.name for field in dataclasses.fields(record_type)}
|
|
)
|
|
|
|
|
|
def test_module_coupling_and_side_effect_guards() -> None:
|
|
path = Path("generate/replay_adapter.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.geometric_search_run",
|
|
"generate.kernel_facts",
|
|
"generate.candidate_operator",
|
|
"generate.run_attempt_binding",
|
|
}
|
|
assert {
|
|
"build_missing_role_candidate",
|
|
"candidate_operator_set_id",
|
|
"bind_candidate_attempt_to_run",
|
|
"initialize_geometric_search_run",
|
|
"build_practice_trace_input",
|
|
"build_bound_practice_trace_input",
|
|
"seal_practice_trace",
|
|
"seal_bound_practice_trace",
|
|
"decide_search_gate",
|
|
"decide_compute_budget",
|
|
"assess_contracts",
|
|
"project_contract_residuals",
|
|
"determine",
|
|
}.isdisjoint(imported_names)
|
|
assert {
|
|
"build_missing_role_candidate",
|
|
"candidate_operator_set_id",
|
|
"bind_candidate_attempt_to_run",
|
|
"initialize_geometric_search_run",
|
|
"build_practice_trace_input",
|
|
"build_bound_practice_trace_input",
|
|
"seal_practice_trace",
|
|
"seal_bound_practice_trace",
|
|
"decide_search_gate",
|
|
"decide_compute_budget",
|
|
"assess_contracts",
|
|
"project_contract_residuals",
|
|
"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/compute_budget.py",
|
|
"generate/search_gate.py",
|
|
"generate/contract_residuals.py",
|
|
):
|
|
assert "replay_adapter" not in Path(upstream).read_text("utf-8")
|
|
|
|
|
|
def test_no_filesystem_network_clock_random_or_environment_identity() -> None:
|
|
source = Path("generate/replay_adapter.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_ordinary_malformed_objects_fail_closed_without_throwing() -> None:
|
|
outcome = build_replay_adapter_input(
|
|
run=SimpleNamespace(), # type: ignore[arg-type]
|
|
attempt=SimpleNamespace(), # type: ignore[arg-type]
|
|
candidate_digest="b" * 64,
|
|
candidate_reconstruction_digest="c" * 64,
|
|
problem_frame_digest="f" * 64,
|
|
original_contract_assessment_id="assessment-a",
|
|
candidate_organ=DEFAULT_CANDIDATE_ORGAN,
|
|
)
|
|
assert isinstance(outcome, ReplayAdapterRefusal)
|
|
|
|
|
|
def test_proof_obligation_refs_preserve_declared_order() -> None:
|
|
run = _run_with_attempt()
|
|
attempt = run.candidate_attempts[0]
|
|
ordered = ("obligation-z", "obligation-a")
|
|
replay_input = build_replay_adapter_input(
|
|
run=run,
|
|
attempt=attempt,
|
|
candidate_digest=attempt.candidate_digest,
|
|
candidate_reconstruction_digest="c" * 64,
|
|
problem_frame_digest=run.problem_frame_digest,
|
|
original_contract_assessment_id=run.contract_assessment_id,
|
|
candidate_organ=DEFAULT_CANDIDATE_ORGAN,
|
|
proof_obligation_refs=ordered,
|
|
)
|
|
assert isinstance(replay_input, ReplayAdapterInput)
|
|
assert replay_input.proof_obligation_refs == ordered
|
|
|
|
|
|
def test_unsupported_candidate_organ_fails_closed() -> None:
|
|
outcome = _build_input(candidate_organ="unknown_organ")
|
|
assert isinstance(outcome, ReplayAdapterRefusal)
|
|
assert "unsupported_candidate_organ" in outcome.reason_codes
|
|
|
|
|
|
def test_refusal_id_matches_canonical_payload() -> None:
|
|
refusal = _build_input(candidate_digest="")
|
|
assert isinstance(refusal, ReplayAdapterRefusal)
|
|
assert refusal.replay_disposition is ReplayRefusalReason.INVALID_REPLAY_INPUT
|
|
assert refusal.replay_refusal_id == _expected_refusal_id(
|
|
input_digest=None,
|
|
run_id=refusal.run_id,
|
|
attempt_id=refusal.attempt_id,
|
|
candidate_digest=None,
|
|
replay_refusal_reason=refusal.replay_disposition,
|
|
reason_codes=refusal.reason_codes,
|
|
) |