"""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: result = hashlib.sha256() for block in iter(lambda: stream.read(1024 * 1024), b''): result.update(block) return result.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)