catalog_runtime.py

"""Generic ComfyUI/Irodori app backend supplied by the HF repository.

The fixed v53 image executes this code as uid10001. HTML is hosted separately.
Only a job-scoped Cloudflare capability is received; no provider keys are needed.
"""
from __future__ import annotations
import base64,hashlib,json,math,mimetypes,os,re,shutil,subprocess,sys,time,urllib.error,urllib.parse,urllib.request,uuid
from pathlib import Path

TRUSTED_ORIGIN=os.environ['APP_SERVICE_ORIGIN']
USER_AGENT='app-HF/1.0'
INPUT_ROOT=Path('/comfyui/input')
OUTPUT_ROOT=Path('/comfyui/output')
TEMP_ROOT=Path('/comfyui/temp')
CHUNK_BYTES=8*1024*1024
MAX_ARTIFACT=512*1024*1024
MAX_TOTAL=2*1024*1024*1024
MAX_FILES=16
MEDIA_MARKERS=('loadimage','loadvideo','loadaudio')
MEDIA_KEYS={'image','images','imagefile','imagepath','video','videos','videofile','videopath','audio','audiofile','audiopath','file','filename','path','filepath','directory','folder','url','uri'}
MODEL_KEYS={'ckpt_name','unet_name','lora_name','vae_name','clip_name','clip_name1','clip_name2','clip_name3','clip_name4','control_net_name','model_name','text_encoder','text_encoder_name','tiny_vae','mmaudio_model','clip_model','vae_model','synchformer_model'}
DENIED_MARKERS={'python','shell','terminal','subprocess','executecommand','systemcommand','commandline','comfyinstall','comfymanager'}
MIMES={'.mp4':'video/mp4','.webm':'video/webm','.mov':'video/quicktime','.mkv':'video/x-matroska','.avi':'video/x-msvideo','.m4v':'video/mp4','.wav':'audio/wav','.flac':'audio/flac','.mp3':'audio/mpeg','.ogg':'audio/ogg','.opus':'audio/ogg','.m4a':'audio/mp4','.png':'image/png','.jpg':'image/jpeg','.jpeg':'image/jpeg','.webp':'image/webp','.gif':'image/gif','.avif':'image/avif'}

class Deadline:
    def __init__(self,seconds=280):self.end=time.monotonic()+seconds
    def remaining(self,limit=None):
        remaining=self.end-time.monotonic()
        if remaining<=0:raise TimeoutError('Generation and artifact transfer exceeded the 280-second app limit')
        return min(remaining,limit) if limit is not None else remaining

class NoRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(self,*args,**kwargs):raise ValueError('Media service redirects are not accepted')

def trusted_url(value):
    if not isinstance(value,str) or len(value)>4096:raise ValueError('Invalid media service URL')
    parsed=urllib.parse.urlsplit(value)
    if parsed.scheme!='https' or parsed.netloc!=urllib.parse.urlsplit(TRUSTED_ORIGIN).netloc or parsed.username or parsed.password or parsed.fragment:
        raise ValueError('Media URL must use the trusted Cloudflare service origin')
    return value

def set_read_timeout(response,deadline):
    remaining=deadline.remaining(25)
    # urllib's read timeout is otherwise reset per read, allowing a slow peer to
    # retain the original timeout after the overall job budget has elapsed.
    sock=getattr(getattr(getattr(response,'fp',None),'raw',None),'_sock',None)
    if sock is not None:sock.settimeout(remaining)

def service_error(label,error,deadline):
    details=['HTTP '+str(error.code)]
    # Never reflect capability URLs, HTML, echoed request data, or credentials.
    try:
        set_read_timeout(error,deadline)
        snippet=error.read(8192).decode('utf-8',errors='replace')
        match=re.search(r'(?:error[ _-]*code|cloudflare[ _-]*error)\s*[:=]\s*["\']?(1[0-9]{3})\b',snippet,re.I)
        if match:details.append('Cloudflare '+match.group(1))
        if error.headers.get('cf-mitigated')=='challenge':details.append('Cloudflare challenge')
        ray=error.headers.get('cf-ray','')
        if re.fullmatch(r'[0-9a-fA-F]{8,32}-[A-Z]{3}',ray):details.append('Ray '+ray)
    except Exception:pass
    return RuntimeError(label+' ('+', '.join(details)+')')

def trusted_network_test(url,deadline):
    trusted_url(url)
    request=urllib.request.Request(url,headers={'User-Agent':USER_AGENT,'Range':'bytes=0-0','Accept':'application/octet-stream'})
    try:
        with urllib.request.build_opener(NoRedirect()).open(request,timeout=deadline.remaining(25)) as response:
            set_read_timeout(response,deadline);received=len(response.read(1))
            return {'status':response.status,'sample_bytes':received,'content_type':response.headers.get('Content-Type'),'content_length':response.headers.get('Content-Length'),'cache_status':response.headers.get('CF-Cache-Status')}
    except urllib.error.HTTPError as error:raise service_error('Network diagnostic failed',error,deadline) from None
    except urllib.error.URLError:raise RuntimeError('Network diagnostic connection failed') from None

def request_json(url,method,body,deadline,headers=None):
    trusted_url(url)
    raw=body if isinstance(body,bytes) else json.dumps(body,separators=(',',':'),allow_nan=False).encode()
    actual={'User-Agent':USER_AGENT,'Content-Type':'application/octet-stream' if isinstance(body,bytes) else 'application/json',**(headers or {})}
    req=urllib.request.Request(url,data=raw,method=method,headers=actual)
    try:
        with urllib.request.build_opener(NoRedirect()).open(req,timeout=deadline.remaining(25)) as response:
            set_read_timeout(response,deadline)
            encoded=response.read(65537)
            if len(encoded)>65536:raise ValueError('Media service JSON response is too large')
            result=json.loads(encoded)
            if not isinstance(result,dict):raise ValueError('Invalid media service response')
            return result
    except urllib.error.HTTPError as error:
        # Capability URLs must never appear in error messages returned to the user.
        raise service_error('Media service request failed',error,deadline) from None
    except urllib.error.URLError as error:
        raise RuntimeError('Media service connection failed: '+str(type(error.reason).__name__)) from None

def relative_name(value):
    return isinstance(value,str) and bool(value) and len(value)<=512 and not value.startswith('/') and '\\' not in value and ':' not in value and not any(part in ('','..','.') for part in value.split('/')) and '\x00' not in value

def reference_name(value):
    if not isinstance(value,str):return None
    if value.startswith('input/'):value=value[6:]
    if value.endswith(' [input]'):value=value[:-8]
    return value if relative_name(value) else None

def safe_cache_path(root,value,directory=False):
    if not relative_name(value):raise ValueError('Invalid cached model path')
    source=Path(root)/value
    # Native HF snapshot files are symlinks into this repository's blob directory.
    resolved=source.resolve();repo=Path(root).parent.parent.resolve()
    if not (resolved.is_relative_to(Path(root).resolve()) or resolved.is_relative_to(repo)):
        raise ValueError('Cached model escaped its repository')
    if not (source.is_dir() if directory else source.is_file()):raise ValueError('Required file is missing from the native HF cache: '+value)
    return source

def prepare_files(files,job,deadline):
    if files is None:files=[]
    if not isinstance(files,list) or len(files)>20:raise ValueError('files must contain at most20 items')
    directory=INPUT_ROOT/'hf-jobs'/job
    directory.mkdir(parents=True,exist_ok=False)
    references={};total=0
    try:
        for item in files:
            deadline.remaining()
            if not isinstance(item,dict):raise ValueError('Invalid input file')
            name=item.get('name')
            if not isinstance(name,str) or not re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9._-]{0,159}',name) or name in references:
                raise ValueError('Input file names must be unique safe basenames')
            if Path(name).suffix.lower() not in MIMES:raise ValueError('Unsupported input media extension: '+name)
            expected=item.get('size');expected_hash=item.get('sha256')
            if expected is not None and (isinstance(expected,bool) or not isinstance(expected,int) or expected<1 or expected>MAX_ARTIFACT):
                raise ValueError('Input file size is invalid or exceeds512MiB')
            if expected_hash is not None and not re.fullmatch(r'[a-f0-9]{64}',str(expected_hash)):
                raise ValueError('Input SHA256 must be lowercase hexadecimal')
            target=directory/(uuid.uuid4().hex+Path(name).suffix.lower())
            digest=hashlib.sha256();size=0
            with target.open('xb') as out:
                if item.get('url'):
                    trusted_url(item['url'])
                    if expected is None or expected_hash is None:raise ValueError('Staged inputs require size and sha256')
                    try:
                        request=urllib.request.Request(item['url'],headers={'User-Agent':USER_AGENT,'Accept':'application/octet-stream'})
                        response=urllib.request.build_opener(NoRedirect()).open(request,timeout=deadline.remaining(25))
                    except urllib.error.HTTPError as error:raise service_error('Input media download failed',error,deadline) from None
                    except urllib.error.URLError as error:raise RuntimeError('Input media download connection failed') from None
                    with response:
                        while True:
                            deadline.remaining()
                            set_read_timeout(response,deadline)
                            chunk=response.read(256*1024)
                            if not chunk:break
                            size+=len(chunk);total+=len(chunk)
                            if size>expected or size>MAX_ARTIFACT or total>MAX_TOTAL:raise ValueError('Input media size exceeds declared or job limits')
                            digest.update(chunk);out.write(chunk)
                elif isinstance(item.get('data'),str):
                    encoded=item['data'].split(',',1)[-1]
                    if len(encoded)>768*1024:raise ValueError('Inline media is too large; stage it through Cloudflare first')
                    try:raw=base64.b64decode(encoded,validate=True)
                    except ValueError:raise ValueError('Invalid input base64') from None
                    size=len(raw);total+=size;digest.update(raw);out.write(raw)
                else:raise ValueError('Input media requires a staged URL')
            if not size or (expected is not None and size!=expected):raise ValueError('Input media size does not match its manifest')
            if expected_hash is not None and digest.hexdigest()!=expected_hash:raise ValueError('Input media SHA256 does not match its manifest')
            references[name]=target.relative_to(INPUT_ROOT).as_posix()
        return references,directory
    except BaseException:
        shutil.rmtree(directory,ignore_errors=True)
        raise

def normalize_and_validate(workflow,object_info,references,job):
    if not isinstance(workflow,dict) or not workflow or len(workflow)>512:raise ValueError('workflow must be a non-empty API graph with at most512 nodes')
    workflow=json.loads(json.dumps(workflow,allow_nan=False))
    for node_id,node in workflow.items():
        if not isinstance(node_id,str) or not re.fullmatch(r'[A-Za-z0-9_-]{1,80}',node_id):raise ValueError('Invalid workflow node ID')
        if not isinstance(node,dict) or not isinstance(node.get('inputs'),dict):raise ValueError('Invalid workflow node: '+node_id)
        node_type=node.get('class_type')
        if not isinstance(node_type,str) or node_type not in object_info:raise ValueError('Node type is not installed: '+str(node_type))
        normalized=re.sub('[^a-z0-9]','',node_type.lower())
        if any(marker in normalized for marker in DENIED_MARKERS):raise ValueError('This node cannot run in the shared runtime: '+node_type)
        definitions=object_info[node_type].get('input',{})
        definitions={**definitions.get('required',{}),**definitions.get('optional',{})}
        if object_info[node_type].get('output_node') and 'filename_prefix' in definitions and 'filename_prefix' not in node['inputs']:
            node['inputs']['filename_prefix']='output'
        for key,value in list(node['inputs'].items()):
            lower=str(key).lower()
            if isinstance(value,str) and len(value)>20000:raise ValueError('Workflow text is too long')
            if isinstance(value,list) and len(value)==2 and isinstance(value[0],str) and isinstance(value[1],int):
                if value[0] not in workflow or not 0<=value[1]<=128:raise ValueError('Broken workflow link in node '+node_id)
            if isinstance(value,(int,float)) and not isinstance(value,bool):
                if not math.isfinite(value):raise ValueError('Workflow numbers must be finite')
                limits={'width':8192,'height':8192,'target_width':8192,'target_height':8192,'output_width':8192,'output_height':8192,
                        'length':2000,'num_frames':2000,'frames':2000,'video_frames':2000,'frame_count':2000,
                        'fps':240,'frame_rate':240,'steps':120,'batch':8,'batch_size':8,'batch_count':8,'duration':300,'duration_seconds':300}
                if lower in limits and not 0<=value<=limits[lower]:raise ValueError('Workflow exceeds '+lower+' limit')
            if (lower in MODEL_KEYS or re.fullmatch(r'lora_\d+',lower)) and isinstance(value,str) and value.lower() not in ('none','disabled'):
                if not relative_name(value):raise ValueError('Unsafe model path in '+node_id)
                definition=definitions.get(key)
                options=definition[0] if isinstance(definition,list) and definition and isinstance(definition[0],list) else None
                if options and value not in options:
                    basename=Path(value).name
                    matches=[option for option in options if isinstance(option,str) and Path(option).name==basename]
                    if len(matches)==1:node['inputs'][key]=matches[0]
                    else:raise ValueError('Model is missing from the native HF cache mapping: '+value)
            if lower in ('filename_prefix','subfolder') and isinstance(value,str):
                if value and not relative_name(value):raise ValueError('Unsafe output path')
                node['inputs'][key]='hf-jobs/'+job+'/'+(value or 'output')
            media_key=re.sub('[^a-z0-9]','',lower)
            if any(marker in normalized for marker in MEDIA_MARKERS) and media_key in MEDIA_KEYS and isinstance(value,str):
                source=reference_name(value)
                if source not in references:raise ValueError('Workflow input media was not supplied by this job: '+str(source))
                mapped=references[source]
                if node_type=='VHS_LoadVideoPath' and media_key=='video':
                    # Path loaders use the path directly, unlike input-folder loaders.
                    target=(INPUT_ROOT/mapped).resolve()
                    job_root=(INPUT_ROOT/'hf-jobs'/job).resolve()
                    if not target.is_relative_to(job_root) or not target.is_file():
                        raise ValueError('Video path must reference an existing input from this job')
                    mapped=str(target)
                node['inputs'][key]=mapped
    return workflow

def snapshot_outputs():
    state={}
    for root in (OUTPUT_ROOT,TEMP_ROOT):
        if not root.exists():continue
        for path in root.rglob('*'):
            try:
                resolved=path.resolve()
                if path.is_file() and resolved.is_relative_to(root.resolve()):
                    stat=path.stat();state[resolved]=(stat.st_size,stat.st_mtime_ns)
            except OSError:pass
    return state

def output_files(history,before):
    paths=[];owned=[];seen=set()
    def visit(value,depth=0):
        if depth>8:return
        if isinstance(value,list):
            for child in value:visit(child,depth+1)
        elif isinstance(value,dict):
            if isinstance(value.get('filename'),str):
                name=value['filename'];subfolder=value.get('subfolder','');kind=value.get('type','output')
                roots={'output':OUTPUT_ROOT,'temp':TEMP_ROOT}
                if kind not in roots or not relative_name(name) or (subfolder and not relative_name(subfolder)):return
                root=roots[kind].resolve();path=(root/subfolder/name).resolve()
                if not path.is_relative_to(root) or not path.is_file() or path in seen:return
                seen.add(path);paths.append(path)
                stat=path.stat()
                if before.get(path)!=(stat.st_size,stat.st_mtime_ns):owned.append(path)
            else:
                for child in value.values():visit(child,depth+1)
    visit(history.get('outputs',{}))
    if not paths:raise RuntimeError('ComfyUI completed without an image, video or audio artifact')
    if len(paths)>MAX_FILES:raise ValueError('Workflow produced more than16 output artifacts')
    return paths,owned

def comfy_execution_ms(history):
    """Use ComfyUI execution lifecycle timestamps; exclude input/output transfer."""
    events={}
    for item in (history.get('status') or {}).get('messages',[]):
        if not isinstance(item,(list,tuple)) or len(item)!=2:continue
        name,data=item
        if name not in ('execution_start','execution_success') or not isinstance(data,dict):continue
        stamp=data.get('timestamp')
        if isinstance(stamp,(int,float)) and not isinstance(stamp,bool) and math.isfinite(stamp):events[name]=(data.get('prompt_id'),stamp)
    start=events.get('execution_start');end=events.get('execution_success')
    if not start or not end or not start[0] or start[0]!=end[0] or end[1]<start[1]:return None
    return math.ceil(end[1]-start[1])


def run_comfy(workflow,context,deadline):
    identifier=None
    with context.lock:
        execution_started=time.monotonic()
        try:
            result=context.comfy('/prompt',{'prompt':workflow,'client_id':'hf-'+uuid.uuid4().hex})
            identifier=result.get('prompt_id')
            if not isinstance(identifier,str):raise RuntimeError('ComfyUI rejected the workflow: '+json.dumps(result)[:1800])
            while True:
                deadline.remaining(10)
                history=context.comfy('/history/'+identifier).get(identifier)
                if isinstance(history,dict):
                    status=history.get('status') or {}
                    if status.get('status_str')=='error':raise RuntimeError('ComfyUI execution failed: '+json.dumps(status.get('messages',[]))[:1800])
                    if status.get('completed') or status.get('status_str')=='success':
                        # Fallback measured inside the lock, before cleanup/transfer.
                        return {**history,'_execution_ms':elapsed_ms(execution_started)}
                time.sleep(min(.25,deadline.remaining()))
        except (TimeoutError,KeyboardInterrupt):
            try:context.comfy('/interrupt',{})
            except Exception:pass
            raise
        except urllib.error.HTTPError as error:
            try:message=error.read(16384).decode(errors='replace')
            except Exception:message=''
            raise RuntimeError('ComfyUI rejected the workflow (HTTP '+str(error.code)+'): '+message[:1800]) from None
        finally:
            if identifier:
                try:context.comfy('/history',{'delete':[identifier]})
                except Exception:pass

def file_metadata(path,deadline):
    size=path.stat().st_size
    if not 0<size<=MAX_ARTIFACT:raise ValueError('Output artifact must be1byte..512MiB')
    digest=hashlib.sha256()
    with path.open('rb') as source:
        while True:
            deadline.remaining();chunk=source.read(CHUNK_BYTES)
            if not chunk:break
            digest.update(chunk)
    return {'filename':path.name,'mime_type':MIMES.get(path.suffix.lower()) or mimetypes.guess_type(path.name)[0] or 'application/octet-stream',
            'size':size,'sha256':digest.hexdigest()}

def upload_outputs(paths,capability,deadline):
    if len(paths)>MAX_FILES or sum(path.stat().st_size for path in paths)>MAX_TOTAL:raise ValueError('Output artifacts exceed the per-job limit')
    if capability:
        trusted_url(capability)
        parsed=urllib.parse.urlsplit(capability)
        if parsed.query or not re.fullmatch(r'/a/[A-Za-z0-9_-]{43}',parsed.path):raise ValueError('Invalid job-scoped artifact capability')
    elif sum(path.stat().st_size for path in paths)>7*1024*1024:
        raise ValueError('A Cloudflare artifact upload capability is required for output larger than7MiB')
    outputs=[]
    for path in paths:
        metadata=file_metadata(path,deadline)
        if not capability:
            outputs.append({**metadata,'data':base64.b64encode(path.read_bytes()).decode()});continue
        response=request_json(capability+'/begin','POST',metadata,deadline)
        identifier=response.get('id')
        if not isinstance(identifier,str) or not re.fullmatch(r'[A-Za-z0-9_-]{8,100}',identifier):raise RuntimeError('Invalid artifact upload ID')
        parts=[]
        with path.open('rb') as source:
            index=0
            while True:
                deadline.remaining();chunk=source.read(CHUNK_BYTES)
                if not chunk:break
                digest=hashlib.sha256(chunk).hexdigest()
                response=request_json(capability+'/'+identifier+'/parts/'+str(index),'PUT',chunk,deadline,{'X-Part-SHA256':digest})
                if response.get('sha256') and response['sha256']!=digest:raise RuntimeError('Artifact part checksum mismatch')
                parts.append({'index':index,'sha256':digest,'size':len(chunk)});index+=1
        completed=request_json(capability+'/'+identifier+'/complete','POST',{'parts':parts,'sha256':metadata['sha256'],'size':metadata['size']},deadline)
        url=completed.get('url');trusted_url(url)
        if completed.get('sha256') and completed['sha256']!=metadata['sha256']:raise RuntimeError('Uploaded artifact checksum mismatch')
        if completed.get('size') is not None and completed['size']!=metadata['size']:raise RuntimeError('Uploaded artifact size mismatch')
        outputs.append({**metadata,'url':url,'expires_at':completed.get('expires_at')})
    return outputs

def finite_number(body,key,default,minimum,maximum,integer=False):
    raw=body.get(key,default)
    if isinstance(raw,bool):raise ValueError(key+' must be numeric')
    try:value=float(raw)
    except (TypeError,ValueError):raise ValueError(key+' must be numeric') from None
    if not math.isfinite(value) or not minimum<=value<=maximum or (integer and value!=int(value)):
        raise ValueError(key+' must be between'+str(minimum)+' and'+str(maximum))
    return int(value) if integer else value

def run_irodori(payload,context,references,job,deadline):
    config=json.loads((context.root/'app-runtime.json').read_text(encoding='utf-8-sig')).get('irodori')
    if not isinstance(config,dict):raise ValueError('HF app is missing its irodori model configuration')
    text=payload.get('text');caption=payload.get('caption','')
    if not isinstance(text,str) or not text.strip() or len(text)>1000:raise ValueError('Irodori text must contain1..1000 characters')
    if not isinstance(caption,str) or len(caption)>2000:raise ValueError('Irodori caption exceeds2000 characters')
    checkpoint=safe_cache_path(context.root,config.get('checkpoint'))
    codec=safe_cache_path(context.root,config.get('codec'))
    tokenizer=safe_cache_path(context.root,config.get('tokenizer','tokenizer'),directory=True)
    if not (tokenizer/'tokenizer_config.json').is_file():raise ValueError('Irodori tokenizer_config.json is missing')
    directory=OUTPUT_ROOT/'hf-jobs'/job;directory.mkdir(parents=True,exist_ok=True)
    output=directory/'irodori.wav'
    reference=payload.get('reference_audio')
    many=payload.get('reference_audios')
    if reference is not None and many is not None:raise ValueError('Use one reference representation')
    selected=many if many is not None else ([reference] if reference else [])
    if not isinstance(selected,list) or len(selected)>20 or any(name not in references for name in selected):raise ValueError('Irodori reference audio was not supplied by this job')
    if sum((INPUT_ROOT/references[name]).stat().st_size for name in selected)>20*1024*1024:raise ValueError('Irodori reference audio exceeds20MiB combined')
    request={'checkpoint':str(checkpoint),'codec':str(codec),'tokenizer':str(tokenizer),'output':str(output),
        'text':text.strip(),'caption':caption.strip(),'reference_wav':str(INPUT_ROOT/references[reference]) if reference else None,
        'reference_wavs':[str(INPUT_ROOT/references[name]) for name in many] if many is not None else None,
        'num_steps':finite_number(payload,'num_steps',60,4,80,True),'duration_scale':finite_number(payload,'duration_scale',1,.5,2),
        'cfg_scale_text':finite_number(payload,'cfg_scale_text',3,0,20),'cfg_scale_caption':finite_number(payload,'cfg_scale_caption',3,0,20),
        'cfg_scale_speaker':finite_number(payload,'cfg_scale_speaker',10,0,20),
        'seed':None if payload.get('seed') is None else finite_number(payload,'seed',0,0,2**53-1,True),
        'seconds':None if payload.get('seconds') is None else finite_number(payload,'seconds',5,.5,45)}
    runner=context.root/'irodori_runner.py'
    if not runner.is_file():raise ValueError('irodori_runner.py is missing from the HF app')
    log=directory/'irodori.stderr'
    with log.open('wb') as error_log:
        try:
            result=subprocess.run(['/opt/irodori-venv/bin/python',str(runner)],input=json.dumps(request),text=True,
                stdout=subprocess.PIPE,stderr=error_log,timeout=deadline.remaining(),check=False,
                env={**os.environ,'HF_HUB_OFFLINE':'1','TRANSFORMERS_OFFLINE':'1'})
        except subprocess.TimeoutExpired:raise TimeoutError('Irodori exceeded the remaining280-second app budget') from None
    if result.returncode:
        with log.open('rb') as error_log:error_log.seek(max(0,log.stat().st_size-1600));detail=error_log.read().decode(errors='replace')
        raise RuntimeError('Irodori generation failed: '+detail)
    try:summary=json.loads(result.stdout)
    except ValueError:raise RuntimeError('Irodori returned invalid metadata') from None
    if summary.get('error'):raise RuntimeError(str(summary['error'])[:1600])
    if not output.is_file():raise RuntimeError('Irodori returned no WAV file')
    return [output],summary

def cleanup_job(directory,roots):
    if directory is None:return
    resolved=directory.resolve()
    if any(resolved.is_relative_to((root/'hf-jobs').resolve()) and resolved!=(root/'hf-jobs').resolve() for root in roots):
        shutil.rmtree(directory,ignore_errors=True)

def elapsed_ms(started):
    return max(0,math.ceil((time.monotonic()-started)*1000))

def generate(payload,context):
    if not isinstance(payload,dict):raise ValueError('Generation payload must be an object')
    deadline=Deadline(280);started=time.monotonic();job=uuid.uuid4().hex
    metadata=payload.get('_app') or {}
    if not isinstance(metadata,dict):raise ValueError('Invalid artifact metadata')
    capability=metadata.get('artifact_upload_url')
    if capability:trusted_url(capability)
    references={};input_directory=None;owned=[];additional=[]
    try:
        references,input_directory=prepare_files(payload.get('files'),job,deadline)
        if payload.get('task')=='diagnostics':
            execution_started=time.monotonic()
            result=diagnostics(payload,context)
            result['summary']={'backend':'diagnostics','executed_ms':elapsed_ms(execution_started)}
            return result
        if payload.get('task')=='chat':
            execution_started=time.monotonic()
            result=run_chat(payload,context,deadline)
            result['summary']['executed_ms']=elapsed_ms(execution_started)
            result['summary']['elapsed_seconds']=round(time.monotonic()-started,3)
            return result
        if payload.get('task')=='irodori':
            execution_started=time.monotonic()
            paths,summary=run_irodori(payload,context,references,job,deadline);owned=paths
            summary['executed_ms']=elapsed_ms(execution_started)
        else:
            additional_ms=0
            if payload.get('task')=='dialogue':
                settings=payload.get('irodori')
                if not isinstance(settings,dict):raise ValueError('Dialogue requires Irodori synthesis settings')
                execution_started=time.monotonic()
                audio,tts_summary=run_irodori(settings,context,references,job,deadline)
                additional_ms=elapsed_ms(execution_started)
                tts_summary['executed_ms']=additional_ms
                reference=input_directory/'dialogue-generated.wav'
                shutil.copyfile(audio[0],reference)
                references['dialogue-generated.wav']=reference.relative_to(INPUT_ROOT).as_posix()
                additional.extend(audio);owned.extend(audio)
            continuation=prepare_continuation(payload.get('continuation'),references,input_directory,deadline)
            workflow=payload.get('workflow')
            if workflow is None:
                default=context.root/'workflow.json'
                if not default.is_file():raise ValueError('Provide an API workflow or include workflow.json in the HF app')
                workflow=json.loads(default.read_text(encoding='utf-8-sig'))
            if payload.get('task')=='dialogue':workflow=dialogue_timing(workflow,tts_summary)
            info=context.comfy('/object_info')
            workflow=normalize_and_validate(workflow,info,references,job)
            before=snapshot_outputs();history=run_comfy(workflow,context,deadline)
            paths,new_outputs=output_files(history,before);owned.extend(new_outputs)
            paths,merged=finish_continuation(paths,continuation,job,deadline);owned.extend(merged)
            executed_ms=comfy_execution_ms(history)
            if executed_ms is None:executed_ms=history['_execution_ms']
            paths.extend(additional);summary={'backend':'comfyui','nodes':len(workflow),'executed_ms':executed_ms+additional_ms}
            if additional:summary['irodori']=tts_summary
            if continuation:summary['continuation']=True
        outputs=upload_outputs(paths,capability,deadline)
        return {'files':outputs,'summary':{**summary,'elapsed_seconds':round(time.monotonic()-started,3)}}
    finally:
        cleanup_job(input_directory,(INPUT_ROOT,))
        for path in owned:
            try:
                resolved=path.resolve()
                if any(resolved.is_relative_to(root.resolve()) for root in (OUTPUT_ROOT,TEMP_ROOT)):path.unlink(missing_ok=True)
            except OSError:pass
        for root in (OUTPUT_ROOT,TEMP_ROOT):cleanup_job(root/'hf-jobs'/job,(root,))

def run_chat(payload,context,deadline):
    config=json.loads((context.root/'app-runtime.json').read_text(encoding='utf-8-sig')).get('chat') or {}
    model_path=context.root if config.get('model_path','.')=='.' else safe_cache_path(context.root,config['model_path'],directory=True)
    if not (model_path/'config.json').is_file():raise ValueError('Chat model config.json is missing from the native HF cache')
    messages=payload.get('messages')
    if messages is None:
        messages=[]
        if payload.get('system_prompt'):messages.append({'role':'system','content':payload['system_prompt']})
        if isinstance(payload.get('history'),list):messages.extend(payload['history'])
        messages.append({'role':'user','content':payload.get('message','')})
    if not isinstance(messages,list) or not 1<=len(messages)<=64:raise ValueError('Chat requires1..64 messages')
    for message in messages:
        if not isinstance(message,dict) or message.get('role') not in ('system','user','assistant') or not isinstance(message.get('content'),str) or not message['content'].strip() or len(message['content'])>20000:
            raise ValueError('Chat messages require role and nonempty text content')
    request={'model_path':str(model_path),'messages':messages,'max_tokens':finite_number(payload,'max_tokens',512,1,2048,True),
             'temperature':finite_number(payload,'temperature',.78,0,2),'top_p':finite_number(payload,'top_p',.9,.01,1),
             'repetition_penalty':finite_number(payload,'repetition_penalty',1.08,1,2),
             'seed':None if payload.get('seed') is None else finite_number(payload,'seed',0,0,2**53-1,True),
             'generation_seconds':deadline.remaining()}
    directory=Path('/workspace/hf-app');directory.mkdir(parents=True,exist_ok=True)
    log=directory/('chat-'+uuid.uuid4().hex+'.stderr')
    try:
        with log.open('wb') as error_log:
            try:result=subprocess.run(['/opt/venv/bin/python',str(context.root/'chat_runner.py')],input=json.dumps(request),text=True,
                stdout=subprocess.PIPE,stderr=error_log,timeout=deadline.remaining(),check=False,
                env={**os.environ,'HF_HUB_OFFLINE':'1','TRANSFORMERS_OFFLINE':'1'})
            except subprocess.TimeoutExpired:raise TimeoutError('Chat exceeded the remaining280-second app budget') from None
        try:output=json.loads(result.stdout)
        except ValueError:raise RuntimeError('Chat returned invalid JSON') from None
        if result.returncode or output.get('error'):raise RuntimeError('Chat failed: '+str(output.get('error','runtime exited'))[:1600])
        return output
    finally:log.unlink(missing_ok=True)

def prepare_continuation(value,references,directory,deadline):
    import bernini_continuation as module
    config=module.validate_continuation(value)
    if config is None:return None
    name=config['source_video_name']
    if name not in references:raise ValueError('Continuation source video was not supplied')
    source=INPUT_ROOT/references[name]
    frame=directory/(uuid.uuid4().hex+'.png')
    resize=f"scale={config['width']}:{config['height']}:force_original_aspect_ratio=decrease,pad={config['width']}:{config['height']}:(ow-iw)/2:(oh-ih)/2:color=black,setsar=1"
    module._DEADLINE=deadline
    errors=[]
    for offset in ('-0.1','-0.5','-1.0'):
        frame.unlink(missing_ok=True)
        try:module._run(['ffmpeg','-hide_banner','-loglevel','error','-y','-sseof',offset,'-i',str(source),'-map','0:v:0','-an','-frames:v','1','-vf',resize,str(frame)],'continuation frame extraction',30)
        except (RuntimeError,subprocess.TimeoutExpired) as error:errors.append(str(error))
        if frame.is_file() and frame.stat().st_size:break
    else:
        frame.unlink(missing_ok=True)
        module._run(['ffmpeg','-hide_banner','-loglevel','error','-y','-i',str(source),'-map','0:v:0','-an','-vf',resize,'-fps_mode','passthrough','-update','1',str(frame)],'continuation full-decode frame extraction',deadline.remaining())
    if not frame.is_file() or not frame.stat().st_size:raise RuntimeError('Continuation produced no reference frame')
    references[config['last_frame_name']]=frame.relative_to(INPUT_ROOT).as_posix()
    return {**config,'source_path':source}

def finish_continuation(paths,config,job,deadline):
    if config is None:return paths,[]
    import bernini_continuation as module
    module._DEADLINE=deadline
    outputs=[];created=[]
    for path in paths:
        if path.suffix.lower() not in module.VIDEO_SUFFIXES:outputs.append(path);continue
        target=OUTPUT_ROOT/'hf-jobs'/job/('continued-'+str(len(created))+'.mp4')
        module.merge_continuation_output(config['source_path'],path,target,config)
        outputs.append(target);created.append(target)
    if not created:raise RuntimeError('Continuation workflow produced no video')
    return outputs,created


def diagnostics(payload,context):
    """Bounded, read-only inspection for version/mapping and workflow smoke checks."""
    network=trusted_network_test(payload['network_url'],Deadline(25)) if payload.get('network_url') else None
    if payload.get('network_only'):
        if network is None:raise ValueError('network_only requires a trusted network_url')
        return {'network':network}
    requested=payload.get('requested_class_types',[])
    if not isinstance(requested,list) or len(requested)>128 or any(not isinstance(name,str) or len(name)>160 for name in requested):
        raise ValueError('requested_class_types must contain at most128 node names')
    info=context.comfy('/object_info')
    result={'network':network,'available_node_count':len(info),'missing':[name for name in requested if name not in info], 'nodes':{}}
    for name in requested:
        if name not in info:continue
        node=info[name]
        result['nodes'][name]={'input':node.get('input',{}),'output':node.get('output',[]),'output_node':bool(node.get('output_node'))}
    stats=context.comfy('/system_stats')
    system=stats.get('system',{})
    result['versions']={key:system.get(key) for key in ('comfyui_version','python_version','pytorch_version','embedded_python') if key in system}
    workflow=payload.get('workflow')
    if workflow is not None:
        references={}
        for item in payload.get('files',[]):
            if isinstance(item,dict) and isinstance(item.get('name'),str):references[item['name']]='hf-jobs/diagnostics/'+item['name']
        normalized=normalize_and_validate(workflow,info,references,'diagnostics')
        result['workflow']={'valid':True,'nodes':len(normalized),'model_inputs':{node_id:{key:value for key,value in node['inputs'].items() if key.lower() in MODEL_KEYS or re.fullmatch(r'lora_\d+',key.lower())} for node_id,node in normalized.items()}}
    if len(json.dumps(result,separators=(',',':')).encode())>512*1024:raise ValueError('Requested diagnostic schema exceeds512KiB; request fewer nodes')
    return result


def dialogue_timing(workflow,summary):
    """Preserve the host dialogue graph's speech-duration dependent frame count."""
    if not isinstance(workflow,dict):raise ValueError('Dialogue workflow must be an object')
    workflow=json.loads(json.dumps(workflow,allow_nan=False))
    expected={'796':'mxSlider','535':'LTXVEmptyLatentAudio','542':'PrimitiveFloat','549':'VHS_VideoCombine'}
    if any(workflow.get(key,{}).get('class_type')!=kind for key,kind in expected.items()):
        raise ValueError('Dialogue workflow is missing its duration/frame-rate controls')
    fps=finite_number(workflow['542']['inputs'],'value',24,1,120)
    duration=finite_number(summary,'duration_seconds',0,.001,45)
    seconds=max(3,math.ceil(duration+.25));frames=int(seconds*fps)
    workflow['796']['inputs'].update(Xi=frames,Xf=frames)
    workflow['535']['inputs'].update(frames_number=frames+1,frame_rate=fps)
    workflow['542']['inputs']['value']=fps
    workflow['549']['inputs']['frame_rate']=fps
    return workflow