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

fix(kernel): fail-closed bound trace structural validation

Browse source 
Return PracticeTraceRefusal before dereferencing bindings or operator
results when structural type/shape checks fail. Guard seal_bound zip
against binding_result_count_mismatch. Add five refusal-without-exception
tests for malformed bound inputs.
This commit is contained in:
Shay 2026-06-23 00:11:14 -07:00
parent 484e8ff9fe
commit b9a6eb9676
2 changed files with 132 additions and 16 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

68
generate/sealed_practice_trace.py
View file

@ -798,6 +798,22 @@ def _valid_binding_record(binding: object) -> bool:
return all(hasattr(binding, name) for name in required)
_BOUND_STRUCTURAL_REASON_CODES = frozenset(
{
"invalid_run_type",
"invalid_binding_type",
"invalid_candidate_operator_result_type",
"invalid_replay_result_type",
"invalid_replay_refusal_type",
"binding_result_count_mismatch",
}
)
def _has_bound_structural_failure(reasons: list[str]) -> bool:
return any(reason in _BOUND_STRUCTURAL_REASON_CODES for reason in reasons)
def _valid_operator_result_record(operator_result: object) -> bool:
required = (
"operator_result_id",
@ -1009,8 +1025,16 @@ def build_bound_practice_trace_input(
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 isinstance(bindings, tuple) and isinstance(candidate_operator_results, tuple):
if len(bindings) != len(candidate_operator_results):
reasons.append("binding_result_count_mismatch")
if _has_bound_structural_failure(reasons):
return _refusal(
input_digest=None,
disposition=_failure_disposition(reasons),
reason_codes=tuple(dict.fromkeys(reasons)),
)
if not reasons and isinstance(run, GeometricSearchRun):
reasons.extend(
@ -1158,6 +1182,13 @@ def seal_bound_practice_trace(
if replay_refusal_ids != trace_input.replay_refusal_ids:
reasons.append("replay_refusal_ids_mismatch")
if _has_bound_structural_failure(reasons):
return _refusal(
input_digest=trace_input.input_digest,
disposition=_failure_disposition(reasons),
reason_codes=tuple(dict.fromkeys(reasons)),
)
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:
@ -1178,24 +1209,29 @@ def seal_bound_practice_trace(
)
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):
elif not reasons:
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_binding_pair(
_validate_bound_replay_bindings(
run=run,
binding=binding,
operator_result=operator_result,
bindings=bindings,
replay_results=replay_results,
replay_refusals=replay_refusals,
)
)
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 bindings and not replay_results and not replay_refusals:
reasons.append("missing_replay_records")
if reasons:
return _refusal(

80
tests/test_sealed_practice_trace_bound_episode.py
View file

@ -643,6 +643,86 @@ def test_invalid_schema_versions_refuse() -> None:
assert "invalid_schema_versions" in outcome.reason_codes
def test_build_bound_refuses_invalid_binding_type_without_exception() -> None:
run, result, _binding = _chain()
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=(object(),),
candidate_operator_results=(result,),
)
assert isinstance(outcome, PracticeTraceRefusal)
assert "invalid_binding_type" in outcome.reason_codes
def test_build_bound_refuses_invalid_operator_result_type_without_exception() -> None:
run, _result, binding = _chain()
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,),
candidate_operator_results=(object(),),
)
assert isinstance(outcome, PracticeTraceRefusal)
assert "invalid_candidate_operator_result_type" in outcome.reason_codes
def test_build_bound_refuses_non_tuple_bindings_without_exception() -> None:
run, result, _binding = _chain()
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=object(), # type: ignore[arg-type]
candidate_operator_results=(result,),
)
assert isinstance(outcome, PracticeTraceRefusal)
assert "invalid_binding_type" in outcome.reason_codes
def test_seal_bound_refuses_binding_result_count_mismatch_without_exception() -> 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)
outcome = seal_bound_practice_trace(
trace_input,
run=run,
bindings=(binding,),
candidate_operator_results=(),
replay_results=(replay,),
)
assert isinstance(outcome, PracticeTraceRefusal)
assert "binding_result_count_mismatch" in outcome.reason_codes
def test_seal_bound_refuses_invalid_binding_type_without_exception() -> 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)
outcome = seal_bound_practice_trace(
trace_input,
run=run,
bindings=(object(),),
candidate_operator_results=(result,),
replay_results=(replay,),
)
assert isinstance(outcome, PracticeTraceRefusal)
assert "invalid_binding_type" 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")