Spaces:
Running
Running
| import pytest | |
| from crosswire.study import RunConfig, run, run_study, summarize | |
| def test_configuration_rejects_nonfinite_unbounded_and_unknown_inputs(): | |
| for values in ({'budget': True}, {'budget': -1}, {'horizon': 7}, {'seed': -1}, | |
| {'policy': 'secret_oracle'}, {'mystery': 2}, {'anchors': 'true'}): | |
| with pytest.raises(ValueError): | |
| RunConfig.from_dict(values) | |
| def test_zero_budget_robust_refuses_semantically_ambiguous_connection(): | |
| result = run(RunConfig(family='symmetric', budget=0)) | |
| assert result['decision']['admitted'] is False | |
| assert result['audit']['contract_satisfied'] is None | |
| assert result['metrics']['target_steps'] == 0 | |
| assert result['audit']['oracle_feasible'] is True | |
| assert result['decision_frozen_hash'] | |
| def test_anchors_and_probes_do_not_drop_true_map_on_supported_case(): | |
| result = run(RunConfig(family='latch', budget=16, anchors=True)) | |
| assert result['audit']['true_map_retained'] is True | |
| assert result['metrics']['target_steps'] <= 16 | |
| assert result['metrics']['target_calls'] == result['metrics']['resets'] + result['metrics']['target_steps'] | |
| if result['decision']['admitted']: | |
| assert result['audit']['contract_satisfied'] is True | |
| def test_related_methods_share_probe_transcript_when_only_admission_changes(): | |
| robust = run(RunConfig(family='latch', budget=8, admission='robust')) | |
| one = run(RunConfig(family='latch', budget=8, admission='one_best')) | |
| assert robust['probes'] == one['probes'] | |
| def test_semantic_corruption_uses_actual_private_events(): | |
| oracle = run(RunConfig(family='semantic_corruption', admission='oracle', budget=0)) | |
| assert oracle['baseline']['privileged'] is True | |
| assert oracle['audit']['contract_satisfied'] is True | |
| assert oracle['audit']['raw_equivalent_maps_count'] > 0 | |
| def test_summary_keeps_abstention_risk_undefined_and_clusters_related_cases(): | |
| row = run(RunConfig(family='symmetric', budget=0)) | |
| summary = summarize([row, row])[0] | |
| assert summary['requests'] == 2 | |
| assert summary['parent_clusters'] == 1 | |
| assert summary['selective_harm'] is None | |
| assert summary['useful_coverage'] == 0 | |
| def test_study_enforces_size_limit_before_work(): | |
| with pytest.raises(ValueError): | |
| run_study({'seeds': list(range(10001))}) | |
| def test_calibration_budget_counts_resets_as_well_as_actions(): | |
| result = run(RunConfig(family='symmetric', policy='fixed', budget=8)) | |
| assert result['metrics']['target_calls'] <= 8 | |
| def test_failed_deployment_never_audits_an_unexecuted_suffix(): | |
| result = run(RunConfig(family='malformed', seed=1, budget=0, admission='identity')) | |
| assert result['deployment']['complete'] is False | |
| assert result['audit']['semantic_events'] == [] | |
| assert result['audit']['hazard'] is None | |
| summary = summarize([result])[0] | |
| assert summary['interface_errors'] == 1 | |
| assert summary['hazardous'] == 0 | |
| def test_forced_fit_exposes_residual_instead_of_implying_consistency(): | |
| result = run(RunConfig(family='incompatible', budget=16, policy='fixed', admission='fit')) | |
| fit = result['baseline']['fit_diagnostics'] | |
| assert fit['chosen_adapter_id'] is not None | |
| assert fit['raw_mismatches'] > 0 | |
| def test_paired_comparison_never_matches_different_budget_conditions(): | |
| from crosswire.study import paired_comparisons | |
| def row(policy, budget, success): | |
| return {'configuration': {'family':'latch','policy':policy,'admission':'robust', | |
| 'seed':17,'budget':budget,'anchors':False,'horizon':4, | |
| 'split':'evaluation','implementation':'procedural'}, | |
| 'parent_instance_id':'p1','audit':{'contract_satisfied':success}} | |
| assert paired_comparisons([row('task',4,True),row('random',8,False)]) == [] | |
| def test_paired_comparison_evaluates_admission_contrasts(): | |
| from crosswire.study import paired_comparisons | |
| def row(admission, success): | |
| return {'configuration': {'family':'latch','policy':'task','admission':admission, | |
| 'seed':17,'budget':4,'anchors':False,'horizon':4, | |
| 'execution_mode':'shielded','temporal':False, | |
| 'split':'evaluation','implementation':'procedural'}, | |
| 'parent_instance_id':'p1','audit':{'contract_satisfied':success}} | |
| pairs = paired_comparisons([row('robust', False), row('adaptive', True)]) | |
| assert len(pairs) == 1 | |
| assert pairs[0]['contrast'] == 'adaptive - robust' | |
| assert pairs[0]['mean_delta'] == 1.0 | |
| def test_unsupported_aalpy_combinations_rejected(): | |
| from crosswire.study import RunConfig | |
| with pytest.raises(ValueError, match='AALpy baseline does not support temporal'): | |
| RunConfig(policy='aalpy', temporal=True).check() | |
| with pytest.raises(ValueError, match='AALpy baseline does not support adaptive'): | |
| RunConfig(policy='aalpy', admission='adaptive', execution_mode='shielded').check() | |
| def _analysis_row(**changes): | |
| from dataclasses import asdict | |
| configuration = asdict(RunConfig()) | |
| configuration.update(changes) | |
| return {'configuration': configuration, 'parent_instance_id': 'parent-latch', | |
| 'decision': {'admitted': True, 'status': 'TASK_ADMISSIBLE'}, | |
| 'audit': {'contract_satisfied': True, 'oracle_feasible': True, | |
| 'hazard': False, 'calibration_hazardous_probes': 0}, | |
| 'metrics': {'target_calls': 0, 'target_steps': 0, 'resets': 0}, | |
| 'deployment': {'target_calls': 0}} | |
| def test_summary_separates_experimental_conditions(field, other): | |
| rows = [_analysis_row(), _analysis_row(**{field: other})] | |
| summary = summarize(rows) | |
| assert len(summary) == 2 | |
| assert all(item['requests'] == 1 for item in summary) | |
| assert {item[field] for item in summary} == {rows[0]['configuration'][field], other} | |
| def test_paired_comparison_never_matches_different_escalation_conditions(): | |
| from crosswire.study import paired_comparisons | |
| rows = [_analysis_row(policy='task', escalate=True), | |
| _analysis_row(policy='random', escalate=False)] | |
| assert paired_comparisons(rows) == [] | |
| def test_paired_comparisons_report_separate_matching_condition_strata(): | |
| from crosswire.study import paired_comparisons | |
| rows = [_analysis_row(policy=policy, escalate=escalate) | |
| for policy in ('task', 'random') for escalate in (False, True)] | |
| pairs = paired_comparisons(rows) | |
| assert len(pairs) == 2 | |
| assert {pair['escalate'] for pair in pairs} == {False, True} | |
| assert all(pair['matched_cells'] == 1 for pair in pairs) | |
| def test_paired_comparisons_reject_duplicate_cells_before_silent_overwrite(): | |
| from crosswire.study import paired_comparisons | |
| first, duplicate = _analysis_row(), _analysis_row() | |
| duplicate['audit']['contract_satisfied'] = False | |
| with pytest.raises(ValueError, match='Duplicate'): | |
| paired_comparisons([first, duplicate, _analysis_row(policy='random')]) | |
| def test_model_escalation_preserves_discarded_class_computation_in_metrics(): | |
| from crosswire.core import CandidateSpace | |
| from crosswire.fixtures import make_case, make_engine | |
| from crosswire.study import _acquire, fit_evidence | |
| case = make_case('delay', seed=17) | |
| def acquire(budget, escalate): | |
| engine, counts = make_engine(case.private), {'target_calls': 0} | |
| def query(word): | |
| engine.reset() | |
| counts['target_calls'] += 1 + len(word) | |
| return [engine.step(action) for action in word] | |
| result = _acquire(case.public, 'task', budget, query, counts, None, None, | |
| escalate=escalate) | |
| engine.close() | |
| return result, counts['target_calls'] | |
| (_, survivors, _, _, _), used = acquire(16, False) | |
| assert not survivors | |
| (static, _, probes, _, _), _ = acquire(used, False) | |
| (escalated, _, replayed, _, info), _ = acquire(used, True) | |
| assert replayed == [dict(probe, survivors_after=replayed[i]['survivors_after']) | |
| for i, probe in enumerate(probes)] | |
| temporal = CandidateSpace(static.machine, static.contract, temporal=True) | |
| temporal.survivors(fit_evidence(probes), {}) | |
| assert info['escalated'] is True | |
| assert escalated.work == {key: value + temporal.work[key] | |
| for key, value in static.work.items()} | |