"""Local-only experiment. No API key, cloud provider or public listener."""
from pathlib import Path
import datetime, hashlib, json, os, socket, subprocess, sys, threading, time, urllib.request

LAB = Path(__file__).resolve().parent
sys.path.insert(0, str(LAB / 'deps'))
import psutil
OUT = LAB.parents[1] / 'outputs' / 'qwen-local-test'
OUT.mkdir(parents=True, exist_ok=True)
PRIVATE = LAB / 'private'
PRIVATE.mkdir(exist_ok=True)
BASE = 'http://127.0.0.1:11434'
LOCAL_HTTP = urllib.request.build_opener(urllib.request.ProxyHandler({}))
FIXTURE = json.loads((LAB/'fixtures.json').read_text(encoding='utf-8'))
MODEL = FIXTURE['model']
EXE = LAB / 'runtime' / 'ollama.exe'
ENV = os.environ.copy()
ENV.update(OLLAMA_HOST='127.0.0.1:11434', OLLAMA_MODELS=str(LAB/'models'), OLLAMA_NO_CLOUD='1', OLLAMA_NUM_PARALLEL='1', OLLAMA_MAX_LOADED_MODELS='1', OLLAMA_VULKAN='0', GGML_VK_VISIBLE_DEVICES='-1')
# Only this child process gets an isolated Windows profile path. This avoids
# accessing the real user-profile directory or changing account-wide settings.
ISOLATED_PROFILE = PRIVATE / 'ollama-profile'
ISOLATED_PROFILE.mkdir(exist_ok=True)
ENV['USERPROFILE'] = str(ISOLATED_PROFILE)

def save(name, value):
    target = OUT/name
    temp = target.with_suffix(target.suffix+'.tmp')
    temp.write_text(json.dumps(value, indent=2, ensure_ascii=False), encoding='utf-8')
    temp.replace(target)

def api(route, payload=None, timeout=180):
    req=urllib.request.Request(BASE+route, data=None if payload is None else json.dumps(payload).encode(), headers={'Content-Type':'application/json'})
    with LOCAL_HTTP.open(req, timeout=timeout) as response:
        return json.load(response)

def cli(*args):
    return subprocess.run([str(EXE),*args], env=ENV, capture_output=True, text=True, timeout=30, creationflags=subprocess.CREATE_NO_WINDOW)

def status(phase, **values):
    record={'phase':phase,'utc':datetime.datetime.now(datetime.timezone.utc).isoformat(),**values}
    save('status.json',record)
    print(json.dumps(record),flush=True)

def memory_sample(server, finish, samples):
    while not finish.is_set():
        try:
            processes=[psutil.Process(server.pid)]+psutil.Process(server.pid).children(recursive=True)
            memories=[p.memory_info() for p in processes if p.is_running()]
            samples.append({'seconds':time.perf_counter(),'rss_bytes_sum':sum(m.rss for m in memories),'private_bytes_sum':sum(getattr(m,'private',0) for m in memories),'process_count':len(memories)})
        except (psutil.NoSuchProcess,psutil.AccessDenied):
            pass
        finish.wait(.25)

def generate(server, task_id, prompt, run, tokens=256):
    payload={'model':MODEL,'prompt':prompt,'stream':True,'think':False,'keep_alive':'5m','options':{**FIXTURE['options'],'num_predict':tokens}}
    start=time.perf_counter(); first=None; chunks=[]; final=None; error=None
    samples=[]; finish=threading.Event()
    monitor=threading.Thread(target=memory_sample,args=(server,finish,samples),daemon=True)
    monitor.start()
    response=None; deadline_hit=threading.Event()
    def deadline():
        deadline_hit.set()
        try:
            owned=psutil.Process(server.pid)
            for child in owned.children(recursive=True):
                try: child.kill()
                except psutil.NoSuchProcess: pass
            owned.kill()
        except psutil.NoSuchProcess: pass
    timer=threading.Timer(180,deadline); timer.daemon=True; timer.start()
    try:
        req=urllib.request.Request(BASE+'/api/generate',data=json.dumps(payload).encode(),headers={'Content-Type':'application/json'})
        response=LOCAL_HTTP.open(req,timeout=180)
        for line in response:
            if time.perf_counter()-start>180:
                raise TimeoutError('180-second operational deadline')
            event=json.loads(line)
            if event.get('error'):
                raise RuntimeError(event['error'])
            text=event.get('response','')
            if text and first is None: first=time.perf_counter()-start
            chunks.append(event)
            if event.get('done'): final=event
        if final is None: error='Stream ended without a completed response'
    except Exception as exc:
        error=type(exc).__name__+': '+str(exc)
    finally:
        elapsed=time.perf_counter()-start
        timer.cancel()
        if response is not None: response.close()
        finish.set(); monitor.join(2)
    if deadline_hit.is_set(): error='180-second watchdog terminated the owned local server and runner'
    result={'task':task_id,'run':run,'model':MODEL,'request':payload,'output':''.join(e.get('response','') for e in chunks),'stream_events':chunks,'client_seconds':elapsed,'first_content_seconds':first,'error':error,'metrics':final,'memory_sampling_interval_seconds':.25,'sampled_peak_rss_sum_bytes':max((s['rss_bytes_sum'] for s in samples),default=0),'sampled_peak_private_sum_bytes':max((s['private_bytes_sum'] for s in samples),default=0),'memory_sample_count':len(samples)}
    if final and final.get('eval_duration'):
        result['generation_tokens_per_second']=final['eval_count']/(final['eval_duration']/1e9)
    save(f'{task_id}-{run}.json',result)
    try:
        result['model_placement']=api('/api/ps',timeout=15)
    except Exception as exc:
        result['placement_error']=type(exc).__name__+': '+str(exc)
    save(f'{task_id}-{run}.json',result)
    status('request-complete',task=task_id,run=run,seconds=round(elapsed,2),error=error)
    return result

def main():
    assert EXE.is_file(), 'Runtime not ready'
    assert not list(OUT.glob('smoke-*.json')) and not list(OUT.glob('extraction-[123].json')), 'Existing first-run evidence: create a separate labeled attempt rather than overwrite'
    sock=socket.socket()
    assert sock.connect_ex(('127.0.0.1',11434))!=0, 'Existing listener: do not disturb it'
    sock.close()
    save('fixtures.json',FIXTURE)
    save('fixture-freeze.json',{'utc':datetime.datetime.now(datetime.timezone.utc).isoformat(),'sha256':hashlib.sha256((LAB/'fixtures.json').read_bytes()).hexdigest()})
    log=(PRIVATE/'server.log').open('wb')
    server=subprocess.Popen([str(EXE),'serve'],env=ENV,stdout=log,stderr=log,creationflags=subprocess.CREATE_NO_WINDOW)
    (PRIVATE/'server-pid.json').write_text(json.dumps({'pid':server.pid}),encoding='utf-8')
    try:
        for attempt in range(30):
            if server.poll() is not None: raise RuntimeError('Server exited; inspect private log')
            try:
                version=api('/api/version',timeout=2); break
            except Exception: time.sleep(1)
        else: raise RuntimeError('Server did not become ready')
        save('configuration.json',{'version':version,'model':MODEL,'cloud_disabled':True,'host':'127.0.0.1:11434','vulkan_disabled':True,'parallel_requests':1,'options':FIXTURE['options'],'psutil_version':psutil.__version__,'date':datetime.date.today().isoformat()})
        status('model-download-started',model=MODEL)
        req=urllib.request.Request(BASE+'/api/pull',data=json.dumps({'model':MODEL,'stream':True}).encode(),headers={'Content-Type':'application/json'})
        last=time.monotonic(); pull_events=[]
        with LOCAL_HTTP.open(req,timeout=300) as response:
            for line in response:
                event=json.loads(line)
                if event.get('error'): raise RuntimeError(event['error'])
                if time.monotonic()-last>20 or event.get('status')=='success':
                    status('model-download',**event); last=time.monotonic()
                    pull_events.append(event)
        save('pull-events.json',pull_events)
        show=api('/api/show',{'model':MODEL})
        save('model-metadata.json',{k:show.get(k) for k in ('details','model_info','parameters','capabilities','modified_at')})
        save('model-list.json',api('/api/tags'))
        manifests=list((LAB/'models'/'manifests').rglob('9b'))
        assert len(manifests)==1
        manifest=manifests[0].read_bytes()
        save('manifest.json',json.loads(manifest))
        save('model-download-record.json',{'manifest_sha256':hashlib.sha256(manifest).hexdigest(),'model_files_bytes':sum(p.stat().st_size for p in (LAB/'models').rglob('*') if p.is_file())})
        for n in range(2):
            smoke=generate(server,'smoke','Reply with exactly: Local model ready.',n+1,128)
            if smoke['error'] or (n==1 and smoke['client_seconds']>90):
                status('diagnosis-required',reason='Smoke error or predeclared latency threshold'); return
        save('placement-cli.json',{'command':'ollama ps','stdout':cli('ps').stdout})
        demo=subprocess.run(['powershell.exe','-NoProfile','-NonInteractive','-File',str(LAB/'first-answer.ps1')],env=ENV,capture_output=True,text=True,timeout=185,creationflags=subprocess.CREATE_NO_WINDOW)
        save('powershell-demo.json',{'exit_code':demo.returncode,'stdout':demo.stdout,'stderr':demo.stderr,'note':'Separate nonstreaming walkthrough check, excluded from timing comparisons.'})
        if demo.returncode:
            status('walkthrough-command-failed',reason='PowerShell demo needs diagnosis before publication')
        for task in FIXTURE['tasks']:
            api('/api/generate',{'model':MODEL,'keep_alive':0})
            unloaded=api('/api/ps')
            save(task['id']+'-before-cold.json',unloaded)
            assert not unloaded.get('models'), 'Model not unloaded'
            for n in range(3):
                result=generate(server,task['id'],task['prompt'],n+1)
                if result['error']:
                    status('diagnosis-required',reason='Task request failed'); return
        status('tests-complete',streamed_requests=11,powershell_walkthrough_attempts=1)
    finally:
        try: owned_children=psutil.Process(server.pid).children(recursive=True)
        except psutil.NoSuchProcess: owned_children=[]
        try: api('/api/generate',{'model':MODEL,'keep_alive':0},timeout=15)
        except Exception: pass
        server.terminate()
        try: server.wait(timeout=10)
        except subprocess.TimeoutExpired: server.kill(); server.wait(timeout=5)
        for child in owned_children:
            try:
                if child.is_running(): child.terminate()
            except psutil.NoSuchProcess: pass
        _, alive=psutil.wait_procs(owned_children,timeout=5)
        for child in alive:
            try: child.kill()
            except psutil.NoSuchProcess: pass
        save('cleanup.json',{'server_stopped':server.poll() is not None,'owned_children_stopped':all(not p.is_running() for p in owned_children)})
        log.close()

if __name__=='__main__': main()
