"""#307 CLI custody lane: private own-tree pool, jobs=2, no live runner overlap."""
import importlib.util
import json
from pathlib import Path
import subprocess
import sys
sys.dont_write_bytecode = True
OUT = Path(__file__).resolve().parent
ROOT = OUT.parents[3]
BASE = OUT / 'final-base-driver.py'
spec = importlib.util.spec_from_file_location('custody_resource_guard', BASE)
g = importlib.util.module_from_spec(spec)
spec.loader.exec_module(g)
g.OUT = OUT
g.ROOT = ROOT
g.TREE = ROOT / '.worktrees/307-cli-resume-custody'
g.TARGET = g.TREE / 'target'
HEAD = 'f27fc4d5243608dd710f6e319198a2277e617bca'
CONTROLLER = ROOT / '.spt/preserved/302/todlando-step3-field/pool-controller.exe'
CONTROLLER_SHA = '2ec9a1f93ce4d52454c49a5c01593e6a292ca75286dc5798d26a0ed6858e01bf'
PHASES = ('pool-claim', 'red', 'green', 'rc', 'compat', 'clippy', 'trace', 'pool-release')
REGRESSIONS = {
    'cli_true_resume_defers_reconcile_until_spawn_returns',
    'cli_fresh_fallback_defers_reconcile_until_spawn_returns',
    'cli_spawn_failure_releases_custody_and_rolls_back_unbound',
}
census = g.census
original_remember_children = g.remember_children

def remember_children(rows, owned):
    original_remember_children(rows, owned)
    runners = [r for r in rows if (r['name'] or '').lower() == 'runner.worker.exe']
    if runners:
        raise RuntimeError('NATIVE HOLD: runner appeared during producer: ' + json.dumps(runners))

def gate(rows):
    refused = [r for r in rows if r['own_target'] or
        r['builder'] or (r['name'] or '').lower() == 'runner.worker.exe']
    if refused:
        raise RuntimeError('NATIVE HOLD: live own-target process or builder: ' + json.dumps(refused))

g.remember_children = remember_children
g.gate = gate

class CustodyPhase(g.Phase):
    def __init__(self, name):
        self.logical = name.split('@', 1)[0]
        super().__init__(name)

    def prepare(self):
        runs = subprocess.run(['gh', 'run', 'list', '--repo', 'BigscreenVR/spt-bs-core',
            '--limit', '8', '--json', 'databaseId,displayTitle,status,workflowName'],
            capture_output=True, timeout=60)
        self.receipt['current_ci_admission'] = dict(native_exit=runs.returncode,
            stdout=runs.stdout.decode('utf-8', 'replace'), stderr=runs.stderr.decode('utf-8', 'replace'))
        if runs.returncode:
            raise RuntimeError('current CI admission unreadable')
        golden = subprocess.run(['gh', 'run', 'view', '35313912102', '--repo',
            'BigscreenVR/spt-bs-core', '--json', 'status,conclusion,jobs'],
            capture_output=True, timeout=60)
        self.receipt['golden_admission'] = dict(native_exit=golden.returncode,
            stdout=golden.stdout.decode('utf-8', 'replace'), stderr=golden.stderr.decode('utf-8', 'replace'))
        self.flush()
        if golden.returncode:
            raise RuntimeError('golden admission unreadable')
        state = json.loads(golden.stdout)
        if any(j['status'] == 'in_progress' and (j['name'] == 'twohost-a' or 'Windows' in j['name']) for j in state['jobs']):
            raise RuntimeError('NATIVE HOLD: golden Windows job still in progress')
        super().prepare()
        self.env['TMPDIR'] = self.env['TMP']
        self.receipt['environment_allowlist']['TMPDIR'] = self.env['TMPDIR']
        if self.receipt['source_before']['head'] != HEAD:
            raise RuntimeError('expected frozen final source commit')
        if self.logical == 'red':
            # Regression-only baseline: product source must be byte-for-byte HEAD.
            product_diff = self.metadata('diff', 'HEAD', '--', 'crates/spt/src',
                'crates/spt-store/src', 'crates/spt-daemon/src')
            if product_diff.strip():
                raise RuntimeError('RED requires unchanged main product sources')
        self.receipt['wrapper_sha256'] = g.digest(Path(__file__))
        self.receipt['base_driver_sha256'] = g.digest(BASE)
        g.shutil.copyfile(Path(__file__), self.dir / 'driver.py')
        g.shutil.copyfile(BASE, self.dir / 'base-driver.py')
        self.flush()

    def execute(self):
        if self.logical in ('pool-claim', 'pool-release'):
            if g.digest(CONTROLLER) != CONTROLLER_SHA:
                raise RuntimeError('retained pool controller provenance mismatch')
            self.receipt['controller'] = dict(path=str(CONTROLLER), sha256=CONTROLLER_SHA)
            self.flush()
            command = [str(CONTROLLER), self.logical, '--pool', str(g.TARGET)]
            if self.logical == 'pool-claim':
                command += ['--label', 'todlando-307-cli-resume-custody']
            self.run(self.name, command, timeout=120)
        elif self.logical in ('red', 'green'):
            scope = ['-p', 'spt', '--test', 'cli_resume_custody']
            selected = self.inventory('inventory', scope, 'all()')
            required = {('spt', 'cli_resume_custody', n) for n in REGRESSIONS}
            if selected != required:
                raise RuntimeError('regression inventory differs from exact non-skipped cells')
            _, stderr = self.run('regression', ['cargo', 'nextest', 'run', *scope,
                '--profile', 'ci-windows', '--test-threads', '1', '--no-fail-fast',
                '--retries', '0', '--failure-output', 'immediate-final'],
                expected=100 if self.logical == 'red' else 0)
            if self.logical == 'red' and stderr.read_bytes().count(b'an unanswered CLI spawn must DEFER, not offline') < len(REGRESSIONS):
                raise RuntimeError('failure lacks the expected injected-reconcile witness for every branch')
        elif self.logical == 'rc':
            self.run('fixture', ['cargo', 'build', '-p', 'mock-adapter', '--bin', 'mock-session'])
            scope = ['-p', 'spt', '--test', 'rc_attach_truth']
            selected = self.inventory('inventory', scope, 'all()')
            if len(selected) != 6:
                raise RuntimeError('rc_attach_truth must select all six cells')
            self.run('tests', ['cargo', 'nextest', 'run', *scope, '--profile', 'ci-windows',
                '--test-threads', '1', '--no-fail-fast', '--retries', '0'])
        elif self.logical == 'compat':
            scope = ['-p', 'spt', '-p', 'spt-store', '-p', 'spt-daemon', '--lib', '--bins',
                '--test', 'resume_custody_aba', '--test', 'driven_by_selfheal', '--test', 'attach',
                '--test', 'attach_idempotent_replay', '--test', 'attach_wedge_e2e']
            expression = ('(package(spt-daemon) & (test(livehost::tests::) | '
                'binary(resume_custody_aba) | binary(driven_by_selfheal) | binary(attach) | binary(attach_idempotent_replay))) | '
                '(package(spt-store) & (test(resume_custody::tests::) | '
                'test(info::tests::resume_unbound) | test(info::tests::rollback_unbound))) | '
                '(package(spt) & (test(rc::tests::) | test(resume_resolution_tests::) | binary(attach_wedge_e2e)))')
            selected = self.inventory('inventory', scope, expression)
            for package, prefix in [('spt', 'rc::tests::'),
                ('spt-daemon', 'livehost::tests::'), ('spt-store', 'resume_custody::tests::')]:
                if not any(p == package and prefix in n for p, b, n in selected):
                    raise RuntimeError('required compatibility band absent: ' + prefix)
            for binary in ('resume_custody_aba', 'driven_by_selfheal', 'attach',
                'attach_idempotent_replay', 'attach_wedge_e2e'):
                if not any(b == binary for p, b, n in selected):
                    raise RuntimeError('required integration binary absent: ' + binary)
            self.run('tests', ['cargo', 'nextest', 'run', *scope, '--profile', 'ci-windows',
                '-E', expression, '--test-threads', '2', '--no-fail-fast', '--retries', '0'])
        elif self.logical == 'clippy':
            self.run('clippy', ['cargo', 'clippy', '--workspace', '--all-targets', '--', '-D', 'warnings'])
        elif self.logical == 'trace':
            self.run('version', ['traceable-reqs', '--version'], timeout=60)
            self.run('check', ['traceable-reqs', 'check', '--json'], timeout=600)

def main():
    if len(sys.argv) != 2 or sys.argv[1].split('@', 1)[0] not in PHASES:
        raise SystemExit('phase required: ' + ', '.join(PHASES))
    phase = CustodyPhase(sys.argv[1])
    code = 1
    try:
        phase.prepare()
        phase.execute()
        phase.receipt['status'] = 'passed'
        code = 0
    except BaseException as error:
        phase.receipt.update(status='refused-or-failed', error=repr(error))
    finally:
        try:
            phase.receipt['source_after'] = phase.source('after')
            phase.receipt['processes_after'] = census()
            phase.receipt['free_bytes_end'] = g.shutil.disk_usage(g.TREE).free
            phase.receipt['target_bytes_end'] = g.target_bytes()
            before, after = phase.receipt.get('source_before', {}), phase.receipt['source_after']
            # Main may advance through hertz's independent thin lane. The tested
            # tree's HEAD and bytes, not that other ref, are the proof identity.
            for key in ('head', 'base', 'diff_sha256', 'source_map_sha256'):
                if before.get(key) != after[key]:
                    raise RuntimeError('tested source changed during phase: ' + key)
        except BaseException as error:
            phase.receipt.update(finalization_error=repr(error), status='refused-or-failed')
            code = 1
        phase.receipt['end'] = g.now()
        phase.receipt['driver_exit'] = code
        phase.flush()
        print(str(phase.dir / 'receipt.json'))
    return code

if __name__ == '__main__':
    sys.exit(main())