| |
| """CPU-only parsers and owned-process supervisor for coordinator window scripts. |
| |
| No remote access occurs except in explicit `supervise`, used only by --execute. |
| Raw evidence stays in the private campaign directory. Summaries contain aggregates. |
| """ |
| from __future__ import annotations |
|
|
| import argparse |
| import csv |
| import hashlib |
| import json |
| import math |
| import os |
| from pathlib import Path |
| import re |
| import selectors |
| import signal |
| import statistics |
| import subprocess |
| import sys |
| import time |
| from typing import Any, IO, cast |
|
|
| MEM_FLOOR_KB = 5_242_880 |
| PAIR = ('GPU-c4e7a371-d8a8-ebf0-918c-da3e38c87a93', |
| 'GPU-7744b44b-8a13-b06d-a67e-f958fd74b8fc') |
| REQUIRED: dict[str, set[str]] = { |
| 'preflight': {'exclusive_window', 'production_drained', 'rollback_saved', 'd2_paused', |
| 'source_pins', 'image_base_hashes', 'owned_processes'}, |
| 'idle': {'experiment_idle', 'no_foreign_requests'}, |
| 'runtime': {'both_rank_hashes', 'model_manifest', 'worker_identity', 'hello_identity', |
| 'rank_env_equal', 'capacity_async_process', 'proxy_torch_absent', 'proxy_mlock', |
| 'local_drafter_absent', 'engram_required', 'kv_layout', 'no_runtime_errors'}, |
| 'clean': {'both_rank_logs', 'counters_before_after', 'no_fallback', 'no_watchdog', |
| 'no_nan', 'no_provenance_stops', 'no_swap_growth', 'healthy'}, |
| 'worker': {'pair_owner_coord', 'owned_transient_only', 'both_gpu_locks', 'source_hashes', |
| 'arm_identity', 'ready_pid', 'four_slots', 'max_length', 'no_emulation', 'no_dumps'}, |
| 'pair-results': {'fresh_results', 'both_arm_soaks', 'dump_identity', 'v1_classified', |
| 'rx_gates', 'pair_timing', 'both_card_peaks', 'runtime_identity'}, |
| 'recalibrate': {'pair_source_matches', 'verify_only_round_ms', 'overlap_audited', |
| 'fit_heldout', 'config_staged', 'both_rank_costs'}, |
| 'up1-prereq': {'k1_qualified', 's18_reviewed'}, |
| 'up1-p0': {'control_k1_equivalent', 'cc_noise', |
| 'short_tokens_identical', 'request_accounting', 'round_within_one_percent'}, |
| 'split-bitwise': {'rank0_rows_33_42_43_63', 'rank1_rows_33_42_43_63', |
| 'per_row_oracle_43_63', 'dispatch_controls_32_64', 'no_host_count_sync'}, |
| 'cpu-pin': {'actual_thread_affinities', 'allowed_cpuset', 'poller_verified', 'proxy_engram_tails'}, |
| 'replay': {'successor_gt_128_both_sites', 'remainders_1_127_128_129', 'partial_budget_le_128', |
| 'retained_l20_kv', 'actual_suffix', 'partial_need_draft_zero', 'full_draft_window', |
| 'cancel_reuse', 'preempt_prefix', 'reconnect_timeout_resync', 'ragged_reject', |
| 'mm_exclusion', 'two_prompt_logprobs', 'final_logits_needles', 'fused_torch_equal', |
| 'zero_provenance_strings', 'zero_internal_error_increment'}, |
| 'k5': {'five_proposals', 'twentyfour_target_rows', 'accepted_round_stats', 'prose_review_tool'}, |
| 'dual-gid': {'rank0_both_ipv4_rocev2_gid3', 'rank1_both_ipv4_rocev2_gid3'}, |
| 'dual-ring': {'collective_byte_identity', 'both_functions_traffic', 'no_nccl_fallback'}, |
| 'vor': {'valid_lens_index_bitwise', 'padded_invalid_replay_300k', 'stock_exclusions', |
| 'no_exception_disable', 'eligible_once_per_step', 'copy_on_change', 'no_request_jit'}, |
| 'rp1-preflight': {'k1_qualified', 'base20_both_images', 'checkpoint_manifest_both', |
| 'cpu_topology_allowed', 'odirect', 'disk_budget_8g', 'cpu_bytes_256m'}, |
| 'rp1-p0': {'no_connector_cli', 'no_rp1_switches', 'same_k1_pool_groups', 'same_fixed_outputs'}, |
| 'rp1-rank': {'independent_preload_payloads', 'both_ranks_store', 'valid_ranges_recorded', |
| 'rank_sha_crc_equal', 'gloo_init', 'cap_29568', 'cpu_bytes_256m'}, |
| 'rp1-restart': {'stores_complete', 'old_engine_recorded', 'old_processes_gone', |
| 'unused_shm_only', 'disk_preserved'}, |
| 'rp1-empty': {'new_engine_empty', 'no_intervening_salts', 'new_engine_id'}, |
| 'rp1-stale': {'new_rank1_owned_mmap', 'bounded_64k_sentinel', 'rank0_untouched', |
| 'mn_verify_off', 'sync_before_gpu', 'selected_payload_equal', 'codes_logits_pass'}, |
| 'rp1-crc': {'stopped_owned_namespace', 'aligned_odirect_flip', 'exact_crc_mismatch', |
| 'block_sidecar_removed', 'no_invalid_completion', 'recompute_pass'}, |
| 'rp1-sidecar': {'stopped_owned_namespace', 'recorded_sidecar_only', 'miss_removed', 'recompute_pass'}, |
| 'rp1-namespace': {'model_and_layout_mismatch', 'old_namespace_unread', 'cached_zero', |
| 'true_identity_restored', 'no_false_namespace_writes'}, |
| 'rp1-concurrency': {'four_distinct_150k_saved_salts', 'empty_restart', 'four_reload_lt_300s', |
| 'each_load_le_29568', 'codes_six_of_six', 'no_foreign_codes', |
| 'no_pending_loop', 'explicit_million_clamped', 'file_gloo_counters'}, |
| 'rp1-reload-evidence': {'cold_fresh_cached_zero', 'reload_positive_le_cap', |
| 'file_reads_and_gloo_increase', 'all_compare_pass', 'no_ambiguous_logits'}, |
| 'depth': {'sp2_configs', 'matching_target_worker', 'max_depth_five', 'same_sampling_budget', |
| 'costs_measured', 'overlap_audited', 'heldout_groups', 'decisions_logged'}, |
| } |
|
|
|
|
| def require(condition: bool, message: str) -> None: |
| if not condition: |
| raise ValueError(message) |
|
|
|
|
| def load(path: Path) -> Any: |
| return json.loads(path.read_text()) |
|
|
|
|
| def kv(text: str, expected: int) -> int: |
| values = [int(x.replace(',', '')) for x in re.findall(r'GPU KV cache size: ([\d,]+)', text)] |
| require(expected > 0 and bool(values), 'missing GPU KV cache size or invalid expectation') |
| require(all(x == expected for x in values), f'KV token mismatch: {values}, expected {expected}') |
| return values[0] |
|
|
|
|
| def memory(text: str) -> tuple[int, int]: |
| fields = dict(re.findall(r'^(MemAvailable|SwapFree):\s*(\d+)\s+kB\s*$', text, re.M)) |
| require(set(fields) == {'MemAvailable', 'SwapFree'}, 'missing/malformed memory fields') |
| available, swap = int(fields['MemAvailable']), int(fields['SwapFree']) |
| require(available >= MEM_FLOOR_KB, f'MemAvailable {available} < {MEM_FLOOR_KB} kB') |
| return available, swap |
|
|
|
|
| def dual(text: str, rank: int) -> None: |
| prefix = f'EXL3 RDMA all-reduce: rank {rank} of 2 on rocep1s0f1,roceP2p1s0f1' |
| require(any(line.startswith(prefix) and re.match(r'(?:,|$)', line[len(prefix):]) |
| for line in text.splitlines()), f'missing exact dual-ring rank {rank} log prefix') |
| require(not re.search(r'RDMA all-reduce off.*NCCL stays', text), 'NCCL-only fallback') |
|
|
|
|
| def clean_logs(text: str) -> None: |
| for stop in ('invalid replay tap provenance', 'replay skip would remove taps', |
| 'RDMA all-reduce off', 'CUDA error', 'illegal memory access', 'Traceback (most recent call last)'): |
| require(stop not in text, f'log stop: {stop}') |
|
|
|
|
| def counter_delta(before: dict[str, Any], after: dict[str, Any]) -> None: |
| |
| require(isinstance(before.get('counters'), dict) and isinstance(after.get('counters'), dict), |
| 'missing counters map') |
| require(after.get('uptime_s', -1) >= before.get('uptime_s', 0), 'stats reset during arm') |
| for key in set(before['counters']) | set(after['counters']): |
| a, b = before['counters'].get(key, 0), after['counters'].get(key, 0) |
| require(isinstance(a, (int, float)) and isinstance(b, (int, float)) and b >= a, |
| f'counter reset/malformed: {key}') |
| if any(s in key for s in ('fallback', 'failed', 'error', 'watchdog', 'preempted', 'skipped')): |
| require(b == a, f'new failure counter: {key}: {a}->{b}') |
| for stats in (before, after): |
| if 'async_watchdog' in stats: |
| require(stats['async_watchdog'].get('fires') == 0, 'async watchdog fired') |
|
|
|
|
| def receipt(path: Path, gate: str, arm: str, root: Path) -> None: |
| require(gate in REQUIRED, f'unknown gate {gate}') |
| data = load(path) |
| require(data.get('schema') == 'wr1-gate-v1', 'wrong receipt schema') |
| require(data.get('gate') == gate and data.get('arm') == arm, 'receipt gate/arm mismatch') |
| require(data.get('run_dir') == str(root.resolve()), 'receipt belongs to another run') |
| checks = data.get('checks', {}) |
| require(isinstance(checks, dict) and all(value is True for value in checks.values()), 'receipt contains a nonpassing check') |
| require(all(checks.get(key) is True for key in REQUIRED[gate]), |
| f'missing/failed {gate} checks: {sorted(key for key in REQUIRED[gate] if checks.get(key) is not True)}') |
| require(bool(data.get('evidence')), 'receipt requires hashed local source evidence') |
| for row in data['evidence']: |
| source = Path(row['path']).resolve() |
| require(source.is_relative_to(root.resolve()) and source != path.resolve(), 'evidence outside run or self-reference') |
| require(source.is_file() and source.stat().st_mtime >= (root / '.run-start').stat().st_mtime - 2, |
| f'missing/stale evidence {source}') |
| require(hashlib.sha256(source.read_bytes()).hexdigest() == row['sha256'], 'evidence SHA256 mismatch') |
| print(f'PASS {gate} {arm}: {len(REQUIRED[gate])} checks, source evidence hashes verified') |
|
|
|
|
| def timing(path: Path) -> None: |
| data = load(path) |
| require(data.get('classification') == 'MEASURED', 'timing must be MEASURED from pair + verify-only evidence') |
| require(data.get('arm') == 'native-hybrid-kx2', 'wrong timing arm') |
| require(data.get('overlap_audited') is True and data.get('verify_only') is True, 'unverified cost semantics') |
| draft, rounds = data['draft_ms'], data['round_ms'] |
| require(isinstance(draft, (int, float)) and math.isfinite(draft) and draft >= 0, 'invalid draft_ms') |
| require(isinstance(rounds, list) and len(rounds) == 6, 'round_ms must contain depths 0..5') |
| require(all(isinstance(x, (int, float)) and math.isfinite(x) and x > 0 for x in rounds), 'invalid round_ms') |
| require(all(a <= b for a, b in zip(rounds, rounds[1:])), 'nonmonotonic verify-only costs') |
| for row in data['sources']: |
| source = Path(row['path']) |
| require(hashlib.sha256(source.read_bytes()).hexdigest() == row['sha256'], 'timing source hash mismatch') |
| require(bool(data['sources']), 'timing source evidence missing') |
| print(f'MEASURED timing from {path}: DRAFT_MS={draft}, ROUND_MS={",".join(map(str, rounds))}') |
|
|
|
|
| def bench(path: Path) -> list[dict[str, Any]]: |
| records = [json.loads(line) for line in path.read_text().splitlines() if line.strip()] |
| runs = [r for r in records if r.get('type') == 'run'] |
| requests = [r for r in records if r.get('type') == 'request'] |
| require(bool(runs) and bool(requests), 'missing run/request measurements') |
| for row in records: |
| require(not row.get('error'), f'benchmark error in {path}') |
| result: list[dict[str, Any]] = [] |
| for run in runs: |
| rows = [r for r in requests if r.get('run_id') == run['run_id']] |
| require(len(rows) == run['request_count'] and bool(rows), 'request accounting mismatch') |
| spec = run['spec'] |
| require(spec.get('rounds', 0) > 0 and spec.get('mean_accepted_per_round') is not None, 'missing accepted/round') |
| for row in rows: |
| require(row.get('ttft_s') is not None and row.get('step_gap_ms', {}).get('p50') is not None, |
| 'missing TTFT/step p50') |
| require(not run.get('telemetry_errors'), 'benchmark telemetry error') |
| result.append({'classification': 'MEASURED', 'source': str(path), 'label': run['config_label'], |
| 'ctx': run['target_ctx'], 'concurrency': run['concurrency'], 'mixed': run['mixed'], |
| 'cache': run['cache_state'], 'rep': run['rep'], 'delivered_tps': run['aggregate_tps'], |
| 'accepted_per_round': spec['mean_accepted_per_round'], |
| 'step_p50_ms': statistics.median(r['step_gap_ms']['p50'] for r in rows), |
| 'ttft_s': statistics.median(r['ttft_s'] for r in rows), |
| 'first_visible_s': [r.get('first_visible_s') for r in rows]}) |
| return result |
|
|
|
|
| def exit_status(proc: subprocess.Popen[Any]) -> int | None: |
| |
| status = os.waitid(os.P_PID, proc.pid, os.WEXITED | os.WNOHANG | os.WNOWAIT) |
| if status is None: |
| return None |
| return status.si_status if status.si_code == os.CLD_EXITED else -status.si_status |
|
|
|
|
| def stop_owned(proc: subprocess.Popen[Any]) -> None: |
| |
| if proc.returncode is not None: |
| return |
| for sig in (signal.SIGTERM, signal.SIGKILL): |
| try: |
| os.killpg(proc.pid, sig) |
| except ProcessLookupError: |
| pass |
| if sig == signal.SIGTERM: |
| time.sleep(0.1) |
| proc.wait(timeout=3) |
|
|
|
|
| def supervise(args: argparse.Namespace) -> int: |
| root, name = args.out, args.name |
| require(not (root / 'STOP').exists(), 'STOP file exists') |
| select = selectors.DefaultSelector() |
| monitors: list[subprocess.Popen[str]] = [] |
| outputs: list[IO[str]] = [] |
| command: subprocess.Popen[Any] | None = None |
| baseline_path = root / 'swap-baseline.json' |
| initial: dict[str, int] = load(baseline_path) if baseline_path.exists() else {} |
| last: dict[str, float] = {} |
| deadline = time.monotonic() + 20 |
| def interrupted(signum: int, _frame: Any) -> None: |
| raise RuntimeError(f'supervisor interrupted by signal {signum}') |
| old_handlers = {s: signal.signal(s, interrupted) for s in (signal.SIGTERM, signal.SIGINT, signal.SIGHUP)} |
| try: |
| ssh = ['ssh', '-o', 'BatchMode=yes', '-o', 'ConnectTimeout=5', '-o', 'ServerAliveInterval=5', '-o', 'ServerAliveCountMax=2'] |
| memscript = "import time; from pathlib import Path\nwhile True:\n print(';'.join(x for x in Path('/proc/meminfo').read_text().splitlines() if x.startswith(('MemAvailable:', 'SwapFree:'))), flush=True); time.sleep(1)" |
| import shlex |
| memcmd = 'exec python3 -u -c ' + shlex.quote(memscript) |
| specs = [(host, memcmd, 'mem') for host in args.hosts] |
| specs.append((args.x86host, f'nvidia-smi -i {",".join(PAIR)} --query-gpu=timestamp,uuid,memory.used,power.draw,temperature.gpu --format=csv,noheader -l 1', 'gpu')) |
| for host, script, kind in specs: |
| monitor_key = f'{kind}-{host}' |
| log = (root / f'{name}.{monitor_key}.log').open('x') |
| err = (root / f'{name}.{monitor_key}.stderr').open('x') |
| outputs.extend([log, err]) |
| proc = subprocess.Popen([*ssh, host, script], stdout=subprocess.PIPE, stderr=err, |
| text=True, start_new_session=True, bufsize=1) |
| monitors.append(proc) |
| assert proc.stdout is not None |
| os.set_blocking(proc.stdout.fileno(), False) |
| select.register(proc.stdout, selectors.EVENT_READ, (kind, host, log)) |
| pending = {key.fileobj: '' for key in select.get_map().values()} |
| while True: |
| now = time.monotonic() |
| if (root / 'STOP').exists(): |
| raise RuntimeError('coordinator STOP file present') |
| for key, _mask in select.select(timeout=0.1): |
| stream = cast(IO[str], key.fileobj) |
| chunk = os.read(stream.fileno(), 65536).decode() |
| require(bool(chunk), 'telemetry stream closed') |
| pending[stream] += chunk |
| while '\n' in pending[stream]: |
| line, pending[stream] = pending[stream].split('\n', 1) |
| kind, host, log = key.data |
| log.write(f'{time.time():.3f} {line}\n'); log.flush() |
| if kind == 'mem': |
| available, swap = memory(line.replace(';', '\n')) |
| initial.setdefault(host, swap) |
| baseline_path.write_text(json.dumps(initial) + '\n') |
| require(swap >= initial[host], f'swap growth on {host}') |
| last[f'{kind}-{host}'] = now |
| else: |
| columns = next(csv.reader([line], skipinitialspace=True)) |
| require(len(columns) == 5 and columns[1] in PAIR, 'invalid GPU telemetry/UUID') |
| require(float(columns[2]) >= 0, 'invalid GPU memory') |
| last[columns[1]] = now |
| require(all(exit_status(p) is None for p in monitors), 'telemetry process exited') |
| expected = [f'mem-{h}' for h in args.hosts] + list(PAIR) |
| if command is None: |
| require(now < deadline, 'no initial telemetry within 20 s') |
| if all(k in last for k in expected): |
| cmd = args.command[1:] if args.command[0] == '--' else args.command |
| command = subprocess.Popen(cmd, start_new_session=True) |
| deadline = now + args.timeout |
| else: |
| require(now < deadline, f'step exceeded {args.timeout}s') |
| require(all(now - last[k] < 15 for k in expected), 'telemetry stale >15s') |
| rc = exit_status(command) |
| if rc is not None: |
| require(rc == 0, f'command exited rc={rc}') |
| return 0 |
| except (Exception, KeyboardInterrupt) as exc: |
| (root / 'STOP').write_text(f'{name}: {exc}\n') |
| print(f'STOP {name}: {exc}', file=sys.stderr) |
| return 1 |
| finally: |
| if command is not None: |
| stop_owned(command) |
| for proc in monitors: |
| stop_owned(proc) |
| select.close() |
| for output in outputs: |
| output.close() |
| for sig, handler in old_handlers.items(): |
| signal.signal(sig, handler) |
|
|
|
|
| def main() -> int: |
| parser = argparse.ArgumentParser(description=__doc__) |
| sub = parser.add_subparsers(dest='mode', required=True) |
| p = sub.add_parser('kv'); p.add_argument('path', type=Path); p.add_argument('expected', type=int) |
| p = sub.add_parser('dual'); p.add_argument('path', type=Path); p.add_argument('rank', type=int, choices=[0, 1]) |
| p = sub.add_parser('logs'); p.add_argument('path', type=Path) |
| p = sub.add_parser('counters'); p.add_argument('before', type=Path); p.add_argument('after', type=Path) |
| p = sub.add_parser('receipt'); p.add_argument('path', type=Path); p.add_argument('gate'); p.add_argument('arm'); p.add_argument('root', type=Path) |
| p = sub.add_parser('timing'); p.add_argument('path', type=Path) |
| p = sub.add_parser('rollback'); p.add_argument('path', type=Path) |
| p = sub.add_parser('bench'); p.add_argument('path', type=Path) |
| p = sub.add_parser('summary'); p.add_argument('root', type=Path); p.add_argument('--pair', action='store_true') |
| p = sub.add_parser('supervise'); p.add_argument('--out', type=Path, required=True); p.add_argument('--name', required=True) |
| p.add_argument('--hosts', nargs=2, required=True); p.add_argument('--x86host', required=True) |
| p.add_argument('--timeout', type=float, default=10800); p.add_argument('command', nargs=argparse.REMAINDER) |
| args = parser.parse_args() |
| if args.mode == 'kv': print(f'MEASURED {args.path}: KV tokens {kv(args.path.read_text(), args.expected)}') |
| elif args.mode == 'dual': dual(args.path.read_text(), args.rank) |
| elif args.mode == 'logs': clean_logs(args.path.read_text()) |
| elif args.mode == 'counters': counter_delta(load(args.before), load(args.after)) |
| elif args.mode == 'receipt': receipt(args.path, args.gate, args.arm, args.root) |
| elif args.mode == 'timing': timing(args.path) |
| elif args.mode == 'rollback': |
| text = args.path.read_text() |
| require(not args.path.is_symlink() and text.startswith('#!/usr/bin/env bash\n'), 'saved coordinator-next.sh must be a local bash script') |
| require(all(fragment in text for fragment in ('prod_window.sh close', 'PRODUCTION HOLD OK', './start.sh', |
| 'window.sh close')), 'incomplete saved coordinator restoration commands') |
| subprocess.run(['bash', '-n', str(args.path)], check=True) |
| print(text, end='') |
| elif args.mode == 'bench': print(json.dumps(bench(args.path), allow_nan=False)) |
| elif args.mode == 'summary': |
| data = [r for path in sorted(args.root.glob('*-cold.jsonl')) + sorted(args.root.glob('*-warm.jsonl')) for r in bench(path)] |
| require(bool(data) or args.pair, 'no benchmark summaries') |
| (args.root / 'summary.json').write_text(json.dumps(data, indent=2, allow_nan=False) + '\n') |
| print(f'MEASURED: {len(data)} cells; sources in {args.root}/summary.json. Coordinator applies plan pass lines; no automatic promotion.') |
| elif args.mode == 'supervise': return supervise(args) |
| return 0 |
|
|
|
|
| if __name__ == '__main__': |
| try: |
| sys.exit(main()) |
| except (ValueError, KeyError, OSError, TypeError) as error: |
| print(f'STOP: {error}', file=sys.stderr) |
| sys.exit(1) |
|
|
