"""One-shot #310 phases; reuse the measured Windows process/disk guard, never retry."""
import importlib.util
import json
from pathlib import Path
import sys

OUT = Path(__file__).resolve().parent
ROOT = OUT.parents[3]
SOURCE = ROOT / '.spt/preserved/308-registry-process-lock/windows-validation/run.py'
spec = importlib.util.spec_from_file_location('guard308', SOURCE)
g = importlib.util.module_from_spec(spec)
spec.loader.exec_module(g)
g.OUT = OUT
g.TREE = ROOT / '.worktrees/310-local-update-docs'
g.TARGET = g.TREE / 'target'
HELPER = SOURCE.parent / 'xtask-e6e9a8ca.exe'
AUTHORIZED = [ROOT / '.worktrees/hertz-307-evidence', ROOT / '.worktrees/gate-226-c9e80511']

def gate(rows):
    table = {r['pid']: r for r in rows}
    def authorized(row):
        chain = [row] + [table[p] for p in row['ancestors'] if p in table]
        # CI hides cwd/argv across service boundaries; attribute its real ancestry,
        # not an empty command line or a process name alone.
        ci = {g.normalized(p['exe']) for p in chain}
        if ('c:/actions-runner/bin.2.337.0/runnerservice.exe' in ci and
                'c:/actions-runner/bin.2.337.0/runner.worker.exe' in ci):
            return True
        return any(g.inside(p['cwd'], tree) or g.inside(p['exe'], tree) or
                   g.normalized(tree) + '/' in g.normalized(p['cmd']) or
                   '"' + g.normalized(tree) + '"' in g.normalized(p['cmd'])
                   for tree in AUTHORIZED for p in chain)
    refused = [r for r in rows if r['own_target'] or (r['builder'] and not authorized(r))]
    if refused:
        raise RuntimeError('unattributed builder/own-target process: ' + repr([(r['pid'], r['name'], r['cwd']) for r in refused]))
g.gate = gate

CLI_NAMES = [
    'update_messages_link_live_node_docs_or_preserve_offline_fallbacks',
    'applied_message_renders_semver_or_counter_fallback_and_states_applied',
    'resident_web_notice_is_shared_by_status_and_apply_only_for_incompatible_versions',
    'daemonless_apply_message_points_at_finish_or_start',
    'finish_message_states_fully_live_no_manual_restart',
]
EXPRESSION = '(package(spt) & (' + ' | '.join('test(' + n + ')' for n in CLI_NAMES) + ' | binary(docs_bundle_e2e))) | (package(spt-daemon) & (test(notif::tests::consent_) | test(docshost::tests::docs_url_encodes_the_canonical_node_prefix)))'
SCOPE = ['-p', 'spt', '-p', 'spt-daemon', '--lib', '--bin', 'spt', '--test', 'docs_bundle_e2e']

class Phase(g.Phase):
    def execute(self):
        self.receipt['driver_sha256'] = g.digest(Path(__file__))
        self.receipt['guard_driver_sha256'] = g.digest(SOURCE)
        self.receipt['authorized_concurrent_source_trees'] = list(map(str, AUTHORIZED))
        if self.name == 'pool-claim':
            assert g.digest(HELPER) == 'e59d3422081ae7406c90075cb145b958b379069e111646d7e8afe28828f9006a'
            self.run('claim', [str(HELPER), 'pool-claim', '--pool', str(g.TARGET), '--label', 'todlando-310-windows'], timeout=120)
        elif self.name == 'verify':
            self.run('build', ['cargo', 'build', '-p', 'spt', '-p', 'xtask', '--bin', 'spt', '--bin', 'xtask', '--example', 'update_links_smoke'])
            selected = self.inventory('inventory', SCOPE, EXPRESSION)
            names = {n.rsplit('::', 1)[-1] for p, b, n in selected}
            required = set(CLI_NAMES) | {'consent_changelog_uses_the_discovered_node_local_docs', 'consent_changelog_retains_release_url_without_live_docs', 'consent_decision_produces_notif_only_when_gated', 'docs_land_on_apply_and_docs_failure_never_touches_the_binary_outcome'}
            if not required <= names:
                raise RuntimeError('missing selected cells: ' + repr(required - names))
            self.run('tests', ['cargo', 'nextest', 'run', *SCOPE, '--profile', 'ci-windows', '-E', EXPRESSION, '--no-fail-fast', '--retries', '0'])
            smoke = g.TARGET / 'debug/examples/update_links_smoke.exe'
            for mode in ('online', 'offline', 'unavailable'):
                out, _ = self.run('smoke-' + mode, [str(smoke), mode, str(g.TARGET / 'debug/spt.exe')], timeout=120)
                if ('SMOKE_PASS ' + mode) not in out.read_text(encoding='utf-8'):
                    raise RuntimeError('smoke lacks its success predicate: ' + mode)
        elif self.name == 'docs':
            self.run('gen', [str(g.TARGET / 'debug/xtask.exe'), 'gen'], timeout=300)
        elif self.name == 'checks':
            self.run('docs-check', [str(g.TARGET / 'debug/xtask.exe'), 'check'], timeout=300)
            self.run('trace', ['traceable-reqs', 'check', '--json'], timeout=600)
        elif self.name == 'pool-release':
            self.run('release', [str(HELPER), 'pool-release', '--pool', str(g.TARGET)], timeout=120)
        else:
            raise RuntimeError('unknown phase')

def main():
    p = Phase(sys.argv[1])
    code = 1
    try:
        p.prepare()
        p.execute()
        p.receipt['status'] = 'passed'
        code = 0
    except BaseException as e:
        p.receipt.update(status='refused-or-failed', error=repr(e))
    finally:
        try:
            p.receipt['source_after'] = p.source('after')
            p.receipt['processes_after'] = g.census()
            p.receipt['free_bytes_end'] = g.shutil.disk_usage(g.TREE).free
            p.receipt['target_bytes_end'] = g.target_bytes()
            if p.name != 'docs':
                for field in ('head', 'diff_sha256', 'source_map_sha256'):
                    if p.receipt.get('source_before', {}).get(field) != p.receipt['source_after'][field]:
                        raise RuntimeError('source changed during phase: ' + field)
        except BaseException as e:
            p.receipt.update(status='refused-or-failed', finalization_error=repr(e))
            code = 1
        p.receipt.update(end=g.now(), driver_exit=code)
        p.flush()
        print(str(p.dir / 'receipt.json'))
    return code

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