"""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)