From ab64003003b2c42e420e95acfab11cb9fc7a4e8d Mon Sep 17 00:00:00 2001 From: Erik Le Blansch <58982453+erikleblansch@users.noreply.github.com> Date: Fri, 16 Oct 2026 00:12:42 +0200 Subject: [PATCH] fix: observe native task memory before its cgroup exits --- scripts/opencode_task_holder.py | 244 +++++++++++++ scripts/smoke_opencode_inference.py | 119 ++++++- tests/test_opencode_ci.py | 10 +- tests/test_opencode_task_holder.py | 507 ++++++++++++++++++++++++++++ 4 files changed, 860 insertions(+), 20 deletions(-) create mode 100644 scripts/opencode_task_holder.py create mode 100644 tests/test_opencode_task_holder.py diff --git a/scripts/opencode_task_holder.py b/scripts/opencode_task_holder.py new file mode 100644 index 0000000..f015ed5 --- /dev/null +++ b/scripts/opencode_task_holder.py @@ -0,0 +1,244 @@ +#!/usr/bin/env python3 +# SPDX-License-Identifier: GPL-3.0-only +"""Fixed guest task lifetime: observe the original cgroup before its last exit. + +The Node task, its permissions, private cleanup and result are unchanged. This +helper joins that child, then stays alone in the original bounded task cgroup +until its owner has read the real counters. Readiness is not task acceptance. +""" +import argparse +import hashlib +import importlib.util +import json +import os +from pathlib import Path +import re +import signal +import socket +import stat +import struct +import subprocess +import sys +import time + +CONTROL = 'task-observation.sock' +FRAME_MAX = 1024 +TASK_SECONDS = 2550 +OWNER_SECONDS = 2565 + + +def require(value, reason='task_holder_contract'): + if not value: + raise ValueError(reason) + + +def remaining(deadline): + require(type(deadline) is int and 0 < deadline < 2 ** 63) + seconds = (deadline - time.monotonic_ns()) / 1_000_000_000 + require(0 < seconds <= OWNER_SECONDS, 'task_holder_deadline') + return seconds + + +def binding(kind, nonce, status, receipt): + require(kind in ('ready', 'release') and type(nonce) is str + and re.fullmatch('[0-9a-f]{64}', nonce)) + require(type(status) is int and -64 <= status <= 255 + and (receipt is None or type(receipt) is str and re.fullmatch('[0-9a-f]{64}', receipt))) + return (json.dumps(dict(version=1, kind='task_' + kind, nonce=nonce, + child_returncode=status, receipt_sha256=receipt), + sort_keys=True, separators=(',', ':')) + '\n').encode() + + +def decode_ready(raw, nonce): + # Re-encoding also rejects duplicate/extra keys, bools, whitespace and EOF + # ambiguity. The peer is a fixed source-bound helper, never a remote client. + value = json.loads(raw) + require(type(value) is dict) + expected = binding('ready', nonce, value.get('child_returncode'), value.get('receipt_sha256')) + require(raw == expected, 'task_holder_binding') + return value + + +def send_frame(channel, raw, deadline): + require(type(raw) is bytes and 0 < len(raw) <= FRAME_MAX) + channel.settimeout(remaining(deadline)) + channel.sendall(raw) + channel.shutdown(socket.SHUT_WR) + + +def receive_frame(channel, deadline): + raw = bytearray() + while True: + channel.settimeout(remaining(deadline)) + block = channel.recv(FRAME_MAX + 1 - len(raw)) + if not block: + require(0 < len(raw) <= FRAME_MAX, 'task_holder_frame') + return bytes(raw) + raw.extend(block) + require(len(raw) <= FRAME_MAX, 'task_holder_frame') + + +def peer(channel, pid): + require(type(pid) is int and 1 < pid < 2 ** 31) + observed = struct.unpack('3i', channel.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12)) + require(observed == (pid, os.getuid(), os.getgid()), 'task_holder_peer') + + +def identity(info): + return info.st_dev, info.st_ino + + +def file_identity(info): + return (info.st_dev, info.st_ino, info.st_mode, info.st_uid, info.st_gid, + info.st_nlink, info.st_size, info.st_mtime_ns, info.st_ctime_ns) + + +def receipt_bytes(path): + require(path.is_absolute() and path.resolve(strict=True) == path, 'task_holder_receipt') + with os.fdopen(os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK | os.O_CLOEXEC), 'rb') as stream: + before = os.fstat(stream.fileno()) + require(stat.S_ISREG(before.st_mode) and before.st_uid == os.getuid() + and stat.S_IMODE(before.st_mode) == 0o600 and before.st_nlink == 1 + and 0 < before.st_size <= 32768, 'task_holder_receipt') + raw = stream.read(32769) + after = os.fstat(stream.fileno()) + require(len(raw) == before.st_size and file_identity(before) == file_identity(after) + and identity(path.lstat()) == identity(before), 'task_holder_receipt_changed') + return raw + + +class Group: + """One fixed task directory, held across both live-member and counter reads.""" + def __init__(self, path, pid): + require(path.is_absolute() and path.resolve(strict=True) == path, 'task_holder_cgroup') + self.path, self.pid = path, pid + self.parent = identity(path.parent.lstat()) + self.fd = os.open(path, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC) + self.bound = identity(os.fstat(self.fd)) + + def close(self): + os.close(self.fd) + + def check(self): + require(identity(self.path.parent.lstat()) == self.parent + and identity(self.path.lstat()) == self.bound + and stat.S_ISDIR(self.path.lstat().st_mode) + and identity(os.fstat(self.fd)) == self.bound, 'task_holder_cgroup_changed') + + def read(self, name): + require(name in ('memory.events', 'memory.max', 'memory.swap.max', 'memory.peak')) + self.check() + with os.fdopen(os.open(name, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK | os.O_CLOEXEC, + dir_fd=self.fd), 'rb') as stream: + require(stat.S_ISREG(os.fstat(stream.fileno()).st_mode), 'task_holder_counter') + raw = stream.read(4097) + require(0 < len(raw) <= 4096, 'task_holder_counter') + self.check() + return raw + + def alone(self): + self.check() + paths = [] + for path in self.path.rglob('cgroup.procs'): + require(len(paths) < 16 and path.resolve(strict=True) == path, 'task_holder_members') + paths.append(path) + require(self.path / 'cgroup.procs' in paths, 'task_holder_members') + members = [] + for path in paths: + with os.fdopen(os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK | os.O_CLOEXEC), 'rb') as stream: + require(stat.S_ISREG(os.fstat(stream.fileno()).st_mode), 'task_holder_members') + raw = stream.read(4097) + require(len(raw) <= 4096 and all(re.fullmatch(rb'[0-9]{1,10}', row) for row in raw.splitlines()), + 'task_holder_members') + members.extend(int(row) for row in raw.splitlines()) + require(members == [self.pid], 'task_holder_members') + self.check() + + def memory(self): + self.alone() + rows = [line.split() for line in self.read('memory.events').splitlines()] + require(rows and all(len(row) == 2 and re.fullmatch(rb'[0-9]{1,20}', row[1]) for row in rows), + 'task_holder_counter') + events = dict(rows) + require(len(events) == len(rows) and b'oom_kill' in events, 'task_holder_counter') + value = {} + for key, name in (('maximum', 'memory.max'), ('swap', 'memory.swap.max'), ('peak', 'memory.peak')): + raw = self.read(name) + require(re.fullmatch(rb'[0-9]{1,20}\n?', raw), 'task_holder_counter') + value[key] = int(raw) + value['oom_kill'] = int(events[b'oom_kill']) + self.alone() + return value + + +def hold(command, path, receipt, nonce, deadline, parent_pid): + """Internal seam for inert child tests; the executable CLI never accepts argv.""" + remaining(deadline) + binding('ready', nonce, 0, None) + child = subprocess.Popen(command) + status = None + try: + status = child.wait(timeout=remaining(deadline)) + try: + digest = hashlib.sha256(receipt_bytes(receipt)).hexdigest() + except (OSError, ValueError, TypeError): + # Preserve the joined child's original status even when receipt + # ownership, metadata or readability fails. Null is no task proof. + digest = None + info = path.lstat() + require(stat.S_ISSOCK(info.st_mode) and info.st_uid == os.getuid() + and stat.S_IMODE(info.st_mode) == 0o600, 'task_holder_socket') + with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as channel: + channel.settimeout(remaining(deadline)) + channel.connect(str(path)) + peer(channel, parent_pid) + send_frame(channel, binding('ready', nonce, status, digest), deadline) + require(receive_frame(channel, deadline) == binding('release', nonce, status, digest), + 'task_holder_release') + return status if status != 0 or digest is not None else 1 + except (OSError, ValueError, TypeError, subprocess.SubprocessError): + # An observation failure cannot turn a failed task into a success or + # overwrite its actual nonzero/signal status. A successful task remains + # non-accepted when the bounded release protocol fails. + return status if status not in (None, 0) else 1 + finally: + if child.poll() is None: + child.terminate() + try: + child.wait(timeout=2) + except subprocess.TimeoutExpired: + child.kill() + child.wait(timeout=2) + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument('--deadline-monotonic-ns', type=int, required=True) + parser.add_argument('--handshake-nonce', required=True) + parser.add_argument('--parent-pid', type=int, required=True) + parser.add_argument('--inference-backend', choices=('llama_cpp_bf16_v1',), required=True) + args = parser.parse_args() + spec = importlib.util.spec_from_file_location('task_holder_trial', Path(__file__).with_name('smoke_opencode_inference.py')) + trial = importlib.util.module_from_spec(spec) + spec.loader.exec_module(trial) + require(os.getuid() > 0 and trial.pwd.getpwuid(os.getuid()).pw_name == 'vpci' + and socket.gethostname() == 'volparossa-alpha' + and trial.run(['systemd-detect-virt', '--vm'], text=True).stdout.strip() == 'kvm', + 'disposable_guest_required') + require(remaining(args.deadline_monotonic_ns) <= TASK_SECONDS) + binding('ready', args.handshake_nonce, 0, None) + status = hold(trial.task_arguments(args.inference_backend), trial.BASE / CONTROL, + trial.BASE / 'task.json', args.handshake_nonce, args.deadline_monotonic_ns, args.parent_pid) + if status < 0: + # Preserve systemd Result=signal/ExecMainStatus for this own process only. + if -status not in (signal.SIGKILL, signal.SIGSTOP): + signal.signal(-status, signal.SIG_DFL) + os.kill(os.getpid(), -status) + return status + + +if __name__ == '__main__': + try: + sys.exit(main()) + except (OSError, ValueError, TypeError, subprocess.SubprocessError): + sys.exit(1) diff --git a/scripts/smoke_opencode_inference.py b/scripts/smoke_opencode_inference.py index c31bdad..f943806 100644 --- a/scripts/smoke_opencode_inference.py +++ b/scripts/smoke_opencode_inference.py @@ -298,7 +298,7 @@ def pack(args): files = {f'code/{path.relative_to(ROOT)}': path for path in sorted((ROOT / 'src').glob('*.cjs'))} names = ['src/opencode-cooperative-tool.js', 'scripts/opencode_session.py', 'scripts/opencode_session.cjs', 'scripts/smoke_opencode_inference.cjs', 'scripts/smoke_opencode_inference.py', - 'scripts/opencode_native_cpu.py', + 'scripts/opencode_native_cpu.py', 'scripts/opencode_task_holder.py', 'third_party/opencode.json', 'third_party/opencode-LICENSE.txt', 'third_party/opencode-build-tools.json', 'patches/opencode-no-runtime-installs.patch', 'LICENSE', 'THIRD_PARTY_LICENSES.md'] @@ -535,6 +535,82 @@ def start(name, command, limit, seconds, public_source_build=False): '--setenv=RUST_LOG=off,volparossa::compute::private_diagnostic=debug', '--setenv=NO_COLOR=1', *command]) +def run_native_task(profile, backend, report): + holder = module(INPUT / 'code/scripts/opencode_task_holder.py', 'opencode_task_holder_inputs') + nonce, started = os.urandom(32).hex(), time.monotonic_ns() + deadline = started + 2550 * 1_000_000_000 + owner_deadline = started + 2565 * 1_000_000_000 + path = BASE / holder.CONTROL + with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as listener: + listener.bind(str(path)) + path.chmod(0o600) + listener.listen(1) + start(TASK_UNIT, [sys.executable, '-B', str(INPUT / 'code/scripts/opencode_task_holder.py'), + '--deadline-monotonic-ns', str(deadline), '--handshake-nonce', nonce, + '--parent-pid', str(os.getpid()), '--inference-backend', backend], 768 * 1024**2, 2550) + while True: + listener.settimeout(min(2.0, holder.remaining(deadline))) + try: + channel, _ = listener.accept() + break + except TimeoutError: + state = properties(TASK_UNIT) + require(state.get('ActiveState') == 'active' and state.get('SubState') == 'running', + 'task_holder_ended_before_ready') + with channel: + state = properties(TASK_UNIT) + require(state.get('ActiveState') == 'active' and state.get('SubState') == 'running' + and re.fullmatch('[0-9]{1,10}', state.get('MainPID', '')), 'task_holder_not_running') + pid = int(state['MainPID']) + holder.peer(channel, pid) + ready = holder.decode_ready(holder.receive_frame(channel, deadline), nonce) + status = ready['child_returncode'] + # Preserve the actual Node status even if later observation fails; + # the holder's systemd terminal status is checked separately below. + report['task_exit_status'] = abs(status) + primary = None + try: + if ready['receipt_sha256'] is not None: + raw = holder.receipt_bytes(BASE / 'task.json') + require(hashlib.sha256(raw).hexdigest() == ready['receipt_sha256'], 'task_receipt_changed') + report['task'] = json.loads(raw) + validate_task_selection(report['task'], profile, backend) + else: + require(status != 0, 'task_holder_missing_success_receipt') + group = holder.Group(cgroup(TASK_UNIT), pid) + try: + report['core_memory'] = memory(CORE_UNIT) + report['task_memory'] = group.memory() + require(report['task_memory']['maximum'] == 768 * 1024**2, 'task_memory_limit') + for field in ('core_memory', 'task_memory'): + require(report[field]['swap'] == report[field]['oom_kill'] == 0, 'resource_violation') + finally: + group.close() + except BaseException as error: + primary = error + raise + finally: + # Release is cleanup, never permission to accept a bad receipt. + # Its failure must not replace the original observation failure. + try: + holder.send_frame(channel, holder.binding('release', nonce, status, + ready['receipt_sha256']), deadline) + except (OSError, ValueError): + if primary is None: + raise + while True: + holder.remaining(owner_deadline) + state = properties(TASK_UNIT) + if state.get('SubState') == 'exited' or state.get('ActiveState') in ('failed', 'inactive'): + break + time.sleep(min(.1, holder.remaining(owner_deadline))) + expected_result = 'success' if status == 0 else 'signal' if status < 0 else 'exit-code' + report['task_unit_result'] = state.get('Result') if state.get('Result') in ( + 'success', 'exit-code', 'signal', 'timeout', 'oom-kill', 'resources') else 'other' + require(state.get('ExecMainStatus') == str(abs(status)) and state.get('Result') == expected_result + and empty(TASK_UNIT), 'task_holder_exit') + + def guest(args): backend = getattr(args, 'inference_backend', 'torch') profile = trial_profile(getattr(args, 'model_profile', MODEL), backend) @@ -638,22 +714,25 @@ def guest(args): require(int(state['MainPID']) > 0, 'core_process') report['phase'] = 'opencode-task' created.append(TASK_UNIT) - start(TASK_UNIT, task_arguments(backend), 768 * 1024**2, 2550) - deadline = time.monotonic() + 2565 - while time.monotonic() < deadline: - state = properties(TASK_UNIT) - if state.get('SubState') == 'exited' or state.get('ActiveState') in ('failed', 'inactive'): - break - time.sleep(2) - report['task_unit_result'] = state.get('Result') if state.get('Result') in ( - 'success', 'exit-code', 'signal', 'timeout', 'oom-kill', 'resources') else 'other' - report['task_exit_status'] = int(state.get('ExecMainStatus', '-1')) - if (BASE / 'task.json').exists(): - report['task'] = load(BASE / 'task.json', 32768) - validate_task_selection(report['task'], profile, backend) - for name, field in ((CORE_UNIT, 'core_memory'), (TASK_UNIT, 'task_memory')): - report[field] = memory(name) - require(report[field]['swap'] == report[field]['oom_kill'] == 0, 'resource_violation') + if backend != 'torch': + run_native_task(profile, backend, report) + else: + start(TASK_UNIT, task_arguments(backend), 768 * 1024**2, 2550) + deadline = time.monotonic() + 2565 + while time.monotonic() < deadline: + state = properties(TASK_UNIT) + if state.get('SubState') == 'exited' or state.get('ActiveState') in ('failed', 'inactive'): + break + time.sleep(2) + report['task_unit_result'] = state.get('Result') if state.get('Result') in ( + 'success', 'exit-code', 'signal', 'timeout', 'oom-kill', 'resources') else 'other' + report['task_exit_status'] = int(state.get('ExecMainStatus', '-1')) + if (BASE / 'task.json').exists(): + report['task'] = load(BASE / 'task.json', 32768) + validate_task_selection(report['task'], profile, backend) + for name, field in ((CORE_UNIT, 'core_memory'), (TASK_UNIT, 'task_memory')): + report[field] = memory(name) + require(report[field]['swap'] == report[field]['oom_kill'] == 0, 'resource_violation') report['core_diagnostics'] = private.service_diagnostic( BASE / (CORE_UNIT + '.log'), **({'memory_enabled': True} if backend != 'torch' else {})) require(report['task_exit_status'] == 0 and report.get('task', {}).get('passed') is True, 'task_failed') @@ -683,7 +762,11 @@ def guest(args): # The original worker monitor owns descendants; stopping the complete # cgroup joins nested native/model processes, not just their leaders. for name in reversed(created): - if cgroup(name).exists(): + # Native task counters were read while the joined-child holder was + # still the sole member. Do not reintroduce a post-exit cgroup race + # or overwrite that bound observation during unconditional cleanup. + observed_task = backend != 'torch' and name == TASK_UNIT and 'task_memory' in report + if not observed_task and cgroup(name).exists(): try: report['core_memory' if name == CORE_UNIT else 'native_build_memory' if name == NATIVE_UNIT else 'task_memory'] = ( diff --git a/tests/test_opencode_ci.py b/tests/test_opencode_ci.py index b89e9ed..460f827 100644 --- a/tests/test_opencode_ci.py +++ b/tests/test_opencode_ci.py @@ -220,10 +220,14 @@ def test_default_guest_keeps_reviewed_task_and_acceptance_with_closed_diagnostic # default. Keep the original fingerprint; do not replace the golden. native_blocks = [child for child in ast.walk(node) if isinstance(child, ast.If) and ast.unparse(child.test) == "backend != 'torch'"] - self.assertEqual(len(native_blocks), 2) + self.assertEqual(len(native_blocks), 3) for child in reversed(sorted(native_blocks, key=lambda item: item.lineno)): lines = source.splitlines(keepends=True) - del lines[child.lineno - node.lineno:child.end_lineno - node.lineno + 1] + # The task-holder branch has an explicit unchanged Torch body. + # Specialize to that body; keep the historical golden unchanged. + replacement = [line[4:] for line in lines[child.orelse[0].lineno - node.lineno: + child.end_lineno - node.lineno + 1]] if child.orelse else [] + lines[child.lineno - node.lineno:child.end_lineno - node.lineno + 1] = replacement source = ''.join(lines) source = specialize_memory_diagnostics(source) changes = ( @@ -250,6 +254,8 @@ def test_default_guest_keeps_reviewed_task_and_acceptance_with_closed_diagnostic "'core_memory' if name == CORE_UNIT else 'task_memory'"), ("(\n native_memory(profile['core_memory_bytes']) if name == NATIVE_UNIT else memory(name))", "memory(name)"), (",\n native_cpu=backend != 'torch'", ''), + (" observed_task = backend != 'torch' and name == TASK_UNIT and 'task_memory' in report\n", ''), + ('if not observed_task and cgroup(name).exists():', 'if cgroup(name).exists():'), ) for added, original in changes: self.assertIn(added, source) diff --git a/tests/test_opencode_task_holder.py b/tests/test_opencode_task_holder.py new file mode 100644 index 0000000..f52baea --- /dev/null +++ b/tests/test_opencode_task_holder.py @@ -0,0 +1,507 @@ +# SPDX-License-Identifier: GPL-3.0-only +"""Inert real-child/socket/counter fixtures, never systemd, a VM or a model.""" +import ast +from contextlib import ExitStack +import hashlib +import importlib.util +import inspect +import json +import os +from pathlib import Path +import signal +import socket +import subprocess +import sys +import tempfile +import time +from types import SimpleNamespace +import unittest +from unittest.mock import patch + +ROOT = Path(__file__).resolve().parents[1] + + +def module(name): + spec = importlib.util.spec_from_file_location(name, ROOT / 'scripts' / (name + '.py')) + value = importlib.util.module_from_spec(spec) + spec.loader.exec_module(value) + return value + + +H = module('opencode_task_holder') +T = module('smoke_opencode_inference') +NONCE = 'a' * 64 +HASH = 'b' * 64 +BACKEND = 'llama_cpp_bf16_v1' + + +def deadline(seconds=5): + return time.monotonic_ns() + int(seconds * 1e9) + + +def counters(root, pid): + root.mkdir() + for name, value in {'cgroup.procs': str(pid), 'memory.max': str(768 * 1024**2), + 'memory.swap.max': '0', 'memory.peak': '123456', + 'memory.events': 'low 0\nhigh 0\nmax 0\noom 0\noom_kill 0'}.items(): + (root / name).write_text(value + '\n') + + +# The test-only runner imports the real helper and uses only a harmless Python +# child. Production main has no command/path option and requires the fixed VM. +RUNNER = '''import importlib.util,json,pathlib,sys +s=importlib.util.spec_from_file_location('holder',sys.argv[1]); h=importlib.util.module_from_spec(s); s.loader.exec_module(h) +status=h.hold(json.loads(sys.argv[2]),pathlib.Path(sys.argv[3]),pathlib.Path(sys.argv[4]),sys.argv[5],int(sys.argv[6]),int(sys.argv[7])) +sys.exit(status) +''' + + +class Cases(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory(prefix='vp-task-') + self.addCleanup(self.temp.cleanup) + self.root = Path(self.temp.name) + self.receipt = self.root / 'task.json' + self.receipt.write_bytes(b'{"inert":true}\n') + self.receipt.chmod(0o600) + self.processes = [] + self.addCleanup(self.join_children) + + def join_children(self): + for process in self.processes: + if process.poll() is None: + process.kill() + process.wait(timeout=3) + + def spawn(self, path, *, status=0, nonce=NONCE, until=None, command=None): + command = command or [sys.executable, '-B', '-c', f'raise SystemExit({status})'] + process = subprocess.Popen([sys.executable, '-B', '-c', RUNNER, + str(ROOT / 'scripts/opencode_task_holder.py'), json.dumps(command), str(path), + str(self.receipt), nonce, str(until or deadline()), str(os.getpid())], + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + self.processes.append(process) + return process + + def listener(self): + path = self.root / H.CONTROL + server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + self.addCleanup(server.close) + server.bind(str(path)) + path.chmod(0o600) + server.listen(1) + server.settimeout(3) + return path, server + + def test_frame_roundtrip_and_closed_shape(self): + raw = H.binding('ready', NONCE, 0, HASH) + self.assertEqual(H.decode_ready(raw, NONCE)['receipt_sha256'], HASH) + value = json.loads(raw) + alternatives = [raw.replace(b'"version":1', b'"version":true'), raw + b' ', + raw.replace(b'"version":1', b'"version":1,"version":1'), + json.dumps(value | {'extra': 'private-canary'}).encode()] + for key, bad in [('kind', 'task_release'), ('version', 2), ('nonce', 'c' * 64), + ('child_returncode', True), ('child_returncode', 256), + ('receipt_sha256', 'private-canary')]: + alternatives.append((json.dumps(value | {key: bad}, sort_keys=True, separators=(',', ':')) + '\n').encode()) + for raw in alternatives: + with self.subTest(raw=raw), self.assertRaises((ValueError, TypeError)): + H.decode_ready(raw, NONCE) + for key in value: + with self.subTest(missing=key), self.assertRaises(ValueError): + H.decode_ready(json.dumps({k: v for k, v in value.items() if k != key}).encode(), NONCE) + + def test_frame_eof_bound_and_original_absolute_deadline(self): + for raw in (b'', b'x' * (H.FRAME_MAX + 1)): + left, right = socket.socketpair() + with left, right: + left.sendall(raw) + left.shutdown(socket.SHUT_WR) + with self.assertRaises(ValueError): + H.receive_frame(right, deadline()) + for until in (True, 0, deadline(-1), deadline(3000)): + with self.assertRaises(ValueError): + H.remaining(until) + left, right = socket.socketpair() + with left, right, self.assertRaises(TimeoutError): + H.receive_frame(right, deadline(.03)) + + def test_peer_binds_pid_uid_gid(self): + left, right = socket.socketpair() + with left, right: + H.peer(right, os.getpid()) + with self.assertRaisesRegex(ValueError, 'task_holder_peer'): + H.peer(right, os.getpid() + 100000) + + def test_joined_child_holds_original_group_until_exact_owner_release(self): + path, server = self.listener() + process = self.spawn(path) + channel, _ = server.accept() + with channel: + H.peer(channel, process.pid) + ready = H.decode_ready(H.receive_frame(channel, deadline()), NONCE) + self.assertEqual(ready['child_returncode'], 0) + self.assertEqual(ready['receipt_sha256'], hashlib.sha256(self.receipt.read_bytes()).hexdigest()) + self.assertIsNone(process.poll()) + group_path = self.root / 'group' + counters(group_path, process.pid) + group = H.Group(group_path, process.pid) + try: + self.assertEqual(group.memory(), dict(maximum=768 * 1024**2, swap=0, peak=123456, oom_kill=0)) + finally: + group.close() + self.assertIsNone(process.poll()) + H.send_frame(channel, H.binding('release', NONCE, 0, ready['receipt_sha256']), deadline()) + self.assertEqual(process.wait(timeout=3), 0) + + def test_wrong_release_and_owner_eof_fail_closed(self): + for raw in (b'', H.binding('release', 'c' * 64, 0, HASH)): + with self.subTest(raw=raw), tempfile.TemporaryDirectory(dir=self.root) as folder: + path = Path(folder) / H.CONTROL + with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as server: + server.bind(str(path)); path.chmod(0o600); server.listen(1); server.settimeout(3) + process = self.spawn(path) + channel, _ = server.accept() + with channel: + H.receive_frame(channel, deadline()) + channel.sendall(raw); channel.shutdown(socket.SHUT_WR) + self.assertEqual(process.wait(timeout=3), 1) + + def test_nonzero_child_status_survives_failed_release(self): + path, server = self.listener() + process = self.spawn(path, status=17) + channel, _ = server.accept() + with channel: + ready = H.decode_ready(H.receive_frame(channel, deadline()), NONCE) + self.assertEqual(ready['child_returncode'], 17) + channel.shutdown(socket.SHUT_WR) + self.assertEqual(process.wait(timeout=3), 17) + + def test_failed_child_without_receipt_still_reports_original_status(self): + self.receipt.unlink() + path, server = self.listener() + process = self.spawn(path, status=19) + channel, _ = server.accept() + with channel: + ready = H.decode_ready(H.receive_frame(channel, deadline()), NONCE) + self.assertIsNone(ready['receipt_sha256']) + H.send_frame(channel, H.binding('release', NONCE, 19, None), deadline()) + self.assertEqual(process.wait(timeout=3), 19) + + def test_signal_status_is_closed_and_propagated_only_to_holder_itself(self): + path, server = self.listener() + process = self.spawn(path, command=[sys.executable, '-B', '-c', + 'import os,signal; os.kill(os.getpid(), signal.SIGTERM)']) + channel, _ = server.accept() + with channel: + ready = H.decode_ready(H.receive_frame(channel, deadline()), NONCE) + self.assertEqual(ready['child_returncode'], -signal.SIGTERM) + H.send_frame(channel, H.binding('release', NONCE, -signal.SIGTERM, + ready['receipt_sha256']), deadline()) + # The inert runner exits normally with the negative result; production + # main instead reproduces the signal on its own process, tested below. + process.wait(timeout=3) + trial = SimpleNamespace(BASE=self.root, task_arguments=lambda _: ['fixed-node-command'], + pwd=SimpleNamespace(getpwuid=lambda _: SimpleNamespace(pw_name='vpci')), + run=lambda *a, **kw: SimpleNamespace(stdout='kvm\n')) + spec = SimpleNamespace(loader=SimpleNamespace(exec_module=lambda _: None)) + argv = ['holder', '--deadline-monotonic-ns', str(deadline()), '--handshake-nonce', NONCE, + '--parent-pid', str(os.getpid()), '--inference-backend', BACKEND] + with patch.object(sys, 'argv', argv), patch.object(H.importlib.util, 'spec_from_file_location', return_value=spec), \ + patch.object(H.importlib.util, 'module_from_spec', return_value=trial), \ + patch.object(H.os, 'getuid', return_value=985), \ + patch.object(H.socket, 'gethostname', return_value='volparossa-alpha'), \ + patch.object(H, 'hold', return_value=-signal.SIGTERM), \ + patch.object(H.signal, 'signal') as handler, patch.object(H.os, 'kill') as kill: + self.assertEqual(H.main(), -signal.SIGTERM) + handler.assert_called_once_with(signal.SIGTERM, signal.SIG_DFL) + kill.assert_called_once_with(os.getpid(), signal.SIGTERM) + + def test_success_without_receipt_retains_status_but_cannot_succeed(self): + self.receipt.unlink() + path, server = self.listener() + process = self.spawn(path) + channel, _ = server.accept() + with channel: + ready = H.decode_ready(H.receive_frame(channel, deadline()), NONCE) + self.assertEqual(ready['child_returncode'], 0) + self.assertIsNone(ready['receipt_sha256']) + H.send_frame(channel, H.binding('release', NONCE, 0, None), deadline()) + self.assertEqual(process.wait(timeout=3), 1) + + def test_original_deadline_terminates_and_joins_running_child(self): + pidfile = self.root / 'child.pid' + command = [sys.executable, '-B', '-c', + 'import os,pathlib,sys,time; pathlib.Path(sys.argv[1]).write_text(str(os.getpid())); time.sleep(60)', + str(pidfile)] + process = self.spawn(self.root / H.CONTROL, until=deadline(.3), command=command) + self.assertEqual(process.wait(timeout=3), 1) + pid = int(pidfile.read_text()) + with self.assertRaises(ProcessLookupError): + os.kill(pid, 0) + + def test_receipt_permissions_links_and_bounds_are_real(self): + self.assertEqual(H.receipt_bytes(self.receipt), b'{"inert":true}\n') + self.receipt.chmod(0o644) + with self.assertRaises(ValueError): H.receipt_bytes(self.receipt) + self.receipt.chmod(0o600) + linked = self.root / 'linked'; os.link(self.receipt, linked) + with self.assertRaises(ValueError): H.receipt_bytes(self.receipt) + linked.unlink(); linked.symlink_to(self.receipt) + with self.assertRaises(ValueError): H.receipt_bytes(linked) + for content in (b'', b'x' * 32769): + self.receipt.write_bytes(content) + with self.assertRaises(ValueError): H.receipt_bytes(self.receipt) + + def test_receipt_foreign_uid_and_path_replacement_rejected(self): + original = H.os.fstat + def foreign(fd): + values = list(original(fd)); values[4] += 1 + return os.stat_result(values) + with patch.object(H.os, 'fstat', side_effect=foreign), self.assertRaises(ValueError): + H.receipt_bytes(self.receipt) + original_lstat = Path.lstat + def replaced(path): + if path == self.receipt: + values = list(original_lstat(path)); values[1] += 1 + return os.stat_result(values) + return original_lstat(path) + with patch.object(Path, 'lstat', replaced), self.assertRaises(ValueError): + H.receipt_bytes(self.receipt) + + def test_group_missing_malformed_duplicate_and_linked_counters_rejected(self): + path = self.root / 'group'; counters(path, os.getpid()) + group = H.Group(path, os.getpid()) + try: + for name, content in [('memory.max', 'max\n'), ('memory.peak', '-1\n'), + ('memory.events', 'oom_kill 0\noom_kill 0\n'), + ('memory.events', 'oom 0\n'), ('memory.events', 'oom_kill unknown\n')]: + target = path / name; old = target.read_bytes(); target.write_text(content) + with self.subTest(name=name, content=content), self.assertRaises(ValueError): group.memory() + target.write_bytes(old) + target = path / 'memory.peak'; target.unlink() + with self.assertRaises(FileNotFoundError): group.memory() + target.symlink_to(self.receipt) + with self.assertRaises(OSError): group.memory() + finally: + group.close() + + def test_group_rejects_extra_descendant_or_foreign_members(self): + path = self.root / 'group'; counters(path, os.getpid()) + group = H.Group(path, os.getpid()) + try: + for members in ('', '1\n', f'{os.getpid()}\n1\n'): + (path / 'cgroup.procs').write_text(members) + with self.assertRaises(ValueError): group.memory() + (path / 'cgroup.procs').write_text(str(os.getpid())) + child = path / 'child'; child.mkdir(); (child / 'cgroup.procs').write_text('1\n') + with self.assertRaises(ValueError): group.memory() + finally: group.close() + + def test_group_rejects_replaced_path_or_parent_identity(self): + parent = self.root / 'parent'; parent.mkdir() + path = parent / 'group'; counters(path, os.getpid()) + group = H.Group(path, os.getpid()) + try: + path.rename(parent / 'old'); counters(path, os.getpid()) + with self.assertRaisesRegex(ValueError, 'cgroup_changed'): group.memory() + finally: group.close() + group = H.Group(path, os.getpid()) + try: + parent.rename(self.root / 'old-parent'); parent.mkdir(); counters(path, os.getpid()) + with self.assertRaisesRegex(ValueError, 'cgroup_changed'): group.memory() + finally: group.close() + + def run_parent(self, *, status=0, broken_counter=False, release_error=False, mismatch=False, + receipt_changed=False, interrupted=False, receipt_mode=0o600, malformed=False): + task = dict(version=2, output_constraint=T.OUTPUT_CONSTRAINT, model_profile=T.LARGE_MODEL, passed=status == 0) + self.receipt.write_text('{invalid private fixture' if malformed else json.dumps(task)) + self.receipt.chmod(receipt_mode) + group_path = self.root / 'group' + process = None + starts = [] + def start(name, command, limit, seconds): + nonlocal process + starts.append((name, command, limit, seconds)) + values = dict(zip(command[3::2], command[4::2])) + process = self.spawn(self.root / H.CONTROL, status=status, + nonce=values['--handshake-nonce'], until=int(values['--deadline-monotonic-ns'])) + counters(group_path, process.pid) + if broken_counter: (group_path / 'memory.peak').unlink() + def properties(name): + rc = process.poll() + if rc is None: return dict(ActiveState='active', SubState='running', MainPID=str(process.pid)) + # Reproduce the original systemd lifetime: after its last process + # leaves, the task cgroup may disappear despite RemainAfterExit. + for path in group_path.iterdir(): path.unlink() + group_path.rmdir() + return dict(ActiveState='active', SubState='exited', MainPID='0', + Result='success' if rc == 0 else 'exit-code', ExecMainStatus=str(rc + int(mismatch))) + report = {} + profile = T.trial_profile(T.LARGE_MODEL, BACKEND) + with ExitStack() as stack: + for key, value in dict(BASE=self.root, INPUT=self.root / 'inputs', module=lambda *_: H, + start=start, properties=properties, cgroup=lambda _: group_path, empty=lambda _: process.poll() is not None, + memory=lambda _: dict(maximum=11 * T.GIB, swap=0, peak=123, oom_kill=0)).items(): + stack.enter_context(patch.object(T, key, value)) + if release_error: + stack.enter_context(patch.object(H, 'send_frame', side_effect=OSError('synthetic release'))) + if receipt_changed: + stack.enter_context(patch.object(H, 'receipt_bytes', return_value=b'changed private fixture')) + if interrupted: + stack.enter_context(patch.object(H.Group, 'memory', side_effect=KeyboardInterrupt)) + try: + T.run_native_task(profile, BACKEND, report) + except BaseException as error: + return report, error, process, starts + return report, None, process, starts + + def test_real_parent_socket_group_observation_and_release_chain(self): + report, error, process, starts = self.run_parent() + self.assertIsNone(error) + self.assertEqual(process.wait(timeout=3), 0) + self.assertTrue(report['task']['passed']) + self.assertEqual(report['task_exit_status'], 0) + self.assertEqual(report['task_unit_result'], 'success') + self.assertEqual(report['task_memory']['peak'], 123456) + self.assertFalse((self.root / 'group').exists()) + # Same post-exit observation as the original implementation now fails: + with patch.object(T, 'cgroup', return_value=self.root / 'group'), self.assertRaises(FileNotFoundError): + T.memory(T.TASK_UNIT) + self.assertEqual(starts[0][2:], (768 * 1024**2, 2550)) + + def test_missing_counter_preserves_task_status_but_never_accepts(self): + report, error, process, _ = self.run_parent(broken_counter=True) + self.assertIsInstance(error, FileNotFoundError) + self.assertEqual(report['task_exit_status'], 0) + self.assertTrue(report['task']['passed']) + self.assertNotIn('task_memory', report) + self.assertNotIn('task_unit_result', report) + self.assertEqual(process.wait(timeout=3), 0) + + def test_release_failure_cannot_mask_first_observation_error(self): + report, error, process, _ = self.run_parent(broken_counter=True, release_error=True) + self.assertIsInstance(error, FileNotFoundError) + self.assertNotIn('task_memory', report) + self.assertEqual(process.wait(timeout=3), 1) + + def test_failed_child_survives_parent_counter_failure(self): + report, error, process, _ = self.run_parent(status=17, broken_counter=True) + self.assertIsInstance(error, FileNotFoundError) + self.assertEqual(report['task_exit_status'], 17) + self.assertFalse(report['task']['passed']) + self.assertEqual(process.wait(timeout=3), 17) + + def test_failed_child_with_invalid_receipt_mode_retains_status(self): + report, error, process, _ = self.run_parent(status=17, receipt_mode=0o644) + self.assertIsNone(error) + self.assertEqual(report['task_exit_status'], 17) + self.assertEqual(report['task_unit_result'], 'exit-code') + self.assertNotIn('task', report) + self.assertEqual(process.wait(timeout=3), 17) + # The unchanged guest task acceptance gate still refuses this result. + self.assertFalse(report['task_exit_status'] == 0 and report.get('task', {}).get('passed') is True) + + def test_successful_child_with_invalid_receipt_mode_is_not_accepted(self): + report, error, process, _ = self.run_parent(receipt_mode=0o644) + self.assertIsInstance(error, ValueError) + self.assertEqual(str(error), 'task_holder_missing_success_receipt') + self.assertEqual(report['task_exit_status'], 0) + self.assertNotIn('task', report) + self.assertNotIn('task_unit_result', report) + self.assertEqual(process.wait(timeout=3), 1) + + def test_failed_child_with_malformed_json_retains_status(self): + report, error, process, _ = self.run_parent(status=17, malformed=True) + self.assertIsInstance(error, json.JSONDecodeError) + self.assertEqual(report['task_exit_status'], 17) + self.assertNotIn('task', report) + self.assertNotIn('task_unit_result', report) + self.assertEqual(process.wait(timeout=3), 17) + + def test_successful_child_with_malformed_json_is_not_accepted(self): + report, error, process, _ = self.run_parent(malformed=True) + self.assertIsInstance(error, json.JSONDecodeError) + self.assertEqual(report['task_exit_status'], 0) + self.assertNotIn('task', report) + self.assertNotIn('task_unit_result', report) + # Holder exit preserves the actual successful child exit, but the + # parent rejected the receipt and never accepted a completed trial. + self.assertEqual(process.wait(timeout=3), 0) + + def test_terminal_unit_must_match_joined_child(self): + report, error, process, _ = self.run_parent(mismatch=True) + self.assertIsInstance(error, ValueError) + self.assertEqual(str(error), 'task_holder_exit') + self.assertEqual(process.wait(timeout=3), 0) + + def test_receipt_hash_mismatch_releases_holder_but_refuses_task(self): + report, error, process, _ = self.run_parent(receipt_changed=True) + self.assertIsInstance(error, ValueError) + self.assertEqual(str(error), 'task_receipt_changed') + self.assertNotIn('task', report) + self.assertNotIn('task_memory', report) + self.assertEqual(process.wait(timeout=3), 0) + + def test_owner_interruption_releases_holder_without_accepting(self): + report, error, process, _ = self.run_parent(interrupted=True) + self.assertIsInstance(error, KeyboardInterrupt) + self.assertNotIn('task_memory', report) + self.assertNotIn('task_unit_result', report) + self.assertEqual(process.wait(timeout=3), 0) + + def test_release_failure_after_valid_counters_still_refuses_completion(self): + report, error, process, _ = self.run_parent(release_error=True) + self.assertIsInstance(error, OSError) + self.assertIn('task_memory', report) + self.assertNotIn('task_unit_result', report) + self.assertEqual(process.wait(timeout=3), 1) + + def test_wiring_native_only_fixed_command_original_caps_and_cleanup(self): + parent = inspect.getsource(T.run_native_task) + self.assertIn('started + 2550 * 1_000_000_000', parent) + self.assertIn('started + 2565 * 1_000_000_000', parent) + self.assertEqual(parent.count('start(TASK_UNIT,'), 1) + self.assertNotIn('task_arguments(', parent) + main = inspect.getsource(H.main) + self.assertIn("choices=('llama_cpp_bf16_v1',)", main) + self.assertIn("trial.task_arguments(args.inference_backend)", main) + self.assertIn("trial.BASE / 'task.json'", main) + self.assertIn("'volparossa-alpha'", main) + self.assertIn("'vpci'", main) + self.assertIn('os.kill(os.getpid(), -status)', main) + self.assertIn("'scripts/opencode_task_holder.py'", inspect.getsource(T.pack)) + guest = inspect.getsource(T.guest) + self.assertIn("if backend != 'torch':\n run_native_task(profile, backend, report)\n else:", guest) + self.assertIn("require(report['task_exit_status'] == 0 and report.get('task', {}).get('passed') is True", guest) + final = guest.split(' finally:', 1)[1] + self.assertIn("unit(name, 'stop', check=False)", final) + self.assertIn('empty(name)', final) + # No command/path/limit CLI parameter was introduced in the helper. + parser = ast.parse(main) + options = [node.args[0].value for node in ast.walk(parser) if isinstance(node, ast.Call) + and isinstance(node.func, ast.Attribute) and node.func.attr == 'add_argument'] + self.assertEqual(set(options), {'--deadline-monotonic-ns', '--handshake-nonce', '--parent-pid', '--inference-backend'}) + + def test_final_cleanup_preserves_live_native_task_sample_without_skipping_stop(self): + tree = ast.parse(inspect.getsource(T.guest)) + condition = next(node.value for node in ast.walk(tree) if isinstance(node, ast.Assign) + and any(isinstance(target, ast.Name) and target.id == 'observed_task' + for target in node.targets)) + for backend in ('torch', BACKEND): + for name in (T.TASK_UNIT, T.CORE_UNIT, T.NATIVE_UNIT): + for observed in (False, True): + skip = eval(compile(ast.Expression(condition), '', 'eval'), + dict(backend=backend, name=name, TASK_UNIT=T.TASK_UNIT, + report={'task_memory': {}} if observed else {})) + self.assertEqual(skip, backend == BACKEND and name == T.TASK_UNIT and observed) + # Stop remains outside the observation guard; a missing observation is + # not replaced with zero and cannot skip the original cleanup. + units = next(node for node in ast.walk(tree) if isinstance(node, ast.For) + and isinstance(node.iter, ast.Call) and isinstance(node.iter.func, ast.Name) + and node.iter.func.id == 'reversed') + self.assertIsInstance(units.body[-1], ast.Expr) + self.assertEqual(units.body[-1].value.func.id, 'unit') + + +if __name__ == '__main__': + unittest.main()