"""Fault injection exercises real orchestration without hosted model requests.""" import copy import json import threading from pathlib import Path import sys import pytest sys.path.insert(0, str(Path(__file__).resolve().parents[2] / 'scripts/ops')) sys.path.insert(0, str(Path(__file__).resolve().parents[2] / 'services/hermes/scripts')) import suite_backends import suite_hierarchical import suite_multipass as workflow import suite_recovery from suite_contract import MODELS, Problem, preflight, validate_request, validate_result from suite_profiles import PROFILE_FIELDS, profile_source, validate_profiles from suite_recovery import Checkpoints from suite_sizing import balanced_sizes from suite_synthetic import fixture, score from test_suite_multipass import natural, public, review, wire_value from suite_capacity_fixture import fixture as realistic_fixture class NoWait: """Cancellation remains testable without real retry sleeps.""" def is_set(self): return False def wait(self, _): return False def source(size=14, large=False): raw, expected = realistic_fixture() if large else fixture(size) raw['routing'] = {'allow_external': True, 'allowed_external_providers': ['claude']} raw['execution'] = {'strategy': 'whole_suite', 'max_seconds': 7200, 'max_cost_usd': 60} return validate_request(raw, ['claude']), expected def oracle(monkeypatch, expected, failures=None): """Use an independent fixture map to check cross-batch control flow and validation.""" calls = [] failures = dict(failures or {}) def backend(request, cancel, *, invocation, progress=None, job_deadline=None): stage = invocation['stage'] calls.append({'stage': stage, 'aliases': sorted(c['alias'] for c in request['cases']), 'seconds': request['execution']['max_seconds'], 'deadline': job_deadline, 'turns': invocation['max_turns'], 'source': copy.deepcopy(request['cases'])}) if failures.get(stage, 0): failures[stage] -= 1 raise Problem('incomplete_generation', 502, final_event_subtype='success', final_is_error=True, exit_code=1, assistant_api_error_seen=True, structured_output_present=False, turns=1, max_turns=invocation['max_turns'], cost_usd_estimate=0, usage={'input_tokens':0,'output_tokens':0}) if stage == 'implementation_profiles': value = {'profiles': {}} for c in request['cases']: name = expected[c['alias']] p = {k: None for k in PROFILE_FIELDS} p.update(target=name, work='Implement '+name+' tests.', setup=name+' fixture', stimulus=name+' stimulus', observations=name+' measurements', machinery=name, evidence={'field':'success_criteria', 'quote':c['success_criteria'][:90]}) value['profiles'][c['alias']] = p else: parts = {} for c in request['cases']: parts.setdefault(expected[c['alias']], []).append(c['alias']) decision = natural(request, list(parts.values()), list(parts)) if stage in ('proposal_a','proposal_b','profile_proposal_a','profile_proposal_b','profile_reconciliation'): decision = public(decision) if stage in ('large_family_review','decision_audit'): decision = review(decision, decision) value = wire_value(decision, invocation) return value, {'model':MODELS['claude']['model'], 'cost_usd_estimate': 25.0, 'usage':{'input_tokens':100,'output_tokens':50}, 'turns':2, 'duration_api_ms':2, 'compaction':False, 'truncation':False, 'cli_diagnostics':{'exit_code':0}} monkeypatch.setattr(suite_backends, 'claude_generate', backend) return calls def run(monkeypatch, *, mode='direct', size=14, large=False, failures=None): request, expected = source(size, large) calls = oracle(monkeypatch, expected, failures) selected = {**preflight(request), 'execution_mode':mode} updates = [] result, metadata = workflow.generate(request, selected, NoWait(), '192.168.22.8', updates.append, credential_scope={'owner':'synthetic-test','providers':['claude']}) validate_result(result, request) return request, expected, calls, result, metadata, updates def test_proposal_b_retry_keeps_validated_a_and_same_source(monkeypatch): request, expected, calls, result, meta, updates = run(monkeypatch, failures={'proposal_b':1}) assert [c['stage'] for c in calls].count('proposal_a') == 1 assert [c['stage'] for c in calls].count('proposal_b') == 2 attempts = [c for c in calls if c['stage']=='proposal_b'] assert attempts[0]['source'] == attempts[1]['source'] assert meta['retry_count']['proposal_b'] == 1 assert 'proposal_a' in meta['checkpointed_passes'] assert meta['cost_usd_estimate'] > 60 and meta['cost_guard_enforced'] is False assert all(c['deadline']==calls[0]['deadline'] for c in calls) assert score({'groups':meta['review_summary']['natural_families']}, expected)['false_merge_pairs']==0 def test_review_failure_does_not_regenerate_prior_passes(monkeypatch): _, _, calls, _, meta, _ = run(monkeypatch, size=75, failures={'large_family_review':1}) for name in ('proposal_a','proposal_b','reconciliation'): assert [c['stage'] for c in calls].count(name)==1 assert [c['stage'] for c in calls].count('large_family_review')==2 assert meta['retry_count']['large_family_review']==1 @pytest.mark.parametrize('mode', ['direct','hierarchical']) def test_large_comparable_fixture_complete_and_globally_reconciled(monkeypatch, mode): request, expected, calls, result, meta, updates = run(monkeypatch, mode=mode, large=True) assert len(request['cases'])==363 assert max(len(g['members']) for g in result['groups'])<=5 assert len(result['groups'])==90 and meta['natural_family_count']==45 quality=score({'groups':meta['review_summary']['natural_families']}, expected) assert quality['false_merge_pairs']==quality['missed_merge_pairs']==0 assert all(d['part_sizes']==balanced_sizes(d['natural_case_count']) for d in meta['review_summary']['capacity_divisions']) if mode=='hierarchical': assert meta['profile_batch_count']==16 for c in calls: if c['stage'] in ('profile_proposal_a','profile_proposal_b','profile_reconciliation'): assert len(c['aliases'])==363 originals={c['alias']:c for c in request['cases']} reopened=[c for call in calls if call['stage']=='source_review' for c in call['source']] assert {c['alias']:c for c in reopened}==originals assert meta['cross_batch_review_count']==1 assert 'Implement ' not in json.dumps(updates) def test_failed_direct_switches_only_after_bounded_recovery(monkeypatch): _, _, calls, _, meta, _ = run(monkeypatch, failures={'proposal_b':3}) assert [c['stage'] for c in calls].count('proposal_a')==1 assert [c['stage'] for c in calls].count('proposal_b')==3 assert meta['execution_mode']=='hierarchical' assert meta['strategy_events'][-1]['reason']=='pass_generation_retries_exhausted' @pytest.mark.parametrize('code', ['provider_authentication','provider_forbidden','cancelled', 'invalid_review_evidence','worker_isolation_failed','job_time_budget_exhausted']) def test_nonretriable_conditions_stop_without_fallback(monkeypatch, code): request, _ = source() seen=[] def fail(*args, **kw): seen.append(kw['invocation']['stage']) raise Problem(code, 502) monkeypatch.setattr(suite_backends,'claude_generate',fail) with pytest.raises(Problem, match='^'+code+'$'): workflow.generate(request, preflight(request), NoWait(),'192.168.22.8') assert seen==['proposal_a'] def test_turn_recovery_grows_only_within_capacity(monkeypatch): request, expected = source() calls=oracle(monkeypatch,expected) original=suite_backends.claude_generate turns=[] def backend(*args,**kw): turns.append(kw['invocation']['max_turns']) if len(turns)<3: raise Problem('pass_turn_limit',502,cost_usd_estimate=0,turns=turns[-1], max_turns=turns[-1],usage={'input_tokens':1,'output_tokens':1}) return original(*args,**kw) monkeypatch.setattr(suite_backends,'claude_generate',backend) workflow.generate(request,preflight(request),NoWait(),'192.168.22.8') assert turns[:3]==[4,5,6] def test_checkpoint_scope_hash_and_restart_behavior(): request,_=source() call={'stage':'proposal_a','input':'synthetic','prior_hash':'a'} store=Checkpoints(request,'claude',{'owner':'a'},float('inf')) key=store.key(call);store.put(key,'proposal_a',{'groups':[]}) assert store.get(key,'proposal_a')=={'groups':[]} assert store.key({**call,'prior_hash':'b'})!=key assert Checkpoints(request,'claude',{'owner':'b'},float('inf')).key(call)!=key assert Checkpoints(request,'claude',{'owner':'a'},float('inf')).get(key,'proposal_a') is None store.clear();assert store.get(key,'proposal_a') is None def test_cli_cost_guard_is_absent_and_legacy_request_cost_is_unenforced(): request,_=source() request['execution']['max_cost_usd']=1e9 assert validate_request(request,['claude'])['execution']['max_cost_usd']==1e9 assert '--max-budget-usd' not in suite_backends.claude_command(MODELS['claude']['model'],60) assert preflight(request)['cost_guard_enforced'] is False def test_full_profile_set_fits_global_comparison_at_service_case_limit(): """Bounded profiles keep every alias globally visible; no neighborhood shortcut.""" from suite_policy import invocation request = {'campaign':'SYNTHETIC','suite':'CAPACITY','cases':[ {'alias':'CASE-'+str(i).zfill(48),'description':'Synthetic'} for i in range(400)]} profiles = {c['alias']:{k:'x'*n for k,n in PROFILE_FIELDS.items()} for c in request['cases']} compact = profile_source(request, profiles) proposal = {'groups':[{'name':'x'*56,'members':[c['alias']]} for c in request['cases']]} call = invocation('profile_reconciliation',compact,{'proposal_a':proposal,'proposal_b':proposal}) result = workflow.capacity(call,'claude',400) assert result['max_turns'] >= 4 and result['context_reserved_tokens'] <= 1000000 assert len(call['schema']['properties']['assignments']['required']) == 400 def test_profile_evidence_and_exact_aliases_are_independently_validated(): request,expected=source() c=request['cases'][0] p={k:None for k in PROFILE_FIELDS} p['evidence']={'field':'success_criteria','quote':c['success_criteria'][:90]} subset={**request,'cases':[c]} validate_profiles({'profiles':{c['alias']:p}},subset) with pytest.raises(Problem,match='invalid_case_assignments'): validate_profiles({'profiles':{'CASE-invented':p}},subset) p['evidence']['quote']='unsupported secret instructions' with pytest.raises(Problem,match='invalid_review_evidence'): validate_profiles({'profiles':{c['alias']:p}},subset) def test_checkpoint_hit_launches_no_new_model_call(monkeypatch): request,expected=source() calls=oracle(monkeypatch,expected) worker=workflow.Workflow(request,preflight(request),NoWait(),'192.168.22.8',credential_scope='synthetic') first=worker.call('proposal_a',request) second=worker.call('proposal_a',request) assert first==second and len(calls)==1 assert worker.safe_state()['checkpoint_reused']==['proposal_a'] assert len(worker.records)==1 def test_one_structured_assignment_repair_preserves_source(monkeypatch): request,expected=source() calls=oracle(monkeypatch,expected) original=suite_backends.claude_generate def backend(*args,**kwargs): value,metadata=original(*args,**kwargs) if len(calls)==1: value['assignments'].pop(next(iter(value['assignments']))) return value,metadata monkeypatch.setattr(suite_backends,'claude_generate',backend) result,meta=workflow.generate(request,preflight(request),NoWait(),'192.168.22.8') assert meta['retry_count']=={'proposal_a':1} assert calls[0]['source']==calls[1]['source'] validate_result(result,request)