"""Finite owner-routed natural request-algorithm windows, not allocator control.""" import math import json import time from . import native from .debug_runtime import NativeRejected, OperationUncertain from .teleport_observation import rejection_boundary_verified def validate(args): if args.get('algorithm', 'approach') != 'approach': raise ValueError('only approach is supported; no runtime node identity is exposed') actors = args.get('actors') if (not isinstance(actors, list) or not 1 <= len(actors) <= 8 or any(not isinstance(ref, str) for ref in actors)): raise ValueError('1..8 distinct generation-qualified actors required') for ref in actors: native.validate('water_depth_actor', ref) parsed = [tuple(map(int, ref.split('|'))) for ref in actors] if len(set(parsed)) != len(parsed): raise ValueError('actors must be distinct') if len({generation for generation, _ in parsed}) != 1: raise ValueError('actors must share a generation') mode = args.get('mode', 'observe') scores = args.get('scores', []) if mode not in ('observe', 'score'): raise ValueError('mode must be observe or score; count mode is unavailable') if mode == 'observe' and scores: raise ValueError('observe accepts no replacement scores') if mode == 'score' and (not isinstance(scores, list) or len(scores) != len(actors) or any(type(v) not in (int, float) or not math.isfinite(v) or not .000001 <= v <= 1000000 for v in scores)): raise ValueError('score requires exactly one finite positive output <=1000000 per actor, in actor order') def verify_status(status, generation, window, actors, inactive=False): if (not isinstance(status, dict) or status.get('generation') != generation or status.get('window_id') != window or status.get('actor_ids') != actors or status.get('instance_count_fields_modified') is not False or status.get('ticket_sources_modified') is not False or (inactive and (status.get('active') is not False or status.get('armed') is not False))): raise OperationUncertain('request_window_status_unverified', dict(status=status)) def collect(runtime, generation, window, final): records, pages, gaps = [], [], [] total = final.get('records') if type(total) is not int or not 0 <= total <= 128 or type(final.get('dropped')) is not int: raise ValueError('request_record_count_invalid') while len(records) < total: page = runtime.call('request_algorithm_page', f'{generation}|{window}|{len(records)}')['data'] rows = page.get('records') if (page.get('generation') != generation or page.get('window_id') != window or page.get('start') != len(records) or page.get('total') != total or page.get('dropped') != final['dropped'] or not isinstance(rows, list) or not 1 <= len(rows) <= min(4, total-len(records))): raise ValueError('request_page_identity_or_accounting_invalid') for row in rows: expected = len(records)+1 if row.get('sequence') != expected: gaps.append(dict(expected=expected, actual=row.get('sequence'))) if (row.get('generation') != generation or row.get('window_id') != window or row.get('actor_id') not in final['actor_ids']): raise ValueError('request_record_identity_invalid') if type(row.get('algorithm_instance')) is not int or not 1 <= row['algorithm_instance'] <= 128: raise ValueError('request_algorithm_instance_invalid') if len(json.dumps(row, separators=(',', ':'), allow_nan=False).encode()) > 1024: raise ValueError('request_record_size_exceeded') tag, tag_status = row.get('binding_tag'), row.get('binding_tag_status') if tag_status == 'invalid_size': if tag is not None: raise ValueError('request_binding_tag_invalid') elif (tag_status not in ('declared', 'untagged') or not isinstance(tag, str) or len(json.dumps(tag, ensure_ascii=True).encode()) > 66 or (tag_status == 'untagged') != (tag == '')): raise ValueError('request_binding_tag_invalid') for field in ('original_score', 'returned_score'): if type(row.get(field)) not in (int, float) or not math.isfinite(row[field]): raise ValueError('request_score_invalid') original, returned = row['original_score'], row['returned_score'] if ((original <= 0 or final['mode'] == 'observe') and returned != original): raise ValueError('request_original_output_not_preserved') if final['mode'] == 'score' and original > 0: score = final['scores'][final['actor_ids'].index(row['actor_id'])] if not math.isclose(returned, score, rel_tol=1e-6, abs_tol=1e-6): raise ValueError('request_replacement_mismatch') records.append(row) pages.append(page) return dict(records=records, pages=pages, gaps=gaps, complete=not gaps and final['dropped'] == 0, dropped=final['dropped']) def run(runtime, args): before = runtime.paused() generation = int(args['actors'][0].split('|')[0]) actors = [int(ref.split('|')[1]) for ref in args['actors']] if before['generation'] != generation: raise ValueError('stale_generation: request window not armed') if before['data']['world']['player']['alive'] is False: return dict(ok=True, stop_reason='player_death', armed=False, before=before) # A monotonic native lease is independent of simulation ticks and broker life. milliseconds = max(1, min(60000, int((runtime.work_deadline-time.monotonic())*1000))) scores = args.get('scores') or [0] * len(actors) payload = '|'.join(map(str, [ generation, ','.join(map(str, actors)), args['ticks'], milliseconds, args.get('mode', 'observe'), ','.join(f'{v:.6f}' for v in scores)])) arm = runtime.call('request_algorithm_arm', payload) status = arm.get('data') window = status.get('window_id') if isinstance(status, dict) else None if type(window) is not int or window < 1: raise OperationUncertain('request_arm_identity_unverified', dict(arm=arm)) verify_status(status, generation, window, actors) if status.get('active') is not True: raise OperationUncertain('request_arm_not_active', dict(arm=arm)) value = failure = stop = boundary = evidence = None cleanup_errors = [] try: def progress(): current = runtime.call('request_algorithm_status')['data'] verify_status(current, generation, window, actors) return current.get('dropped') == 0 value = runtime.block(args['ticks'], drain=progress) except Exception as error: failure = error finally: previous = runtime.cleaning runtime.cleaning = True try: if failure: try: boundary = runtime.halt() except Exception as error: cleanup_errors.append(str(error)) try: stop = runtime.call('request_algorithm_stop', f'{generation}|{window}') verify_status(stop['data'], generation, window, actors, inactive=True) evidence = collect(runtime, generation, window, stop['data']) except Exception as error: cleanup_errors.append(str(error)) finally: runtime.cleaning = previous audit = dict(arm=arm, stop=stop, block=value, collection=evidence, cleanup_boundary=boundary, cleanup_errors=cleanup_errors, error=str(failure) if failure else None) if cleanup_errors: raise OperationUncertain('request_window_cleanup_unverified', audit) from failure if failure: if (isinstance(failure, NativeRejected) and evidence['complete'] and rejection_boundary_verified(failure, boundary, generation)): failure.evidence.update(audit) raise failure raise OperationUncertain('request_window_outcome_unverified', audit) from failure return dict(value, request_algorithm=dict( **audit, runtime_restoration_verified=True, source_modified=False, binding_marker_semantics='a declared tag identifies this naturally evaluated instance for this actor; not native resource identity; missing callbacks are unknown', scope='actor_and_instrumented_algorithm_not_node', effective_requested_count=None, final_allocator_importance=None, native_qualification='unverified'))