"""Local documentation checks only; never imports an engine or calls a service.""" import argparse from collections import Counter from datetime import datetime, timezone, timedelta import hashlib import importlib.util import json import math from pathlib import Path import re import subprocess import sys sys.dont_write_bytecode = True ROOT = Path(__file__).resolve().parents[1] REPO = Path('/workspace/contextra-analytics-runtime-lifecycle-refactor') EXPECTED_HEAD = '4c1fc6dcbdb3e1229104b6761de156239b3282b8' SOURCE_PATHS = [ 'app/fight_r5/config.py', 'app/fight_r5/web.py', 'app/fight_r5/decision.py', 'app/fight_r5/rules.py', 'app/fall_r2/config.py', 'app/fall_r2/web.py', 'app/fall_r2/core.py', 'app/fall_r2/reference.py', 'app/fall_r2/decision.py', 'app/fall_r2/geometry.py', 'app/fall_r2/base.json', 'app/fall_r2/rules.json', 'app/fall_runtime.py', 'app/scenario_parameters.py', 'app/secure_control.py', 'app/stream_settings.py', 'app/event_cooldown.py', 'docs/integration/FIGHT_R5_WEB_CONFIG.ko.md', 'docs/integration/FALL_R2_WEB_CONFIG.ko.md', 'docs/deployment/SCENARIO_DEMO_COMPATIBILITY.ko.md', 'docs/profiles/ltb1021/topics/runtime-lifecycle/README.ko.md', ] def strict_load(text): def pairs(items): result = {} for key, value in items: if key in result: raise ValueError('duplicate JSON key') result[key] = value return result def nonfinite(value): raise ValueError('nonfinite JSON token') return json.loads(text, object_pairs_hook=pairs, parse_constant=nonfinite) def read_json(path): return strict_load(path.read_text(encoding='utf-8')) def git(*args): return subprocess.check_output( ['git', '--no-optional-locks', '-C', str(REPO), *args], text=True).strip() def fingerprints(): return {path: hashlib.sha256((REPO / path).read_bytes()).hexdigest() for path in SOURCE_PATHS} def load_config(name, path): spec = importlib.util.spec_from_file_location(name, REPO / path) module = importlib.util.module_from_spec(spec) sys.modules[name] = module spec.loader.exec_module(module) return module def finite_number(value, minimum, maximum): return (type(value) in (int, float) and math.isfinite(value) and minimum <= value <= maximum) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--write-results', action='store_true') args = parser.parse_args() output_paths = [ROOT / 'validation/context.json', ROOT / 'validation/result.json'] if args.write_results and any(path.exists() for path in output_paths): raise SystemExit('Existing evidence: use read-only mode; do not overwrite.') before = fingerprints() before_git = dict(branch=git('branch', '--show-current'), head=git('rev-parse', 'HEAD'), status=git('status', '--short', '--untracked-files=all')) rows = [] pending = [] def check(group, name, ok, detail=None): row = dict(group=group, name=name, passed=bool(ok)) if detail is not None: row['detail'] = detail rows.append(row) fight = load_config('packet_current_fight_config', 'app/fight_r5/config.py') fall = load_config('packet_current_fall_config', 'app/fall_r2/config.py') cases = read_json(ROOT / 'examples/validation_cases.json') controls = read_json(ROOT / 'contracts/controls.json') for case in cases['fight_current']: current = fight.R5Config(**case.get('stored', {})) try: value = current.patched(case['patch']).payload() outcome = 'accept' except fight.R5ConfigError as exc: outcome = str(exc) value = None check('current_fight_validator', case['id'], outcome == case['expected'], outcome) if value and case['id'] == 'partial_preserves_saved': check('current_fight_validator', 'hidden_and_window_preserved', value['contact_min_samples'] == 5 and value['window_seconds'] == 1.5) if value and case['id'] == 'legacy_reset_is_not_new_ui': check('current_fight_validator', 'existing_reset_defaults', value == fight.R5Config().payload()) for case in cases['fall_current']: try: fall.block(case['block'], request=True) outcome = 'accept' except fall.R2ConfigError as exc: outcome = str(exc) check('current_fall_validator', case['id'], outcome == case['expected'], outcome) # This checks the written proposal, not an implementation of the future API. def proposed_angles_valid(block): if type(block) is not dict or set(block) - ({'schema_version'} | set(controls['fall']['controls'])): return False if type(block.get('schema_version')) is not int or block['schema_version'] != 1: return False return all(finite_number(value, controls['fall']['controls'][key]['minimum'], controls['fall']['controls'][key]['maximum']) for key, value in block.items() if key != 'schema_version') for case in cases['fall_design_only']: if case['expected'] == 'pending_main_decision': pending.append(dict(id=case['id'], status='not_decided_not_tested')) continue outcome = 'accept' if proposed_angles_valid(case['block']) else 'reject' check('fall_design_only_not_runtime', case['id'], outcome == case['expected'], outcome) for key in controls['fall']['controls']: for label, value in [('nan', float('nan')), ('infinity', float('inf'))]: check('fall_design_only_not_runtime', f'{key}_{label}', not proposed_angles_valid({'schema_version': 1, key: value})) for label, value in [('nan', float('nan')), ('infinity', float('inf'))]: try: fight.R5Config().patched({'schema_version': 1, 'window_seconds': value}) outcome = 'accept' except fight.R5ConfigError as exc: outcome = str(exc) check('current_fight_validator', f'window_{label}', outcome == 'FIGHT_R5_WINDOW_INVALID') for name, module in [('fight', fight), ('fall', fall)]: for minimum in [.25, .10]: for label, value, expected in [('low', 0., minimum), ('equal', minimum, minimum), ('higher', .45, .45), ('upper', 1., 1.)]: check('current_score_validator', f'{name}_{minimum}_{label}', module.score_floor(value, minimum) == expected) for label, value in [('negative', -.01), ('over_one', 1.01), ('bool', True), ('string', '0.35'), ('null', None), ('nan', float('nan')), ('infinity', float('inf'))]: error = False try: module.score_floor(value, minimum) except ValueError: error = True check('current_score_validator', f'{name}_{minimum}_{label}', error) for path in sorted((ROOT / 'examples/CURRENT').glob('*.json')): params = read_json(path) engine = 'fight_r5' if path.name.startswith('FIGHT') else 'fall_r2' check('example_shape', path.name, not (set(params) - {engine, 'person_confidence', 'pose_confidence', 'cooldown_seconds'})) if engine == 'fight_r5': fight.R5Config().patched(params[engine]) check('example_current', path.name, 'contact_min_samples' not in params[engine]) else: fall.block(params[engine], request=True) check('example_current', path.name, True) for key in ('person_confidence', 'pose_confidence', 'cooldown_seconds'): if key in params: definition = controls['common']['controls'][key] check('example_common_numeric', path.name + ':' + key, finite_number(params[key], definition['minimum'], definition['maximum'])) for path in sorted((ROOT / 'examples/PLANNED_DO_NOT_SEND').glob('*.json')): params = read_json(path) check('example_shape', path.name, set(params) == {'fall_r2'}) check('example_future_design_only', path.name, proposed_angles_valid(params['fall_r2'])) try: fall.block(params['fall_r2'], request=True) outcome = 'accept' except fall.R2ConfigError as exc: outcome = str(exc) check('example_current_expected_rejection', path.name, outcome == 'FALL_R2_FIELDS_INVALID', outcome) base = read_json(REPO / 'app/fall_r2/base.json') rules = read_json(REPO / 'app/fall_r2/rules.json') defaults = dict(r1_torso_min_deg=base['torso_down_deg'], r1_leg_min_deg=base['leg_down_deg'], d1_knee_min_deg=rules['D1']['knee_angle_min']) check('source_defaults', 'fall_three_constants', all(defaults[k] == d['default'] for k, d in controls['fall']['controls'].items())) check('source_defaults', 'other_constants_unchanged', base['thigh_only_down_deg'] == 65 and rules['D1']['box_aspect_min'] == 1.1 and rules['D2']['box_aspect_min'] == 1.1 and rules['D1']['torso_angle_min'] == 20) fight_default = fight.R5Config().payload() check('source_defaults', 'fight_visible_defaults', all(fight_default[k] == d['default'] for k, d in controls['fight']['controls'].items())) check('source_defaults', 'fight_hidden_default', fight_default['contact_min_samples'] == 3) for path in sorted(ROOT.rglob('*.json')): read_json(path) check('json_syntax', str(path.relative_to(ROOT)), True) for path in sorted(ROOT.rglob('*.md')): text = path.read_text(encoding='utf-8') check('markdown_format', str(path.relative_to(ROOT)), text.endswith('\n') and all(line == line.rstrip() for line in text.splitlines())) for i, snippet in enumerate(re.findall(r'```json\s*\n(.*?)\n```', text, flags=re.S)): strict_load(snippet) check('markdown_json', f'{path.name}:{i}', True) for match in re.finditer(r'\[[^\]]+\]\(([^)]+)\)', text): target = match.group(1) if target.startswith(('https://', 'http://', '#')): continue target = re.sub(r':\d+$', '', target.split('#', 1)[0]) resolved = (path.parent / target).resolve() generated = resolved in set(output_paths + [ROOT / 'manifest.json']) check('local_link', path.name + ':' + target, resolved.exists() or generated, 'generated_output_verified_after_write' if generated and not resolved.exists() else 'exists') after = fingerprints() after_git = dict(branch=git('branch', '--show-current'), head=git('rev-parse', 'HEAD'), status=git('status', '--short', '--untracked-files=all')) check('read_only_boundary', 'source_fingerprints_unchanged', before == after) check('read_only_boundary', 'git_unchanged_during_checks', before_git == after_git) check('read_only_boundary', 'draft_source_head', before_git['head'] == EXPECTED_HEAD) check('read_only_boundary', 'named_branch', before_git['branch'] == 'task/runtime-lifecycle-refactor') check('read_only_boundary', 'no_html_video_or_cache_in_packet', not any(p.suffix.lower() in {'.html', '.mp4', '.onnx', '.pt', '.pyc'} for p in ROOT.rglob('*'))) context = dict( checked_at_kst=datetime.now(timezone(timedelta(hours=9))).isoformat(), profile='ltb1021', server='test', repo=str(REPO), artifact_root=str(ROOT), git_before=before_git, git_after=after_git, source_sha256=before, artifact_kind='user_requested_handoff_draft_snapshot', repo_mutation=False, loaded_modules=['app/fight_r5/config.py', 'app/fall_r2/config.py'], network_runtime_engine_called=False) groups = Counter(row['group'] for row in rows) result = dict(status='passed' if all(row['passed'] for row in rows) else 'failed', performed_checks=len(rows), passed_checks=sum(row['passed'] for row in rows), group_counts=dict(groups), pending=pending, checks=rows, design_is_implementation=False, not_run=['Start/API', 'Runtime/engine', 'DB/save/restart/no-op', 'SSH/site/Web', 'cache/model/Pose/Tracking', 'HTML/video', 'deployment']) if args.write_results: for path, payload in zip(output_paths, [context, result]): with path.open('x', encoding='utf-8') as stream: stream.write(json.dumps(payload, ensure_ascii=False, indent=2, allow_nan=False) + '\n') print(json.dumps(dict(status=result['status'], performed_checks=len(rows), passed_checks=result['passed_checks'], group_counts=dict(groups), pending=pending, failures=[r for r in rows if not r['passed']]), ensure_ascii=False, indent=2)) return 0 if result['status'] == 'passed' else 1 if __name__ == '__main__': raise SystemExit(main())