"""Fixed manager-owned executor. All authoritative transitions go to live owner."""
import fcntl
import importlib.util
import json
import os
import pathlib
import socket
import struct
import time

from d09_store import atomic, read, secure, sync_dir
from d09_wire import MAX, canonical, parse

# Same implementation as the authority, loaded only from the protected package.
spec = importlib.util.spec_from_file_location('d09_owner', pathlib.Path(__file__).with_name('d09-lifecycle-owner.py'))
owner = importlib.util.module_from_spec(spec)
spec.loader.exec_module(owner)


def call(incarnation, method, **fields):
    req = dict(incarnation_id=incarnation, method=method, **fields)
    with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as conn:
        # A package checkpoint can wait for the explicitly bounded manager-bus
        # proof window (max120s). This changes reply waiting, not authority life.
        conn.settimeout(150 if method == 'os_checkpoint' else 10)
        conn.connect(str(owner.PRIVATE))
        pid, uid, gid = struct.unpack('3i', conn.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12))
        if uid != 0:
            raise PermissionError('owner peer')
        conn.sendall(canonical(req))
        conn.shutdown(socket.SHUT_WR)
        raw = b''
        while True:
            block = conn.recv(8192)
            if not block:
                break
            raw += block
            if len(raw) > MAX:
                raise ValueError('owner response size')
        result = parse(raw)
        if 'value' not in result:
            raise PermissionError('owner refused continuity')
        return result


class Worker:
    def __init__(self, transaction, incarnation):
        self.transaction, self.incarnation = transaction, incarnation
        self.refresh()
        if transaction != self.value['transaction_id']:
            raise PermissionError('transaction mismatch')

    def refresh(self):
        result = call(self.incarnation, 'get')
        self.binding, self.value = result['binding'], result['value']

    def transition(self, phase, intent=None, consume=None, **fields):
        result = call(self.incarnation, 'transition', updates=dict(phase=phase, **fields),
                      intent=intent, consume=consume)
        self.binding, self.value = result['binding'], result['value']

    def observed_start(self, action):
        current = owner.service()
        if current['ActiveState'] != 'active' or not current['InvocationID'] or current['MainPID'] == '0':
            raise ValueError('activation not observed')
        # Read the kernel's live executable reference, not the mutable pathname.
        import hashlib
        with open('/proc/' + str(int(current['MainPID'])) + '/exe', 'rb') as executable:
            hashed = hashlib.file_digest(executable, 'sha256').hexdigest()
        artifacts = self.binding['artifacts' if action == 'APPLY' else 'restore_artifacts']
        if hashed != artifacts['backend']:
            raise ValueError('unauthorized running artifact')
        if current['InvocationID'] == self.value['receipt']['prior_invocation_id']:
            raise ValueError('prior invocation still running')
        try:
            owner.PERMIT.unlink()
            sync_dir(owner.RUN)
        except FileNotFoundError:
            pass
        return current['InvocationID']

    def disposition(self, name, action):
        dest = owner.destinations(self.binding)[name]
        spec = self.binding['objects'][name]
        suffix = '.d09-' + self.transaction
        staged = owner.snapshot(self.binding,name,'.new' if action=='APPLY' else '.restore')
        retained = owner.snapshot(self.binding,name,'.old' if action=='APPLY' else '.failed')
        wanted = spec['new' if action == 'APPLY' else 'restore']
        before = spec['old' if action == 'APPLY' else 'new']
        absent = owner.digest(b'ABSENT')
        phase = 'CONFIG_INTENT' if name == 'config' else 'PAYLOAD_INTENT'
        intent = dict(object=name, action=action, before=before, after=wanted,
                      stage=owner.object_digest(staged), retained=owner.object_digest(retained))
        current = owner.object_digest(dest)
        if action == 'RESTORE' and current == spec['old']:
            before = current
        old_intent = self.value['intent']
        # Already-desired bytes are completion only after an attributable intent.
        # Recovery may legitimately find the exact original preimage (including
        # absent config), but still journals/restores/reloads before activation.
        if current == wanted:
            if action == 'APPLY' and wanted != spec['old'] and not (old_intent and old_intent.get('object') == name):
                if self.value['receipt']['phase'] not in ('CONFIG_COMMITTED', 'PAYLOAD_COMMITTED', 'RELOAD_INTENT', 'RELOADED', 'START_INTENT', 'AWAITING_OBSERVATION'):
                    raise ValueError('unattributed postimage')
            self.transition(phase, intent=intent)
            return
        if current != before and not (current == absent and owner.object_digest(retained) == before and old_intent and old_intent.get('object') == name):
            raise ValueError('foreign payload/config')
        if wanted != absent and owner.object_digest(staged) != wanted:
            raise ValueError('missing/foreign staged payload')
        self.transition(phase, intent=intent)
        fd = secure(dest.parent, True)
        stage_fd = None
        try:
            stage_fd = secure(staged.parent, True)
            # Verify again under the externally provisioned writer exclusion.
            if owner.object_digest(dest) == before and before != absent:
                if retained.exists() or retained.is_symlink():
                    raise ValueError('retained evidence collision')
                os.rename(dest.name, retained.name, src_dir_fd=fd, dst_dir_fd=stage_fd)
                os.fsync(fd)
                os.fsync(stage_fd)
            if wanted != absent:
                os.rename(staged.name, dest.name, src_dir_fd=stage_fd, dst_dir_fd=fd)
                os.fsync(fd)
                os.fsync(stage_fd)
            if owner.object_digest(dest) != wanted:
                raise ValueError('postimage mismatch')
        finally:
            if stage_fd is not None:
                os.close(stage_fd)
            os.close(fd)

    def await_start(self, action):
        end = time.monotonic() + 130
        while time.monotonic() < end:
            self.refresh()
            current = owner.service()
            if current['ActiveState'] == 'active':
                return self.observed_start(action)
            if (self.value.get('intent') or {}).get('start_submission')=='ACKNOWLEDGED' and current['ActiveState'] == 'failed' and current['MainPID'] == '0' and current['Job'] in ('', '0', '0 /'):
                raise ValueError('manager observed failed start')
            time.sleep(.1)
        raise StartPending('one submitted start remains unresolved')

    def generation(self, method):
        result=call(self.incarnation,'access_generation',action=method)
        self.binding,self.value=result['binding'],result['value']
        return result['access_generation']

    def prepare_generation(self):
        fresh=self.value['receipt']['phase']=='STOPPED'
        if fresh:
            # Inspect until the actual old supervisor/role lifetime is retired.
            # No deadline can turn an unreaped role into transition authority.
            while not self.generation('inspect')['quiescent']:
                time.sleep(.1)
            self.transition('ACCESS_INTENT')
        state=self.generation('begin' if fresh else 'inspect')
        if state['state']!='PENDING':
            raise ValueError('C01 generation begin outcome requires external reconciliation')
        self.transition('ACCESS_PREPARED')

    def commit_generation(self):
        fresh=self.value['receipt']['phase']=='PAYLOAD_COMMITTED'
        if fresh:
            self.transition('ACCESS_COMMIT_INTENT')
        state=self.generation('commit' if fresh else 'inspect')
        if state['state']!='COMMITTED':
            raise ValueError('C01 generation commit outcome requires external reconciliation')
        self.transition('ACCESS_COMMITTED')

    def execute(self, key, request):
        if request['operation'] == 'ACK_OBSERVATION':
            result = call(self.incarnation, 'observation', digest=request['observer_receipt_digest'])
            obs = result['observation']
            self.transition('COMPLETE', consume=key, outcome='COMPLETE', reason='NONE',
                            namespace_observation_digest=obs['namespace_observation_digest'],
                            preservation_observation_digest=obs['preservation_observation_digest'])
            return
        action = 'APPLY' if request['operation'] == 'APPLY' else 'RESTORE'
        if 'os_update' in self.binding:
            import os_update_executor
            try:
                os_update_executor.stage(self.binding, action, 'prepare',
                    checkpoint=lambda: call(self.incarnation, 'os_checkpoint'))
            except Exception as exc:
                if self.value['receipt']['phase'] in ('DISPATCH_INTENT', 'HANDOFF_READY', 'EXECUTING'):
                    raise OSPreparationRefused('OS preparation failed before service effects') from exc
                raise
        r = self.value['receipt']
        if r['phase'] in ('AWAITING_OBSERVATION', 'COMPLETE', 'RETAINED') and r['config_action'] == action:
            if action == 'RESTORE' and request['reason'] == 'AUTOMATIC_FAILURE':
                self.transition('CONFLICT', consume=key, outcome='RECOVERY_REQUIRED', reason='PARTIAL_RECOVERY')
                return
            if owner.service()['InvocationID'] != r['observed_invocation_id']:
                raise ValueError('no repeat activation under stale invocation')
            self.transition(r['phase'], consume=key)
            return
        # A worker lost around start must observe the existing exact successor;
        # it may never stop it and blindly repeat the activation.
        if r['phase'] == 'START_INTENT':
            current = owner.service()
            if current['ActiveState'] == 'active' and current['InvocationID'] != r['prior_invocation_id']:
                self.transition('AWAITING_OBSERVATION', consume=key,
                                observed_invocation_id=self.observed_start(action))
                return
            invocation = self.await_start(action)
            self.transition('AWAITING_OBSERVATION', consume=key, observed_invocation_id=invocation)
            return
        if r['phase'] in ('DISPATCH_INTENT', 'HANDOFF_READY', 'COMPLETE', 'AWAITING_OBSERVATION'):
            self.transition('EXECUTING', config_action=action, outcome='PENDING', reason='NONE',
                            namespace_observation_digest=None, preservation_observation_digest=None,
                            observed_invocation_id=None)
        if self.value['receipt']['phase'] in ('EXECUTING', 'STOP_INTENT'):
            current = owner.service()
            stopped = current['MainPID'] == '0' and current['ActiveState'] in ('inactive', 'failed') and current['Job'] in ('', '0', '0 /')
            if self.value['receipt']['phase'] == 'STOP_INTENT' and not stopped and current['Job'] not in ('', '0', '0 /'):
                raise ValueError('outstanding stop job must resolve before retry')
            self.transition('STOP_INTENT', prior_invocation_id=current['InvocationID'] or None,
                            config_action=action)
            try:
                owner.PERMIT.unlink()
                sync_dir(owner.RUN)
            except FileNotFoundError:
                pass
            if not stopped:
                owner.command('/usr/bin/systemctl', 'stop', '--no-block', owner.SERVICE)
            while True:
                self.refresh()
                current = owner.service()
                if current['MainPID'] == '0' and current['ActiveState'] in ('inactive', 'failed') and current['Job'] in ('', '0', '0 /'):
                    break
                time.sleep(.1)
            self.transition('STOPPED')
        if 'c01' in self.binding and self.value['receipt']['phase'] in ('STOPPED','ACCESS_INTENT'):
            self.prepare_generation()
        if 'os_update' in self.binding and self.value['receipt']['phase'] in ('STOPPED', 'ACCESS_PREPARED'):
            import os_update_executor
            os_update_executor.stage(self.binding, action, 'apply',
                checkpoint=lambda: call(self.incarnation, 'os_checkpoint'))
        phase = self.value['receipt']['phase']
        if phase in ('STOPPED', 'ACCESS_PREPARED', 'CONFIG_INTENT'):
            if action == 'APPLY':
                helper = pathlib.Path('/usr/libexec/archipelago-d09-prepare-workspace')
                if owner.digest(read(helper)) != self.binding['helper_digest']:
                    raise ValueError('helper changed after admission')
                # This helper deliberately rejects root and preserves existing
                # Work/history. Its executable and HOME are fixed, not wire data.
                if 'c01' in self.binding:
                    import d09_c01
                    d09_c01.workspace(self.binding)
                else:
                    owner.command('/usr/bin/setpriv', '--reuid=' + str(self.binding['requester_uid']),
                                  '--regid=' + str(self.binding['requester_gid']), '--init-groups',
                                  '/usr/bin/env', '-i', 'HOME=/home/archipelago', 'PATH=/usr/bin:/bin',
                                  '/bin/bash', str(helper), '/home/archipelago')
            spec = self.binding['objects']['config']
            self.transition('CONFIG_INTENT', intent=self.value['intent'], config_action=action,
                            config_pre_digest=spec['old'], config_post_digest=spec['new'],
                            config_current_digest=owner.object_digest(owner.DESTINATIONS['config']))
            self.disposition('config', action)
            self.transition('CONFIG_COMMITTED', config_current_digest=owner.object_digest(owner.DESTINATIONS['config']))
        if self.value['receipt']['phase'] in ('CONFIG_COMMITTED', 'PAYLOAD_INTENT'):
            start = (self.value['intent'] or {}).get('object')
            names = ['frontend', 'backend'] + (['launch'] if 'c01' in self.binding else [])
            if start in names:
                names = names[names.index(start):]
            for name in names:
                self.disposition(name, action)
                # A committed per-object intent is retained until the next one.
                self.transition('PAYLOAD_INTENT', intent=dict(object=name, action=action, committed=True))
            self.transition('PAYLOAD_COMMITTED', frontend_restoration='SUCCEEDED' if action == 'RESTORE' else 'NOT_REQUIRED',
                            backend_restoration='SUCCEEDED' if action == 'RESTORE' else 'NOT_REQUIRED')
        if 'c01' in self.binding and self.value['receipt']['phase'] in ('PAYLOAD_COMMITTED','ACCESS_COMMIT_INTENT'):
            self.commit_generation()
        if self.value['receipt']['phase'] in ('PAYLOAD_COMMITTED', 'ACCESS_COMMITTED', 'RELOAD_INTENT'):
            reopening = self.value['receipt']['phase'] == 'RELOAD_INTENT'
            self.transition('RELOAD_INTENT')
            if reopening:
                if owner.command('/usr/bin/systemctl', 'show', owner.SERVICE, '-p', 'NeedDaemonReload', '--value') != 'no':
                    raise ValueError('reload result unresolved; no blind manager retry')
            else:
                owner.command('/usr/bin/systemctl', 'daemon-reload')
            effective = owner.effective_configuration()
            expected = self.binding['effective_config'][action]
            if owner.digest(effective.encode()) != expected:
                raise ValueError('effective config differs from admission')
            self.transition('RELOADED', effective_config_observation_digest=expected)
        if self.value['receipt']['phase'] == 'RELOADED':
            self.transition('START_INTENT',intent={'start_submission':'INTENT','action':action})
            atomic(owner.PERMIT, canonical({'transaction_id': self.transaction}), 0o600)
            # Exactly one ordinary start. Lost replies leave START_INTENT.
            owner.command('/usr/bin/systemctl', 'start', '--no-block', owner.SERVICE)
            self.transition('START_INTENT',intent={'start_submission':'ACKNOWLEDGED','action':action,'manager_job':owner.service()['Job']})
            invocation = self.await_start(action)
            self.transition('AWAITING_OBSERVATION', consume=key, observed_invocation_id=invocation)

    def reactivate(self):
        """Externally consumed one-use grant for the retained exact generation."""
        r = self.value['receipt']
        action = r['config_action']
        if r['phase'] == 'START_INTENT':
            # Lost manager/owner acknowledgement: observation only, never start.
            invocation = self.observed_start(action)
            self.transition('AWAITING_OBSERVATION', observed_invocation_id=invocation)
            return
        if r['phase'] != 'REACTIVATION_READY' or action not in ('APPLY', 'RESTORE'):
            raise ValueError('invalid external activation phase')
        current = owner.service()
        if current['MainPID'] != '0' or current['ActiveState'] not in ('inactive', 'failed') or current['Job'] not in ('', '0', '0 /'):
            raise ValueError('activation process/job not drained')
        generation = 'new' if action == 'APPLY' else 'restore'
        for name, dest in owner.destinations(self.binding).items():
            if owner.object_digest(dest) != self.binding['objects'][name][generation]:
                raise ValueError('retained generation changed')
        effective = owner.effective_configuration()
        if owner.digest(effective.encode()) != self.binding['effective_config'][action]:
            raise ValueError('retained effective configuration changed')
        self.transition('START_INTENT', intent={'start_submission':'INTENT','action':action}, effective_config_observation_digest=owner.digest(effective.encode()))
        atomic(owner.PERMIT, canonical({'transaction_id': self.transaction,
            'execution_grant_digest': self.value['activation']['execution_grant_digest']}))
        owner.command('/usr/bin/systemctl', 'start', '--no-block', owner.SERVICE)
        self.transition('START_INTENT',intent={'start_submission':'ACKNOWLEDGED','action':action,'manager_job':owner.service()['Job']})
        self.transition('AWAITING_OBSERVATION', observed_invocation_id=self.await_start(action))


class StartPending(Exception):
    pass


class OSPreparationRefused(Exception):
    pass


def run(transaction, incarnation):
    lock = os.open(owner.ROOT / 'lifecycle.lock', os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW, 0o600)
    fcntl.flock(lock, fcntl.LOCK_EX)
    worker = Worker(transaction, incarnation)
    try:
        if worker.value.get('activation'):
            worker.reactivate()
            worker.refresh()
        while worker.value['inbox']:
            key = worker.value['inbox'][0]
            worker.execute(key, worker.value['requests'][key]['request'])
            worker.refresh()
    except StartPending:
        # Preserve the exact intent/permit/notice. No timeout authorizes a start.
        return
    except OSPreparationRefused:
        # A rejected dependency/repository plan must not enter the generic
        # post-effects recovery path, which deliberately stops the manager.
        worker.transition('CONFLICT', intent=worker.value['intent'], outcome='RECOVERY_REQUIRED', reason='PARTIAL_RECOVERY')
        raise
    except Exception:
        # A definitively failed ordinary activation may use the admitted failure
        # scope once. Unknown manager jobs/processes must remain held instead.
        r = worker.value['receipt']
        if worker.value.get('activation') and r['phase'] == 'START_INTENT':
            try:
                invocation = worker.observed_start(r['config_action'])
                worker.transition('AWAITING_OBSERVATION', observed_invocation_id=invocation)
                return
            except Exception:
                pass
        current = owner.service()
        if not worker.value.get('activation') and (worker.value.get('intent') or {}).get('start_submission')=='ACKNOWLEDGED' and r['phase'] == 'START_INTENT' and r['config_action'] == 'APPLY' and current['MainPID'] == '0' and current['ActiveState'] in ('inactive', 'failed') and current['Job'] in ('', '0', '0 /'):
            try:
                worker.transition('EXECUTING', config_action='RESTORE', outcome='PENDING', reason='AUTOMATIC_FAILURE')
                worker.execute(key, dict(worker.value['requests'][key]['request'], operation='RECOVER'))
                return
            except Exception:
                pass
        # Preserve all bytes and the outstanding notice. No local downgrade or
        # cleanup, even if the authority was lost while the worker was alive.
        try:
            owner.PERMIT.unlink()
            sync_dir(owner.RUN)
        except FileNotFoundError:
            pass
        owner.command('/usr/bin/systemctl', '--no-block', 'stop', owner.SERVICE)
        worker.transition('CONFLICT', intent=worker.value['intent'], outcome='RECOVERY_REQUIRED', reason='PARTIAL_RECOVERY')
        raise