"""Task-only transparent nonstream recorder. Root must separately freeze and arm. The incoming JSON body is validated, recorded and forwarded without any edits. No models endpoint, retries, subprocesses, prompt rewriting or quality reruns. """ import datetime import hashlib import http.server import json import os import pathlib import threading import time import urllib.error import urllib.request ROOT = pathlib.Path(__file__).resolve().parent PRIVATE = ROOT / 'private' RECORDS = ROOT / 'records' PRIVATE.mkdir(exist_ok=True) RECORDS.mkdir(exist_ok=True) HEADERS = PRIVATE / 'headers' HEADERS.mkdir(exist_ok=True) STATE_PATH = PRIVATE / 'recorder-state.json' CONTRACT_PATH = ROOT / 'original-case-contract-v1.json' RUNTIME_PATH = ROOT / 'runtime-frozen-v1.json' ARM_PATH = PRIVATE / 'armed-case.json' UPSTREAM = 'http://127.0.0.1:36280/v1/chat/completions' PREFERRED_PORT = 37445 LOCK = threading.RLock() HOP_HEADERS = {'host', 'content-length', 'connection', 'keep-alive', 'transfer-encoding', 'upgrade', 'proxy-authorization', 'proxy-authenticate', 'te', 'trailer'} utc = lambda: datetime.datetime.now(datetime.timezone.utc).isoformat() sha = lambda raw: hashlib.sha256(raw).hexdigest() jbytes = lambda value: (json.dumps(value, ensure_ascii=False, indent=2) + '\n').encode('utf-8') class NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): return None OPENER = urllib.request.build_opener(urllib.request.ProxyHandler({}), NoRedirect()) if STATE_PATH.exists(): raise RuntimeError('Existing recorder execution state: do not restart or overwrite') contract_bytes = CONTRACT_PATH.read_bytes() contract = json.loads(contract_bytes) case_ids = [case['id'] for case in contract['cases']] if case_ids != ['fabric-primary', 'fabric-boundary']: raise RuntimeError('Unexpected original case identity') STATE = {'startedAt': utc(), 'pid': os.getpid(), 'port': None, 'upstream': UPSTREAM, 'forwards': 0, 'deniedRequests': 0, 'inFlight': 0, 'cases': {cid: {'forwards': 0, 'completedResponses': 0} for cid in case_ids}, 'payloadMutations': 0, 'providerOutputMutations': 0, 'recorderRetries': 0, 'qualityReruns': 0, 'newModelDownloads': 0, 'originalContractSha256': sha(contract_bytes), 'sdkRetryPolicy': 'unknown; every actual additional native request is recorded', 'armedByThisProgram': False} def save(): STATE_PATH.write_bytes(jbytes(STATE)) def put(directory, name, raw): with (directory / name).open('xb') as stream: stream.write(raw) def validate(value): if not RUNTIME_PATH.exists() or not ARM_PATH.exists(): return None, None, 'Missing immutable runtime or unarmed' runtime_bytes = RUNTIME_PATH.read_bytes() runtime = json.loads(runtime_bytes) armed = json.loads(ARM_PATH.read_bytes()) cid = armed.get('caseId') if cid not in STATE['cases']: return None, None, 'Unarmed case' if sha(CONTRACT_PATH.read_bytes()) != runtime.get('contractSha256') or sha(pathlib.Path(__file__).read_bytes()) != runtime.get('recorderSha256'): return None, None, 'Immutable original contract or recorder hash mismatch' if runtime.get('contractSha256') != STATE['originalContractSha256']: return None, None, 'Original contract differs from startup snapshot' if armed.get('runtimeSha256') != sha(runtime_bytes): return None, None, 'Armed runtime hash mismatch' if runtime.get('recorderPort') != STATE['port'] or runtime.get('maximumProviderForwardsPerCase') != 12 or runtime.get('caseTimeoutSeconds') != 180: return None, None, 'Frozen recorder port, forward budget or timeout changed' start = armed.get('startedAtEpoch') deadline = armed.get('deadlineEpoch') if type(start) not in (int, float) or type(deadline) not in (int, float) or not start < deadline <= start + 180: return None, None, 'Invalid fixed native case deadline' if time.time() < start or time.time() >= deadline: return None, None, 'Fixed native case deadline inactive or elapsed' if STATE['cases'][cid]['forwards'] >= 12: return None, None, 'Fixed native forward budget exhausted' fields = {'model', 'messages', 'temperature', 'top_p', 'max_tokens', 'stream'} if not isinstance(value, dict) or set(value) != fields: return None, None, 'Native Chat Completions field set differs from original contract' case = next(item for item in contract['cases'] if item['id'] == cid) messages = value.get('messages') expected_messages = [{'role': 'system', 'content': case['instruction'].strip()}, {'role': 'user', 'content': case['input'].strip()}] if messages != expected_messages or any(not isinstance(message, dict) or set(message) != {'role', 'content'} for message in messages): return None, None, 'Original complete native TrimSpace system or input mismatch' params = contract['nativeParameters'] if value['model'] != runtime.get('expectedModelId') or not isinstance(value['model'], str) or not value['model']: return None, None, 'Frozen existing model ID mismatch' if value['temperature'] != params['temperature'] or value['top_p'] != params['top_p'] or type(value['max_tokens']) is not int or value['max_tokens'] != params['max_tokens'] or value['stream'] is not False: return None, None, 'Frozen nonstream sampling or token limit mismatch' return cid, deadline, None class Recorder(http.server.BaseHTTPRequestHandler): protocol_version = 'HTTP/1.1' def log_message(self, *args): pass def reply(self, status, body, content_type='application/json'): self.send_response(status) self.send_header('Content-Type', content_type) self.send_header('Content-Length', str(len(body))) self.send_header('Connection', 'close') self.end_headers() self.close_connection = True self.wfile.write(body) self.wfile.flush() def deny(self, status, reason, raw, raw_body_complete=True): with LOCK: STATE['deniedRequests'] += 1 number = STATE['deniedRequests'] prefix = f'fabric-denied-{number:02}' put(RECORDS, prefix + '-native-request.bin', raw) put(HEADERS, prefix + '-native-headers.json', jbytes(list(self.headers.items()))) put(RECORDS, prefix + '-receipt.json', jbytes({'at': utc(), 'method': self.command, 'path': self.path, 'status': status, 'reason': reason, 'upstreamForwarded': False, 'requestSha256': sha(raw), 'rawBodyComplete': raw_body_complete, 'declaredContentLength': self.headers.get('Content-Length'), 'guardOutcome': 'denied'})) save() self.reply(status, jbytes({'error': {'message': reason}})) def do_GET(self): if self.path != '/task-state': self.deny(403, 'Endpoint outside task scope; models/discovery forwarding is forbidden', b'') return with LOCK: snapshot = dict(STATE) snapshot['armed'] = ARM_PATH.exists() snapshot['runtimeFrozen'] = RUNTIME_PATH.exists() self.reply(200, jbytes(snapshot)) def do_POST(self): try: length = int(self.headers.get('Content-Length', '0')) except ValueError: length = 0 if length < 1 or length > 65536: self.deny(413, 'Invalid task request size', b'', False) return raw = self.rfile.read(length) try: value = json.loads(raw) except Exception: value = None with LOCK: try: cid, deadline, reason = validate(value) if self.path == '/v1/chat/completions' else (None, None, 'Endpoint outside task scope') except Exception as error: cid, deadline, reason = None, None, 'Freeze/arm validation failed: ' + type(error).__name__ if reason: self.deny(403, reason, raw) return STATE['forwards'] += 1 STATE['cases'][cid]['forwards'] += 1 STATE['inFlight'] += 1 ordinal = STATE['cases'][cid]['forwards'] prefix = f'{cid}-request-{ordinal:02}' started = utc() put(RECORDS, prefix + '-native-request.json', raw) put(HEADERS, prefix + '-native-headers.json', jbytes(list(self.headers.items()))) save() status = 502 body = b'' response_headers = [] error = None content_type = 'application/json' delivered = False provider_reply_received = False timeout_seconds = max(0.001, min(180, deadline - time.time())) try: forwarded_headers = {key: value for key, value in self.headers.items() if key.lower() not in HOP_HEADERS} request = urllib.request.Request(UPSTREAM, data=raw, headers=forwarded_headers, method='POST') with OPENER.open(request, timeout=timeout_seconds) as response: status = response.status body = response.read() response_headers = list(response.headers.items()) content_type = response.headers.get('Content-Type', 'application/json') provider_reply_received = True except urllib.error.HTTPError as exc: status = exc.code body = exc.read() response_headers = list(exc.headers.items()) content_type = exc.headers.get('Content-Type', 'application/json') provider_reply_received = True error = {'type': type(exc).__name__, 'message': str(exc)} except Exception as exc: error = {'type': type(exc).__name__, 'message': str(exc)} put(RECORDS, prefix + '-upstream-response.json', body) put(HEADERS, prefix + '-upstream-headers.json', jbytes(response_headers)) try: self.reply(status, body, content_type) delivered = True except (BrokenPipeError, ConnectionResetError, OSError) as exc: error = error or {'type': type(exc).__name__, 'message': str(exc)} put(RECORDS, prefix + '-served-response.json', body if delivered else b'') parsed = {} try: parsed = json.loads(body) except Exception: pass candidate_choices = parsed.get('choices', []) if isinstance(parsed, dict) else [] choices = [choice for choice in candidate_choices if isinstance(choice, dict)] if isinstance(candidate_choices, list) else [] message = choices[0].get('message', {}) if choices else {} first_content = message.get('content') if isinstance(message, dict) else None output = first_content if isinstance(first_content, str) else '' put(RECORDS, prefix + '-first-provider-output.txt', output.encode('utf-8')) receipt = {'caseId': cid, 'ordinal': ordinal, 'startedAt': started, 'completedAt': utc(), 'completedBeforeCaseDeadline': time.time() <= deadline, 'fixedCaseDeadlineEpoch': deadline, 'upstreamTimeoutSecondsAtForward': timeout_seconds, 'status': status, 'providerReplyReceived': provider_reply_received, 'requestBytesForwardedUnchanged': True, 'nativeRequestSha256': sha(raw), 'upstreamRequestSha256': sha(raw), 'upstreamResponseSha256': sha(body), 'servedResponseSha256': sha(body if delivered else b''), 'bodyBytesServedUnchanged': delivered, 'downstreamDisconnected': not delivered, 'finishReasons': [choice.get('finish_reason') for choice in choices], 'usage': parsed.get('usage') if isinstance(parsed, dict) else None, 'error': error, 'providerOutputSha256': sha(output.encode('utf-8')), 'providerOutputSelection': 'choices[0].message.content, matching official native OpenAI nonstream Send; full response retained separately', 'guardOutcome': 'passed', 'additionalNativeForwardForCase': ordinal > 1, 'retryCause': 'not inferred; actual requests retained', 'privateNativeHeadersPath': 'private/headers/' + prefix + '-native-headers.json', 'privateUpstreamHeadersPath': 'private/headers/' + prefix + '-upstream-headers.json'} put(RECORDS, prefix + '-receipt.json', jbytes(receipt)) with LOCK: STATE['inFlight'] -= 1 STATE['cases'][cid]['completedResponses'] += 1 STATE['cases'][cid].setdefault('receipts', []).append(receipt) save() try: server = http.server.ThreadingHTTPServer(('127.0.0.1', PREFERRED_PORT), Recorder) except OSError as error: if getattr(error, 'winerror', None) != 10048 and getattr(error, 'errno', None) != 10048: raise server = http.server.ThreadingHTTPServer(('127.0.0.1', 0), Recorder) STATE['port'] = server.server_address[1] save() print(json.dumps({'pid': STATE['pid'], 'port': STATE['port'], 'forwards': 0, 'armed': False, 'runtimeFrozen': False}), flush=True) server.serve_forever()