"""Ocean6 local reference v0.2. Python 3.9+, standard library only. Deterministic line retrieval, not AI inference or a distributed node runtime. Run: python3 reference_runtime.py run task.json result.json receipt.json Verify: python3 reference_runtime.py verify task.json result.json receipt.json Receipts are unsigned local records; they do not authenticate an executor. """ import argparse from datetime import datetime, timezone import hashlib import json from pathlib import Path import re import sys VERSION = '0.2' OPERATION = 'literal-line-search' def require(condition, code): if not condition: raise ValueError(code) def fold(value): return re.sub('[A-Z]', lambda m: chr(ord(m[0]) + 32), value) def scalar_text(value): return isinstance(value, str) and not any(0xD800 <= ord(c) <= 0xDFFF for c in value) def execute(task): require(isinstance(task, dict) and set(task) == {'version', 'operation', 'document', 'terms', 'top_k'}, 'INVALID_FIELDS') require(task['version'] == VERSION and task['operation'] == OPERATION, 'UNSUPPORTED_VERSION_OR_OPERATION') document, terms, top_k = task['document'], task['terms'], task['top_k'] require(scalar_text(document) and 1 <= len(document.encode('utf-8')) <= 20000, 'INVALID_DOCUMENT') require(isinstance(terms, list) and 1 <= len(terms) <= 16, 'INVALID_TERMS') require(type(top_k) in (int, float) and 1 <= top_k <= 10 and top_k == int(top_k), 'INVALID_TOP_K') top_k = int(top_k) normalized = [] for term in terms: require(scalar_text(term) and 1 <= len(term) <= 80 and '\r' not in term and '\n' not in term, 'INVALID_TERM') term = fold(term.strip(' \t')) require(bool(term), 'EMPTY_TERM') if term not in normalized: normalized.append(term) matches = [] for number, line in enumerate(document.replace('\r\n', '\n').replace('\r', '\n').split('\n'), 1): matched = [term for term in normalized if term in fold(line)] if matched: matches.append({'line': number, 'score': len(matched), 'text': line, 'matched_terms': matched}) matches.sort(key=lambda row: (-row['score'], row['line'])) return {'version': VERSION, 'operation': OPERATION, 'total_matches': len(matches), 'matches': matches[:top_k]} def encode(value): return (json.dumps(value, ensure_ascii=False, indent=2, allow_nan=False) + '\n').encode('utf-8') def digest(data): return hashlib.sha256(data).hexdigest() def now(): return datetime.now(timezone.utc).isoformat(timespec='microseconds').replace('+00:00', 'Z') def strict_json(data): def pairs(values): obj = {} for key, value in values: require(key not in obj, 'DUPLICATE_JSON_KEY') obj[key] = value return obj def invalid(value): raise ValueError('INVALID_JSON_NUMBER') return json.loads(data, object_pairs_hook=pairs, parse_constant=invalid) def canonical(value): if isinstance(value, dict): return {key: canonical(item) for key, item in value.items()} if isinstance(value, list): return [canonical(item) for item in value] if type(value) is float and value.is_integer(): return int(value) return value def compare(task_bytes, result_bytes): expected = execute(strict_json(task_bytes)) actual = strict_json(result_bytes) require(json.dumps(canonical(actual), sort_keys=True, ensure_ascii=False) == json.dumps(expected, sort_keys=True, ensure_ascii=False), 'REPLAY_MISMATCH') return True def verify(task_bytes, result_bytes, receipt_bytes): compare(task_bytes, result_bytes) receipt = strict_json(receipt_bytes) fields = {'version','execution','implementation','task_sha256','result_sha256','started_at','finished_at','signature','settlement'} require(isinstance(receipt, dict) and set(receipt) == fields, 'INVALID_RECEIPT') require(receipt['version'] == VERSION and receipt['execution'] == 'local-cpu', 'INVALID_RECEIPT_SCOPE') require(receipt['implementation'] == 'ocean6-python-reference/0.2', 'INVALID_IMPLEMENTATION') require(receipt['task_sha256'] == digest(task_bytes), 'TASK_DIGEST_MISMATCH') require(receipt['result_sha256'] == digest(result_bytes), 'RESULT_DIGEST_MISMATCH') require(receipt['signature'] is None and receipt['settlement'] is None, 'INVALID_RECEIPT_SCOPE') times = [] for key in ['started_at', 'finished_at']: value = receipt[key] require(isinstance(value, str) and re.fullmatch(r'\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{6}Z', value), 'INVALID_TIMESTAMP') times.append(datetime.fromisoformat(value.replace('Z', '+00:00'))) require(times[1] >= times[0], 'INVALID_TIME_ORDER') return True def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('command', choices=['run', 'verify', 'compare']) parser.add_argument('task', type=Path) parser.add_argument('result', type=Path) parser.add_argument('receipt', type=Path, nargs='?') args = parser.parse_args() try: require(args.command == 'compare' or args.receipt is not None, 'RECEIPT_PATH_REQUIRED') paths = [args.task.resolve(), args.result.resolve()] + ([args.receipt.resolve()] if args.receipt else []) require(len(set(paths)) == len(paths), 'PATHS_MUST_DIFFER') require(args.task.stat().st_size <= 150000, 'TASK_FILE_TOO_LARGE') task_bytes = args.task.read_bytes() if args.command == 'run': started = now() result = encode(execute(strict_json(task_bytes))) receipt = {'version': VERSION, 'execution': 'local-cpu', 'implementation': 'ocean6-python-reference/0.2', 'task_sha256': digest(task_bytes), 'result_sha256': digest(result), 'started_at': started, 'finished_at': now(), 'signature': None, 'settlement': None} args.result.write_bytes(result) args.receipt.write_bytes(encode(receipt)) print('LOCAL RUN COMPLETE: result and unsigned receipt written. No network execution.') elif args.command == 'compare': require(args.result.stat().st_size <= 150000, 'OUTPUT_FILE_TOO_LARGE') compare(task_bytes, args.result.read_bytes()) print('PASS: result matches independent Python replay. No identity or digest claim.') else: require(args.result.stat().st_size <= 150000 and args.receipt.stat().st_size <= 10000, 'OUTPUT_FILE_TOO_LARGE') verify(task_bytes, args.result.read_bytes(), args.receipt.read_bytes()) print('PASS: deterministic replay and byte digests. Executor identity and timestamps are not authenticated.') return 0 except (ValueError, OSError, TypeError, UnicodeError, RecursionError) as error: print('FAIL: ' + str(error), file=sys.stderr) return 1 if __name__ == '__main__': sys.exit(main())