#!/usr/bin/env python3
"""Real child/cgroup failures, only inside this fixture's delegated systemd unit."""
import errno
import os
from pathlib import Path
import signal
import resource
import select
import socket
import sys
import tempfile
import time
from unittest.mock import patch

sys.path.insert(0, str(Path(__file__).resolve().parents[2] / 'scripts'))
from c01_children import Children, RoleGroup, drop_credentials, process_group_path
import c01_children as lifecycle

assert os.geteuid() == 0
unit = Path(process_group_path(os.getpid())).parent.name
assert unit.startswith('opencode-c01-children-') and unit.endswith('.service')
assert os.environ.get('ARCHY_C01_CHILDREN_FIXTURE') == '1'


def settle(owner, close=True):
    deadline = time.monotonic() + 5
    while owner.children:
        assert not owner.poll(), 'child lifecycle reported a kernel failure'
        assert time.monotonic() < deadline, 'fixture children did not retire'
        time.sleep(0.005)
    if close:
        owner.close()


def waiting(control):
    parent = os.getppid()
    drop_credentials(65534, 65534, parent)
    control.send(b'ready')
    while True:
        signal.pause()


def exercise(name, action):
    owner = Children()
    try:
        action(owner)
        settle(owner)
        assert not any(owner.root.glob('c01-*'))
        print('PASS child ownership:', name, flush=True)
    except BaseException:
        owner.request_stop()
        settle(owner)
        raise


with tempfile.TemporaryDirectory(prefix='c01-children-') as temporary:
    root = Path(temporary); root.chmod(0o777)

    def ordinary(owner):
        def child(control):
            drop_credentials(65534, 65534, os.getppid())
            assert os.getresuid() == (65534, 65534, 65534)
            control.send(b'dropped')
        record = owner.spawn('application', child)
        record.channel.settimeout(2)
        assert record.channel.recv(32) == b'dropped'
        assert record.pidfd is not None
    exercise('permanent credential drop and actual bounded role cgroup', ordinary)

    def escaped_descendant(owner):
        def child(control):
            drop_credentials(65534, 65534, os.getppid())
            parent, ready = socket.socketpair()
            descendant = os.fork()
            if descendant == 0:
                parent.close()
                os.setsid()
                grandchild = os.fork()
                if grandchild:
                    os._exit(0)
                os.setsid()
                ready.send(str(os.getpid()).encode())
                while True:
                    signal.pause()
            ready.close(); escaped = parent.recv(32); parent.close()
            assert os.waitpid(descendant, 0)[0] == descendant
            control.send(escaped)
        record = owner.spawn('application', child)
        record.channel.settimeout(2)
        descendant = int(record.channel.recv(32))
        # Do not use PID liveness after reaping as an identity assertion.
        handle = os.pidfd_open(descendant)
        try:
            settle(owner, close=False)
            try:
                os.waitpid(descendant, os.WNOHANG)
            except ChildProcessError:
                pass
            else:
                raise AssertionError('escaped descendant was not reaped')
        finally:
            os.close(handle)
    exercise('double-fork setsid descendant killed and reaped after leader exit', escaped_descendant)

    for stage in ('pidfd', 'cgroup'):
        def admission_failure(owner, stage=stage):
            marker = root / (stage + '-body-ran')
            def forbidden(control):
                marker.write_text('ran')
            manager = patch.object(os, 'pidfd_open', side_effect=OSError(errno.EMFILE, 'fixture')) if stage == 'pidfd' else \
                patch.object(RoleGroup, 'attach', side_effect=OSError(errno.EPERM, 'fixture'))
            with manager:
                try:
                    owner.spawn('host-io', forbidden)
                except OSError:
                    pass
                else:
                    raise AssertionError('injected admission failure was accepted')
            assert owner.stopping and len(owner.children) == 1
            assert next(iter(owner.children.values())).pid is not None
            settle(owner, close=False)
            assert not marker.exists(), 'unregistered child crossed its execution barrier'
        exercise(stage + ' failure retains and retires the unadmitted direct child', admission_failure)

    def interrupted_registration(owner):
        original = os.pidfd_open
        def interrupted(pid):
            result = original(pid)
            os.kill(os.getpid(), signal.SIGTERM)
            return result
        with patch.object(os, 'pidfd_open', side_effect=interrupted):
            owner.spawn('authority', waiting)
        assert owner.stopping and len(owner.children) == 1
    exercise('termination during registration cannot escape ownership', interrupted_registration)

    def retained_waitability(owner):
        record = owner.spawn('host-io', waiting)
        record.channel.settimeout(2); assert record.channel.recv(32) == b'ready'
        owner.request_stop()
        # Model an unreapable observation, not an actual D-state filesystem.
        with patch.object(os, 'waitid', return_value=None):
            assert owner.poll() == []
            assert owner.children['host-io'] is record
            try:
                owner.spawn('administrator', waiting)
            except ValueError:
                pass
            else:
                raise AssertionError('replacement created over retained worker')
    exercise('unreapable observation retains role and rejects replacement', retained_waitability)

    def channel_failure(owner):
        original = socket.socketpair
        first = original(socket.AF_UNIX, socket.SOCK_SEQPACKET)
        with patch.object(socket, 'socketpair', side_effect=[first, OSError(errno.EMFILE, 'fixture')]):
            try:
                owner.spawn('administrator', waiting)
            except OSError:
                pass
            else:
                raise AssertionError('socket allocation failure was accepted')
        assert not owner.children and all(channel.fileno() == -1 for channel in first)
        assert not (owner.root / 'c01-administrator').exists()
    exercise('pre-fork channel allocation failure releases only owned resources', channel_failure)

    def live_orphans(owner):
        def child(control):
            drop_credentials(65534, 65534, os.getppid())
            for _ in range(32):  # More than this role's entire eight-task budget.
                middle = os.fork()
                if middle == 0:
                    orphan = os.fork()
                    if orphan == 0:
                        os._exit(0)
                    os._exit(0)
                os.waitpid(middle, 0)
                control.send(b'round')
                assert control.recv(4) == b'next'
        record = owner.spawn('host-io', child)
        record.channel.settimeout(2)
        for _ in range(32):
            assert record.channel.recv(8) == b'round'
            deadline = time.monotonic() + 2
            while True:
                assert not owner.poll()
                assert record.status is None, 'live leader was prematurely retired'
                if (record.group.path / 'pids.current').read_text().strip() == '1':
                    break
                assert time.monotonic() < deadline, 'completed orphan consumed the live role PID budget'
                time.sleep(0.005)
            record.channel.send(b'next')
    exercise('live leader reclaims completed adopted orphans beyond PID budget', live_orphans)

    def waiting_tree(control):
        drop_credentials(65534, 65534, os.getppid())
        parent, ready = socket.socketpair()
        descendant = os.fork()
        if descendant == 0:
            parent.close(); os.setsid(); ready.send(b'ready')
            while True:
                signal.pause()
        ready.close(); assert parent.recv(5) == b'ready'; parent.close()
        control.send(str(descendant).encode())
        while True:
            signal.pause()

    def group_kill_failure(owner):
        record = owner.spawn('authority', waiting_tree)
        record.channel.settimeout(2); record.channel.recv(32)
        observer = select.poll(); observer.register(record.pidfd, select.POLLIN)
        with patch.object(RoleGroup, 'kill', side_effect=OSError(errno.EIO, 'fixture')):
            owner.request_stop()
            assert owner.poll(), 'group termination error was hidden'
            assert observer.poll(2000), 'group kill error suppressed pidfd leader signal'
            assert owner.children['authority'] is record and not record.group.empty()
            try:
                owner.spawn('authority', waiting)
            except ValueError:
                pass
            else:
                raise AssertionError('replacement escaped a retained populated group')
        settle(owner, close=False)
    exercise('group kill failure still signals leader and retains descendants', group_kill_failure)

    def exhausted_descriptors(owner):
        record = owner.spawn('authority', waiting_tree)
        record.channel.settimeout(2); descendant = int(record.channel.recv(32))
        descendant_handle = os.pidfd_open(descendant)
        observer = select.poll()
        observer.register(record.pidfd, select.POLLIN); observer.register(descendant_handle, select.POLLIN)
        original = resource.getrlimit(resource.RLIMIT_NOFILE)
        ceiling = max(int(name) for name in os.listdir('/proc/self/fd')) + 16
        held = []
        try:
            resource.setrlimit(resource.RLIMIT_NOFILE, (ceiling, original[1]))
            while True:
                try:
                    held.append(os.open('/dev/null', os.O_RDONLY | os.O_CLOEXEC))
                except OSError as error:
                    assert error.errno == errno.EMFILE
                    break
            owner.request_stop()
            owner.poll()  # proc scans can fail; reserved termination FDs cannot.
            exited = set()
            deadline = time.monotonic() + 2
            while len(exited) < 2:
                exited.update(fd for fd, _ in observer.poll(20))
                assert time.monotonic() < deadline, 'EMFILE prevented descendant/leader termination'
            assert owner.children['authority'] is record, 'unknown retirement was discarded'
        finally:
            for fd in held:
                os.close(fd)
            resource.setrlimit(resource.RLIMIT_NOFILE, original)
            os.close(descendant_handle)
    exercise('real descriptor exhaustion preserves whole-group termination', exhausted_descriptors)

    def cleanup_retry(owner):
        record = owner.spawn('administrator', waiting)
        record.channel.settimeout(2); assert record.channel.recv(32) == b'ready'
        owner.request_stop()
        with patch.object(RoleGroup, 'close', side_effect=OSError(errno.EBUSY, 'fixture')):
            deadline = time.monotonic() + 2
            while not record.reaped:
                owner.poll()
                assert time.monotonic() < deadline
                time.sleep(0.005)
            assert owner.children['administrator'] is record and record.pid is None and record.pidfd is None
            assert record.channel.fileno() == -1
            try:
                owner.close()
            except ValueError:
                pass
            else:
                raise AssertionError('owner closed over pending cgroup retirement')
        reused = os.open('/dev/null', os.O_RDONLY | os.O_CLOEXEC)
        try:
            with patch.object(os, 'kill', side_effect=AssertionError('released PID signalled')), \
                 patch.object(signal, 'pidfd_send_signal', side_effect=AssertionError('retired pidfd signalled')), \
                 patch.object(os, 'waitpid', side_effect=AssertionError('released PID waited')):
                assert owner.poll() == []
            os.fstat(reused)
            assert not owner.children and not record.group.path.exists()
        finally:
            os.close(reused)
    exercise('post-reap cleanup remains owned and retries without stale PID or FD reuse', cleanup_retry)

    def replenished_observations(owner):
        record = owner.spawn('host-io', waiting)
        record.channel.settimeout(2); assert record.channel.recv(32) == b'ready'
        snapshots = 0
        def replenished(path, *args, **kwargs):
            nonlocal snapshots
            assert str(path).endswith('/children')
            snapshots += 1
            assert snapshots <= 64, 'live adoption drain never yielded to control'
            return str(1_000_000 + snapshots)
        # Deterministic non-quiescing kernel-observation seam. No fake PID is
        # ever signaled/waited on: both wait calls are explicitly substituted.
        with patch.object(Path, 'read_text', replenished), \
             patch.object(lifecycle, 'process_group_path', return_value=record.group.kernel_path), \
             patch.object(os, 'waitid', return_value=object()), patch.object(os, 'waitpid', return_value=(1, 0)) as reap:
            assert owner._reap_adopted(record) is False
            assert snapshots < 64 and reap.call_count > 0
            owner.request_stop(); reap.reset_mock()
            assert owner._reap_adopted(record) is False
            reap.assert_not_called()
    exercise('non-quiescing adoption observation yields and honors stop', replenished_observations)

    def continuous_orphans(owner):
        def producer(control):
            drop_credentials(65534, 65534, os.getppid())
            control.send(b'ready')
            while True:
                try:
                    middle = os.fork()
                except BlockingIOError:
                    time.sleep(0.001)
                    continue
                if middle == 0:
                    try:
                        orphan = os.fork()
                        if orphan == 0:
                            os._exit(0)
                    finally:
                        os._exit(0)
                os.waitpid(middle, 0)
        record = owner.spawn('host-io', producer)
        record.channel.settimeout(2); assert record.channel.recv(32) == b'ready'
        for _ in range(16):
            assert not owner.poll()
            time.sleep(0.005)
        os.kill(os.getpid(), signal.SIGTERM)
        assert owner.stopping
    exercise('continuous real orphan producer remains stoppable', continuous_orphans)

    def zombie_behind_live_children(owner):
        def producer(control):
            drop_credentials(65534, 65534, os.getppid())
            for index in range(33):
                parent, ready = socket.socketpair()
                middle = os.fork()
                if middle == 0:
                    parent.close()
                    orphan = os.fork()
                    if orphan == 0:
                        ready.send(b'ready')
                        if index == 32:
                            os._exit(0)
                        while True:
                            signal.pause()
                    os._exit(0)
                ready.close(); assert parent.recv(5) == b'ready'; parent.close()
                os.waitpid(middle, 0)
            control.send(b'constructed')
            while True:
                signal.pause()
        record = owner.spawn('application', producer)
        record.channel.settimeout(5); assert record.channel.recv(32) == b'constructed'
        adopted = [int(text) for text in Path(f'/proc/self/task/{os.getpid()}/children').read_text().split()
                   if int(text) != record.pid]
        assert len(adopted) == 33
        assert all(os.waitid(os.P_PID, pid, os.WEXITED | os.WNOHANG | os.WNOWAIT) is None for pid in adopted[:32])
        deadline = time.monotonic() + 2
        while os.waitid(os.P_PID, adopted[-1], os.WEXITED | os.WNOHANG | os.WNOWAIT) is None:
            assert time.monotonic() < deadline
            time.sleep(0.005)
        for _ in range(4):
            assert owner.poll() == []
        assert (record.group.path / 'pids.current').read_text().strip() == '33', 'live prefix starved the waitable orphan'
        owner.request_stop()
    exercise('32 live adopted children cannot starve a later zombie', zombie_behind_live_children)