scripts/map_references.py

#!/usr/bin/env python3
"""Deterministic static IXPL/resource index; does not compile or execute references."""
import argparse
import ast
from collections import Counter, defaultdict
import hashlib
import json
from pathlib import Path, PurePosixPath
import re
import sys

import references

ROOT = Path(__file__).resolve().parents[1]
OUT = ROOT / 'assets/analysis/source-map.json'
TEXT_SUFFIXES = {'.ixpl', '.txt', '.cfg', '.shader', '.bat'}


def lex(source):
    """Return offset/line-preserving tokens; comments disappear, strings stay opaque.

    IXPL delimiter strings (backslash quote delimiter ... delimiter quote) may
    contain arbitrary multiline text. This is a source index, not an IXPL parser.
    """
    tokens = []
    i, line = 0, 1
    while i < len(source):
        start, begin = i, line
        c = source[i]
        if c.isspace():
            line += c == '\n'
            i += 1
            continue
        if source.startswith('//', i):
            end = source.find('\n', i)
            i = len(source) if end < 0 else end
            continue
        if source.startswith('/*', i):
            end = source.find('*/', i + 2)
            if end < 0:
                raise ValueError(f'Unterminated comment at line {line}')
            i = end + 2
            line += source[start:i].count('\n')
            continue
        if source.startswith('\\"', i):
            if i + 2 >= len(source):
                raise ValueError(f'Incomplete delimiter string at line {line}')
            delimiter = source[i + 2]
            closing = {'[': ']', '{': '}', '(': ')', '<': '>'}.get(delimiter, delimiter) + '"'
            end = source.find(closing, i + 3)
            if end < 0:
                raise ValueError(f'Unterminated delimiter string at line {line}')
            i = end + 2
            kind = 'string'
        elif c in '\"\'':
            i += 1
            while i < len(source):
                if source[i] == '\\':
                    i += 2
                elif source[i] == c:
                    i += 1
                    break
                else:
                    i += 1
            else:
                raise ValueError(f'Unterminated quoted literal at line {line}')
            kind = 'string' if c == '"' else 'char'
        elif c.isalpha() or c in '_$@':
            i += 1
            while i < len(source) and (source[i].isalnum() or source[i] in '_$'):
                i += 1
            kind = 'identifier'
        elif c.isdigit():
            i += 1
            while i < len(source) and (source[i].isalnum() or source[i] == '.'):
                i += 1
            kind = 'number'
        else:
            i += 2 if source[i:i+2] in ('::', '+=', '==', '!=', '@=', '<=', '>=') else 1
            kind = 'symbol'
        tokens.append({'kind': kind, 'text': source[start:i], 'offset': start, 'end': i, 'line': begin})
        line += source[start:i].count('\n')
    return tokens


def masked(source, tokens):
    result = list(source)
    for i, c in enumerate(source):
        if c not in '\r\n':
            result[i] = ' '
    for token in tokens:
        if token['kind'] not in ('string', 'char'):
            result[token['offset']:token['end']] = source[token['offset']:token['end']]
    return ''.join(result)


def arguments(tokens, opening):
    """Split a call's balanced parentheses/arrays/braces, without parsing types."""
    stack, values, start = [], [], opening + 1
    pairs = {')': '(', ']': '[', '}': '{'}
    for i in range(opening, len(tokens)):
        t = tokens[i]['text'] if tokens[i]['kind'] == 'symbol' else ''
        if t in ('(', '[', '{'):
            stack.append(t)
        elif t in pairs:
            if not stack or stack[-1] != pairs[t]:
                raise ValueError(f'Unbalanced call at line {tokens[opening]["line"]}')
            stack.pop()
            if not stack:
                if i > start:
                    values.append(tokens[start:i])
                return values, i
        elif t == ',' and len(stack) == 1:
            values.append(tokens[start:i]); start = i + 1
    raise ValueError(f'Unterminated call at line {tokens[opening]["line"]}')


def text_of(source, tokens):
    return source[tokens[0]['offset']:tokens[-1]['end']] if tokens else ''


def constant_value(expression, values):
    node = ast.parse(expression, mode='eval').body
    def visit(n):
        if isinstance(n, ast.Constant) and type(n.value) is int:
            return n.value
        if isinstance(n, ast.Name) and n.id in values:
            return values[n.id]
        if isinstance(n, ast.UnaryOp) and isinstance(n.op, (ast.UAdd, ast.USub)):
            return visit(n.operand) * (-1 if isinstance(n.op, ast.USub) else 1)
        if isinstance(n, ast.BinOp) and isinstance(n.op, (ast.Add, ast.Sub)):
            return visit(n.left) + (visit(n.right) if isinstance(n.op, ast.Add) else -visit(n.right))
        raise ValueError('Unsupported or unresolved constant expression')
    return visit(node)


def scan_ixpl(source):
    tokens = lex(source)
    code = masked(source, tokens)
    aliases = dict(re.findall(r'\bnamespace\s+(\w+)\s*=\s*(\w+)\s*;', code))
    constants = []
    for m in re.finditer(r'\b((?:_?INT_\w+|CMD_\w+|coreVersion|akulaCoreVersion))\s*=\s*([^,;\r\n]+)', code):
        constants.append({'name': m[1], 'expression': m[2].strip(), 'line': source.count('\n', 0, m.start(1)) + 1})
    calls = []
    for i, token in enumerate(tokens[:-1]):
        if token['kind'] != 'identifier' or tokens[i+1]['text'] != '(':
            continue
        qualified = i >= 2 and tokens[i-1]['text'] == '::'
        if not qualified and token['text'] != 'interrupt':
            continue
        args, end = arguments(tokens, i+1)
        qualifier = tokens[i-2]['text'] if qualified else None
        calls.append({'qualifier': aliases.get(qualifier, qualifier), 'method': token['text'],
                      'line': token['line'], 'offset': token['offset'],
                      'arguments': [text_of(source, a) for a in args]})
    includes = [{'include': m[1], 'line': source.count('\n', 0, m.start(1)) + 1}
                for m in re.finditer(r'^[ \t]*#[ \t]*include[ \t]+"([^"\r\n]+)"', source, re.M)
                if any(t['text'] == '#' and t['offset'] <= m.start(1) and source.count('\n',0,t['offset']) + 1 == source.count('\n',0,m.start(1)) + 1 for t in tokens)]
    # Source-level wrappers, with complete balanced signatures (including tuple args).
    headers = []
    token_offsets = {t['offset']:i for i,t in enumerate(tokens)}
    known_types = {"void", "int", "long", "float", "double", "vec3", "string", "Tool", "Plane", "UV", "AABB", "Config"} | set(re.findall(r"\btype\s+(\w+)\s*=",code))
    type_pattern = "|".join(re.escape(t) for t in sorted(known_types))
    for match in re.finditer(r'^[ \t]*((?:(?:' + type_pattern + r')\b[^\n\r;={}]*?|\([^\n\r;={}]*\)[\[\]& \t]*))[ \t&]+([A-Za-z_]\w*)[ \t]*\(', code, re.M):
        prefix = match[1].strip()
        if not prefix or prefix.split()[0] in ('if', 'while', 'for', 'return', 'switch', 'else'):
            continue
        opening = token_offsets.get(match.end()-1)
        if opening is None:
            continue
        args, end = arguments(tokens, opening)
        headers.append({'name': match[2], 'line': source.count('\n',0,match.start(2))+1,
                        'offset': match.start(2), 'signature': re.sub(r'\s+', ' ', source[match.start(1):tokens[end]['end']]).strip(),
                        'declaration': tokens[end+1]['text'] == ';' if end+1 < len(tokens) else False})
    wrappers = []
    for call in calls:
        if call['method'] == 'interrupt':
            preceding = [h for h in headers if h['offset'] < call['offset'] and not h['declaration']]
            wrapper = preceding[-1] if preceding else None
            wrappers.append({**call, 'wrapper': wrapper})
    bindings = {}
    for i,t in enumerate(tokens[:-2]):
        if t['kind'] == 'identifier' and tokens[i+1]['text'] == '=':
            j, depth = i+2, 0
            while j < len(tokens):
                value = tokens[j]['text'] if tokens[j]['kind'] == 'symbol' else ''
                if value in ('(', '[', '{'): depth += 1
                elif value in (')', ']', '}'): depth -= 1
                if depth < 0 or (depth == 0 and value in (',', ';')): break
                j += 1
            expr = tokens[i+2:j]
            if expr and all(x['kind'] == 'string' or x['text'] == '+' or x['kind'] == 'identifier' for x in expr):
                bindings.setdefault(t['text'], []).append({'expression': text_of(source,expr), 'line':t['line']})
    return {'includes': includes, 'aliases': aliases, 'constants': constants, 'qualified_calls': calls,
            'interrupt_wrappers': wrappers, 'string_bindings': bindings}


def literal_string(expression, bindings, trail=()):
    """Resolve only an unambiguous concatenation of literals and local bindings.

    Scope/flow is not reconstructed: conflicting same-named declarations remain
    unresolved, even when a human could infer the active one.
    """
    parts = lex(expression)
    if not parts or len(parts) % 2 == 0 or any(t['text'] != '+' for t in parts[1::2]):
        return None
    result = ''
    for token in parts[::2]:
        value = token['text']
        if token['kind'] == 'string':
            if value.startswith('\\"'):
                result += value[3:-2]
            else:
                result += re.sub(r"\\([rnt\\\"'])", lambda m: {'r':'\r','n':'\n','t':'\t'}.get(m[1],m[1]), value[1:-1])
        elif token['kind'] == 'identifier' and value in bindings and value not in trail:
            candidates = [literal_string(x['expression'],bindings,(*trail,value)) for x in bindings[value]]
            if not candidates or any(x is None for x in candidates) or len(set(candidates)) != 1:
                return None
            result += candidates[0]
        else:
            return None
    return result


def category(path):
    p = PurePosixPath(path)
    if p.suffix == '.ixpl':
        if '/plugins/extra stuff/' in path: return 'experimental-plugin'
        if '/plugins/' in path: return 'plugin'
        if '/models/' in path: return 'model-generator'
        if '/include/core/' in path: return 'host-api'
        if '/include/tasks/' in path: return 'task-client'
        return 'ixpl-library'
    if p.suffix == '.exe': return 'executable-unexamined'
    if 'grammar' in p.name: return 'grammar'
    return {'.bat':'batch-script','.cfg':'configuration','.shader':'shader','.png':'image'}.get(p.suffix,'reference-text')


def index_build(root, inventory):
    files = {x['path']:x for x in inventory['files']}
    paths = set(files)
    include_roots = [PurePosixPath('tools') / p.replace('\\','/') for p in (root/'tools/includePaths.txt').read_text().splitlines() if p.strip()]
    # Normalize relative traversal in trusted include search directories only.
    include_roots = [Path(root/p).resolve().relative_to(root.resolve()).as_posix() for p in include_roots]
    records = []
    for path, item in sorted(files.items()):
        raw = (root/path).read_bytes()
        if hashlib.sha256(raw).hexdigest() != item['sha256']:
            raise ValueError(f'Modified/unhydrated source: {root/path}')
        record = {'path':path,'sha256':item['sha256'],'bytes':len(raw),'category':category(path),
                  'coverage':'lexically-indexed' if PurePosixPath(path).suffix.lower() in {'.ixpl','.shader','.cfg'} or category(path)=='grammar' else 'text-inventoried' if PurePosixPath(path).suffix.lower() in TEXT_SUFFIXES else 'inventory-only',
                  'owner':'Nanobrush reference research',
                  'followup':{'host-api':'P4.03','plugin':'P4.08','experimental-plugin':'P4.09','task-client':'P4.05','ixpl-library':'P4.07','model-generator':'P9.06','grammar':'P6.01','executable-unexamined':'P3.06','shader':'P9.01','configuration':'P8.09','image':'P9.05','batch-script':'P4.12'}.get(category(path),'P3.02')}
        if PurePosixPath(path).suffix.lower() in TEXT_SUFFIXES:
            try: text,encoding = raw.decode('utf-8-sig'),'utf-8-sig'
            except UnicodeDecodeError: text,encoding = raw.decode('cp1252'),'cp1252-fallback'
            record.update(encoding=encoding,lines=len(text.splitlines()))
            if path.endswith('.ixpl'):
                record.update(scan_ixpl(text))
                for edge in record['includes']:
                    candidates = [PurePosixPath(path).parent/edge['include'],*(PurePosixPath(d)/edge['include'] for d in include_roots)]
                    edge['candidates'] = sorted({c.as_posix() for c in candidates if c.as_posix() in paths})
                    edge['status'] = 'resolved-candidate' if len(edge['candidates'])==1 else 'ambiguous' if edge['candidates'] else 'missing'
            elif path.endswith('.shader'):
                record['sections'] = [{'name':m[1],'line':text.count('\n',0,m.start())+1} for m in re.finditer(r'\[@([^\]]+)\]',text)]
                record['uniforms'] = [{'type':m[1],'name':m[2],'line':text.count('\n',0,m.start())+1} for m in re.finditer(r'\buniform\s+(\w+)\s+(\w+)\s*;',text)]
            elif path.endswith('.cfg'):
                record['keys'] = [{'type':m[1],'key':m[2],'line':text.count('\n',0,m.start())+1} for m in re.finditer(r'^\s*(bool|int|long|float|double|vec3|string)\s+([\w.]+)\s*=',text,re.M)]
            elif record['category']=='grammar':
                record['rules'] = [{'name':m[1],'line':text.count('\n',0,m.start())+1} for m in re.finditer(r'^\s*(#?\w+)\s*=\s*\{',text,re.M)]
            elif path.endswith('.bat'):
                record['lines_text'] = [{'line':i,'text':line} for i,line in enumerate(text.splitlines(),1) if line.strip()]
        records.append(record)
    core = [r for r in records if r['category']=='host-api']
    values, pending = {},[(r,c) for r in core for c in r.get('constants',[])]
    while pending:
        progress = False
        for r,c in pending[:]:
            try: value = constant_value(c['expression'],values)
            except (ValueError,SyntaxError): continue
            values[c['name']] = value;c['value']=value;pending.remove((r,c));progress=True
        if not progress:break
    for r in records:
        if r['category']!='host-api':continue
        for wrapper in r.get('interrupt_wrappers',[]):
            argument = wrapper['arguments'][0]
            try: wrapper['interrupt_id'] = constant_value(argument,values)
            except (ValueError,SyntaxError): wrapper['interrupt_id']=None
    graph = {r['path']:sorted({p for e in r.get('includes',[]) for p in e['candidates']}) for r in records if r['path'].endswith('.ixpl')}
    for r in records:
        if r['category'] not in ('plugin','experimental-plugin','model-generator'):continue
        seen,stack = set(),list(graph[r['path']])
        while stack:
            path=stack.pop()
            if path in seen:continue
            seen.add(path);stack.extend(graph.get(path,[]))
        r['transitive_includes']=sorted(seen)
        r['core_calls']=sorted({c['method'] for c in r['qualified_calls'] if c['qualifier']=='Core'})
        r['interface_lookups']=[c for c in r['qualified_calls'] if c['method']=='getToolID']
        r['registrations']=[c for c in r['qualified_calls'] if c['method'] in ('initTool','registerPackage','setRuntimeOnAllLoaded','setRuntimeOnFinalization')]
        r['registered_names'] = []
        for call in r['registrations']:
            if call['method'] != 'initTool': continue
            args = call['arguments']
            name = literal_string(args[0],r['string_bindings']) if args else None
            if name is None and len(args)>1: name = literal_string(args[1],r['string_bindings'])
            r['registered_names'].append({'line':call['line'],'name':name,'arguments':args})
        r['resource_calls']=[c for c in r['qualified_calls'] if c['method'] in ('loadConfigFile','saveConfigFile','loadGUITexture','loadTextFile','saveTextFile','callProgram','saveWorldToFile','loadWorldFromFile')]
        for call in r['resource_calls']:
            literal = literal_string(call['arguments'][0],r['string_bindings']) if call['arguments'] else None
            call['literal_path'] = literal
            call['exact_name_candidates'] = sorted(p for p in paths if literal is not None and (p == literal.replace('\\','/') or PurePosixPath(p).name == PurePosixPath(literal.replace('\\','/')).name))
            call['status'] = 'dynamic' if literal is None else 'exact-name-candidate' if call['exact_name_candidates'] else 'not-present-by-exact-name'

    return {'build':inventory['build'],'archive_sha256':inventory['archive_sha256'],'include_search_roots':include_roots,
            'categories':dict(sorted(Counter(r['category'] for r in records).items())),
            'unresolved_core_constants':[{'path':r['path'],**c} for r,c in pending], 'files':records}


def differences(builds):
    result=[]
    for old,new in zip(builds,builds[1:]):
        a={f['path']:f['sha256'] for f in old['files']};b={f['path']:f['sha256'] for f in new['files']}
        added,removed=sorted(b.keys()-a.keys()),sorted(a.keys()-b.keys())
        result.append({'from':old['build'],'to':new['build'],'added':added,'removed':removed,
                       'changed':[p for p in sorted(a.keys()&b.keys()) if a[p]!=b[p]],
                       'identical_rename_candidates':[{'from':p,'to':q} for p in removed for q in added if a[p]==b[q]],
                       'identical_copy_candidates':[{'from':p,'to':q} for p in sorted(a) for q in added if p not in removed and a[p]==b[q]],
                       'modified_same_basename_candidates':[{'from':p,'to':q} for p in removed for q in added if PurePosixPath(p).name==PurePosixPath(q).name and a[p]!=b[q]]})
    return result


def validate_report(builds):
    """Corpus-level closure checks, separate from the permissive lexical scanner."""
    for build in builds:
        if build['unresolved_core_constants']:
            raise ValueError('Unresolved Core constants: ' + build['build'])
        core = [f for f in build['files'] if f['category'] == 'host-api']
        declared = {c['value'] for f in core for c in f['constants'] if c['name'].startswith('INT_')}
        wrappers = [w for f in core for w in f.get('interrupt_wrappers', [])]
        if {w.get('interrupt_id') for w in wrappers} != declared or declared != set(range(max(declared)+1)):
            raise ValueError('Core endpoint coverage/range differs: ' + build['build'])
        if any(w['wrapper'] is None or w['wrapper']['line'] != w['line'] for w in wrappers):
            raise ValueError('A Core interrupt has no same-line signature: ' + build['build'])
        if any(len(e['candidates']) != 1 for f in build['files'] for e in f.get('includes', [])):
            raise ValueError('Include candidate closure differs: ' + build['build'])


def render_api_index(build):
    lines = ['# Core interrupt index', '',
        'Source-only endpoint enumeration for build `2016-03-10-c`. IDs are evaluated source constants, not a binary dispatch trace. Full signatures, arguments and older builds are in [source-map.json](../assets/analysis/source-map.json). See [HOST_APIS.md](HOST_APIS.md).', '']
    for item in build['files']:
        if item['category'] != 'host-api' or not item.get('interrupt_wrappers'):
            continue
        lines += ['## ' + PurePosixPath(item['path']).name, '',
                  'Source: `' + item['path'] + '`; SHA-256 `' + item['sha256'] + '`.', '',
                  '| ID | Constant(s) | Wrapper(s), source lines |', '| ---: | --- | --- |']
        groups = defaultdict(list)
        for wrapper in item['interrupt_wrappers']:
            groups[wrapper['interrupt_id']].append(wrapper)
        for value, wrappers in sorted(groups.items()):
            constants = sorted({w['arguments'][0].lstrip('+').strip() for w in wrappers})
            names = ', '.join('`' + w['wrapper']['name'] + '`:' + str(w['line']) for w in wrappers)
            lines.append('| ' + str(value) + ' | ' + ', '.join('`' + c + '`' for c in constants) + ' | ' + names + ' |')
        lines.append('')
    return '\n'.join(lines)


def render_plugin_index(build):
    lines = ['# Plugin tool registration index', '',
        'Baseline `2016-03-10-c`. These are lexical `initTool` call sites, including internal/service tools and experiments. Conservative name resolution does not prove runtime reachability, load order or simultaneous registration. See [PLUGINS.md](PLUGINS.md).', '']
    for item in build['files']:
        if not item.get('registered_names'):
            continue
        lines += ['## ' + PurePosixPath(item['path']).stem, '',
                  'Source: `' + item['path'] + '`; SHA-256 `' + item['sha256'] + '`.', '',
                  '| Tool name expression resolved to | Line | Press callback expression |', '| --- | ---: | --- |']
        for entry in item['registered_names']:
            args = entry['arguments']
            direct = literal_string(args[0], item['string_bindings']) is not None
            index = 1 if direct else 2
            callback = args[index] if len(args) > index else 'unknown'
            escape = lambda value: value.replace('|', '\\|').replace('\n', ' ')
            lines.append('| `' + escape(entry['name'] or '(unresolved)') + '` | ' + str(entry['line']) + ' | `' + escape(callback) + '` |')
        lines.append('')
    return '\n'.join(lines)


def main():
    parser=argparse.ArgumentParser(description=__doc__);parser.add_argument('--check',action='store_true');args=parser.parse_args()
    builds=[]
    for p in json.loads(references.MANIFEST.read_text())['packages']:
        inventory=references.inspect_archive(references.REFERENCES/'archives'/p['filename'],p)
        root=references.REFERENCES/'extracted'/p['build'];references.check_extracted(root,inventory)
        builds.append(index_build(root,inventory))
    validate_report(builds)
    report={'schema':1,'method':'static lexical source index; no reference execution','generator_sha256':hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
            'limitations':['Include resolution records candidates, not measured compiler search order.','Qualified calls are lexical references, not proven execution or a typed call graph.','CP1252 is an explicit fallback, not an established encoding contract.','No host implementation, binary serializer, or bytecode layout is inferred from an API declaration.'],
            'builds':builds,'version_differences':differences(builds)}
    documents = {ROOT/'docs/API_INDEX.md':render_api_index(builds[-1]), ROOT/'docs/PLUGIN_INDEX.md':render_plugin_index(builds[-1])}
    if args.check:
        if not OUT.exists() or json.loads(OUT.read_text())!=report:raise ValueError('Static map is absent/stale; regenerate it')
        for path,content in documents.items():
            if not path.exists() or path.read_text()!=content:raise ValueError(f'Stale generated index: {path}')
    else:
        references.write_json(OUT,report)
        for path,content in documents.items():path.write_text(content)
    print(json.dumps({'builds':len(builds),'files':sum(len(b['files']) for b in builds),'latest_categories':builds[-1]['categories'],'report':str(OUT.relative_to(ROOT))},indent=2))


if __name__=='__main__':
    try:main()
    except (OSError,ValueError) as error:print(f'Source map: {error}',file=sys.stderr);sys.exit(1)