app.py

"""Flowframes on Spaces: RIFE frame interpolation that runs on CPU or GPU.

One code path, one model file. The execution provider is chosen at startup
(TensorRT, else CUDA, else CPU) and whatever was actually selected is printed in
the header, because a missing CUDA library drops ORT to CPU silently and a Space
running 50x slower looks identical to one that is not.

Work is cut into segments sized from a measured per-frame rate. On ZeroGPU each
segment is its own GPU call, so the 120 s cap bounds a segment rather than the
job and clip length stops being a limit. On CPU the same seams are checkpoints,
so cancelling keeps the minutes already spent.
"""

import os
import pathlib
import re
import shutil
import sys
import tempfile
import threading
import time
import uuid

# Gradio copies every file it serves into GRADIO_TEMP_DIR and never prunes it,
# so each delivered output existed twice and only one copy was ever cleaned.
# Point that cache somewhere sweepable. Must happen before gradio is imported.
SCRATCH = pathlib.Path(tempfile.gettempdir()) / "flowframes"
OUTDIR = SCRATCH / "out"
CACHEDIR = SCRATCH / "gradio"
for _d in (OUTDIR, CACHEDIR):
    _d.mkdir(parents=True, exist_ok=True)
os.environ.setdefault("GRADIO_TEMP_DIR", str(CACHEDIR))
CACHEDIR = pathlib.Path(os.environ["GRADIO_TEMP_DIR"])   # setdefault may not have won

import gradio as gr

import rife_core as R

try:
    import spaces
    GPU = spaces.GPU
except Exception:                       # noqa: BLE001
    def GPU(*_a, **_k):
        return lambda f: f

# The import above is NOT the test. `spaces` is in requirements.txt and installs
# anywhere; off ZeroGPU its decorator simply no-ops, so `import spaces` succeeded
# on cpu-basic too and every ZeroGPU branch was taken there: BUDGET_S 90 s
# instead of 900, segments ten times smaller than intended, and a header
# announcing a GPU. Only the environment knows. Read from the live container:
#   SPACES_ZERO_GPU='1'  ZEROGPU_V2='true'  SPACES_ZERO_DEVICE_API_URL=...
ZEROGPU = bool(os.environ.get("SPACES_ZERO_GPU"))

HERE = pathlib.Path(__file__).parent
LOG_LINES = 14
CANCELLED: set[str] = set()      # session hashes that asked to stop
LOCK = threading.Lock()
OWNER: str | None = None         # session hash of the job currently holding LOCK

# Built on CPU at import: on ZeroGPU there is no CUDA device until a @spaces.GPU
# call is running, so a session created here could never see one.
BACKEND = R.OnnxBackend(prefer_gpu=False)
_GPU_BE = None
# (height, width, device) -> (seconds per frame, backend name, build seconds).
# Survives between jobs because it lives in the parent, not in a @spaces.GPU fork.
_RATE_CACHE: dict[tuple[int, int, str], tuple[float, str, float]] = {}
# Set when a GPU call fails on quota, cleared when one next succeeds. Module
# level because the header is rendered per page load by visitors who are not
# running a job, and a Space whose daily quota is gone was still announcing
# "GPU (ZeroGPU), about 39s" while every job crawled along on CPU.
GPU_DEAD = False
# Whether NVENC actually encodes in a GPU fork. False until the first fork
# reports back, because the parent process has no GPU and cannot find out.
NVENC_OK = False


def worker() -> R.Backend:
    """The backend to actually infer with. Inside a GPU call this upgrades to
    CUDA or TensorRT once, then stays."""
    global _GPU_BE
    if not ZEROGPU:
        return BACKEND
    if _GPU_BE is None:
        t0 = time.time()
        _GPU_BE = R.OnnxBackend(prefer_gpu=True, trt_cache=str(HERE / ".trt"))
        print(f"[ff] BUILT gpu worker {_GPU_BE.name} in {time.time()-t0:.0f}s "
              f"pid={os.getpid()}", flush=True)
    else:
        print(f"[ff] REUSED gpu worker pid={os.getpid()}", flush=True)
    return _GPU_BE
# Reference rate at 720p so the header can quote a real number rather than a table.
REF_RATE = R.measure_rate(BACKEND, 720, 1280)
# One GPU call must finish inside its ZeroGPU budget; CPU has no such cap but
# still benefits from the checkpoint seams.
#
# 60, not 90, and not the 75 I first reasoned my way to. plan_gaps charges only
# the inference term, so a segment costs more than it plans for. Measured on this
# Space against a hard 120 s @spaces.GPU window:
#
#   852x480 f2   1.416x, 1.456x        360x640 f2   1.393x, 1.496x
#   852x480 f4   1.566x   <-- worst
#
# I predicted f4 would be BELOW f2, on the theory that a fixed per-gap overhead
# is diluted by three inferences instead of one. It came in highest. The overhead
# is 0.0235 s/gap at f2 and 0.0957 at f4 - four times larger, not fixed - and per
# generated frame it is superlinear too, so there is no model here, only a worst
# case. At the worst observed 1.566x: 90 plans 141 s (killed), 75 plans 117 s
# (2% of headroom, which is not headroom), 60 plans 94 s.
#
# The cost of being conservative is refunded by the re-cut below, which grows the
# plan back when a segment proves cheaper than budgeted. CPU stays 900: no window
# to overrun, and bigger segments mean fewer ffmpeg restarts.
BUDGET_S = 60.0 if ZEROGPU else 900.0
print(f"[ff] provider={BACKEND.provider} rate720p={REF_RATE:.3f}s/frame "
      f"budget={BUDGET_S:.0f}s zerogpu={ZEROGPU}", flush=True)
# Kept: if HF ever renames the marker, this line is how it gets noticed.
print("[ff] env " + " ".join(sorted(k for k in os.environ if "ZERO" in k.upper())),
      flush=True)
# On GPU the plan-check ratio says 40-66% of a segment is decode, emit and x264
# on two vCPUs rather than the model, so whether ffmpeg can offload any of that
# is the largest open speed question. It depends on the image's ffmpeg build,
# which nothing else here can see.
R.report_ffmpeg_hw()
# Whether the thread count came from the cgroup or from a failed read matters:
# one is the container's real allowance, the other is the host's count and means
# ORT is oversubscribing the two vCPUs a Space is supposed to have.
print(f"[rife] threads={R.CPUS} from: {R.cgroup_report()}", flush=True)
# One-shot: is the 12-thread cap right for THIS container? The sweep behind it
# measured 12 against 32 on a dev box, and the ZeroGPU container allows 16, which
# was never tested. Both measured in the same boot so the comparison is not
# across two containers with different noise. Costs a few seconds once, against a
# 1h startup budget, and the CPU path is what users hit when quota runs out.
if ZEROGPU and R.container_cpus() > R.CPUS:
    try:
        _alt = R.OnnxBackend(prefer_gpu=False, threads=R.container_cpus())
        _r_alt = R.measure_rate(_alt, 720, 1280)
        del _alt
        print(f"[rife] thread sweep at 720p: {R.CPUS} threads {REF_RATE:.3f}s/frame, "
              f"{R.container_cpus()} threads {_r_alt:.3f}s/frame -> "
              f"{'raise the cap' if _r_alt < REF_RATE * 0.95 else 'cap is right'}",
              flush=True)
    except Exception as _e:                                  # noqa: BLE001
        print(f"[rife] thread sweep failed ({type(_e).__name__})", flush=True)


def hms(s: float) -> str:
    s = int(s)
    return f"{s//3600}:{(s%3600)//60:02d}:{s%60:02d}" if s >= 3600 else f"{s//60}:{s%60:02d}"


def plain(seconds: float) -> str:
    """Wall clock the way a person says it, not seconds-per-frame."""
    if seconds < 90:
        return f"{int(seconds)}s"
    if seconds < 5400:
        return f"{seconds/60:.0f} min"
    return f"{seconds/3600:.1f} h"


# Measured at 1280x720 on the ZeroGPU device (RTX PRO 6000 Blackwell MIG 2g.48gb),
# fp32 CUDA, by running a real job through this Space. Used only as the a-priori
# number in the header before any job has run here, because there is no GPU at
# import time to measure on. Quoted at 720p directly rather than scaled up from a
# smaller measurement, so the header carries a number and not an extrapolation.
GPU_REF_RATE_720P = 0.130


def measured() -> tuple[float, str, int, int] | None:
    """The rate this Space actually runs at, with the device that produced it.

    One snapshot and one preference order, so the name and the number in the
    header always describe the same device. Two independent lookups over this
    dict used to disagree: the estimate preferred a GPU entry while the name took
    whatever was inserted first, which on a Space that ran on CPU before the
    quota reset printed a CPU backend beside a GPU rate.

    The snapshot also matters: demo.load runs outside the queue on every page
    load, so a job in another session can insert here mid-iteration.
    """
    snap = list(_RATE_CACHE.items())
    devices = ("cpu",) if GPU_DEAD else ("gpu", "cpu")
    # 1280x720 first, because "10s 720p clip" is the phrase a person can act on
    # and the default Max height produces exactly that key on any 16:9 source,
    # which is most video. Failing that, quote whatever WAS measured and name its
    # size: a number that is true about an odd resolution beats a familiar one
    # restated from a measurement taken somewhere else.
    for exact in (True, False):
        for want in devices:
            for (h, w, dev), (rate, name, _b) in snap:
                if dev == want and (not exact or (h, w) == (720, 1280)):
                    return rate, name, h, w
    return None


def header() -> str:
    """A 10 s 30fps clip at 2x is 300 generated frames.

    The measured rate is quoted at the resolution it was measured at, never
    restated at 720p. Scaling by pixel count looked exact from 640x360 (0.032
    measured, 0.033 predicted) and was 15% out from 852x480 (0.067 measured
    predicts 45 s; the directly measured 720p answer is 39 s), because a small
    frame leaves the GPU idle. Naming the resolution costs nothing and means the
    number is always something that was actually observed.
    """
    hit = measured()
    if hit:
        rate, name, h, w = hit
        size = "720p" if (h, w) == (720, 1280) else f"{w}x{h}"
        cost = f"about {plain(rate * 300)} for a 10s {size} clip at 2x"
    elif ZEROGPU:
        name = "GPU (ZeroGPU)"
        cost = f"about {plain(GPU_REF_RATE_720P * 300)} for a 10s 720p clip at 2x"
    else:
        name = BACKEND.name
        cost = f"about {plain(REF_RATE * 300)} for a 10s 720p clip at 2x"
    return (f"[flowframes](https://github.com/n00mkrad/flowframes): "
            f"RIFE 4.9 frame interpolation · **{name}** · **{cost}** · "
            f"streams two frames at a time, so length costs time and not memory")


class Log:
    """Ring buffer with elapsed-time prefixes.

    Every yield re-sends the whole joined string over SSE, so an unbounded log
    grows each frame. Elapsed rather than wall clock because on a long job the
    delta between two lines is the only ETA instrument the viewer has.
    """

    def __init__(self, cap: int = 400):
        self.t0 = time.time()
        self.lines: list[str] = []
        self.cap = cap

    def __call__(self, msg: str) -> str:
        self.lines.append(f"[+{hms(time.time() - self.t0)}] {msg}")
        if len(self.lines) > self.cap:
            del self.lines[: self.cap // 2]
        return "\n".join(self.lines[-LOG_LINES:])

    @property
    def text(self) -> str:
        return "\n".join(self.lines[-LOG_LINES:])


def stop(request: gr.Request):
    """Cancel only the caller's own job.

    This handler runs outside the queue, so a second visitor waiting behind the
    first used to be able to kill the first visitor's run by pressing Stop, and
    anyone could POST the endpoint directly. Cancellation is keyed on the session
    that owns the running job.
    """
    me = getattr(request, "session_hash", None) or f"anon-{uuid.uuid4()}"
    if OWNER is not None and me != OWNER:
        return "That job belongs to another visitor. Yours has not started yet."
    CANCELLED.add(me)
    return "Stopping at the next frame boundary. The finished part is still saved."


def _segment_duration(src, dst, start, gaps, factor, clip, crf, write_tail,
                      me, tw, th) -> int:
    """Ask for what this particular segment needs, not for the ceiling.

    Takes the same arguments as the call it sizes, which is how a `duration`
    callable is invoked. Reuses the planner's own arithmetic and the measured
    `_OVERHEAD`, so a segment cut for a 60s budget asks for roughly 60s rather
    than the 120s cap, and a one-gap segment asks for seconds.

    Under-asking is not free: ZeroGPU kills a call that outlives its declaration.
    That lands in the existing `is_timeout_error` path, which halves the segment
    and retries up to four times, so it degrades into a slower job rather than a
    failed one. Hence the 1.4x on top of an overhead figure that already carries
    the measured 1.40-1.66 correction.
    """
    # A measurement of THIS shape beats the pooled mean. _PER_GAP holds the wall
    # cost per gap of the last full segment at this (h, w, factor), which is the
    # same quantity the re-cut sizes `gaps` from - so the declaration and the
    # plan agree by construction instead of drifting apart with the mean.
    seen = _PER_GAP.get((th, tw, int(factor), "gpu"))
    if seen:
        # 1.35, not 1.4: it gives every segment the same headroom the re-cut
        # already assumes is enough. REAL_BUDGET_S is 0.75 of the cap, so a 25%
        # swing - the deadband's own definition of ordinary - still lands under
        # 119. A short tail segment then gets the same margin as a full one,
        # instead of whatever the pooled mean happened to imply for it.
        return max(15, min(GPU_MAX_S, int(gaps * seen * 1.35) + 5))
    hit = _RATE_CACHE.get((th, tw, "gpu"))
    if not hit:
        return GPU_MAX_S                 # nothing measured yet, keep the ceiling
    est = gaps * hit[0] * max(1, int(factor) - 1) * _OVERHEAD["gpu"]
    return max(15, min(GPU_MAX_S, int(est * 1.4) + 5))


@GPU(duration=_segment_duration)
def _segment_gpu(src, dst, start, gaps, factor, clip, crf, write_tail, me, tw, th):
    """One segment, start to finish, inside a single GPU allocation.

    Returns (frames, seconds_inside). The caller times this call from the parent,
    which means its clock also covers the ZeroGPU allocation queue - 45s on a
    session's first call. That number is not the cost of the work, and using it
    as one shrank the next segment ~2x and permanently skewed the overhead mean.
    """
    t0 = time.perf_counter()
    n = R.interpolate_segment(src, dst, start, gaps, factor, worker(), clip,
                              crf=crf, cancel=lambda: me in CANCELLED,
                              write_tail=write_tail, out_w=tw, out_h=th,
                              hw_encode=NVENC_OK)
    return n, time.perf_counter() - t0


def _segment_cpu(src, dst, start, gaps, factor, clip, crf, write_tail, me, tw, th):
    """The same work without a GPU allocation, for when ZeroGPU quota is gone.

    Same (frames, seconds_inside) shape as the GPU worker so the caller does not
    branch on which one ran.
    """
    t0 = time.perf_counter()
    n = R.interpolate_segment(src, dst, start, gaps, factor, BACKEND, clip,
                              crf=crf, cancel=lambda: me in CANCELLED,
                              write_tail=write_tail, out_w=tw, out_h=th)
    return n, time.perf_counter() - t0


# Two different failures used to share one bucket. "GPU task aborted" is what a
# segment that overran its 120 s window looks like, and calling that quota
# exhaustion moved the whole rest of the job to CPU, told the user the day's
# budget was gone, and never mentioned the real fault. A timeout means the
# segment was too big; only the quota messages mean there is no GPU left today.
QUOTA_HINTS = ("exceeded your zerogpu", "zerogpu runs limit", "proxy token",
               "no gpu available", "gpu quota")
TIMEOUT_HINTS = ("gpu task aborted", "illegal duration", "exceeded the maximum",
                 "timeout")


def is_quota_error(e: BaseException) -> bool:
    return any(h in str(e).lower() for h in QUOTA_HINTS)


def is_timeout_error(e: BaseException) -> bool:
    return any(h in str(e).lower() for h in TIMEOUT_HINTS)


def cpu_rate(h: int, w: int) -> tuple[float, str, float]:
    """Same caching for the CPU path. Measuring costs a few seconds of the two
    vCPUs a job is about to need."""
    key = (h, w, "cpu")
    if key not in _RATE_CACHE:
        _RATE_CACHE[key] = (R.measure_rate(BACKEND, h, w), BACKEND.name, 0.0)
    return _RATE_CACHE[key]


def gpu_rate(h: int, w: int) -> tuple[float, str, float]:
    """Cached in the parent process, which survives between jobs even though each
    @spaces.GPU call runs in a fork. ZeroGPU meters runs rather than seconds, so a
    90 s measurement costs the same as a 90 s segment; paying it on every job was
    spending a fifth of a short job's budget to re-learn a constant."""
    global NVENC_OK
    key = (h, w, "gpu")
    if key not in _RATE_CACHE:
        rate, name, build, nvenc = _gpu_rate(h, w)
        # The cache stays a 3-tuple. measured() unpacks it as (rate, name, _b),
        # and an element-count mismatch here has already cost one debugging
        # session, so the new field rides beside the cache rather than inside it.
        _RATE_CACHE[key] = (rate, name, build)
        NVENC_OK = nvenc
        print(f"[ff] nvenc usable in the GPU fork: {nvenc}", flush=True)
    return _RATE_CACHE[key]


# measure_rate times three samples. At keep-source 4K that is 3 x ~1.1s, plus a
# session build the logs show landing in under a second once the CUDA libs are
# preloaded. 90s was reserving a minute and a half of quota to spend four
# seconds, on the one call whose queue wait the user actually sits through.
@GPU(duration=30)
def _gpu_rate(h: int, w: int) -> tuple[float, str, float]:
    """Measure the per-frame rate where the work will actually happen.

    Measuring outside a GPU call times the CPU session, and the segment planner
    would then size GPU segments from a CPU number, making them far too small.
    """
    t0 = time.time()
    be = worker()                       # first call in this worker builds the session
    build = time.time() - t0
    rate = R.measure_rate(be, h, w)
    # Probed here because this fork HAS a GPU, and returned rather than cached,
    # because a global set inside a fork dies with it. Riding the rate call keeps
    # it free: a separate @spaces.GPU probe would burn a metered run on one bool.
    # Printed, not returned: this goes to the container log, and adding a fifth
    # element to the tuple is how the 3-vs-2 unpack bug happened once already.
    # FIXED 720p, not the job's resolution. This runs inside _gpu_rate, which
    # declares duration=30; profiling four extra inferences at a keep-source 4K
    # would add ~5-8s plus a second session build to a probe that already spends
    # ~4s measuring, and a probe that outlives its declared duration is KILLED -
    # which would drop the whole job to CPU. The op MIX is what decides between
    # CUDA graphs and IO binding, and that does not need the job's own size.
    print(f"[ff] profile 1280x720: {R.profile_ops(720, 1280)}", flush=True)
    return rate, be.name, build, R.nvenc_works()


def example_run(video, factor, max_h, crf):
    """Examples cache one value per output, so drain the generator and hand back
    only the video and the final log. Routed through run() so a cached example is
    the same code path the button takes."""
    class _Anon:
        session_hash = "example"
    vid_out, log_out = None, ""
    for v, lg, *_ in run(video, factor, max_h, crf, _Anon()):
        vid_out, log_out = v or vid_out, lg
    if vid_out is None:
        # Raising keeps Gradio from writing the failure into the cache. Returning
        # None here cached a null video permanently, so the example row stayed
        # broken for every later visitor with nothing to explain why.
        raise gr.Error(f"Example produced no video. {log_out.splitlines()[-1:]}")
    return vid_out, log_out


# 50 GB of non-persistent container disk, shared with the model, the upload and
# every segment. Four is room for several 4K jobs and nowhere near the ceiling.
DISK_BUDGET = 4 * 1024 ** 3
SWEEP_MIN_AGE = 3600          # never touch anything younger than an hour
SWEEP_KEEP = 3                # ...nor the newest few, whatever the budget says
ABANDONED_AGE = 6 * 3600      # work dirs left by a kill, not by the finally


VIDEO_NAME = re.compile(r"[\w.\-]+\.(?:mp4|webm|mkv|mov)")


def pinned_names() -> set[str]:
    """Filenames the lazily cached example row points at.

    Gradio postprocesses the example's output through save_file_to_cache, so the
    video is copied into GRADIO_TEMP_DIR and the cached_examples CSV stores that
    path. The row is written once and never rewritten, so its mtime never moves
    and it becomes the OLDEST file in the cache: precisely the first thing an
    oldest-first sweep takes. Losing it breaks the example permanently, for
    everyone, with nothing in the log to say why.

    Matching on the bare filename over-protects at worst, which is the right way
    to be wrong here.
    """
    names: set[str] = set()
    try:
        for csv in pathlib.Path(".gradio").rglob("*.csv"):
            names.update(VIDEO_NAME.findall(
                csv.read_text(encoding="utf-8", errors="replace")))
    except OSError:
        pass
    return names


def sweep_outputs(budget: int = DISK_BUDGET, protect: str | None = None) -> None:
    """Hold finished outputs and Gradio's copies of them under a byte budget.

    Counting files is the wrong unit: three 4K outputs are far more disk than
    thirty 480p ones. Nothing pruned either directory before, so disk-full was
    the expected end state of a busy day, and the first thing that fails when the
    disk is full is the mkdtemp at the top of the next job.

    Four things must survive regardless of the budget. `protect` is the source
    video of the job doing the sweeping, pinned by resolved path: GRADIO_TEMP_DIR
    is also Gradio's UPLOAD folder, and the age floor does NOT cover it. That
    claim used to sit here and was wrong - SWEEP_MIN_AGE is one hour while this
    code permits a six hour job, so an upload that sat for ninety minutes before
    the user pressed Interpolate could be unlinked two lines before
    `R.probe(video)` read it, and the failure surfaced as "ffmpeg cannot open
    this file as a video", blaming the user's file. The cached example is pinned
    by name. And the newest few are always kept, because a budget smaller than
    one job's output would otherwise delete its own input.
    """
    now = time.time()
    pinned = pinned_names()
    safe = None
    if protect:
        try:
            safe = os.path.realpath(protect)
        except OSError:
            safe = None
    files = []
    for root in (OUTDIR, CACHEDIR):
        for f in root.rglob("*"):
            try:
                if f.is_file():
                    st = f.stat()
                    files.append((st.st_mtime, st.st_size, f))
            except OSError:
                pass
    total = sum(sz for _, sz, _ in files)
    files.sort()                                  # oldest first
    keep = {f for _, _, f in files[-SWEEP_KEEP:]}
    for mt, sz, f in files:
        if total <= budget:
            break
        if f in keep or f.name in pinned or now - mt < SWEEP_MIN_AGE:
            continue
        if safe:
            try:
                if os.path.realpath(f) == safe:
                    continue
            except OSError:
                pass
        try:
            f.unlink()
            total -= sz
        except OSError:
            pass      # an open handle on Windows; still counted, so the next
                      # candidate is tried rather than the entry being forgotten

    # Work directories are removed by run()'s finally on every normal path, but
    # not after a hard kill: an OOM, a container restart, a ZeroGPU teardown.
    # Those hold a whole segment set, gigabytes for a 4K job, and were invisible
    # to the accounting above, so the budget could report itself satisfied while
    # the disk filled anyway.
    for d in SCRATCH.glob("ff_*"):
        try:
            if d.is_dir() and now - d.stat().st_mtime > ABANDONED_AGE:
                shutil.rmtree(d, ignore_errors=True)
        except OSError:
            pass


def preflight(video):
    """Reject before the buttons swap, never after."""
    if video is None:
        raise gr.Error("Upload a video first, or click the example row below.")
    return BUSY


# The heights the Max height control actually offers, descending. Advice about
# that control reads the same list, so a suggestion can never name a value the
# dropdown does not have - which it did: at 480p it proposed 240, and at 360p it
# claimed 480 was "the smallest step" when 480 IS the smallest step.
HEIGHTS = (1080, 720, 480)


def next_lower_height(th: int) -> int | None:
    """The largest offered height below the current one, or None at the floor."""
    return next((h for h in HEIGHTS if h < th), None)


# What the plan does NOT charge, seeded from this Space's own `plan-check` lines
# and then re-fitted from every full segment that runs. plan_gaps budgets only
# inference; a real segment also decodes, emits factor-1 frames per gap and runs
# x264. Seeds are the medians of what was actually logged:
#   gpu  1.400 1.417 1.456 1.566 1.663
#   cpu  1.050 1.187 1.192 1.204 1.230
# Not a model: a constant would be wrong, because the ratio moves with device and
# resolution - an earlier additive attempt fitted three points and was falsified
# by a fourth. This is a running mean of measured ratios and converges on
# whatever this container actually does.
# The gpu seed moved 1.46 -> 2.10 when the model went to opset 18 and Resize
# stopped falling back to the CPU provider. Inference went from 0.105 to 0.021
# s/frame, so ffmpeg - decode, emit, x264, none of which got faster - became the
# dominant share of a segment and the ratio rose accordingly. Measured 2.120 and
# 2.111 on two consecutive 179-gap 720p segments.
#
# Leaving it at 1.46 was not merely inaccurate: _segment_duration falls back to
# `gaps * rate * overhead * 1.4` when no _PER_GAP measurement exists, so the
# FIRST segment of a long clip would have declared ~85s for ~89s of work and been
# killed by ZeroGPU. Making the model faster broke the duration declaration.
_OVERHEAD = {"gpu": 2.10, "cpu": 1.19}
_OVERHEAD_N = {"gpu": 5, "cpu": 5}

# Wall seconds per gap of the last FULL segment at each (h, w, factor, device).
# Set in the parent after a segment returns, read in the parent by the duration
# callable, so no fork boundary is crossed. Empty until a segment completes,
# which is why the reconstruction below it still has to exist.
#
# The DEVICE belongs in the key. The ratio moves with resolution and factor - the
# note in plan_gaps is explicit about that - but the per-gap cost itself moves
# with the hardware by roughly 4x, so a CPU-measured value would size a GPU
# segment at four times its true length and hand it a duration to match.
_PER_GAP: dict[tuple[int, int, int, str], float] = {}

# The wait the estimate always ignored. A ZeroGPU allocation queues before it
# forks: measured 45-46s for the first call of a session, then 1-2s per later
# call. On a short clip that first wait is the LARGEST term - a real job logged
# "estimate 0:26", did 31s of work and took 1:17 wall clock, and the 46s nobody
# was accounting for is the whole difference.
# Seeded from a COLD container, where the first allocation of a session queued
# 45-46s. That is the worst case, not the normal one: a warm Space allocates in
# ~1s, and a browser run whose button promised 59s finished in 22 - 39 phantom
# seconds, all of them this constant. Now that `took - spent` measures the queue
# directly, it is a running mean like _OVERHEAD rather than a guess.
GPU_QUEUE_S = 45.0
_QUEUE_N = 1
GPU_ACQUIRE_S = 1.5

# ZeroGPU's per-call ceiling. Declaring above a tier's cap does not run slower,
# it is refused outright with "ZeroGPU illegal duration", so this is a hard limit
# rather than a target and every duration is clamped to it.
GPU_MAX_S = 119

# What the RE-CUT aims a segment at, in wall-clock seconds.
#
# BUDGET_S is not this. plan_gaps charges it against measure_rate, which times
# one bare inference, so BUDGET_S = 60 means 60s of INFERENCE - about 88s of wall
# time once decode, emit and x264 are added. The re-cut divides a wall-clock
# measurement, so it needs a wall-clock budget or it silently re-plans every job
# down by the overhead factor: want = gaps_per / 1.46, which trips the 0.8
# deadband on the first segment of every GPU job and then settles there. That put
# segments at 60s inside a 119s window and cost 1.46x more GPU runs than
# intended, reported in the log as ordinary adaptive behaviour.
#
# 0.75 of the cap restores the planner's own intent (~88s) with 30s of headroom.
# Safe because real_per_gap is a measurement of the segment that just ran, not
# the 1.566x worst case that forced BUDGET_S to half the cap - and an overrun
# still lands in the halve-and-retry path at the cost of one run.
REAL_BUDGET_S = 0.75 * GPU_MAX_S if ZEROGPU else 900.0


def note_queue(seconds: float) -> None:
    """Fold one observed allocation wait into the running mean.

    Bounded below at 0 because a clock skew or a cancelled segment can make the
    subtraction negative, and above at 120 because anything larger is not a queue
    - it is a stall the estimate should not learn from.
    """
    global GPU_QUEUE_S, _QUEUE_N
    if not 0.0 <= seconds <= 120.0:
        return
    _QUEUE_N = min(_QUEUE_N + 1, 20)
    GPU_QUEUE_S += (seconds - GPU_QUEUE_S) / _QUEUE_N


def note_overhead(dev: str, ratio: float) -> None:
    """Fold one full segment's measured/assumed ratio into the running mean."""
    if not 0.2 < ratio < 20:                 # a cancelled or starved segment
        return
    n = _OVERHEAD_N[dev] = min(_OVERHEAD_N[dev] + 1, 40)
    _OVERHEAD[dev] += (ratio - _OVERHEAD[dev]) / n


# How far pixel-linear scaling may be trusted before the estimate stops being a
# single number. It was 15% out at 852x480, which is fine, and 31x out at
# 3840x2160, which is not. Two is deliberately conservative: past it, say a
# range rather than a number nobody can stand behind.
TRUST_RATIO = 2.0


def _rate_for(h: int, w: int, dev: str | None = None
              ) -> tuple[float, bool, float] | None:
    """Per-frame rate at this size, whether it was measured AT this size, and how
    far the answer had to be extrapolated (target pixels / measured pixels)."""
    if dev is None:
        dev = "gpu" if (ZEROGPU and not GPU_DEAD) else "cpu"
    hit = _RATE_CACHE.get((h, w, dev))
    if hit:
        return hit[0], True, 1.0
    # Fall back to pixel-scaling from the nearest measurement on the same device.
    # That model was measured 15% out at 852x480, which is why it only ever feeds
    # a rounded label carrying a "~", and never the segment planner.
    same = [(hh, ww, v[0]) for (hh, ww, d), v in list(_RATE_CACHE.items())
            if d == dev]
    if not same:
        # Cold container: _RATE_CACHE is empty until a job runs, and seeding it
        # here would be a bug - gpu_rate/cpu_rate treat a present key as measured
        # and would plan the whole job off an a-priori number. So fall back the
        # way header() already does, without writing anything back. REF_RATE was
        # measured at import on THIS container; GPU_REF_RATE_720P is the only
        # a-priori constant in the estimate, and it is replaced the moment a real
        # GPU rate lands.
        same = [(720, 1280, GPU_REF_RATE_720P if dev == "gpu" else REF_RATE)]
    hh, ww, r = min(same, key=lambda t: abs(t[0] * t[1] - h * w))
    ratio = (h * w) / (hh * ww)
    return r * ratio, False, ratio


def eta(clip: R.Clip, factor: int, max_h: int) -> tuple[float, float] | None:
    """Wall clock from click to result, as (likely, worst) seconds.

    The log's `estimate` line has always been rate x frames, so it promised less
    than the user waits: it left out the per-segment overhead the plan
    under-charges, and the ZeroGPU allocation queue entirely.

    The two numbers differ only where the rate had to be extrapolated well past
    anything measured on this container. A 3840x2160 job was quoted 13 min and
    ran 3.5 h: the label scaled the measured 720p GPU rate up by pixel count
    while the job itself fell back to the CPU, whose 4K cost the planner had
    right all along. Past TRUST_RATIO the CPU price is therefore the honest
    upper bound.
    """
    tw, th = scale_for(clip, int(max_h))
    dev = "gpu" if (ZEROGPU and not GPU_DEAD) else "cpu"
    hit = _rate_for(th, tw, dev)
    if hit is None:
        return None
    rate, _exact, ratio = hit

    def total_for(r: float, d: str) -> float:
        t = r * clip.frames * (int(factor) - 1) * _OVERHEAD[d]
        if d == "gpu":
            gaps_per = max(1, R.plan_gaps(clip, int(factor), r, BUDGET_S))
            segs = max(1, -(-(clip.frames - 1) // gaps_per))
            t += segs * GPU_ACQUIRE_S
            if (th, tw, "gpu") not in _RATE_CACHE:
                t += GPU_QUEUE_S      # the cold probe the user waits through
        return t

    likely = total_for(rate, dev)
    if ratio <= TRUST_RATIO or dev != "gpu":
        return likely, likely
    slow = _rate_for(th, tw, "cpu")
    worst = total_for(slow[0], "cpu") if slow else likely
    return likely, max(likely, worst)


_ETA_PROBE: dict[str, R.Clip] = {}


def button_label(video, factor, max_h):
    """Put the cost in the control that starts it, recomputed as inputs change.

    Never raises into the UI and never blocks it: a probe failure or an unknown
    rate just leaves the plain label, because a button that cannot be clicked is
    worse than a button with no number on it.
    """
    if not video:
        return gr.update(value="Interpolate")
    try:
        clip = _ETA_PROBE.get(video)
        if clip is None:
            if len(_ETA_PROBE) > 32:
                _ETA_PROBE.clear()
            clip = _ETA_PROBE[video] = R.probe(video)
        pair = eta(clip, factor, max_h)
    except Exception:                                        # noqa: BLE001
        return gr.update(value="Interpolate")
    if not pair or not pair[0]:
        return gr.update(value="Interpolate")
    likely, worst = pair
    if worst > likely * 1.5:
        # Never a lone "~13 min" for a size nothing was measured at. The spread
        # IS the message: it says the height cap is the lever worth moving.
        return gr.update(value=f"Interpolate ({plain(likely)} - {plain(worst)})")
    return gr.update(value=f"Interpolate (~{plain(likely)})")


def scale_for(clip: R.Clip, max_h: int) -> tuple[int, int]:
    """Both dimensions are forced even unconditionally: yuv420p subsamples chroma
    2x2 and x264 refuses an odd side, so a 1920x1038 or phone-cropped source used
    to die on the first frame written even though nothing else about it was hard."""
    if max_h <= 0 or clip.height <= max_h:
        return clip.width // 2 * 2, clip.height // 2 * 2
    w = int(round(clip.width * max_h / clip.height)) // 2 * 2
    return w, max_h // 2 * 2


def run(video: str, factor: int, max_h: int, crf: float,
        request: gr.Request, progress=gr.Progress()):
    """Interpolate a video to a higher frame rate with RIFE.

    Args:
        video: Path to the source clip. Any container ffmpeg can read.
        factor: Frame rate multiplier, 2 doubles the frame rate and 4 quadruples it.
        max_h: Cap the output height, 0 keeps the source resolution. Lowering this is
            the single biggest lever on runtime, since cost scales with pixels.
        crf: x264 quality, lower is better and larger. 18 is visually lossless.
    """
    global OWNER, GPU_DEAD
    log = Log()
    me = getattr(request, "session_hash", None) or f"anon-{uuid.uuid4()}"
    if not LOCK.acquire(blocking=False):
        raise gr.Error("A job is already running. Two vCPUs, one job. "
                       "Press Stop to cancel it.")
    CANCELLED.discard(me)
    OWNER = me
    closing = False
    stopped = ""                     # why the run ended early, if it did
    out_path = None
    parts: list[str] = []
    work = None
    try:
        # Before the mkdtemp, because mkdtemp is the first thing that fails on a
        # full disk. Inside the try, not before it: failing between acquire() and
        # try: held the lock with nothing to release it, wedging every later job.
        sweep_outputs(protect=video)
        work = tempfile.mkdtemp(prefix="ff_", dir=str(SCRATCH))
        clip = R.probe(video)
        tw, th = scale_for(clip, int(max_h))
        src = video

        yield None, log(f"source {clip.width}x{clip.height} {clip.fps:.2f}fps "
                        f"{clip.frames} frames {clip.duration:.1f}s"), *BUSY

        if (tw, th) != (clip.width, clip.height):
            yield None, log(f"scaling to {tw}x{th} inside each segment decoder, "
                            f"no pre-transcode"), *BUSY

        # Estimate from a measured rate at this resolution, before committing.
        # Start from what the last job learned: a cached GPU rate never raises,
        # so an optimistic retry costs a failed segment and a re-plan per job.
        gpu_dead = GPU_DEAD
        if ZEROGPU:
            try:
                rate, be_name, build_s = gpu_rate(th, tw)
                # Clear what the LAST job learned. Without this, GPU_DEAD from an
                # earlier quota-out job survived a successful probe: gpu_rate
                # returned a GPU rate and name, the header and the estimate
                # quoted them, and then every segment ran on _segment_cpu because
                # gpu_dead was still True. Seen live on 2026-09-06 - "GPU CUDA:
                # 0.053s/frame ... estimate 0:09" against a measured 57.4s, with
                # the plan-check line labelling its own segment "cpu".
                #
                # The comment above this block already described the intended
                # behaviour ("an optimistic retry costs a failed segment and a
                # re-plan per job"); the retry just never happened. A cached rate
                # does not re-prove the quota, so this is deliberately
                # optimistic: the segment loop catches the quota error, sets the
                # flag and re-cuts the rest for CPU. One failed segment is the
                # price of not pinning the Space to CPU until it restarts, and it
                # keeps the one invariant that matters - the rate that plans the
                # job is the rate of the device that runs it.
                gpu_dead = GPU_DEAD = False
            except Exception as e:                           # noqa: BLE001
                if not (is_quota_error(e) or is_timeout_error(e)):
                    raise
                gpu_dead = GPU_DEAD = True
                # str(e), not type(e).__name__: ZeroGPU raises gradio's own
                # Error, so the class name rendered as the literal "no usable
                # GPU (Error)" - true, and carrying none of the quota message
                # that tells the user when the GPU comes back.
                why = " ".join(str(e).split())[:110] or type(e).__name__
                yield None, log(f"no usable GPU ({why}), "
                                f"running on CPU"), *BUSY
                rate, be_name, build_s = cpu_rate(th, tw)
        else:
            rate, be_name, build_s = cpu_rate(th, tw)
        new_frames = clip.frames * (int(factor) - 1)
        est = rate * new_frames
        if build_s > 1:
            yield None, log(f"{be_name} session built in {build_s:.0f}s"), *BUSY
        yield None, log(f"{be_name}: {rate:.3f}s/frame here, "
                        f"{new_frames} frames to generate, estimate {hms(est)}"), *BUSY

        if est > 6 * 3600:
            # BACKEND is always the CPU session, built at import with
            # prefer_gpu=False, so testing BACKEND.on_gpu here named the CPU
            # ceiling on GPU hardware every time. Ask the flag that knows.
            fits = int(clip.duration * (6 * 3600) / est)
            raise gr.Error(
                f"Estimated {hms(est)} at {tw}x{th}, past the 6 hour ceiling"
                f"{' and well past a ZeroGPU day' if ZEROGPU else ''}. "
                f"Lower 'Max height' to {min(720, max(240, th // 2))}, which is "
                f"about 4x faster, or trim the clip to under "
                f"{hms(max(1, fits))} at this size.")

        # Plan segment 0 from a MEASUREMENT when this container has already run
        # this shape on this device. plan_gaps charges BUDGET_S against a bare
        # inference rate, so it deliberately mis-sizes the first segment and the
        # re-cut then fires once to correct it - on every repeat job, forever.
        #
        # It also mis-states the plan LENGTH, and `len(plan) > 30` below is what
        # gates the "you will run out of ZeroGPU quota, lower your resolution"
        # warning. At a clean overhead of 1.26 a job that really needs 30 calls
        # was announced as 35, so the user was told to halve their height for a
        # limit they were not going to hit.
        dev0 = "cpu" if gpu_dead else "gpu"
        seen0 = _PER_GAP.get((th, tw, int(factor), dev0))
        if seen0:
            gaps_per = max(1, int(REAL_BUDGET_S / seen0))
            budget_note = f"{REAL_BUDGET_S:.0f}s measured budget"
        else:
            # Charge the overhead here too. plan_gaps bills against measure_rate,
            # which times bare inference, so BUDGET_S seconds of inference is
            # BUDGET_S * overhead of WALL time - and the cap is on wall time.
            #
            # That gap widened when the model got five times faster: ffmpeg did
            # not, so the ratio went 1.46 -> 1.85 and the same BUDGET_S = 60 went
            # from 87.6s (74% of the 119s cap, the intent) to 111.0s (93%). The
            # first segment of every job was being planned with 7% headroom, and
            # the re-cut that would fix it only runs AFTER that segment lands.
            #
            # Billing rate * overhead against REAL_BUDGET_S puts the first
            # segment on the same 75%-of-cap target the re-cut aims for, and it
            # stays correct automatically as the ratio moves.
            gaps_per = R.plan_gaps(clip, int(factor), rate * _OVERHEAD[dev0],
                                   REAL_BUDGET_S)
            budget_note = f"{REAL_BUDGET_S:.0f}s wall budget"
        plan = R.segment_bounds(clip, gaps_per)
        total_gaps = max(1, clip.frames - 1)
        yield None, log(f"{len(plan)} segment(s) of up to {gaps_per} gaps "
                        f"({gaps_per / clip.fps:.1f}s), inside the "
                        f"{budget_note}"), *BUSY

        # On ZeroGPU each segment is a separate allocation and the daily budget
        # is counted in RUNS, so segment count - not estimated hours - is what
        # decides whether a job can finish on the GPU. The 6 hour ceiling above
        # is checked against the GPU estimate, so a 4K five minute clip at source
        # resolution sails through it at ~2.8h and then needs ~165 allocations:
        # quota dies partway, the CPU fallback takes over at roughly 6 s/frame,
        # and the user who was shown "2:45" is waiting most of a day. Say so
        # before any of it starts, with the number, rather than letting them find
        # out hours in.
        if ZEROGPU and not gpu_dead and len(plan) > 30:
            # The remedy has to be an action the user can actually take. A
            # suggestion to cap height at the height they are already using is
            # not advice, and 240 is the floor of the dropdown - below that the
            # only lever left is a shorter clip. Cost scales with pixels, so
            # halving the height is ~4x fewer calls; say the number either way.
            lower = next_lower_height(th)
            if lower:
                # Cost scales with pixels, so the saving is the area ratio.
                saved = max(1, int(len(plan) * (lower / th) ** 2))
                remedy = (f"Capping 'Max height' at {lower} (from {th}) would "
                          f"cut it to about {saved} calls.")
            else:
                fits = clip.duration * 30 / len(plan)
                # Reached when no offered cap is BELOW the current height -
                # either it is already the lowest option, or the source is
                # smaller than every option. Saying "the control's lowest" would
                # be wrong in the second case, so say what is true in both.
                remedy = (f"no lower cap is available for a {th}p source, so the "
                          f"lever left is length: about "
                          f"{hms(max(1, int(fits)))} of this clip fits in "
                          f"30 calls.")
            yield None, log(
                f"heads up: {len(plan)} GPU calls needed, and ZeroGPU meters a "
                f"daily number of them. This will likely run out partway and "
                f"finish on CPU, which is far slower. " + remedy), *BUSY

        done = 0
        cursor = 0                       # gaps already rendered
        i = 0
        shrinks = 0                      # GPU-window overruns absorbed so far
        while cursor < total_gaps:
            if me in CANCELLED:
                # Set `stopped` like every other early exit. Without it the
                # summary read "done: 121 frames" on a job that expected 599 -
                # the word done, and a number with nothing to compare it to.
                # Cancel is the most frequently taken of the early exits and was
                # the only one not marking the result partial.
                stopped = "cancelled at your request"
                yield None, log("cancelled, keeping what finished"), *BUSY
                break
            if i >= len(plan):
                # Never silent. cursor advancing without a segment to consume it
                # is a planner bug, and the two guards below cover the ways it
                # has actually happened; this catches any third.
                stopped = (f"the plan ran out at {cursor} of {total_gaps} gaps")
                yield None, log(f"stopped early: {stopped}"), *BUSY
                break
            t0, gaps, is_last = plan[i]
            part = os.path.join(work, f"seg{i:04d}.mp4")
            t_seg = time.perf_counter()
            try:
                worker_fn = _segment_cpu if gpu_dead else _segment_gpu
                n, inside = worker_fn(src, part, t0, gaps, int(factor), clip,
                                      int(crf), is_last, me, tw, th)
            except Exception as e:                           # noqa: BLE001
                # ZEROGPU gates this: only there is there a window to overrun,
                # and TIMEOUT_HINTS is broad enough to match ordinary ffmpeg
                # errors that retrying cannot help.
                overran = ZEROGPU and not gpu_dead and is_timeout_error(e)
                if overran and shrinks < 4 and gaps_per > 1:
                    # The segment was too big for its GPU window, not the day's
                    # budget. measure_rate times one bare inference while a real
                    # segment also runs factor-1 encodes and emits per gap, and
                    # at 4x that overhead is where a 90 s plan meets a 120 s cap.
                    # Halve and retry the same range rather than abandoning the
                    # GPU for the rest of the job.
                    shrinks += 1
                    gaps_per = max(1, gaps_per // 2)
                    plan = plan[:i] + R.segment_bounds(clip, gaps_per, cursor)
                    yield None, log(f"segment overran its GPU window, re-cut to "
                                    f"{gaps_per} gaps and retrying "
                                    f"({len(plan) - i} segments left)"), *BUSY
                    continue
                if gpu_dead or not (overran or is_quota_error(e)):
                    raise
                # Two ways to get here. Quota exhausted, or a single gap that
                # cannot fit the GPU window even after shrinking - which used to
                # kill the job outright, the one case where the GPU truly cannot
                # proceed being the one case that refused the existing fallback.
                why = ("one gap will not fit the GPU window" if overran
                       else "ZeroGPU quota exhausted")
                #
                # Re-cutting the remainder is the part that matters. These
                # segments were sized from a GPU rate, so one of them is roughly
                # six times more work than the CPU budget allows: a segment
                # yields once, at the end, so an over-sized one freezes progress,
                # gives Stop that much latency, and stops the seams being
                # checkpoints at all. Caching the GPU rate made this the common
                # path, because a cached rate never raises the quota error that
                # used to be caught before any planning happened.
                gpu_dead = GPU_DEAD = True
                rate, be_name, _ = cpu_rate(th, tw)
                gaps_per = R.plan_gaps(clip, int(factor), rate, BUDGET_S)
                plan = plan[:i] + R.segment_bounds(clip, gaps_per, cursor)
                left = rate * (total_gaps - cursor) * max(1, int(factor) - 1)
                yield None, log(f"{why}. Finishing on {be_name} at "
                                f"{rate:.3f}s/frame: re-cut the rest into "
                                f"{len(plan) - i} segments, about {hms(left)} to "
                                f"go. Frames so far are kept."), *BUSY
                continue
            # plan_gaps budgets rate*(factor-1) per gap from measure_rate, which
            # times one bare inference. A real segment also runs `factor` emit
            # passes and `factor` x264 frames per gap. That overhead is roughly
            # fixed per gap while the inference term scales with the device, so
            # the plan under-charges least on CPU and most on a fast GPU - which
            # is exactly where the 120 s cap lives. Logged rather than guessed,
            # so any real job on any device reports its own correction factor.
            took = time.perf_counter() - t_seg      # wall, includes the queue
            spent = inside                          # what the work actually cost
            assumed = gaps * rate * max(1, int(factor) - 1)
            if assumed > 0:
                # Only a segment that rendered its whole assignment describes the
                # overhead. A cancelled or starved one returns early and logs
                # ratios like 0.399x, which would drag the mean toward nonsense.
                if n // max(1, int(factor)) >= gaps:
                    note_overhead("gpu" if (ZEROGPU and not gpu_dead) else "cpu",
                                  spent / assumed)
                    if ZEROGPU and not gpu_dead:
                        note_queue(took - spent)
                    if gaps > 0:
                        _PER_GAP[(th, tw, int(factor),
                                  "cpu" if gpu_dead else "gpu")] = spent / gaps
                print(f"[ff] plan-check seg{i} {gaps} gaps f{factor} {tw}x{th} "
                      f"{'gpu' if (ZEROGPU and not gpu_dead) else 'cpu'}: "
                      f"assumed {assumed:.2f}s "
                      f"measured {spent:.2f}s ratio {spent / assumed:.3f}x "
                      f"(wall {took:.2f}s, queue {took - spent:.2f}s)",
                      flush=True)
            if not gpu_dead:
                GPU_DEAD = False          # a GPU segment landed, quota is back
            if n:                         # a cancelled segment writes no file
                parts.append(part)
            done += n
            # A segment cancelled part-way returns fewer frames than its gap
            # assignment. Crediting the whole assignment made the last line of a
            # cancelled job overstate progress: 60 of 179 gaps reported against
            # 156 frames, which at 4x is 39 gaps of real work. Each gap writes
            # `factor` frames, so the frames are the honest count.
            got_gaps = min(gaps, n // max(1, int(factor)))
            cursor += got_gaps
            i += 1
            if got_gaps < gaps and me not in CANCELLED:
                # The decoder handed back fewer frames than this segment covers,
                # and nobody cancelled: the source is shorter than its header
                # claims. On CPU nothing re-cuts the plan, so cursor stopped
                # short of total_gaps while i walked past the last entry, and the
                # next pass raised "IndexError: list index out of range" - which
                # is what the live Space showed for a truncated upload while this
                # machine's ffmpeg raised the clean starved-decoder error before
                # ever reaching here. Same file, two decoders, one path untested.
                stopped = (f"source ended early: this segment covers {gaps} gaps "
                           f"but the decoder supplied {got_gaps}. The file is "
                           f"probably truncated or corrupt - its header claims "
                           f"{clip.frames} frames.")
                yield None, log(f"stopped early: {stopped}"), *BUSY
                break

            # Re-cut from what this segment ACTUALLY cost, not from a constant.
            # The plan charges only inference, so it is optimistic by a fixed
            # per-gap amount that is a bigger fraction the faster the device is -
            # measured 1.456x at 852x480 on the GPU against 1.03x on CPU. That
            # ratio cannot be bounded ahead of time: a smaller frame makes
            # inference cheaper without making ffmpeg cheaper, so it climbs as
            # resolution drops, and four measurements were neither constant nor
            # pixel-linear. So do not model it - measure it here and shrink the
            # rest of the plan if the segment that just ran says the next one
            # would not fit the window. Costs nothing when the plan is right.
            # Every device, not just the GPU. The gate used to read
            # `ZEROGPU and not gpu_dead`, on the reasoning that only a GPU has a
            # window to overrun - but a segment 5x its plan is a problem on any
            # device, because a segment yields once at the END. A 2290x1708 job
            # planned at 59s a segment took 318s and showed the user a frozen log
            # for 5m18s with no progress line, no ETA and 5 minutes of Stop
            # latency (2026-09-06). The 120 s cap is why the GPU MUST re-cut; the
            # seams being checkpoints at all is why everything else should.
            #
            # Cheap when the plan is right: the measured CPU under-charge is
            # 1.05-1.23x, which lands inside the 20% deadband and re-cuts nothing.
            if gaps > 0 and cursor < total_gaps:
                real_per_gap = spent / gaps
                # Both directions. Shrinking keeps a segment inside the window;
                # growing refunds a conservative BUDGET_S when the plan turns out
                # pessimistic, so starting safe does not cost a GPU run per
                # segment for the whole job. real_per_gap already contains the
                # under-charge, so this is a measurement and not a correction
                # factor. The 20% deadband stops one noisy segment churning the
                # plan, and BUDGET_S sits at half the hard cap, so even a segment
                # twice as slow as the one just measured still lands inside it.
                want = max(1, int(REAL_BUDGET_S / real_per_gap))
                if want < gaps_per * 0.8 or want > gaps_per * 1.25:
                    grew = want > gaps_per
                    gaps_per = want
                    plan = plan[:i] + R.segment_bounds(clip, gaps_per, cursor)
                    yield None, log(f"segment cost {real_per_gap:.3f}s/gap against "
                                    f"{rate * max(1, int(factor) - 1):.3f}s planned; "
                                    f"{'widened' if grew else 're-cut'} the rest into "
                                    f"{len(plan) - i} segments"), *BUSY
            progress(cursor / total_gaps, desc=f"{cursor}/{total_gaps} gaps")
            yield None, log(f"[{cursor}/{total_gaps} gaps] {n} frames, "
                            f"{done} total"), *BUSY
    except GeneratorExit:
        # Tab closed or SSE torn down. Yielding from finally here would raise
        # "generator ignored GeneratorExit" and block teardown while ffmpeg muxes.
        closing = True
        CANCELLED.add(me)
        raise
    except gr.Error as e:
        # Mirror the generic handler below. This clause exists so a validation
        # error raised BEFORE any work reaches the user as a clean toast, and
        # with no parts that is still exactly what happens. But ZeroGPU itself
        # raises gr.Error, so a mid-job GPU failure whose text matches neither
        # QUOTA_HINTS nor TIMEOUT_HINTS lands here with segments already on disk:
        # the finally delivered them labelled "done", for a job that produced a
        # fraction of its frames, with the log replaced by an Error badge.
        if not parts:
            raise
        stopped = f"{type(e).__name__}: {e}"
        yield None, log(f"stopped early: {stopped}"), *IDLE
    except Exception as e:                                   # noqa: BLE001
        if parts:
            # Segments landed, so there IS a result. Raising here would hand
            # Gradio an errored generator, and verified in Chrome on 2026-09-06
            # against truncated.mp4 that means: the video the finally yields
            # survives and stays downloadable, but the log component is replaced
            # by a bare "Error" badge. The user gets a silently truncated clip
            # with the one line explaining it wiped off the screen. Record the
            # reason and let the finally deliver both.
            stopped = f"{type(e).__name__}: {e}"
            yield None, log(f"stopped early: {stopped}"), *IDLE
        else:
            yield None, log(f"failed: {type(e).__name__}: {e}"), *IDLE
            raise
    finally:
        # Package from finally so cancel, error and success all hand back the
        # frames that were actually produced. The flag is re-derived here because
        # a GeneratorExit raised at the yield inside `except Exception` is a
        # sibling of the GeneratorExit handler and never reaches it.
        closing = closing or sys.exc_info()[0] is GeneratorExit
        try:
            if closing:
                # Do not YIELD - that raises "generator ignored GeneratorExit"
                # and blocks teardown. But keeping is a separate decision from
                # yielding, and the inner finally is about to rmtree `work` with
                # every finished segment inside it. Concat into OUTDIR so the
                # result survives the cleanup and is recoverable from the log,
                # instead of a dropped SSE costing a 40 minute job outright.
                if parts:
                    try:
                        fd, sv = tempfile.mkstemp(suffix=".mp4", dir=str(OUTDIR))
                        os.close(fd)
                        R.concat(parts, sv)
                        print(f"[ff] stream closed mid-job; kept {len(parts)} "
                              f"segments at {sv}", flush=True)
                    except Exception as e:                   # noqa: BLE001
                        # Best effort on a path that is already unwinding.
                        print(f"[ff] stream closed mid-job and the salvage "
                              f"failed: {type(e).__name__}: {e}", flush=True)
            elif parts:
                # Guarded for the same reason the probe below is: describing OR
                # packaging the result must never be able to discard it. A raise
                # from concat, mkstemp or the copy escapes this finally so
                # nothing is yielded, REPLACES the exception already in flight -
                # overwriting a segment failure already recorded in `stopped` -
                # and still reaches the inner finally that deletes `work`. The
                # parts would be destroyed by cleanup for a delivery that never
                # happened. Falling back to the largest single segment keeps most
                # of the job rather than none of it.
                out_path = None
                try:
                    joined = os.path.join(work, "joined.mp4")
                    R.concat(parts, joined)
                    fd, out_path = tempfile.mkstemp(suffix=".mp4", dir=str(OUTDIR))
                    os.close(fd)
                    if not R.mux_audio(joined, video, out_path):
                        shutil.copy(joined, out_path)
                except Exception as e:                       # noqa: BLE001
                    stopped = stopped or f"{type(e).__name__}: {e}"
                    big = max(parts, key=lambda p: os.path.getsize(p)
                              if os.path.exists(p) else 0)
                    try:
                        fd, out_path = tempfile.mkstemp(suffix=".mp4",
                                                        dir=str(OUTDIR))
                        os.close(fd)
                        shutil.copy(big, out_path)
                        log(f"could not join the segments ({e}); handing back the "
                            f"longest one instead")
                    except Exception:                        # noqa: BLE001
                        out_path = None
                if out_path is None:
                    # No `return` here: a return inside a finally DISCARDS an
                    # exception that is still propagating. An else instead, so
                    # the summary below cannot run on a None path and probe it.
                    yield None, log("the frames were rendered but could not be "
                                    "packaged; nothing to hand back"), *IDLE
                else:
                    # Describing the result must never be able to discard it.
                    # probe raises when it cannot establish a frame count, and
                    # this call happens after the work is finished and written.
                    head = "partial" if stopped else "done"
                    try:
                        got = R.probe(out_path)
                        summary = (f"{head}: {got.width}x{got.height} "
                                   f"{got.fps:.2f}fps {got.frames} frames")
                        if stopped:
                            summary += (f" of the "
                                        f"{(clip.frames - 1) * int(factor) + 1} "
                                        f"expected - the download below is the "
                                        f"part that finished")
                    except Exception as e:                   # noqa: BLE001
                        summary = (f"{head}, {done} frames written "
                                   f"({type(e).__name__})")
                    yield out_path, log(summary), *IDLE
            else:
                yield None, log("no frames produced"), *IDLE
        finally:
            if work:
                shutil.rmtree(work, ignore_errors=True)
            OWNER = None
            LOCK.release()


BUSY = (gr.update(visible=False), gr.update(visible=True))
IDLE = (gr.update(visible=True), gr.update(visible=False))

CSS = """
footer{display:none}
.gradio-container{padding:4px !important;max-width:100% !important}
.form{gap:2px !important}
#hdr{max-height:34px}
#hdr p{margin:0 !important;font-size:.82em;opacity:.75}
#log textarea{font-family:ui-monospace,monospace;font-size:.78em}
#vid video{max-height:190px}
:where(button,input,textarea,select,[role="button"],[tabindex]):focus-visible{
  outline:2px solid var(--color-accent) !important;outline-offset:2px}
@media (prefers-reduced-motion:reduce){*,*::before,*::after{
  animation-duration:.01ms !important;transition-duration:.01ms !important}}
"""

with gr.Blocks(title="Flowframes", fill_width=True) as demo:
    hdr = gr.Markdown(header(), elem_id="hdr")
    with gr.Row():
        with gr.Column(scale=4):
            vid = gr.Video(label="Source clip", elem_id="vid", height=190)
            go = gr.Button("Interpolate", variant="primary")
            halt = gr.Button("Stop", variant="stop", visible=False)
            # label with show_label=False: hidden visually, still the control's
            # accessible name. Without it this is the one element on the page a
            # screen reader announces as just "Textbox", and it is the element
            # carrying every progress line, the ETA and every error message.
            status = gr.Textbox(lines=8, max_lines=LOG_LINES + 2, autoscroll=True,
                                label="Progress log", show_label=False,
                                container=False, elem_id="log")
        with gr.Column(scale=2):
            factor = gr.Radio([2, 4], value=2, label="Frame rate multiplier",
                              info="4x generates three frames per gap, so it costs "
                                   "three times 2x.")
            max_h = gr.Dropdown(
                [("Keep source resolution", 0)]
                + [(f"Cap at {h}p" + (" (fastest)" if h == 720 else ""), h)
                   for h in HEIGHTS],
                value=720, label="Max height",
                info="Cost scales with pixels. Halving height is roughly 4x faster.")
            with gr.Accordion("Advanced", open=False):
                crf = gr.Slider(12, 28, value=18, step=1, label="x264 CRF",
                                info="Lower is better quality and a larger file.")
        with gr.Column(scale=4):
            out = gr.Video(label="Result", height=190, interactive=False)
            dl = gr.DownloadButton("Download result", size="sm", interactive=False)

    # Registered before the work handler and outside the queue, so the swap lands
    # immediately rather than waiting behind a job that may run for hours.
    # Validation lives in the swap handler, so a rejected click never swaps the
    # buttons. Raising after the swap stranded the UI showing Stop forever.
    # The trailing .then restores the buttons on paths that never reach run()'s
    # final yield: the lock-busy gr.Error is raised before the try, so the swap
    # that preflight already made would otherwise leave the page showing Stop
    # forever. .then runs on failure too, which .success would not.
    go.click(preflight, inputs=[vid], outputs=[go, halt], queue=False).then(
        run, [vid, factor, max_h, crf], [out, status, go, halt]).then(
        lambda: IDLE, outputs=[go, halt], queue=False)
    out.change(lambda v: gr.update(value=v, interactive=v is not None),
               inputs=out, outputs=dl, queue=False)
    # queue=False so the label never waits behind a running job, which under
    # concurrency_limit=1 would mean the number arrives after the work finishes.
    for _c in (vid, factor, max_h):
        _c.change(button_label, [vid, factor, max_h], [go], queue=False)
    halt.click(stop, outputs=status, queue=False)
    # Recomputed per page load, so the first real job replaces the reference
    # estimate with the rate measured on the hardware actually serving this Space.
    demo.load(header, outputs=[hdr], queue=False)

    ex = HERE / "examples" / "sample.mp4"
    if ex.exists():
        # Cached so the first visitor is not asked to supply a video before they
        # can see anything work. Lazy, not eager: eager runs the whole pipeline
        # before the Space serves its first request, and on ZeroGPU that would
        # also spend quota on every rebuild. Cached after the first click, then
        # instant for everyone.
        gr.Examples(examples=[[str(ex), 2, 480, 20]],
                    inputs=[vid, factor, max_h, crf],
                    outputs=[out, status],
                    fn=example_run,
                    cache_examples=True, cache_mode="lazy",
                    label="Example (cached after first run)")

if __name__ == "__main__":
    demo.queue(max_size=4, default_concurrency_limit=1).launch(css=CSS, show_error=True)