152 lines
11 KiB
Python
152 lines
11 KiB
Python
"""Independent r26b final driver boundary review. In-memory dispatch only."""
|
|
import ast
|
|
import contextlib
|
|
import copy
|
|
import hashlib
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
OUT = ROOT / 'app/artifacts/ai_concurrency'
|
|
spec = importlib.util.spec_from_file_location('r26c_independent_driver', OUT / 'private_tcm_r26_acceptance_driver.py')
|
|
d = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(d)
|
|
|
|
|
|
def generation(kind='terminal', ordinal=1):
|
|
no_execution = kind == 'rejected'
|
|
terminal = kind == 'terminal'
|
|
identified = kind in ('terminal', 'running')
|
|
lifecycle = None
|
|
if identified:
|
|
lifecycle = dict(producer_terminal_status='stopped', persisted_run_status='stopped' if terminal else 'running',
|
|
independently_terminal=terminal, generation_transfer_closed=True, lifecycle_conflict=not terminal, stop_accepted=False,
|
|
verification_attempts=1, verification_elapsed_ms=10, observations=[dict(status='stopped' if terminal else 'running', observed_at=1800000000.2, within_budget=True)],
|
|
producer_terminal_observed_at=1800000000.0, persisted_run_observed_at=1800000000.2, stop_accepted_at=None,
|
|
transfer_closed_at=1800000000.1, verification_started_at=12345.0,
|
|
clock_domains=dict(wall='unix_seconds_microtime', verification_started_at='monotonic_seconds_hrtime', verification_elapsed_ms='monotonic_milliseconds'))
|
|
return dict(generation=ordinal, serialized_request=True, submitted=True,
|
|
submission=dict(submitted_at=1800000000.0, transfer_closed=True, transfer_closed_at=1800000000.1,
|
|
disposition='rejected_before_execution' if no_execution else 'unconfirmed'),
|
|
not_submitted=False, no_execution_confirmed=no_execution, original_task_captured=identified,
|
|
original_pair_captured=identified, transfer_closed=True, closed_at_request_ms=12.0, transfer_exit='natural_eof',
|
|
independently_terminal=terminal, lifecycle=lifecycle)
|
|
|
|
|
|
def summary(kinds=('terminal',)):
|
|
gs = [generation(k, i + 1) for i, k in enumerate(kinds)]
|
|
total, terminal, rejected, unknown = len(gs), kinds.count('terminal'), kinds.count('rejected'), kinds.count('unknown')
|
|
pending = total - terminal - rejected
|
|
return dict(audit_schema='v43r26b_submissions', generation_count=total, submitted_attempt_count=total,
|
|
submitted_transfer_closed_count=total, submitted_pending_attempt_count=pending, submitted_without_identity_count=unknown,
|
|
pre_execution_closed_attempt_count=rejected, not_submitted_attempt_count=0, rejected_before_execution_attempt_count=rejected,
|
|
never_submitted_generation_count=0, no_execution_generation_count=rejected, persisted=True, serialized_request_count=total,
|
|
original_task_count=total-rejected-unknown, original_run_count=total-rejected-unknown,
|
|
audit_bytes=4096, audit_sha256='c'*64, closed_generation_count=total, independently_terminal_generation_count=terminal,
|
|
pending_generation_count=pending, unidentified_generation_count=rejected+unknown,
|
|
lifecycle_consistent=True, independent_cleanup_confirmed=terminal > 0 and pending == 0, generations=gs)
|
|
|
|
|
|
def row(kinds=('terminal',)):
|
|
a = summary(kinds)
|
|
state = {k: a[k] for k in ('submitted_attempt_count','submitted_transfer_closed_count','submitted_pending_attempt_count',
|
|
'submitted_without_identity_count','pre_execution_closed_attempt_count','not_submitted_attempt_count','rejected_before_execution_attempt_count')}
|
|
state.update(transport_closed=True, upstream_terminal_confirmed=a['pending_generation_count']==0,
|
|
upstream_stop_confirmed=True, upstream_pending_task_count=a['pending_generation_count'], cancel_requested=False, timeout_reached=True)
|
|
return dict(ok=True, profile='qwen', preparation_isolation_verified=True, private_wire_audit=a, state=state,
|
|
production_acceptance_path=True, diagnostic_only=False, error_code='UPSTREAM_TIMEOUT')
|
|
|
|
|
|
class IndependentFinalBoundary(unittest.TestCase):
|
|
def test_current_production_and_integration_pins(self):
|
|
pins=json.loads((OUT/'v43r26c_harness_reader_independent_source_pins.safe.json').read_text(encoding='utf-8'))
|
|
for r in pins['source_pins']:
|
|
self.assertEqual(hashlib.sha256((ROOT/r['file']).read_bytes()).hexdigest(),r['sha256'])
|
|
for path, digest in pins['production_pins'].items():
|
|
self.assertEqual(hashlib.sha256((ROOT/path).read_bytes()).hexdigest(),digest)
|
|
self.assertNotIn('private_local_fcgi', __import__('sys').modules)
|
|
|
|
def test_no_execution_and_unknown_are_distinct_and_not_predecessor_proof(self):
|
|
for kinds in [('rejected',), ('unknown',), ('rejected','terminal'), ('terminal','running')]:
|
|
a=summary(kinds); self.assertEqual(d.safe_private_wire_summary(a),a)
|
|
self.assertEqual(d.independent_cleanup_confirmed(row(kinds)),kinds==('rejected','terminal'))
|
|
self.assertTrue(d.independent_cleanup_confirmed(row()))
|
|
|
|
def test_all_nested_private_identity_injections_refuse_safe_export(self):
|
|
for path in [(),('generations',0),('generations',0,'submission'),('generations',0,'lifecycle'),
|
|
('generations',0,'lifecycle','observations',0),('generations',0,'lifecycle','clock_domains')]:
|
|
a=summary(); target=a
|
|
for key in path: target=target[key]
|
|
target['submission_id']='private-admission-sentinel'
|
|
self.assertEqual(d.safe_private_wire_summary(a),{'persisted':False})
|
|
|
|
def test_false_counts_exits_clocks_and_budget_failclosed(self):
|
|
for name in [k for k in summary() if k.endswith('_count')]:
|
|
a=summary(); a[name]+=1; self.assertEqual(d.safe_private_wire_summary(a),{'persisted':False},name)
|
|
for kind in ('raw_exit','foreign_schema','late','clock','budget','source_time'):
|
|
a=summary(); g=a['generations'][0]; life=g['lifecycle']
|
|
if kind=='raw_exit':g['transfer_exit']='private-task-sentinel'
|
|
if kind=='foreign_schema':a['audit_schema']='v43r26_private_run_wire'
|
|
if kind=='late':life['verification_elapsed_ms']=2001
|
|
if kind=='clock':life['clock_domains']['verification_started_at']='unix_seconds_microtime'
|
|
if kind=='budget':life['observations'][0]['within_budget']=False
|
|
if kind=='source_time':life['observations'][0]['observed_at']=life['transfer_closed_at']-1
|
|
self.assertEqual(d.safe_private_wire_summary(a),{'persisted':False},kind)
|
|
|
|
def test_predecessor_gate_keeps_every_undispatched_slot(self):
|
|
for kinds in [('unknown',),('rejected',),('terminal','running')]:
|
|
writes=[]; real_open=Path.open
|
|
@contextlib.contextmanager
|
|
def memory_open(path,mode='r',*args,**kwargs):
|
|
if mode=='x':
|
|
stream=io.StringIO(); yield stream; stream.close()
|
|
else:
|
|
with real_open(path,mode,*args,**kwargs) as stream: yield stream
|
|
with patch.object(d,'run',return_value=row(kinds)) as dispatch, patch.object(d,'_local_fcgi',side_effect=AssertionError('No auth')) as auth, \
|
|
patch.object(Path,'exists',return_value=False),patch.object(Path,'open',new=memory_open), \
|
|
patch.object(Path,'write_text',side_effect=lambda text,**kw:writes.append(json.loads(text))),contextlib.redirect_stdout(io.StringIO()):
|
|
d.run_deadline_sequence('independent_final', ['qwen','openai'], d.argparse.ArgumentParser())
|
|
auth.assert_not_called(); self.assertEqual(dispatch.call_count,1)
|
|
final=writes[-1];self.assertEqual((len(final['expected_requests']),final['completed_requests'],len(final['unexecuted_requests'])),(4,1,3))
|
|
self.assertFalse(final['suite_complete']);self.assertEqual({r['reason'] for r in final['unexecuted_requests']},{'PREDECESSOR_CLEANUP_UNCONFIRMED'})
|
|
self.assertEqual([r['generation_budget_seconds'] for r in final['expected_requests']],[1,90,1,90])
|
|
|
|
def test_exact_generated_label_boundary_before_auth_and_dispatch(self):
|
|
for profile in ('qwen','openai'):
|
|
label='L'*(100-len(profile));self.assertEqual(len(d.validated_request_label(label,profile)),101)
|
|
with self.assertRaises(ValueError):d.validated_request_label(label+'L',profile)
|
|
real_open=Path.open;writes=[]
|
|
def guarded_open(path,mode='r',*args,**kwargs):
|
|
if any(c in mode for c in 'wax+'):
|
|
writes.append(mode);raise AssertionError('No result writes')
|
|
return real_open(path,mode,*args,**kwargs)
|
|
with patch.object(d,'_local_fcgi',side_effect=AssertionError('No auth')) as auth,patch.object(d,'uuid4',side_effect=AssertionError('No UUID')) as uid,\
|
|
patch.object(Path,'open',new=guarded_open):
|
|
out=d.run(profile,label+'L',save_body=False)
|
|
auth.assert_not_called();uid.assert_not_called();self.assertEqual(writes,[]);self.assertEqual(out['error_code'],'PRIVATE_DRIVER_REQUEST_FAILED')
|
|
|
|
def test_mapper_auth_get_and_clinical_exporter_stay_unchanged(self):
|
|
mapper=(OUT/'clinical_ttft_r26_review.py').read_text(encoding='utf-8')
|
|
before=(OUT/'v43r26b_before_clinical_ttft_r26_review.py.snapshot').read_text(encoding='utf-8')
|
|
for a,b in [('9c264a7f2871e62998b1328d2e892e35d8038a976120c591705ca887e83dcc10','b14facc5f3a32c684a37a1768ac6052e41f4b83fe93d31418cff87b1c7479291'),
|
|
('80e8eb0b4feac04d4c5d6837e89e9a4c67942a257986bbb3c5e19bd87678773f','46c570dfcd7ed364a927bfb5fca59c2fa719dec87bf8e17a9be605f7e88c8fa1')]:mapper=mapper.replace(a,b)
|
|
self.assertEqual(ast.dump(ast.parse(mapper)),ast.dump(ast.parse(before)))
|
|
for name in ('private_tcm_r26_run_input_probe.php',):self.assertEqual((OUT/name).read_bytes(),(OUT/f'v43r26b_before_{name}.snapshot').read_bytes())
|
|
name='private_tcm_r26_run_input_inspector.php';now=(OUT/name).read_text(encoding='utf-8');old=(OUT/f'v43r26b_before_{name}.snapshot').read_text(encoding='utf-8')
|
|
marker=' private static function connection(';self.assertEqual(now[now.index(marker):],old[old.index(marker):])
|
|
for kind in ('acceptance','deadline'):
|
|
name=f'private_tcm_r26_{kind}_probe.php';now=(OUT/name).read_text(encoding='utf-8');old=(OUT/f'v43r26b_before_{name}.snapshot').read_text(encoding='utf-8')
|
|
for start,end in [(' // No real preparation',' $testedHashes='),(' $reply = ',' if ($reply === null)')]:
|
|
self.assertEqual(now[now.index(start):now.index(end)],old[old.index(start):old.index(end)])
|
|
now=(OUT/'private_tcm_r26_acceptance_driver.py').read_text(encoding='utf-8');old=(OUT/'v43r26b_before_private_tcm_r26_acceptance_driver.py.snapshot').read_text(encoding='utf-8')
|
|
start=' if result.get("preparation_isolation_verified") is True and save_body:'
|
|
self.assertEqual(now[now.index(start):now.index(' return result',now.index(start))],old[old.index(start):old.index(' return result',old.index(start))])
|
|
|
|
|
|
if __name__=='__main__':unittest.main()
|