kangshifu-board/scripts/verify-installed-voc-database.py

"""Verify the installed database with the already-installed query Skill.

Read-only; no new environment variables, sessions, reports or collection calls.
The only optional write is a new receipt, outside the database and skill folders.
"""
import argparse
import hashlib
import json
import os
from pathlib import Path
import subprocess
import sys
import time

sys.dont_write_bytecode = True


def digest(path):
    with Path(path).open('rb') as stream:
        return hashlib.file_digest(stream, 'sha256').hexdigest()


def skill_snapshot(skill):
    manifest = skill / 'FILES.sha256'
    hashes = {'FILES.sha256': digest(manifest)}
    for line in manifest.read_text(encoding='utf-8').splitlines():
        sha, name = line.split('  ', 1)
        path = (skill / name).resolve()
        if not path.is_relative_to(skill.resolve()) or digest(path) != sha:
            raise ValueError('SKILL_MANIFEST_MISMATCH')
        hashes[name] = sha
    return hashes


def verify(package, skill, db):
    package, skill, db = [Path(p).absolute() for p in (package, skill, db)]
    release = json.loads((package / 'DATABASE.json').read_text(encoding='utf-8'))
    expected = release['database']
    hashes = skill_snapshot(skill)
    if hashes['FILES.sha256'] != release['requiredResearchSkillManifestSha256']:
        raise ValueError('EXPECTED_NOV6_QUERY_SKILL_REQUIRED')
    before = digest(db)
    if before != expected['sha256']:
        raise ValueError('INSTALLED_DATABASE_HASH_MISMATCH')
    cli = skill / 'scripts/query_voc.py'
    calls = []
    def call(command, *args, generation=None, success=True):
        argv = [sys.executable, '-B', '-X', 'utf8', str(cli), command, '--db', str(db), *args]
        if generation:
            argv += ['--generation', generation]
        start = time.perf_counter()
        proc = subprocess.run(argv, text=True, encoding='utf-8', capture_output=True, timeout=60)
        if (proc.returncode == 0) != success:
            raise ValueError('QUERY_CALL_FAILED:' + command + ':' + proc.stderr[:300])
        payload = json.loads(proc.stdout if success else proc.stderr)
        if payload.get('readOnly') is not True or payload.get('databaseWrites') != 0:
            raise ValueError('READ_ONLY_RECEIPT_REQUIRED')
        calls.append(dict(command=command, exitCode=proc.returncode, elapsedMs=round((time.perf_counter()-start)*1000, 3)))
        return payload
    contract = call('contract')
    generation = contract['generation']
    if generation != expected['generation'] or contract['retrieval']['agentRulesVersion'] != 'ksf.agent-data-rules.20261106.v1':
        raise ValueError('CONTRACT_OR_GENERATION_MISMATCH')
    if contract['counts'] != expected['entityCounts']:
        raise ValueError('CONTRACT_COUNTS_MISMATCH')
    checked = call('verify', generation=generation)
    if checked['generation'] != generation:
        raise ValueError('VERIFIED_GENERATION_MISMATCH')
    scenarios, sampled = [], []
    for path in sorted((package / 'examples').glob('*.json')):
        command = 'aggregate' if path.name.startswith(('03-', '08-')) else 'search'
        result = call(command, '--input', str(path), generation=generation)
        if result['generation'] != generation:
            raise ValueError('QUERY_GENERATION_MISMATCH')
        if 'filters' not in result['effectiveQuery']:
            raise ValueError('EFFECTIVE_QUERY_REQUIRED')
        if path.name.startswith(('01-', '02-', '06-', '07-', '09-')):
            for row in result.get('records', []):
                if (row['platform'], row['brand'], row['category']) != ('jd', 'ksf', 'banmian'):
                    raise ValueError('EXPLICIT_SCOPE_MISMATCH')
        if path.name.startswith('02-'):
            for row in result['records']:
                if row['is_default'] or row['identity_state'] != 'matched' or row['category_state'] != 'consistent':
                    raise ValueError('REVIEW_ANALYSIS_POLICY_MISMATCH')
        if path.name.startswith(('01-', '02-')) and result['records']:
            sampled.append(result['records'][0])
        scenarios.append(dict(example=path.name, command=command, totalMatched=result.get('totalMatched'),
            groups=result.get('groups'), effectiveQuery=result['effectiveQuery'],
            firstRefs=[dict(entity=row['entity'], id=row['record_id'], sourceId=row['source_record_id']) for row in result.get('records', [])[:2]]))
    citations = []
    for row in sampled:
        detail = call('get', '--entity', row['entity'], '--id', row['record_id'], generation=generation)['record']
        if detail['source_record_id'] != row['source_record_id']:
            raise ValueError('SOURCE_RECORD_MISMATCH')
        source = call('source', '--id', detail['source_record_id'], generation=generation)
        citation = source['citation']
        if citation['id'] != row['source_record_id'] or not citation['archiveSha256'] or not citation['payload_sha256']:
            raise ValueError('SOURCE_CITATION_INCOMPLETE')
        citations.append(dict(entity=row['entity'], id=row['record_id'], sourceId=row['source_record_id'],
            archiveSha256=citation['archiveSha256'], asset=citation['asset_path'], line=citation['line_number']))
    stale = call('get', '--entity', sampled[0]['entity'], '--id', sampled[0]['record_id'], generation='deliberately-stale', success=False)
    if stale['error']['code'] != 'LIBRARY_GENERATION_CHANGED':
        raise ValueError('GENERATION_GUARD_NOT_ENFORCED')
    if digest(db) != before or skill_snapshot(skill) != hashes:
        raise ValueError('DATABASE_OR_SKILL_CHANGED_DURING_READS')
    return dict(ok=True, readOnly=True, databaseWrites=0, generation=generation, databaseSha256=before,
        databaseLocation=str(db), queryScope='installed static snapshot, not the whole industry or a production board connection',
        counts=contract['counts'], scenarios=scenarios, citations=citations, calls=calls,
        databaseUnchanged=True, skillUnchanged=True, generationGuardVerified=True,
        storageMode='static-snapshot', automaticUpdatesConfigured=False, productionBoardAligned=False,
        realMessageVerified=False, browserVerified=False)


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--package', required=True)
    parser.add_argument('--skill-dir', default='/opt/data/skills/ksf-voc-library')
    parser.add_argument('--db', default='/opt/data/ksf-voc-library/library.sqlite')
    parser.add_argument('--out')
    args = parser.parse_args()
    try:
        if args.out:
            output = Path(args.out).absolute()
            if output.is_relative_to(Path(args.skill_dir).resolve()) or output.is_relative_to(Path(args.db).resolve().parent) or output.is_relative_to(Path(args.package).resolve()):
                raise ValueError('RECEIPT_MUST_BE_OUTSIDE_PACKAGE_SKILL_AND_LIBRARY')
        result = verify(args.package, args.skill_dir, args.db)
        if args.out:
            fd = os.open(args.out, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
            with os.fdopen(fd, 'w', encoding='utf-8') as stream:
                json.dump(result, stream, ensure_ascii=False, indent=2)
                stream.write('\n')
        print(json.dumps(result, ensure_ascii=False, indent=2))
    except (ValueError, OSError, KeyError, subprocess.TimeoutExpired) as error:
        print(json.dumps(dict(ok=False, error=dict(code=str(error))), ensure_ascii=False), file=sys.stderr)
        raise SystemExit(1)