Qualify and enable MTP 2 with paired workload evidence and fallback

scripts/make_qualification_assets.py

@@ -0,0 +1,16 @@
+"""Generate nonprivate OCR image and 120-second video; requires Pillow/OpenCV/numpy."""
+from pathlib import Path
+import argparse
+import cv2,numpy as np
+from PIL import Image,ImageDraw,ImageFont
+parser=argparse.ArgumentParser();parser.add_argument('--output',type=Path,default=Path('/data/flash-next/cache/qualification'));p=parser.parse_args().output;p.mkdir(parents=True,exist_ok=True)
+im=Image.new('RGB',(2048,1536),'white');draw=ImageDraw.Draw(im)
+font=ImageFont.truetype('/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf',72)
+for i,s in enumerate(['INVOICE','Verification: SPARK-82741','Amount A: 125','Amount B: 478','Amount C: 765']):draw.text((100,120+i*220),s,font=font,fill='black')
+im.save(p/'chart.png')
+w=cv2.VideoWriter(str(p/'sequence120.mp4'),cv2.VideoWriter_fourcc(*'mp4v'),2,(1280,720))
+assert w.isOpened()
+for name,color in [('RED',(0,0,255)),('GREEN',(0,255,0)),('BLUE',(255,0,0)),('YELLOW',(0,255,255)),('PURPLE',(128,0,128)),('WHITE',(255,255,255))]:
+ for n in range(40):
+  frame=np.full((720,1280,3),color,dtype=np.uint8);cv2.putText(frame,name,(300,400),cv2.FONT_HERSHEY_SIMPLEX,3,(0,0,0),7);w.write(frame)
+w.release();print('Assets generated',[(f.name,f.stat().st_size) for f in p.iterdir()])

scripts/qualification.py

@@ -0,0 +1,98 @@
+"""Paired API qualification. Run identically on baseline and MTP; synthetic data only."""
+import argparse,base64,concurrent.futures,hashlib,json,random,time,urllib.request
+from pathlib import Path
+p=argparse.ArgumentParser();p.add_argument('--output',type=Path,required=True);p.add_argument('--soak-seconds',type=int,default=600);p.add_argument('--assets',type=Path,default=Path('/data/flash-next/cache/qualification'));a=p.parse_args()
+key=(Path(__file__).resolve().parents[1]/'secrets/api-key').read_text().strip();url='http://127.0.0.1:8000';model='qwen3.8-flash-next'
+def req(path,body=None):
+ return urllib.request.urlopen(urllib.request.Request(url+path,data=None if body is None else json.dumps(body).encode(),headers={'Authorization':'Bearer '+key,'Content-Type':'application/json'}),timeout=1200)
+def run(name,messages,limit=4096,expected=None,forced=False,effort='low',extra=None):
+ body=dict(model=model,messages=messages,temperature=0,seed=6000,max_tokens=limit,reasoning_effort=effort,stream=True,stream_options={'include_usage':True});body.update(extra or {})
+ if forced:body['ignore_eos']=True
+ start=time.monotonic();first=None;content='';reasoning='';finish=None;usage={};events=0;max_gap=0;prev=None
+ try:
+  with req('/v1/chat/completions',body) as resp:
+   for raw in resp:
+    if not raw.startswith(b'data: '):continue
+    data=raw[6:].strip()
+    if data==b'[DONE]':break
+    x=json.loads(data)
+    if x.get('error'):raise RuntimeError(str(x['error']))
+    usage=x.get('usage') or usage
+    for c in x.get('choices',[]):
+     d=c.get('delta',{});s=d.get('content') or '';r=d.get('reasoning') or d.get('reasoning_content') or ''
+     if s or r:
+      now=time.monotonic();first=first or now;events+=1
+      if prev is not None:max_gap=max(max_gap,now-prev)
+      prev=now
+     content+=s;reasoning+=r;finish=c.get('finish_reason') or finish
+  end=time.monotonic();n=usage.get('completion_tokens',0)
+  passed=bool(first and n and finish)
+  if forced:passed=passed and n==limit and finish=='length'
+  elif expected is not None:
+   if isinstance(expected,list):passed=passed and all(str(v).lower() in content.lower() for v in expected) and finish=='stop'
+   else:passed=passed and content.strip()==expected and finish=='stop'
+  else:passed=passed and finish=='stop' and bool(content.strip())
+  return dict(name=name,passed=passed,finish=finish,usage=usage,ttft_s=None if first is None else first-start,elapsed_s=end-start,decode_tps=None if first is None else (n-1)/max(end-first,1e-6),max_stream_gap_s=max_gap,events=events,content=content,reasoning=reasoning,sha256=hashlib.sha256((content+reasoning).encode()).hexdigest())
+ except Exception as e:
+  detail=e.read().decode(errors='replace') if hasattr(e,'read') else str(e)
+  return dict(name=name,passed=False,error=detail,elapsed_s=time.monotonic()-start)
+def text(s):return [{'role':'user','content':s}]
+def media(kind,file):
+ mime='image/png' if kind=='image' else 'video/mp4';f=a.assets/file
+ return {'type':kind+'_url',kind+'_url':{'url':'data:'+mime+';base64,'+base64.b64encode(f.read_bytes()).decode()}}
+image=media('image','chart.png');video=media('video','sequence120.mp4')
+image_msgs=[{'role':'user','content':[{'type':'text','text':'Read the verification code and the three invoice amounts. Reply with the code and the sum of the amounts.'},image]}]
+video_msgs=[{'role':'user','content':[{'type':'text','text':'List the six background colors in chronological order. Reply only with the English color names.'},video]}]
+cases=[]
+for i in range(12):
+ x=137+i*23;y=29+i*7
+ cases.append((f'arithmetic-{i}',text(f'Compute {x} * {y} + {i}. Reply only with the integer.'),2048,str(x*y+i)))
+cases += [('json',text('Return ONLY valid JSON with exactly these values: {"name":"设备A","count":7,"enabled":true}. No markdown.'),2048,None),('multiturn',[{'role':'user','content':'Remember the code AQUA-58321.'},{'role':'assistant','content':'I will remember it.'},{'role':'user','content':'What was the code? Reply only with the code.'}],2048,'AQUA-58321'),('image-ocr',image_msgs,4096,['SPARK-82741','1368']),('video-120s',video_msgs,4096,['red','green','blue','yellow','purple','white']),('multi-image',[{'role':'user','content':[{'type':'text','text':'These four copies show the same invoice. What is its verification code? Reply only with the code.'},image,image,image,image]}],4096,'SPARK-82741')]
+a.output.parent.mkdir(parents=True,exist_ok=True)
+with a.output.open('x') as out:
+ def record(r):
+  out.write(json.dumps(r,ensure_ascii=False)+'\n');out.flush();print(json.dumps({k:v for k,v in r.items() if k not in ('content','reasoning')},ensure_ascii=False),flush=True)
+ def check(r):
+  if r['name']=='json' and r.get('passed'):
+   try:r['passed']=json.loads(r['content'])=={'name':'设备A','count':7,'enabled':True}
+   except Exception:r['passed']=False
+  if r['name'].startswith('video') and r.get('passed'):
+   indices=[r['content'].lower().find(c) for c in ['red','green','blue','yellow','purple','white']];r['passed']=indices==sorted(indices)
+  record(r)
+ for name,msgs,limit,expected in cases:check(run(name,msgs,limit,expected))
+ # Different useful workloads, fixed-length timing includes reasoning, not a quality score.
+ prompts=[('code','Write a production-quality Python CSV import utility with validation, duplicate handling, error reporting, unit tests, and a detailed explanation.'),('chinese','请写一份企业采购系统的完整需求说明，覆盖审批、预算、供应商、库存和审计，并逐项说明异常处理。'),('reasoning','Explain and prove why Dijkstra works with nonnegative weights. Include a counterexample with negative edges and compare Bellman-Ford.')]
+ for tag,prompt in prompts:
+  record(run('warmup-'+tag,text(prompt),256,forced=True,effort='xhigh'))
+  for i in range(3):record(run(f'perf-{tag}-{i}',text(prompt),2048,forced=True,effort='xhigh'))
+ record(run('long-output-16384',text('Write an extensive technical textbook chapter explaining algorithms, with many worked examples and Python implementations. Continue with additional examples throughout.'),16384,forced=True,effort='xhigh'))
+ # Exact token count, fresh random filler, two needles at separated positions.
+ rng=random.Random(927);lines=[f'Item {i:06d}: batch {rng.randrange(10**8):08d}, archived in shelf {rng.randrange(1000):03d}.\n' for i in range(16000)]
+ lines[20]='The northern access code is NORTH-49273.\n';lines[2500]='The southern access code is SOUTH-18362.\n'
+ question='\nGive both northern and southern access codes only.'
+ def longmsg(n):return text(''.join(lines[:n])+question)
+ lo,hi=2501,len(lines)
+ while lo<hi:
+  mid=(lo+hi+1)//2
+  with req('/tokenize',dict(model=model,messages=longmsg(mid),chat_template_kwargs={'reasoning_effort':'low'})) as resp:count=json.load(resp)['count']
+  if count<=126000:lo=mid
+  else:hi=mid-1
+ longmessages=longmsg(lo)
+ for i in range(3):record(run(f'long-context-{i}',longmessages,4096,['NORTH-49273','SOUTH-18362']))
+ # Smaller text budget leaves room for image tokens; request still exercises mixed long input.
+ mixed=[{'role':'user','content':[{'type':'text','text':''.join(lines[:max(2501,lo-600)])+'\nGive both access codes and the invoice verification code.'},image]}]
+ record(run('long-context-image',mixed,4096,['NORTH-49273','SOUTH-18362','SPARK-82741']))
+ # Queue pressure with max_num_seqs=1. This tests queued clients, not parallel GPU decoding.
+ with concurrent.futures.ThreadPoolExecutor(max_workers=4) as pool:
+  fs=[pool.submit(run,f'queued-{i}',text(f'Compute {43+i} * 17. Reply only with the integer.'),2048,str((43+i)*17)) for i in range(4)]
+  for f in concurrent.futures.as_completed(fs):record(f.result())
+ start=time.monotonic();i=0
+ while time.monotonic()-start<a.soak_seconds:
+  mode=i%5
+  if mode==0:r=run(f'soak-{i}-long',longmessages,4096,['NORTH-49273','SOUTH-18362'])
+  elif mode==1:r=run(f'soak-{i}-image',image_msgs,4096,['SPARK-82741','1368'])
+  elif mode==2:r=run(f'soak-{i}-video',video_msgs,4096,['red','green','blue','yellow','purple','white'])
+  else:r=run(f'soak-{i}-decode',text(prompts[i%3][1]+f'\nExample identifier: {i}.'),2048,forced=True,effort='xhigh')
+  record(r);i+=1
+  if 'error' in r:break
+ print('SUITE_COMPLETE',flush=True)

scripts/quality_tasks.py

@@ -0,0 +1,54 @@
+"""Normal-EOS task checks: extraction, scheduling, tool arguments, and guarded code tests."""
+import argparse,ast,json,subprocess,time,urllib.request
+from pathlib import Path
+p=argparse.ArgumentParser();p.add_argument('--output',type=Path,required=True);p.add_argument('--only');p.add_argument('--effort',default='xhigh',choices=['low','medium','xhigh']);p.add_argument('--max-tokens',type=int,default=8192);a=p.parse_args();key=(Path(__file__).resolve().parents[1]/'secrets/api-key').read_text().strip()
+def call(messages,**extra):
+ body=dict(model='qwen3.8-flash-next',messages=messages,temperature=0,seed=6000,max_tokens=a.max_tokens,reasoning_effort=a.effort);body.update(extra)
+ req=urllib.request.Request('http://127.0.0.1:8000/v1/chat/completions',data=json.dumps(body).encode(),headers={'Authorization':'Bearer '+key,'Content-Type':'application/json'})
+ with urllib.request.urlopen(req,timeout=600) as r:return json.load(r)
+# Only independently expected data; JSON parse accepts whitespace, not Markdown wrappers.
+cases=[('invoice-extraction','Extract this invoice to JSON with keys supplier, currency, total, due_date: Supplier North Star Ltd; currency CNY; rows: 3 items at 125 each, 2 items at 48 each; shipping 20; discount 15. Due date 2026-10-03. total must be an integer. Reply only JSON.',{'supplier':'North Star Ltd','currency':'CNY','total':476,'due_date':'2026-10-03'}),('deduplicate','Input orders: A pending 12; B paid 20; A paid 15; C cancelled 99; D paid 8; B paid 25. Keep the last occurrence of each ID, discard cancelled orders, and output ONLY JSON {"ids":[...],"total":...}. Sort IDs alphabetically.',{'ids':['A','B','D'],'total':48}),('schedule','Jobs A takes 3 hours; B takes 5 hours after A; C takes 4 hours after A; D takes 2 hours after both B and C. Unlimited workers, start at hour 0. Output only JSON with earliest_finish and critical_path (array of job letters).',{'earliest_finish':10,'critical_path':['A','B','D']}),('chinese-policy','制度：金额不超过5000元由主管审批；超过5000且不超过20000由经理审批；超过20000由总监审批。加急不改变审批人。申请：甲5000，乙5001加急，丙20000，丁20001。请仅输出JSON，键为甲乙丙丁，值为主管/经理/总监。',{'甲':'主管','乙':'经理','丙':'经理','丁':'总监'})]
+# No imports or arbitrary method calls permitted; child has CPU and address-space limits.
+harness='''import ast,json,resource,sys
+resource.setrlimit(resource.RLIMIT_CPU,(2,2));resource.setrlimit(resource.RLIMIT_AS,(256*1024**2,256*1024**2))
+x=json.load(sys.stdin);tree=ast.parse(x['code'])
+for n in ast.walk(tree):
+ if isinstance(n,(ast.Import,ast.ImportFrom,ast.ClassDef,ast.With,ast.AsyncWith,ast.Global,ast.Nonlocal)):raise ValueError('disallowed syntax')
+ if isinstance(n,ast.Name) and n.id.startswith('__'):raise ValueError('dunder name')
+ if isinstance(n,ast.Attribute) and n.attr not in ('append','sort','copy'):raise ValueError('disallowed attribute')
+ if isinstance(n,ast.Call):
+  if isinstance(n.func,ast.Name) and n.func.id not in ('sorted','len','range','min','max','list','enumerate','sum'):raise ValueError('disallowed call')
+  if not isinstance(n.func,(ast.Name,ast.Attribute)):raise ValueError('indirect call')
+ if isinstance(n,ast.FunctionDef) and (n.decorator_list or n.returns or any(z.annotation for z in n.args.args)):raise ValueError('annotations/decorators unsupported')
+if any(not isinstance(n,ast.FunctionDef) for n in tree.body):raise ValueError('functions only')
+g={'__builtins__':{k:__builtins__.__dict__[k] for k in ['sorted','len','range','min','max','list','enumerate','sum']}}
+exec(compile(tree,'candidate','exec'),g)
+for args,expected in x['tests']:
+ result=g[x['function']](*args)
+ if result!=expected:raise ValueError('wrong result: '+repr(result))
+print('passed')
+'''
+codecases=[('merge-intervals','merge_intervals','Merge overlapping integer intervals, including touching endpoints. Return sorted merged intervals as lists.',[[[[]],[]],[[[[1,3],[2,6],[8,10],[15,18]]],[[1,6],[8,10],[15,18]]],[[[[1,4],[4,5]]],[[1,5]]],[[[[5,7],[1,9],[2,3]]],[[1,9]]]]),('binary-search','lower_bound','Return the index of the first element >= target in a sorted list nums, or len(nums) if none. Parameters nums, target.',[[[[],3],0],[[[1,2,2,4],2],1],[[[1,2,2,4],3],3],[[[1,2,2,4],9],4],[[[1,2,2,4],0],0]])]
+with a.output.open('x') as out:
+ def save(row):out.write(json.dumps(row,ensure_ascii=False)+'\n');out.flush();print(json.dumps({k:v for k,v in row.items() if k!='response'},ensure_ascii=False),flush=True)
+ for name,prompt,expected in cases:
+  if a.only and a.only!=name:continue
+  r=None
+  start=time.monotonic()
+  try:
+   r=call([{'role':'user','content':prompt}]);c=r['choices'][0];actual=json.loads(c['message']['content']);passed=actual==expected and c['finish_reason']=='stop';save(dict(name=name,passed=passed,elapsed_s=time.monotonic()-start,response=r))
+  except Exception as e:save(dict(name=name,passed=False,error=str(e),response=r))
+ for name,fn,task,tests in codecases:
+  if a.only and a.only!=name:continue
+  r=None
+  try:
+   r=call([{'role':'user','content':f'Write Python function {fn}. {task} Reply ONLY a JSON object with a code string. Use no imports, type annotations, decorators, top-level statements or helper functions. Allowed builtins: sorted,len,range,min,max,list,enumerate,sum. Allowed methods: append,sort,copy.'}]);c=r['choices'][0];code=json.loads(c['message']['content'])['code'];result=subprocess.run(['python3','-I','-c',harness],input=json.dumps(dict(code=code,function=fn,tests=tests)),capture_output=True,text=True,timeout=5);save(dict(name=name,passed=result.returncode==0 and c['finish_reason']=='stop',validation_stdout=result.stdout,validation_stderr=result.stderr,response=r))
+  except Exception as e:save(dict(name=name,passed=False,error=str(e),response=r))
+ tools=[dict(type='function',function=dict(name='create_ticket',description='Create a support ticket',parameters=dict(type='object',properties={'title':{'type':'string'},'priority':{'type':'string','enum':['low','normal','urgent']},'asset':{'type':'string'}},required=['title','priority','asset'])))]
+ r=None
+ try:
+  if a.only and a.only!='tool-arguments':
+   print('QUALITY_COMPLETE',flush=True);raise SystemExit(0)
+  r=call([{'role':'user','content':'Call create_ticket with title exactly GPU cooling failure, priority urgent, and asset exactly GPU-092.'}],tools=tools,tool_choice='auto');c=r['choices'][0];tc=c['message']['tool_calls'];passed=len(tc)==1 and tc[0]['function']['name']=='create_ticket' and json.loads(tc[0]['function']['arguments'])=={'title':'GPU cooling failure','priority':'urgent','asset':'GPU-092'};save(dict(name='tool-arguments',passed=passed,response=r))
+ except Exception as e:save(dict(name='tool-arguments',passed=False,error=str(e),response=r))
+print('QUALITY_COMPLETE',flush=True)

scripts/summarize_qualification.py

@@ -0,0 +1,31 @@
+"""Summarize complete/partial receipts without dropping failures."""
+import argparse,json,statistics
+from pathlib import Path
+p=argparse.ArgumentParser();p.add_argument('directory',type=Path);a=p.parse_args();result={}
+for variant in ['baseline','mtp2']:
+ path=a.directory/(variant+'.jsonl')
+ if not path.exists():continue
+ rows=[json.loads(l) for l in path.read_text().splitlines() if l];summary={'requests':len(rows),'failures':[{'name':x['name'],'error':x.get('error'),'content':x.get('content')} for x in rows if not x['passed']],'timing':{}}
+ for tag in ['code','chinese','reasoning']:
+  samples=[r for r in rows if r['name'].startswith('perf-'+tag+'-') and r.get('decode_tps')]
+  if samples:summary['timing'][tag]={'samples':len(samples),'decode_tps_median':statistics.median(r['decode_tps'] for r in samples),'ttft_s_median':statistics.median(r['ttft_s'] for r in samples),'elapsed_s_median':statistics.median(r['elapsed_s'] for r in samples),'reasoning_tokens':[r['usage'].get('completion_tokens_details',{}).get('reasoning_tokens') for r in samples]}
+ soak=[r for r in rows if r['name'].startswith('soak-')];summary['soak']={'requests':len(soak),'sum_request_seconds':sum(r.get('elapsed_s',0) for r in soak),'failures':sum(not r['passed'] for r in soak)}
+ long=[r for r in rows if r['name']=='long-output-16384'];summary['long_output']=[{k:v for k,v in r.items() if k not in ('content','reasoning')} for r in long]
+ quality=a.directory/(variant+'-quality.jsonl')
+ if quality.exists():
+  q=[json.loads(l) for l in quality.read_text().splitlines()];summary['quality']={'tasks':len(q),'failures':[r['name'] for r in q if not r['passed']]}
+ result[variant]=summary
+if all(v in result for v in ['baseline','mtp2']):
+ result['speedup']={k:result['mtp2']['timing'][k]['decode_tps_median']/v['decode_tps_median'] for k,v in result['baseline']['timing'].items() if k in result['mtp2']['timing']}
+for variant in ['baseline','mtp2']:
+ path=a.directory/(variant+'-quality-medium.jsonl')
+ if path.exists() and variant in result:
+  q=[json.loads(l) for l in path.read_text().splitlines()];result[variant]['quality_medium']={'tasks':len(q),'failures':[r['name'] for r in q if not r['passed']]}
+rpath=a.directory/'resources.jsonl'
+if rpath.exists():
+ rows=[json.loads(l) for f in [rpath,a.directory/'resources-resumed.jsonl'] if f.exists() for l in f.read_text().splitlines() if l];gpus=[]
+ for r in rows:
+  try:gpus.append([float(x.strip()) for x in r['gpu'].split(',')])
+  except (ValueError,KeyError):pass
+ result['resources']={'samples':len(rows),'min_host_available_gib':min(r['available_gib'] for r in rows),'oom_kill_max':max(r['oom_kill'] for r in rows),'safeguard_triggered':any(r['stop'] for r in rows),'gpu_memory_peak_mib':max((g[0] for g in gpus),default=None),'gpu_temperature_peak_c':max((g[3] for g in gpus),default=None)}
+print(json.dumps(result,ensure_ascii=False,indent=2))