"""CHANNELPROC-01: parent-side channel supervision, bounded IPC and run fencing.""" from copy import deepcopy import fcntl import multiprocessing import os from pathlib import Path import queue import signal import threading import time from uuid import uuid4 from app.channel_child import run_channel from app.stream_lifecycle import make_stream_failure from app.stream_manager import StreamGroup class ProcessSlots: def __init__(self, limit=0, lock_path=None): # 0 means no AI-side channel-count admission cap. Ownership/flock guards remain. if type(limit) is not int or limit < 0: raise ValueError('Invalid channel process limit.') self.limit = limit self._owned = set() self._lock = threading.Lock() self._file = None self.lock_path = lock_path or '/var/lib/contextra/runtime/channel-parent.lock' def guard(self): with self._lock: if self._file is None: path = Path(self.lock_path) path.parent.mkdir(parents=True, exist_ok=True) handle = path.open('a+') try: fcntl.flock(handle, fcntl.LOCK_EX | fcntl.LOCK_NB) except Exception: handle.close() raise RuntimeError('Another channel supervisor owns this runtime.') from None self._file = handle def acquire(self, sid): self.guard() with self._lock: if sid in self._owned or (self.limit > 0 and len(self._owned) >= self.limit): raise RuntimeError('Channel process capacity or ownership conflict.') self._owned.add(sid) def release(self, sid): with self._lock: self._owned.discard(sid) def close(self): with self._lock: if self._owned: return False if self._file is not None: self._file.close() self._file = None return True class ProcessStreamGroup(StreamGroup): def __init__(self, spec, *, slots, metadata_publisher=None, child_target=run_channel): super().__init__(spec, metadata_publisher=metadata_publisher) self._slots = slots self._target = child_target self._process = None self._monitor_thread = None self._cached = {} self._writer_token = None self._last_report = None self._stop_deadline = None self._replies = {} self._reply_lock = threading.Lock() self._exit_diagnostic = {} def _record_child_event(self, name): events = self._exit_diagnostic.setdefault('events', []) events.append(dict(name=name, monotonic_ns=time.monotonic_ns(), wall_ms=time.time_ns()//1_000_000)) del events[:-16] def start(self, *, assigned_run_id=None, assigned_generation=None): with self._teardown_lock: if self._pending_restart is not None: return self.run_generation if self._process is not None or not self._cleanup_done.is_set(): if self.desired_state == 'running' and self.state in ('starting','running'): return self.run_generation raise RuntimeError('Previous child is not reaped.') if self._cleanup_failure is not None: raise RuntimeError('Writer ownership cleanup is incomplete.') self._slots.acquire(self.stream_id) generation = self.lifecycle.begin(assigned_run_id=assigned_run_id, assigned_generation=assigned_generation) run_id = self.lifecycle.snapshot()['run_id'] self._cached = {} self._exit_diagnostic = {} self._record_child_event('start_requested') self._last_report = None self.started_at = time.time() self._stop_deadline = None self._cleanup_done.clear() context = multiprocessing.get_context('spawn') self._commands = context.Queue(maxsize=4) self._output = context.Queue(maxsize=2) self._stop = context.Event() self._terminal_ack = context.Event() publisher = (self.metadata_publisher if self.spec.get('server_id') and self.spec.get('source_id') else None) settings = None try: if publisher is not None: self._writer_token = publisher.claim(self.spec['source_id'], run_id, generation) settings = publisher.settings self._process = context.Process(target=self._target, name='channel-' + self.stream_id, args=(deepcopy(self.spec), run_id, generation, settings, self._writer_token, self._commands, self._output, self._stop, self._terminal_ack, os.getpid()), daemon=True) launch = dict(ready=threading.Event(), error=None) self._monitor_thread = threading.Thread(target=self._monitor, args=(self._process, run_id, generation, launch), name='channel-supervisor', daemon=True) self._monitor_thread.start() if not launch['ready'].wait(10): raise TimeoutError('Child spawn did not complete.') if launch['error'] is not None: raise launch['error'] except TimeoutError: # The owner thread retains cleanup responsibility if spawn is delayed. self._request_stop('spawn_timeout') raise RuntimeError('Child launch is pending cleanup.') from None except Exception: if self._process is not None and self._process.pid is not None: self._process.terminate() self._process.join(2) if self._process.is_alive(): self._cleanup_failure = make_stream_failure('STOP_CLEANUP_FAILED','child_launch_cleanup') raise RuntimeError('Child launch cleanup failed.') from None self._finish(run_id, generation, None, True, False) raise RuntimeError('Channel process could not start.') from None return generation def stop(self): with self._teardown_lock: self._pending_restart = None generation = self.lifecycle.request_stop() if self._process is None: self.lifecycle.finish_stop(generation, self._cleanup_failure) return self._request_stop() def _request_stop(self, reason='stop_requested'): if self._stop_deadline is None: self._record_child_event(reason) self._stop_deadline = time.monotonic() + self.CLEANUP_TIMEOUT_S + 3 self._stop.set() def restart(self, candidate=None): with self._teardown_lock: target = deepcopy(self.spec if candidate is None else candidate) self.check_restart(target) if self._pending_restart is not None: return if self._process is None and self._cleanup_done.is_set(): self._apply_spec(target) self.start() return self._pending_restart = target self.lifecycle.request_stop() self._request_stop('restart_requested') def _accept(self, message, run_id, generation): if (message.get('run_id'), message.get('run_generation')) != (run_id, generation): return False kind = message.get('kind') if kind == 'reply': with self._reply_lock: target = self._replies.get(message.get('id')) if target is not None: target[1].update(message) target[0].set() return True if kind not in ('status','terminal'): return False with self._teardown_lock: if (self.run_generation, self.lifecycle.snapshot()['run_id']) != (generation, run_id): return False value = message.get('value') if value is not None: if value.get('last_error') and 'first_error' not in self._exit_diagnostic: self._exit_diagnostic['first_error'] = deepcopy(value['last_error']) self._record_child_event('first_child_error') if not self.lifecycle.observe_child(value): return False self._cached = value self._last_report = time.monotonic() if self.metadata_publisher is not None and self._writer_token is not None: self.metadata_publisher.update_stats(self.spec['source_id'], self._writer_token, value.get('publisher_metrics',{})) if kind == 'terminal': if not self._terminal_ack.is_set(): self._record_child_event('terminal_received') self._exit_diagnostic['child_exception'] = message.get('exception_type') self._terminal_ack.set() return True def _monitor(self, process, run_id, generation, launch): # PDEATHSIG follows the spawning thread. Keep that owner alive until reap. try: self._exit_diagnostic['spawn_thread_id'] = threading.get_native_id() process.start() except Exception as error: launch['error'] = error finally: launch['ready'].set() if launch['error'] is not None: return terminal = None forced = False started = time.monotonic() try: while process.is_alive(): try: message = self._output.get(timeout=.2) if self._accept(message,run_id,generation) and message.get('kind') == 'terminal': terminal = message except queue.Empty: pass now = time.monotonic() with self._teardown_lock: reference = self._last_report or started timeout = 10 if self._last_report is not None else 90 if now-reference > timeout and self._stop_deadline is None: self.lifecycle.signal_error(generation, make_stream_failure('ROLE_RUNTIME_FAILED','child_heartbeat')) self.lifecycle.request_stop() self._request_stop('heartbeat_timeout') if terminal is not None and self._stop_deadline is None: # Normal interpreter/SDK finalization was observed beyond three seconds. self._stop_deadline = now+5 deadline = self._stop_deadline if deadline is not None and now >= deadline: if not process.is_alive(): break forced = True with self._teardown_lock: self._record_child_event('terminate_sent') process.terminate() process.join(2) if process.is_alive(): with self._teardown_lock: self._record_child_event('kill_sent') process.kill() process.join(2) break process.join(2) # A terminal report can still be buffered when the child exits. for _ in range(2): try: message = self._output.get_nowait() if self._accept(message,run_id,generation) and message.get('kind') == 'terminal': terminal = message except queue.Empty: break failed = forced or process.exitcode != 0 or terminal is None or terminal.get('failed',False) self._finish(run_id,generation,terminal,failed,forced) except Exception: with self._teardown_lock: self._cleanup_failure = make_stream_failure('STOP_CLEANUP_FAILED','child_supervisor') self.lifecycle.record_cleanup(generation,self._cleanup_failure) def _finish(self, run_id, generation, terminal, failed, forced): with self._teardown_lock: process = self._process if process is not None and process.pid is not None and process.is_alive(): self._cleanup_failure = make_stream_failure('STOP_CLEANUP_FAILED','child_still_alive') self.lifecycle.record_cleanup(generation,self._cleanup_failure) return handoff_error = None if self.metadata_publisher is not None and self._writer_token is not None: try: self.metadata_publisher.retire(self.spec['source_id'], self._writer_token, run_id,generation,failed) except Exception: handoff_error = make_stream_failure('STOP_CLEANUP_FAILED','writer_handoff') cleanup_error = handoff_error or (make_stream_failure('STOP_CLEANUP_FAILED','child_forced_exit') if forced else None) self._cleanup_failure = handoff_error self.lifecycle.finish_stop(generation,cleanup_error) if failed: self.lifecycle.signal_error(generation,make_stream_failure('ROLE_RUNTIME_FAILED','child_exit')) self._last_cleanup = dict(generation=generation,completed=not failed and handoff_error is None, child_reaped=True,forced=forced,writer_released=handoff_error is None, finished_monotonic_ns=time.monotonic_ns()) self._record_child_event('child_reaped') code = process.exitcode if process is not None else None self._exit_diagnostic.update(exit_code=code, exit_signal=signal.Signals(-code).name if code is not None and code < 0 else None, terminal_received=terminal is not None, terminal_failed=terminal.get('failed') if terminal is not None else None, forced=forced) self._last_cleanup['diagnostic'] = deepcopy(self._exit_diagnostic) self._process = None self._writer_token = None for resource in (self._commands,self._output): resource.cancel_join_thread() resource.close() if process is not None: process.close() self._slots.release(self.stream_id) self.started_at = None self._cleanup_done.set() pending,self._pending_restart = self._pending_restart,None if pending is not None and not failed and handoff_error is None: self._apply_spec(pending) self.start() def configure_metrics(self, generation, enabled, seconds): with self._teardown_lock: if generation != self.run_generation or self.state != 'running' or self._process is None: raise RuntimeError('Measurement run is not active.') commands = self._commands request_id = uuid4().hex event, reply = threading.Event(), {} with self._reply_lock: if len(self._replies) >= 4: raise RuntimeError('Diagnostics queue is full.') self._replies[request_id] = event, reply try: commands.put_nowait(dict(op='metrics',id=request_id,enabled=enabled,seconds=seconds)) if not event.wait(3) or 'error' in reply: raise RuntimeError('Child diagnostics unavailable.') return reply['value'] except queue.Full: raise RuntimeError('Child command queue is full.') from None finally: with self._reply_lock: self._replies.pop(request_id,None) def status(self, detail=False): with self._teardown_lock: out = deepcopy(self._cached) out.update(self.lifecycle.snapshot()) out.update(stream_id=self.stream_id,site_id=self.site_id,source_type=self.source_type, cleanup_pending=not self._cleanup_done.is_set(),restart_pending=self._pending_restart is not None, last_cleanup=deepcopy(self._last_cleanup),started_at=self.started_at, uptime_s=round(time.time()-self.started_at,1) if self.started_at else None, execution_mode='process',child_pid=self._process.pid if self._process else None, heartbeat_age_s=round(time.monotonic()-self._last_report,3) if self._last_report else None) out.setdefault('metadata_path',dict(state='NOT_CONNECTED')) out.setdefault('scenarios',dict(active=[],disabled=[])) if self._process is None: out['metadata_path'] = dict(state='NOT_CONNECTED') return out