gstack/bin/gstack-brain-sync-index.py

#!/usr/bin/env python3
"""Read-only verification of brain-sync's post-add index; never repairs it."""
import json
import os
from pathlib import Path
import re
import stat
import subprocess
import sys
import unicodedata


class Refusal(Exception):
    def __init__(self, message, spellings=()):
        super().__init__(message)
        self.spellings = spellings


def git(root, *args, data=None, allow_missing=False):
    try:
        result = subprocess.run(
            ['git', '-C', str(root), *args], input=data,
            stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, timeout=30)
    except (OSError, subprocess.TimeoutExpired) as error:
        raise Refusal('index verification could not read Git') from error
    if result.returncode == 1 and allow_missing:
        return b''
    if result.returncode:
        raise Refusal('index verification Git command failed')
    return result.stdout


def prefixes(paths):
    groups = {}
    for path in paths:
        parts = path.split('/')
        for i in range(1, len(parts) + 1):
            spelling = '/'.join(parts[:i])
            key = unicodedata.normalize('NFC', spelling).casefold()
            groups.setdefault(key, set()).add(spelling)
    return groups


def names(spellings):
    # Only the colliding component names, clamped to a slug's charset: skill
    # start prints them, and the artifacts repo's paths are not trusted text.
    leaves = sorted({re.sub(r'[^A-Za-z0-9._-]', '?', s.rsplit('/', 1)[-1])[:64] for s in spellings})
    return ','.join(leaves[:4])


def first_collision(groups):
    found = sorted((key.count('/'), key) for key, spellings in groups.items() if len(spellings) > 1)
    return groups[found[0][1]] if found else ()


def tree_paths(root, *args):
    return {p.decode('utf-8') for p in git(root, *args).split(b'\0') if p}


def refuse_new_collisions(root, paths, baseline='HEAD'):
    # Compare complete component prefixes: Foo/a and foo/b collide too.
    # Existing aliases are historical data; this check does not migrate them
    # or block unrelated writes, but it refuses any additional spelling.
    head = git(root, 'ls-tree', '-rz', '--name-only', baseline) if baseline else b''
    previous = prefixes(p.decode('utf-8') for p in head.split(b'\0') if p)
    groups = prefixes(paths)
    for key, spellings in sorted(groups.items(), key=lambda item: (item[0].count('/'), item[0])):
        if len(spellings) > 1 and not spellings <= previous.get(key, set()):
            raise Refusal('new casefold path collision; inspect path spellings before retrying', spellings)
    return first_collision(groups)


def hold(root, requested_file, held_file):
    # A requested path is held when one of its prefixes has another spelling
    # in the index or in this batch, unless its own spelling is the only one
    # the index already tracks (an established project keeps syncing; the new
    # alias waits). Held paths are never staged, so git cannot record them
    # under the other spelling, and their queue records stay.
    requested = {p for p in Path(requested_file).read_text(encoding='utf-8').replace('\r', '').split('\n') if p}
    tracked_paths = tree_paths(root, 'ls-files', '-z')
    tracked = prefixes(tracked_paths)
    groups = prefixes(tracked_paths | requested)
    held, colliding = set(), []
    for path in requested:
        parts = path.split('/')
        for i in range(1, len(parts) + 1):
            spelling = '/'.join(parts[:i])
            key = unicodedata.normalize('NFC', spelling).casefold()
            if len(groups[key]) > 1 and tracked.get(key) != {spelling}:
                held.add(path)
                colliding.append((key.count('/'), key))
                break
    with open(held_file, 'w', encoding='utf-8', newline='\n') as f:
        f.writelines(p + '\n' for p in sorted(held))
    return groups[min(colliding)[1]] if colliding else ()


def verify(root, requested_file, names_file):
    requested = set(Path(requested_file).read_text(encoding='utf-8').splitlines())
    requested.discard('')
    entries = {}
    for record in git(root, 'ls-files', '--stage', '-z').split(b'\0'):
        if not record:
            continue
        header, name = record.split(b'\t', 1)
        mode, oid, stage = header.split()
        name = name.decode('utf-8')
        if stage != b'0':
            raise Refusal('unmerged index entries; inspect the index')
        entries[name] = (mode, oid)
    changed = {p.decode('utf-8') for p in Path(names_file).read_bytes().split(b'\0') if p}
    refuse_new_collisions(root, set(entries) | requested)
    if requested - entries.keys() or changed - requested:
        raise Refusal('requested/index path mismatch; inspect exact path spellings and unrelated staged entries')

    regular = []
    for name in sorted(requested):
        mode, oid = entries[name]
        file = root / name
        actual = file.lstat().st_mode
        if mode == b'120000' and stat.S_ISLNK(actual):
            hashed = git(root, 'hash-object', '--stdin', data=os.fsencode(os.readlink(file))).strip()
            if hashed != oid:
                raise Refusal('requested/index content mismatch')
        elif mode in (b'100644', b'100755') and stat.S_ISREG(actual):
            regular.append(name)
        else:
            raise Refusal('requested/index file type mismatch')
    if regular:
        # --stdin-paths applies the same clean filters/EOL conversion as add.
        # Quote each path as Git's C-style input (including quotes/backslashes).
        paths = ''.join(json.dumps(p, ensure_ascii=False) + '\n' for p in regular).encode('utf-8')
        hashes = git(root, 'hash-object', '--stdin-paths', data=paths).splitlines()
        if len(hashes) != len(regular) or any(entries[p][1] != oid for p, oid in zip(regular, hashes)):
            raise Refusal('requested/index content mismatch; queued content is not in the index')


def outgoing(root):
    if not git(root, 'rev-parse', '--verify', '--quiet', 'HEAD', allow_missing=True):
        return ()
    branch = git(root, 'rev-parse', '--abbrev-ref', 'HEAD').decode('utf-8').strip()
    remote = 'refs/remotes/origin/' + branch
    baseline = remote if git(root, 'rev-parse', '--verify', '--quiet', remote, allow_missing=True) else None
    return refuse_new_collisions(root, tree_paths(root, 'ls-tree', '-rz', '--name-only', 'HEAD'), baseline)


if __name__ == '__main__':
    try:
        if sys.argv[1] == '--outgoing':
            print(names(outgoing(Path(sys.argv[2]))))
        elif sys.argv[1] == '--hold':
            print(names(hold(Path(sys.argv[2]), sys.argv[3], sys.argv[4])))
        else:
            verify(Path(sys.argv[1]), sys.argv[2], sys.argv[3])
    except (Refusal, OSError, UnicodeError, ValueError, IndexError) as error:
        print(names(getattr(error, 'spellings', ())))
        print('BRAIN_SYNC: staged artifact validation refused: ' + str(error), file=sys.stderr)
        sys.exit(1)