kangshifu-board/scripts/probe-agent-library-read.py

"""Run the installed Agent read path without installing or changing anything.

Only the explicit new output directory is written. No upstream collection,
publication, database writes, shell commands or credentials are exported.
"""
import sys
sys.dont_write_bytecode = True
import argparse
import datetime
import hashlib
import importlib
import json
import os
from pathlib import Path
import re
import subprocess
import urllib.request

REVIEW = 'jd:100297147639:106531290235443828'
SOURCE = '554b71fd087b7c0ef33016c736e1b822ff5be66b832ce523d2c96dc926d90735'
FIELDS = ('id', 'product_id', 'platform_family', 'review_date_raw',
          'review_calendar_date', 'review_date_utc', 'rating_raw',
          'rating_scale', 'is_default', 'source_record_id')
SMOKE = {'entities': ['reviews'], 'filters': {'platform': 'jd', 'brand': 'ksf',
         'category': 'banmian'}, 'quality': 'analysis', 'limit': 20}


def sha(raw):
    return hashlib.sha256(raw).hexdigest()


def code(value):
    # Never export raw traceback, arbitrary messages or stderr.
    return value if isinstance(value, str) and re.fullmatch(r'[A-Z][A-Z0-9_]{0,100}', value) else 'UNCLASSIFIED_ERROR'


def summary(result):
    out = {k: result[k] for k in ('ok', 'generation', 'readOnly', 'databaseWrites',
           'dataAccessMode', 'totalMatched', 'countsByEntity', 'effectiveQuery') if k in result}
    if isinstance(result.get('error'), dict):
        out['error'] = {'code': code(result['error'].get('code'))}
        diagnostics = result['error'].get('diagnostics', {})
        out['error']['diagnostics'] = {k: diagnostics[k] for k in (
            'stage', 'exitCode', 'stdoutBytes', 'stderrBytes',
            'stdoutSha256', 'stderrSha256') if k in diagnostics}
    if isinstance(result.get('records'), list):
        out['recordRefs'] = [{k: r[k] for k in ('entity', 'record_id', 'source_record_id',
            'review_day', 'rating_value', 'rating_scale', 'is_default') if k in r}
            for r in result['records']]
    return out


def valid_read(value, generation):
    return (value.get('ok') is True and value.get('readOnly') is True and
            value.get('databaseWrites') == 0 and value.get('generation') == generation and
            ('process' not in value or value['process'].get('exitCode') == 0))


def search_agrees(results, generation, keyword=None):
    """Require successful reads with identical scope, counts and ordered IDs."""
    if not results or not all(valid_read(r, generation) for r in results):
        return False
    expected_filters = {k: {'eq': v} for k, v in SMOKE['filters'].items()}
    for result in results:
        effective = result.get('effectiveQuery', {})
        if (effective.get('entities') != SMOKE['entities'] or
                effective.get('filters') != expected_filters or
                effective.get('quality') != 'analysis' or
                effective.get('text', {}).get('groupsAll') != ([[keyword]] if keyword else [])):
            return False
        if type(result.get('totalMatched')) is not int or result['totalMatched'] <= 0:
            return False
        refs = result.get('recordRefs', [])
        ids = [r.get('record_id') for r in refs]
        if not refs or any(not item for item in ids) or len(ids) != len(set(ids)):
            return False
        if len(refs) != min(result['totalMatched'], SMOKE['limit']):
            return False
    first = results[0]
    # Preview prose and gateway metadata may differ, but result scope and IDs may not.
    for result in results[1:]:
        if any(result.get(k) != first.get(k) for k in ('totalMatched', 'countsByEntity', 'effectiveQuery', 'recordRefs')):
            return False
    return True


def probe(args):
    skill = Path(args.skill_root).resolve()
    instructions = Path(getattr(args, 'instructions_root', None) or args.skill_root).resolve()
    scripts = skill / 'scripts'
    channel_path = args.channel or os.environ.get('KSF_LIBRARY_CHANNEL')
    if not channel_path:
        raise ValueError('CHANNEL_NOT_CONFIGURED')
    output = Path(args.out).absolute()
    if output.exists() or any(p.is_symlink() for p in (output, *output.parents)):
        raise ValueError('OUTPUT_ALREADY_EXISTS_OR_REDIRECTED')
    expected_state_path = getattr(args, 'expected_installed_state', None)
    if expected_state_path:
        expected_state = json.loads(Path(expected_state_path).read_text(encoding='utf-8-sig'))
        if sha((instructions / 'SKILL.md').read_bytes()) != expected_state['skillFingerprint']:
            raise ValueError('INSTALLED_SKILL_BASELINE_CHANGED')
        for name in ('query_voc.py', 'library_channel.py', 'voc_db.py', 'retrieval.py'):
            if sha((scripts / name).read_bytes()) != expected_state['installedFingerprints'][name]:
                raise ValueError('INSTALLED_READ_CODE_BASELINE_CHANGED')
    original_raw = Path(args.original_input).read_bytes() if args.original_input else None
    original = json.loads(original_raw.decode('utf-8-sig')) if original_raw is not None else None
    if original is not None and (not isinstance(original, dict) or set(original) - {
            'entities', 'filters', 'text', 'quality', 'limit', 'cursor', 'groupBy', 'sort'}):
        raise ValueError('ORIGINAL_INPUT_NOT_SEARCH_OBJECT')
    sys.path.insert(0, str(scripts))
    channel = importlib.import_module('library_channel')
    transport = []

    def runner(command, **kwargs):
        try:
            run = subprocess.run(command, **kwargs)
        except (OSError, subprocess.TimeoutExpired) as error:
            transport.append({'stage': 'ssh-execution', 'exceptionType': type(error).__name__})
            raise
        transport.append({'stage': 'ssh-response', 'exitCode': run.returncode,
            'requestSha256': sha(kwargs['input']), 'stdoutBytes': len(run.stdout),
            'stderrBytes': len(run.stderr), 'stdoutSha256': sha(run.stdout),
            'stderrSha256': sha(run.stderr)})
        return run

    def remote(command, params):
        try:
            return channel.send(channel_path, {'action': 'read', 'command': command,
                                'params': params}, runner=runner)
        except (ValueError, OSError) as error:
            return {'ok': False, 'error': {'code': code(str(error))}}

    def cli(command, extra, stdin=None):
        argv = [sys.executable, '-B', '-X', 'utf8', str(scripts / 'query_voc.py'),
                command, '--channel', channel_path, *extra]
        try:
            run = subprocess.run(argv, input=stdin, stdout=subprocess.PIPE,
                                 stderr=subprocess.PIPE, timeout=330, check=False)
            body = run.stdout if run.stdout.strip() else run.stderr
            result = json.loads(body.decode('utf-8-sig'))
            safe = summary(result)
            safe['process'] = {'exitCode': run.returncode, 'stdoutBytes': len(run.stdout),
                              'stderrBytes': len(run.stderr), 'stdoutSha256': sha(run.stdout),
                              'stderrSha256': sha(run.stderr)}
            return safe
        except (ValueError, OSError, subprocess.TimeoutExpired):
            return {'ok': False, 'error': {'code': 'CLI_RESULT_UNAVAILABLE'}}

    receipt = {'schema': 'ksf.agent-library-read-probe.v2',
        'checkedAt': datetime.datetime.now(datetime.timezone.utc).isoformat(),
        'readOnly': True, 'databaseWrites': 0, 'paidCalls': 0, 'upstreamCalls': 0,
        'installedFilesModified': False, 'originalRequestProvided': original is not None,
        'credentialsExported': False,
        'runtime': {'python': sys.version.split()[0]},
        'installedFingerprints': {name: sha((scripts / name).read_bytes()) for name in (
            'query_voc.py', 'library_channel.py', 'voc_db.py', 'retrieval.py')},
        'skillFingerprint': sha((instructions / 'SKILL.md').read_bytes()),
        'instructionSkillName': instructions.name}
    def board(request):
        req = urllib.request.Request('https://kangshifu.fmode.cn/api/library',
            data=json.dumps(request).encode('utf-8'), headers={'Content-Type': 'application/json'}, method='POST')
        with urllib.request.urlopen(req, timeout=40) as response:
            return json.load(response)

    public = board({'op': 'summary'})
    contract = remote('contract', {})
    receipt['contract'] = summary(contract)
    receipt['board'] = summary(public)
    generation = contract.get('generation')
    expected = getattr(args, 'expected_generation', None)
    if (not valid_read(contract, generation) or not valid_read(public, generation) or
            not generation or expected is not None and generation != expected):
        receipt['sameGeneration'] = False
        receipt['ok'] = False
        receipt['stoppedReason'] = 'CONTRACT_OR_BOARD_GENERATION_MISMATCH'
    else:
        receipt['sameGeneration'] = True
        pinned = ['--generation', generation]
        query = original if original is not None else SMOKE
        accept_current = getattr(args, 'accept_current', False)
        receipt['queryKind'] = ('original-input' if original is not None else
            'current-functional-acceptance' if accept_current else 'documented-smoke-only')
        receipt['query'] = query
        if original_raw is not None:
            receipt['input'] = {'sha256': sha(original_raw), 'bytes': len(original_raw),
                                'utf8Bom': original_raw.startswith(b'\xef\xbb\xbf')}
            receipt['searchOriginalFileCli'] = cli('search', pinned + ['--input', args.original_input])
            receipt['searchOriginalStdinCli'] = cli('search', pinned + ['--input', '-'], original_raw)
        receipt['searchStructuredChannel'] = summary(remote('search', {'generation': generation, 'input': query}))
        receipt['currentCliVerified'] = False
        receipt['keywordSearchVerified'] = False
        if accept_current:
            output.mkdir(mode=0o700, parents=True)
            receipt['currentChecks'] = []
            for name, keyword in (('scope', None), ('keyword', '味道')):
                current_query = {**SMOKE, **({'text': {'all': [keyword]}} if keyword else {})}
                raw = (json.dumps(current_query, ensure_ascii=False, indent=2) + '\n').encode('utf-8')
                path = output / (name + '-request.json')
                with path.open('xb') as stream:
                    stream.write(raw)
                os.chmod(path, 0o600)
                check = {'name': name, 'query': current_query, 'requestSha256': sha(raw),
                    'fileCli': cli('search', pinned + ['--input', str(path)]),
                    'stdinCli': cli('search', pinned + ['--input', '-'], raw),
                    'structuredChannel': summary(remote('search', {'generation': generation, 'input': current_query}))}
                board_request = {'op': 'browse', 'entity': 'reviews', 'filters': SMOKE['filters'],
                    'quality': SMOKE['quality'], 'limit': SMOKE['limit'], 'generation': generation}
                if keyword: board_request['keyword'] = keyword
                try:
                    check['board'] = summary(board(board_request))
                except (ValueError, OSError):
                    check['board'] = {'ok': False, 'error': {'code': 'BOARD_READ_UNAVAILABLE'}}
                check['verified'] = search_agrees([check[k] for k in (
                    'fileCli', 'stdinCli', 'structuredChannel', 'board')], generation, keyword)
                receipt['currentChecks'].append(check)
            receipt['currentCliVerified'] = all(c['verified'] for c in receipt['currentChecks'])
            receipt['keywordSearchVerified'] = receipt['currentChecks'][1]['verified']
        detail = remote('get', {'generation': generation, 'entity': 'reviews', 'id': REVIEW})
        record = detail.get('record', {})
        receipt['get'] = summary(detail)
        receipt['get']['recordFields'] = {k: {'present': k in record, 'value': record.get(k)} for k in FIELDS}
        receipt['get']['unknownFieldsPreserved'] = (
            'is_default' in record and record['is_default'] is None and
            'rating_scale' in record and record['rating_scale'] is None)
        source = remote('source', {'generation': generation, 'id': SOURCE})
        receipt['source'] = summary(source)
        citation = source.get('citation', {})
        receipt['source']['citationRef'] = {k: citation[k] for k in (
            'id', 'line_number', 'payload_sha256', 'assetSha256', 'archiveSha256') if k in citation}
        receipt['originalConditionVerified'] = original is not None and all(
            valid_read(receipt[k], generation) for k in (
                'searchOriginalFileCli', 'searchOriginalStdinCli', 'searchStructuredChannel'))
        receipt['fieldReadVerified'] = valid_read(detail, generation) and all(k in record for k in FIELDS) and (
            record['review_date_raw'] == '2026-11-06 10:49:00' and record['rating_raw'] == '5' and
            record['source_record_id'] == SOURCE and receipt['get']['unknownFieldsPreserved'])
        receipt['historicalRootCauseVerified'] = False
        receipt['acceptanceScope'] = 'current-functionality' if accept_current else 'original-condition'
        search_verified = receipt['currentCliVerified'] if accept_current else receipt['originalConditionVerified']
        receipt['ok'] = search_verified and receipt['fieldReadVerified'] and valid_read(source, generation)
    receipt['transport'] = transport
    receipt['installedFingerprintsAfter'] = {name: sha((scripts / name).read_bytes())
        for name in receipt['installedFingerprints']}
    receipt['skillFingerprintAfter'] = sha((instructions / 'SKILL.md').read_bytes())
    receipt['installedFilesUnchanged'] = (
        receipt['installedFingerprintsAfter'] == receipt['installedFingerprints'] and
        receipt['skillFingerprintAfter'] == receipt['skillFingerprint'])
    if not receipt['installedFilesUnchanged']:
        receipt['ok'] = False
        receipt['stoppedReason'] = 'INSTALLED_FILES_CHANGED_DURING_PROBE'
    output.mkdir(mode=0o700, parents=True, exist_ok=True)
    path = output / 'agent-library-read-receipt.json'
    path.write_text(json.dumps(receipt, ensure_ascii=False, indent=2) + '\n', encoding='utf-8')
    os.chmod(path, 0o600)
    print(json.dumps({'ok': receipt['ok'], 'receipt': str(path), 'sha256': sha(path.read_bytes()),
                      'installedFilesModified': False, 'databaseWrites': 0}, ensure_ascii=False))
    return 0 if receipt['ok'] else 1


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--skill-root', default='/opt/data/skills/ksf-voc-library')
    parser.add_argument('--instructions-root', help='Unified Skill instruction directory; runtime stays at --skill-root')
    parser.add_argument('--channel', help='Existing private config path; never copied to output')
    parser.add_argument('--original-input', help='Unchanged JSON input of the failed search')
    parser.add_argument('--accept-current', action='store_true', help='Verify current CLI file/stdin/keyword reads; never claims historical reproduction')
    parser.add_argument('--expected-generation', help='Stop if the current generation differs from this accepted generation')
    parser.add_argument('--expected-installed-state', help='Stop if public Skill/code fingerprints differ from the prior receipt')
    parser.add_argument('--out', required=True, help='New private output directory')
    options = parser.parse_args()
    try:
        sys.exit(probe(options))
    except (ValueError, OSError, KeyError) as error:
        print(json.dumps({'ok': False, 'error': {'code': code(str(error))},
                          'readOnly': True, 'databaseWrites': 0}, ensure_ascii=False))
        sys.exit(1)