Private CPU-first synthetic compute tests with temporary-file cleanup

app.py

@@ -1,43 +1,27 @@
-import spaces  # Must precede libraries that interact with CUDA.
+import spaces
 import gradio as gr
-from probe_core import qualify, verify_artifacts
-from training_smoke import validate_uploads, execute_smoke
+from cloud_probe import run_probe
 
-# Integrity only: no device initialization or CUDA library loading at startup.
-verify_artifacts()
-
-@spaces.GPU(duration=45)
-def probe():
-    return qualify()
+def cpu_smoke():
+    return run_probe(False)
 
 @spaces.GPU(duration=90)
-def leased_smoke(template):
-    return execute_smoke(template)
-
-def smoke(track, network):
-    # Reject unsupported inputs on CPU before requesting a GPU lease.
-    return leased_smoke(validate_uploads(track, network))
+def gpu_smoke():
+    return run_probe(True)
 
 with gr.Blocks() as demo:
-    gr.Markdown("# TAO · ZeroGPU compatibility check\n"
-                "Private, bounded native-library qualification. No training, CPU fallback, "
-                "or automatic runs. Uses up to 45 seconds of the caller's ZeroGPU allowance. "
-                "Check included quota before calling. Actual GPU identity is recorded; no H200 assumption.")
-    button = gr.Button("Run one compatibility check")
-    result = gr.JSON(label="Qualification receipt")
-    button.click(probe, inputs=[], outputs=result, api_name="probe", concurrency_limit=1)
-    gr.Markdown("## Bounded upload test\nUpload the supplied A01 track and a 6–4–2 Formula .evolution. "
-                "Fixed at 4 vehicles, 1 generation, 60 physics ticks, seed 42. "
-                "Checks compatibility again inside a new lease. Up to 90 seconds of included quota. "
-                "No CPU fallback, parity or performance claim. A candidate is not evidence of improvement.")
-    track = gr.File(label="A01 Beginner track", file_types=[".track"], type="filepath")
-    network = gr.File(label="Formula network", file_types=[".evolution"], type="filepath")
-    run = gr.Button("Run bounded training test")
-    report = gr.JSON(label="Training receipt")
-    candidate = gr.File(label="Candidate network")
-    receipt = gr.File(label="Download receipt")
-    run.click(smoke, inputs=[track, network], outputs=[report, candidate, receipt],
-              api_name="smoke", concurrency_limit=1)
+    gr.Markdown('''# AI Learns To Drive — private compute test
+Simulator by [Zorro909](https://github.com/Zorro909/driving_sim_rs).
+UI and HF implementation by TiGa (seeker-cyber-maker).
+
+Fixed synthetic network, 4 cars, 1 generation, 60 ticks. CPU runs without a GPU lease.
+GPU uses up to 90 seconds of included allowance; check quota first. No automatic runs.
+No user uploads; temporary model files are deleted before the receipt is returned.
+This tests backend execution, not useful learning, performance, or full UI integration.
+''')
+    report=gr.JSON(label='Execution receipt')
+    gr.Button('Run CPU-only test').click(cpu_smoke,outputs=report,api_name='cpu_smoke',concurrency_limit=1,concurrency_id='native')
+    gr.Button('Run GPU test').click(gpu_smoke,outputs=report,api_name='gpu_smoke',concurrency_limit=1,concurrency_id='native')
 
-if __name__ == "__main__":
+if __name__=='__main__':
     demo.queue(max_size=1).launch()

cloud_probe.py

@@ -0,0 +1,63 @@
+"""Fixed synthetic CPU/CUDA smoke test; no uploads or retained model artifacts."""
+import gzip
+import hashlib
+import json
+import os
+from pathlib import Path
+import subprocess
+import tempfile
+import time
+
+ROOT = Path(__file__).resolve().parent
+
+def run_probe(gpu=False):
+    from probe_core import verify_artifacts, qualify
+    verify_artifacts()
+    qualification = qualify() if gpu else None
+    if gpu and not qualification['qualified_for_next_test']:
+        return {'status':'qualification_failed','qualification':qualification}
+    binary=ROOT/'bin/altd-sim'
+    binary.chmod(0o755)
+    network={'shape':[1,2],'weights':[[[0.0,0.0]]],'biases':[[0.0,1.0]],
+             'inputs':['speed'],'outputs':['Steering','Acceleration']}
+    model={'sensor_layout':{'names':['speed'],'sensors':[{'$type':'speed'}]}}
+    settings={'population':4,'selection_algorithm':'best','selection_size':1,
+        'crossover':'none','mutation_rate':0.01,'adaptive_mutation':False,'weight_decay':0,
+        'preserve_parents':'on_custom','preserve_parents_size':1,
+        'rewards':[{'metric':'total_score','type':'default','weight':1}]}
+    report={'backend_requested':'cuda' if gpu else 'cpu','population':4,'generations':1,
+        'ticks':60,'seed':42,'qualification':qualification,'speedup_verified':False}
+    with tempfile.TemporaryDirectory(prefix='tiga-probe-') as directory:
+        work=Path(directory)
+        for name,value in [('network',network),('model',model),('settings',settings)]:
+            (work/(name+'.json')).write_text(json.dumps(value))
+        command=[str(binary),'--math-profile','proton','--threads','1','train-scratch']
+        if gpu: command+=['--gpu']
+        command+=['--scene',str(ROOT/'tracks/A01 Beginner.scene.json'),
+            '--spawn-trace',str(ROOT/'tracks/A01 Beginner.spawn.json'),
+            '--network',str(work/'network.json'),'--init-network',str(work/'network.json'),
+            '--model',str(work/'model.json'),'--settings',str(work/'settings.json'),
+            '--shape','1,2','--population','4','--generations','1','--ticks','60',
+            '--seed','42','--batch-count','2','--out-dir',str(work/'out'),
+            '--mutation-start','.01','--mutation-end','.01','--schedule','linear','--save-final-candidate']
+        # Preserve only native-loader/device plumbing, never provider credentials.
+        keys=('PATH','LD_LIBRARY_PATH','LD_PRELOAD','CUDA_VISIBLE_DEVICES','NVIDIA_VISIBLE_DEVICES')
+        env={k:os.environ[k] for k in keys if k in os.environ}
+        env.update(HOME=directory,TMPDIR=directory,ALTD_GPU_LIB=str(ROOT/'bin/libaltd_gpu.so'))
+        started=time.perf_counter()
+        try:
+            result=subprocess.run(command,cwd=work,env=env,capture_output=True,text=True,timeout=30)
+            report.update(returncode=result.returncode,wall_seconds=time.perf_counter()-started,
+                          stderr=result.stderr[-4000:],stdout=result.stdout[-4000:])
+            candidate=work/'out/candidate.json'
+            report['status']='failed'
+            if result.returncode==0 and candidate.is_file():
+                value=json.loads(candidate.read_text())
+                report.update(status='candidate_created',shape=value['shape'],
+                    candidate_sha256=hashlib.sha256(candidate.read_bytes()).hexdigest(),
+                    history=[json.loads(line) for line in (work/'out/log.jsonl').read_text().splitlines()])
+        except subprocess.TimeoutExpired:
+            report.update(status='timeout',wall_seconds=time.perf_counter()-started)
+    report['temporary_directory_removed']=not work.exists()
+    report['candidate_retained']=False
+    return report