From 7f46f3215948ec681bdc21743b2b635a4e729211 Mon Sep 17 00:00:00 2001
From: shard872 <106123738+shard872@users.noreply.github.com>
Date: Thu, 6 Aug 2026 01:28:31 -0400
Subject: [PATCH] fix: stop downloads when destination is full (#116)

---
 moon_cli.py                | 148 ++++++---
 moon_download.py           | 129 +++++++-
 moon_engine.py             | 172 ++++++----
 tests/test_disk_full.py    | 659 +++++++++++++++++++++++++++++++++++++
 tests/test_exit_cleanup.py |   1 +
 5 files changed, 986 insertions(+), 123 deletions(-)
 create mode 100644 tests/test_disk_full.py

diff --git a/moon_cli.py b/moon_cli.py
index a8b3e3b..3edfa16 100644
--- a/moon_cli.py
+++ b/moon_cli.py
@@ -19,6 +19,7 @@
     _close_sess,
     _sanitize_filename,
     download_file,
+    RunFatalControl,
 )
 
 # ── EXTRACTION ────────────────────────────────────────────────────────────────
@@ -76,6 +77,7 @@ async def run(urls: list[str], output_dir: str, n_workers: int,
     ok_count      = 0
     fail_count    = 0
     dls_active    = 0
+    fatal_control = RunFatalControl()
 
     cfg = {"browsers": n_workers, "dl_streams": max_dl, "retries": max_retries,
            "total_links": len(urls)}
@@ -101,7 +103,12 @@ def mark_done():
     stop_progress = asyncio.Event()
     async def progress_loop():
         while not stop_progress.is_set():
-            await asyncio.sleep(2.0)
+            try:
+                await asyncio.wait_for(stop_progress.wait(), timeout=2.0)
+            except asyncio.TimeoutError:
+                pass
+            if stop_progress.is_set():
+                break
             snap = list(bytes_acc)
             now  = time.monotonic()
             cut  = now - 3.0
@@ -126,49 +133,61 @@ async def progress_loop():
     async def do_dl(proxy_url, cookies, filename, orig_url, rec):
         nonlocal ok_count, fail_count, dls_active
         async with dl_sem:
+            if fatal_control.is_set():
+                rec.status = "aborted"
+                return
             with lock: dls_active += 1
-            dest = os.path.join(output_dir, filename)
-
-            if os.path.exists(dest):
-                with lock:
-                    ok_count += 1; dls_active -= 1
-                print(f"  [exists] {filename}")
-                rec.status = "ok"; rec.dl_s = 0.0
-                mark_done(); return
-
-            kc       = kill_counts.get(orig_url, 0)
-            kill_evt = asyncio.Event()
-            ok, msg, bytes_done = await download_file(
-                proxy_url, cookies, dest, rec, bytes_acc, kill_evt, kc,
-                on_event=lambda msg, tag: print(f"  [{tag}] {msg}", flush=True))
-            rec.dl_s = max(time.monotonic() - rec.dl_start, 0.001)
-
-            if ok:
-                with lock: ok_count += 1
-                spd = f"  ({rec.avg_mbs:.1f} MB/s)" if rec.avg_mbs > 0 else ""
-                print(f"  [ok] {filename}{spd}")
-                rec.status = "ok"; mark_done()
-            elif msg == "stall_killed":
-                new_kc = kc + 1; kill_counts[orig_url] = new_kc
-                print(f"  [kill#{new_kc}] {filename}  ({bytes_done//(1<<20)}MB) -> re-extract")
-                rec.queued_at = time.monotonic(); rec.status = "pending"
-                await q.put((orig_url, 1, rec))
-            else:
-                with lock: fail_count += 1
-                failed_urls.append(orig_url)
-                rec.status = "fail"; rec.error = msg
-                print(f"  [fail] {filename}: {msg}")
-                mark_done()
-
-            with lock: dls_active -= 1
+            try:
+                dest = os.path.join(output_dir, filename)
+
+                if os.path.exists(dest):
+                    with lock: ok_count += 1
+                    print(f"  [exists] {filename}")
+                    rec.status = "ok"; rec.dl_s = 0.0
+                    mark_done(); return
+
+                kc       = kill_counts.get(orig_url, 0)
+                kill_evt = asyncio.Event()
+                ok, msg, bytes_done = await download_file(
+                    proxy_url, cookies, dest, rec, bytes_acc, kill_evt, kc,
+                    on_event=lambda msg, tag: print(f"  [{tag}] {msg}", flush=True),
+                    fatal_control=fatal_control)
+                rec.dl_s = max(time.monotonic() - rec.dl_start, 0.001)
+
+                if ok:
+                    with lock: ok_count += 1
+                    spd = f"  ({rec.avg_mbs:.1f} MB/s)" if rec.avg_mbs > 0 else ""
+                    print(f"  [ok] {filename}{spd}")
+                    rec.status = "ok"; mark_done()
+                elif msg == "stall_killed":
+                    new_kc = kc + 1; kill_counts[orig_url] = new_kc
+                    print(f"  [kill#{new_kc}] {filename}  ({bytes_done//(1<<20)}MB) -> re-extract")
+                    rec.queued_at = time.monotonic(); rec.status = "pending"
+                    if not fatal_control.is_set():
+                        await q.put((orig_url, 1, rec))
+                elif msg == "aborted_disk_full":
+                    rec.status = "aborted"
+                else:
+                    with lock: fail_count += 1
+                    failed_urls.append(orig_url)
+                    rec.status = "fail"; rec.error = msg
+                    if msg != "disk_full":
+                        print(f"  [fail] {filename}: {msg}")
+                    mark_done()
+            finally:
+                with lock: dls_active -= 1
 
     async def browser_worker(get_browser, wid):
         nonlocal ok_count, fail_count
-        while True:
+        while not fatal_control.is_set():
             if all_done.is_set() and q.empty(): break
             try:
                 url, attempt, rec = await asyncio.wait_for(q.get(), timeout=1.0)
             except asyncio.TimeoutError: continue
+            if fatal_control.is_set():
+                rec.status = "aborted"
+                q.task_done()
+                break
 
             rec.worker_id = wid
             t_start = time.monotonic()
@@ -187,6 +206,10 @@ async def browser_worker(get_browser, wid):
                     # reaches for a window if /go refuses it.
                     link = await extract_fuckingfast(url, get_browser)
                     rec.extract_s = time.monotonic() - t_start
+                    if fatal_control.is_set():
+                        rec.status = "aborted"
+                        q.task_done()
+                        break
                     if not link:
                         print("  [fail] No link found")
                     else:
@@ -205,6 +228,10 @@ async def _task(pu=link, fn=filename, ou=url, r=rec):
                     # the persistent lane pool internally.
                     proxy_url, cookies = await extract_datanodes(await get_browser(), url)
                     rec.extract_s = time.monotonic() - t_start
+                    if fatal_control.is_set():
+                        rec.status = "aborted"
+                        q.task_done()
+                        break
                     if not proxy_url:
                         print("  [fail] No URL extracted")
                     else:
@@ -225,15 +252,25 @@ async def _task(pu=proxy_url, co=cookies, fn=filename, ou=url, r=rec):
             except Exception as e:
                 print(f"  [error] {e}")
 
-            if not success and not unsupported and not is_re and attempt < max_retries:
+            if fatal_control.is_set():
+                rec.status = "aborted"
+                q.task_done()
+                break
+
+            if (not success and not unsupported and not is_re and attempt < max_retries
+                    and not fatal_control.is_set()):
                 backoff = min(2**(attempt-1), 6)
                 print(f"  [retry in {backoff}s]")
                 await asyncio.sleep(backoff)
+                if fatal_control.is_set():
+                    rec.status = "aborted"
+                    q.task_done()
+                    break
                 rec.queued_at = time.monotonic()
                 await q.put((url, attempt+1, rec))
                 q.task_done(); continue
 
-            if not success and not is_re:
+            if not success and not is_re and not fatal_control.is_set():
                 with lock: fail_count += 1
                 failed_urls.append(url)
                 rec.status = "fail"; mark_done()
@@ -262,33 +299,48 @@ async def _task(pu=proxy_url, co=cookies, fn=filename, ou=url, r=rec):
         await gate.aclose()
 
     async with tasks_lock:
-        stragglers = [t for t in all_tasks if not t.done()]
+        recorded_tasks = list(all_tasks)
+    stragglers = [t for t in recorded_tasks if not t.done()]
     if stragglers:
         print(f"  [wait] {len(stragglers)} downloads still finishing...")
-        await asyncio.gather(*stragglers, return_exceptions=True)
+    if recorded_tasks:
+        await asyncio.gather(*recorded_tasks, return_exceptions=True)
 
     stop_progress.set()
+    await progress_t
     await _close_sess()
     await close_ff_session()
     await _PROXY_POOL.close_all()
     telem.finish()
 
     base = os.path.dirname(os.path.abspath(__file__))
-    lp, jp = telem.save(base)
+    try:
+        lp, jp = telem.save(base)
+        print(f"Log: {os.path.basename(lp)}")
+    except (OSError, TypeError) as e:
+        # Report persistence is independent of the fatal download state.
+        print(f"[warn] Log save error: {e}")
 
     el = time.monotonic() - t0
     total_bytes = sum(b for _, b in bytes_acc)
+    disk_full = fatal_control.disk_full
     print(f"\n{'='*60}")
-    print(f"Done in {int(el//60)}m {int(el%60)}s  |  "
-          f"ok={ok_count}  fail={fail_count}  |  "
-          f"{total_bytes/1e9:.2f} GB  @  {total_bytes/el/1e6:.1f} MB/s")
-    print(f"Log: {os.path.basename(lp)}")
+    if disk_full is not None:
+        print(f"Run stopped: disk full in {disk_full.folder}  |  "
+              f"ok={ok_count}  fail={fail_count}")
+    else:
+        print(f"Done in {int(el//60)}m {int(el%60)}s  |  "
+              f"ok={ok_count}  fail={fail_count}  |  "
+              f"{total_bytes/1e9:.2f} GB  @  {total_bytes/el/1e6:.1f} MB/s")
 
     if failed_urls:
         fp = os.path.join(base, "failed_links.txt")
-        with open(fp, "w", encoding="utf-8") as f:
-            f.write("\n".join(failed_urls) + "\n")
-        print(f"Failed ({len(failed_urls)}): {fp}")
+        try:
+            with open(fp, "w", encoding="utf-8") as f:
+                f.write("\n".join(failed_urls) + "\n")
+            print(f"Failed ({len(failed_urls)}): {fp}")
+        except OSError as e:
+            print(f"[warn] Failed links save error: {e}")
 
 # ── ENTRY POINT ────────────────────────────────────────────────────────────────
 def main():
diff --git a/moon_download.py b/moon_download.py
index 2ee87ef..2f779e6 100644
--- a/moon_download.py
+++ b/moon_download.py
@@ -5,6 +5,7 @@
 import asyncio
 import collections
 import datetime
+import errno
 import io
 import json
 import os
@@ -305,7 +306,8 @@ def W(*parts):
         if ok_r:
             tb = sum(r.file_bytes for r in ok_r)
             W(f"Total    : {tb/1e9:.2f} GB  @  {tb/el/1e6:.1f} MB/s")
-        W(f"OK: {len(ok_r)}  /  Fail: {len(recs)-len(ok_r)}")
+        fail_r = [r for r in recs if r.status == "fail"]
+        W(f"OK: {len(ok_r)}  /  Fail: {len(fail_r)}")
         W()
         W(f"{'#':<4} {'Filename':<48} {'DL':>7} {'Speed':>10} {'Status'}")
         W("-" * 80)
@@ -326,7 +328,7 @@ def W(*parts):
                 "version": VERSION,
                 "duration_s": round(el, 2),
                 "ok": len(ok_r),
-                "fail": len(recs) - len(ok_r),
+                "fail": len(fail_r),
                 "files": [{k: getattr(r, k) for k in cli_fields} for r in recs],
             }, f, indent=2)
         return lp, jp
@@ -361,7 +363,8 @@ def W(*parts):
         W("── SUMMARY ─────────────────────────────────────────────────────────")
         W(f"  Total links    : {len(recs)}")
         W(f"  Completed OK   : {len(ok_r)}")
-        W(f"  Failed         : {len(recs)-len(ok_r)}")
+        fail_r = [r for r in recs if r.status == "fail"]
+        W(f"  Failed         : {len(fail_r)}")
         W(f"  Stall kills    : {sum(r.stall_kills for r in recs)}")
         if ok_r:
             tb = sum(r.file_bytes for r in ok_r)
@@ -427,7 +430,7 @@ def W(*parts):
                     "config": self.cfg,
                     "total": len(recs),
                     "ok": len(ok_r),
-                    "fail": len(recs) - len(ok_r),
+                    "fail": len(fail_r),
                     "stall_kills": sum(r.stall_kills for r in recs),
                     "median_dl_s": round(med, 2),
                 },
@@ -446,6 +449,38 @@ class _StallKill(Exception):
     pass
 
 
+@dataclass(frozen=True)
+class DiskFullError:
+    """The first ENOSPC observed during one front-end run."""
+
+    folder: str
+    needed_bytes: int
+
+
+class RunFatalControl:
+    """Share one terminal disk-full outcome without reusing stall-kill state."""
+
+    def __init__(self):
+        self._lock = threading.Lock()
+        self._disk_full: DiskFullError | None = None
+
+    def report_disk_full(self, folder: str, needed_bytes: int) -> DiskFullError | None:
+        """Keep the first failure so every peer agrees on the same run outcome."""
+        with self._lock:
+            if self._disk_full is not None:
+                return None
+            self._disk_full = DiskFullError(folder, max(1, needed_bytes))
+            return self._disk_full
+
+    @property
+    def disk_full(self) -> DiskFullError | None:
+        with self._lock:
+            return self._disk_full
+
+    def is_set(self) -> bool:
+        return self.disk_full is not None
+
+
 async def download_file(
     proxy_url: str,
     cookies: str,
@@ -456,10 +491,13 @@ async def download_file(
     kills_so_far: int,
     telem: Telemetry | None = None,
     on_event: Callable[[str, str], None] | None = None,
+    *,
+    fatal_control: RunFatalControl | None = None,
 ) -> tuple[bool, str, int]:
     """Download a single file with resume support, stall detection, and proxy rotation."""
     tmp = dest + ".tmp"
     loop = asyncio.get_running_loop()
+    fatal_control = fatal_control or RunFatalControl()
 
     def note(msg: str, tag: str = "warn") -> None:
         """Record a mid-transfer event in the report and surface it live if a front-end is listening."""
@@ -469,10 +507,21 @@ def note(msg: str, tag: str = "warn") -> None:
     detect = kills_so_far < STALL_MAX_KILL
 
     def _write(f, data: bytes):
-        f.write(data)
+        view = memoryview(data)
+        while view:
+            written = f.write(view)
+            if written is None or written <= 0:
+                raise OSError(errno.EIO, "download write made no progress")
+            view = view[written:]
 
     for att in range(DL_INNER_RETRIES):
+        if fatal_control.is_set():
+            return False, "aborted_disk_full", 0
         resume = os.path.getsize(tmp) if os.path.exists(tmp) else 0
+        file_size = 0
+        downloaded = resume
+        last_write_bytes = 0
+        write_persisted_before = resume
         ref = referer_for(proxy_url)
         hdrs = {
             "User-Agent": random.choice(USER_AGENTS),
@@ -498,7 +547,11 @@ def _write(f, data: bytes):
                 req_kwargs["proxy_auth"] = dl_proxy_auth
 
             async with dl_session.get(proxy_url, **req_kwargs) as r:
+                if fatal_control.is_set():
+                    return False, "aborted_disk_full", downloaded
                 if r.status == 416:
+                    if fatal_control.is_set():
+                        return False, "aborted_disk_full", downloaded
                     if os.path.exists(tmp):
                         os.replace(tmp, dest)
                     rec.file_bytes = os.path.getsize(dest) if os.path.exists(dest) else 0
@@ -514,7 +567,9 @@ def _write(f, data: bytes):
                     rec.file_bytes = file_size
 
                 effective_detect = detect and (file_size == 0 or file_size >= STALL_MIN_FILE_BYTES)
-                f = open(tmp, "ab" if resume > 0 else "wb")
+                # Each executor write must reach the filesystem before a peer can
+                # abort the run; buffered close could otherwise flush after ENOSPC.
+                f = open(tmp, "ab" if resume > 0 else "wb", buffering=0)
                 speed_win: collections.deque = collections.deque(maxlen=8000)
                 pub_win: collections.deque = collections.deque(maxlen=600)
                 last_pub = dl_t0
@@ -525,6 +580,8 @@ def _write(f, data: bytes):
                     buf: list[bytes] = []
                     bufsz = 0
                     async for chunk in r.content.iter_chunked(RECV_CHUNK):
+                        if fatal_control.is_set():
+                            return False, "aborted_disk_full", downloaded
                         if not chunk:
                             break
                         if kill_evt.is_set():
@@ -540,6 +597,10 @@ def _write(f, data: bytes):
                             data = b"".join(buf)
                             buf = []
                             bufsz = 0
+                            if fatal_control.is_set():
+                                return False, "aborted_disk_full", downloaded
+                            last_write_bytes = len(data)
+                            write_persisted_before = os.path.getsize(tmp)
                             await loop.run_in_executor(_POOL, _write, f, data)
 
                         elapsed = now - dl_t0
@@ -574,11 +635,18 @@ def _write(f, data: bytes):
                                         raise _StallKill()
 
                     if buf:
-                        bytes_acc.append((time.monotonic(), sum(len(b) for b in buf)))
-                        await loop.run_in_executor(_POOL, _write, f, b"".join(buf))
+                        data = b"".join(buf)
+                        if fatal_control.is_set():
+                            return False, "aborted_disk_full", downloaded
+                        bytes_acc.append((time.monotonic(), len(data)))
+                        last_write_bytes = len(data)
+                        write_persisted_before = os.path.getsize(tmp)
+                        await loop.run_in_executor(_POOL, _write, f, data)
                 finally:
                     f.close()
 
+            if fatal_control.is_set():
+                return False, "aborted_disk_full", downloaded
             os.replace(tmp, dest)
             dl_s = max(time.monotonic() - dl_t0, 0.001)
             net = downloaded - resume
@@ -589,20 +657,53 @@ def _write(f, data: bytes):
             return True, "ok", 0
 
         except _StallKill:
+            if fatal_control.is_set():
+                return False, "aborted_disk_full", downloaded
             return False, "stall_killed", downloaded
-        except (aiohttp.ClientPayloadError, aiohttp.ServerDisconnectedError):
-            note(f"connection dropped att {att+1}", "retry")
-            if att < DL_INNER_RETRIES - 1:
-                await asyncio.sleep(0.5 * (att + 1))
-                continue
-            return False, "connection dropped", downloaded
         except asyncio.TimeoutError:
+            if fatal_control.is_set():
+                return False, "aborted_disk_full", downloaded
             note(f"timeout att {att+1}", "retry")
             if att < DL_INNER_RETRIES - 1:
                 await asyncio.sleep(1 + att)
                 continue
             return False, "timeout", downloaded
+        except OSError as e:
+            if e.errno == errno.ENOSPC:
+                try:
+                    persisted = os.path.getsize(tmp)
+                except OSError:
+                    persisted = 0
+                if file_size > 0:
+                    needed = max(file_size - persisted, 1)
+                else:
+                    persisted_in_flush = max(persisted - write_persisted_before, 0)
+                    needed = max(last_write_bytes - persisted_in_flush, 1)
+                disk_full = fatal_control.report_disk_full(
+                    os.path.dirname(os.path.abspath(dest)), needed)
+                if disk_full is not None:
+                    note(
+                        f"Disk full in {disk_full.folder}: need {disk_full.needed_bytes:,} bytes to continue",
+                        "fail",
+                    )
+                    return False, "disk_full", downloaded
+                return False, "aborted_disk_full", downloaded
+            if fatal_control.is_set():
+                return False, "aborted_disk_full", downloaded
+            err = str(e)
+            note(f"error att {att+1}: {err}", "retry")
+            return False, err, downloaded
+        except (aiohttp.ClientPayloadError, aiohttp.ServerDisconnectedError):
+            if fatal_control.is_set():
+                return False, "aborted_disk_full", downloaded
+            note(f"connection dropped att {att+1}", "retry")
+            if att < DL_INNER_RETRIES - 1:
+                await asyncio.sleep(0.5 * (att + 1))
+                continue
+            return False, "connection dropped", downloaded
         except Exception as e:
+            if fatal_control.is_set():
+                return False, "aborted_disk_full", downloaded
             err = str(e)
             note(f"error att {att+1}: {err}", "retry")
             if att < DL_INNER_RETRIES - 1 and ("ContentLengthError" in err or "not enough data" in err.lower()):
diff --git a/moon_engine.py b/moon_engine.py
index 306ec6b..dc02e39 100644
--- a/moon_engine.py
+++ b/moon_engine.py
@@ -41,6 +41,7 @@
     _sanitize_filename,
     download_file,
     count_usable_proxies,
+    RunFatalControl,
 )
 
 # ── THEME ──────────────────────────────────────────────────────────────────────
@@ -161,70 +162,82 @@ def log(self, msg, tag=""):
 
     async def _do_dl(self, proxy_url, cookies, filename, orig_url, rec,
                      kill_counts, dl_sem, dest_folder, telem, mark_done_fn,
-                     failed_urls, q):
+                     failed_urls, q, fatal_control):
         async with dl_sem:
+            if fatal_control.is_set():
+                rec.status = "aborted"
+                return
             self._inc("_dls")
-            rec.dl_start = time.monotonic(); rec.status = "downloading"
-            self._track(rec)
-            dest = os.path.join(dest_folder, filename)
-
-            if os.path.exists(dest):
-                self._inc("_ok")
-                self.log(f"    ✓  Exists: {filename}", "ok")
-                rec.status="ok"; rec.dl_s=0.0
-                mark_done_fn(); self._inc("_dl_done"); self._inc("_dls",-1); return
-
-            kc       = kill_counts.get(orig_url, 0)
-            kill_evt = asyncio.Event()
-            with self._lock:
-                self._active_kill_events.add(kill_evt)
             try:
-                ok, msg, bytes_done = await download_file(
-                    proxy_url, cookies, dest, rec, self._bytes_acc, kill_evt, kc,
-                    telem=telem, on_event=self.log)
-            finally:
+                rec.dl_start = time.monotonic(); rec.status = "downloading"
+                self._track(rec)
+                dest = os.path.join(dest_folder, filename)
+
+                if os.path.exists(dest):
+                    self._inc("_ok")
+                    self.log(f"    ✓  Exists: {filename}", "ok")
+                    rec.status="ok"; rec.dl_s=0.0
+                    mark_done_fn(); self._inc("_dl_done"); return
+
+                kc       = kill_counts.get(orig_url, 0)
+                kill_evt = asyncio.Event()
                 with self._lock:
-                    self._active_kill_events.discard(kill_evt)
-            rec.dl_s = max(time.monotonic()-rec.dl_start, 0.001)
-
-            if ok:
-                self._inc("_ok")
-                spd = f"  ({rec.avg_mbs:.1f} MB/s)" if rec.avg_mbs > 0 else ""
-                self.log(f"    ✓  Saved: {filename}{spd}", "ok")
-                rec.status="ok"; mark_done_fn(); self._inc("_dl_done")
-            elif msg == "stall_killed" and self._get("_stop_flag"):
-                rec.status = "stopped"
-                self.log(f"    stop: {filename} interrupted; partial file kept", "warn")
-                mark_done_fn(); self._inc("_dl_done")
-            elif msg == "stall_killed":
-                done_mb = bytes_done//(1<<20)
-                new_kc = kc + 1; kill_counts[orig_url] = new_kc
-                self._inc("_kills"); rec.stall_kills += 1
-                if new_kc <= STALL_MAX_KILL:
-                    self.log(f"    ⚡  Kill #{new_kc}: {filename}  ({done_mb}MB) → re-extract", "kill")
+                    self._active_kill_events.add(kill_evt)
+                try:
+                    ok, msg, bytes_done = await download_file(
+                        proxy_url, cookies, dest, rec, self._bytes_acc, kill_evt, kc,
+                        telem=telem, on_event=self.log, fatal_control=fatal_control)
+                finally:
+                    with self._lock:
+                        self._active_kill_events.discard(kill_evt)
+                rec.dl_s = max(time.monotonic()-rec.dl_start, 0.001)
+
+                if ok:
+                    self._inc("_ok")
+                    spd = f"  ({rec.avg_mbs:.1f} MB/s)" if rec.avg_mbs > 0 else ""
+                    self.log(f"    ✓  Saved: {filename}{spd}", "ok")
+                    rec.status="ok"; mark_done_fn(); self._inc("_dl_done")
+                elif msg == "stall_killed" and self._get("_stop_flag"):
+                    rec.status = "stopped"
+                    self.log(f"    stop: {filename} interrupted; partial file kept", "warn")
+                    mark_done_fn(); self._inc("_dl_done")
+                elif msg == "aborted_disk_full":
+                    rec.status = "aborted"
+                elif msg == "stall_killed":
+                    done_mb = bytes_done//(1<<20)
+                    new_kc = kc + 1; kill_counts[orig_url] = new_kc
+                    self._inc("_kills"); rec.stall_kills += 1
+                    if new_kc <= STALL_MAX_KILL:
+                        self.log(f"    ⚡  Kill #{new_kc}: {filename}  ({done_mb}MB) → re-extract", "kill")
+                    else:
+                        self.log(f"    ⚡  Kill #{new_kc}: {filename}  ({done_mb}MB) → continue", "warn")
+                    rec.queued_at=time.monotonic(); rec.status="pending"
+                    if not fatal_control.is_set():
+                        await q.put((orig_url, 1, rec))
                 else:
-                    self.log(f"    ⚡  Kill #{new_kc}: {filename}  ({done_mb}MB) → continue", "warn")
-                rec.queued_at=time.monotonic(); rec.status="pending"
-                await q.put((orig_url, 1, rec))
-                self._inc("_dls",-1); return
-            else:
-                self._inc("_fail"); failed_urls.append(orig_url)
-                rec.status="fail"; rec.error=msg
-                self.log(f"    ✗  {filename}: {msg}", "fail")
-                mark_done_fn(); self._inc("_dl_done")
-
-            self._inc("_dls",-1)
+                    self._inc("_fail"); failed_urls.append(orig_url)
+                    rec.status="fail"; rec.error=msg
+                    if msg != "disk_full":
+                        self.log(f"    ✗  {filename}: {msg}", "fail")
+                    mark_done_fn(); self._inc("_dl_done")
+            finally:
+                self._inc("_dls",-1)
 
     async def _browser_worker(self, get_browser, wid, q, dl_sem, all_done, mark_done_fn,
                                kill_counts, all_tasks, tasks_lock,
-                               output_links, failed_urls, dest_folder, mode, max_retries, telem):
+                               output_links, failed_urls, dest_folder, mode, max_retries, telem,
+                               fatal_control):
         my_tasks = []
         try:
-            while not self._get("_stop_flag"):
+            while not self._get("_stop_flag") and not fatal_control.is_set():
                 if all_done.is_set() and q.empty(): break
                 try:
                     url, attempt, rec = await asyncio.wait_for(q.get(), timeout=1.0)
                 except asyncio.TimeoutError: continue
+                if fatal_control.is_set():
+                    rec.status = "aborted"
+                    q.task_done()
+                    break
 
                 rec.worker_id    = wid
                 t_start          = time.monotonic()
@@ -247,6 +260,10 @@ async def _browser_worker(self, get_browser, wid, q, dl_sem, all_done, mark_done
                         # reaches for a window if /go refuses it.
                         link = await extract_fuckingfast(url, get_browser)
                         rec.extract_s = time.monotonic()-t_start
+                        if fatal_control.is_set():
+                            rec.status = "aborted"
+                            q.task_done()
+                            break
                         if not link:
                             self.log("    ✗  No link found", "fail")
                         elif mode == "links":
@@ -258,7 +275,7 @@ async def _browser_worker(self, get_browser, wid, q, dl_sem, all_done, mark_done
                             async def _task(pu=link, fn=filename, ou=url, r=rec):
                                 await self._do_dl(pu, "", fn, ou, r, kill_counts,
                                                    dl_sem, dest_folder, telem, mark_done_fn,
-                                                   failed_urls, q)
+                                                   failed_urls, q, fatal_control)
                             t = asyncio.create_task(_task())
                             my_tasks.append(t)
                             async with tasks_lock: all_tasks.append(t)
@@ -272,6 +289,10 @@ async def _task(pu=link, fn=filename, ou=url, r=rec):
                         # the persistent lane pool internally.
                         proxy_url, cookies = await extract_datanodes(await get_browser(), url)
                         rec.extract_s = time.monotonic()-t_start
+                        if fatal_control.is_set():
+                            rec.status = "aborted"
+                            q.task_done()
+                            break
                         if not proxy_url:
                             rec.notes.append("extraction failed")
                             self.log("    ✗  No URL extracted", "fail")
@@ -284,7 +305,7 @@ async def _task(pu=link, fn=filename, ou=url, r=rec):
                             async def _task(pu=proxy_url, co=cookies, fn=filename, ou=url, r=rec):
                                 await self._do_dl(pu, co, fn, ou, r, kill_counts,
                                                    dl_sem, dest_folder, telem, mark_done_fn,
-                                                   failed_urls, q)
+                                                   failed_urls, q, fatal_control)
                             t = asyncio.create_task(_task())
                             my_tasks.append(t)
                             async with tasks_lock: all_tasks.append(t)
@@ -304,16 +325,25 @@ async def _task(pu=proxy_url, co=cookies, fn=filename, ou=url, r=rec):
                     rec.notes.append(f"exception: {e}")
                     self.log(f"    ✗  {e}", "fail")
 
+                if fatal_control.is_set():
+                    rec.status = "aborted"
+                    q.task_done()
+                    break
+
                 if (not success and not unsupported and not is_re and attempt < max_retries
-                        and not self._get("_stop_flag")):
+                        and not self._get("_stop_flag") and not fatal_control.is_set()):
                     backoff = min(2**(attempt-1), 6)
                     self.log(f"    ↻  retry in {backoff}s", "warn")
                     await asyncio.sleep(backoff)
+                    if fatal_control.is_set():
+                        rec.status = "aborted"
+                        q.task_done()
+                        break
                     rec.queued_at = time.monotonic()
                     await q.put((url, attempt+1, rec))
                     q.task_done(); continue
 
-                if not success and not is_re:
+                if not success and not is_re and not fatal_control.is_set():
                     self._inc("_fail"); failed_urls.append(url)
                     rec.status="fail"; mark_done_fn()
 
@@ -333,6 +363,7 @@ async def _run(self, urls, n_workers, max_dl, max_retries):
         all_tasks    : list      = []
         tasks_lock   = asyncio.Lock()
         kill_counts  : dict[str,int] = {}
+        fatal_control = RunFatalControl()
         dest_folder  = self._cfg["out_folder"]
         mode         = self._cfg["mode"]
         n_done       = 0
@@ -368,7 +399,10 @@ async def snap_task():
                 with self._lock:
                     b, d, ok, fail = self._browsers, self._dls, self._ok, self._fail
                 telem.snap(b, d, q.qsize(), ok, fail)
-                await asyncio.sleep(1.0)
+                try:
+                    await asyncio.wait_for(snap_stop.wait(), timeout=1.0)
+                except asyncio.TimeoutError:
+                    pass
 
         # Kept in a local: a task with no reference can be collected mid-run.
         snap_t = asyncio.create_task(snap_task())  # noqa: F841
@@ -392,7 +426,8 @@ async def _launch(wid):
             await self._browser_worker(
                 gate.get, wid, q, dl_sem, all_done, mark_done,
                 kill_counts, all_tasks, tasks_lock,
-                output_links, failed_urls, dest_folder, mode, max_retries, telem)
+                output_links, failed_urls, dest_folder, mode, max_retries, telem,
+                fatal_control)
 
         try:
             worker_results = await asyncio.gather(
@@ -414,6 +449,7 @@ async def _launch(wid):
             await asyncio.gather(*stragglers, return_exceptions=True)
 
         snap_stop.set()
+        await snap_t
         await _close_sess()
         await close_ff_session()
         await _PROXY_POOL.close_all()
@@ -430,18 +466,29 @@ async def _launch(wid):
             # downloads already completed; a report-write failure shouldn't crash finalize.
             self.log(f"⚠  Log save error: {e}", "warn")
 
-        if output_links and mode == "links":
+        if output_links and mode == "links" and not fatal_control.is_set():
             with open(os.path.join(base,"output_links.txt"),"w",encoding="utf-8") as f:
                 f.write("\n".join(output_links)+"\n")
             self.log("✓  Links → output_links.txt", "info")
         if failed_urls:
-            with open(os.path.join(base,"failed_links.txt"),"w",encoding="utf-8") as f:
-                f.write("\n".join(failed_urls)+"\n")
-            self.log(f"⚠  {len(failed_urls)} failed → failed_links.txt", "warn")
+            try:
+                with open(os.path.join(base,"failed_links.txt"),"w",encoding="utf-8") as f:
+                    f.write("\n".join(failed_urls)+"\n")
+                self.log(f"⚠  {len(failed_urls)} failed → failed_links.txt", "warn")
+            except OSError as e:
+                self.log(f"⚠  Failed links save error: {e}", "warn")
 
         el = time.monotonic()-t0; m, s = divmod(int(el), 60)
         with self._lock: ok, fail, kills = self._ok, self._fail, self._kills
-        self.log(f"\n✓  Done in {m}m {s}s  ·  ✓ {ok}  ✗ {fail}  ⚡ {kills} kills", "ok")
+        disk_full = fatal_control.disk_full
+        if disk_full is not None:
+            self.log(
+                f"\nRun stopped: disk full in {disk_full.folder} after {m}m {s}s  ·  "
+                f"✓ {ok}  ✗ {fail}",
+                "fail",
+            )
+        else:
+            self.log(f"\n✓  Done in {m}m {s}s  ·  ✓ {ok}  ✗ {fail}  ⚡ {kills} kills", "ok")
         self._on_done()
 
     def _on_done(self):
@@ -472,6 +519,7 @@ def scan_tmp(self) -> int:
         "pending":     "queue",
         "extracting":  "extract",
         "downloading": "download",
+        "aborted":     "queue",
         "ok":          "ok",
         "fail":        "fail",
         # A user-initiated stop is not a failure: the partial .tmp is kept and the
@@ -690,6 +738,8 @@ def snapshot(self, cursor: int = 0) -> dict:
         # so the engine ships numbers and a stage name instead of prose.
         if not running and state == "idle":
             stage = "idle"
+        elif not running and state == "done":
+            stage = "done"
         elif url_done < url_tot:
             stage = "extracting"
         elif dl_done < dl_tot:
diff --git a/tests/test_disk_full.py b/tests/test_disk_full.py
new file mode 100644
index 0000000..3ee49ef
--- /dev/null
+++ b/tests/test_disk_full.py
@@ -0,0 +1,659 @@
+"""Regression coverage for run-level ENOSPC handling (#116)."""
+from __future__ import annotations
+
+import asyncio
+import builtins
+import collections
+import errno
+import json
+import os
+
+import moon_cli
+import moon_download
+import moon_engine
+from moon_download import FileRecord, RunFatalControl, download_file
+
+
+class _Chunks:
+    def __init__(self, chunks):
+        self._chunks = iter(chunks)
+
+    def __aiter__(self):
+        return self
+
+    async def __anext__(self):
+        try:
+            return next(self._chunks)
+        except StopIteration as exc:
+            raise StopAsyncIteration from exc
+
+
+class _Response:
+    def __init__(self, status=206, content_length=90, chunks=(b"x" * 20,)):
+        self.status = status
+        self.headers = {"Content-Length": str(content_length)}
+        self.content = type("Content", (), {
+            "iter_chunked": lambda _self, _size: _Chunks(chunks),
+        })()
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *exc):
+        return False
+
+
+class _Session:
+    def __init__(self, response):
+        self._response = response
+
+    def get(self, *_args, **_kwargs):
+        return self._response
+
+
+class _NoSpaceFile:
+    def write(self, _data):
+        raise OSError(errno.ENOSPC, "No space left on device")
+
+    def close(self):
+        pass
+
+
+def test_download_file_reports_errno_enospc_and_preserves_tmp(monkeypatch, tmp_path):
+    """A write ENOSPC creates the shared fatal reason and keeps the partial file."""
+    dest = tmp_path / "file.bin"
+    partial = dest.with_suffix(".bin.tmp")
+    partial.write_bytes(b"p" * 10)
+
+    def no_space_open(path, mode="r", *args, **kwargs):
+        assert os.fspath(path) == os.fspath(partial)
+        assert mode == "ab"
+        assert kwargs["buffering"] == 0
+        return _NoSpaceFile()
+
+    monkeypatch.setattr(moon_download, "open", no_space_open, raising=False)
+    monkeypatch.setattr(moon_download._PROXY_POOL, "next", lambda: None)
+    monkeypatch.setattr(moon_download, "_sess", lambda: _Session(_Response()))
+    control = RunFatalControl()
+    events = []
+    rec = FileRecord("https://example.invalid/one", "file.bin")
+
+    ok, result, _ = asyncio.run(download_file(
+        "https://download.invalid/one", "", str(dest), rec, collections.deque(),
+        asyncio.Event(), 0, fatal_control=control,
+        on_event=lambda message, tag: events.append((message, tag)),
+    ))
+
+    assert (ok, result) == (False, "disk_full")
+    assert partial.read_bytes() == b"p" * 10
+    assert control.disk_full is not None
+    assert control.disk_full.folder == str(tmp_path)
+    assert control.disk_full.needed_bytes == 90
+    disk_events = [message for message, _tag in events if message.startswith("Disk full")]
+    assert len(disk_events) == 1
+    assert str(tmp_path) in disk_events[0]
+    assert "90" in disk_events[0]
+
+    peer = FileRecord("https://example.invalid/two", "two.bin")
+    peer_result = asyncio.run(download_file(
+        "https://download.invalid/two", "", str(tmp_path / "two.bin"), peer,
+        collections.deque(), asyncio.Event(), 0, fatal_control=control,
+    ))
+    assert peer_result == (False, "aborted_disk_full", 0)
+
+
+def test_timeout_retries_before_returning_timeout(monkeypatch, tmp_path):
+    """TimeoutError stays on the existing retry path despite inheriting OSError."""
+    attempts = []
+
+    class TimeoutResponse:
+        async def __aenter__(self):
+            attempts.append("request")
+            raise asyncio.TimeoutError()
+
+        async def __aexit__(self, *exc):
+            return False
+
+    async def no_sleep(_seconds):
+        pass
+
+    monkeypatch.setattr(moon_download, "DL_INNER_RETRIES", 2)
+    monkeypatch.setattr(moon_download._PROXY_POOL, "next", lambda: None)
+    monkeypatch.setattr(moon_download, "_sess", lambda: _Session(TimeoutResponse()))
+    monkeypatch.setattr(moon_download.asyncio, "sleep", no_sleep)
+    events = []
+
+    result = asyncio.run(download_file(
+        "https://download.invalid/timeout", "", str(tmp_path / "timeout.bin"),
+        FileRecord("timeout", "timeout.bin"), collections.deque(), asyncio.Event(), 0,
+        on_event=lambda message, _tag: events.append(message),
+    ))
+
+    assert result == (False, "timeout", 0)
+    assert attempts == ["request", "request"]
+    assert events == ["timeout att 1", "timeout att 2"]
+
+
+def test_short_unbuffered_writes_persist_the_full_payload(monkeypatch, tmp_path):
+    """A partial FileIO.write result is retried until the chunk is complete."""
+    dest = tmp_path / "short.bin"
+    writes = []
+    original_open = builtins.open
+
+    class ShortFile:
+        def __init__(self, raw):
+            self._raw = raw
+
+        def write(self, data):
+            data = bytes(data)
+            writes.append(data)
+            return self._raw.write(data[:2])
+
+        def close(self):
+            self._raw.close()
+
+    def short_open(path, mode="r", *args, **kwargs):
+        assert os.fspath(path) == f"{dest}.tmp"
+        assert mode == "wb"
+        assert kwargs["buffering"] == 0
+        return ShortFile(original_open(path, mode, *args, **kwargs))
+
+    monkeypatch.setattr(moon_download, "WRITE_BUF", 1)
+    monkeypatch.setattr(moon_download, "open", short_open, raising=False)
+    monkeypatch.setattr(moon_download._PROXY_POOL, "next", lambda: None)
+    monkeypatch.setattr(
+        moon_download, "_sess", lambda: _Session(_Response(status=200, content_length=6, chunks=(b"abcdef",))),
+    )
+
+    result = asyncio.run(download_file(
+        "https://download.invalid/short", "", str(dest),
+        FileRecord("short", "short.bin"), collections.deque(), asyncio.Event(), 0,
+    ))
+
+    assert result[0] is True
+    assert writes == [b"abcdef", b"cdef", b"ef"]
+    assert dest.read_bytes() == b"abcdef"
+
+
+def test_short_write_then_enospc_reports_actual_remaining_bytes(monkeypatch, tmp_path):
+    """Known content length wins over the original buffer when a write is partial."""
+    dest = tmp_path / "partial.bin"
+    original_open = builtins.open
+
+    class PartialThenFull:
+        def __init__(self, raw):
+            self._raw = raw
+            self._writes = 0
+
+        def write(self, data):
+            self._writes += 1
+            if self._writes == 1:
+                return self._raw.write(bytes(data)[:2])
+            raise OSError(errno.ENOSPC, "No space left on device")
+
+        def close(self):
+            self._raw.close()
+
+    def partial_open(path, mode="r", *args, **kwargs):
+        assert os.fspath(path) == f"{dest}.tmp"
+        assert mode == "wb"
+        assert kwargs["buffering"] == 0
+        return PartialThenFull(original_open(path, mode, *args, **kwargs))
+
+    control = RunFatalControl()
+    monkeypatch.setattr(moon_download, "WRITE_BUF", 1)
+    monkeypatch.setattr(moon_download, "open", partial_open, raising=False)
+    monkeypatch.setattr(moon_download._PROXY_POOL, "next", lambda: None)
+    monkeypatch.setattr(
+        moon_download, "_sess", lambda: _Session(_Response(status=200, content_length=6, chunks=(b"abcdef",))),
+    )
+
+    result = asyncio.run(download_file(
+        "https://download.invalid/partial", "", str(dest),
+        FileRecord("partial", "partial.bin"), collections.deque(), asyncio.Event(), 0,
+        fatal_control=control,
+    ))
+
+    assert result[1] == "disk_full"
+    assert (tmp_path / "partial.bin.tmp").read_bytes() == b"ab"
+    assert control.disk_full is not None
+    assert control.disk_full.needed_bytes == 4
+
+
+def test_active_peer_unwinds_after_shared_enospc(monkeypatch, tmp_path):
+    """Two live streams share the first ENOSPC without peer writes after it."""
+    monkeypatch.setattr(moon_download, "WRITE_BUF", 1)
+    monkeypatch.setattr(moon_download._PROXY_POOL, "next", lambda: None)
+    control = RunFatalControl()
+    entered = set()
+    both_streaming = asyncio.Event()
+    writes = {"bad": [], "peer": []}
+    original_open = builtins.open
+    bad_tmp = tmp_path / "bad.bin.tmp"
+    peer_tmp = tmp_path / "peer.bin.tmp"
+    bad_tmp.write_bytes(b"bad")
+    peer_tmp.write_bytes(b"peer")
+
+    class CoordinatedContent:
+        def __init__(self, name):
+            self._name = name
+
+        async def iter_chunked(self, _size):
+            entered.add(self._name)
+            if len(entered) == 2:
+                both_streaming.set()
+            await both_streaming.wait()
+            if self._name == "bad":
+                yield b"boom"
+            else:
+                while not control.is_set():
+                    await asyncio.sleep(0)
+                yield b"peer"
+
+    class CoordinatedResponse:
+        status = 206
+        headers = {"Content-Length": "10"}
+
+        def __init__(self, name):
+            self.content = CoordinatedContent(name)
+
+        async def __aenter__(self):
+            return self
+
+        async def __aexit__(self, *exc):
+            return False
+
+    class CoordinatedSession:
+        def get(self, url, **_kwargs):
+            return CoordinatedResponse("bad" if url.endswith("bad") else "peer")
+
+    class TrackingFile:
+        def __init__(self, name, raw):
+            self._name = name
+            self._raw = raw
+
+        def write(self, data):
+            writes[self._name].append(bytes(data))
+            if self._name == "bad":
+                raise OSError(errno.ENOSPC, "No space left on device")
+            return self._raw.write(data)
+
+        def close(self):
+            self._raw.close()
+
+    def tracked_open(path, mode="r", *args, **kwargs):
+        path = os.fspath(path)
+        assert mode == "ab"
+        assert kwargs["buffering"] == 0
+        if path == os.fspath(bad_tmp):
+            return TrackingFile("bad", original_open(path, mode, *args, **kwargs))
+        if path == os.fspath(peer_tmp):
+            return TrackingFile("peer", original_open(path, mode, *args, **kwargs))
+        raise AssertionError(f"unexpected file open: {path}")
+
+    monkeypatch.setattr(moon_download, "open", tracked_open, raising=False)
+    monkeypatch.setattr(moon_download, "_sess", lambda: CoordinatedSession())
+    events = []
+    async def run_both():
+        return await asyncio.gather(
+            download_file("https://download.invalid/bad", "", str(tmp_path / "bad.bin"),
+                          FileRecord("bad", "bad.bin"), collections.deque(), asyncio.Event(), 0,
+                          fatal_control=control, on_event=lambda message, _tag: events.append(message)),
+            download_file("https://download.invalid/peer", "", str(tmp_path / "peer.bin"),
+                          FileRecord("peer", "peer.bin"), collections.deque(), asyncio.Event(), 0,
+                          fatal_control=control),
+        )
+
+    results = asyncio.run(run_both())
+
+    assert results[0][1] == "disk_full"
+    assert results[1] == (False, "aborted_disk_full", 4)
+    assert entered == {"bad", "peer"}
+    assert writes == {"bad": [b"boom"], "peer": []}
+    assert bad_tmp.read_bytes() == b"bad"
+    assert peer_tmp.read_bytes() == b"peer"
+    assert [message for message in events if message.startswith("Disk full")] == [
+        f"Disk full in {tmp_path}: need 10 bytes to continue",
+    ]
+    assert control.disk_full is not None
+    assert control.disk_full.folder == str(tmp_path)
+    assert control.disk_full.needed_bytes == 10
+
+
+def test_network_exception_after_peer_enospc_is_cooperative_abort(monkeypatch, tmp_path):
+    """A peer exception after ENOSPC must not become a second ordinary failure."""
+    monkeypatch.setattr(moon_download, "WRITE_BUF", 1)
+    monkeypatch.setattr(moon_download._PROXY_POOL, "next", lambda: None)
+    control = RunFatalControl()
+    entered = set()
+    both_streaming = asyncio.Event()
+    original_open = builtins.open
+    writes = {"bad": [], "peer": []}
+
+    class RaceContent:
+        def __init__(self, name):
+            self._name = name
+
+        async def iter_chunked(self, _size):
+            entered.add(self._name)
+            if len(entered) == 2:
+                both_streaming.set()
+            await both_streaming.wait()
+            if self._name == "bad":
+                yield b"boom"
+            else:
+                while not control.is_set():
+                    await asyncio.sleep(0)
+                raise moon_download.aiohttp.ClientPayloadError("peer dropped")
+
+    class RaceResponse:
+        status = 200
+        headers = {"Content-Length": "6"}
+
+        def __init__(self, name):
+            self.content = RaceContent(name)
+
+        async def __aenter__(self):
+            return self
+
+        async def __aexit__(self, *exc):
+            return False
+
+    class RaceSession:
+        def get(self, url, **_kwargs):
+            return RaceResponse("bad" if url.endswith("bad") else "peer")
+
+    class RaceFile:
+        def __init__(self, name, raw):
+            self._name = name
+            self._raw = raw
+
+        def write(self, data):
+            writes[self._name].append(bytes(data))
+            if self._name == "bad":
+                raise OSError(errno.ENOSPC, "No space left on device")
+            return self._raw.write(data)
+
+        def close(self):
+            self._raw.close()
+
+    def race_open(path, mode="r", *args, **kwargs):
+        path = os.fspath(path)
+        assert mode == "wb"
+        assert kwargs["buffering"] == 0
+        name = "bad" if path.endswith("bad.bin.tmp") else "peer"
+        return RaceFile(name, original_open(path, mode, *args, **kwargs))
+
+    monkeypatch.setattr(moon_download, "open", race_open, raising=False)
+    monkeypatch.setattr(moon_download, "_sess", lambda: RaceSession())
+    events = []
+
+    async def run_both():
+        return await asyncio.gather(
+            download_file("https://download.invalid/bad", "", str(tmp_path / "bad.bin"),
+                          FileRecord("bad", "bad.bin"), collections.deque(), asyncio.Event(), 0,
+                          fatal_control=control, on_event=lambda message, _tag: events.append(message)),
+            download_file("https://download.invalid/peer", "", str(tmp_path / "peer.bin"),
+                          FileRecord("peer", "peer.bin"), collections.deque(), asyncio.Event(), 0,
+                          fatal_control=control),
+        )
+
+    results = asyncio.run(run_both())
+
+    assert results[0][1] == "disk_full"
+    assert results[1] == (False, "aborted_disk_full", 0)
+    assert entered == {"bad", "peer"}
+    assert writes == {"bad": [b"boom"], "peer": []}
+    assert [message for message in events if message.startswith("Disk full")] == [
+        f"Disk full in {tmp_path}: need 6 bytes to continue",
+    ]
+
+
+class _Gate:
+    def __init__(self, *_args, **_kwargs):
+        pass
+
+    async def get(self):
+        return None
+
+    async def aclose(self):
+        pass
+
+
+async def _fake_extract(url, _get_browser):
+    return f"https://download.invalid/{url.rsplit('/', 1)[-1]}"
+
+
+async def _noop():
+    pass
+
+
+def _fatal_download(events, calls, completed):
+    active = set()
+    both_active = asyncio.Event()
+
+    async def fake_download(_proxy_url, _cookies, dest, rec, _bytes_acc, _kill_evt,
+                            _kills_so_far, telem=None, on_event=None, *, fatal_control=None):
+        calls.append(rec.url)
+        os.makedirs(os.path.dirname(dest), exist_ok=True)
+        with builtins.open(dest + ".tmp", "wb") as f:
+            f.write(rec.filename.encode())
+        active.add(rec.url)
+        if len(active) == 2:
+            both_active.set()
+        await both_active.wait()
+        if rec.url.endswith("file-one.bin"):
+            await asyncio.sleep(0)
+            disk_full = fatal_control.report_disk_full(os.path.dirname(dest), 77)
+            message = f"Disk full in {disk_full.folder}: need {disk_full.needed_bytes} bytes to continue"
+            events.append(message)
+            if on_event:
+                on_event(message, "fail")
+            completed.append(rec.url)
+            return False, "disk_full", 0
+        while not fatal_control.is_set():
+            await asyncio.sleep(0)
+        completed.append(rec.url)
+        return False, "aborted_disk_full", 0
+
+    return fake_download
+
+
+def _patch_frontend(monkeypatch, module, tmp_path, events, calls, completed):
+    monkeypatch.setattr(module, "BrowserGate", _Gate)
+    monkeypatch.setattr(module, "extract_fuckingfast", _fake_extract)
+    monkeypatch.setattr(module, "download_file", _fatal_download(events, calls, completed))
+    monkeypatch.setattr(module, "_close_sess", _noop)
+    monkeypatch.setattr(module, "close_ff_session", _noop)
+    monkeypatch.setattr(module._PROXY_POOL, "close_all", _noop)
+    monkeypatch.setattr(module, "__file__", str(tmp_path / f"{module.__name__}.py"))
+
+
+def _reject_retry_file(monkeypatch, module):
+    original_open = builtins.open
+
+    def retry_file_full(path, mode="r", *args, **kwargs):
+        if os.path.basename(os.fspath(path)) == "failed_links.txt" and "w" in mode:
+            raise OSError(errno.ENOSPC, "No space left on device")
+        return original_open(path, mode, *args, **kwargs)
+
+    monkeypatch.setattr(module, "open", retry_file_full, raising=False)
+
+
+def _frontend_urls():
+    return [
+        "https://fuckingfast.co/one/file-one.bin",
+        "https://fuckingfast.co/two/file-two.bin",
+        "https://fuckingfast.co/three/file-three.bin",
+    ]
+
+
+def test_engine_stops_queue_and_waits_for_downloads(monkeypatch, tmp_path):
+    events, calls, completed = [], [], []
+    _patch_frontend(monkeypatch, moon_engine, tmp_path, events, calls, completed)
+    engine = moon_engine.Engine()
+    engine._cfg["out_folder"] = str(tmp_path / "downloads")
+    urls = _frontend_urls()
+
+    asyncio.run(engine._run(urls, 2, 2, 2))
+
+    failed = (tmp_path / "failed_links.txt").read_text(encoding="utf-8").splitlines()
+    snapshot = engine.snapshot(0)
+    assert set(calls) == set(urls[:2])
+    assert urls[2] not in calls
+    assert set(completed) == set(urls[:2])
+    assert failed == [urls[0]]
+    assert snapshot["metrics"]["fail"] == 1
+    assert snapshot["metrics"]["active"] == 0
+    assert snapshot["metrics"]["stage"] == "done"
+    assert snapshot["state"] == "done"
+    assert len(events) == 1 and str(tmp_path / "downloads") in events[0]
+    assert (tmp_path / "downloads" / "file-one.bin.tmp").exists()
+    assert (tmp_path / "downloads" / "file-two.bin.tmp").exists()
+    report = json.loads(next(tmp_path.glob("moontech_*.json")).read_text(encoding="utf-8"))
+    assert report["session"]["fail"] == 1
+    assert any(rec.status == "aborted" for rec in engine._tracked.values())
+    assert not any(file["state"] in ("extract", "download") for file in snapshot["files"])
+    assert not any("✓  Done" in message for message, _tag in snapshot["log"])
+
+
+def test_cli_stops_queue_and_records_only_triggering_url(monkeypatch, tmp_path, capsys):
+    events, calls, completed = [], [], []
+    _patch_frontend(monkeypatch, moon_cli, tmp_path, events, calls, completed)
+    urls = _frontend_urls()
+
+    asyncio.run(moon_cli.run(urls, str(tmp_path / "downloads"), 2, 2, 2, "proxies.txt"))
+
+    failed = (tmp_path / "failed_links.txt").read_text(encoding="utf-8").splitlines()
+    output = capsys.readouterr().out
+    assert set(calls) == set(urls[:2])
+    assert urls[2] not in calls
+    assert set(completed) == set(urls[:2])
+    assert failed == [urls[0]]
+    assert len(events) == 1 and str(tmp_path / "downloads") in events[0]
+    assert (tmp_path / "downloads" / "file-one.bin.tmp").exists()
+    assert (tmp_path / "downloads" / "file-two.bin.tmp").exists()
+    report = json.loads(next(tmp_path.glob("moontech_cli_*.json")).read_text(encoding="utf-8"))
+    assert report["fail"] == 1
+    assert "Run stopped: disk full" in output
+    assert "Done in" not in output
+
+
+def test_engine_retry_file_enospc_keeps_disk_full_outcome(monkeypatch, tmp_path):
+    events, calls, completed = [], [], []
+    _patch_frontend(monkeypatch, moon_engine, tmp_path, events, calls, completed)
+    _reject_retry_file(monkeypatch, moon_engine)
+    engine = moon_engine.Engine()
+    engine._cfg["out_folder"] = str(tmp_path / "downloads")
+
+    asyncio.run(engine._run(_frontend_urls(), 2, 2, 2))
+
+    snapshot = engine.snapshot(0)
+    messages = [message for message, _tag in snapshot["log"]]
+    assert snapshot["state"] == "done"
+    assert snapshot["metrics"]["fail"] == 1
+    assert any("Failed links save error" in message for message in messages)
+    assert any(message.startswith("\nRun stopped: disk full") for message in messages)
+    assert list(tmp_path.glob("moontech_*.json"))
+    assert events == [f"Disk full in {tmp_path / 'downloads'}: need 77 bytes to continue"]
+
+
+def test_cli_retry_file_enospc_keeps_disk_full_outcome(monkeypatch, tmp_path, capsys):
+    events, calls, completed = [], [], []
+    _patch_frontend(monkeypatch, moon_cli, tmp_path, events, calls, completed)
+    _reject_retry_file(monkeypatch, moon_cli)
+
+    asyncio.run(moon_cli.run(_frontend_urls(), str(tmp_path / "downloads"), 2, 2, 2, "proxies.txt"))
+
+    output = capsys.readouterr().out
+    assert "[warn] Failed links save error" in output
+    assert "Run stopped: disk full" in output
+    assert list(tmp_path.glob("moontech_cli_*.json"))
+    assert events == [f"Disk full in {tmp_path / 'downloads'}: need 77 bytes to continue"]
+
+
+def _install_backoff_fatal(monkeypatch, module, tmp_path):
+    controls = []
+
+    class CapturingControl(RunFatalControl):
+        def __init__(self):
+            super().__init__()
+            controls.append(self)
+
+    async def fatal_sleep(_seconds):
+        controls[-1].report_disk_full(str(tmp_path / "downloads"), 33)
+
+    monkeypatch.setattr(module, "RunFatalControl", CapturingControl)
+    monkeypatch.setattr(module.asyncio, "sleep", fatal_sleep)
+    return controls
+
+
+def test_engine_does_not_enqueue_retry_after_fatal_backoff(monkeypatch, tmp_path):
+    events, calls, completed = [], [], []
+    _patch_frontend(monkeypatch, moon_engine, tmp_path, events, calls, completed)
+    controls = _install_backoff_fatal(monkeypatch, moon_engine, tmp_path)
+    extract_calls = []
+
+    async def no_link(url, _get_browser):
+        extract_calls.append(url)
+        return None
+
+    monkeypatch.setattr(moon_engine, "extract_fuckingfast", no_link)
+    engine = moon_engine.Engine()
+    engine._cfg["out_folder"] = str(tmp_path / "downloads")
+    url = "https://fuckingfast.co/retry/file.bin"
+    asyncio.run(engine._run([url], 2, 2, 2))
+
+    snapshot = engine.snapshot(0)
+    assert extract_calls == [url]
+    assert controls[0].disk_full is not None
+    assert snapshot["metrics"]["stage"] == "done"
+    assert engine._tracked[url].status == "aborted"
+    assert not any(file["state"] in ("extract", "download") for file in snapshot["files"])
+
+
+def test_cli_does_not_enqueue_retry_after_fatal_backoff(monkeypatch, tmp_path, capsys):
+    events, calls, completed = [], [], []
+    _patch_frontend(monkeypatch, moon_cli, tmp_path, events, calls, completed)
+    controls = _install_backoff_fatal(monkeypatch, moon_cli, tmp_path)
+    extract_calls = []
+
+    async def no_link(url, _get_browser):
+        extract_calls.append(url)
+        return None
+
+    monkeypatch.setattr(moon_cli, "extract_fuckingfast", no_link)
+    url = "https://fuckingfast.co/retry/file.bin"
+    asyncio.run(moon_cli.run([url], str(tmp_path / "downloads"), 2, 2, 2, "proxies.txt"))
+
+    assert extract_calls == [url]
+    assert controls[0].disk_full is not None
+    assert "Run stopped: disk full" in capsys.readouterr().out
+
+
+def test_cli_observes_completed_download_task_errors(monkeypatch, tmp_path):
+    """Done tasks are gathered too, so their exceptions do not reach the loop handler."""
+    events, calls, completed = [], [], []
+    _patch_frontend(monkeypatch, moon_cli, tmp_path, events, calls, completed)
+
+    async def exploding_download(_proxy_url, _cookies, dest, _rec, _bytes_acc, _kill_evt,
+                                 _kills_so_far, telem=None, on_event=None, *, fatal_control=None):
+        fatal_control.report_disk_full(os.path.dirname(dest), 11)
+        raise RuntimeError("already-completed task")
+
+    monkeypatch.setattr(moon_cli, "download_file", exploding_download)
+    loop_errors = []
+
+    async def run_cli():
+        loop = asyncio.get_running_loop()
+        old_handler = loop.get_exception_handler()
+        loop.set_exception_handler(lambda _loop, context: loop_errors.append(context))
+        try:
+            await moon_cli.run(
+                ["https://fuckingfast.co/error/file.bin"], str(tmp_path / "downloads"),
+                2, 2, 2, "proxies.txt")
+        finally:
+            loop.set_exception_handler(old_handler)
+
+    asyncio.run(run_cli())
+    assert loop_errors == []
diff --git a/tests/test_exit_cleanup.py b/tests/test_exit_cleanup.py
index 6b71531..5f71cf8 100644
--- a/tests/test_exit_cleanup.py
+++ b/tests/test_exit_cleanup.py
@@ -72,6 +72,7 @@ async def fake_download(
         kills_so_far,
         telem=None,
         on_event=None,
+        fatal_control=None,
     ):
         download_started.set()
         partial_path = pathlib.Path(dest + ".tmp")