kangshifu-board/scripts/verify-query-skills.py

"""Exercise the delivered CLI on the current local corpus, with no database writes."""
import hashlib
import importlib.util
import json
from pathlib import Path
import shutil
import subprocess
import sys
import tempfile

ROOT = Path(__file__).resolve().parents[1]
WORK = ROOT / 'work/agent-query-20261106'
RELEASE = ROOT / 'artifacts/数据库精准检索规则-20261106/ksf-agent-query-rules-20261106-r1'
sys.path.insert(0, str(ROOT / 'server/voc-library'))
import voc_db
import verify_retrieval

spec = importlib.util.spec_from_file_location('query_installer', RELEASE / 'install-query-skills.py')
installer = importlib.util.module_from_spec(spec)
spec.loader.exec_module(installer)
registry = ROOT / 'work/library-preview/registry'
db = voc_db.registry_database(registry)
before = installer.digest(db)
pointer_before = installer.digest(registry / 'current.json')
research = verify_retrieval.verify(db)
with tempfile.TemporaryDirectory(prefix='query-skill-', dir=WORK) as temp:
    data = Path(temp) / 'data'
    for name in installer.NAMES:
        shutil.copytree(WORK / 'baseline-skills' / name, data / 'skills' / name)
    (data / 'skills/ksf-workshop-runtime/credential-binding.json').write_text('{"testSentinel":true}', encoding='utf-8')
    state = data / 'ksf-workshop-state/sessions'
    state.mkdir(parents=True)
    (state / 'test-session.json').write_text('{"testSentinel":true}', encoding='utf-8')
    old = installer.snapshot(data)
    plan = installer.install(RELEASE, data)
    assert installer.snapshot(data) == old
    expected_workshop = ['ksf-workshop-runtime/' + rel for rel in ('SKILL.md', 'references/workshop-tasks.md', 'FILES.sha256')]
    assert sorted(x for x in plan['changed'] if x.startswith('ksf-workshop-runtime/')) == sorted(expected_workshop)
    applied = installer.install(RELEASE, data, True)
    assert applied['filesystemVerified'] and applied['credentialBindingPreserved'] and applied['stateDirectoryPreserved']
    assert not installer.install(RELEASE, data, True)['applied']
    cli = data / 'skills/ksf-voc-library/scripts/query_voc.py'
    def call(command, *args, generation=None, ok=True):
        argv = [sys.executable, '-B', '-X', 'utf8', str(cli), command, '--registry', str(registry), *args]
        if generation:
            argv.extend(['--generation', generation])
        result = subprocess.run(argv, text=True, encoding='utf-8', capture_output=True, timeout=30)
        assert (result.returncode == 0) == ok, result.stderr
        payload = json.loads(result.stdout if ok else result.stderr)
        assert payload['readOnly'] and payload['databaseWrites'] == 0
        return payload
    contract = call('contract')
    generation = contract['generation']
    assert contract['retrieval']['agentRulesVersion'] == 'ksf.agent-data-rules.20261106.v1'
    examples = []
    first_product = first_review = None
    for path in sorted((RELEASE / 'examples').glob('*.json')):
        aggregate = path.name.startswith(('03-', '08-'))
        response = call('aggregate' if aggregate else 'search', '--input', str(path), generation=generation)
        assert response['generation'] == generation
        for row in response.get('records', []):
            if path.name.startswith(('01-', '02-', '06-', '07-', '09-')):
                assert (row['platform'], row['brand'], row['category']) == ('jd', 'ksf', 'banmian')
            if row['entity'] == 'products' and first_product is None:
                first_product = row
            if row['entity'] == 'reviews' and first_review is None:
                first_review = row
        examples.append(dict(file=path.name, command='aggregate' if aggregate else 'search',
            totalMatched=response.get('totalMatched'), groups=response.get('groups'), effectiveQuery=response['effectiveQuery']))
    citations = []
    for row in [first_product, first_review]:
        if row is None:
            continue
        record = call('get', '--entity', row['entity'], '--id', row['record_id'], generation=generation)['record']
        source = call('source', '--id', record['source_record_id'], generation=generation)
        assert source['citation']['id'] == row['source_record_id']
        citations.append(dict(entity=row['entity'], id=row['record_id'], sourceId=row['source_record_id'], archiveSha256=source['citation']['archiveSha256']))
    stale = call('get', '--entity', 'products', '--id', first_product['record_id'], generation='stale-generation', ok=False)
    assert stale['error']['code'] == 'LIBRARY_GENERATION_CHANGED'
assert installer.digest(db) == before and installer.digest(registry / 'current.json') == pointer_before
result = dict(ok=True, generation=generation, databaseSha256=before, queryExamples=examples, sourceChecks=citations,
    realCorpusScenarios=research['scenarios'], benchmark=research['benchmark'], installPlan=plan,
    installSimulationVerified=True, generationGuardVerified=True, databaseWrites=0,
    productionInstalled=False, realMessageVerified=False)
(WORK / 'validation.json').write_text(json.dumps(result, ensure_ascii=False, indent=2) + '\n', encoding='utf-8')
print(json.dumps(dict(ok=True, examples=len(examples), researchScenarios=len(research['scenarios']), sourceChecks=len(citations),
    benchmark=research['benchmark'], generation=generation, changed=plan['changed']), ensure_ascii=False, indent=2))