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

docs+kernel: support bound practice episode sealing

Browse source 
Extends SealedPracticeTrace for externally bound candidate episodes per
ADR-0233. Adds candidate_attempt_binding_ids to trace identity, bound trace
input/seal entry points, and end-to-end bound episode tests without mutating
GeometricSearchRun.candidate_attempts.
This commit is contained in:
Shay 2026-06-23 00:04:26 -07:00
parent 5b30ec877f
commit 484e8ff9fe
4 changed files with 1388 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

123
docs/adr/ADR-0233-bound-practice-episode-sealing.md Normal file
View file

@ -0,0 +1,123 @@
# ADR-0233: Bound Practice Episode Sealing
**Status:** Proposed
**Date:** 2026-06-23
**Scope:** Kernel diagnostics, immutable bound practice evidence, residual-gated practice loop
**Depends on:**
- ADR-0230 SealedPracticeTrace boundary
- ADR-0231 First Candidate Operator boundary
- ADR-0232 CandidateAttempt Run-Binding boundary
- PR #878 inert run-attempt binding shell
- PR #879 bound replay adapter input
## 1. Summary
This ADR authorizes bound practice episode sealing: the bridge from
`CandidateAttemptRunBinding` and bound replay evidence into
`SealedPracticeTrace`.
The bound spine is:
```text
GeometricSearchRun
-> CandidateOperatorResult
-> CandidateAttemptRunBinding
-> ReplayAdapterInput
-> ReplayAdapterResult | ReplayAdapterRefusal
-> SealedPracticeTrace
```
`CandidateAttemptRunBinding` is **membership evidence only**. It is not truth,
not proof, not replay success, and not serving authority.
Sealed traces must preserve binding evidence explicitly through:
```text
candidate_attempt_binding_ids
```
The bound path must **not** mutate `GeometricSearchRun.candidate_attempts`.
The bound path must **not** call candidate operators, search, replay input
builders, replay classifiers, or binding producers. It consumes only
already-built upstream records:
```text
GeometricSearchRun
CandidateAttemptRunBinding
CandidateOperatorResult
ReplayAdapterResult | ReplayAdapterRefusal
```
The bound path emits only:
```text
PracticeTraceInput | PracticeTraceRefusal
SealedPracticeTrace | PracticeTraceRefusal
```
This boundary is diagnostic and inert. It does not answer, rank, select, serve,
promote, repair, teach, mutate packs, mutate policy, mutate identity, or write
artifacts.
## 2. Decision
Extend `PracticeTraceInput` and `SealedPracticeTrace` with
`candidate_attempt_binding_ids`.
Add:
```text
build_bound_practice_trace_input(...)
seal_bound_practice_trace(...)
```
Legacy `build_practice_trace_input(...)` and `seal_practice_trace(...)` remain
unchanged in authority and continue to use `candidate_attempt_binding_ids=()`
for run-embedded attempts.
Canonical trace digests include binding IDs in:
```text
input_digest
trace_id
upstream_identity_chain
```
Identity chain order:
```text
problem_frame_digest
original_contract_assessment_id
residual_ids
search_gate_decision_id
compute_budget_id
geometric_search_run_id
candidate_attempt_ids
candidate_attempt_binding_ids
replay_result_ids
replay_refusal_ids
```
## 3. Authorized next PR
This ADR authorizes exactly one implementation PR:
```text
docs+kernel: support bound practice episode sealing
```
That PR may add only:
- binding-aware trace input construction
- binding-aware trace sealing
- tests for the bound episode path
- this ADR
It explicitly excludes candidate operators, Workbench, serving, ranking,
promotion, teaching mutation, pack/policy/identity mutation, runtime
integration, and artifact writing.

550
generate/sealed_practice_trace.py
View file

@ -65,6 +65,7 @@ class PracticeTraceInput:
compute_budget_id: str
geometric_search_run_id: str
candidate_attempt_ids: tuple[str, ...]
candidate_attempt_binding_ids: tuple[str, ...]
replay_result_ids: tuple[str, ...]
replay_refusal_ids: tuple[str, ...]
schema_versions: tuple[tuple[str, str], ...]
@ -83,6 +84,7 @@ class SealedPracticeTrace:
compute_budget_id: str
geometric_search_run_id: str
candidate_attempt_ids: tuple[str, ...]
candidate_attempt_binding_ids: tuple[str, ...]
replay_result_ids: tuple[str, ...]
replay_refusal_ids: tuple[str, ...]
upstream_identity_chain: tuple[str, ...]
@ -184,6 +186,7 @@ def _input_digest_payload(
compute_budget_id: str,
geometric_search_run_id: str,
candidate_attempt_ids: tuple[str, ...],
candidate_attempt_binding_ids: tuple[str, ...],
replay_result_ids: tuple[str, ...],
replay_refusal_ids: tuple[str, ...],
schema_versions: tuple[tuple[str, str], ...],
@ -198,6 +201,7 @@ def _input_digest_payload(
"compute_budget_id": compute_budget_id,
"geometric_search_run_id": geometric_search_run_id,
"candidate_attempt_ids": list(candidate_attempt_ids),
"candidate_attempt_binding_ids": list(candidate_attempt_binding_ids),
"replay_result_ids": list(replay_result_ids),
"replay_refusal_ids": list(replay_refusal_ids),
"schema_versions": _version_pairs_payload(schema_versions),
@ -216,6 +220,7 @@ def _trace_id_payload(
compute_budget_id: str,
geometric_search_run_id: str,
candidate_attempt_ids: tuple[str, ...],
candidate_attempt_binding_ids: tuple[str, ...],
replay_result_ids: tuple[str, ...],
replay_refusal_ids: tuple[str, ...],
upstream_identity_chain: tuple[str, ...],
@ -235,6 +240,7 @@ def _trace_id_payload(
"compute_budget_id": compute_budget_id,
"geometric_search_run_id": geometric_search_run_id,
"candidate_attempt_ids": list(candidate_attempt_ids),
"candidate_attempt_binding_ids": list(candidate_attempt_binding_ids),
"replay_result_ids": list(replay_result_ids),
"replay_refusal_ids": list(replay_refusal_ids),
"upstream_identity_chain": list(upstream_identity_chain),
@ -414,6 +420,7 @@ def _upstream_identity_chain(
compute_budget_id: str,
geometric_search_run_id: str,
candidate_attempt_ids: tuple[str, ...],
candidate_attempt_binding_ids: tuple[str, ...],
replay_result_ids: tuple[str, ...],
replay_refusal_ids: tuple[str, ...],
) -> tuple[str, ...]:
@ -425,6 +432,7 @@ def _upstream_identity_chain(
compute_budget_id,
geometric_search_run_id,
*candidate_attempt_ids,
*candidate_attempt_binding_ids,
*replay_result_ids,
*replay_refusal_ids,
)
@ -441,6 +449,22 @@ def _failure_disposition(reasons: list[str]) -> PracticeDisposition:
"budget_id_refusal_mismatch",
"geometric_search_run_id_mismatch",
"candidate_attempt_ids_mismatch",
"candidate_attempt_binding_ids_mismatch",
"invalid_binding_type",
"invalid_candidate_operator_result_type",
"binding_result_count_mismatch",
"binding_run_mismatch",
"binding_result_mismatch",
"binding_attempt_mismatch",
"binding_reconstruction_mismatch",
"binding_not_structurally_bound",
"binding_not_successful",
"binding_evidence_mismatch",
"operator_result_run_mismatch",
"operator_reconstruction_run_mismatch",
"duplicate_candidate_attempt_binding_id",
"duplicate_candidate_attempt_id",
"duplicate_candidate_digest",
"replay_run_id_mismatch",
"replay_orphan_attempt_id",
"replay_candidate_digest_mismatch",
@ -560,6 +584,7 @@ def build_practice_trace_input(
compute_budget_id=compute_budget_id,
geometric_search_run_id=geometric_search_run_id,
candidate_attempt_ids=candidate_attempt_ids,
candidate_attempt_binding_ids=(),
replay_result_ids=replay_result_ids,
replay_refusal_ids=replay_refusal_ids,
schema_versions=schema_versions,
@ -576,6 +601,7 @@ def build_practice_trace_input(
compute_budget_id=compute_budget_id,
geometric_search_run_id=geometric_search_run_id,
candidate_attempt_ids=candidate_attempt_ids,
candidate_attempt_binding_ids=(),
replay_result_ids=replay_result_ids,
replay_refusal_ids=replay_refusal_ids,
schema_versions=schema_versions,
@ -669,6 +695,7 @@ def seal_practice_trace(
compute_budget_id=trace_input.compute_budget_id,
geometric_search_run_id=trace_input.geometric_search_run_id,
candidate_attempt_ids=trace_input.candidate_attempt_ids,
candidate_attempt_binding_ids=trace_input.candidate_attempt_binding_ids,
replay_result_ids=trace_input.replay_result_ids,
replay_refusal_ids=trace_input.replay_refusal_ids,
schema_versions=trace_input.schema_versions,
@ -702,6 +729,7 @@ def seal_practice_trace(
compute_budget_id=trace_input.compute_budget_id,
geometric_search_run_id=trace_input.geometric_search_run_id,
candidate_attempt_ids=trace_input.candidate_attempt_ids,
candidate_attempt_binding_ids=trace_input.candidate_attempt_binding_ids,
replay_result_ids=trace_input.replay_result_ids,
replay_refusal_ids=trace_input.replay_refusal_ids,
)
@ -717,6 +745,7 @@ def seal_practice_trace(
compute_budget_id=trace_input.compute_budget_id,
geometric_search_run_id=trace_input.geometric_search_run_id,
candidate_attempt_ids=trace_input.candidate_attempt_ids,
candidate_attempt_binding_ids=trace_input.candidate_attempt_binding_ids,
replay_result_ids=trace_input.replay_result_ids,
replay_refusal_ids=trace_input.replay_refusal_ids,
upstream_identity_chain=identity_chain,
@ -741,6 +770,525 @@ def seal_practice_trace(
compute_budget_id=trace_input.compute_budget_id,
geometric_search_run_id=trace_input.geometric_search_run_id,
candidate_attempt_ids=trace_input.candidate_attempt_ids,
candidate_attempt_binding_ids=trace_input.candidate_attempt_binding_ids,
replay_result_ids=trace_input.replay_result_ids,
replay_refusal_ids=trace_input.replay_refusal_ids,
upstream_identity_chain=identity_chain,
practice_disposition=practice_disposition,
trace_records=trace_records,
evidence_spans=evidence_spans,
created_by_policy=CREATED_BY_POLICY,
explanation=safe_explanation,
)
def _valid_binding_record(binding: object) -> bool:
required = (
"binding_id",
"original_run_id",
"operator_result_id",
"candidate_attempt_id",
"attempt_index",
"candidate_digest",
"candidate_reconstruction_digest",
"run_attempt_membership",
"reason_codes",
"evidence_spans",
)
return all(hasattr(binding, name) for name in required)
def _valid_operator_result_record(operator_result: object) -> bool:
required = (
"operator_result_id",
"attempt_id",
"attempt_index",
"candidate_digest",
"candidate_reconstruction_digest",
"geometric_search_run_id",
"evidence_spans",
"candidate_attempt",
"candidate_reconstruction",
)
return all(hasattr(operator_result, name) for name in required)
def _validate_bound_binding_pair(
*,
run: GeometricSearchRun,
binding: object,
operator_result: object,
) -> list[str]:
reasons: list[str] = []
attempt = operator_result.candidate_attempt
reconstruction = operator_result.candidate_reconstruction
if binding.original_run_id != run.run_id:
reasons.append("binding_run_mismatch")
if binding.operator_result_id != operator_result.operator_result_id:
reasons.append("binding_result_mismatch")
if binding.candidate_attempt_id != operator_result.attempt_id:
reasons.append("binding_attempt_mismatch")
if binding.candidate_attempt_id != attempt.attempt_id:
reasons.append("binding_attempt_mismatch")
if binding.attempt_index != operator_result.attempt_index:
reasons.append("binding_attempt_mismatch")
if binding.attempt_index != attempt.attempt_index:
reasons.append("binding_attempt_mismatch")
if binding.candidate_digest != operator_result.candidate_digest:
reasons.append("binding_reconstruction_mismatch")
if binding.candidate_digest != attempt.candidate_digest:
reasons.append("binding_reconstruction_mismatch")
if (
binding.candidate_reconstruction_digest
!= operator_result.candidate_reconstruction_digest
):
reasons.append("binding_reconstruction_mismatch")
if (
binding.candidate_reconstruction_digest
!= reconstruction.candidate_reconstruction_digest
):
reasons.append("binding_reconstruction_mismatch")
if binding.run_attempt_membership != "structurally_bound":
reasons.append("binding_not_structurally_bound")
if binding.reason_codes != ():
reasons.append("binding_not_successful")
if binding.evidence_spans != operator_result.evidence_spans:
reasons.append("binding_evidence_mismatch")
if binding.evidence_spans != attempt.evidence_spans:
reasons.append("binding_evidence_mismatch")
if binding.evidence_spans != reconstruction.evidence_spans:
reasons.append("binding_evidence_mismatch")
if operator_result.geometric_search_run_id != run.run_id:
reasons.append("operator_result_run_mismatch")
if reconstruction.problem_frame_digest != run.problem_frame_digest:
reasons.append("operator_reconstruction_run_mismatch")
if reconstruction.original_contract_assessment_id != run.contract_assessment_id:
reasons.append("operator_reconstruction_run_mismatch")
if reconstruction.source_residual_id not in run.residual_ids:
reasons.append("operator_reconstruction_run_mismatch")
return reasons
def _validate_bound_replay_bindings(
*,
run: GeometricSearchRun,
bindings: tuple[object, ...],
replay_results: tuple[ReplayAdapterResult, ...],
replay_refusals: tuple[ReplayAdapterRefusal, ...],
) -> list[str]:
reasons: list[str] = []
attempt_ids = {binding.candidate_attempt_id for binding in bindings} # type: ignore[attr-defined]
digest_by_attempt = {
binding.candidate_attempt_id: binding.candidate_digest # type: ignore[attr-defined]
for binding in bindings
}
for record in (*replay_results, *replay_refusals):
identity_reason = _validate_replay_record_identity(record)
if identity_reason is not None:
reasons.append(identity_reason)
run_id = record.run_id
attempt_id = record.attempt_id
candidate_digest = record.candidate_digest
if run_id != run.run_id:
reasons.append("replay_run_id_mismatch")
if attempt_id is None or not _nonempty_text(attempt_id):
reasons.append("replay_orphan_attempt_id")
elif attempt_id not in attempt_ids:
reasons.append("replay_orphan_attempt_id")
if candidate_digest is not None and _nonempty_text(candidate_digest):
if (
attempt_id in digest_by_attempt
and candidate_digest != digest_by_attempt[attempt_id]
):
reasons.append("replay_candidate_digest_mismatch")
elif attempt_id in digest_by_attempt:
reasons.append("replay_candidate_digest_mismatch")
return reasons
def _derive_bound_disposition(
*,
replay_results: tuple[ReplayAdapterResult, ...],
replay_refusals: tuple[ReplayAdapterRefusal, ...],
) -> PracticeDisposition | None:
if not replay_results and not replay_refusals:
return None
if any(
result.replay_disposition is ReplayDisposition.CONTRACT_AND_PROOF_CLOSED
for result in replay_results
):
return PracticeDisposition.SEALED_CANDIDATE_REPLAY_CLOSED
if any(
result.replay_disposition
is ReplayDisposition.CONTRACT_CLOSED_BUT_PROOF_REFUSED
for result in replay_results
):
return PracticeDisposition.SEALED_CONTRACT_CLOSED_PROOF_REFUSED
if replay_results and all(
result.replay_disposition is ReplayDisposition.CONTRACT_REFUSED
for result in replay_results
):
return PracticeDisposition.SEALED_ALL_CANDIDATES_REFUSED
if replay_refusals and not replay_results:
return PracticeDisposition.SEALED_REPLAY_UNAVAILABLE
return None
def build_bound_practice_trace_input(
*,
problem_frame_digest: str,
original_contract_assessment_id: str,
residual_ids: tuple[str, ...],
search_gate_decision_id: str,
compute_budget_id: str,
run: GeometricSearchRun,
bindings: tuple[object, ...],
candidate_operator_results: tuple[object, ...],
replay_results: tuple[ReplayAdapterResult, ...] = (),
replay_refusals: tuple[ReplayAdapterRefusal, ...] = (),
schema_versions: tuple[tuple[str, str], ...] = (),
policy_versions: tuple[tuple[str, str], ...] = (),
trace_policy_version: str = SEALED_PRACTICE_TRACE_POLICY_VERSION,
) -> PracticeTraceInput | PracticeTraceRefusal:
"""Validate bound episode identities and emit a trace input or refusal."""
if trace_policy_version != SEALED_PRACTICE_TRACE_POLICY_VERSION:
return _refusal(
input_digest=None,
trace_policy_version=trace_policy_version,
disposition=PracticeDisposition.TRACE_POLICY_UNSUPPORTED,
reason_codes=("unsupported_trace_policy_version",),
)
reasons: list[str] = []
if not isinstance(run, GeometricSearchRun):
reasons.append("invalid_run_type")
if not isinstance(bindings, tuple) or not all(
_valid_binding_record(binding) for binding in bindings
):
reasons.append("invalid_binding_type")
if not isinstance(candidate_operator_results, tuple) or not all(
_valid_operator_result_record(result) for result in candidate_operator_results
):
reasons.append("invalid_candidate_operator_result_type")
if not isinstance(replay_results, tuple) or not all(
isinstance(result, ReplayAdapterResult) for result in replay_results
):
reasons.append("invalid_replay_result_type")
if not isinstance(replay_refusals, tuple) or not all(
isinstance(refusal, ReplayAdapterRefusal) for refusal in replay_refusals
):
reasons.append("invalid_replay_refusal_type")
if not _sha256_text(problem_frame_digest):
reasons.append("invalid_problem_frame_digest")
if not _nonempty_text(original_contract_assessment_id):
reasons.append("missing_original_contract_assessment_id")
if not isinstance(residual_ids, tuple) or not all(
_nonempty_text(residual_id) for residual_id in residual_ids
):
reasons.append("invalid_residual_ids")
if not _sha256_text(search_gate_decision_id):
reasons.append("invalid_search_gate_decision_id")
if not _sha256_text(compute_budget_id):
reasons.append("invalid_compute_budget_id")
if not _valid_version_pairs(schema_versions):
reasons.append("invalid_schema_versions")
if not _valid_version_pairs(policy_versions):
reasons.append("invalid_policy_versions")
geometric_search_run_id = run.run_id if isinstance(run, GeometricSearchRun) else ""
if not _sha256_text(geometric_search_run_id):
reasons.append("invalid_geometric_search_run_id")
if len(bindings) != len(candidate_operator_results):
reasons.append("binding_result_count_mismatch")
if not reasons and isinstance(run, GeometricSearchRun):
reasons.extend(
_validate_run_bindings(
run=run,
problem_frame_digest=problem_frame_digest,
original_contract_assessment_id=original_contract_assessment_id,
residual_ids=residual_ids,
search_gate_decision_id=search_gate_decision_id,
compute_budget_id=compute_budget_id,
)
)
binding_ids: list[str] = []
attempt_ids: list[str] = []
candidate_digests: list[str] = []
for binding, operator_result in zip(bindings, candidate_operator_results, strict=True):
if not _sha256_text(binding.binding_id):
reasons.append("invalid_candidate_attempt_binding_id")
if binding.binding_id in binding_ids:
reasons.append("duplicate_candidate_attempt_binding_id")
binding_ids.append(binding.binding_id)
if binding.candidate_attempt_id in attempt_ids:
reasons.append("duplicate_candidate_attempt_id")
attempt_ids.append(binding.candidate_attempt_id)
if binding.candidate_digest in candidate_digests:
reasons.append("duplicate_candidate_digest")
candidate_digests.append(binding.candidate_digest)
reasons.extend(
_validate_bound_binding_pair(
run=run,
binding=binding,
operator_result=operator_result,
)
)
reasons.extend(
_validate_bound_replay_bindings(
run=run,
bindings=bindings,
replay_results=replay_results,
replay_refusals=replay_refusals,
)
)
candidate_attempt_ids = tuple(binding.candidate_attempt_id for binding in bindings)
candidate_attempt_binding_ids = tuple(binding.binding_id for binding in bindings)
replay_result_ids = tuple(result.replay_result_id for result in replay_results)
replay_refusal_ids = tuple(refusal.replay_refusal_id for refusal in replay_refusals)
if reasons:
return _refusal(
input_digest=None,
disposition=_failure_disposition(reasons),
reason_codes=tuple(dict.fromkeys(reasons)),
)
input_digest = _canonical_digest(
_input_digest_payload(
trace_policy_version=trace_policy_version,
problem_frame_digest=problem_frame_digest,
original_contract_assessment_id=original_contract_assessment_id,
residual_ids=residual_ids,
search_gate_decision_id=search_gate_decision_id,
compute_budget_id=compute_budget_id,
geometric_search_run_id=geometric_search_run_id,
candidate_attempt_ids=candidate_attempt_ids,
candidate_attempt_binding_ids=candidate_attempt_binding_ids,
replay_result_ids=replay_result_ids,
replay_refusal_ids=replay_refusal_ids,
schema_versions=schema_versions,
policy_versions=policy_versions,
)
)
return PracticeTraceInput(
input_digest=input_digest,
trace_policy_version=trace_policy_version,
problem_frame_digest=problem_frame_digest,
original_contract_assessment_id=original_contract_assessment_id,
residual_ids=residual_ids,
search_gate_decision_id=search_gate_decision_id,
compute_budget_id=compute_budget_id,
geometric_search_run_id=geometric_search_run_id,
candidate_attempt_ids=candidate_attempt_ids,
candidate_attempt_binding_ids=candidate_attempt_binding_ids,
replay_result_ids=replay_result_ids,
replay_refusal_ids=replay_refusal_ids,
schema_versions=schema_versions,
policy_versions=policy_versions,
)
def seal_bound_practice_trace(
trace_input: PracticeTraceInput,
*,
run: GeometricSearchRun,
bindings: tuple[object, ...],
candidate_operator_results: tuple[object, ...],
replay_results: tuple[ReplayAdapterResult, ...] = (),
replay_refusals: tuple[ReplayAdapterRefusal, ...] = (),
evidence_spans: tuple[SourceSpan, ...] = (),
explanation: str = "",
) -> SealedPracticeTrace | PracticeTraceRefusal:
"""Seal a validated bound practice episode as immutable diagnostic evidence."""
if not isinstance(trace_input, PracticeTraceInput):
return _refusal(
input_digest=None,
disposition=PracticeDisposition.TRACE_INVALID_INPUT,
reason_codes=("invalid_trace_input_type",),
)
if trace_input.trace_policy_version != SEALED_PRACTICE_TRACE_POLICY_VERSION:
return _refusal(
input_digest=trace_input.input_digest,
trace_policy_version=trace_input.trace_policy_version,
disposition=PracticeDisposition.TRACE_POLICY_UNSUPPORTED,
reason_codes=("unsupported_trace_policy_version",),
)
if not _valid_spans(evidence_spans):
return _refusal(
input_digest=trace_input.input_digest,
disposition=PracticeDisposition.TRACE_INVALID_INPUT,
reason_codes=("invalid_evidence_spans",),
)
reasons: list[str] = []
if not isinstance(run, GeometricSearchRun):
reasons.append("invalid_run_type")
if not isinstance(bindings, tuple) or not all(
_valid_binding_record(binding) for binding in bindings
):
reasons.append("invalid_binding_type")
if not isinstance(candidate_operator_results, tuple) or not all(
_valid_operator_result_record(result) for result in candidate_operator_results
):
reasons.append("invalid_candidate_operator_result_type")
if run.run_id != trace_input.geometric_search_run_id:
reasons.append("geometric_search_run_id_mismatch")
replay_result_ids = tuple(result.replay_result_id for result in replay_results)
replay_refusal_ids = tuple(refusal.replay_refusal_id for refusal in replay_refusals)
if replay_result_ids != trace_input.replay_result_ids:
reasons.append("replay_result_ids_mismatch")
if replay_refusal_ids != trace_input.replay_refusal_ids:
reasons.append("replay_refusal_ids_mismatch")
binding_ids = tuple(binding.binding_id for binding in bindings)
attempt_ids = tuple(binding.candidate_attempt_id for binding in bindings)
if binding_ids != trace_input.candidate_attempt_binding_ids:
reasons.append("candidate_attempt_binding_ids_mismatch")
if attempt_ids != trace_input.candidate_attempt_ids:
reasons.append("candidate_attempt_ids_mismatch")
if not reasons:
reasons.extend(
_validate_run_bindings(
run=run,
problem_frame_digest=trace_input.problem_frame_digest,
original_contract_assessment_id=trace_input.original_contract_assessment_id,
residual_ids=trace_input.residual_ids,
search_gate_decision_id=trace_input.search_gate_decision_id,
compute_budget_id=trace_input.compute_budget_id,
)
)
if len(bindings) != len(candidate_operator_results):
reasons.append("binding_result_count_mismatch")
for binding, operator_result in zip(bindings, candidate_operator_results, strict=True):
reasons.extend(
_validate_bound_binding_pair(
run=run,
binding=binding,
operator_result=operator_result,
)
)
reasons.extend(
_validate_bound_replay_bindings(
run=run,
bindings=bindings,
replay_results=replay_results,
replay_refusals=replay_refusals,
)
)
if bindings and not replay_results and not replay_refusals:
reasons.append("missing_replay_records")
if reasons:
return _refusal(
input_digest=trace_input.input_digest,
disposition=_failure_disposition(reasons),
reason_codes=tuple(dict.fromkeys(reasons)),
)
recomputed_input_digest = _canonical_digest(
_input_digest_payload(
trace_policy_version=trace_input.trace_policy_version,
problem_frame_digest=trace_input.problem_frame_digest,
original_contract_assessment_id=trace_input.original_contract_assessment_id,
residual_ids=trace_input.residual_ids,
search_gate_decision_id=trace_input.search_gate_decision_id,
compute_budget_id=trace_input.compute_budget_id,
geometric_search_run_id=trace_input.geometric_search_run_id,
candidate_attempt_ids=trace_input.candidate_attempt_ids,
candidate_attempt_binding_ids=trace_input.candidate_attempt_binding_ids,
replay_result_ids=trace_input.replay_result_ids,
replay_refusal_ids=trace_input.replay_refusal_ids,
schema_versions=trace_input.schema_versions,
policy_versions=trace_input.policy_versions,
)
)
if recomputed_input_digest != trace_input.input_digest:
return _refusal(
input_digest=trace_input.input_digest,
disposition=PracticeDisposition.TRACE_IDENTITY_MISMATCH,
reason_codes=("input_digest_mismatch",),
)
practice_disposition = _derive_bound_disposition(
replay_results=replay_results,
replay_refusals=replay_refusals,
)
if practice_disposition is None or practice_disposition not in _SEALED_DISPOSITIONS:
return _refusal(
input_digest=trace_input.input_digest,
disposition=PracticeDisposition.TRACE_UPSTREAM_INCOMPLETE,
reason_codes=("unsupported_bound_practice_disposition",),
)
identity_chain = _upstream_identity_chain(
problem_frame_digest=trace_input.problem_frame_digest,
original_contract_assessment_id=trace_input.original_contract_assessment_id,
residual_ids=trace_input.residual_ids,
search_gate_decision_id=trace_input.search_gate_decision_id,
compute_budget_id=trace_input.compute_budget_id,
geometric_search_run_id=trace_input.geometric_search_run_id,
candidate_attempt_ids=trace_input.candidate_attempt_ids,
candidate_attempt_binding_ids=trace_input.candidate_attempt_binding_ids,
replay_result_ids=trace_input.replay_result_ids,
replay_refusal_ids=trace_input.replay_refusal_ids,
)
trace_records = identity_chain
trace_id = _canonical_digest(
_trace_id_payload(
trace_policy_version=trace_input.trace_policy_version,
input_digest=trace_input.input_digest,
problem_frame_digest=trace_input.problem_frame_digest,
original_contract_assessment_id=trace_input.original_contract_assessment_id,
residual_ids=trace_input.residual_ids,
search_gate_decision_id=trace_input.search_gate_decision_id,
compute_budget_id=trace_input.compute_budget_id,
geometric_search_run_id=trace_input.geometric_search_run_id,
candidate_attempt_ids=trace_input.candidate_attempt_ids,
candidate_attempt_binding_ids=trace_input.candidate_attempt_binding_ids,
replay_result_ids=trace_input.replay_result_ids,
replay_refusal_ids=trace_input.replay_refusal_ids,
upstream_identity_chain=identity_chain,
practice_disposition=practice_disposition,
trace_records=trace_records,
evidence_spans=evidence_spans,
created_by_policy=CREATED_BY_POLICY,
)
)
safe_explanation = explanation or (
"Sealed bound practice trace disposition: " + practice_disposition.value + "."
)
return SealedPracticeTrace(
trace_id=trace_id,
trace_policy_version=trace_input.trace_policy_version,
input_digest=trace_input.input_digest,
problem_frame_digest=trace_input.problem_frame_digest,
original_contract_assessment_id=trace_input.original_contract_assessment_id,
residual_ids=trace_input.residual_ids,
search_gate_decision_id=trace_input.search_gate_decision_id,
compute_budget_id=trace_input.compute_budget_id,
geometric_search_run_id=trace_input.geometric_search_run_id,
candidate_attempt_ids=trace_input.candidate_attempt_ids,
candidate_attempt_binding_ids=trace_input.candidate_attempt_binding_ids,
replay_result_ids=trace_input.replay_result_ids,
replay_refusal_ids=trace_input.replay_refusal_ids,
upstream_identity_chain=identity_chain,
@ -761,4 +1309,6 @@ __all__ = [
"PracticeTraceOutcome",
"build_practice_trace_input",
"seal_practice_trace",
"build_bound_practice_trace_input",
"seal_bound_practice_trace",
]

12
tests/test_sealed_practice_trace.py
View file

@ -404,6 +404,7 @@ def _expected_input_digest(
"compute_budget_id": compute_budget_id,
"geometric_search_run_id": geometric_search_run_id,
"candidate_attempt_ids": list(candidate_attempt_ids),
"candidate_attempt_binding_ids": [],
"replay_result_ids": list(replay_result_ids),
"replay_refusal_ids": list(replay_refusal_ids),
"schema_versions": [[name, version] for name, version in schema_versions],
@ -426,6 +427,7 @@ def _expected_trace_id(
trace_input.compute_budget_id,
trace_input.geometric_search_run_id,
*trace_input.candidate_attempt_ids,
*trace_input.candidate_attempt_binding_ids,
*trace_input.replay_result_ids,
*trace_input.replay_refusal_ids,
)
@ -441,6 +443,7 @@ def _expected_trace_id(
"compute_budget_id": trace_input.compute_budget_id,
"geometric_search_run_id": trace_input.geometric_search_run_id,
"candidate_attempt_ids": list(trace_input.candidate_attempt_ids),
"candidate_attempt_binding_ids": list(trace_input.candidate_attempt_binding_ids),
"replay_result_ids": list(trace_input.replay_result_ids),
"replay_refusal_ids": list(trace_input.replay_refusal_ids),
"upstream_identity_chain": list(identity_chain),
@ -481,6 +484,8 @@ def test_public_api_exports_are_exact() -> None:
"PracticeTraceOutcome",
"build_practice_trace_input",
"seal_practice_trace",
"build_bound_practice_trace_input",
"seal_bound_practice_trace",
)
assert SEALED_PRACTICE_TRACE_POLICY_VERSION == "sealed_practice_trace.v1"
assert tuple(sealed_trace.__all__) == expected
@ -1036,7 +1041,10 @@ def test_module_coupling_and_side_effect_guards() -> None:
"project_contract_residuals",
"determine",
"build_replay_adapter_input",
"build_replay_adapter_input_from_binding",
"classify_replay_result",
"build_missing_role_candidate",
"bind_candidate_attempt_to_run",
}.isdisjoint(imported_names)
assert {
"decide_search_gate",
@ -1045,6 +1053,10 @@ def test_module_coupling_and_side_effect_guards() -> None:
"assess_contracts",
"project_contract_residuals",
"determine",
"build_missing_role_candidate",
"bind_candidate_attempt_to_run",
"build_replay_adapter_input_from_binding",
"classify_replay_result",
"repair",
"serve",
"store",

703
tests/test_sealed_practice_trace_bound_episode.py Normal file
View file

@ -0,0 +1,703 @@
from __future__ import annotations
import ast
import dataclasses
import hashlib
import json
from dataclasses import replace
from pathlib import Path
from generate.candidate_operator import (
CandidateOperatorResult,
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 (
VACUOUS_PROOF_DECLARATION,
ReplayAdapterInput,
ReplayAdapterRefusal,
ReplayAdapterResult,
ReplayDisposition,
build_replay_adapter_input_from_binding,
classify_replay_result,
)
from generate.run_attempt_binding import (
CandidateAttemptRunBinding,
bind_candidate_attempt_to_run,
)
from generate.search_gate import SearchGateDecision, SearchGateStatus
from generate.sealed_practice_trace import (
PracticeDisposition,
PracticeTraceInput,
PracticeTraceRefusal,
SealedPracticeTrace,
build_bound_practice_trace_input,
seal_bound_practice_trace,
)
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,
CandidateOperatorResult,
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 isinstance(result, CandidateOperatorResult)
binding = bind_candidate_attempt_to_run(
original_run=run,
candidate_operator_result=result,
)
assert isinstance(binding, CandidateAttemptRunBinding)
return run, result, binding
def _bound_replay_input(
run: GeometricSearchRun,
result: CandidateOperatorResult,
binding: CandidateAttemptRunBinding,
**kwargs: object,
) -> ReplayAdapterInput | ReplayAdapterRefusal:
outcome = build_replay_adapter_input_from_binding(
run=run,
binding=binding,
candidate_operator_result=result,
**kwargs, # type: ignore[arg-type]
)
assert isinstance(outcome, ReplayAdapterInput)
return outcome
def _bound_replay_result(
run: GeometricSearchRun,
result: CandidateOperatorResult,
binding: CandidateAttemptRunBinding,
*,
disposition: ReplayDisposition = ReplayDisposition.CONTRACT_REFUSED,
schema_versions: tuple[tuple[str, str], ...] = (),
proof_obligation_refs: tuple[str, ...] = (),
) -> ReplayAdapterResult:
if disposition is ReplayDisposition.CONTRACT_AND_PROOF_CLOSED:
schema_versions = schema_versions or (VACUOUS_PROOF_DECLARATION,)
if disposition is ReplayDisposition.CONTRACT_CLOSED_BUT_PROOF_REFUSED:
proof_obligation_refs = proof_obligation_refs or ("obligation-a",)
replay_input = _bound_replay_input(
run,
result,
binding,
schema_versions=schema_versions,
proof_obligation_refs=proof_obligation_refs,
)
if disposition is ReplayDisposition.CONTRACT_REFUSED:
outcome = classify_replay_result(
replay_input,
contract_replay_assessment_id="e" * 64,
contract_closed=False,
)
else:
outcome = classify_replay_result(
replay_input,
contract_replay_assessment_id="e" * 64,
contract_closed=True,
)
assert isinstance(outcome, ReplayAdapterResult)
assert outcome.replay_disposition is disposition
return outcome
def _bound_replay_refusal(
run: GeometricSearchRun,
result: CandidateOperatorResult,
binding: CandidateAttemptRunBinding,
) -> ReplayAdapterRefusal:
replay_input = _bound_replay_input(run, result, binding)
outcome = classify_replay_result(
replay_input,
contract_replay_assessment_id=None,
contract_closed=True,
)
assert isinstance(outcome, ReplayAdapterRefusal)
return outcome
def _build_bound_trace_input(
run: GeometricSearchRun,
result: CandidateOperatorResult,
binding: CandidateAttemptRunBinding,
*,
replay_results: tuple[ReplayAdapterResult, ...] = (),
replay_refusals: tuple[ReplayAdapterRefusal, ...] = (),
**changes: object,
) -> PracticeTraceInput | PracticeTraceRefusal:
values: dict[str, object] = {
"problem_frame_digest": run.problem_frame_digest,
"original_contract_assessment_id": run.contract_assessment_id,
"residual_ids": run.residual_ids,
"search_gate_decision_id": run.gate_decision_id,
"compute_budget_id": run.budget_id,
"run": run,
"bindings": (binding,),
"candidate_operator_results": (result,),
"replay_results": replay_results,
"replay_refusals": replay_refusals,
}
values.update(changes)
return build_bound_practice_trace_input(**values) # type: ignore[arg-type]
def _seal_bound(
trace_input: PracticeTraceInput,
run: GeometricSearchRun,
result: CandidateOperatorResult,
binding: CandidateAttemptRunBinding,
*,
replay_results: tuple[ReplayAdapterResult, ...] = (),
replay_refusals: tuple[ReplayAdapterRefusal, ...] = (),
evidence_spans: tuple[SourceSpan, ...] = (),
explanation: str = "",
) -> SealedPracticeTrace | PracticeTraceRefusal:
return seal_bound_practice_trace(
trace_input,
run=run,
bindings=(binding,),
candidate_operator_results=(result,),
replay_results=replay_results,
replay_refusals=replay_refusals,
evidence_spans=evidence_spans,
explanation=explanation,
)
def test_bound_contract_refused_seals_all_candidates_refused() -> None:
run, result, binding = _chain()
replay = _bound_replay_result(run, result, binding)
trace_input = _build_bound_trace_input(run, result, binding, replay_results=(replay,))
assert isinstance(trace_input, PracticeTraceInput)
sealed = _seal_bound(trace_input, run, result, binding, replay_results=(replay,))
assert isinstance(sealed, SealedPracticeTrace)
assert sealed.practice_disposition is PracticeDisposition.SEALED_ALL_CANDIDATES_REFUSED
def test_bound_vacuous_proof_closed_seals_candidate_replay_closed() -> None:
run, result, binding = _chain()
replay = _bound_replay_result(
run,
result,
binding,
disposition=ReplayDisposition.CONTRACT_AND_PROOF_CLOSED,
)
trace_input = _build_bound_trace_input(run, result, binding, replay_results=(replay,))
assert isinstance(trace_input, PracticeTraceInput)
sealed = _seal_bound(trace_input, run, result, binding, replay_results=(replay,))
assert isinstance(sealed, SealedPracticeTrace)
assert sealed.practice_disposition is PracticeDisposition.SEALED_CANDIDATE_REPLAY_CLOSED
def test_bound_contract_closed_proof_refused_seals_disposition() -> None:
run, result, binding = _chain()
replay = _bound_replay_result(
run,
result,
binding,
disposition=ReplayDisposition.CONTRACT_CLOSED_BUT_PROOF_REFUSED,
)
trace_input = _build_bound_trace_input(run, result, binding, replay_results=(replay,))
assert isinstance(trace_input, PracticeTraceInput)
sealed = _seal_bound(trace_input, run, result, binding, replay_results=(replay,))
assert isinstance(sealed, SealedPracticeTrace)
assert (
sealed.practice_disposition
is PracticeDisposition.SEALED_CONTRACT_CLOSED_PROOF_REFUSED
)
def test_bound_replay_refusal_only_seals_replay_unavailable() -> None:
run, result, binding = _chain()
refusal = _bound_replay_refusal(run, result, binding)
trace_input = _build_bound_trace_input(run, result, binding, replay_refusals=(refusal,))
assert isinstance(trace_input, PracticeTraceInput)
sealed = _seal_bound(trace_input, run, result, binding, replay_refusals=(refusal,))
assert isinstance(sealed, SealedPracticeTrace)
assert sealed.practice_disposition is PracticeDisposition.SEALED_REPLAY_UNAVAILABLE
def test_run_remains_unmutated_after_bound_sealing() -> None:
run, result, binding = _chain()
before = run.candidate_attempts
replay = _bound_replay_result(run, result, binding)
trace_input = _build_bound_trace_input(run, result, binding, replay_results=(replay,))
assert isinstance(trace_input, PracticeTraceInput)
sealed = _seal_bound(trace_input, run, result, binding, replay_results=(replay,))
assert isinstance(sealed, SealedPracticeTrace)
assert run.candidate_attempts == before == ()
def test_trace_includes_bound_identity_fields() -> None:
run, result, binding = _chain()
replay = _bound_replay_result(run, result, binding)
trace_input = _build_bound_trace_input(run, result, binding, replay_results=(replay,))
assert isinstance(trace_input, PracticeTraceInput)
sealed = _seal_bound(trace_input, run, result, binding, replay_results=(replay,))
assert isinstance(sealed, SealedPracticeTrace)
assert sealed.candidate_attempt_ids == (binding.candidate_attempt_id,)
assert sealed.candidate_attempt_binding_ids == (binding.binding_id,)
assert sealed.replay_result_ids == (replay.replay_result_id,)
assert sealed.replay_refusal_ids == ()
def test_bound_trace_id_is_deterministic() -> None:
run, result, binding = _chain()
replay = _bound_replay_result(run, result, binding)
trace_input = _build_bound_trace_input(run, result, binding, replay_results=(replay,))
assert isinstance(trace_input, PracticeTraceInput)
first = _seal_bound(trace_input, run, result, binding, replay_results=(replay,))
second = _seal_bound(trace_input, run, result, binding, replay_results=(replay,))
assert isinstance(first, SealedPracticeTrace)
assert isinstance(second, SealedPracticeTrace)
assert first.trace_id == second.trace_id
def test_bound_explanation_changes_do_not_affect_trace_id() -> None:
run, result, binding = _chain()
replay = _bound_replay_result(run, result, binding)
trace_input = _build_bound_trace_input(run, result, binding, replay_results=(replay,))
assert isinstance(trace_input, PracticeTraceInput)
first = _seal_bound(
trace_input,
run,
result,
binding,
replay_results=(replay,),
explanation="first prose",
)
second = _seal_bound(
trace_input,
run,
result,
binding,
replay_results=(replay,),
explanation="second prose",
)
assert isinstance(first, SealedPracticeTrace)
assert isinstance(second, SealedPracticeTrace)
assert first.trace_id == second.trace_id
def test_bound_duplicate_evidence_spans_are_preserved() -> None:
run, result, binding = _chain()
replay = _bound_replay_result(run, result, binding)
trace_input = _build_bound_trace_input(run, result, binding, replay_results=(replay,))
assert isinstance(trace_input, PracticeTraceInput)
span = SourceSpan("dup", 1, 4, 0)
sealed = _seal_bound(
trace_input,
run,
result,
binding,
replay_results=(replay,),
evidence_spans=(span, span),
)
assert isinstance(sealed, SealedPracticeTrace)
assert sealed.evidence_spans == (span, span)
def test_binding_run_mismatch_refuses() -> None:
run, result, binding = _chain()
replay = _bound_replay_result(run, result, binding)
bad_binding = replace(binding, original_run_id="wrong" * 8)
outcome = _build_bound_trace_input(
run,
result,
bad_binding,
replay_results=(replay,),
)
assert isinstance(outcome, PracticeTraceRefusal)
assert "binding_run_mismatch" in outcome.reason_codes
def test_binding_result_mismatch_refuses() -> None:
run, result, binding = _chain()
replay = _bound_replay_result(run, result, binding)
bad_binding = replace(binding, operator_result_id="wrong" * 8)
outcome = _build_bound_trace_input(
run,
result,
bad_binding,
replay_results=(replay,),
)
assert isinstance(outcome, PracticeTraceRefusal)
assert "binding_result_mismatch" in outcome.reason_codes
def test_binding_attempt_mismatch_refuses() -> None:
run, result, binding = _chain()
replay = _bound_replay_result(run, result, binding)
bad_binding = replace(binding, candidate_attempt_id="wrong" * 8)
outcome = _build_bound_trace_input(
run,
result,
bad_binding,
replay_results=(replay,),
)
assert isinstance(outcome, PracticeTraceRefusal)
assert "binding_attempt_mismatch" in outcome.reason_codes
def test_binding_reconstruction_mismatch_refuses() -> None:
run, result, binding = _chain()
replay = _bound_replay_result(run, result, binding)
bad_binding = replace(binding, candidate_reconstruction_digest="wrong" * 8)
outcome = _build_bound_trace_input(
run,
result,
bad_binding,
replay_results=(replay,),
)
assert isinstance(outcome, PracticeTraceRefusal)
assert "binding_reconstruction_mismatch" in outcome.reason_codes
def test_binding_not_structurally_bound_refuses() -> None:
run, result, binding = _chain()
replay = _bound_replay_result(run, result, binding)
bad_binding = replace(binding, run_attempt_membership="pending")
outcome = _build_bound_trace_input(
run,
result,
bad_binding,
replay_results=(replay,),
)
assert isinstance(outcome, PracticeTraceRefusal)
assert "binding_not_structurally_bound" in outcome.reason_codes
def test_binding_with_reason_codes_refuses() -> None:
run, result, binding = _chain()
replay = _bound_replay_result(run, result, binding)
bad_binding = replace(binding, reason_codes=("leftover",))
outcome = _build_bound_trace_input(
run,
result,
bad_binding,
replay_results=(replay,),
)
assert isinstance(outcome, PracticeTraceRefusal)
assert "binding_not_successful" in outcome.reason_codes
def test_duplicate_binding_ids_refuse() -> None:
run, result, binding = _chain()
replay = _bound_replay_result(run, result, binding)
outcome = build_bound_practice_trace_input(
problem_frame_digest=run.problem_frame_digest,
original_contract_assessment_id=run.contract_assessment_id,
residual_ids=run.residual_ids,
search_gate_decision_id=run.gate_decision_id,
compute_budget_id=run.budget_id,
run=run,
bindings=(binding, binding),
candidate_operator_results=(result, result),
replay_results=(replay, replay),
)
assert isinstance(outcome, PracticeTraceRefusal)
assert "duplicate_candidate_attempt_binding_id" in outcome.reason_codes
def test_duplicate_attempt_ids_refuse() -> None:
run, result, binding = _chain()
replay = _bound_replay_result(run, result, binding)
second_binding = replace(binding, binding_id="b" * 64)
outcome = build_bound_practice_trace_input(
problem_frame_digest=run.problem_frame_digest,
original_contract_assessment_id=run.contract_assessment_id,
residual_ids=run.residual_ids,
search_gate_decision_id=run.gate_decision_id,
compute_budget_id=run.budget_id,
run=run,
bindings=(binding, second_binding),
candidate_operator_results=(result, result),
replay_results=(replay,),
)
assert isinstance(outcome, PracticeTraceRefusal)
assert "duplicate_candidate_attempt_id" in outcome.reason_codes
def test_duplicate_candidate_digests_refuse() -> None:
run, result, binding = _chain()
replay = _bound_replay_result(run, result, binding)
second_binding = replace(
binding,
binding_id="b" * 64,
candidate_attempt_id="c" * 64,
)
outcome = build_bound_practice_trace_input(
problem_frame_digest=run.problem_frame_digest,
original_contract_assessment_id=run.contract_assessment_id,
residual_ids=run.residual_ids,
search_gate_decision_id=run.gate_decision_id,
compute_budget_id=run.budget_id,
run=run,
bindings=(binding, second_binding),
candidate_operator_results=(result, result),
replay_results=(replay,),
)
assert isinstance(outcome, PracticeTraceRefusal)
assert "duplicate_candidate_digest" in outcome.reason_codes
def test_replay_result_wrong_run_id_refuses() -> None:
run, result, binding = _chain()
replay = _bound_replay_result(run, result, binding)
bad_replay = replace(replay, run_id="wrong" * 8)
outcome = _build_bound_trace_input(run, result, binding, replay_results=(bad_replay,))
assert isinstance(outcome, PracticeTraceRefusal)
assert "replay_run_id_mismatch" in outcome.reason_codes
def test_replay_result_orphan_attempt_id_refuses() -> None:
run, result, binding = _chain()
replay = _bound_replay_result(run, result, binding)
bad_replay = replace(replay, attempt_id="orphan" * 8)
outcome = _build_bound_trace_input(run, result, binding, replay_results=(bad_replay,))
assert isinstance(outcome, PracticeTraceRefusal)
assert "replay_orphan_attempt_id" in outcome.reason_codes
def test_replay_result_wrong_candidate_digest_refuses() -> None:
run, result, binding = _chain()
replay = _bound_replay_result(run, result, binding)
bad_replay = replace(replay, candidate_digest="wrong" * 8)
outcome = _build_bound_trace_input(run, result, binding, replay_results=(bad_replay,))
assert isinstance(outcome, PracticeTraceRefusal)
assert "replay_candidate_digest_mismatch" in outcome.reason_codes
def test_replay_refusal_wrong_run_id_refuses() -> None:
run, result, binding = _chain()
refusal = _bound_replay_refusal(run, result, binding)
bad_refusal = replace(refusal, run_id="wrong" * 8)
outcome = _build_bound_trace_input(run, result, binding, replay_refusals=(bad_refusal,))
assert isinstance(outcome, PracticeTraceRefusal)
assert "replay_run_id_mismatch" in outcome.reason_codes
def test_replay_refusal_orphan_attempt_id_refuses() -> None:
run, result, binding = _chain()
refusal = _bound_replay_refusal(run, result, binding)
bad_refusal = replace(refusal, attempt_id="orphan" * 8)
outcome = _build_bound_trace_input(run, result, binding, replay_refusals=(bad_refusal,))
assert isinstance(outcome, PracticeTraceRefusal)
assert "replay_orphan_attempt_id" in outcome.reason_codes
def test_unsupported_trace_policy_refuses() -> None:
run, result, binding = _chain()
replay = _bound_replay_result(run, result, binding)
outcome = _build_bound_trace_input(
run,
result,
binding,
replay_results=(replay,),
trace_policy_version="unsupported.v9",
)
assert isinstance(outcome, PracticeTraceRefusal)
assert outcome.practice_disposition is PracticeDisposition.TRACE_POLICY_UNSUPPORTED
def test_invalid_schema_versions_refuse() -> None:
run, result, binding = _chain()
replay = _bound_replay_result(run, result, binding)
outcome = _build_bound_trace_input(
run,
result,
binding,
replay_results=(replay,),
schema_versions=(("b", "1"), ("a", "1")),
)
assert isinstance(outcome, PracticeTraceRefusal)
assert "invalid_schema_versions" in outcome.reason_codes
def test_bound_module_does_not_call_upstream_producers() -> None:
path = Path("generate/sealed_practice_trace.py")
source = path.read_text("utf-8")
tree = ast.parse(source)
imported_names: set[str] = set()
calls: set[str] = set()
for node in ast.walk(tree):
if isinstance(node, ast.ImportFrom):
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)
forbidden = {
"build_missing_role_candidate",
"bind_candidate_attempt_to_run",
"build_replay_adapter_input_from_binding",
"classify_replay_result",
}
assert forbidden.isdisjoint(imported_names)
assert forbidden.isdisjoint(calls)
def test_bound_dataclasses_have_no_forbidden_authority_fields() -> None:
from generate.sealed_practice_trace import PracticeTraceRefusal, SealedPracticeTrace
forbidden = {
"answer",
"final_answer",
"served_output",
"proof",
"verdict",
"promotion",
"mutation",
"teaching_update",
"pack_update",
"policy_update",
"identity_update",
"workbench_state",
"runtime_effect",
"confidence",
"score",
"rank",
"priority",
"selected",
"selected_candidate",
"best",
"best_candidate",
"serving_allowed",
"runnable",
}
for record_type in (PracticeTraceInput, SealedPracticeTrace, PracticeTraceRefusal):
assert forbidden.isdisjoint(
{field.name for field in dataclasses.fields(record_type)}
)