"""One-shot Windows #308 evidence phases; never edits source or deletes targets.
Usage: python run.py {pool-claim,red,green,trace,pool-release}
Receipts/logs remain beside this file; private homes/temp remain outside Git.
"""
import argparse
import ctypes
from ctypes import wintypes
import hashlib
import json
import os
from pathlib import Path
import shutil
import stat
import subprocess
import sys
import tempfile
import time
from datetime import datetime, timezone

try:
    import psutil
except ImportError:
    psutil = None

OUT = Path(__file__).resolve().parent
ROOT = OUT.parents[3]
TREE = ROOT / '.worktrees' / '308-registry-process-lock'
TARGET = TREE / 'target'
HERTZ = ROOT / '.worktrees' / 'hertz-307-evidence'
PHASES = ('pool-claim', 'red', 'green', 'trace', 'pool-release')
GIB = 1024 ** 3
ADMISSION, EMERGENCY, GROWTH = 96 * GIB, 32 * GIB, 64 * GIB
DEADLINE = 3 * 60 * 60
RED_TEST = 'loaded_reaper_serializes_registration_without_resurrecting_expiry'
RED_FILTER = 'test(=' + RED_TEST + ')'
GREEN_FILTER = ('(package(spt-store) & (test(serving::tests::) | '
                'binary(serving_registry_two_process_int))) | '
                '(package(spt-daemon) & test(servehost::tests::)) | '
                '(package(spt) & (binary(webserve_attachment_e2e) | '
                'binary(webserve_cross_node_e2e)))')
PROCESS_TESTS = {RED_TEST,
    'later_reap_preserves_a_registration_published_while_it_was_waiting',
    'abrupt_holder_exit_releases_the_sentinel_for_the_waiting_reaper'}
FAIL_CLOSED = 'servehost::tests::unavailable_registry_lock_cannot_publish_an_add'
COMPOSITES = {
    'webserve_attachment_e2e': 'an_attachment_is_snapshot_served_fetched_back_and_named_by_its_message',
    'webserve_cross_node_e2e': 'a_peers_url_is_served_by_its_owner_through_the_local_listener',
}
BUILDERS = {'cargo.exe', 'cargo-nextest.exe', 'rustc.exe', 'rustdoc.exe',
            'cl.exe', 'link.exe', 'lld-link.exe', 'cmake.exe', 'ninja.exe',
            'msbuild.exe', 'sccache.exe', 'cc.exe', 'gcc.exe', 'clang.exe'}
CIM = r"""$ErrorActionPreference='Stop'; [Console]::OutputEncoding=[Text.UTF8Encoding]::new();
@(Get-CimInstance Win32_Process | ForEach-Object {
  [pscustomobject]@{pid=[int]$_.ProcessId; ppid=[int]$_.ParentProcessId;
  name=$_.Name; exe=$_.ExecutablePath; cmd=$_.CommandLine; cwd=$null;
  birth=if ($_.CreationDate) {([DateTimeOffset]$_.CreationDate).ToUnixTimeMilliseconds()/1000.0} else {$null}}
}) | ConvertTo-Json -Depth 4 -Compress"""


def now():
    return datetime.now(timezone.utc).isoformat()


def digest(path):
    h = hashlib.sha256()
    with path.open('rb') as source:
        for chunk in iter(lambda: source.read(1024 * 1024), b''):
            h.update(chunk)
    return h.hexdigest()


def save(path, value):
    path.write_text(json.dumps(value, indent=2, sort_keys=True) + '\n', encoding='utf-8')


def normalized(value):
    return str(value or '').replace('\\', '/').lower().rstrip('/')


def inside(path, parent):
    p, root = normalized(path), normalized(parent)
    return p == root or p.startswith(root + '/')


def reparse_guard(path):
    for entry in (path, *path.parents):
        if entry.exists() or entry.is_symlink():
            if entry.lstat().st_file_attributes & stat.FILE_ATTRIBUTE_REPARSE_POINT:
                raise RuntimeError('reparse point refused: ' + str(entry))


def target_bytes():
    reparse_guard(TARGET)
    total = 0
    if TARGET.exists():
        for root, dirs, files in os.walk(TARGET, followlinks=False):
            for name in dirs + files:
                path = Path(root) / name
                try:
                    info = path.lstat()
                except FileNotFoundError:
                    continue  # Cargo may rename temporary files during the census.
                if info.st_file_attributes & stat.FILE_ATTRIBUTE_REPARSE_POINT:
                    raise RuntimeError('target contains reparse point: ' + str(path))
                if stat.S_ISREG(info.st_mode):
                    total += info.st_size
    return total


def census():
    if psutil:
        rows = []
        for proc in psutil.process_iter():
            try:
                row = proc.as_dict(attrs=['pid', 'ppid', 'name', 'exe', 'cmdline', 'cwd', 'create_time'], ad_value=None)
                rows.append(dict(pid=row['pid'], ppid=row['ppid'], name=row['name'],
                    exe=row['exe'], cmd=subprocess.list2cmdline(row['cmdline'] or []),
                    cwd=row['cwd'], birth=row['create_time']))
            except (psutil.NoSuchProcess, psutil.AccessDenied):
                continue
    else:
        result = subprocess.run(['powershell.exe', '-NoProfile', '-NonInteractive', '-Command', CIM],
                                capture_output=True, timeout=45, check=True)
        rows = json.loads(result.stdout.decode('utf-8-sig'))
        if isinstance(rows, dict):
            rows = [rows]
    table = {row['pid']: row for row in rows}
    for row in rows:
        ancestors, seen = [], {row['pid']}
        parent = table.get(row['ppid'])
        while parent and parent['pid'] not in seen:
            if row['birth'] and parent['birth'] and parent['birth'] > row['birth']:
                break
            ancestors.append(parent['pid'])
            seen.add(parent['pid'])
            parent = table.get(parent['ppid'])
        row['ancestors'] = ancestors
        row['builder'] = (row['name'] or '').lower() in BUILDERS
        row['own_target'] = inside(row['exe'], TARGET)
        # Attribution is based on the process or an ancestor's actual path/cwd
        # or explicit path-bearing argv, never just a shared executable name.
        chain = [row] + [table[pid] for pid in ancestors]
        row['authorized_hertz'] = any(
            inside(p['cwd'], HERTZ) or inside(p['exe'], HERTZ) or
            normalized(HERTZ) + '/' in normalized(p['cmd']) or
            ('"' + normalized(HERTZ) + '"') in normalized(p['cmd'])
            for p in chain)
    return rows


def gate(rows):
    refused = [r for r in rows if r['own_target'] or (r['builder'] and not r['authorized_hertz'])]
    if refused:
        raise RuntimeError('unattributed builder or live own-target process: ' +
                           ', '.join(str(r['pid']) + ':' + str(r['name']) for r in refused))


def remember_children(rows, owned):
    # Retain birth identities after parents exit. A reused PID never inherits ownership.
    changed = True
    while changed:
        changed = False
        for row in rows:
            parent = next((r for r in rows if r['pid'] == row['ppid']), None)
            if (row['birth'] is not None and parent and
                    owned.get(parent['pid']) == parent['birth'] and
                    row['birth'] >= parent['birth'] and row['pid'] not in owned):
                owned[row['pid']] = row['birth']
                changed = True


def kill_verified(owned, events):
    # Open a handle, verify its birth, then terminate THAT handle (not a PID lookup).
    kernel = ctypes.WinDLL('kernel32', use_last_error=True)
    kernel.OpenProcess.argtypes = [wintypes.DWORD, wintypes.BOOL, wintypes.DWORD]
    kernel.OpenProcess.restype = wintypes.HANDLE
    kernel.GetProcessTimes.argtypes = [wintypes.HANDLE] + [ctypes.POINTER(wintypes.FILETIME)] * 4
    kernel.TerminateProcess.argtypes = [wintypes.HANDLE, wintypes.UINT]
    kernel.WaitForSingleObject.argtypes = [wintypes.HANDLE, wintypes.DWORD]
    kernel.CloseHandle.argtypes = [wintypes.HANDLE]
    try:
        remember_children(census(), owned)
    except Exception as exc:
        events.append(dict(time=now(), action='final child census', error=repr(exc)))
    # Parent first prevents further spawns; captured descendants are independently verified.
    for pid, birth in owned.items():
        event = dict(time=now(), pid=pid, birth=birth, action='TerminateProcess verified handle')
        handle = kernel.OpenProcess(0x1000 | 0x0001 | 0x00100000, False, pid)
        if not handle:
            event.update(result='not opened; no signal', winerror=ctypes.get_last_error())
        else:
            try:
                creation, exit_time, system, user = (wintypes.FILETIME() for _ in range(4))
                if not kernel.GetProcessTimes(handle, ctypes.byref(creation), ctypes.byref(exit_time),
                                              ctypes.byref(system), ctypes.byref(user)):
                    event.update(result='birth unreadable; no signal', winerror=ctypes.get_last_error())
                else:
                    actual = ((creation.dwHighDateTime << 32) | creation.dwLowDateTime) / 10000000 - 11644473600
                    if abs(actual - birth) > 0.002:
                        event.update(result='birth mismatch; no signal', actual_birth=actual)
                    else:
                        ok = bool(kernel.TerminateProcess(handle, 124))
                        event.update(result='terminated' if ok else 'termination failed',
                                     winerror=0 if ok else ctypes.get_last_error())
                        if ok:
                            event['wait_result'] = kernel.WaitForSingleObject(handle, 10000)
            finally:
                kernel.CloseHandle(handle)
        events.append(event)


class Phase:
    def __init__(self, name):
        self.name = name
        self.dir = OUT / name
        self.dir.mkdir()  # Exclusive: a phase receipt is never overwritten or retried.
        self.receipt = dict(phase=name, start=now(), cwd=str(TREE), target=str(TARGET),
                            commands=[], termination_events=[], status='running',
                            thresholds=dict(admission=ADMISSION, emergency=EMERGENCY,
                                            target_growth=GROWTH, producer_deadline_seconds=DEADLINE))
        self.env = {k: v for k, v in os.environ.items() if not k.upper().startswith(('OWL_', 'SPT_'))}
        self.receipt['scrubbed_names'] = sorted(k for k in os.environ if k.upper().startswith(('OWL_', 'SPT_')))
        self.receipt['driver_sha256'] = digest(Path(__file__))
        self.flush()

    def flush(self):
        save(self.dir / 'receipt.json', self.receipt)

    def metadata(self, *args):
        command = ['git', '-C', str(TREE), *args]
        result = subprocess.run(command, env=self.env, capture_output=True, timeout=60)
        self.receipt.setdefault('metadata_commands', []).append(dict(argv=command, native_exit=result.returncode))
        if result.returncode:
            raise RuntimeError('git metadata failed: ' + result.stderr.decode('utf-8', 'replace'))
        return result.stdout

    def source(self, tag):
        diff = self.dir / (tag + '-tracked.diff')
        diff.write_bytes(self.metadata('diff', 'HEAD', '--binary', '--no-ext-diff'))
        paths = self.metadata('ls-files', '-z', '--cached', '--others', '--exclude-standard').split(b'\0')
        hashes = {}
        for raw in paths:
            if raw:
                rel = os.fsdecode(raw)
                path = TREE / rel
                hashes[rel] = digest(path) if path.is_file() else None
        map_path = self.dir / (tag + '-source-sha256.json')
        save(map_path, hashes)
        return dict(head=self.metadata('rev-parse', 'HEAD').decode().strip(),
                    base=self.metadata('merge-base', 'HEAD', 'main').decode().strip(),
                    main=self.metadata('rev-parse', 'main').decode().strip(),
                    diff=str(diff), diff_sha256=digest(diff),
                    source_map=str(map_path), source_map_sha256=digest(map_path))

    def prepare(self):
        if os.name != 'nt':
            raise RuntimeError('Windows only')
        self.receipt['source_before'] = self.source('before')
        self.receipt['free_bytes_start'] = shutil.disk_usage(TREE).free
        self.receipt['processes_before'] = census()
        reparse_guard(TARGET)
        foreign = self.env.get('CARGO_TARGET_DIR')
        if foreign and (TREE / foreign).resolve() != TARGET.resolve():
            raise RuntimeError('foreign CARGO_TARGET_DIR refused: ' + foreign)
        if self.name == 'pool-claim':
            if TARGET.exists() and next(TARGET.iterdir(), None) is not None:
                raise RuntimeError('cold pool target must be absent or empty')
        else:
            claim = json.loads((OUT / 'pool-claim' / 'receipt.json').read_text(encoding='utf-8'))
            if claim['status'] != 'passed' or claim['source_before']['head'] != self.receipt['source_before']['head']:
                raise RuntimeError('successful same-HEAD pool claim required')
        predecessor = {'green': 'red', 'trace': 'green'}.get(self.name)
        if predecessor:
            previous = json.loads((OUT / predecessor / 'receipt.json').read_text(encoding='utf-8'))
            if previous['status'] != 'passed':
                raise RuntimeError('successful ' + predecessor + ' required')
        base = Path(os.environ['LOCALAPPDATA']) / 'Temp'
        base.mkdir(parents=True, exist_ok=True)
        reparse_guard(base)
        if inside(base.resolve(), ROOT):
            raise RuntimeError('private temporary storage must be outside repository')
        private = Path(tempfile.mkdtemp(prefix='spt-308-' + self.name + '-', dir=base))
        home, tmp = private / 'home', private / 'tmp'
        home.mkdir()
        tmp.mkdir()
        overlay = dict(CARGO_TARGET_DIR=str(TARGET), CARGO_BUILD_JOBS='2',
                       NEXTEST_PROFILE='ci-windows', SPT_HOME=str(home), TEMP=str(tmp), TMP=str(tmp),
                       SPT_INSTALL_NO_FIREWALL='1', SPT_TEST_EPHEMERAL_ADVISORY_PORTS='1',
                       SPT_ATTACH_IPC_DEADLINE_MS='30000', SPT_ATTACH_GATE_WATCHDOG_MS='120000')
        self.env.update(overlay)
        self.receipt['environment_allowlist'] = overlay
        self.receipt['unchanged_toolchain_environment'] = {
            k: self.env[k] for k in ('USERPROFILE', 'HOME', 'CARGO_HOME', 'RUSTUP_HOME', 'RUSTUP_TOOLCHAIN') if k in self.env}
        self.receipt['process_backend'] = 'psutil' if psutil else 'PowerShell CIM'
        self.flush()

    def run(self, label, argv, expected=0, timeout=DEADLINE):
        row = dict(label=label, argv=argv, cwd=str(TREE), start=now(), native_exit=None)
        self.receipt['commands'].append(row)
        stdout, stderr = self.dir / (label + '.stdout'), self.dir / (label + '.stderr')
        proc, owned = None, {}
        failure = None
        try:
            row['processes_before'] = census()
            row['free_bytes_start'] = shutil.disk_usage(TREE).free
            row['target_bytes_start'] = target_bytes()
            gate(row['processes_before'])
            if row['free_bytes_start'] < ADMISSION:
                raise RuntimeError('producer admission requires >=96 GiB free')
            if row['target_bytes_start'] > GROWTH:
                raise RuntimeError('cold target growth already exceeds 64 GiB')
            self.flush()
            with stdout.open('xb') as out, stderr.open('xb') as err:
                proc = subprocess.Popen(argv, cwd=TREE, env=self.env, stdin=subprocess.DEVNULL,
                                        stdout=out, stderr=err)
                row['pid'] = proc.pid
                started = time.monotonic()
                rows = census()
                root = next((r for r in rows if r['pid'] == proc.pid), None)
                if root and root['birth'] is not None:
                    owned[proc.pid] = root['birth']
                elif proc.poll() is None:
                    raise RuntimeError('cannot establish child birth identity')
                last_disk = 0
                while proc.poll() is None:
                    rows = census()
                    remember_children(rows, owned)
                    if time.monotonic() - started > timeout:
                        raise RuntimeError('producer deadline exceeded')
                    if time.monotonic() - last_disk >= 10:
                        free, size = shutil.disk_usage(TREE).free, target_bytes()
                        row.setdefault('disk_samples', []).append(dict(time=now(), free_bytes=free, target_bytes=size))
                        if free <= EMERGENCY or size > GROWTH:
                            raise RuntimeError('emergency free-space or target-growth bound reached')
                        last_disk = time.monotonic()
                        self.flush()
                    time.sleep(1)
                row['native_exit'] = proc.wait()
            if row['native_exit'] != expected:
                raise RuntimeError(f'{label}: native exit {row["native_exit"]}, expected {expected}')
        except BaseException as exc:
            failure = exc
            row['error'] = repr(exc)
            if proc:
                if proc.pid not in owned and proc.poll() is None:
                    event = dict(time=now(), pid=proc.pid,
                                 action='TerminateProcess original Popen child handle')
                    self.receipt['termination_events'].append(event)
                    try:
                        proc.kill()  # Windows Popen retains the original process handle.
                        event['result'] = 'terminated'
                        row['native_exit'] = proc.wait(timeout=10)
                    except Exception as cleanup_error:
                        event['error'] = repr(cleanup_error)
                kill_verified(owned, self.receipt['termination_events'])
                if proc.poll() is not None:
                    row['native_exit'] = proc.returncode
        finally:
            row['end'] = now()
            row['owned_birth_identities'] = owned
            for stream in (stdout, stderr):
                if stream.exists():
                    row[stream.suffix[1:]] = dict(path=str(stream), bytes=stream.stat().st_size, sha256=digest(stream))
            row['free_bytes_end'] = shutil.disk_usage(TREE).free
            row['target_bytes_end'] = target_bytes()
            row['processes_after'] = census()
            self.flush()
        if failure:
            raise failure
        gate(row['processes_after'])
        if row['free_bytes_end'] <= EMERGENCY or row['target_bytes_end'] > GROWTH:
            raise RuntimeError('producer ended outside disk bounds')
        return stdout, stderr

    def inventory(self, label, scope, expression):
        stdout, _ = self.run(label, ['cargo', 'nextest', 'list', *scope, '--profile', 'ci-windows',
                                     '-E', expression, '--message-format', 'json'])
        data = json.loads(stdout.read_bytes())
        selected = []
        for suite in data['rust-suites'].values():
            for name, case in suite['testcases'].items():
                if case['filter-match']['status'] == 'matches' and not case['ignored']:
                    if suite['status'] != 'listed':
                        raise RuntimeError('matching test belongs to skipped suite')
                    selected.append((suite['package-name'], suite['binary-name'], name))
        if not selected or not data['test-count']:
            raise RuntimeError('empty or zero-selected inventory refused')
        self.receipt[label + '_selected'] = selected
        self.flush()
        return set(selected)

    def execute(self):
        if self.name == 'pool-claim':
            self.run('claim', ['cargo', 'run', '-p', 'xtask', '--', 'pool-claim', '--pool',
                              str(TARGET), '--label', 'todlando-307-308-windows'])
        elif self.name == 'red':
            scope = ['-p', 'spt-store', '--test', 'serving_registry_two_process_int']
            selected = self.inventory('inventory', scope, RED_FILTER)
            if selected != {('spt-store', 'serving_registry_two_process_int', RED_TEST)}:
                raise RuntimeError('red must select exactly the one non-skipped regression')
            _, stderr = self.run('regression', ['cargo', 'nextest', 'run', *scope,
                '--profile', 'ci-windows', '-E', RED_FILTER, '--no-fail-fast', '--retries', '0',
                '--failure-output', 'immediate-final'], expected=100)
            if b'sentinel was unlocked' not in stderr.read_bytes():
                raise RuntimeError('native 100 lacked the independent fs2 witness; not expected red')
        elif self.name == 'green':
            self.run('fixture-build', ['cargo', 'build', '-p', 'mock-adapter', '--bin', 'capture-player'])
            if not (TARGET / 'debug' / 'capture-player.exe').is_file():
                raise RuntimeError('capture-player fixture missing after native success')
            scope = ['--workspace', '--lib', '--test', 'serving_registry_two_process_int',
                     '--test', 'webserve_attachment_e2e', '--test', 'webserve_cross_node_e2e']
            selected = self.inventory('inventory', scope, GREEN_FILTER)
            required = {('spt-store', 'serving_registry_two_process_int', name) for name in PROCESS_TESTS}
            required.update(('spt', binary, name) for binary, name in COMPOSITES.items())
            if not required <= selected or not any(p == 'spt-daemon' and t == FAIL_CLOSED for p, b, t in selected):
                raise RuntimeError('green inventory missing a required non-skipped process/fail-closed/composite test')
            if not any(p == 'spt-store' and t.startswith('serving::tests::') for p, b, t in selected):
                raise RuntimeError('green inventory missing serving unit tests')
            self.run('tests', ['cargo', 'nextest', 'run', *scope, '--profile', 'ci-windows',
                     '-E', GREEN_FILTER, '--no-fail-fast', '--retries', '0'])
        elif self.name == 'trace':
            stdout, _ = self.run('version', ['traceable-reqs', '--version'], timeout=60)
            if stdout.read_text(encoding='utf-8').strip() != 'traceable-reqs 0.4.1':
                raise RuntimeError('literal traceable-reqs 0.4.1 required')
            self.run('check', ['traceable-reqs', 'check', '--json'], timeout=600)
        else:
            xtask = TARGET / 'debug' / 'xtask.exe'
            if not xtask.is_file():
                raise RuntimeError('own built xtask.exe missing; never fall back to cargo')
            self.run('release', [str(xtask), 'pool-release', '--pool', str(TARGET)], timeout=120)


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('phase', choices=PHASES)
    args = parser.parse_args()
    phase = Phase(args.phase)
    code = 1
    try:
        phase.prepare()
        phase.execute()
        phase.receipt['status'] = 'passed'
        code = 0
    except BaseException as exc:
        phase.receipt.update(status='refused-or-failed', error=repr(exc))
    finally:
        try:
            phase.receipt['source_after'] = phase.source('after')
            phase.receipt['processes_after'] = census()
            phase.receipt['free_bytes_end'] = shutil.disk_usage(TREE).free
            phase.receipt['target_bytes_end'] = target_bytes()
            if phase.receipt.get('source_before') != phase.receipt['source_after']:
                # Filenames differ by tag; compare evidence hashes rather than artifact paths.
                before, after = phase.receipt.get('source_before', {}), phase.receipt['source_after']
                for key in ('head', 'base', 'main', 'diff_sha256', 'source_map_sha256'):
                    if before.get(key) != after[key]:
                        raise RuntimeError('source changed during phase: ' + key)
        except BaseException as exc:
            phase.receipt['finalization_error'] = repr(exc)
            phase.receipt['status'] = 'refused-or-failed'
            code = 1
        phase.receipt['end'] = now()
        phase.receipt['driver_exit'] = code
        phase.flush()
        print(str(phase.dir / 'receipt.json'))
    return code


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