"""Deterministic client reproduction. No Ollama daemon, model, SDK or external requests. Run: python3 ollama_client_lab.py [--output result.json] Only a temporary loopback HTTP server is started; it closes on completion. """ import argparse, json, platform, threading from datetime import datetime, timezone from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from urllib.request import Request, urlopen from urllib.error import HTTPError class IncompleteGeneration(RuntimeError): pass def read_native_generate(response): """Consume Ollama-native generate NDJSON; do not feed OpenAI SSE here.""" chunks = [] completed = False for raw in response: if not raw.strip(): continue event = json.loads(raw) if event.get('error'): raise IncompleteGeneration(event['error']) chunks.append(event.get('response', '')) if event.get('done') is True: completed = True break if not completed: raise IncompleteGeneration('Stream ended without done:true; output is incomplete') return ''.join(chunks) class Fixture(BaseHTTPRequestHandler): def log_message(self, *args): pass def do_POST(self): request=json.loads(self.rfile.read(int(self.headers['Content-Length']))) self.server.requests.append(self.path) if self.path=='/api/generate': self.send_response(200) if request.get('stream') is False: self.send_header('Content-Type','application/json'); self.end_headers() self.wfile.write(json.dumps({'response':'A complete answer.','done':True}).encode());return self.send_header('Content-Type','application/x-ndjson');self.end_headers() events=[{'response':'A ','done':False},{'response':'complete answer.','done':False},{'response':'','done':True}] if request['prompt']=='fixture_error': events=[events[0],{'error':'synthetic model failure'}] if request['prompt']=='fixture_truncated': events=events[:1] for event in events: self.wfile.write((json.dumps(event)+'\n').encode());self.wfile.flush() elif self.path=='/api/chat': self.send_response(200);self.send_header('Content-Type','application/json');self.end_headers() self.wfile.write(json.dumps({'message':{'role':'assistant','content':'Native shape'},'done':True}).encode()) elif self.path=='/v1/chat/completions': self.send_response(200);self.send_header('Content-Type','application/json');self.end_headers() self.wfile.write(json.dumps({'choices':[{'message':{'role':'assistant','content':'Compatible shape'}}]}).encode()) else: self.send_response(404);self.send_header('Content-Type','application/json');self.end_headers() self.wfile.write(b'{"error":"synthetic fixture route not found"}') def run(): server=ThreadingHTTPServer(('127.0.0.1',0),Fixture);server.requests=[] thread=threading.Thread(target=server.serve_forever,daemon=True);thread.start() base=f'http://127.0.0.1:{server.server_port}' results=[] def post(path,**data): return urlopen(Request(base+path,data=json.dumps({'model':'fixture-only',**data}).encode(),headers={'Content-Type':'application/json'}),timeout=5) try: # Reproduce the actual Python JSON decoder failure over HTTP, then both fixes. with post('/api/generate',prompt='fixture_ok') as response: try: json.load(response) except json.JSONDecodeError as e: failure=e.msg else: raise AssertionError('Expected JSONDecodeError') assert failure=='Extra data' with post('/api/generate',prompt='fixture_ok',stream=False) as response: assert json.load(response)['response']=='A complete answer.' with post('/api/generate',prompt='fixture_ok') as response: assert read_native_generate(response)=='A complete answer.' results.append({'guide':'extra-data','reproduced':failure,'fixed':'stream:false and line-by-line reader both return complete answer','pass':True}) # Endpoint route mismatch and the separate response-shape mismatch. try: post('/api/chat/completions',messages=[]) except HTTPError as e: status=e.code;e.close() else: raise AssertionError('Expected HTTP 404') assert status==404 with post('/api/chat',messages=[],stream=False) as response: native=json.load(response) try: native['choices'][0]['message']['content'] except KeyError as e: shape_error=str(e) else: raise AssertionError('Expected wrong-shape failure') assert native['message']['content']=='Native shape' with post('/v1/chat/completions',messages=[],stream=False) as response: assert json.load(response)['choices'][0]['message']['content']=='Compatible shape' results.append({'guide':'endpoint-mismatch','reproduced':{'fixture_http_status':status,'wrong_shape_key_error':shape_error},'fixed':'paired native route/reader and compatible route/reader both pass','pass':True}) # HTTP 200 is insufficient when the response subsequently contains an error. with post('/api/generate',prompt='fixture_error') as response: assert response.status==200 naive=''.join(json.loads(line).get('response','') for line in response) assert naive=='A ' failures=[] for prompt in ('fixture_error','fixture_truncated'): with post('/api/generate',prompt=prompt) as response: try: read_native_generate(response) except IncompleteGeneration as e: failures.append(str(e)) else: raise AssertionError('Expected rejected incomplete generation') results.append({'guide':'http-200-incomplete','reproduced':{'http_status':200,'naive_partial_output':naive},'fixed':failures,'pass':True}) return {'checked_at':datetime.now(timezone.utc).isoformat(),'python':platform.python_version(),'platform':platform.system(),'scope':'Kingy client-level test against synthetic loopback HTTP fixtures; no Ollama daemon, model inference, OpenAI SDK, or Docker tested','external_requests':0,'model_downloads':0,'results':results,'requests':server.requests} finally: server.shutdown();server.server_close();thread.join(timeout=5) if __name__=='__main__': parser=argparse.ArgumentParser();parser.add_argument('--output');args=parser.parse_args() result=run();rendered=json.dumps(result,indent=2)+'\n' if args.output: from pathlib import Path Path(args.output).write_text(rendered) print(rendered)