diff --git a/coordinator/robot_bt_coordinator/provenance.py b/coordinator/robot_bt_coordinator/provenance.py new file mode 100644 index 0000000..1dd2135 --- /dev/null +++ b/coordinator/robot_bt_coordinator/provenance.py @@ -0,0 +1,70 @@ +"""Content identities with explicit scope; source files do not attest remote binaries.""" +import hashlib +from pathlib import Path +import subprocess + +from .plan import canonical + + +def digest(value): + return hashlib.sha256(canonical(value).encode('utf-8')).hexdigest() + + +def missing(reason): + return {'status': 'missing', 'reason': reason} + + +def file_identity(path, root): + try: + data = path.read_bytes() + except OSError: + return missing('source file unavailable') + return {'status': 'captured', 'scope': 'source_only', + 'path': path.relative_to(root).as_posix(), + 'sha256': hashlib.sha256(data).hexdigest()} + + +def capture_sources(root=None): + """Capture once at coordinator startup, never assert this is a binary build ID.""" + root = Path(root) if root is not None else Path(__file__).resolve().parents[2] + code = missing('source checkout or git unavailable') + try: + def git(*args): + return subprocess.check_output(['git', '-C', str(root), *args], + stderr=subprocess.DEVNULL, timeout=3).decode().strip() + # Installed packages inside an unrelated git checkout must not inherit its identity. + if (root / '.git').exists(): + commit = git('rev-parse', 'HEAD') + dirty = bool(git('status', '--porcelain', '--untracked-files=normal')) + code = {'status': 'captured', 'scope': 'checkout_at_coordinator_start', + 'commit': commit, 'dirty': dirty, + 'binary_build_identity': missing('build attestation unavailable')} + except (OSError, subprocess.SubprocessError, UnicodeError): + pass + definitions = sorted((root / 'ros2/bt_skill_interfaces').glob('*/*')) + files = [file_identity(p, root) for p in definitions if p.suffix in ('.msg', '.srv', '.action')] + idl = {'status': 'captured', 'scope': 'source_only', 'files': files, + 'sha256': digest(files)} if files else missing('IDL source unavailable') + code_files = [] + for folder in ('coordinator', 'core', 'robobrain', 'ros2'): + for path in sorted((root / folder).rglob('*')): + if path.suffix in ('.py', '.cpp', '.hpp', '.h', '.cmake') or path.name == 'CMakeLists.txt': + code_files.append(file_identity(path, root)) + content = ({'status': 'captured', 'scope': 'source_only_at_coordinator_start', + 'sha256': digest(code_files), 'files': code_files} + if code_files else missing('code source unavailable')) + return {'code': code, 'code_content': content, 'xml': file_identity(root / 'ros2/bt_executor/trees/fixed_workflow.xml', root), + 'idl': idl, 'planner_source': file_identity(root / 'robobrain/robot_robobrain/service.py', root)} + + +def execution_versions(sources, task, backend_config): + # Hash configuration only: credentials and operator identities must not be + # copied into task events. Remote service configuration is explicitly unknown. + config = {'context': task['context'], 'backend': backend_config} + return {'provenance_schema': 1, 'sources': sources, + 'runtime_config': {'status': 'captured', 'scope': 'coordinator_dispatch', 'sha256': digest(config)}, + 'approved_plan': {'status': 'captured', 'sha256': digest(task['execution_plan'])}, + 'instruction': {'status': 'captured', 'sha256': digest(task['request']['instruction'])}, + 'planning_record_ref': task.get('planning_record_ref'), + 'instruction_template': missing('remote planner template not attested; inspect linked planning record if available'), + 'deployed_executor': missing('loaded XML, generated IDL, binary and executor parameters are not attested by current transport')} diff --git a/coordinator/robot_bt_coordinator/replay.py b/coordinator/robot_bt_coordinator/replay.py index 829530c..e9035e1 100644 --- a/coordinator/robot_bt_coordinator/replay.py +++ b/coordinator/robot_bt_coordinator/replay.py @@ -3,6 +3,7 @@ No ROS import or execution transport exists in this module. Sensor/model inference is not rerun, and this result must never be used as live robot state. """ +from copy import deepcopy import argparse import json import sqlite3 @@ -11,7 +12,7 @@ from .plan import validate_plan from .plan_v2 import instances def replay_events(events): - status='RECEIVED';version=0;cursor=0;delivered={};stages=[];plans=[];errors=[];progress=[];reconciliations=[] + status='RECEIVED';version=0;cursor=0;delivered={};stages=[];plans=[];errors=[];progress=[];reconciliations=[];planning_progress=[] for e in events: if type(e['event_id']) is not int or e['event_id']<=cursor:raise ValueError('non-monotonic event cursor') cursor=e['event_id'] @@ -20,6 +21,8 @@ def replay_events(events): version=e['status_version'];status=e['status'] if e.get('error_code'):errors.append(e['error_code']) if e['kind']=='progress':progress.append({k:e[k] for k in ('event_id','run_id','stage','sequence','detail') if k in e}) + elif e['kind']=='planning_progress': + planning_progress.append(deepcopy(e['payload'])) elif e['kind']=='plan_approved': plans.append({'run_id':e['run_id'],'plan':validate_plan(e['plan'])}) elif e['kind']=='delivery_committed': @@ -31,7 +34,7 @@ def replay_events(events): elif e['kind']=='reconciliation': reconciliations.append({k:e[k] for k in ('event_id','run_id','resolution','evidence_ref','credited_items') if k in e}) if status=='SUCCEEDED' and (not plans or set(delivered)!=set(range(len(instances(plans[-1]['plan']))))):raise ValueError('success without committed delivery') - return {'mode':'offline_audit_replay','status':status,'status_version':version,'last_event_id':cursor,'completed_quantity':len(delivered),'plans':plans,'results':stages,'progress':progress,'reconciliations':reconciliations,'errors':errors,'connects_to_robot':False} + return {'mode':'offline_audit_replay','status':status,'status_version':version,'last_event_id':cursor,'completed_quantity':len(delivered),'plans':plans,'results':stages,'progress':progress,'planning_progress':planning_progress,'reconciliations':reconciliations,'errors':errors,'connects_to_robot':False} def main(): p=argparse.ArgumentParser();p.add_argument('--db',required=True);p.add_argument('--task-id',required=True);args=p.parse_args() diff --git a/coordinator/robot_bt_coordinator/ros_backend.py b/coordinator/robot_bt_coordinator/ros_backend.py index bc2fd9a..bddfcc2 100644 --- a/coordinator/robot_bt_coordinator/ros_backend.py +++ b/coordinator/robot_bt_coordinator/ros_backend.py @@ -135,9 +135,37 @@ class RosBackend: constraints['clarification_confirmed'] = True goal.constraints_json = canonical(constraints) self._duration(goal.timeout, self.config.get('planning_timeout', 25)) - future = self._planner.send_goal_async(goal) + future = self._planner.send_goal_async( + goal, feedback_callback=lambda feedback: self._plan_feedback(key, feedback)) future.add_done_callback(lambda completed: self._plan_accepted(key, completed)) + def _plan_feedback(self, key, wrapped): + """Retain actual Planner fields; feedback never decides task completion.""" + with self._lock: + rec = self._planning.get(key) + if rec is None or rec['done']: + return + try: + msg = wrapped.feedback + sequence, phase = msg.sequence, msg.phase + sec, nanosec = msg.stamp.sec, msg.stamp.nanosec + if (type(sequence) is not int or not 0 < sequence < 2**32 or + sequence <= rec.get('feedback_sequence', 0) or + type(phase) is not int or not 0 <= phase <= 255 or + type(sec) is not int or not -(2**31) <= sec < 2**31 or + type(nanosec) is not int or not 0 <= nanosec < 1_000_000_000 or + not isinstance(msg.message, str)): + return + event = dict(type='planning_progress', task_id=rec['task_id'], + task_revision=rec['task_revision'], + planning_generation=rec['planning_generation'], + stamp=dict(sec=sec, nanosec=nanosec), sequence=sequence, + phase=phase, message=msg.message) + except (AttributeError, TypeError, ValueError): + return + rec['feedback_sequence'] = sequence + self._emit(event) + def _plan_accepted(self, key, future): with self._lock: rec = self._planning.get(key) diff --git a/coordinator/robot_bt_coordinator/service.py b/coordinator/robot_bt_coordinator/service.py index f59ac06..793fa45 100644 --- a/coordinator/robot_bt_coordinator/service.py +++ b/coordinator/robot_bt_coordinator/service.py @@ -10,6 +10,7 @@ from .plan import canonical, text, validate_known, validate_plan from .store import Store from .plan_v2 import instances, item_plan from .intent import conflicts_with_instruction +from .provenance import capture_sources, execution_versions TERMINAL = {'SUCCEEDED','FAILED','CANCELED','EXPIRED'} EXECUTION = {'READY','EXECUTING','PAUSING','CANCELING','INTERVENTION_REQUIRED'} @@ -30,6 +31,7 @@ class Coordinator: self.robots=set(robots);self.site=site if site is not None else demo_site() self.clock=clock;self.queue_timeout=queue_timeout;self.closed=False self.steady=steady;self.planning_timeout=planning_timeout;self._planning_started={} + self._source_provenance=capture_sources() with self.lock,self.store.db: for t in self.store.active(): if t['status'] in EXECUTION or (t['status']=='PAUSED' and t.get('motion_dispatched')): @@ -221,7 +223,17 @@ class Coordinator: self._set(t,'QUEUED' if t['planning_attempts']<2 else 'FAILED',error) def _event(self,t,e): typ=e.get('type') - if typ=='plan': + if typ=='planning_progress': + if t['status']!='PLANNING' or e.get('task_revision')!=t['task_revision'] or e.get('planning_generation')!=t['planning_generation']:return + seq=e.get('sequence');phase=e.get('phase');stamp=e.get('stamp') + if type(seq) is not int or not 0=self.planning_timeout: @@ -252,8 +264,9 @@ class Coordinator: except (ApiError,ValueError,KeyError,TypeError) as ex: self._planning_failure(t,getattr(ex,'code','INVALID_PLAN'));return t['plan']=p;t['plan_version']=p['plan_version'];t['active_item_index']=0;t['requested_quantity']=len(instances(p)) + t['planning_record_ref']=e.get('planning_record_ref') t['run_id']=str(uuid.uuid4()) - self.store.event(t,'plan_approved',{'plan':p,'task_revision':t['task_revision'],'planning_generation':t['planning_generation'],'run_id':t['run_id']},self.clock()) + self.store.event(t,'plan_approved',{'plan':p,'task_revision':t['task_revision'],'planning_generation':t['planning_generation'],'run_id':t['run_id'],'planning_record_ref':t['planning_record_ref']},self.clock()) self._dispatch_item(t,initial=True) elif typ in {'execution_result','progress','cancel_ack','advisory'}: if e.get('run_id')!=t['run_id'] or t['status'] in TERMINAL or not t['motion_dispatched']:return @@ -325,7 +338,10 @@ class Coordinator: t['context']={**self.site,'robot_id':t['robot_id'],'target_id':slots['target_name'],'source_shelf':source['shelf_id'],'destination_id':dest['region_ref'],'observe_location':source['observe_location'],'destination_location':dest['location'],'task_revision':t['task_revision']} if t['plan']['schema_version']==2:t['context'].update(item_index=index,route=t['plan']['route']) self._set(t,'READY');t['motion_dispatched']=True;t['stop_confirmed']=False;self._set(t,'EXECUTING') - self.store.event(t,'execution_dispatched',{'run_id':t['run_id'],'item_index':index,'context':t['context'],'versions':{'coordinator':'1.2.0','schema':t['plan']['schema_version'],'tree':'1.2.0'}},self.clock()) + versions=execution_versions(self._source_provenance,t,{ + 'adapter':getattr(self.backend,'config',{}), + 'coordinator':{'queue_timeout':self.queue_timeout,'planning_timeout':self.planning_timeout,'robots':sorted(self.robots)}}) + self.store.event(t,'execution_dispatched',{'run_id':t['run_id'],'item_index':index,'context':t['context'],'versions':versions},self.clock()) self.store.db.commit() try:self.backend.start_execution(t) except Exception:self._set(t,'INTERVENTION_REQUIRED','DISPATCH_ACCEPTANCE_UNKNOWN') diff --git a/robobrain/robot_robobrain/service.py b/robobrain/robot_robobrain/service.py index d1861a5..54e87d0 100644 --- a/robobrain/robot_robobrain/service.py +++ b/robobrain/robot_robobrain/service.py @@ -26,9 +26,12 @@ class BrainService: with open(path,'xb') as f:os.chmod(path,0o600);f.write(data);f.flush();os.fsync(f.fileno()) except FileExistsError:pass result.update(image_path=str(path.resolve()),sha256=digest);return result - def _record(self,capability,goal,raw,result,observation=None): + def _record(self,capability,goal,raw,result,observation=None,prompt=None): path=self.records/(uuid.uuid4().hex+'.json') body=dict(schema_version=1,capability=capability,recorded_at_ns=time.time_ns(),model_version=self.backend.model_version,prompt_version=PROMPT_VERSION,input=goal,observation=observation,raw_output=raw,result=result) + body['prompt_provenance']=({'status':'captured','sha256':hashlib.sha256(prompt.encode('utf-8')).hexdigest(), + 'template_sha256':hashlib.sha256(PLANNER_RULES.encode('utf-8')).hexdigest()} if prompt is not None and capability=='plan' + else {'status':'missing','reason':'no captured inference prompt for this record'}) with open(path,'x',encoding='utf-8') as f: os.chmod(path,0o600);f.write(canonical(body));f.flush();os.fsync(f.fileno()) return str(path.resolve()) @@ -44,7 +47,7 @@ class BrainService: if not isinstance(raw,str) or len(raw.encode())>262144:raise InferenceError('OUTPUT_TOO_LARGE') return raw def plan(self,goal,cancel=None): - raw='' + raw='';prompt=None try: if not isinstance(goal.get('instruction'),str) or not 0 0 and isinstance(row['message'], str) and + 0 <= row['stamp']['nanosec'] < 1000000000 for row in planning_feedback) + assert [row['sequence'] for row in planning_feedback] == sorted(set(row['sequence'] for row in planning_feedback)) + dispatches = [event for event in events if event['kind']=='execution_dispatched'] + assert dispatches and dispatches[-1]['versions']['provenance_schema'] == 1 + assert dispatches[-1]['versions']['runtime_config']['status'] == 'captured' assert any(event['kind']=='delivery_committed' for event in events) report['normal'] = {'task_id':first['task_id'],'run_id':first['run_id'], 'status':first['status'],'completed_quantity':first['completed_quantity'], 'planning_record_ref':planning[-1]['payload']['planning_record_ref'], + 'planning_feedback':planning_feedback, + 'execution_versions':dispatches[-1]['versions'], 'counts':counts()} second = submit('lost-final-result') diff --git a/tests/test_native_soak.py b/tests/test_native_soak.py new file mode 100644 index 0000000..f172269 --- /dev/null +++ b/tests/test_native_soak.py @@ -0,0 +1,128 @@ +"""Real short-duration process tests; these never stand in for an eight-hour run.""" +import importlib.util +import json +import os +from pathlib import Path +import subprocess +import sys +import tempfile +import time +import unittest + +ROOT = Path(__file__).resolve().parents[1] +TOOL = ROOT / 'tools/native_soak.py' + + +class VerdictTests(unittest.TestCase): + def test_eight_hour_liveness_is_not_full_dr_acceptance(self): + # Classification logic only: this test supplies a number, not duration evidence. + spec = importlib.util.spec_from_file_location('native_soak', TOOL) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + report = module.verdict(28800, True) + self.assertTrue(report['duration_liveness_pass']) + self.assertFalse(report['full_dr_acceptance']) + self.assertTrue(report['resource_growth_review_required']) + self.assertFalse(report['no_deadlock_claim']) + for seconds, progress, gap in [(28799, True, False), (28800, False, False), (28800, True, True)]: + self.assertFalse(module.verdict(seconds, progress, sampling_gap=gap)['eight_hour_pass']) + +@unittest.skipUnless(sys.platform.startswith('linux'), 'Linux /proc required') +class NativeSoakTests(unittest.TestCase): + def run_monitor(self, directory, *args): + output = Path(directory) / 'evidence' + process = subprocess.run([sys.executable, str(TOOL), '--output', str(output), + '--duration-seconds', '0.45', '--sample-interval-seconds', '0.05', + *args], capture_output=True, text=True, timeout=8) + self.assertTrue((output / 'summary.json').exists(), process.stderr) + return process, json.loads((output / 'summary.json').read_text()), output + + def test_late_increment_cannot_erase_elapsed_progress_deadline(self): + # Fault-injection unit test only; this clock is never duration evidence. + from unittest.mock import patch + spec = importlib.util.spec_from_file_location('native_soak_deadline', TOOL) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + with tempfile.TemporaryDirectory() as directory: + progress = Path(directory) / 'counter.json' + progress.write_text('{"completed":0}') + output = Path(directory) / 'evidence' + clock = [0.0] + + def advance(seconds): + clock[0] += seconds + # The new completion is only visible after the .12-second deadline. + if clock[0] >= .16: + progress.write_text('{"completed":1}') + + with patch.object(module.time, 'monotonic', side_effect=lambda: clock[0]), \ + patch.object(module.time, 'sleep', side_effect=advance), patch('builtins.print'): + code = module.main(['--pid', str(os.getpid()), '--output', str(output), + '--duration-seconds', '.16', '--sample-interval-seconds', '.08', + '--max-progress-gap-seconds', '.12', '--progress-file', str(progress)]) + report = json.loads((output / 'summary.json').read_text()) + self.assertEqual(code, 1) + self.assertEqual(report['status'], 'failed') + self.assertIn('progress deadline', report['reason']) + def test_real_short_workload_is_incomplete_and_one_process(self): + with tempfile.TemporaryDirectory() as directory: + progress = Path(directory) / 'progress.json' + script = ('import json,os,time,sys; from pathlib import Path; p=Path(sys.argv[1]); n=0\n' + 'while True:\n n+=1; q=p.with_suffix(".tmp"); q.write_text(json.dumps({"completed":n})); os.replace(q,p); time.sleep(.025)\n') + process, report, output = self.run_monitor(directory, '--progress-file', str(progress), + '--max-progress-gap-seconds', '0.25', '--command', sys.executable, '-u', '-c', script, str(progress)) + self.assertEqual(process.returncode, 2, process.stderr) + self.assertEqual(report['status'], 'qualification_incomplete') + self.assertFalse(report['eight_hour_pass']) + self.assertGreaterEqual(report['elapsed_seconds'], .45) + self.assertTrue(report['progress_observed']) + samples = [json.loads(line) for line in (output / 'samples.jsonl').read_text().splitlines()] + self.assertGreater(len(samples), 3) + self.assertEqual(len({(s['pid'], s['start_ticks']) for s in samples}), 1) + self.assertGreater(samples[-1]['rss_bytes'], 0) + self.assertIn('cpu_seconds', samples[-1]) + self.assertIn('fd_count', samples[-1]) + self.assertFalse(report['no_deadlock_claim']) + + def test_attached_process_is_left_alive_and_no_metric_not_pass(self): + with tempfile.TemporaryDirectory() as directory: + child = subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(10)']) + try: + process, report, _ = self.run_monitor(directory, '--pid', str(child.pid)) + self.assertEqual(process.returncode, 2, process.stderr) + self.assertIsNone(child.poll()) + self.assertFalse(report['progress_observed']) + self.assertFalse(report['no_deadlock_claim']) + finally: + child.terminate() + child.wait(timeout=3) + + def test_early_exit_fails(self): + with tempfile.TemporaryDirectory() as directory: + process, report, _ = self.run_monitor(directory, '--command', sys.executable, '-c', 'pass') + self.assertEqual(process.returncode, 1) + self.assertEqual(report['status'], 'failed') + self.assertIn('process', report['reason']) + + def test_stalled_metric_fails_even_if_process_alive(self): + with tempfile.TemporaryDirectory() as directory: + progress = Path(directory) / 'progress.json' + progress.write_text('{"completed":1}') + process, report, _ = self.run_monitor(directory, '--progress-file', str(progress), + '--max-progress-gap-seconds', '.15', '--command', sys.executable, '-c', 'import time; time.sleep(10)') + self.assertEqual(process.returncode, 1) + self.assertIn('progress', report['reason']) + + def test_journal_size_is_observed(self): + with tempfile.TemporaryDirectory() as directory: + journal = Path(directory) / 'goals.log' + journal.write_bytes(b'goal\n') + process, report, output = self.run_monitor(directory, '--journal', str(journal), + '--command', sys.executable, '-c', 'import time; time.sleep(10)') + self.assertEqual(process.returncode, 2) + sample = json.loads((output / 'samples.jsonl').read_text().splitlines()[-1]) + self.assertEqual(sample['journal_bytes'][str(journal.resolve())], 5) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_physics_delivery.py b/tests/test_physics_delivery.py new file mode 100644 index 0000000..ad51850 --- /dev/null +++ b/tests/test_physics_delivery.py @@ -0,0 +1,59 @@ +import copy +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'tools')) +from physics_delivery_probe import verify_delivery + + +class DeliveryPredicateTests(unittest.TestCase): + def setUp(self): + self.frames = [dict(time=i / 240, body_id=4, destination_id=2, + aabb=[[-.02, -.02, .01], [.02, .02, .05]], + linear_speed=0., angular_speed=0., + floor_contact=True, held=False) for i in range(121)] + self.bounds = [[-.2, -.2, 0.], [.2, .2, .3]] + + def verify(self, frames=None): + return verify_delivery(self.frames if frames is None else frames, + self.bounds, 4, 2) + + def test_settled_released_object(self): + self.assertTrue(self.verify()['verified']) + + def test_reject_each_unsafe_condition(self): + for update in ({'held': True}, {'floor_contact': False}, + {'linear_speed': .1}, {'angular_speed': .3}, + {'body_id': 8}, {'destination_id': 9}, + {'aabb': [[.19, 0., .01], [.23, .04, .05]]}, + {'linear_speed': float('nan')}, {'held': None}, + {'floor_contact': 'true'}): + with self.subTest(update=update): + frames = copy.deepcopy(self.frames) + frames[-1].update(update) + self.assertFalse(self.verify(frames)['verified']) + + def test_missing_or_short_evidence(self): + for frames in ([], self.frames[-20:], [{}]): + self.assertFalse(self.verify(frames)['verified']) + + def test_time_must_increase_without_sampling_gap(self): + for value in (self.frames[-2]['time'], 1., float('nan')): + frames = copy.deepcopy(self.frames) + frames[-1]['time'] = value + self.assertFalse(self.verify(frames)['verified']) + + def test_stability_requires_whole_window(self): + frames = copy.deepcopy(self.frames) + frames[-40]['held'] = True + self.assertFalse(self.verify(frames)['verified']) + + def test_invalid_earlier_timestamp_cannot_be_filtered_out(self): + frames = copy.deepcopy(self.frames) + frames[-30]['time'] = float('nan') + self.assertFalse(self.verify(frames)['verified']) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_planner_feedback.py b/tests/test_planner_feedback.py new file mode 100644 index 0000000..a6b15d8 --- /dev/null +++ b/tests/test_planner_feedback.py @@ -0,0 +1,65 @@ +import sys, threading, tempfile, unittest +from pathlib import Path +from types import SimpleNamespace as NS +from collections import deque +ROOT=Path(__file__).resolve().parents[1] +sys.path.insert(0,str(ROOT/'coordinator')) +from robot_bt_coordinator.ros_backend import RosBackend +from robot_bt_coordinator.service import Coordinator +from robot_bt_coordinator.backends import ManualBackend +from robot_bt_coordinator.replay import replay_events + +class PlannerFeedbackTests(unittest.TestCase): + def backend(self): + b=RosBackend.__new__(RosBackend);b._lock=threading.RLock();b._events=deque();b._planning={};b._closed=False;b.config={} + class Planner: + def server_is_ready(self):return True + def send_goal_async(self,goal,**kwargs): + self.callback=kwargs.get('feedback_callback');return NS(add_done_callback=lambda cb:None) + b._planner=Planner();b._PlanTask=NS(Goal=lambda:NS(timeout=NS())) + b.start_planning(dict(task_id='t',task_revision=1,planning_generation=2,robot_id='r',request={'instruction':'move item'})) + return b + def feedback(self,sequence=1,phase=7,message='actual phase text'): + return NS(feedback=NS(stamp=NS(sec=12,nanosec=345),sequence=sequence,phase=phase,message=message)) + def test_real_fields_are_copied_without_phase_invention(self): + b=self.backend();self.assertTrue(callable(b._planner.callback)) + b._planner.callback(self.feedback());e=b._events.popleft() + self.assertEqual(e,dict(type='planning_progress',task_id='t',task_revision=1,planning_generation=2,stamp={'sec':12,'nanosec':345},sequence=1,phase=7,message='actual phase text')) + def test_duplicate_reordered_and_terminal_feedback_ignored(self): + b=self.backend();self.assertTrue(callable(b._planner.callback)) + for seq in (2,2,1):b._planner.callback(self.feedback(seq)) + b._planning[('t',1,2)]['done']=True;b._planner.callback(self.feedback(3)) + self.assertEqual(len(b._events),1) + def test_invalid_fields_do_not_kill_callback(self): + b=self.backend();self.assertTrue(callable(b._planner.callback)) + for field,value in [('sequence',True),('sequence',-1),('sequence',2**32),('phase',256),('message',object())]: + f=self.feedback();setattr(f.feedback,field,value);b._planner.callback(f) + f=self.feedback();f.feedback.stamp.nanosec=1000000000;b._planner.callback(f) + self.assertFalse(b._events) + def test_feedback_is_persisted_and_stale_generation_rejected(self): + with tempfile.TemporaryDirectory() as tmp: + path=str(Path(tmp)/'tasks.db');backend=ManualBackend();c=Coordinator(path,backend,{'r'}) + try: + tid=c.submit(dict(client_request_id='req',robot_id='r',instruction='move item',known_info={}))['task_id'];c.tick();t=c.get(tid) + event=dict(type='planning_progress',task_id=tid,task_revision=t['task_revision'],planning_generation=t['planning_generation'],stamp={'sec':12,'nanosec':345},sequence=1,phase=7,message='actual phase text') + backend.emit(event);backend.emit(event);backend.emit(dict(event,planning_generation=0,sequence=2));c.tick() + rows=[e for e in c.events(tid) if e['kind']=='planning_progress'];self.assertEqual(len(rows),1);self.assertEqual(rows[0]['payload'],event) + self.assertEqual(c.get(tid)['status'],'PLANNING') + # A new generation starts its own sequence; old callbacks cannot + # contaminate it, and feedback after leaving planning is ignored. + current=c.store.get(tid);current['planning_generation']+=1 + with c.store.db:c.store.put(current) + newer=dict(event,planning_generation=current['planning_generation'],message='next actual phase') + backend.emit(dict(event,sequence=99));backend.emit(newer);c.tick() + rows=[e['payload'] for e in c.events(tid) if e['kind']=='planning_progress'] + self.assertEqual(rows,[event,newer]) + c.control(tid,'cancel');backend.emit(dict(newer,sequence=2));c.tick() + self.assertEqual(c.get(tid)['status'],'CANCELED') + replay=replay_events(c.events(tid));self.assertEqual(replay['planning_progress'],[event,newer]);self.assertFalse(replay['connects_to_robot']) + self.assertEqual([e['payload'] for e in c.events(tid) if e['kind']=='planning_progress'],[event,newer]) + finally:c.close() + c=Coordinator(path,ManualBackend(),{'r'}) + try:self.assertEqual([e['payload'] for e in c.events(tid) if e['kind']=='planning_progress'],[event,newer]) + finally:c.close() + +if __name__=='__main__':unittest.main() diff --git a/tests/test_provenance.py b/tests/test_provenance.py new file mode 100644 index 0000000..3a66ad9 --- /dev/null +++ b/tests/test_provenance.py @@ -0,0 +1,97 @@ +import hashlib +import json +from pathlib import Path +import sys +import tempfile +import subprocess +import unittest + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'coordinator')) +from robot_bt_coordinator.provenance import capture_sources, execution_versions +import test_coordinator as fixture + + +class ProvenanceTest(unittest.TestCase): + def test_git_commit_and_dirty_are_measured(self): + with tempfile.TemporaryDirectory() as tmp: + subprocess.run(['git', 'init', '-q', tmp], check=True) + subprocess.run(['git', '-C', tmp, '-c', 'user.name=Test', '-c', 'user.email=test@example.invalid', + 'commit', '--allow-empty', '-qm', 'fixture'], check=True) + clean = capture_sources(tmp)['code'] + self.assertEqual(clean['status'], 'captured') + self.assertEqual(len(clean['commit']), 40) + self.assertFalse(clean['dirty']) + Path(tmp, 'changed.py').write_text('changed') + dirty = capture_sources(tmp)['code'] + self.assertTrue(dirty['dirty']) + self.assertEqual(clean['commit'], dirty['commit']) + def test_actual_planner_prompt_is_hashed_in_linked_record(self): + sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'robobrain')) + from robot_robobrain.service import BrainService, PLANNER_RULES + from robot_robobrain.backends import FixtureBackend + from robot_bt_coordinator.plan import canonical + from test_robobrain import RoboBrainTests + requests = [] + def infer(request): + requests.append(request) + return '{"missing_information":["which destination?"]}' + with tempfile.TemporaryDirectory() as tmp: + result = BrainService(FixtureBackend(infer), tmp).plan(RoboBrainTests().goal()) + record = json.loads(Path(result['record_ref']).read_text()) + self.assertEqual(record['prompt_provenance']['sha256'], hashlib.sha256(requests[0]['prompt'].encode()).hexdigest()) + self.assertEqual(record['prompt_provenance']['template_sha256'], hashlib.sha256(PLANNER_RULES.encode()).hexdigest()) + self.assertEqual(requests[0]['prompt'], PLANNER_RULES + '\nINPUT: ' + canonical(record['input'])) + def test_source_hash_changes_and_missing_is_explicit(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + xml = root / 'ros2/bt_executor/trees/fixed_workflow.xml' + xml.parent.mkdir(parents=True) + xml.write_bytes(b'') + first = capture_sources(root) + self.assertEqual(first['xml']['sha256'], hashlib.sha256(b'').hexdigest()) + self.assertEqual(first['xml']['scope'], 'source_only') + self.assertEqual(first['idl']['status'], 'missing') + self.assertEqual(first['code']['status'], 'missing') + source = root / 'coordinator/example.py' + source.parent.mkdir() + source.write_text('version_one = True') + before = capture_sources(root)['code_content'] + source.write_text('version_two = True') + self.assertNotEqual(before['sha256'], capture_sources(root)['code_content']['sha256']) + xml.write_bytes(b'') + self.assertNotEqual(first['xml'], capture_sources(root)['xml']) + + def test_actual_configuration_hash_is_order_independent_and_sensitive(self): + task = {'context': {'a': 1, 'b': 2}, 'execution_plan': {'schema_version': 1}, + 'request': {'instruction': 'move water'}, 'planning_record_ref': '/record/1'} + first = execution_versions({}, task, {'timeout': 5, 'recovery_token': 'secret'}) + task['context'] = {'b': 2, 'a': 1} + self.assertEqual(first, execution_versions({}, task, {'recovery_token': 'secret', 'timeout': 5})) + task['context']['a'] = 3 + self.assertNotEqual(first['runtime_config'], execution_versions({}, task, {})['runtime_config']) + self.assertNotIn('secret', json.dumps(first)) + self.assertEqual(first['deployed_executor']['status'], 'missing') + self.assertEqual(first['instruction_template']['status'], 'missing') + self.assertEqual(first['planning_record_ref'], '/record/1') + + +class DispatchProvenanceTest(unittest.TestCase): + setUp = fixture.CoordinatorTest.setUp + tearDown = fixture.CoordinatorTest.tearDown + plan = fixture.CoordinatorTest.plan + def test_dispatch_identity_survives_database_reopen(self): + tid = self.c.submit(fixture.REQ)['task_id'] + self.plan(tid) + events = self.c.store.events(tid, 0, 100) + versions = next(e['versions'] for e in events if e['kind'] == 'execution_dispatched') + self.assertEqual(versions['runtime_config']['status'], 'captured') + self.assertEqual(len(versions['sources']['xml']['sha256']), 64) + self.assertIn(versions['sources']['code']['status'], ('captured', 'missing')) + import sqlite3 + with sqlite3.connect(self.db) as connection: + persisted = json.loads(connection.execute("SELECT data FROM events WHERE kind='execution_dispatched'").fetchone()[0]) + self.assertEqual(versions, persisted['versions']) + + +if __name__ == '__main__': + unittest.main() diff --git a/tools/native_soak.py b/tools/native_soak.py new file mode 100644 index 0000000..57a801a --- /dev/null +++ b/tools/native_soak.py @@ -0,0 +1,230 @@ +#!/usr/bin/env python3 +"""Monitor one Linux process for a real eight-hour duration/liveness qualification. + +Examples (never selects robot endpoints): + python tools/native_soak.py --pid 1234 --output /tmp/soak-unique \ + --progress-file /tmp/completed.json --journal /tmp/goals.log + python tools/native_soak.py --output /tmp/qualification --duration-seconds 30 \ + --command python -u workload.py + +The workload must atomically replace progress JSON {"completed": N} after actual +business work completes. A timer/heartbeat is NOT a valid business progress metric. +Counter reset, stale progress, process exit/replacement and sampling failure fail. +An attached process is never signaled. A spawned command runs once (no shell or +restart); after monitoring it receives SIGTERM, then SIGKILL after five seconds. +Only that direct child is owned/monitored: use a foreground executor, not a launcher +or shell with detached children. Logs/resources of descendants are not aggregated. + +Exit 0: observed >=8h and progressing; NOT full DR acceptance. Resource growth and +workload validity still need review. Exit 2: short/missing-metric/interrupted run. +Exit 1: observed failure. Missing final summary means interrupted/incomplete, never +pass. No simulated/ROS clock, injectable clock or shortened acceptance threshold. +""" +import argparse +from datetime import datetime, timezone +import json +import math +import os +from pathlib import Path +import signal +import subprocess +import sys +import time + +EIGHT_HOURS = 8 * 60 * 60 + + +def verdict(elapsed, progress_observed, failure=None, interrupted=False, sampling_gap=False): + qualified = (elapsed >= EIGHT_HOURS and progress_observed and not failure + and not interrupted and not sampling_gap) + return dict(status='failed' if failure else 'duration_liveness_pass' if qualified else 'qualification_incomplete', + eight_hour_pass=qualified, no_deadlock_claim=False, + duration_liveness_pass=qualified, full_dr_acceptance=False, + resource_growth_review_required=True, + scope='Same process survival and declared business progress only; not proof of physical safety or absence of all deadlocks') + + +def positive(text): + value = float(text) + if not math.isfinite(value) or value <= 0: + raise argparse.ArgumentTypeError('must be finite and positive') + return value + + +def process_sample(pid): + base = Path('/proc') / str(pid) + raw = (base / 'stat').read_text() + fields = raw[raw.rfind(')') + 2:].split() + if fields[0] in ('Z', 'X', 'x'): + raise RuntimeError('process exited (zombie/dead)') + status = dict(line.split(':', 1) for line in (base / 'status').read_text().splitlines() if ':' in line) + result = dict(pid=pid, start_ticks=int(fields[19]), + rss_bytes=int(status['VmRSS'].split()[0]) * 1024, + threads=int(status['Threads']), fd_count=len(list((base / 'fd').iterdir())), + cpu_seconds=(int(fields[11]) + int(fields[12])) / os.sysconf('SC_CLK_TCK')) + # Detect exit/replacement during multi-file sampling as well as between samples. + again = (base / 'stat').read_text().rsplit(')', 1)[1].split() + if int(again[19]) != result['start_ticks'] or again[0] in ('Z', 'X', 'x'): + raise RuntimeError('process identity changed during sampling') + return result + + +def atomic_json(path, value): + temporary = path.with_suffix('.tmp') + with temporary.open('w') as stream: + json.dump(value, stream, indent=2, allow_nan=False) + stream.write('\n') + stream.flush() + os.fsync(stream.fileno()) + os.replace(temporary, path) + + +def main(argv=None): + parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + source = parser.add_mutually_exclusive_group(required=True) + source.add_argument('--pid', type=int) + source.add_argument('--command', nargs=argparse.REMAINDER, help='Foreground argv; must be last option; no shell') + parser.add_argument('--output', type=Path, required=True, help='New evidence directory; existing paths refused') + parser.add_argument('--duration-seconds', type=positive, default=EIGHT_HOURS) + parser.add_argument('--sample-interval-seconds', type=positive, default=5.) + parser.add_argument('--max-progress-gap-seconds', type=positive, default=300.) + parser.add_argument('--progress-file', type=Path) + parser.add_argument('--journal', type=Path, action='append', default=[]) + args = parser.parse_args(argv) + if not sys.platform.startswith('linux'): + parser.error('Linux /proc required') + if args.pid is not None and args.pid <= 0: + parser.error('--pid must be positive') + if args.command == []: + parser.error('--command requires argv') + if args.progress_file and args.sample_interval_seconds >= args.max_progress_gap_seconds: + parser.error('sample interval must be shorter than progress gap') + args.output.mkdir(parents=True, exist_ok=False) + summary_path = args.output / 'summary.json' + report = dict(schema_version=1, started_utc=datetime.now(timezone.utc).isoformat(), + duration_requested_seconds=args.duration_seconds, acceptance_duration_seconds=EIGHT_HOURS, + clock='time.monotonic (host Linux; excludes suspend)', + mode='attach' if args.pid else 'spawn', command=args.command, + progress_file=str(args.progress_file) if args.progress_file else None, + progress_contract='Atomic JSON completed counter incremented by actual workload completion', + sample_interval_seconds=args.sample_interval_seconds, + max_progress_gap_seconds=args.max_progress_gap_seconds, + boot_id=Path('/proc/sys/kernel/random/boot_id').read_text().strip(), + status='running', eight_hour_pass=False, full_dr_acceptance=False, + resource_growth_review_required=True) + atomic_json(summary_path, report) + child = None + interrupted = False + failure = None + gap = False + start = time.monotonic() + last_progress = start + previous_counter = None + progress_observed = False + previous = None + identity = None + sample_count = 0 + peaks = dict(rss_bytes=0, threads=0, fd_count=0) + old_handlers = {} + + def request_stop(signum, frame): + nonlocal interrupted + interrupted = True + + try: + for sig in (signal.SIGINT, signal.SIGTERM): + old_handlers[sig] = signal.signal(sig, request_stop) + with (args.output / 'process.log').open('w') as process_log, (args.output / 'samples.jsonl').open('w') as samples: + if args.command: + child = subprocess.Popen(args.command, stdin=subprocess.DEVNULL, stdout=process_log, stderr=subprocess.STDOUT) + pid = child.pid if child else args.pid + report['pid'] = pid + start = time.monotonic() + last_progress = start + while not interrupted: + if child and child.poll() is not None: + raise RuntimeError('process exited before observation completed: ' + str(child.returncode)) + sample = process_sample(pid) + now = time.monotonic() + sample['elapsed_seconds'] = now - start + if identity is None: + identity = sample['start_ticks'] + report['start_ticks'] = identity + if sample['start_ticks'] != identity: + raise RuntimeError('process identity changed (PID reused)') + sample['cpu_percent'] = None + if previous: + delta = sample['elapsed_seconds'] - previous['elapsed_seconds'] + sample['cpu_percent'] = 100 * (sample['cpu_seconds'] - previous['cpu_seconds']) / delta + if delta > max(5., args.sample_interval_seconds * 3): + gap = True + sample['journal_bytes'] = {str(p.resolve()): p.stat().st_size for p in args.journal} + sample['completed'] = None + if args.progress_file: + # Startup and subsequent progress share the same strict bound. + # Late evidence cannot retroactively erase an observed gap. + if now - last_progress > args.max_progress_gap_seconds: + raise RuntimeError('progress deadline exceeded; process liveness is insufficient') + try: + value = json.loads(args.progress_file.read_text())['completed'] + except FileNotFoundError: + if previous_counter is not None: + raise RuntimeError('progress metric disappeared') + value = None # Allow initial workload startup, bounded by the same deadline. + if value is not None: + if type(value) is not int or value < 0: + raise RuntimeError('progress counter must be a nonnegative integer') + if previous_counter is not None: + if value < previous_counter: + raise RuntimeError('progress counter regressed') + if value > previous_counter: + progress_observed = True + last_progress = now + previous_counter = value + sample['completed'] = value + samples.write(json.dumps(sample, allow_nan=False) + '\n') + samples.flush() + os.fsync(samples.fileno()) + sample_count += 1 + for name in peaks: + peaks[name] = max(peaks[name], sample[name]) + previous = sample + report.update(elapsed_seconds=now - start, samples=sample_count, resource_peaks=peaks, + progress_observed=progress_observed, last_completed=previous_counter) + atomic_json(summary_path, report) + if now - start >= args.duration_seconds: + break + deadline = min(now + args.sample_interval_seconds, start + args.duration_seconds) + while not interrupted and time.monotonic() < deadline: + time.sleep(max(0., min(1., deadline - time.monotonic()))) + except (Exception, KeyboardInterrupt) as exc: + failure = str(exc) or type(exc).__name__ + finally: + elapsed = previous['elapsed_seconds'] if previous else 0. + report.update(verdict(elapsed, progress_observed, failure, interrupted, gap)) + report.update(elapsed_seconds=elapsed, reason=failure or ('interrupted' if interrupted else 'monitoring completed'), + sampling_gap=gap, samples=sample_count, progress_observed=progress_observed, + resource_peaks=peaks, ended_utc=datetime.now(timezone.utc).isoformat(), + child_cleanup='not applicable; attached process untouched') + if child: + if child.poll() is None: + child.terminate() + try: + child.wait(timeout=5) + report['child_cleanup'] = 'direct child terminated after monitoring' + except subprocess.TimeoutExpired: + child.kill() + child.wait() + report['child_cleanup'] = 'direct child killed after five-second termination timeout' + else: + report['child_cleanup'] = 'direct child had already exited' + report['child_exit_code'] = child.returncode + for sig, old in old_handlers.items(): + signal.signal(sig, old) + atomic_json(summary_path, report) + print(json.dumps(report)) + return 1 if failure else 0 if report['duration_liveness_pass'] else 2 + + +if __name__ == '__main__': + raise SystemExit(main()) diff --git a/tools/physics_delivery_probe.py b/tools/physics_delivery_probe.py new file mode 100644 index 0000000..6c48ec6 --- /dev/null +++ b/tools/physics_delivery_probe.py @@ -0,0 +1,151 @@ +#!/usr/bin/env python3 +"""CPU physics fixture for delivery evidence; not robot or NAV/VLA acceptance. + +Requires pybullet==3.2.7 for the CLI; the independent predicate uses stdlib. +Objects fall under gravity into generic trays. No robot, grasp controller, +perception model, ROS endpoint or production VerifyState server is exercised. +""" +import argparse +import json +import math +from pathlib import Path +import time + + +def verify_delivery(frames, bounds, body_id, destination_id): + """Require 0.25 simulated seconds of fresh, continuous, released stability. + + Simulator truth is privileged test evidence, not a production sensor. + Bounds describe the tray interior. Whole AABB containment is conservative. + """ + def reject(reason): + return dict(verified=False, reason=reason) + try: + if not frames: + return reject('missing_evidence') + if (len(bounds) != 2 or any(len(v) != 3 for v in bounds) or + not all(math.isfinite(x) for v in bounds for x in v) or + any(bounds[0][i] >= bounds[1][i] for i in range(3))): + return reject('invalid_bounds') + previous = None + for frame in frames: + stamp = frame['time'] + if not math.isfinite(stamp) or (previous is not None and stamp <= previous): + return reject('invalid_time') + previous = stamp + end = frames[-1]['time'] + selected = [f for f in frames if f['time'] >= end - .25 - 1e-9] + if len(selected) < 2 or end - selected[0]['time'] < .25 - 1e-9: + return reject('insufficient_stability_window') + previous = None + for frame in selected: + stamp = frame['time'] + if not math.isfinite(stamp) or (previous is not None and + not 0 < stamp - previous <= .02): + return reject('discontinuous_time') + previous = stamp + if frame['body_id'] != body_id or frame['destination_id'] != destination_id: + return reject('identity_mismatch') + if frame['held'] is not False: + return reject('not_released') + if frame['floor_contact'] is not True: + return reject('no_destination_support') + speeds = [frame['linear_speed'], frame['angular_speed']] + if not all(math.isfinite(x) and x >= 0 for x in speeds): + return reject('invalid_velocity') + if speeds[0] > .02 or speeds[1] > .1: + return reject('not_stationary') + box = frame['aabb'] + if (len(box) != 2 or any(len(v) != 3 for v in box) or + not all(math.isfinite(x) for v in box for x in v) or + any(box[0][i] > box[1][i] for i in range(3))): + return reject('invalid_geometry') + if any(box[0][i] < bounds[0][i] - .001 or + box[1][i] > bounds[1][i] + .001 for i in range(3)): + return reject('outside_destination') + return dict(verified=True, reason='released_supported_contained_stable') + except (KeyError, IndexError, TypeError, ValueError, OverflowError): + return reject('invalid_evidence') + + +def run_probe(): + import pybullet as p + from importlib.metadata import version + started = time.monotonic() + client = p.connect(p.DIRECT) + cases = [] + try: + # Reset per case so no earlier object's state can satisfy later proof. + for name, xyz, steps, hold, velocity, expected in ( + ('settled_in_correct_tray', (0, 0, .25), 480, False, None, True), + ('settled_in_wrong_tray', (.7, 0, .25), 480, False, None, False), + ('outside_both_trays', (1.3, 0, .25), 480, False, None, False), + ('brief_transit_above_tray', (0, 0, .25), 30, False, None, False), + ('held_inside_tray', (0, 0, .10), 480, True, None, False), + ('moving_inside_tray', (0, 0, .10), 480, False, (.3, 0, 0), False), + ): + p.resetSimulation(physicsClientId=client) + p.setGravity(0, 0, -9.81, physicsClientId=client) + p.setTimeStep(1 / 240, physicsClientId=client) + p.setPhysicsEngineParameter(numSolverIterations=100, physicsClientId=client) + def box(half, pos, mass=0): + shape = p.createCollisionShape(p.GEOM_BOX, halfExtents=half, physicsClientId=client) + return p.createMultiBody(baseMass=mass, baseCollisionShapeIndex=shape, + basePosition=pos, physicsClientId=client) + box([2, 2, .02], [0, 0, -.04]) + floors = [] + for x in (0., .7): + floors.append(box([.21, .21, .01], [x, 0, -.01])) + for sign in (-1, 1): + box([.01, .21, .15], [x + sign * .21, 0, .15]) + box([.21, .01, .15], [x, sign * .21, .15]) + obj = box([.025, .025, .025], xyz, .1) + if hold: + p.createConstraint(obj, -1, -1, -1, p.JOINT_FIXED, [0, 0, 0], + [0, 0, 0], xyz, physicsClientId=client) + frames = [] + for step in range(steps): + # The negative moving fixture applies velocity only; no pose + # teleport or manufactured result is used as measured evidence. + if velocity and step >= steps - 60: + p.resetBaseVelocity(obj, velocity, physicsClientId=client) + p.stepSimulation(physicsClientId=client) + lin, ang = p.getBaseVelocity(obj, physicsClientId=client) + held = any(p.getConstraintInfo(p.getConstraintUniqueId(i, physicsClientId=client), + physicsClientId=client)[0] == obj + for i in range(p.getNumConstraints(physicsClientId=client))) + frames.append(dict(time=(step + 1) / 240, body_id=obj, + destination_id=floors[0], aabb=p.getAABB(obj, physicsClientId=client), + linear_speed=math.sqrt(sum(x*x for x in lin)), + angular_speed=math.sqrt(sum(x*x for x in ang)), held=held, + floor_contact=bool(p.getContactPoints(obj, floors[0], physicsClientId=client)))) + verdict = verify_delivery(frames, [[-.2, -.2, 0], [.2, .2, .3]], obj, floors[0]) + cases.append(dict(case=name, expected_verified=expected, verdict=verdict, + passed=verdict['verified'] == expected, + simulated_seconds=steps / 240, evidence_window=frames[-61:])) + return dict(scope='generic_delivery_predicate_physics_fixture', + simulator='PyBullet', version=version('pybullet'), + renderer='DIRECT', wall_seconds=time.monotonic()-started, + passed=all(c['passed'] for c in cases), cases=cases, + target_robot_acceptance=False, isaac_sim_executed=False, + production_verify_state_integrated=False, + grasp_navigation_perception_exercised=False) + finally: + p.disconnect(physicsClientId=client) + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('--output', type=Path, required=True) + args = parser.parse_args() + result = run_probe() + args.output.parent.mkdir(parents=True, exist_ok=True) + temporary = args.output.with_suffix(args.output.suffix + '.tmp') + temporary.write_text(json.dumps(result, indent=2, allow_nan=False) + '\n') + temporary.replace(args.output) + print(json.dumps({k: v for k, v in result.items() if k != 'cases'})) + raise SystemExit(0 if result['passed'] else 1) + + +if __name__ == '__main__': + main()