import argparse import json import os import socket import ssl import subprocess import sys import tempfile import threading import time import yaml from datetime import datetime, timezone from types import SimpleNamespace from urllib.parse import urlsplit sys.dont_write_bytecode = True if not sys.dont_write_bytecode: raise RuntimeError('worker CLI could not disable bytecode writes') from remote_worker_client import WorkerHTTPClient, default_worker_paths from runtime_security import ( atomic_write_private_json, canonical_path, durable_unlink, ensure_private_directory, private_file_ready, read_private_json, ) from worker_contracts import WORKER_EVENT_SCHEMA from worker_local_state import ( DEFAULT_LOG_BYTES, DEFAULT_LOG_FILES, DEFAULT_RETENTION_BYTES, DEFAULT_RETENTION_DAYS, RETENTION_SCHEMA, WorkerLocalState, WorkerLocalStateError, ) from worker_package import verify_worker_package, worker_package_manifest_sha256 from worker_supervisor import ( WorkerAlreadyRunning, WorkerSupervisor, WorkerSupervisorError, CONTROL_SCHEMA, PROJECTION_SCHEMA, capture_spawned_process_identity, classify_instance, detached_command, load_shutdown_receipt, send_control_request, spawn_detached, terminate_spawned_process, wait_for_startup, ) from worker_assignment_runner import run_assignment CLI_SCHEMA = 1 CONFIG_SCHEMA = 2 LEGACY_CONFIG_SCHEMA = 1 LAUNCH_SCHEMA = 1 MAX_INSTALL_CONFIG_BYTES = 64 * 1024 EXIT_SUCCESS = 0 EXIT_NOT_RUNNING = 3 EXIT_STALE_OR_UNVERIFIABLE = 4 EXIT_STARTUP_FAILED = 6 EXIT_STOP_INCOMPLETE = 7 EXIT_INVALID_INVOCATION = 64 class InvocationError(ValueError): pass class WorkerArgumentParser(argparse.ArgumentParser): def error(self, message): raise InvocationError(message) def _canonical(value): return json.dumps( value, ensure_ascii=True, sort_keys=True, separators=(',', ':'), allow_nan=False, ) def _print_json(value): print(_canonical(value), flush=True) def _config_path(paths): return os.path.join(paths['state_dir'], 'control', 'worker.config.json') def _validate_config(value): legacy_fields = { 'schema', 'server', 'token', 'parallelism', 'installed_at', } current_fields = { 'schema', 'server', 'token', 'parallelism', 'installed_at', 'retention_days', 'retention_bytes', 'log_bytes', 'log_files', } if not isinstance(value, dict): raise InvocationError('installed worker configuration is invalid') if value.get('schema') == LEGACY_CONFIG_SCHEMA: if set(value) == legacy_fields: value = { **value, 'schema': CONFIG_SCHEMA, 'retention_days': DEFAULT_RETENTION_DAYS, 'retention_bytes': DEFAULT_RETENTION_BYTES, 'log_bytes': DEFAULT_LOG_BYTES, 'log_files': DEFAULT_LOG_FILES, } elif set(value) == current_fields: value = {**value, 'schema': CONFIG_SCHEMA} else: raise InvocationError('installed worker configuration is invalid') elif value.get('schema') == CONFIG_SCHEMA and set(value) == current_fields: value = dict(value) else: raise InvocationError('installed worker configuration is invalid') if type(value.get('parallelism')) is not int or not 1 <= value['parallelism'] <= 128: raise InvocationError('installed worker parallelism is invalid') WorkerHTTPClient(value.get('server'), value.get('token')) if not isinstance(value.get('installed_at'), str): raise InvocationError('installed worker timestamp is invalid') if type(value.get('retention_days')) is not int or not 1 <= value['retention_days'] <= 3650: raise InvocationError('installed worker retention days are invalid') if type(value.get('retention_bytes')) is not int or not 16 * 1024 * 1024 <= value['retention_bytes'] <= 1024 ** 4: raise InvocationError('installed worker retention bytes are invalid') if type(value.get('log_bytes')) is not int or not 64 * 1024 <= value['log_bytes'] <= 1024 ** 3: raise InvocationError('installed worker log bytes are invalid') if type(value.get('log_files')) is not int or not 1 <= value['log_files'] <= 20: raise InvocationError('installed worker log file count is invalid') return dict(value) def _load_config(paths): try: return _validate_config(read_private_json(_config_path(paths))) except OSError as exc: raise InvocationError('worker is not installed; run truf-worker install') from exc def _load_install_config(path): if path == '-': payload = sys.stdin.buffer.read(MAX_INSTALL_CONFIG_BYTES + 1) else: path = os.path.abspath(os.fspath(path)) if not private_file_ready(path): raise InvocationError('worker install configuration must be a private regular file') with open(path, 'rb') as handle: payload = handle.read(MAX_INSTALL_CONFIG_BYTES + 1) if len(payload) > MAX_INSTALL_CONFIG_BYTES: raise InvocationError('worker install configuration exceeds its byte bound') try: class UniqueKeyLoader(yaml.SafeLoader): def compose_node(self, parent, index): if self.check_event(yaml.AliasEvent): event = self.get_event() raise yaml.composer.ComposerError( None, None, 'YAML aliases are not supported', event.start_mark, ) return super().compose_node(parent, index) def construct_mapping(loader, node, deep=False): loader.flatten_mapping(node) value = {} for key_node, value_node in node.value: key = loader.construct_object(key_node, deep=deep) if key in value: raise yaml.constructor.ConstructorError( 'while constructing a mapping', node.start_mark, f'duplicate key: {key}', key_node.start_mark, ) value[key] = loader.construct_object(value_node, deep=deep) return value UniqueKeyLoader.add_constructor( yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG, construct_mapping, ) value = yaml.load(payload.decode('utf-8', errors='strict'), Loader=UniqueKeyLoader) except (UnicodeDecodeError, yaml.YAMLError, TypeError, ValueError) as exc: raise InvocationError('worker install configuration is invalid YAML') from exc if not isinstance(value, dict) or set(value) != {'server', 'token', 'parallelism'}: raise InvocationError( 'worker install configuration must contain only server, token, and parallelism' ) return value def _configuration(args, paths, *, require_explicit=False): server = getattr(args, 'server', None) token = getattr(args, 'token', None) parallelism = getattr(args, 'parallelism', None) config_path = getattr(args, 'config', None) if config_path and any(value is not None for value in (server, token, parallelism)): raise InvocationError('--config cannot be combined with --server, --token, or --parallelism') if config_path: document = _load_install_config(config_path) server = document['server'] token = document['token'] parallelism = document['parallelism'] if bool(server) != bool(token): raise InvocationError('--server and --token must be supplied together') if server: value = { 'schema': CONFIG_SCHEMA, 'server': server, 'token': token, 'parallelism': 1 if parallelism is None else parallelism, 'installed_at': datetime.now(timezone.utc).isoformat().replace('+00:00', 'Z'), 'retention_days': int(getattr(args, 'retention_days', DEFAULT_RETENTION_DAYS)), 'retention_bytes': int(getattr(args, 'retention_bytes', DEFAULT_RETENTION_BYTES)), 'log_bytes': int(getattr(args, 'log_bytes', DEFAULT_LOG_BYTES)), 'log_files': int(getattr(args, 'log_files', DEFAULT_LOG_FILES)), } return _validate_config(value) if require_explicit: raise InvocationError('--server and --token are required') value = _load_config(paths) if parallelism is not None: value['parallelism'] = parallelism return _validate_config(value) def _client_args(config, paths): return SimpleNamespace( package_manifest=paths['package_manifest'], state_dir=paths['state_dir'], bundle_dir=paths['bundle_dir'], work_dir=paths['work_dir'], server=config['server'], token=config['token'], parallelism=config['parallelism'], poll_seconds=5.0, error_delay_seconds=15.0, http_timeout=120, retention_days=config['retention_days'], retention_bytes=config['retention_bytes'], log_bytes=config['log_bytes'], log_files=config['log_files'], ) def _add_configuration(parser, *, required=False, allow_config=False): parser.add_argument('--server', required=required) parser.add_argument('--token', required=required) parser.add_argument('--parallelism', type=int) if allow_config: parser.add_argument('--config') def build_parser(): parser = WorkerArgumentParser(prog='truf-worker', description='TRUF remote worker operator CLI') commands = parser.add_subparsers(dest='command', required=True) install = commands.add_parser('install') _add_configuration(install, allow_config=True) install.add_argument('--retention-days', type=int, default=DEFAULT_RETENTION_DAYS) install.add_argument('--retention-bytes', type=int, default=DEFAULT_RETENTION_BYTES) install.add_argument('--log-bytes', type=int, default=DEFAULT_LOG_BYTES) install.add_argument('--log-files', type=int, default=DEFAULT_LOG_FILES) run = commands.add_parser('run') _add_configuration(run) start = commands.add_parser('start') _add_configuration(start) start.add_argument('--startup-timeout', type=float, default=30.0) stop = commands.add_parser('stop') stop.add_argument('--timeout', type=float, default=30.0) stop.add_argument('--json', action='store_true') status = commands.add_parser('status') status.add_argument('--json', action='store_true') attach = commands.add_parser('attach') attach_output = attach.add_mutually_exclusive_group() attach_output.add_argument('--json', action='store_true') attach_output.add_argument('--ndjson', action='store_true') attach.add_argument('--follow-seconds', type=float) watch = commands.add_parser('watch') watch.add_argument('--follow-seconds', type=float) watch.set_defaults(json=False, ndjson=False) logs = commands.add_parser('logs') logs.add_argument('--tail', type=int, default=100) logs.add_argument('--follow', action='store_true') logs_output = logs.add_mutually_exclusive_group() logs_output.add_argument('--json', action='store_true') logs_output.add_argument('--ndjson', action='store_true') logs.add_argument('--follow-seconds', type=float, default=60.0) history = commands.add_parser('history') history.add_argument('--limit', type=int, default=100) history.add_argument('--reservation', type=int) history_output = history.add_mutually_exclusive_group() history_output.add_argument('--json', action='store_true') history_output.add_argument('--ndjson', action='store_true') doctor = commands.add_parser('doctor') doctor.add_argument('--json', action='store_true') child = commands.add_parser('_supervise', help=argparse.SUPPRESS) child.add_argument('--launch-file', required=True) child.add_argument('--startup-file', required=True) child.add_argument('--launch-nonce', required=True) runner = commands.add_parser('_assignment_runner', help=argparse.SUPPRESS) runner.add_argument('--root', required=True) return parser def parse_args(argv=None): values = list(sys.argv[1:] if argv is None else argv) if values and values[0].startswith('-'): values.insert(0, 'run') args = build_parser().parse_args(values) if getattr(args, 'parallelism', None) is not None and not 1 <= args.parallelism <= 128: raise InvocationError('--parallelism must be between 1 and 128') if getattr(args, 'startup_timeout', 1) is not None and not 0.1 <= getattr(args, 'startup_timeout', 1) <= 300: raise InvocationError('--startup-timeout must be between 0.1 and 300 seconds') if getattr(args, 'timeout', 1) is not None and not 0.1 <= getattr(args, 'timeout', 1) <= 3600: raise InvocationError('--timeout must be between 0.1 and 3600 seconds') if getattr(args, 'tail', 1) is not None and not 1 <= getattr(args, 'tail', 1) <= 10000: raise InvocationError('--tail must be between 1 and 10000') if getattr(args, 'limit', 1) is not None and not 1 <= getattr(args, 'limit', 1) <= 1000: raise InvocationError('--limit must be between 1 and 1000') follow = getattr(args, 'follow_seconds', None) if follow is not None and not 0 <= follow <= 3600: raise InvocationError('--follow-seconds must be between 0 and 3600') if getattr(args, 'reservation', None) is not None and args.reservation <= 0: raise InvocationError('--reservation must be positive') return args def _seconds(value, now=None): if not value: return None try: parsed = datetime.fromisoformat(str(value).replace('Z', '+00:00')) except ValueError: return None return round(((now or datetime.now(timezone.utc)) - parsed).total_seconds(), 3) def _local_layout_ready(state_dir): return all(os.path.isdir(os.path.join(state_dir, name)) for name in ( 'control', 'events', 'history', 'diagnostics', 'logs', )) def _local_settings(paths): try: config = _load_config(paths) except (InvocationError, ValueError): config = {} return { 'retention_days': config.get('retention_days', DEFAULT_RETENTION_DAYS), 'retention_bytes': config.get('retention_bytes', DEFAULT_RETENTION_BYTES), 'log_bytes': config.get('log_bytes', DEFAULT_LOG_BYTES), 'log_files': config.get('log_files', DEFAULT_LOG_FILES), } def _open_local_reader(paths): return WorkerLocalState( paths['state_dir'], read_only=True, **_local_settings(paths), ) def _directory_usage(path): byte_count = 0 file_count = 0 if os.path.isdir(path): for current, directories, files in os.walk(path, followlinks=False): directories[:] = [ name for name in directories if not os.path.islink(os.path.join(current, name)) ] for name in files: candidate = os.path.join(current, name) try: if os.path.isfile(candidate) and not os.path.islink(candidate): byte_count += os.path.getsize(candidate) file_count += 1 except OSError: continue return {'bytes': byte_count, 'files': file_count} def _root_file_usage(path): byte_count = 0 file_count = 0 if os.path.isdir(path): with os.scandir(path) as entries: for entry in entries: try: if entry.is_file(follow_symlinks=False): byte_count += entry.stat(follow_symlinks=False).st_size file_count += 1 except OSError: continue return {'bytes': byte_count, 'files': file_count} def _fallback_retention(paths): settings = _local_settings(paths) def non_evictable(item): return { **item, 'evictable_bytes': 0, 'evictable_files': 0, 'non_evictable_bytes': item['bytes'], 'non_evictable_files': item['files'], } categories = { 'control': non_evictable(_directory_usage( os.path.join(paths['state_dir'], 'control'), )), 'state': non_evictable(_root_file_usage(paths['state_dir'])), 'events': non_evictable({'bytes': 0, 'files': 0}), 'history': non_evictable({'bytes': 0, 'files': 0}), 'diagnostics': non_evictable({'bytes': 0, 'files': 0}), 'logs': non_evictable({'bytes': 0, 'files': 0}), 'bundles': non_evictable(_directory_usage(paths['bundle_dir'])), 'work': non_evictable(_directory_usage(paths['work_dir'])), } total_bytes = sum(item['bytes'] for item in categories.values()) return { 'schema': RETENTION_SCHEMA, 'maximum_age_days': settings['retention_days'], 'maximum_bytes': settings['retention_bytes'], 'total_bytes': total_bytes, 'total_files': sum(item['files'] for item in categories.values()), 'evictable_bytes': 0, 'evictable_files': 0, 'non_evictable_bytes': total_bytes, 'non_evictable_files': sum(item['files'] for item in categories.values()), 'over_limit': total_bytes > settings['retention_bytes'], 'categories': categories, } def _slot_document(slot, now=None): now = now or datetime.now(timezone.utc) progress = dict(slot.get('progress') or {}) assignment_remaining = _seconds(slot.get('assignment_deadline_at'), now) scan_remaining = _seconds(slot.get('scan_deadline_at'), now) measured = { key: value for key, value in progress.items() if type(value) in (int, float) and key not in { 'attempt', 'retry_after_seconds', } } counters = progress.get('counters') if isinstance(counters, dict): measured.update({ key: value for key, value in counters.items() if type(value) in (int, float) }) return { 'slot_id': int(slot['slot_id']), 'sequence': int(slot.get('sequence') or 0), 'phase': str(slot['phase']), 'phase_age_seconds': _seconds(slot.get('phase_started_at'), now), 'reservation_id': slot.get('reservation_id'), 'source': slot.get('source'), 'attempt': progress.get('attempt'), 'scan_deadline_at': slot.get('scan_deadline_at'), 'scan_deadline_remaining_seconds': ( -scan_remaining if scan_remaining is not None else None ), 'assignment_deadline_at': slot.get('assignment_deadline_at'), 'assignment_remaining_seconds': ( -assignment_remaining if assignment_remaining is not None else None ), 'last_progress_at': slot.get('timestamp'), 'last_progress_age_seconds': _seconds(slot.get('timestamp'), now), 'progress': progress, 'measured_progress': measured, 'child_state': progress.get('child_state'), 'idle_reason': progress.get('reason') if slot.get('phase') in {'idle', 'backoff'} else None, 'next_claim_at': progress.get('next_claim_at'), } def status_document(paths, classification=None): classification = classification or classify_instance(paths['state_dir']) state = classification['state'] instance = classification.get('instance') package = instance.get('package') if instance else None runtime = instance.get('runtime') if instance else None protocol = instance.get('protocol') if instance else None if instance is None: try: verified = verify_worker_package(paths['package_manifest']) manifest = verified['manifest'] package = { 'schema': manifest['schema'], 'manifest_sha256': worker_package_manifest_sha256(manifest), 'code_manifest_sha256': verified['code_manifest_sha256'], 'platform_tag': manifest['platform_tag'], } runtime = { 'python': '.'.join(str(item) for item in sys.version_info[:3]), 'platform': sys.platform, 'executable': canonical_path(sys.executable), 'mode': 'inactive', } protocol = { 'worker_protocol': manifest['protocol_version'], 'bundle_format': manifest['bundle_format_version'], 'event': WORKER_EVENT_SCHEMA, 'control': CONTROL_SCHEMA, 'projection': PROJECTION_SCHEMA, } except (OSError, ValueError): pass worker = None slots = [] retention = None if state in {'running', 'draining'}: try: snapshot = send_control_request(classification['record'], 'snapshot', {}) worker = snapshot['worker'] slots = [_slot_document(item) for item in snapshot['projection']['slots']] retention = snapshot['retention'] except (OSError, ValueError, WorkerSupervisorError, WorkerLocalStateError) as exc: state = 'unverifiable' classification = dict(classification) classification['state'] = state classification['detail'] = f'live worker snapshot failed: {type(exc).__name__}' if state not in {'running', 'draining'} and os.path.isdir(paths['state_dir']): local = None if _local_layout_ready(paths['state_dir']): try: local = _open_local_reader(paths) except (OSError, ValueError, WorkerLocalStateError): local = None projection = ( local.snapshot() if local is not None else {'aggregate': {'slot_count': 0, 'phases': {}}, 'slots': []} ) slots = [_slot_document(item) for item in projection['slots']] retention = ( local.retention_usage({ 'bundles': paths['bundle_dir'], 'work': paths['work_dir'], }) if local is not None else _fallback_retention(paths) ) try: configured_parallelism = _load_config(paths)['parallelism'] except (InvocationError, ValueError): configured_parallelism = None worker = { 'state': state, 'parallelism': configured_parallelism, 'slot_cap': configured_parallelism, 'configured_slots': configured_parallelism, 'recovery_slots': None, 'started_at': None, 'drain_deadline_at': None, 'aggregate': projection['aggregate'], } return { 'schema': CLI_SCHEMA, 'command': 'status', 'state': state, 'detail': classification.get('detail'), 'instance': instance, 'package': package, 'runtime': runtime, 'protocol': protocol, 'worker': worker, 'slots': slots, 'retention': retention, } def _human_status(document): print(f"Worker: {document['state']} ({document.get('detail') or 'no detail'})") instance = document.get('instance') if instance: print(f"Instance: {instance['instance_id']} PID: {instance['pid']} Mode: {instance['runtime']['mode']}") print( f"Package: {instance['package']['manifest_sha256']} " f"Protocol: {instance['protocol']['worker_protocol']}" ) worker = document.get('worker') if worker: print( f"Slots: {worker['aggregate']['slot_count']} Cap: {worker['slot_cap']} " f"State: {worker['state']}" ) for slot in document.get('slots') or []: reason = slot['idle_reason'] or '' visibility = slot['progress'].get('execution_visibility') or '' counters = _canonical(slot['measured_progress']) if slot['measured_progress'] else '{}' print( f"slot {slot['slot_id']}: {slot['phase']} age={slot['phase_age_seconds']}s " f"source={slot['source'] or '-'} reservation={slot['reservation_id'] or '-'} " f"attempt={slot['attempt'] or '-'} " f"scan_deadline={slot['scan_deadline_at'] or '-'} " f"scan_remaining={slot['scan_deadline_remaining_seconds']}s " f"assignment_deadline={slot['assignment_deadline_at'] or '-'} " f"assignment_remaining={slot['assignment_remaining_seconds']}s " f"progress_age={slot['last_progress_age_seconds']}s " f"next_claim_at={slot['next_claim_at'] or '-'} " f"child_state={slot['child_state'] or '-'} counters={counters} " f"{reason} {visibility}" ) retention = document.get('retention') if retention: print(f"Local retention: {retention['total_bytes']} bytes in {retention['total_files']} files") def command_install(args, paths): config = _configuration(args, paths, require_explicit=True) try: verify_worker_package(paths['package_manifest']) except Exception as exc: print(f'Worker installation failed: {type(exc).__name__}', file=sys.stderr) return EXIT_STARTUP_FAILED ensure_private_directory(paths['state_dir'], reject_reparse=True) ensure_private_directory(os.path.join(paths['state_dir'], 'control'), reject_reparse=True) ensure_private_directory(paths['bundle_dir'], reject_reparse=True) ensure_private_directory(paths['work_dir'], reject_reparse=True) atomic_write_private_json(_config_path(paths), config) print(f"Installed worker configuration at {_config_path(paths)}") return EXIT_SUCCESS def command_run(args, paths): config = _configuration(args, paths) try: supervisor = WorkerSupervisor(_client_args(config, paths), foreground=True) return supervisor.run() except WorkerAlreadyRunning as exc: print(str(exc), file=sys.stderr) return EXIT_STARTUP_FAILED except Exception as exc: print(f'Worker startup failed: {type(exc).__name__}', file=sys.stderr) return EXIT_STARTUP_FAILED def _launch_paths(paths, nonce): control = ensure_private_directory(os.path.join(paths['state_dir'], 'control'), reject_reparse=True) return ( os.path.join(control, f'worker.launch.{nonce}.json'), os.path.join(control, f'worker.startup.{nonce}.json'), ) def command_start(args, paths): current = classify_instance(paths['state_dir'], remove_stale=True) if current['state'] == 'running': document = status_document(paths, current) _human_status(document) return EXIT_SUCCESS if current['state'] in {'unverifiable', 'starting', 'draining'} or ( current['state'] == 'stale' and not current.get('removed') ): _human_status(status_document(paths, current)) return EXIT_STALE_OR_UNVERIFIABLE config = _configuration(args, paths) nonce = os.urandom(16).hex() launch_file, startup_file = _launch_paths(paths, nonce) launch = { 'schema': LAUNCH_SCHEMA, 'nonce': nonce, 'server': config['server'], 'token': config['token'], 'parallelism': config['parallelism'], 'retention_days': config['retention_days'], 'retention_bytes': config['retention_bytes'], 'log_bytes': config['log_bytes'], 'log_files': config['log_files'], 'package_manifest': paths['package_manifest'], 'state_dir': paths['state_dir'], 'bundle_dir': paths['bundle_dir'], 'work_dir': paths['work_dir'], } atomic_write_private_json(launch_file, launch) bootstrap = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'remote_worker_bootstrap.py') process = None spawned_identity = None started = False try: process = spawn_detached(detached_command(bootstrap, launch_file, startup_file, nonce)) spawned_identity = capture_spawned_process_identity(process) result = wait_for_startup(startup_file, nonce, process, timeout=args.startup_timeout) if result['outcome'] != 'ready': print(f"Worker startup failed: {result['outcome']}", file=sys.stderr) return EXIT_STARTUP_FAILED verified = classify_instance(paths['state_dir']) if ( verified['state'] != 'running' or verified['instance']['instance_id'] != result['instance_id'] or result['pid'] != spawned_identity['pid'] or verified['instance']['pid'] != spawned_identity['pid'] or verified['instance']['process_creation_time'] != spawned_identity['creation_time'] or canonical_path(verified['instance']['executable']) != canonical_path( spawned_identity['executable'] ) ): print('Worker startup identity could not be verified', file=sys.stderr) return EXIT_STARTUP_FAILED started = True _human_status(status_document(paths, verified)) return EXIT_SUCCESS except (OSError, ValueError, WorkerSupervisorError) as exc: print(f'Worker startup failed: {exc}', file=sys.stderr) return EXIT_STARTUP_FAILED finally: try: if process is not None and not started: if spawned_identity is not None: terminate_spawned_process(process, spawned_identity) elif process.poll() is None: process.terminate() try: process.wait(timeout=5) except subprocess.TimeoutExpired: process.kill() process.wait(timeout=5) classify_instance(paths['state_dir'], remove_stale=True) finally: for path in (launch_file, startup_file): try: if os.path.exists(path): durable_unlink(path) except OSError: pass def command_stop(args, paths): classification = classify_instance(paths['state_dir']) if classification['state'] == 'stopped': value = {'schema': CLI_SCHEMA, 'command': 'stop', 'state': 'stopped', 'receipt': None} _print_json(value) if args.json else print('Worker is not running') return EXIT_NOT_RUNNING if classification['state'] not in {'running', 'draining'}: value = { 'schema': CLI_SCHEMA, 'command': 'stop', 'state': classification['state'], 'receipt': None, } _print_json(value) if args.json else print(f"Worker is {classification['state']}") return EXIT_STALE_OR_UNVERIFIABLE record = classification['record'] try: result = send_control_request( record, 'stop', {'timeout_seconds': args.timeout}, timeout=min(args.timeout, 10), ) except (OSError, TimeoutError, WorkerSupervisorError): final = status_document(paths, classify_instance(paths['state_dir'])) value = { 'schema': CLI_SCHEMA, 'command': 'stop', 'state': 'control_disconnected', 'receipt': None, 'final_status': final, } if args.json: _print_json(value) else: print('Worker control disconnected; final state:') _human_status(final) return ( EXIT_NOT_RUNNING if final['state'] == 'stopped' else EXIT_STALE_OR_UNVERIFIABLE ) deadline = time.monotonic() + args.timeout + 5.0 receipt = None while time.monotonic() < deadline: try: receipt = load_shutdown_receipt(paths['state_dir'], record['instance_id']) break except (OSError, WorkerSupervisorError): time.sleep(0.1) clean = bool( receipt is not None and receipt.get('drained') is True and receipt.get('exit_code') == 0 ) state = 'stopped' if clean else ( 'non_drained' if receipt is not None else 'stop_timeout' ) value = { 'schema': CLI_SCHEMA, 'command': 'stop', 'state': state, 'accepted': result['accepted'], 'receipt': receipt, } _print_json(value) if args.json else print( 'Worker stopped cleanly' if clean else ( 'Worker exited without a clean drain receipt' if receipt is not None else 'Worker stop timed out without a shutdown receipt' ) ) return EXIT_SUCCESS if clean else EXIT_STOP_INCOMPLETE def command_status(args, paths): document = status_document(paths) _print_json(document) if args.json else _human_status(document) if document['state'] in {'running', 'starting', 'draining'}: return EXIT_SUCCESS if document['state'] == 'stopped': return EXIT_NOT_RUNNING return EXIT_STALE_OR_UNVERIFIABLE def _follow_events(record, sequence, seconds, emit): deadline = None if seconds is None else time.monotonic() + seconds while deadline is None or time.monotonic() < deadline: response = send_control_request( record, 'events', {'after_sequence': sequence, 'limit': 256}, ) for event in response['events']: sequence = max(sequence, int(event['sequence'])) emit(event) time.sleep(0.2) return sequence def command_attach(args, paths): command_name = getattr(args, 'command', 'attach') classification = classify_instance(paths['state_dir']) if classification['state'] not in {'running', 'draining'}: document = status_document(paths, classification) if args.json: _print_json({'schema': CLI_SCHEMA, 'command': command_name, 'status': document}) elif args.ndjson: _print_json({'schema': CLI_SCHEMA, 'type': 'snapshot', 'status': document}) else: _human_status(document) return ( EXIT_NOT_RUNNING if classification['state'] == 'stopped' else EXIT_STALE_OR_UNVERIFIABLE ) record = classification['record'] snapshot = status_document(paths, classification) if args.json: _print_json({ 'schema': CLI_SCHEMA, 'command': command_name, 'status': snapshot, }) return EXIT_SUCCESS if args.ndjson: _print_json({'schema': CLI_SCHEMA, 'type': 'snapshot', 'status': snapshot}) else: _human_status(snapshot) action = 'Watching' if command_name == 'watch' else 'Attached' print(f'{action}; q, EOF, or Ctrl-C detaches without stopping the worker.') stop = threading.Event() if not args.ndjson: def read_input(): try: while not stop.is_set(): value = sys.stdin.read(1) if value == '' or value.lower() == 'q': stop.set() return except (EOFError, OSError): stop.set() threading.Thread(target=read_input, daemon=True).start() sequence = max((slot.get('sequence', 0) for slot in snapshot.get('slots', [])), default=0) deadline = None if args.follow_seconds is None else time.monotonic() + args.follow_seconds next_refresh = time.monotonic() + 1.0 disconnected = False try: while not stop.is_set() and (deadline is None or time.monotonic() < deadline): response = send_control_request( record, 'events', {'after_sequence': sequence, 'limit': 256}, ) for event in response['events']: sequence = max(sequence, int(event['sequence'])) if args.ndjson: _print_json(event) now = time.monotonic() if not args.ndjson and (response['events'] or now >= next_refresh): refreshed = status_document(paths, classification) print('---') _human_status(refreshed) next_refresh = now + 1.0 stop.wait(0.2) except (OSError, TimeoutError, WorkerSupervisorError): disconnected = True final = status_document(paths, classify_instance(paths['state_dir'])) if args.ndjson: _print_json({'schema': CLI_SCHEMA, 'type': 'snapshot', 'status': final}) else: print('Worker control disconnected; final state:') _human_status(final) except (KeyboardInterrupt, EOFError): pass finally: stop.set() if disconnected: return ( EXIT_NOT_RUNNING if final['state'] == 'stopped' else EXIT_STALE_OR_UNVERIFIABLE ) return EXIT_SUCCESS def command_logs(args, paths): if not os.path.isdir(paths['state_dir']): value = {'schema': CLI_SCHEMA, 'command': 'logs', 'lines': []} if args.follow and (args.json or args.ndjson): print('Worker logs follow unavailable: worker state is not initialized', file=sys.stderr) elif args.json: _print_json(value) elif not args.ndjson: print('No worker logs') return EXIT_NOT_RUNNING layout_ready = _local_layout_ready(paths['state_dir']) local = _open_local_reader(paths) if layout_ready else None lines = local.log_tail(args.tail) if local is not None else [] machine_follow = args.follow and (args.json or args.ndjson) if machine_follow: sequence = max(0, (local.snapshot()['sequence'] if local else 0) - args.tail) events = local.events_after(sequence, args.tail) if local else [] for event in events: _print_json(event) if events: sequence = events[-1]['sequence'] elif args.ndjson: for index, line in enumerate(lines, 1): _print_json({'schema': CLI_SCHEMA, 'type': 'log', 'index': index, 'line': line}) elif args.json and not args.follow: _print_json({'schema': CLI_SCHEMA, 'command': 'logs', 'lines': lines}) else: for line in lines: print(line) if not args.follow: return EXIT_SUCCESS classification = classify_instance(paths['state_dir']) if classification['state'] not in {'running', 'draining'}: if machine_follow: print( f"Worker logs follow ended: {classification['state']}", file=sys.stderr, ) return EXIT_NOT_RUNNING if local is None: local = _open_local_reader(paths) if not machine_follow: sequence = local.snapshot()['sequence'] try: _follow_events( classification['record'], sequence, args.follow_seconds, _print_json if machine_follow else lambda event: print( f"#{event['sequence']} slot {event['slot_id']} {event['phase']}" ), ) except (OSError, TimeoutError, WorkerSupervisorError): final = status_document(paths, classify_instance(paths['state_dir'])) if machine_follow: print( f"Worker logs follow control disconnected: {final['state']}", file=sys.stderr, ) else: print('Worker control disconnected; final state:') _human_status(final) return ( EXIT_NOT_RUNNING if final['state'] == 'stopped' else EXIT_STALE_OR_UNVERIFIABLE ) except KeyboardInterrupt: pass return EXIT_SUCCESS def command_history(args, paths): if not _local_layout_ready(paths['state_dir']): values = [] else: values = _open_local_reader(paths).history(args.limit, args.reservation) if args.ndjson: for value in values: _print_json(value) elif args.json: _print_json({'schema': CLI_SCHEMA, 'command': 'history', 'assignments': values}) elif not values: print('No terminal assignment history') else: for value in values: print( f"reservation {value['reservation_id']}: {value['outcome']} " f"source={value.get('source') or '-'} slot={value['slot_id']} " f"completed={value['completed_at']} duration={value.get('duration_seconds')}s" ) timeline = ' -> '.join( f"{event['phase']}#{event['sequence']}" for event in value.get('timeline') or [] ) if timeline: print(f' timeline {timeline}') for diagnostic in value.get('diagnostics') or []: print( f" diagnostic {diagnostic.get('diagnostic_uid')}: " f"{diagnostic.get('record')} available={diagnostic.get('available')}" ) return EXIT_SUCCESS def _doctor_check(name, applicable, status, summary): return { 'name': name, 'applicable': bool(applicable), 'status': str(status), 'summary': str(summary), } def command_doctor(args, paths): checks = [] package = None installed = os.path.isfile(paths['package_manifest']) checks.append(_doctor_check( 'installation', True, 'ok' if installed else 'error', 'worker package manifest is installed' if installed else 'worker package manifest is absent', )) try: package = verify_worker_package(paths['package_manifest']) checks.append(_doctor_check('package_identity', True, 'ok', 'package identity verified')) except Exception as exc: checks.append(_doctor_check('package_identity', True, 'error', f'package verification failed: {type(exc).__name__}')) config = None try: config = _load_config(paths) checks.append(_doctor_check('configuration', True, 'ok', 'installed configuration is valid')) checks.append(_doctor_check('credentials', True, 'ok', 'worker credential shape is valid')) except (InvocationError, ValueError) as exc: checks.append(_doctor_check('configuration', True, 'error', str(exc))) checks.append(_doctor_check('credentials', False, 'not_applicable', 'configuration is unavailable')) path_values = (paths['state_dir'], paths['bundle_dir'], paths['work_dir']) missing_paths = [path for path in path_values if not os.path.isdir(path)] if missing_paths: checks.append(_doctor_check( 'paths', False, 'not_applicable', 'runtime paths are not initialized; run install first', )) else: probe_error = None for path in path_values: descriptor = None temporary = None try: descriptor, temporary = tempfile.mkstemp(prefix='.doctor-', dir=path) except OSError as exc: probe_error = exc break finally: if descriptor is not None: os.close(descriptor) if temporary is not None: try: os.unlink(temporary) except FileNotFoundError: pass except OSError as exc: probe_error = probe_error or exc if probe_error is None: checks.append(_doctor_check('paths', True, 'ok', 'initialized state and data paths are writable')) else: checks.append(_doctor_check( 'paths', True, 'error', f'path validation failed: {type(probe_error).__name__}', )) for name, key in (('git', 'git_path'), ('trufflehog', 'trufflehog_path')): if package is None: checks.append(_doctor_check(name, False, 'not_applicable', 'verified package is unavailable')) continue try: completed = subprocess.run( [package[key], '--version'], stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=10, check=False, ) status = 'ok' if completed.returncode == 0 else 'error' checks.append(_doctor_check(name, True, status, f'{name} executable returned {completed.returncode}')) except (OSError, subprocess.TimeoutExpired) as exc: checks.append(_doctor_check(name, True, 'error', f'{name} validation failed: {type(exc).__name__}')) if config is None: checks.append(_doctor_check('server_reachability', False, 'not_applicable', 'configuration is unavailable')) else: parsed = urlsplit(config['server']) try: context = ssl.create_default_context() with socket.create_connection((parsed.hostname, parsed.port or 443), timeout=3) as raw: with context.wrap_socket(raw, server_hostname=parsed.hostname): pass checks.append(_doctor_check('server_reachability', True, 'ok', 'server TLS endpoint is reachable')) except OSError as exc: checks.append(_doctor_check('server_reachability', True, 'error', f'server is unreachable: {type(exc).__name__}')) classification = classify_instance(paths['state_dir']) status = ( 'ok' if classification['state'] in {'running', 'starting', 'draining', 'stopped'} else 'error' ) checks.append(_doctor_check('singleton_process', True, status, classification['detail'])) if _local_layout_ready(paths['state_dir']): try: local = _open_local_reader(paths) retention = local.retention_usage({ 'bundles': paths['bundle_dir'], 'work': paths['work_dir'], }) except (OSError, ValueError, WorkerLocalStateError): retention = _fallback_retention(paths) checks.append(_doctor_check( 'local_state', True, 'error', 'local state is unreadable', )) else: retention = _fallback_retention(paths) checks.append(_doctor_check('retention', True, 'ok', f"{retention['total_bytes']} bytes retained")) overall = 'ok' if all(item['status'] in {'ok', 'not_applicable'} for item in checks) else 'error' document = { 'schema': CLI_SCHEMA, 'command': 'doctor', 'overall': overall, 'checks': checks, 'retention': retention, } if args.json: _print_json(document) else: print(f'Doctor: {overall}') for check in checks: print(f"[{check['status']}] {check['name']}: {check['summary']}") return EXIT_SUCCESS if overall == 'ok' else EXIT_STARTUP_FAILED def _load_launch(path, nonce): value = read_private_json(path) if not isinstance(value, dict) or set(value) != { 'schema', 'nonce', 'server', 'token', 'parallelism', 'package_manifest', 'state_dir', 'bundle_dir', 'work_dir', 'retention_days', 'retention_bytes', 'log_bytes', 'log_files', } or value.get('schema') != LAUNCH_SCHEMA or value.get('nonce') != nonce: raise InvocationError('detached launch record is invalid') _validate_config({ 'schema': CONFIG_SCHEMA, 'server': value['server'], 'token': value['token'], 'parallelism': value['parallelism'], 'installed_at': '', 'retention_days': value['retention_days'], 'retention_bytes': value['retention_bytes'], 'log_bytes': value['log_bytes'], 'log_files': value['log_files'], }) return value def command_supervise(args): try: value = _load_launch(args.launch_file, args.launch_nonce) durable_unlink(args.launch_file) client = SimpleNamespace( package_manifest=value['package_manifest'], state_dir=value['state_dir'], bundle_dir=value['bundle_dir'], work_dir=value['work_dir'], server=value['server'], token=value['token'], parallelism=value['parallelism'], poll_seconds=5.0, error_delay_seconds=15.0, http_timeout=120, retention_days=value['retention_days'], retention_bytes=value['retention_bytes'], log_bytes=value['log_bytes'], log_files=value['log_files'], ) return WorkerSupervisor(client, foreground=False).run( startup_file=args.startup_file, launch_nonce=args.launch_nonce, ) except WorkerAlreadyRunning: return EXIT_STARTUP_FAILED except BaseException as exc: from worker_supervisor import write_startup_result try: write_startup_result( args.startup_file, args.launch_nonce, 'failed', error=type(exc).__name__, ) except OSError: pass return EXIT_STARTUP_FAILED def command_assignment_runner(args, paths): work_root = canonical_path(ensure_private_directory( os.path.abspath(paths['work_dir']), reject_reparse=True, )) root = canonical_path(os.path.abspath(args.root)) if os.path.dirname(root) != work_root or not os.path.basename(root).startswith( 'worker-assignment-' ): raise InvocationError('assignment runner root is outside worker work storage') package = verify_worker_package(paths['package_manifest']) manifest = package.pop('manifest') build_compatibility = package.pop('build_compatibility') package.pop('runtime_trees') runtime = { **package, 'build_compatibility': build_compatibility, 'capabilities': tuple( (item['source'], item['platform'], item['planning_kind']) for item in manifest['capabilities'] ), } return run_assignment(root, runtime) def main(argv=None): try: args = parse_args(argv) if args.command == '_supervise': return command_supervise(args) paths = default_worker_paths(module_path=__file__) if args.command == '_assignment_runner': return command_assignment_runner(args, paths) commands = { 'install': command_install, 'run': command_run, 'start': command_start, 'stop': command_stop, 'status': command_status, 'attach': command_attach, 'watch': command_attach, 'logs': command_logs, 'history': command_history, 'doctor': command_doctor, } return commands[args.command](args, paths) except (InvocationError, ValueError) as exc: print(f'truf-worker: {exc}', file=sys.stderr) return EXIT_INVALID_INVOCATION except (WorkerSupervisorError, WorkerLocalStateError) as exc: print(f'truf-worker: {exc}', file=sys.stderr) return EXIT_STALE_OR_UNVERIFIABLE except KeyboardInterrupt: return 130 if __name__ == '__main__': raise SystemExit(main())