release: archivebox 0.9.32rc4

archivebox/cli/archivebox_run.py

@@ -244,7 +244,7 @@ def process_stdin_records() -> int:
     return 0
 
 
-def run_runner(daemon: bool = False) -> int:
+def run_runner(daemon: bool = False, crawl_id: str | None = None) -> int:
     """
     Run the background runner loop.
 
@@ -269,7 +269,7 @@ def run_runner(daemon: bool = False) -> int:
         current.save(update_fields=["process_type", "modified_at"])
 
     try:
-        run_pending_crawls(daemon=daemon)
+        run_pending_crawls(daemon=daemon, crawl_id=crawl_id)
         return 0
     except KeyboardInterrupt:
         return 0
@@ -319,19 +319,7 @@ def main(daemon: bool, crawl_id: str, snapshot_id: str, binary_id: str):
             sys.exit(1)
 
     if crawl_id:
-        try:
-            from archivebox.services.runner import run_crawl
-
-            run_crawl(crawl_id)
-            sys.exit(0)
-        except KeyboardInterrupt:
-            sys.exit(0)
-        except Exception as e:
-            rprint(f"[red]Runner error: {type(e).__name__}: {e}[/red]", file=sys.stderr)
-            import traceback
-
-            traceback.print_exc()
-            sys.exit(1)
+        sys.exit(run_runner(daemon=False, crawl_id=crawl_id))
 
     if daemon:
         sys.exit(run_runner(daemon=True))

archivebox/cli/archivebox_snapshot.py

@@ -35,7 +35,7 @@ from collections.abc import Iterable
 
 import rich_click as click
 from rich import print as rprint
-from django.db.models import Q, Sum
+from django.db.models import Case, IntegerField, Q, Sum, When
 from django.db.models.functions import Coalesce
 
 from archivebox.cli.cli_utils import apply_filters
@@ -213,7 +213,7 @@ def list_snapshots(
 
     is_tty = sys.stdout.isatty() and not csv
 
-    queryset = Snapshot.objects.annotate(output_size_sum=Coalesce(Sum("archiveresult__output_size"), 0)).order_by("-created_at")
+    queryset = Snapshot.objects.order_by("-created_at")
 
     # Apply filters
     filter_kwargs = {
@@ -264,7 +264,17 @@ def list_snapshots(
 
     if sort:
         queryset = queryset.order_by(sort)
-    if limit:
+
+    if not is_tty:
+        if limit:
+            limited_ids = list(queryset.values_list("id", flat=True)[:limit])
+            preserved_order = Case(
+                *(When(id=snapshot_id, then=position) for position, snapshot_id in enumerate(limited_ids)),
+                output_field=IntegerField(),
+            )
+            queryset = Snapshot.objects.filter(id__in=limited_ids).order_by(preserved_order)
+        queryset = queryset.annotate(output_size_sum=Coalesce(Sum("archiveresult__output_size"), 0)).prefetch_related("tags")
+    elif limit:
         queryset = queryset[:limit]
 
     count = 0
@@ -276,8 +286,36 @@ def list_snapshots(
         rows: list[str] = []
         if with_headers:
             rows.append(",".join(cols))
+        simple_cols = {
+            "id",
+            "crawl_id",
+            "url",
+            "title",
+            "timestamp",
+            "depth",
+            "status",
+            "fs_version",
+            "bookmarked_at",
+            "created_at",
+            "modified_at",
+            "retry_at",
+            "downloaded_at",
+        }
+        from archivebox.misc.util import to_json
+
         for snapshot in queryset.iterator(chunk_size=500):
-            rows.append(snapshot.to_csv(cols=cols, separator=","))
+            if set(cols).issubset(simple_cols):
+                rows.append(
+                    ",".join(
+                        to_json(
+                            value.isoformat() if hasattr((value := getattr(snapshot, col, "")), "isoformat") else value,
+                            indent=None,
+                        )
+                        for col in cols
+                    ),
+                )
+            else:
+                rows.append(snapshot.to_csv(cols=cols, separator=","))
             count += 1
         output = "\n".join(rows)
         if output:

archivebox/config/common.py

@@ -1,6 +1,7 @@
 __package__ = "archivebox.config"
 
 import json
+import os
 import re
 import secrets
 import sys
@@ -76,6 +77,7 @@ class StorageConfig(BaseConfigSet):
     # ARCHIVE_DIR / USERS_DIR are resolved dynamically via get_config().
     ARCHIVE_DIR: Path = Field(default=CONSTANTS.ARCHIVE_DIR)
     USERS_DIR: Path = Field(default=CONSTANTS.USERS_DIR)
+    PERSONAS_DIR: Path = Field(default=CONSTANTS.PERSONAS_DIR)
 
     # TMP_DIR must be a local, fast, readable/writable dir by archivebox user,
     # must be a short path due to unix path length restrictions for socket files (<100 chars)
@@ -461,6 +463,7 @@ def get_config(
     snapshot: Any = None,
     archiveresult: Any = None,
     machine: Any = None,
+    include_machine: bool = True,
     resolve_plugins: bool = True,
 ) -> ArchiveBoxBaseConfig:
     """
@@ -484,7 +487,7 @@ def get_config(
     if crawl is None and snapshot is not None:
         crawl = snapshot.crawl
 
-    if machine is None:
+    if include_machine and machine is None:
         try:
             from django.apps import apps
 
@@ -520,10 +523,10 @@ def get_config(
 
     scope_overrides: ConfigPayload = {}
 
-    if machine is not None and machine.config:
+    if include_machine and machine is not None and machine.config:
         from archivebox.machine.models import _sanitize_machine_config
 
-        scope_overrides.update(_sanitize_machine_config(machine.config))
+        scope_overrides.update(_sanitize_machine_config(machine.config, lib_dir=config_data.get("LIB_DIR")))
 
     if persona is not None:
         scope_overrides.update(persona.get_derived_config())
@@ -564,6 +567,7 @@ def get_config(
     config_data["ABX_RUNTIME"] = "archivebox"
 
     config = ArchiveBoxConfig.model_validate(config_data)
+    os.environ["ABXPKG_LIB_DIR"] = str(config.LIB_DIR)
     archiving_warning_key = (config.TIMEOUT, config.USE_COLOR)
     if archiving_warning_key not in _WARNED_ARCHIVING_CONFIGS:
         config.warn_if_invalid()

archivebox/config/paths.py

@@ -13,7 +13,7 @@ from typing import TYPE_CHECKING
 
 from benedict import benedict
 
-from .permissions import SudoPermission, IS_ROOT, ARCHIVEBOX_USER
+from .permissions import SudoPermission, IS_ROOT, ARCHIVEBOX_USER, ARCHIVEBOX_GROUP
 
 if TYPE_CHECKING:
     from archivebox.config.common import ArchiveBoxConfig
@@ -158,11 +158,23 @@ def assert_dir_can_contain_unix_sockets(dir_path: Path) -> bool:
 
 
 def create_and_chown_dir(dir_path: Path) -> None:
+    """Create a required runtime dir and fix only that dir's ownership when needed."""
+    dir_existed = dir_path.exists()
+    dir_path.mkdir(parents=True, exist_ok=True)
+
+    try:
+        stat = dir_path.stat()
+    except OSError:
+        return
+
+    if dir_existed and stat.st_uid == ARCHIVEBOX_USER and stat.st_gid == ARCHIVEBOX_GROUP:
+        return
+
     with SudoPermission(uid=0, fallback=True):
-        dir_path.mkdir(parents=True, exist_ok=True)
-        subprocess.run(["chown", str(ARCHIVEBOX_USER), str(dir_path)], stderr=subprocess.DEVNULL)
-        for child in dir_path.iterdir():
-            subprocess.run(["chown", str(ARCHIVEBOX_USER), str(child)], stderr=subprocess.DEVNULL)
+        try:
+            os.chown(dir_path, ARCHIVEBOX_USER, ARCHIVEBOX_GROUP)
+        except (OSError, PermissionError):
+            pass
 
 
 def tmp_dir_socket_path_is_short_enough(dir_path: Path) -> bool:

archivebox/core/admin_site.py

@@ -10,6 +10,9 @@ from admin_data_views.admin import (
     get_app_list as adv_get_app_list,
 )
 
+from archivebox.config import VERSION
+from archivebox.config.version import get_COMMIT_HASH
+
 if TYPE_CHECKING:
     from django.http import HttpRequest
     from django.template.response import TemplateResponse
@@ -24,6 +27,12 @@ class ArchiveBoxAdmin(admin.AdminSite):
     site_title = "Admin"
     namespace = "admin"
 
+    def each_context(self, request: "HttpRequest") -> dict[str, Any]:
+        context = super().each_context(request)
+        context["VERSION"] = VERSION
+        context["STATIC_CACHE_KEY"] = (get_COMMIT_HASH() or VERSION or "dev").strip()
+        return context
+
     @staticmethod
     def _format_object_count(count: int) -> tuple[int, str, str]:
         if count >= 1_000_000_000:

archivebox/core/models.py

@@ -2537,7 +2537,7 @@ class Snapshot(ModelWithOutputDir, ModelWithConfig, ModelWithNotes, ModelWithHea
             lower = (path or "").lower()
             return lower.endswith(text_exts)
 
-        hashes_index = self.hashes_index
+        hashes_index = self.hashes_index if include_filesystem_fallback else {}
         for result in self.archiveresult_set.all().order_by("start_ts"):
             output_file_map = result.output_file_map()
             embed_path = result.embed_path_db(output_file_map=output_file_map)
@@ -2616,6 +2616,8 @@ class Snapshot(ModelWithOutputDir, ModelWithConfig, ModelWithNotes, ModelWithHea
 
         if not include_filesystem_fallback or hashes_index:
             return outputs
+        if not snap_dir.is_dir():
+            return outputs
 
         embeddable_exts = {
             "html",
@@ -2829,7 +2831,8 @@ class Snapshot(ModelWithOutputDir, ModelWithConfig, ModelWithNotes, ModelWithHea
         outputs: list[dict] | None = None,
         hidden_card_plugins: set[str] | None = None,
     ) -> tuple[list[dict[str, object]], list[dict[str, object]]]:
-        outputs = outputs or self.discover_outputs(include_filesystem_fallback=True)
+        if outputs is None:
+            outputs = self.discover_outputs(include_filesystem_fallback=True)
         hidden_card_plugins = hidden_card_plugins or set()
         accounted_entries: set[str] = set()
         for output in outputs:

archivebox/core/urls.py

@@ -69,7 +69,7 @@ urlpatterns = [
     path("accounts/login/", RedirectView.as_view(url="/admin/login/")),
     path("accounts/logout/", RedirectView.as_view(url="/admin/logout/")),
     path("accounts/", include("django.contrib.auth.urls")),
-    path("admin/live-progress/", live_progress_view, name="live_progress"),
+    path("admin/live-progress/", archivebox_admin.admin_view(live_progress_view), name="live_progress"),
     path("admin/", archivebox_admin.urls),
     path("api/", include("archivebox.api.urls"), name="api"),
     path("health/", HealthCheckView.as_view(), name="healthcheck"),

archivebox/core/views.py

@@ -18,8 +18,7 @@ from django.utils.safestring import mark_safe
 from django.views import View
 from django.views.generic.list import ListView
 from django.views.generic import FormView
-from django.db import connection
-from django.db.models import Q, Prefetch
+from django.db.models import Count, Q, Prefetch
 from django.contrib import messages
 from django.contrib.auth.mixins import UserPassesTestMixin
 from django.views.decorators.csrf import csrf_exempt
@@ -174,7 +173,7 @@ class SnapshotView(View):
         hidden_card_plugins = {"archivedotorg", "favicon", "title"}
         outputs = [
             out
-            for out in snapshot.discover_outputs(include_filesystem_fallback=True)
+            for out in snapshot.discover_outputs(include_filesystem_fallback=False)
             if (out.get("size") or 0) > 0 and out.get("name") not in hidden_card_plugins
         ]
         archiveresults = {out["name"]: out for out in outputs}
@@ -1319,6 +1318,19 @@ def live_progress_view(request):
         from archivebox.crawls.models import Crawl
         from archivebox.core.models import Snapshot, ArchiveResult
         from archivebox.machine.models import Process, Machine
+        from archivebox.machine.detect import get_host_guid
+
+        if not request.user.is_authenticated or not request.user.is_active or not request.user.is_staff:
+            return JsonResponse({"error": "Permission denied"}, status=403)
+
+        now = timezone.now()
+        crawl_scope = Crawl.objects.all()
+        snapshot_scope = Snapshot.objects.all()
+        archiveresult_scope = ArchiveResult.objects.all()
+        if not request.user.is_superuser:
+            crawl_scope = crawl_scope.filter(created_by=request.user)
+            snapshot_scope = snapshot_scope.filter(crawl__created_by=request.user)
+            archiveresult_scope = archiveresult_scope.filter(snapshot__crawl__created_by=request.user)
 
         def is_current_run_timestamp(event_ts, run_started_at) -> bool:
             if run_started_at is None:
@@ -1376,9 +1388,18 @@ def live_progress_view(request):
 
             return hook_details(Path(hook_path).name, plugin=Path(hook_path).parent.name or "setup")
 
-        machine = Machine.current()
-        Process.cleanup_stale_running(machine=machine)
-        Process.cleanup_orphaned_workers()
+        def snapshot_output_url(snapshot, output_path: str) -> str:
+            return build_snapshot_url(str(snapshot.id), output_path, request=request, config=getattr(request, "archivebox_config", None))
+
+        def snapshot_view_url(snapshot, output_path: str = "") -> str:
+            anchor = f"#{output_path}" if output_path else ""
+            return build_web_url(
+                f"/{snapshot.archive_path_from_db}/index.html{anchor}",
+                request=request,
+                config=getattr(request, "archivebox_config", None),
+            )
+
+        machine = Machine.objects.filter(guid=get_host_guid()).first()
         orchestrator_proc = (
             Process.objects.filter(
                 machine=machine,
@@ -1387,6 +1408,8 @@ def live_progress_view(request):
             )
             .order_by("-started_at")
             .first()
+            if machine is not None
+            else None
         )
         runner_worker = None
         try:
@@ -1402,85 +1425,110 @@ def live_progress_view(request):
         orchestrator_running = orchestrator_proc is not None or runner_worker_running
         orchestrator_pid = orchestrator_proc.pid if orchestrator_proc else runner_worker_pid
 
-        def sqlite_approx_count(model) -> int | None:
-            if connection.vendor != "sqlite":
-                return None
-            with connection.cursor() as cursor:
-                try:
-                    cursor.execute("SELECT stat FROM sqlite_stat1 WHERE tbl = %s", [model._meta.db_table])
-                    stats = [int(str(row[0]).split()[0]) for row in cursor.fetchall() if row and row[0]]
-                except Exception:
-                    stats = []
-            return max(stats) if stats else None
+        def count_statuses(queryset, statuses) -> dict[str, int]:
+            return {status: queryset.filter(status=status).count() for status in statuses}
 
         # Get model counts by status
-        crawls_pending = Crawl.objects.filter(status=Crawl.StatusChoices.QUEUED).count()
-        crawls_started = Crawl.objects.filter(status=Crawl.StatusChoices.STARTED).count()
+        crawl_status_counts = count_statuses(crawl_scope, (Crawl.StatusChoices.QUEUED, Crawl.StatusChoices.STARTED))
+        crawls_pending = crawl_status_counts.get(Crawl.StatusChoices.QUEUED, 0)
+        crawls_started = crawl_status_counts.get(Crawl.StatusChoices.STARTED, 0)
 
         # Get recent crawls (last 24 hours)
         from datetime import timedelta
 
-        one_day_ago = timezone.now() - timedelta(days=1)
-        crawls_recent = Crawl.objects.filter(created_at__gte=one_day_ago).count()
+        one_day_ago = now - timedelta(days=1)
+        recently_cancelled_after = now - timedelta(minutes=10)
+        crawls_recent = crawl_scope.filter(created_at__gte=one_day_ago).count()
 
-        snapshots_pending = Snapshot.objects.filter(status=Snapshot.StatusChoices.QUEUED).count()
-        snapshots_started = Snapshot.objects.filter(status=Snapshot.StatusChoices.STARTED).count()
+        snapshot_status_counts = count_statuses(snapshot_scope, (Snapshot.StatusChoices.QUEUED, Snapshot.StatusChoices.STARTED))
+        snapshots_pending = snapshot_status_counts.get(Snapshot.StatusChoices.QUEUED, 0)
+        snapshots_started = snapshot_status_counts.get(Snapshot.StatusChoices.STARTED, 0)
 
-        archiveresults_pending = ArchiveResult.objects.filter(status=ArchiveResult.StatusChoices.QUEUED).count()
-        archiveresults_started = ArchiveResult.objects.filter(status=ArchiveResult.StatusChoices.STARTED).count()
-        archiveresults_failed = ArchiveResult.objects.filter(status=ArchiveResult.StatusChoices.FAILED).count()
-        archiveresults_backoff = ArchiveResult.objects.filter(status=ArchiveResult.StatusChoices.BACKOFF).count()
-        archiveresults_skipped = ArchiveResult.objects.filter(status=ArchiveResult.StatusChoices.SKIPPED).count()
-        archiveresults_noresults = ArchiveResult.objects.filter(status=ArchiveResult.StatusChoices.NORESULTS).count()
-        archiveresults_total = sqlite_approx_count(ArchiveResult)
-        if archiveresults_total is not None:
-            archiveresults_succeeded = max(
-                archiveresults_total
-                - archiveresults_pending
-                - archiveresults_started
-                - archiveresults_failed
-                - archiveresults_backoff
-                - archiveresults_skipped
-                - archiveresults_noresults,
-                0,
-            )
-        else:
-            archiveresults_succeeded = ArchiveResult.objects.filter(status=ArchiveResult.StatusChoices.SUCCEEDED).count()
+        archiveresult_status_counts = count_statuses(
+            archiveresult_scope,
+            (
+                ArchiveResult.StatusChoices.QUEUED,
+                ArchiveResult.StatusChoices.STARTED,
+                ArchiveResult.StatusChoices.SUCCEEDED,
+                ArchiveResult.StatusChoices.FAILED,
+            ),
+        )
+        archiveresults_pending = archiveresult_status_counts.get(ArchiveResult.StatusChoices.QUEUED, 0)
+        archiveresults_started = archiveresult_status_counts.get(ArchiveResult.StatusChoices.STARTED, 0)
+        archiveresults_succeeded = archiveresult_status_counts.get(ArchiveResult.StatusChoices.SUCCEEDED, 0)
+        archiveresults_failed = archiveresult_status_counts.get(ArchiveResult.StatusChoices.FAILED, 0)
 
         # Build hierarchical active crawls with nested snapshots and archive results
 
         active_crawls_qs = (
-            Crawl.objects.filter(status__in=[Crawl.StatusChoices.QUEUED, Crawl.StatusChoices.STARTED])
-            .prefetch_related("snapshot_set")
+            crawl_scope.filter(
+                Q(status__in=[Crawl.StatusChoices.QUEUED, Crawl.StatusChoices.STARTED])
+                | Q(status=Crawl.StatusChoices.SEALED, modified_at__gte=recently_cancelled_after),
+            )
+            .select_related("created_by")
             .distinct()
             .order_by("-modified_at")[:10]
         )
+        active_crawls_list = list(active_crawls_qs)
+        active_crawl_ids = [crawl.id for crawl in active_crawls_list]
+        snapshot_counts_by_crawl: dict[str, dict[str, int]] = {str(crawl_id): {} for crawl_id in active_crawl_ids}
+        cancelled_snapshot_counts_by_crawl: dict[str, int] = {str(crawl_id): 0 for crawl_id in active_crawl_ids}
+        if active_crawl_ids:
+            for row in snapshot_scope.filter(crawl_id__in=active_crawl_ids).values("crawl_id", "status").annotate(count=Count("id")):
+                snapshot_counts_by_crawl.setdefault(str(row["crawl_id"]), {})[row["status"]] = row["count"]
+            for row in (
+                snapshot_scope.filter(
+                    crawl_id__in=active_crawl_ids,
+                    status=Snapshot.StatusChoices.SEALED,
+                    downloaded_at__isnull=True,
+                    modified_at__gte=recently_cancelled_after,
+                )
+                .values("crawl_id")
+                .annotate(count=Count("id"))
+            ):
+                cancelled_snapshot_counts_by_crawl[str(row["crawl_id"])] = row["count"]
 
-        running_processes = Process.objects.filter(
-            machine=machine,
-            status=Process.StatusChoices.RUNNING,
-            process_type__in=[
-                Process.TypeChoices.HOOK,
-                Process.TypeChoices.BINARY,
-            ],
-        )
-        recent_processes = Process.objects.filter(
-            machine=machine,
-            process_type__in=[
-                Process.TypeChoices.HOOK,
-                Process.TypeChoices.BINARY,
-            ],
-            modified_at__gte=timezone.now() - timedelta(minutes=10),
-        ).order_by("-modified_at")
+        if machine is not None:
+            running_processes = Process.objects.filter(
+                machine=machine,
+                status=Process.StatusChoices.RUNNING,
+                process_type__in=[
+                    Process.TypeChoices.HOOK,
+                    Process.TypeChoices.BINARY,
+                ],
+            ).only("id", "machine_id", "process_type", "status", "pwd", "cmd", "pid", "exit_code", "started_at", "modified_at")
+            recent_processes = (
+                Process.objects.filter(
+                    machine=machine,
+                    process_type__in=[
+                        Process.TypeChoices.HOOK,
+                        Process.TypeChoices.BINARY,
+                    ],
+                    modified_at__gte=now - timedelta(minutes=10),
+                )
+                .only("id", "machine_id", "process_type", "status", "pwd", "cmd", "pid", "exit_code", "started_at", "modified_at")
+                .order_by("-modified_at")
+            )
+        else:
+            running_processes = Process.objects.none()
+            recent_processes = Process.objects.none()
         crawl_process_pids: dict[str, int] = {}
         snapshot_process_pids: dict[str, int] = {}
         process_records_by_crawl: dict[str, list[tuple[dict[str, object], object | None]]] = {}
         process_records_by_snapshot: dict[str, list[tuple[dict[str, object], object | None]]] = {}
         seen_process_records: set[str] = set()
         active_snapshot_statuses = {Snapshot.StatusChoices.QUEUED, Snapshot.StatusChoices.STARTED}
-        snapshots = [
-            snapshot for crawl in active_crawls_qs for snapshot in crawl.snapshot_set.all() if snapshot.status in active_snapshot_statuses
-        ]
+        recently_cancelled_snapshots_q = Q(
+            status=Snapshot.StatusChoices.SEALED,
+            downloaded_at__isnull=True,
+            modified_at__gte=recently_cancelled_after,
+        )
+        crawls_by_id = {str(crawl.id): crawl for crawl in active_crawls_list}
+        snapshots = list(
+            snapshot_scope.filter(Q(status__in=active_snapshot_statuses) | recently_cancelled_snapshots_q, crawl_id__in=active_crawl_ids)
+            .select_related("crawl")
+            .order_by("crawl_id", "status", "modified_at")[:100],
+        )
         snapshots_by_id = {str(snapshot.id): snapshot for snapshot in snapshots}
 
         def find_snapshot_for_process(proc_pwd: Path) -> Snapshot | None:
@@ -1490,14 +1538,29 @@ def live_progress_view(request):
                     return snapshot
             return None
 
+        def find_crawl_for_process(proc_pwd: Path) -> Crawl | None:
+            for path_part in reversed(proc_pwd.parts):
+                crawl = crawls_by_id.get(path_part)
+                if crawl:
+                    return crawl
+            return None
+
+        running_worker_ids: set[str] = set()
         for proc in running_processes:
             if not proc.pwd:
                 continue
-            matched_snapshot = find_snapshot_for_process(Path(proc.pwd))
+            proc_pwd = Path(proc.pwd)
+            matched_snapshot = find_snapshot_for_process(proc_pwd)
+            matched_crawl = matched_snapshot.crawl if matched_snapshot is not None else find_crawl_for_process(proc_pwd)
             if matched_snapshot is None:
-                continue
-            crawl_id = str(matched_snapshot.crawl_id)
-            snapshot_id = str(matched_snapshot.id)
+                if matched_crawl is None:
+                    continue
+                crawl_id = str(matched_crawl.id)
+                snapshot_id = ""
+            else:
+                crawl_id = str(matched_snapshot.crawl_id)
+                snapshot_id = str(matched_snapshot.id)
+            running_worker_ids.add(str(proc.id))
             _plugin, _label, phase, _hook_name = process_label(proc.cmd)
             if crawl_id and proc.pid:
                 crawl_process_pids.setdefault(crawl_id, proc.pid)
@@ -1507,11 +1570,13 @@ def live_progress_view(request):
         for proc in recent_processes:
             if not proc.pwd:
                 continue
-            matched_snapshot = find_snapshot_for_process(Path(proc.pwd))
-            if matched_snapshot is None:
+            proc_pwd = Path(proc.pwd)
+            matched_snapshot = find_snapshot_for_process(proc_pwd)
+            matched_crawl = matched_snapshot.crawl if matched_snapshot is not None else find_crawl_for_process(proc_pwd)
+            if matched_snapshot is None and matched_crawl is None:
                 continue
-            crawl_id = str(matched_snapshot.crawl_id)
-            snapshot_id = str(matched_snapshot.id)
+            crawl_id = str(matched_snapshot.crawl_id if matched_snapshot is not None else matched_crawl.id)
+            snapshot_id = str(matched_snapshot.id) if matched_snapshot is not None else ""
 
             plugin, label, phase, hook_name = process_label(proc.cmd)
 
@@ -1545,19 +1610,21 @@ def live_progress_view(request):
                 process_records_by_crawl.setdefault(crawl_id, []).append((payload, proc_started_at))
 
         active_crawls = []
-        total_workers = 0
-        for crawl in active_crawls_qs:
-            # Get ALL snapshots for this crawl to count status (already prefetched)
-            all_crawl_snapshots = list(crawl.snapshot_set.all())
-
-            # Count snapshots by status from ALL snapshots
-            total_snapshots = len(all_crawl_snapshots)
-            completed_snapshots = sum(1 for s in all_crawl_snapshots if s.status == Snapshot.StatusChoices.SEALED)
-            started_snapshots = sum(1 for s in all_crawl_snapshots if s.status == Snapshot.StatusChoices.STARTED)
-            pending_snapshots = sum(1 for s in all_crawl_snapshots if s.status == Snapshot.StatusChoices.QUEUED)
+        total_workers = len(running_worker_ids)
+        for crawl in active_crawls_list:
+            crawl_snapshot_counts = snapshot_counts_by_crawl.get(str(crawl.id), {})
+            total_snapshots = sum(crawl_snapshot_counts.values())
+            completed_snapshots = crawl_snapshot_counts.get(Snapshot.StatusChoices.SEALED, 0)
+            started_snapshots = crawl_snapshot_counts.get(Snapshot.StatusChoices.STARTED, 0)
+            pending_snapshots = crawl_snapshot_counts.get(Snapshot.StatusChoices.QUEUED, 0)
+            cancelled_snapshots = cancelled_snapshot_counts_by_crawl.get(str(crawl.id), 0)
 
             # Get only ACTIVE snapshots to display (limit to 5 most recent)
-            active_crawl_snapshots = [s for s in all_crawl_snapshots if s.status in active_snapshot_statuses][:5]
+            active_crawl_snapshots = list(
+                snapshot_scope.filter(Q(status__in=active_snapshot_statuses) | recently_cancelled_snapshots_q, crawl=crawl)
+                .select_related("crawl")
+                .order_by("status", "modified_at")[:5],
+            )
 
             # Count URLs in the crawl (for when snapshots haven't been created yet)
             urls_count = 0
@@ -1572,7 +1639,6 @@ def live_progress_view(request):
                 for payload, proc_started_at in process_records_by_crawl.get(str(crawl.id), [])
                 if is_current_run_timestamp(proc_started_at, crawl_run_started_at)
             ]
-            total_workers += sum(1 for item in crawl_setup_plugins if item.get("source") == "process" and item.get("status") == "started")
             crawl_setup_total = len(crawl_setup_plugins)
             crawl_setup_completed = sum(1 for item in crawl_setup_plugins if item.get("status") == "succeeded")
             crawl_setup_failed = sum(1 for item in crawl_setup_plugins if item.get("status") == "failed")
@@ -1591,10 +1657,31 @@ def live_progress_view(request):
                     if archiveresult_matches_current_run(ar, snapshot_run_started_at)
                 ]
 
-                now = timezone.now()
                 plugin_progress_values: list[int] = []
                 all_plugins: list[dict[str, object]] = []
                 seen_plugin_keys: set[str] = set()
+                snapshot_title = snapshot._normalize_title_candidate(snapshot.title, snapshot_url=snapshot.url)
+                snapshot_favicon_url = ""
+                snapshot_preview_url = ""
+                snapshot_preview_link = snapshot_view_url(snapshot)
+                snapshot_fallback_urls: list[str] = []
+                result_by_plugin = {result.plugin: result for result in snapshot_results}
+                title_result = result_by_plugin.get("title")
+                if not snapshot_title and title_result is not None:
+                    snapshot_title = snapshot._normalize_title_candidate(title_result.output_str, snapshot_url=snapshot.url)
+                favicon_result = result_by_plugin.get("favicon")
+                if favicon_result is not None and favicon_result.status == ArchiveResult.StatusChoices.SUCCEEDED:
+                    favicon_path = favicon_result.embed_path_db() or "favicon/favicon.ico"
+                    snapshot_favicon_url = snapshot_output_url(snapshot, favicon_path)
+                screenshot_result = result_by_plugin.get("screenshot")
+                if screenshot_result is not None and screenshot_result.status == ArchiveResult.StatusChoices.SUCCEEDED:
+                    screenshot_path = screenshot_result.embed_path_db() or "screenshot/screenshot.png"
+                    snapshot_preview_url = snapshot_output_url(snapshot, screenshot_path)
+                    snapshot_preview_link = snapshot_view_url(snapshot, screenshot_path)
+                    if snapshot_favicon_url:
+                        snapshot_fallback_urls.append(snapshot_favicon_url)
+                elif snapshot_favicon_url:
+                    snapshot_preview_url = snapshot_favicon_url
 
                 def plugin_sort_key(ar):
                     status_order = {
@@ -1618,7 +1705,7 @@ def live_progress_view(request):
                         progress_value = 100
                     elif status == ArchiveResult.StatusChoices.STARTED:
                         started_at = ar.start_ts or (ar.process.started_at if ar.process_id and ar.process else None)
-                        timeout = ar.timeout or 120
+                        timeout = ar.process.timeout if ar.process_id and ar.process else 120
                         if started_at and timeout:
                             elapsed = max(0.0, (now - started_at).total_seconds())
                             progress_value = int(min(99, max(1, (elapsed / float(timeout)) * 100)))
@@ -1638,12 +1725,17 @@ def live_progress_view(request):
                         "phase": phase,
                         "status": status,
                         "process_id": str(ar.process_id) if ar.process_id else None,
+                        "admin_url": f"/admin/core/archiveresult/{ar.id}/change/",
                     }
+                    output_path = ar.embed_path_db()
+                    if output_path:
+                        plugin_payload["output_path"] = output_path
+                        plugin_payload["output_url"] = snapshot_view_url(snapshot, output_path)
                     if status == ArchiveResult.StatusChoices.STARTED and ar.process_id and ar.process:
                         plugin_payload["pid"] = ar.process.pid
                     if status == ArchiveResult.StatusChoices.STARTED:
                         plugin_payload["progress"] = progress_value
-                        plugin_payload["timeout"] = ar.timeout or 120
+                        plugin_payload["timeout"] = ar.process.timeout if ar.process_id and ar.process else 120
                     plugin_payload["source"] = "archiveresult"
                     all_plugins.append(plugin_payload)
                     seen_plugin_keys.add(str(ar.process_id) if ar.process_id else f"{ar.plugin}:{hook_name}")
@@ -1662,7 +1754,6 @@ def live_progress_view(request):
                         plugin_progress_values.append(100)
                     elif proc_status == "started":
                         plugin_progress_values.append(1)
-                        total_workers += 1
                     else:
                         plugin_progress_values.append(0)
 
@@ -1672,11 +1763,30 @@ def live_progress_view(request):
                 pending_plugins = sum(1 for item in all_plugins if item.get("status") == "queued")
 
                 snapshot_progress = int(sum(plugin_progress_values) / len(plugin_progress_values)) if plugin_progress_values else 0
+                worker_state = "running" if snapshot_process_pids.get(str(snapshot.id)) else "waiting"
+                if snapshot.status == Snapshot.StatusChoices.SEALED and not snapshot.downloaded_at:
+                    worker_state = "cancelled"
+                if (
+                    snapshot.status == Snapshot.StatusChoices.STARTED
+                    and worker_state == "waiting"
+                    and not all_plugins
+                    and snapshot.modified_at
+                    and (now - snapshot.modified_at).total_seconds() > 30
+                ):
+                    worker_state = "stalled" if orchestrator_running else "crashed"
 
                 active_snapshots_for_crawl.append(
                     {
                         "id": str(snapshot.id),
                         "url": snapshot.url[:80],
+                        "full_url": snapshot.url,
+                        "title": snapshot_title,
+                        "admin_url": f"/admin/core/snapshot/{snapshot.id}/change/",
+                        "view_url": snapshot_view_url(snapshot),
+                        "favicon_url": snapshot_favicon_url,
+                        "preview_url": snapshot_preview_url,
+                        "preview_link": snapshot_preview_link,
+                        "preview_fallbacks": snapshot_fallback_urls,
                         "status": snapshot.status,
                         "started": (snapshot.downloaded_at or snapshot.created_at).isoformat()
                         if (snapshot.downloaded_at or snapshot.created_at)
@@ -1688,6 +1798,7 @@ def live_progress_view(request):
                         "pending_plugins": pending_plugins,
                         "all_plugins": all_plugins,
                         "worker_pid": snapshot_process_pids.get(str(snapshot.id)),
+                        "worker_state": worker_state,
                     },
                 )
 
@@ -1696,8 +1807,19 @@ def live_progress_view(request):
             urls_preview = crawl.urls[:60] if crawl.urls else None
 
             # Check if retry_at is in the future (would prevent worker from claiming)
-            retry_at_future = crawl.retry_at > timezone.now() if crawl.retry_at else False
-            seconds_until_retry = int((crawl.retry_at - timezone.now()).total_seconds()) if crawl.retry_at and retry_at_future else 0
+            retry_at_future = crawl.retry_at > now if crawl.retry_at else False
+            seconds_until_retry = int((crawl.retry_at - now).total_seconds()) if crawl.retry_at and retry_at_future else 0
+            crawl_worker_state = (
+                "running"
+                if crawl_process_pids.get(str(crawl.id)) or any(snapshot.get("worker_pid") for snapshot in active_snapshots_for_crawl)
+                else "waiting"
+            )
+            if crawl.status == Crawl.StatusChoices.SEALED and cancelled_snapshots:
+                crawl_worker_state = "cancelled"
+            elif (
+                crawl.status == Crawl.StatusChoices.STARTED and crawl_worker_state == "waiting" and (started_snapshots or pending_snapshots)
+            ):
+                crawl_worker_state = "stalled" if orchestrator_running else "crashed"
 
             active_crawls.append(
                 {
@@ -1713,6 +1835,7 @@ def live_progress_view(request):
                     "started_snapshots": started_snapshots,
                     "failed_snapshots": 0,
                     "pending_snapshots": pending_snapshots,
+                    "cancelled_snapshots": cancelled_snapshots,
                     "setup_plugins": crawl_setup_plugins,
                     "setup_total_plugins": crawl_setup_total,
                     "setup_completed_plugins": crawl_setup_completed,
@@ -1724,6 +1847,7 @@ def live_progress_view(request):
                     "retry_at_future": retry_at_future,
                     "seconds_until_retry": seconds_until_retry,
                     "worker_pid": crawl_process_pids.get(str(crawl.id)),
+                    "worker_state": crawl_worker_state,
                 },
             )
 

archivebox/crawls/models.py

@@ -1,6 +1,7 @@
 __package__ = "archivebox.crawls"
 
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
+from collections.abc import Iterable, Mapping
 from io import StringIO
 import uuid
 import json
@@ -410,7 +411,9 @@ class Crawl(ModelWithOutputDir, ModelWithConfig, ModelWithHealthStats, ModelWith
     def url_passes_filters(self, url: str, *, snapshot=None, use_effective_config: bool = True) -> bool:
         denylist = self.get_url_denylist(use_effective_config=use_effective_config, snapshot=snapshot)
         allowlist = self.get_url_allowlist(use_effective_config=use_effective_config, snapshot=snapshot)
+        return self.url_passes_compiled_filters(url, allowlist=allowlist, denylist=denylist)
 
+    def url_passes_compiled_filters(self, url: str, *, allowlist: list[str], denylist: list[str]) -> bool:
         for pattern in denylist:
             if self._pattern_matches_url(url, pattern):
                 return False
@@ -748,56 +751,158 @@ class Crawl(ModelWithOutputDir, ModelWithConfig, ModelWithHealthStats, ModelWith
         created_by_id: int | None = None,
     ):
         """Create one child snapshot if it passes crawl filters and limits."""
-        from archivebox.core.models import Snapshot
+        snapshots = self.create_discovered_snapshots(
+            parent_snapshot,
+            [{"url": url, "title": title, "tags": tags}],
+            depth=depth,
+            created_by_id=created_by_id,
+        )
+        return snapshots[0] if snapshots else None
+
+    def create_discovered_snapshots(
+        self,
+        parent_snapshot,
+        records: Iterable[Mapping[str, Any]],
+        *,
+        depth: int,
+        created_by_id: int | None = None,
+    ) -> list["Snapshot"]:
+        """Create child snapshots from discovered URL records after filtering and deduping once."""
+        from archivebox.core.models import Snapshot, SnapshotTag, Tag
+        from archivebox.config.common import get_config
         from archivebox.misc.util import fix_url_from_markdown, sanitize_extracted_url
+        from archivebox.core.host_utils import get_admin_host, get_api_host, get_listen_host, get_public_host, get_web_host, split_host_port
 
         if self.status == self.StatusChoices.SEALED:
-            return None
+            return []
 
-        url = sanitize_extracted_url(fix_url_from_markdown(str(url or "").strip()))
-        if not url:
-            return None
-        try:
-            validate_url_length(url)
-        except ValueError as err:
-            print(f"[yellow][!] Skipping over-long discovered snapshot URL: {url[:120]}... ({err})[/yellow]")
-            return None
         if depth > self.max_depth:
-            return None
-        if not self.url_passes_filters(url, snapshot=parent_snapshot):
-            return None
-        if self.snapshot_set.filter(url=url).exists():
-            return None
-        if not self.has_remaining_snapshot_capacity():
-            return None
+            return []
+
+        config = get_config(crawl=self, snapshot=parent_snapshot)
+        allowlist = self.split_filter_patterns(config.get("URL_ALLOWLIST", ""))
+        denylist = self.split_filter_patterns(config.get("URL_DENYLIST", ""))
+        protected_subdomains = {"admin", "web", "api", "public"}
+        protected_hosts = set()
+        protected_roots = set()
+        for host_value in (
+            get_listen_host(config=config),
+            get_admin_host(config=config),
+            get_web_host(config=config),
+            get_api_host(config=config),
+            get_public_host(config=config),
+        ):
+            if not host_value:
+                continue
+            protected_host = split_host_port(host_value)[0].strip(".")
+            if not protected_host:
+                continue
+            protected_hosts.add(protected_host)
+            host_parts = protected_host.split(".", 1)
+            if len(host_parts) == 2 and (host_parts[0] in protected_subdomains or host_parts[0].startswith("snap-")):
+                protected_roots.add(host_parts[1])
+            else:
+                protected_roots.add(protected_host)
+        uses_subdomain_routing = bool(config.get("USES_SUBDOMAIN_ROUTING", False))
+
+        deduped_records: dict[str, Mapping[str, Any]] = {}
+        for record in records:
+            url = sanitize_extracted_url(fix_url_from_markdown(str(record.get("url") or "").strip()))
+            if not url or url in deduped_records:
+                continue
+            try:
+                validate_url_length(url)
+            except ValueError as err:
+                print(f"[yellow][!] Skipping over-long discovered snapshot URL: {url[:120]}... ({err})[/yellow]")
+                continue
+            parsed = urlparse(url)
+            host = (parsed.hostname or "").lower().strip(".")
+            is_internal_url = False
+            if parsed.scheme in ("http", "https") and host:
+                if host in protected_hosts:
+                    is_internal_url = True
+                elif uses_subdomain_routing:
+                    for protected_root in protected_roots:
+                        if not protected_root or not host.endswith(f".{protected_root}"):
+                            continue
+                        subdomain = host[: -(len(protected_root) + 1)]
+                        if subdomain in protected_subdomains or subdomain.startswith("snap-"):
+                            is_internal_url = True
+                            break
+            if is_internal_url:
+                print(f"[yellow][!] Skipping internal ArchiveBox discovered snapshot URL: {url}[/yellow]")
+                continue
+            if self.url_passes_compiled_filters(url, allowlist=allowlist, denylist=denylist):
+                deduped_records[url] = record
+
+        if not deduped_records:
+            return []
+
+        existing_urls = set(self.snapshot_set.filter(url__in=deduped_records.keys()).values_list("url", flat=True))
+        urls = [url for url in deduped_records.keys() if url not in existing_urls]
+        remaining = self.remaining_snapshot_capacity()
+        if remaining is not None:
+            urls = urls[:remaining]
+        if not urls:
+            return []
+
+        now = timezone.now()
+        snapshots = [
+            Snapshot(
+                url=url,
+                timestamp=str((now + timedelta(microseconds=index)).timestamp()),
+                title=str(deduped_records[url].get("title") or "").strip()[:512] or None,
+                crawl=self,
+                parent_snapshot=parent_snapshot,
+                depth=depth,
+                status=Snapshot.StatusChoices.QUEUED,
+                retry_at=now,
+                bookmarked_at=now,
+                created_at=now,
+            )
+            for index, url in enumerate(urls)
+        ]
 
         try:
-            snapshot = Snapshot.from_json(
-                {
-                    "url": url,
-                    "depth": depth,
-                    "title": title,
-                    "tags": tags,
-                    "parent_snapshot_id": str(parent_snapshot.id),
-                    "crawl_id": str(self.id),
-                },
-                overrides={
-                    "crawl": self,
-                    "snapshot": parent_snapshot,
-                    "created_by_id": created_by_id or self.created_by_id,
-                },
-                queue_for_extraction=False,
-            )
+            created_snapshots = list(Snapshot.objects.bulk_create(snapshots))
         except ValidationError as err:
-            print(f"[yellow][!] Skipping blocked discovered snapshot URL: {url} ({err})[/yellow]")
-            return None
-        if snapshot is None or snapshot.status == Snapshot.StatusChoices.SEALED:
-            return None
+            print(f"[yellow][!] Skipping blocked discovered snapshots: {err}[/yellow]")
+            return []
 
-        snapshot.status = Snapshot.StatusChoices.QUEUED
-        snapshot.retry_at = timezone.now()
-        snapshot.save(update_fields=["status", "retry_at", "modified_at"])
-        return snapshot
+        crawl_urls = {url for _raw_line, url in self._iter_url_lines() if url}
+        new_url_lines = [snapshot.url for snapshot in created_snapshots if snapshot.url not in crawl_urls]
+        if new_url_lines:
+            self.urls = (self.urls.rstrip() + "\n" + "\n".join(new_url_lines)).lstrip("\n")
+            self.save(update_fields=["urls", "modified_at"])
+
+        tag_names_by_url: dict[str, set[str]] = {}
+        for snapshot in created_snapshots:
+            tags = str(deduped_records[snapshot.url].get("tags") or "").strip()
+            if tags:
+                tag_names_by_url[snapshot.url] = {tag.strip() for tag in re.split(config.TAG_SEPARATOR_PATTERN, tags) if tag.strip()}
+            try:
+                snapshot.ensure_crawl_symlink()
+            except Exception:
+                pass
+
+        tag_names = {tag for tags in tag_names_by_url.values() for tag in tags}
+        if tag_names:
+            tags_by_name = {tag.name: tag for tag in Tag.objects.filter(name__in=tag_names)}
+            missing_tags = [Tag(name=name) for name in sorted(tag_names - tags_by_name.keys())]
+            if missing_tags:
+                Tag.objects.bulk_create(missing_tags, ignore_conflicts=True)
+                tags_by_name = {tag.name: tag for tag in Tag.objects.filter(name__in=tag_names)}
+            SnapshotTag.objects.bulk_create(
+                [
+                    SnapshotTag(snapshot=snapshot, tag=tags_by_name[tag_name])
+                    for snapshot in created_snapshots
+                    for tag_name in tag_names_by_url.get(snapshot.url, set())
+                    if tag_name in tags_by_name
+                ],
+                ignore_conflicts=True,
+            )
+
+        return created_snapshots
 
     def install_declared_binaries(self, binary_names: set[str], machine=None) -> None:
         """

archivebox/hooks.py

@@ -442,6 +442,7 @@ def run_hook(
     lib_bin_dir = resolved_config.LIB_BIN_DIR
     if lib_dir:
         env["LIB_DIR"] = str(lib_dir)
+        env["ABXPKG_LIB_DIR"] = str(lib_dir)
     if lib_bin_dir:
         env["LIB_BIN_DIR"] = str(lib_bin_dir)
 
@@ -468,6 +469,7 @@ def run_hook(
     SKIP_KEYS = {
         "PATH",
         "LIB_DIR",
+        "ABXPKG_LIB_DIR",
         "LIB_BIN_DIR",
         "NODE_PATH",
         "NODE_MODULES_DIR",

archivebox/machine/migrations/0014_process_machine_pro_pid_6eec8b_idx_and_more.py

@@ -0,0 +1,20 @@
+# Generated by Django 6.0.5 on 2026-05-26 18:19
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("machine", "0013_alter_machine_config"),
+    ]
+
+    operations = [
+        migrations.AddIndex(
+            model_name="process",
+            index=models.Index(fields=["pid", "started_at"], name="machine_pro_pid_6eec8b_idx"),
+        ),
+        migrations.AddIndex(
+            model_name="process",
+            index=models.Index(fields=["process_type", "worker_type", "pwd", "started_at"], name="machine_pro_process_b0411b_idx"),
+        ),
+    ]

archivebox/machine/models.py

@@ -52,8 +52,6 @@ PROCESS_RECHECK_INTERVAL = 60  # Re-validate every 60 seconds
 PID_REUSE_WINDOW = timedelta(hours=24)  # Max age for considering a PID match valid
 PROCESS_TIMEOUT_GRACE = timedelta(seconds=30)  # Extra margin before force-cleaning timed-out RUNNING rows
 START_TIME_TOLERANCE = 5.0  # Seconds tolerance for start time matching
-LEGACY_MACHINE_CONFIG_KEYS = frozenset({"CHROMIUM_VERSION"})
-MACHINE_CONFIG_ALWAYS_ALLOWED_KEYS = frozenset({"ABX_INSTALL_CACHE"})
 
 
 def _find_existing_binary_for_reference(machine: Machine, reference: str) -> Binary | None:
@@ -76,6 +74,13 @@ def _find_existing_binary_for_reference(machine: Machine, reference: str) -> Bin
     return qs.filter(name=reference).order_by("-modified_at").first()
 
 
+def _canonical_binary_name(name: Any) -> str:
+    name = str(name or "").strip()
+    if "/" in name or "\\" in name or name.startswith("~"):
+        return Path(name).expanduser().name
+    return name
+
+
 def _get_process_binary_env_keys(plugin_name: str, hook_path: str, env: dict[str, Any] | None) -> list[str]:
     env = env or {}
     plugin_name = str(plugin_name or "").strip()
@@ -122,13 +127,12 @@ def _get_process_binary_env_keys(plugin_name: str, hook_path: str, env: dict[str
     return keys
 
 
-def _sanitize_machine_config(config: dict[str, Any] | None) -> dict[str, Any]:
+def _sanitize_machine_config(config: dict[str, Any] | None, *, lib_dir: str | Path | None = None) -> dict[str, Any]:
     if not isinstance(config, dict):
         return {}
 
-    sanitized = {key: value for key, value in config.items() if key in MACHINE_CONFIG_ALWAYS_ALLOWED_KEYS or str(key).endswith("_BINARY")}
-    for key in LEGACY_MACHINE_CONFIG_KEYS:
-        sanitized.pop(key, None)
+    sanitized = {key: value for key, value in config.items() if str(key).endswith("_BINARY")}
+    active_lib_dir = Path(lib_dir).expanduser().resolve(strict=False) if lib_dir else None
     for key, value in list(sanitized.items()):
         if not str(key).endswith("_BINARY"):
             continue
@@ -140,8 +144,16 @@ def _sanitize_machine_config(config: dict[str, Any] | None) -> dict[str, Any]:
             continue
         if "/" in value or value.startswith("~"):
             try:
-                if not Path(value).expanduser().exists():
+                path = Path(value).expanduser()
+                if not path.exists():
                     sanitized.pop(key, None)
+                    continue
+                if active_lib_dir is not None:
+                    resolved_path = path.resolve(strict=False)
+                    try:
+                        resolved_path.relative_to(active_lib_dir)
+                    except ValueError:
+                        sanitized.pop(key, None)
             except OSError:
                 sanitized.pop(key, None)
     return sanitized
@@ -237,7 +249,9 @@ class Machine(ModelWithHealthStats):
 
     @classmethod
     def _sanitize_config(cls, machine: Machine) -> Machine:
-        sanitized = _sanitize_machine_config(machine.config)
+        from archivebox.config.constants import CONSTANTS
+
+        sanitized = _sanitize_machine_config(machine.config, lib_dir=os.environ.get("LIB_DIR") or CONSTANTS.DEFAULT_LIB_DIR)
         current = machine.config or {}
         if sanitized != current:
             machine.config = sanitized
@@ -521,7 +535,7 @@ class Binary(ModelWithHealthStats, ModelWithStateMachine):
         Returns:
             Binary instance or None
         """
-        name = record.get("name")
+        name = _canonical_binary_name(record.get("name"))
         if not name:
             return None
 
@@ -1094,6 +1108,8 @@ class Process(models.Model):
         indexes = [
             models.Index(fields=["machine", "status", "retry_at"]),
             models.Index(fields=["binary", "exit_code"]),
+            models.Index(fields=["pid", "started_at"]),
+            models.Index(fields=["process_type", "worker_type", "pwd", "started_at"]),
         ]
 
     def __str__(self) -> str:

archivebox/services/binary_service.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+from pathlib import Path
+
 from asgiref.sync import sync_to_async
 
 from abx_dl.events import BinaryRequestEvent, BinaryEvent
@@ -20,6 +22,7 @@ class BinaryService(BaseService):
 
         machine = await sync_to_async(Machine.current, thread_sensitive=True)()
         existing = await Binary.objects.filter(machine=machine, name=event.name).afirst()
+        cache_invalidated = False
         if existing and existing.status == Binary.StatusChoices.INSTALLED:
             changed = False
             if event.binproviders and existing.binproviders != event.binproviders:
@@ -29,7 +32,10 @@ class BinaryService(BaseService):
                 existing.overrides = event.overrides
                 changed = True
             if changed:
-                await existing.asave(update_fields=["binproviders", "overrides", "modified_at"])
+                existing.status = Binary.StatusChoices.QUEUED
+                existing.retry_at = None
+                cache_invalidated = True
+                await existing.asave(update_fields=["binproviders", "overrides", "status", "retry_at", "modified_at"])
         elif existing is None:
             await Binary.objects.acreate(
                 machine=machine,
@@ -39,19 +45,50 @@ class BinaryService(BaseService):
                 status=Binary.StatusChoices.QUEUED,
             )
 
-        installed = (
-            await Binary.objects.filter(machine=machine, name=event.name, status=Binary.StatusChoices.INSTALLED)
-            .exclude(abspath="")
-            .exclude(abspath__isnull=True)
-            .order_by("-modified_at")
-            .afirst()
-        )
+        installed = None
+        if not cache_invalidated:
+            installed = (
+                await Binary.objects.filter(machine=machine, name=event.name, status=Binary.StatusChoices.INSTALLED)
+                .exclude(abspath="")
+                .exclude(abspath__isnull=True)
+                .order_by("-modified_at")
+                .afirst()
+            )
+            if installed is not None and not await sync_to_async(Path(installed.abspath).expanduser().exists, thread_sensitive=True)():
+                installed.status = Binary.StatusChoices.QUEUED
+                installed.retry_at = None
+                await installed.asave(update_fields=["status", "retry_at", "modified_at"])
+                installed = None
+            if installed is not None and event.overrides and installed.overrides != event.overrides:
+                installed.status = Binary.StatusChoices.QUEUED
+                installed.retry_at = None
+                await installed.asave(update_fields=["status", "retry_at", "modified_at"])
+                installed = None
         cached = None
         if installed is not None:
+            from archivebox.config.common import get_config
             from abxpkg import BinProvider, PROVIDER_CLASS_BY_NAME
 
             binary_env: dict[str, str] = {}
+            installed_path = Path(installed.abspath).expanduser().resolve(strict=False)
+            active_lib_dir = (
+                Path(str((await sync_to_async(get_config, thread_sensitive=True)()).get("LIB_DIR", "")))
+                .expanduser()
+                .resolve(
+                    strict=False,
+                )
+            )
             provider_name = (installed.binprovider or installed.binproviders.split(",", 1)[0]).strip()
+            if active_lib_dir and provider_name in {"npm", "pip", "puppeteer", "uv", "deno", "gem", "cargo", "goget", "nix", "bash"}:
+                try:
+                    installed_path.relative_to(active_lib_dir)
+                except ValueError:
+                    installed.status = Binary.StatusChoices.QUEUED
+                    installed.retry_at = None
+                    await installed.asave(update_fields=["status", "retry_at", "modified_at"])
+                    installed = None
+            if installed is None:
+                return None
             provider_class = PROVIDER_CLASS_BY_NAME.get(provider_name)
             if provider_class is not None:
                 provider = provider_class()

archivebox/services/machine_service.py

@@ -16,15 +16,17 @@ class MachineService(BaseService):
 
     async def on_MachineEvent__save_to_db(self, event: MachineEvent) -> None:
         from archivebox.machine.models import Machine, _sanitize_machine_config
+        from archivebox.config.common import get_config
 
         if event.config_type != "derived":
             return
 
         machine = await sync_to_async(Machine.current, thread_sensitive=True)()
+        lib_dir = await sync_to_async(lambda: get_config(include_machine=False).LIB_DIR, thread_sensitive=True)()
         config = dict(machine.config or {})
 
         if event.config is not None:
-            config.update(_sanitize_machine_config(event.config))
+            config.update(_sanitize_machine_config(event.config, lib_dir=lib_dir))
         elif event.method == "update":
             key = event.key.replace("config/", "", 1).strip()
             if key:
@@ -36,5 +38,5 @@ class MachineService(BaseService):
         else:
             return
 
-        machine.config = _sanitize_machine_config(config)
+        machine.config = _sanitize_machine_config(config, lib_dir=lib_dir)
         await machine.asave(update_fields=["config", "modified_at"])

archivebox/services/process_service.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import asyncio
 from datetime import datetime
 from typing import ClassVar
 
@@ -7,7 +8,7 @@ from asgiref.sync import sync_to_async
 from django.utils import timezone
 
 from abxbus import BaseEvent
-from abx_dl.events import ProcessCompletedEvent, ProcessStartedEvent
+from abx_dl.events import CrawlCleanupEvent, CrawlCompletedEvent, ProcessCompletedEvent, ProcessStartedEvent
 from abx_dl.services.base import BaseService
 
 
@@ -26,23 +27,38 @@ def parse_event_datetime(value: str | None):
 def current_network_interface_with_machine():
     from archivebox.machine.models import NetworkInterface
 
-    current_iface = NetworkInterface.current(refresh=True)
+    current_iface = NetworkInterface.current()
     return NetworkInterface.objects.select_related("machine").get(id=current_iface.id)
 
 
 class ProcessService(BaseService):
-    LISTENS_TO: ClassVar[list[type[BaseEvent]]] = [ProcessStartedEvent, ProcessCompletedEvent]
+    LISTENS_TO: ClassVar[list[type[BaseEvent]]] = [
+        ProcessStartedEvent,
+        ProcessCompletedEvent,
+        CrawlCleanupEvent,
+        CrawlCompletedEvent,
+    ]
     EMITS: ClassVar[list[type[BaseEvent]]] = []
 
     def __init__(self, bus):
+        self._iface = None
+        self._completed_queue: asyncio.Queue[ProcessCompletedEvent | None] = asyncio.Queue()
+        self._completed_worker: asyncio.Task | None = None
         super().__init__(bus)
         self.bus.on(ProcessStartedEvent, self.on_ProcessStartedEvent__save_to_db)
         self.bus.on(ProcessCompletedEvent, self.on_ProcessCompletedEvent__save_to_db)
+        self.bus.on(CrawlCleanupEvent, self.on_CrawlCleanupEvent__flush_completed)
+        self.bus.on(CrawlCompletedEvent, self.on_CrawlCompletedEvent__flush_completed)
+
+    async def current_iface(self):
+        if self._iface is None:
+            self._iface = await sync_to_async(current_network_interface_with_machine, thread_sensitive=True)()
+        return self._iface
 
     async def on_ProcessStartedEvent__save_to_db(self, event: ProcessStartedEvent) -> None:
         from archivebox.machine.models import Process
 
-        iface = await sync_to_async(current_network_interface_with_machine, thread_sensitive=True)()
+        iface = await self.current_iface()
         process_type = event.process_type or (
             Process.TypeChoices.BINARY if event.hook_name.startswith("on_BinaryRequest") else Process.TypeChoices.HOOK
         )
@@ -50,15 +66,15 @@ class ProcessService(BaseService):
         started_at = parse_event_datetime(event.start_ts)
         if started_at is None:
             raise ValueError("ProcessStartedEvent.start_ts is required")
-        process_query = Process.objects.filter(
-            process_type=process_type,
-            worker_type=worker_type,
-            pwd=event.output_dir,
-            cmd=[event.hook_path, *event.hook_args],
-            started_at=started_at,
-        )
         if event.pid:
-            process_query = process_query.filter(pid=event.pid)
+            process_query = Process.objects.filter(pid=event.pid, started_at=started_at)
+        else:
+            process_query = Process.objects.filter(
+                process_type=process_type,
+                worker_type=worker_type,
+                pwd=event.output_dir,
+                started_at=started_at,
+            )
         process = await process_query.order_by("-modified_at").afirst()
         if process is None:
             process = await Process.objects.acreate(
@@ -96,12 +112,50 @@ class ProcessService(BaseService):
             plugin_name=event.plugin_name,
             hook_path=event.hook_path,
         )
-        await process.asave()
+        await Process.objects.filter(id=process.id).aupdate(
+            pwd=process.pwd,
+            cmd=process.cmd,
+            env=process.env,
+            timeout=process.timeout,
+            pid=process.pid,
+            url=process.url,
+            process_type=process.process_type,
+            worker_type=process.worker_type,
+            started_at=process.started_at,
+            status=process.status,
+            retry_at=process.retry_at,
+            binary_id=process.binary_id,
+            modified_at=timezone.now(),
+        )
+
+    async def _completed_worker_loop(self) -> None:
+        while True:
+            event = await self._completed_queue.get()
+            try:
+                if event is None:
+                    return
+                await self._save_completed_process_to_db(event)
+            finally:
+                self._completed_queue.task_done()
+
+    def _ensure_completed_worker(self) -> None:
+        if self._completed_worker is None or self._completed_worker.done():
+            self._completed_worker = asyncio.create_task(self._completed_worker_loop())
 
     async def on_ProcessCompletedEvent__save_to_db(self, event: ProcessCompletedEvent) -> None:
+        self._ensure_completed_worker()
+        await self._completed_queue.put(event)
+
+    async def on_CrawlCleanupEvent__flush_completed(self, event: CrawlCleanupEvent) -> None:
+        await self._completed_queue.join()
+
+    async def on_CrawlCompletedEvent__flush_completed(self, event: CrawlCompletedEvent) -> None:
+        await self._completed_queue.join()
+
+    async def _save_completed_process_to_db(self, event: ProcessCompletedEvent) -> None:
         from archivebox.machine.models import Process
 
-        iface = await sync_to_async(current_network_interface_with_machine, thread_sensitive=True)()
+        iface = await self.current_iface()
         process_type = event.process_type or (
             Process.TypeChoices.BINARY if event.hook_name.startswith("on_BinaryRequest") else Process.TypeChoices.HOOK
         )
@@ -109,18 +163,18 @@ class ProcessService(BaseService):
         started_at = parse_event_datetime(event.start_ts)
         if started_at is None:
             raise ValueError("ProcessCompletedEvent.start_ts is required")
-        process_query = Process.objects.filter(
-            process_type=process_type,
-            worker_type=worker_type,
-            pwd=event.output_dir,
-            cmd=[event.hook_path, *event.hook_args],
-            started_at=started_at,
-        )
         if event.pid:
-            process_query = process_query.filter(pid=event.pid)
+            process_query = Process.objects.filter(pid=event.pid, started_at=started_at)
+        else:
+            process_query = Process.objects.filter(
+                process_type=process_type,
+                worker_type=worker_type,
+                pwd=event.output_dir,
+                started_at=started_at,
+            )
         process = await process_query.order_by("-modified_at").afirst()
         if process is None:
-            process = await Process.objects.acreate(
+            await Process.objects.acreate(
                 machine=iface.machine,
                 iface=iface,
                 process_type=process_type,
@@ -135,28 +189,28 @@ class ProcessService(BaseService):
                 status=Process.StatusChoices.RUNNING,
                 retry_at=None,
             )
-        elif process.iface_id != iface.id or process.machine_id != iface.machine_id:
-            process.iface = iface
-            process.machine = iface.machine
-            await process.asave(update_fields=["iface", "machine", "modified_at"])
+            process = await process_query.order_by("-modified_at").afirst()
+            if process is None:
+                return
 
-        process.pwd = event.output_dir
-        if not process.cmd:
-            process.cmd = [event.hook_path, *event.hook_args]
-        process.env = event.env
-        process.pid = event.pid or process.pid
-        process.url = event.url or process.url
-        process.process_type = process_type or process.process_type
-        process.worker_type = worker_type or process.worker_type
-        process.started_at = started_at
-        process.ended_at = parse_event_datetime(event.end_ts) or timezone.now()
-        process.stdout = event.stdout
-        process.stderr = event.stderr
-        process.exit_code = event.exit_code
-        process.status = process.StatusChoices.EXITED
-        process.retry_at = None
-        await sync_to_async(process.hydrate_binary_from_context, thread_sensitive=True)(
-            plugin_name=event.plugin_name,
-            hook_path=event.hook_path,
-        )
-        await process.asave()
+        missing_cmd = not process.cmd
+        updates = {
+            "machine_id": iface.machine_id,
+            "iface_id": iface.id,
+            "pwd": event.output_dir,
+            "pid": event.pid or process.pid,
+            "url": event.url or process.url,
+            "process_type": process_type or process.process_type,
+            "worker_type": worker_type or process.worker_type,
+            "started_at": started_at,
+            "ended_at": parse_event_datetime(event.end_ts) or timezone.now(),
+            "stdout": event.stdout,
+            "stderr": event.stderr,
+            "exit_code": event.exit_code,
+            "status": Process.StatusChoices.EXITED,
+            "retry_at": None,
+            "modified_at": timezone.now(),
+        }
+        if missing_cmd:
+            updates["cmd"] = [event.hook_path, *event.hook_args]
+        await Process.objects.filter(id=process.id).aupdate(**updates)

archivebox/services/runner.py

@@ -4,6 +4,7 @@ import asyncio
 import contextvars
 import json
 import os
+import signal
 import shutil
 import subprocess
 import sys
@@ -206,6 +207,7 @@ class CrawlRunner:
         self.initial_snapshot_ids = snapshot_ids
         self.snapshot_tasks: dict[str, asyncio.Task[None]] = {}
         self.snapshot_semaphore = asyncio.Semaphore(1)
+        self.max_concurrent_snapshots = 1
         self.persona = None
         self.base_config: dict[str, Any] = {}
         self.derived_config: dict[str, Any] = {}
@@ -214,11 +216,60 @@ class CrawlRunner:
         self._live_stream = None
         self.root_crawl_event_id: str | None = None
         self.root_crawl_start_event_id: str | None = None
+        self._run_task: asyncio.Task[None] | None = None
         self._skip_wait_until_idle = False
+        self._signal_abort_requested = False
+
+    def _install_signal_handlers(self) -> list[tuple[signal.Signals, Any, bool]]:
+        loop = asyncio.get_running_loop()
+        installed: list[tuple[signal.Signals, Any, bool]] = []
+        for sig in (signal.SIGINT, signal.SIGTERM):
+            previous = signal.getsignal(sig)
+
+            def request_abort(sig=sig) -> None:
+                self._request_abort_from_signal(sig)
+
+            try:
+                loop.add_signal_handler(sig, request_abort)
+                installed.append((sig, previous, True))
+            except (NotImplementedError, RuntimeError):
+                signal.signal(sig, lambda _signum, _frame, sig=sig: self._request_abort_from_signal(sig))
+                installed.append((sig, previous, False))
+        return installed
+
+    def _restore_signal_handlers(self, installed: list[tuple[signal.Signals, Any, bool]]) -> None:
+        loop = asyncio.get_running_loop()
+        for sig, previous, installed_on_loop in reversed(installed):
+            if installed_on_loop:
+                loop.remove_signal_handler(sig)
+            signal.signal(sig, previous)
+
+    def _request_abort_from_signal(self, sig: signal.Signals) -> None:
+        if self._signal_abort_requested:
+            if self._run_task is not None and not self._run_task.done():
+                self._run_task.cancel()
+            return
+        self._signal_abort_requested = True
+        self._skip_wait_until_idle = True
+        asyncio.create_task(self.abort_from_signal(sig.name))
+
+    async def abort_from_signal(self, signal_name: str) -> None:
+        from archivebox.crawls.models import Crawl
+
+        await sync_to_async(
+            Crawl.objects.filter(id=self.crawl.id).exclude(status=Crawl.StatusChoices.SEALED).update,
+            thread_sensitive=False,
+        )(
+            status=Crawl.StatusChoices.STARTED,
+            retry_at=timezone.now(),
+            modified_at=timezone.now(),
+        )
 
     async def crawl_is_cancelled(self) -> bool:
         from archivebox.crawls.models import Crawl
 
+        if self._signal_abort_requested:
+            return True
         return await Crawl.objects.filter(id=self.crawl.id, status=Crawl.StatusChoices.SEALED).aexists()
 
     async def watch_for_cancelled_crawl(self, parent_event: BaseEvent, *, poll_interval: float = 1.0) -> None:
@@ -239,10 +290,13 @@ class CrawlRunner:
             runtime="archivebox",
             crawl_id=str(self.crawl.id),
         )
+        installed_signal_handlers = self._install_signal_handlers()
         root_snapshot_id: str | None = None
         try:
+            self._run_task = asyncio.current_task()
             snapshot_ids = await sync_to_async(self.load_run_state, thread_sensitive=True)()
-            max_concurrent_snapshots = max(1, int(self.base_config.CRAWL_MAX_CONCURRENT_SNAPSHOTS))
+            max_concurrent_snapshots = max(1, int(self.base_config.get("CRAWL_MAX_CONCURRENT_SNAPSHOTS", 1)))
+            self.max_concurrent_snapshots = max_concurrent_snapshots
             self.snapshot_semaphore = asyncio.Semaphore(max_concurrent_snapshots)
             live_ui = self._create_live_ui()
             with live_ui if live_ui is not None else nullcontext():
@@ -259,6 +313,8 @@ class CrawlRunner:
                     root_snapshot_id = snapshot_ids[0]
                     await self.run_crawl(root_snapshot_id, snapshot_ids)
         finally:
+            self._run_task = None
+            self._restore_signal_handlers(installed_signal_handlers)
             await heartbeat.stop()
             if not self._skip_wait_until_idle:
                 await self.bus.wait_until_idle()
@@ -271,6 +327,8 @@ class CrawlRunner:
             await sync_to_async(self.finalize_run_state, thread_sensitive=True)()
 
     async def enqueue_snapshot(self, snapshot_id: str) -> None:
+        if await self.crawl_is_cancelled():
+            return
         task = self.snapshot_tasks.get(snapshot_id)
         if task is not None and not task.done():
             return
@@ -285,6 +343,7 @@ class CrawlRunner:
 
     async def wait_for_snapshot_tasks(self) -> None:
         task_errors: list[Exception] = []
+        stop_scheduling = False
         while True:
             pending_tasks: list[asyncio.Task[None]] = []
             for snapshot_id, task in list(self.snapshot_tasks.items()):
@@ -300,15 +359,18 @@ class CrawlRunner:
                         await sync_to_async(recover_orphaned_crawls, thread_sensitive=True)()
                     except Exception as err:
                         task_errors.append(err)
+                        stop_scheduling = True
                     continue
                 pending_tasks.append(task)
             if not pending_tasks:
+                if task_errors:
+                    if len(task_errors) == 1:
+                        raise task_errors[0]
+                    raise ExceptionGroup("One or more snapshot tasks failed", task_errors)
+                if stop_scheduling:
+                    return
                 await self.enqueue_pending_snapshots_from_projection()
                 if not self.snapshot_tasks:
-                    if task_errors:
-                        if len(task_errors) == 1:
-                            raise task_errors[0]
-                        raise ExceptionGroup("One or more snapshot tasks failed", task_errors)
                     return
                 continue
             done, _pending = await asyncio.wait(pending_tasks, return_when=asyncio.FIRST_COMPLETED)
@@ -326,22 +388,55 @@ class CrawlRunner:
                     await sync_to_async(recover_orphaned_crawls, thread_sensitive=True)()
                 except Exception as err:
                     task_errors.append(err)
-            await self.enqueue_pending_snapshots_from_projection()
+                    stop_scheduling = True
+            if self.snapshot_tasks and await self.crawl_is_cancelled():
+                stop_scheduling = True
+            if not stop_scheduling:
+                await self.enqueue_pending_snapshots_from_projection()
+
+    async def drain_snapshot_tasks(self) -> None:
+        task_errors: list[Exception] = []
+        while self.snapshot_tasks:
+            done, _pending = await asyncio.wait(list(self.snapshot_tasks.values()), return_when=asyncio.FIRST_COMPLETED)
+            for task in done:
+                for snapshot_id, tracked_task in list(self.snapshot_tasks.items()):
+                    if tracked_task is task:
+                        self.snapshot_tasks.pop(snapshot_id, None)
+                        break
+                try:
+                    task.result()
+                except asyncio.CancelledError as err:
+                    if _is_external_task_cancelled(err):
+                        raise
+                    await sync_to_async(recover_orphaned_snapshots, thread_sensitive=True)()
+                    await sync_to_async(recover_orphaned_crawls, thread_sensitive=True)()
+                except Exception as err:
+                    task_errors.append(err)
+        if task_errors:
+            if len(task_errors) == 1:
+                raise task_errors[0]
+            raise ExceptionGroup("One or more snapshot tasks failed", task_errors)
 
     async def enqueue_pending_snapshots_from_projection(self) -> None:
         from archivebox.core.models import Snapshot
 
         if not isinstance(get_current_event(), CrawlStartEvent):
             return
+        if await self.crawl_is_cancelled():
+            return
 
+        active_snapshot_ids = [snapshot_id for snapshot_id, task in self.snapshot_tasks.items() if not task.done()]
+        available_slots = max(0, self.max_concurrent_snapshots - len(active_snapshot_ids))
+        if available_slots <= 0:
+            return
         pending_snapshot_ids = await sync_to_async(
-            lambda: [
-                str(snapshot_id)
-                for snapshot_id in self.crawl.snapshot_set.exclude(status=Snapshot.StatusChoices.SEALED)
+            lambda: list(
+                self.crawl.snapshot_set.exclude(status=Snapshot.StatusChoices.SEALED)
+                .exclude(id__in=active_snapshot_ids)
                 .filter(retry_at__lte=timezone.now())
                 .order_by("depth", "created_at")
-                .values_list("id", flat=True)
-            ],
+                .values_list("id", flat=True)[:available_slots],
+            ),
             thread_sensitive=True,
         )()
         for snapshot_id in pending_snapshot_ids:
@@ -361,8 +456,8 @@ class CrawlRunner:
             current_process.machine = current_iface.machine
             current_process.save(update_fields=["iface", "machine", "modified_at"])
         self.persona = self.crawl.resolve_persona()
-        self.base_config = get_config(crawl=self.crawl)
-        self.derived_config = _sanitize_machine_config(Machine.current().config)
+        self.base_config = get_config(crawl=self.crawl, include_machine=False)
+        self.derived_config = _sanitize_machine_config(Machine.current().config, lib_dir=self.base_config["LIB_DIR"])
         self.crawl_output_dir = str(self.crawl.output_dir)
         self.base_config["ABX_RUNTIME"] = "archivebox"
         if self.selected_plugins is None:
@@ -463,7 +558,7 @@ class CrawlRunner:
         from archivebox.config.common import get_config
 
         snapshot = Snapshot.objects.select_related("crawl").get(id=snapshot_id)
-        config = get_config(crawl=self.crawl, snapshot=snapshot)
+        config = get_config(crawl=self.crawl, snapshot=snapshot, include_machine=False)
         config.update(self.base_config)
         config["CRAWL_DIR"] = self.crawl_output_dir
         config["SNAP_DIR"] = str(snapshot.output_dir)
@@ -514,24 +609,13 @@ class CrawlRunner:
         if parent_snapshot is None:
             return
 
-        for record in discovered_urls:
-            url = str(record.get("url") or "").strip()
-            if not url:
-                continue
-            child_snapshot = await sync_to_async(self.crawl.create_discovered_snapshot, thread_sensitive=True)(
-                parent_snapshot,
-                url=url,
-                depth=parent_snapshot.depth + 1,
-                title=str(record.get("title") or "").strip(),
-                tags=str(record.get("tags") or "").strip(),
-            )
-            if child_snapshot is None:
-                has_capacity = await sync_to_async(self.crawl.has_remaining_snapshot_capacity, thread_sensitive=True)()
-                if has_capacity:
-                    continue
-                break
-            if self.process_discovered_snapshots_inline and isinstance(get_current_event(), CrawlStartEvent):
-                await self.enqueue_snapshot(str(child_snapshot.id))
+        await sync_to_async(self.crawl.create_discovered_snapshots, thread_sensitive=True)(
+            parent_snapshot,
+            discovered_urls,
+            depth=parent_snapshot.depth + 1,
+        )
+        if self.process_discovered_snapshots_inline and isinstance(get_current_event(), CrawlStartEvent):
+            await self.enqueue_pending_snapshots_from_projection()
 
     async def run_crawl(self, root_snapshot_id: str, snapshot_ids: list[str]) -> None:
         snapshot = await sync_to_async(self.load_snapshot_payload, thread_sensitive=True)(root_snapshot_id)
@@ -618,6 +702,10 @@ class CrawlRunner:
             if event.event_id != self.root_crawl_start_event_id:
                 return
             for snapshot_id in snapshot_ids:
+                if sum(1 for task in self.snapshot_tasks.values() if not task.done()) >= self.max_concurrent_snapshots:
+                    break
+                if await self.crawl_is_cancelled():
+                    break
                 await self.enqueue_snapshot(snapshot_id)
             await self.wait_for_snapshot_tasks()
 
@@ -652,6 +740,8 @@ class CrawlRunner:
                         self.root_crawl_start_event_id = crawl_start_event.event_id
                         await _run_event_now(event.emit(crawl_start_event), None)
                 finally:
+                    if self.snapshot_tasks:
+                        await self.drain_snapshot_tasks()
                     await _run_event_now(
                         event.emit(
                             CrawlCleanupEvent(
@@ -749,7 +839,7 @@ class CrawlRunner:
             snapshot_hooks = [(plugin, hook) for plugin in plugins.values() for hook in plugin.filter_hooks("Snapshot")]
             snapshot_phase_timeout = compute_phase_timeout(snapshot_hooks, config)
             await _emit_machine_config(self.bus, config=config, derived_config=derived_config, parent_event=crawl_start_event)
-            HookSnapshotService(
+            snapshot_service = HookSnapshotService(
                 self.bus,
                 url=snapshot["url"],
                 snapshot=abx_snapshot,
@@ -760,27 +850,30 @@ class CrawlRunner:
                 snapshot_cleanup_phase_timeout=snapshot_phase_timeout,
                 abort_requested=self.crawl_is_cancelled,
             )
-            snapshot_event = SnapshotEvent(
-                url=snapshot["url"],
-                snapshot_id=snapshot["id"],
-                output_dir=str(output_dir),
-                depth=int(snapshot["depth"]),
-                event_timeout=snapshot_phase_timeout,
-                event_handler_slow_timeout=slow_warning_timeout(snapshot_phase_timeout),
-            )
-            emitted_snapshot_event = crawl_start_event.emit(snapshot_event)
-            await _run_event_now(emitted_snapshot_event, snapshot_phase_timeout)
-            completed_snapshot = await self.bus.find(
-                SnapshotCompletedEvent,
-                child_of=emitted_snapshot_event,
-                past=True,
-                future=snapshot_phase_timeout,
-            )
-            if completed_snapshot is None:
-                raise RuntimeError(f"Snapshot {snapshot_id} did not complete")
-            await completed_snapshot.wait(timeout=snapshot_phase_timeout)
-            await completed_snapshot.event_results_list()
-            await self.enqueue_discovered_snapshots_from_outputs(snapshot)
+            try:
+                snapshot_event = SnapshotEvent(
+                    url=snapshot["url"],
+                    snapshot_id=snapshot["id"],
+                    output_dir=str(output_dir),
+                    depth=int(snapshot["depth"]),
+                    event_timeout=snapshot_phase_timeout,
+                    event_handler_slow_timeout=slow_warning_timeout(snapshot_phase_timeout),
+                )
+                emitted_snapshot_event = crawl_start_event.emit(snapshot_event)
+                await _run_event_now(emitted_snapshot_event, snapshot_phase_timeout)
+                completed_snapshot = await self.bus.find(
+                    SnapshotCompletedEvent,
+                    child_of=emitted_snapshot_event,
+                    past=True,
+                    future=snapshot_phase_timeout,
+                )
+                if completed_snapshot is None:
+                    raise RuntimeError(f"Snapshot {snapshot_id} did not complete")
+                await completed_snapshot.wait(timeout=snapshot_phase_timeout)
+                await completed_snapshot.event_results_list()
+                await self.enqueue_discovered_snapshots_from_outputs(snapshot)
+            finally:
+                snapshot_service.close()
 
     def seal_snapshot_due_to_limit(self, snapshot_id: str) -> None:
         from archivebox.core.models import Snapshot
@@ -815,13 +908,13 @@ def run_crawl(
 
 async def _run_binary(binary_id: str) -> None:
     from archivebox.config.common import get_config
-    from archivebox.machine.models import Binary, Machine
+    from archivebox.machine.models import Binary, Machine, _sanitize_machine_config
 
     binary = await Binary.objects.aget(id=binary_id)
     plugins = discover_plugins()
-    config = get_config()
+    config = get_config(include_machine=False)
     machine = await sync_to_async(Machine.current, thread_sensitive=True)()
-    derived_config = _normalize_runtime_config(dict(machine.config))
+    derived_config = _normalize_runtime_config(_sanitize_machine_config(machine.config, lib_dir=config["LIB_DIR"]))
     config["ABX_RUNTIME"] = "archivebox"
     config = _normalize_runtime_config(config)
     bus = create_bus(name=_bus_name("ArchiveBox_binary", str(binary.id)), total_timeout=1800.0)
@@ -867,12 +960,12 @@ def run_binary(binary_id: str) -> None:
 
 async def _run_install(plugin_names: list[str] | None = None) -> None:
     from archivebox.config.common import get_config
-    from archivebox.machine.models import Machine
+    from archivebox.machine.models import Machine, _sanitize_machine_config
 
     plugins = discover_plugins()
-    config = get_config()
+    config = get_config(include_machine=False)
     machine = await sync_to_async(Machine.current, thread_sensitive=True)()
-    derived_config = _normalize_runtime_config(dict(machine.config))
+    derived_config = _normalize_runtime_config(_sanitize_machine_config(machine.config, lib_dir=config["LIB_DIR"]))
     config["ABX_RUNTIME"] = "archivebox"
     config = _normalize_runtime_config(config)
     bus = create_bus(name="ArchiveBox_install", total_timeout=3600.0)

archivebox/services/snapshot_service.py

@@ -1,7 +1,5 @@
 from __future__ import annotations
 
-from pathlib import Path
-
 from asgiref.sync import sync_to_async
 from django.utils import timezone
 from abx_dl.events import SnapshotCompletedEvent, SnapshotEvent
@@ -20,67 +18,18 @@ class SnapshotService(BaseService):
         self.bus.on(SnapshotEvent, self.on_SnapshotEvent)
         self.bus.on(SnapshotCompletedEvent, self.on_SnapshotCompletedEvent)
 
-    async def _upsert_discovered_snapshot(self, parent_snapshot, *, url: str, depth: int, title: str = "", tags: str = "") -> str | None:
-        crawl = parent_snapshot.crawl
-        if depth > crawl.max_depth:
-            return None
-        stop_reason = await sync_to_async(self._crawl_limit_stop_reason, thread_sensitive=True)(crawl)
-        if stop_reason == "crawl_max_size":
-            return None
-        snapshot = await sync_to_async(crawl.create_discovered_snapshot, thread_sensitive=True)(
-            parent_snapshot,
-            url=url,
-            depth=depth,
-            title=title,
-            tags=tags,
-        )
-        if snapshot is None:
-            return None
-        return str(snapshot.id)
-
     async def on_SnapshotEvent(self, event: SnapshotEvent) -> None:
         from archivebox.core.models import Snapshot
-        from archivebox.crawls.models import Crawl
 
-        crawl = await Crawl.objects.aget(id=self.crawl_id)
-        snapshot_id: str | None = None
-        snapshot = await Snapshot.objects.filter(id=event.snapshot_id, crawl=crawl).afirst()
+        snapshot = await Snapshot.objects.filter(id=event.snapshot_id, crawl_id=self.crawl_id).afirst()
 
         if snapshot is not None:
             snapshot.status = Snapshot.StatusChoices.STARTED
             snapshot.retry_at = None
             await snapshot.asave(update_fields=["status", "retry_at", "modified_at"])
-            snapshot_id = str(snapshot.id)
-        elif event.depth > 0:
-            parent_event = await self.bus.find(
-                SnapshotEvent,
-                past=True,
-                future=False,
-                where=lambda candidate: candidate.depth == event.depth - 1 and self.bus.event_is_child_of(event, candidate),
-            )
-            parent_snapshot = None
-            if parent_event is not None:
-                parent_snapshot = (
-                    await Snapshot.objects.select_related("crawl", "crawl__created_by")
-                    .filter(id=parent_event.snapshot_id, crawl=crawl)
-                    .afirst()
-                )
-            if parent_snapshot is not None:
-                snapshot_id = await self._upsert_discovered_snapshot(
-                    parent_snapshot,
-                    url=event.url,
-                    depth=event.depth,
-                )
-
-        if snapshot_id:
-            snapshot = await Snapshot.objects.filter(id=snapshot_id).select_related("crawl", "crawl__created_by").afirst()
-            if snapshot is not None:
-                await sync_to_async(snapshot.ensure_crawl_symlink, thread_sensitive=True)()
-        if snapshot_id and event.depth > 0:
-            await self.schedule_snapshot(snapshot_id)
+            await sync_to_async(snapshot.ensure_crawl_symlink, thread_sensitive=True)()
 
     async def on_SnapshotCompletedEvent(self, event: SnapshotCompletedEvent) -> None:
-        from archivebox.crawls.models import Crawl
         from archivebox.core.models import Snapshot
 
         snapshot = await Snapshot.objects.select_related("crawl", "crawl__created_by").filter(id=event.snapshot_id).afirst()
@@ -112,27 +61,8 @@ class SnapshotService(BaseService):
                     await sync_to_async(snapshot.write_index_jsonl, thread_sensitive=True)()
                     await sync_to_async(snapshot.write_json_details, thread_sensitive=True)()
                     await sync_to_async(snapshot.write_html_details, thread_sensitive=True)()
-                    stop_reason = await sync_to_async(self._crawl_limit_stop_reason, thread_sensitive=True)(snapshot.crawl)
-                    if snapshot.depth < snapshot.crawl.max_depth and stop_reason != "crawl_max_size":
-                        from archivebox.hooks import collect_urls_from_plugins
-
-                        discovered_urls = await sync_to_async(collect_urls_from_plugins, thread_sensitive=True)(Path(snapshot.output_dir))
-                        for record in discovered_urls:
-                            discovered_snapshot_id = await self._upsert_discovered_snapshot(
-                                snapshot,
-                                url=str(record.get("url") or "").strip(),
-                                depth=snapshot.depth + 1,
-                                title=str(record.get("title") or "").strip(),
-                                tags=str(record.get("tags") or "").strip(),
-                            )
-                            if discovered_snapshot_id:
-                                await self.schedule_snapshot(discovered_snapshot_id)
                 finally:
-                    is_finished = await sync_to_async(snapshot.crawl.is_finished, thread_sensitive=True)()
-                    if is_finished and snapshot.crawl.status != Crawl.StatusChoices.SEALED:
-                        snapshot.crawl.status = Crawl.StatusChoices.SEALED
-                        snapshot.crawl.retry_at = None
-                        await snapshot.crawl.asave(update_fields=["status", "retry_at", "modified_at"])
+                    pass
 
     def _crawl_limit_stop_reason(self, crawl) -> str:
         config = dict(crawl.config or {})

archivebox/templates/admin/base.html

@@ -1506,7 +1506,7 @@
         <link href="{% static 'select2.min.css' %}" rel="stylesheet"/>
         <script src="{% static 'select2.min.js' %}"></script>
 
-        <link rel="stylesheet" type="text/css" href="{% static "admin.css" %}">
+        <link rel="stylesheet" type="text/css" href="{% static "admin.css" %}?v={{ STATIC_CACHE_KEY|default:VERSION|urlencode }}">
         
         <script>
             function selectSnapshotListView(e) {

archivebox/templates/admin/progress_monitor.html

@@ -309,10 +309,61 @@
         text-align: center;
         color: #58a6ff;
     }
+    #progress-monitor .snapshot-preview {
+        display: flex;
+        align-items: center;
+        justify-content: center;
+        width: 84px;
+        height: 52px;
+        flex: 0 0 84px;
+        border-radius: 6px;
+        border: 1px solid #30363d;
+        background: #161b22;
+        color: #6e7681;
+        overflow: hidden;
+        text-decoration: none;
+    }
+    #progress-monitor .snapshot-preview:hover {
+        border-color: #58a6ff;
+        box-shadow: 0 0 0 1px rgba(88, 166, 255, 0.2);
+    }
+    #progress-monitor .snapshot-preview img {
+        width: 100%;
+        height: 100%;
+        object-fit: cover;
+        object-position: top center;
+        display: block;
+    }
+    #progress-monitor .snapshot-preview.placeholder {
+        font-size: 20px;
+    }
     #progress-monitor .snapshot-info {
         flex: 1;
         min-width: 0;
     }
+    #progress-monitor .snapshot-title-line {
+        display: flex;
+        align-items: center;
+        gap: 7px;
+        min-width: 0;
+    }
+    #progress-monitor .snapshot-favicon {
+        width: 16px;
+        height: 16px;
+        border-radius: 3px;
+        object-fit: contain;
+        background: #fff;
+        flex: 0 0 16px;
+    }
+    #progress-monitor .snapshot-title {
+        color: #f0f6fc;
+        font-size: 12px;
+        font-weight: 600;
+        line-height: 1.3;
+        white-space: nowrap;
+        overflow: hidden;
+        text-overflow: ellipsis;
+    }
     #progress-monitor .snapshot-url {
         font-family: 'Monaco', 'Menlo', 'Ubuntu Mono', monospace;
         font-size: 11px;
@@ -351,6 +402,12 @@
         background: #21262d;
         overflow: hidden;
         white-space: nowrap;
+        color: inherit;
+        text-decoration: none;
+    }
+    #progress-monitor a.extractor-badge:hover {
+        background: #30363d;
+        color: #f0f6fc;
     }
     #progress-monitor .extractor-badge .progress-fill {
         position: absolute;
@@ -416,14 +473,6 @@
     #progress-monitor .extractor-badge .badge-icon {
         font-size: 10px;
     }
-    #progress-monitor .extractor-badge.started .badge-icon {
-        animation: spin 1s linear infinite;
-    }
-    @keyframes spin {
-        from { transform: rotate(0deg); }
-        to { transform: rotate(360deg); }
-    }
-
     /* Status Badge */
     #progress-monitor .status-badge {
         font-size: 10px;
@@ -593,11 +642,11 @@
                     <span class="stat-label">Queued</span>
                     <span class="stat-value warning" id="total-queued">0</span>
                 </div>
-                <div class="stat clickable" id="stat-succeeded" title="Click to reset counter">
+                <div class="stat" id="stat-succeeded">
                     <span class="stat-label">Done</span>
                     <span class="stat-value success" id="total-succeeded">0</span>
                 </div>
-                <div class="stat clickable" id="stat-failed" title="Click to reset counter">
+                <div class="stat" id="stat-failed">
                     <span class="stat-label">Failed</span>
                     <span class="stat-value error" id="total-failed">0</span>
                 </div>
@@ -629,10 +678,6 @@
     let isCollapsed = localStorage.getItem('progress-monitor-collapsed') === 'true';
     let knownThumbnailIds = new Set();
 
-    // Baselines for resettable counters
-    let succeededBaseline = parseInt(localStorage.getItem('progress-succeeded-baseline') || '0');
-    let failedBaseline = parseInt(localStorage.getItem('progress-failed-baseline') || '0');
-
     function getApiKey() {
         return (window.ARCHIVEBOX_API_KEY || '').trim();
     }
@@ -650,21 +695,28 @@
         if (apiKey) headers['X-ArchiveBox-API-Key'] = apiKey;
         return headers;
     }
-    let lastSucceeded = 0;
-    let lastFailed = 0;
-
-    // Click handlers for resetting counters
-    document.getElementById('stat-succeeded').addEventListener('click', function() {
-        succeededBaseline = lastSucceeded;
-        localStorage.setItem('progress-succeeded-baseline', succeededBaseline);
-        document.getElementById('total-succeeded').textContent = '0';
-    });
-    document.getElementById('stat-failed').addEventListener('click', function() {
-        failedBaseline = lastFailed;
-        localStorage.setItem('progress-failed-baseline', failedBaseline);
-        document.getElementById('total-failed').textContent = '0';
-    });
-
+    function escapeHtml(value) {
+        return String(value || '').replace(/[&<>"']/g, char => ({
+            '&': '&amp;',
+            '<': '&lt;',
+            '>': '&gt;',
+            '"': '&quot;',
+            "'": '&#39;',
+        }[char]));
+    }
+    function escapeAttr(value) {
+        return escapeHtml(value).replace(/`/g, '&#96;');
+    }
+    window.nextPreviewFallback = function(img) {
+        const fallbacks = (img.dataset.fallbacks || '').split(',').filter(Boolean);
+        if (fallbacks.length > 0) {
+            img.src = fallbacks.shift();
+            img.dataset.fallbacks = fallbacks.join(',');
+        } else {
+            img.closest('.snapshot-preview')?.classList.add('placeholder');
+            img.closest('.snapshot-preview').innerHTML = '<span>▤</span>';
+        }
+    };
     function formatUrl(url) {
         if (!url) return '(no URL)';
         try {
@@ -677,19 +729,20 @@
 
     function getPluginIcon(plugin) {
         const icons = {
-            'screenshot': '&#128247;',
-            'favicon': '&#11088;',
-            'dom': '&#128196;',
-            'pdf': '&#128462;',
-            'title': '&#128221;',
-            'headers': '&#128203;',
-            'singlefile': '&#128230;',
-            'readability': '&#128214;',
-            'mercury': '&#9884;',
-            'wget': '&#128229;',
-            'media': '&#127909;',
+            'screenshot': '▧',
+            'chrome_mhtml': '▧',
+            'favicon': '◆',
+            'dom': '&lt;/&gt;',
+            'pdf': 'PDF',
+            'title': 'T',
+            'headers': '{}',
+            'singlefile': '▣',
+            'readability': 'R',
+            'mercury': 'M',
+            'wget': '↓',
+            'media': '▶',
         };
-        return icons[plugin] || '&#128196;';
+        return icons[plugin] || '•';
     }
 
 
@@ -697,37 +750,52 @@
     function updateThumbnails(thumbnails) {}
 
     function renderExtractor(extractor) {
-        const icon = extractor.status === 'started' ? '&#8635;' :
-                    extractor.status === 'succeeded' ? '&#10003;' :
-                    extractor.status === 'failed' ? '&#10007;' :
-                    extractor.status === 'backoff' ? '&#8987;' :
-                    extractor.status === 'skipped' ? '&#8674;' : '&#9675;';
+        const icon = extractor.status === 'started' ? '▶' :
+                    extractor.status === 'succeeded' ? '✓' :
+                    extractor.status === 'failed' ? '!' :
+                    extractor.status === 'backoff' ? 'wait' :
+                    extractor.status === 'skipped' ? 'skip' :
+                    extractor.status === 'noresults' ? '∅' : getPluginIcon(extractor.plugin);
         const progress = typeof extractor.progress === 'number'
             ? Math.max(0, Math.min(100, extractor.progress))
             : null;
         const progressStyle = progress !== null ? ` style="width: ${progress}%;"` : '';
         const pidHtml = extractor.status === 'started' && extractor.pid ? `<span class="pid-label compact">pid ${extractor.pid}</span>` : '';
+        const href = extractor.output_url || extractor.admin_url || '';
+        const tag = href ? 'a' : 'span';
+        const hrefAttr = href ? ` href="${escapeAttr(href)}"` : '';
+        const title = extractor.output_path
+            ? `${extractor.plugin || 'output'}: ${extractor.output_path}`
+            : `${extractor.plugin || 'hook'}${extractor.hook_name ? `: ${extractor.hook_name}` : ''}`;
 
         return `
-            <span class="extractor-badge ${extractor.status || 'queued'}">
+            <${tag} class="extractor-badge ${extractor.status || 'queued'}"${hrefAttr} title="${escapeAttr(title)}">
                 <span class="progress-fill"${progressStyle}></span>
                 <span class="badge-content">
                     <span class="badge-icon">${icon}</span>
-                    <span>${extractor.label || extractor.plugin || 'unknown'}</span>
+                    <span>${escapeHtml(extractor.label || extractor.plugin || 'unknown')}</span>
                     ${pidHtml}
                 </span>
-            </span>
+            </${tag}>
         `;
     }
 
     function renderSnapshot(snapshot, crawlId) {
-        const statusIcon = snapshot.status === 'started' ? '&#8635;' : '&#128196;';
-        const adminUrl = `/admin/core/snapshot/${snapshot.id || 'unknown'}/change/`;
+        const statusIcon = snapshot.status === 'started' ? '▤' : '▢';
+        const adminUrl = snapshot.admin_url || `/admin/core/snapshot/${snapshot.id || 'unknown'}/change/`;
         const canCancel = snapshot.status === 'queued';
         const cancelBtn = canCancel
             ? `<button class="cancel-item-btn" data-cancel-type="snapshot" data-snapshot-id="${snapshot.id}" data-label="✕" title="Cancel snapshot">✕</button>`
             : '';
         const snapshotPidHtml = snapshot.worker_pid ? `<span class="pid-label compact">pid ${snapshot.worker_pid}</span>` : '';
+        const titleText = snapshot.title || formatUrl(snapshot.full_url || snapshot.url);
+        const urlText = snapshot.full_url || snapshot.url || '';
+        const faviconHtml = snapshot.favicon_url
+            ? `<img class="snapshot-favicon" src="${escapeAttr(snapshot.favicon_url)}" alt="" decoding="async" loading="lazy" onerror="this.remove()">`
+            : '';
+        const previewHtml = snapshot.preview_url
+            ? `<a class="snapshot-preview" href="${escapeAttr(snapshot.preview_link || snapshot.view_url || adminUrl)}" title="Open snapshot output"><img src="${escapeAttr(snapshot.preview_url)}" alt="" decoding="async" loading="lazy" data-fallbacks="${escapeAttr((snapshot.preview_fallbacks || []).join(','))}" onerror="nextPreviewFallback(this)"></a>`
+            : `<a class="snapshot-preview placeholder" href="${escapeAttr(snapshot.view_url || adminUrl)}" title="Open snapshot"><span>${statusIcon}</span></a>`;
 
         let extractorHtml = '';
         if (snapshot.all_plugins && snapshot.all_plugins.length > 0) {
@@ -757,15 +825,25 @@
                 : hasProcessEntries
                     ? `${snapshot.completed_plugins || 0}/${snapshot.total_plugins || 0} tasks${(snapshot.failed_plugins || 0) > 0 ? ` <span style="color:#f85149">(${snapshot.failed_plugins} failed)</span>` : ''}${runningProcessCount > 0 ? ` <span style="color:#d29922">(${runningProcessCount} hooks running)</span>` : ''}`
                     : `${snapshot.completed_plugins || 0}/${snapshot.total_plugins || 0} extractors${(snapshot.failed_plugins || 0) > 0 ? ` <span style="color:#f85149">(${snapshot.failed_plugins} failed)</span>` : ''}`
-            : 'Waiting for extractors...';
+            : snapshot.worker_state === 'crashed'
+                ? '<span style="color:#f85149">Worker stopped before extractors started</span>'
+                : snapshot.worker_state === 'stalled'
+                    ? '<span style="color:#d29922">Waiting for runner to resume</span>'
+                    : snapshot.worker_state === 'cancelled'
+                        ? '<span style="color:#8b949e">Cancelled before completion</span>'
+                        : 'Waiting for extractors...';
 
         return `
             <div class="snapshot-item">
                 <div class="snapshot-header">
+                    ${previewHtml}
                     <a class="snapshot-header-link" href="${adminUrl}">
-                        <span class="snapshot-icon">${statusIcon}</span>
                         <div class="snapshot-info">
-                            <div class="snapshot-url">Snapshot: ${formatUrl(snapshot.url)}</div>
+                            <div class="snapshot-title-line">
+                                ${faviconHtml}
+                                <span class="snapshot-title">${escapeHtml(titleText)}</span>
+                            </div>
+                            <div class="snapshot-url">${escapeHtml(urlText)}</div>
                             <div class="snapshot-meta">
                                 ${snapshotMeta}
                             </div>
@@ -787,7 +865,7 @@
     }
 
     function renderCrawl(crawl) {
-        const statusIcon = crawl.status === 'started' ? '&#8635;' : '&#128269;';
+        const statusIcon = crawl.status === 'started' ? '▦' : '▥';
         const adminUrl = `/admin/crawls/crawl/${crawl.id || 'unknown'}/change/`;
         const canCancel = crawl.status === 'queued' || crawl.status === 'started';
         const cancelBtn = canCancel
@@ -828,21 +906,39 @@
         if (crawl.status === 'queued' && !crawl.can_start) {
             warningHtml = `
                 <div style="padding: 8px 14px; background: rgba(248, 81, 73, 0.1); border-top: 1px solid #f85149; color: #f85149; font-size: 11px;">
-                    ⚠️ Crawl cannot start: ${crawl.urls_preview ? 'unknown error' : 'no URLs'}
+                    Crawl cannot start: ${crawl.urls_preview ? 'unknown error' : 'no URLs'}
                 </div>
             `;
         } else if (crawl.status === 'queued' && crawl.retry_at_future) {
             // Queued but retry_at is in future (was claimed by worker, will retry)
             warningHtml = `
                 <div style="padding: 8px 14px; background: rgba(88, 166, 255, 0.1); border-top: 1px solid #58a6ff; color: #58a6ff; font-size: 11px;">
-                    🔄 Trying in ${crawl.seconds_until_retry || 0}s...${crawl.urls_preview ? ` (${crawl.urls_preview})` : ''}
+                    Trying in ${crawl.seconds_until_retry || 0}s...${crawl.urls_preview ? ` (${escapeHtml(crawl.urls_preview)})` : ''}
                 </div>
             `;
         } else if (crawl.status === 'queued' && crawl.total_snapshots === 0) {
             // Queued and waiting to be picked up by worker
             warningHtml = `
                 <div style="padding: 8px 14px; background: rgba(210, 153, 34, 0.1); border-top: 1px solid #d29922; color: #d29922; font-size: 11px;">
-                    ⏳ Waiting for the runner to pick up...${crawl.urls_preview ? ` (${crawl.urls_preview})` : ''}
+                    Waiting for the runner to pick up...${crawl.urls_preview ? ` (${escapeHtml(crawl.urls_preview)})` : ''}
+                </div>
+            `;
+        } else if (crawl.status === 'started' && crawl.worker_state === 'crashed') {
+            warningHtml = `
+                <div style="padding: 8px 14px; background: rgba(248, 81, 73, 0.1); border-top: 1px solid #f85149; color: #f85149; font-size: 11px;">
+                    Runner stopped with ${crawl.started_snapshots || 0} active and ${crawl.pending_snapshots || 0} pending snapshots. It will resume when the runner starts again.
+                </div>
+            `;
+        } else if (crawl.status === 'started' && crawl.worker_state === 'stalled') {
+            warningHtml = `
+                <div style="padding: 8px 14px; background: rgba(210, 153, 34, 0.1); border-top: 1px solid #d29922; color: #d29922; font-size: 11px;">
+                    Runner is online but no worker is attached to this crawl yet.
+                </div>
+            `;
+        } else if (crawl.worker_state === 'cancelled') {
+            warningHtml = `
+                <div style="padding: 8px 14px; background: rgba(139, 148, 158, 0.1); border-top: 1px solid #6e7681; color: #8b949e; font-size: 11px;">
+                    Crawl was cancelled. ${crawl.cancelled_snapshots || 0} snapshot${(crawl.cancelled_snapshots || 0) === 1 ? '' : 's'} stopped before completion.
                 </div>
             `;
         }
@@ -854,22 +950,23 @@
         } else if ((crawl.urls_count || 0) > 0) {
             metaText += ` | ${crawl.urls_count} URLs`;
         } else if (crawl.urls_preview) {
-            metaText += ` | ${crawl.urls_preview.substring(0, 40)}${crawl.urls_preview.length > 40 ? '...' : ''}`;
+            metaText += ` | ${escapeHtml(crawl.urls_preview.substring(0, 40))}${crawl.urls_preview.length > 40 ? '...' : ''}`;
         }
 
         return `
             <div class="crawl-item" data-crawl-id="${crawl.id || 'unknown'}">
                 <div class="crawl-header">
                     <a class="crawl-header-link" href="${adminUrl}">
-                        <span class="crawl-icon">${statusIcon}</span>
-                        <div class="crawl-info">
-                            <div class="crawl-label">Crawl: ${crawl.label || '(no label)'}</div>
+                            <span class="crawl-icon">${statusIcon}</span>
+                            <div class="crawl-info">
+                            <div class="crawl-label">Crawl: ${escapeHtml(crawl.label || '(no label)')}</div>
                             <div class="crawl-meta">${metaText}</div>
                         </div>
                         <div class="crawl-stats">
                             <span style="color:#3fb950">${crawl.completed_snapshots || 0} done</span>
                             <span style="color:#d29922">${crawl.started_snapshots || 0} active</span>
                             <span style="color:#8b949e">${crawl.pending_snapshots || 0} pending</span>
+                            ${(crawl.cancelled_snapshots || 0) > 0 ? `<span style="color:#8b949e">${crawl.cancelled_snapshots} cancelled</span>` : ''}
                         </div>
                         ${crawlPidHtml}
                         <span class="status-badge ${crawl.status || 'unknown'}">${crawl.status || 'unknown'}</span>
@@ -921,11 +1018,20 @@
         const text = document.getElementById('orchestrator-text');
         const pidEl = document.getElementById('orchestrator-pid');
         const hasWorkers = data.total_workers > 0;
+        const hasBlockedCrawl = (data.active_crawls || []).some(c => c.worker_state === 'crashed' || c.worker_state === 'stalled');
 
-        if (hasWorkers || hasActivity) {
+        if (hasWorkers) {
             dot.classList.remove('stopped', 'idle');
             dot.classList.add('running');
             text.textContent = 'Running';
+        } else if (hasActivity && hasBlockedCrawl) {
+            dot.classList.remove('idle', 'running');
+            dot.classList.add('stopped');
+            text.textContent = 'Runner stopped';
+        } else if (hasActivity) {
+            dot.classList.remove('stopped', 'running');
+            dot.classList.add('idle');
+            text.textContent = 'Waiting';
         } else {
             // No activity - show as idle (whether orchestrator process exists or not)
             dot.classList.remove('stopped', 'running');
@@ -950,22 +1056,8 @@
         document.getElementById('total-queued').textContent =
             data.crawls_pending + data.snapshots_pending + data.archiveresults_pending;
 
-        // Store raw values and display relative to baseline
-        lastSucceeded = data.archiveresults_succeeded;
-        lastFailed = data.archiveresults_failed;
-
-        // If baseline is higher than current (e.g. after DB reset), reset baseline
-        if (succeededBaseline > lastSucceeded) {
-            succeededBaseline = 0;
-            localStorage.setItem('progress-succeeded-baseline', '0');
-        }
-        if (failedBaseline > lastFailed) {
-            failedBaseline = 0;
-            localStorage.setItem('progress-failed-baseline', '0');
-        }
-
-        document.getElementById('total-succeeded').textContent = lastSucceeded - succeededBaseline;
-        document.getElementById('total-failed').textContent = lastFailed - failedBaseline;
+        document.getElementById('total-succeeded').textContent = data.archiveresults_succeeded;
+        document.getElementById('total-failed').textContent = data.archiveresults_failed;
 
         // Render crawl tree
         if (data.active_crawls.length > 0) {

archivebox/tests/test_cli_real_flows.py

@@ -3,15 +3,178 @@
 
 import json
 import os
+import signal
 import sqlite3
 import subprocess
 import sys
+import time
 
 import pytest
 
 from .conftest import _find_system_browser
 
 
+def _pid_is_alive(pid: int) -> bool:
+    try:
+        os.kill(pid, 0)
+    except ProcessLookupError:
+        return False
+    except PermissionError:
+        return True
+    return True
+
+
+def _wait_for_pid_exit(pid: int, *, timeout: float = 5.0) -> None:
+    deadline = time.time() + timeout
+    while time.time() < deadline:
+        if not _pid_is_alive(pid):
+            return
+        time.sleep(0.05)
+    raise AssertionError(f"PID {pid} is still alive")
+
+
+def _cleanup_process_group(group_pid: int | None, *child_pids: int | None) -> None:
+    if group_pid and _pid_is_alive(group_pid):
+        try:
+            os.killpg(group_pid, signal.SIGKILL)
+        except ProcessLookupError:
+            pass
+        except OSError:
+            try:
+                os.kill(group_pid, signal.SIGKILL)
+            except ProcessLookupError:
+                pass
+    for pid in child_pids:
+        if pid and _pid_is_alive(pid):
+            try:
+                os.kill(pid, signal.SIGKILL)
+            except ProcessLookupError:
+                pass
+
+
+@pytest.mark.timeout(90)
+def test_cli_run_signal_cleans_background_hook_process_group(tmp_path, process):
+    os.chdir(tmp_path)
+    assert process.returncode == 0, process.stderr
+
+    plugins_root = tmp_path / "runtime_plugins"
+    plugin_dir = plugins_root / "cancel_group"
+    plugin_dir.mkdir(parents=True)
+    daemon_hook = plugin_dir / "on_CrawlSetup__10_daemon.daemon.bg.sh"
+    foreground_hook = plugin_dir / "on_CrawlSetup__20_foreground.sh"
+    daemon_hook.write_text(
+        "\n".join(
+            [
+                "#!/usr/bin/env bash",
+                "set -euo pipefail",
+                'test_dir="${LEAK_TEST_DIR:?}"',
+                "sleep 600 &",
+                'echo $$ > "$test_dir/daemon.pid"',
+                'echo $! > "$test_dir/daemon-child.pid"',
+                'echo ready > "$test_dir/daemon.ready"',
+                "trap 'echo cleaned > \"$test_dir/daemon.cleaned\"; exit 0' TERM INT",
+                "wait",
+                "",
+            ],
+        ),
+    )
+    foreground_hook.write_text(
+        "\n".join(
+            [
+                "#!/usr/bin/env bash",
+                "set -euo pipefail",
+                'test_dir="${LEAK_TEST_DIR:?}"',
+                'echo $$ > "$test_dir/foreground.pid"',
+                'echo ready > "$test_dir/foreground.ready"',
+                "trap 'echo cleaned > \"$test_dir/foreground.cleaned\"; exit 0' TERM INT",
+                "while true; do sleep 1; done",
+                "",
+            ],
+        ),
+    )
+    daemon_hook.chmod(0o755)
+    foreground_hook.chmod(0o755)
+
+    leak_test_dir = tmp_path / "leak-check"
+    leak_test_dir.mkdir()
+    env = os.environ.copy()
+    env.update(
+        {
+            "ABX_PLUGINS_DIR": str(plugins_root),
+            "LEAK_TEST_DIR": str(leak_test_dir),
+            "PLUGINS": "cancel_group",
+            "TIMEOUT": "30",
+            "USE_COLOR": "false",
+            "SHOW_PROGRESS": "false",
+        },
+    )
+
+    create_result = subprocess.run(
+        [sys.executable, "-m", "archivebox", "crawl", "create", "https://example.com"],
+        cwd=tmp_path,
+        capture_output=True,
+        text=True,
+        env=env,
+        timeout=60,
+    )
+    assert create_result.returncode == 0, create_result.stderr or create_result.stdout
+    crawl_records = [json.loads(line) for line in create_result.stdout.splitlines() if line.strip().startswith("{")]
+    crawl_id = next(record["id"] for record in crawl_records if record.get("type") == "Crawl")
+
+    daemon_pid: int | None = None
+    daemon_child_pid: int | None = None
+    foreground_pid: int | None = None
+    run_process = subprocess.Popen(
+        [sys.executable, "-m", "archivebox", "run", f"--crawl-id={crawl_id}"],
+        cwd=tmp_path,
+        env=env,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,
+        text=True,
+        start_new_session=True,
+    )
+    try:
+        deadline = time.time() + 20
+        while time.time() < deadline:
+            if (leak_test_dir / "daemon.ready").exists() and (leak_test_dir / "foreground.ready").exists():
+                break
+            if run_process.poll() is not None:
+                output = run_process.communicate(timeout=1)[0]
+                raise AssertionError(f"archivebox run exited before hooks were ready:\n{output}")
+            time.sleep(0.05)
+        assert (leak_test_dir / "daemon.ready").exists()
+        assert (leak_test_dir / "foreground.ready").exists()
+
+        daemon_pid = int((leak_test_dir / "daemon.pid").read_text().strip())
+        daemon_child_pid = int((leak_test_dir / "daemon-child.pid").read_text().strip())
+        foreground_pid = int((leak_test_dir / "foreground.pid").read_text().strip())
+        assert _pid_is_alive(daemon_pid)
+        assert _pid_is_alive(daemon_child_pid)
+        assert _pid_is_alive(foreground_pid)
+
+        run_process.send_signal(signal.SIGTERM)
+        time.sleep(0.1)
+        if run_process.poll() is None:
+            run_process.send_signal(signal.SIGTERM)
+        output = run_process.communicate(timeout=20)[0]
+        assert "Runner error" not in output
+
+        _wait_for_pid_exit(daemon_pid)
+        _wait_for_pid_exit(daemon_child_pid)
+        _wait_for_pid_exit(foreground_pid)
+        assert (leak_test_dir / "daemon.cleaned").read_text().strip() == "cleaned"
+        assert (leak_test_dir / "foreground.cleaned").read_text().strip() == "cleaned"
+    finally:
+        if run_process.poll() is None:
+            try:
+                os.killpg(run_process.pid, signal.SIGKILL)
+            except ProcessLookupError:
+                pass
+            run_process.communicate(timeout=5)
+        _cleanup_process_group(daemon_pid, daemon_child_pid)
+        _cleanup_process_group(foreground_pid)
+
+
 @pytest.mark.timeout(180)
 def test_cli_add_real_urls_with_options_writes_inspectable_outputs(tmp_path, process):
     os.chdir(tmp_path)

archivebox/tests/test_machine_models.py

@@ -12,9 +12,12 @@ Tests cover:
 """
 
 import os
+import subprocess
+import sys
+import tempfile
 from datetime import timedelta
+from pathlib import Path
 from typing import cast
-from unittest.mock import Mock, patch
 
 import pytest
 from django.test import TestCase
@@ -94,10 +97,16 @@ class TestMachineModel(TestCase):
 
     def test_machine_from_jsonl_update(self):
         """Machine.from_json() should update machine config."""
+        from archivebox.config.constants import CONSTANTS
+
         Machine.current()  # Ensure machine exists
+        wget_path = CONSTANTS.DEFAULT_LIB_DIR / "wget"
+        wget_path.parent.mkdir(parents=True, exist_ok=True)
+        wget_path.write_text("#!/bin/sh\n")
+        self.addCleanup(lambda: wget_path.exists() and wget_path.unlink())
         record = {
             "config": {
-                "WGET_BINARY": "/usr/bin/wget",
+                "WGET_BINARY": str(wget_path),
             },
         }
 
@@ -105,15 +114,22 @@ class TestMachineModel(TestCase):
 
         self.assertIsNotNone(result)
         assert result is not None
-        self.assertEqual(result.config.get("WGET_BINARY"), "/usr/bin/wget")
+        self.assertEqual(result.config.get("WGET_BINARY"), str(wget_path))
+
+    def test_machine_from_jsonl_keeps_only_valid_binary_paths(self):
+        """Machine.from_json() should persist only valid LIB_DIR binary paths."""
+        from archivebox.config.constants import CONSTANTS
 
-    def test_machine_from_jsonl_strips_legacy_chromium_version(self):
-        """Machine.from_json() should ignore legacy browser version keys."""
         Machine.current()  # Ensure machine exists
+        wget_path = CONSTANTS.DEFAULT_LIB_DIR / "wget"
+        wget_path.parent.mkdir(parents=True, exist_ok=True)
+        wget_path.write_text("#!/bin/sh\n")
+        self.addCleanup(lambda: wget_path.exists() and wget_path.unlink())
         record = {
             "config": {
-                "WGET_BINARY": "/usr/bin/wget",
+                "WGET_BINARY": str(wget_path),
                 "CHROMIUM_VERSION": "123.4.5",
+                "YTDLP_BINARY": "/tmp/archivebox-test-missing-yt-dlp",
             },
         }
 
@@ -121,8 +137,9 @@ class TestMachineModel(TestCase):
 
         self.assertIsNotNone(result)
         assert result is not None
-        self.assertEqual(result.config.get("WGET_BINARY"), "/usr/bin/wget")
+        self.assertEqual(result.config.get("WGET_BINARY"), str(wget_path))
         self.assertNotIn("CHROMIUM_VERSION", result.config)
+        self.assertNotIn("YTDLP_BINARY", result.config)
 
     def test_machine_from_jsonl_invalid(self):
         """Machine.from_json() should return None for invalid records."""
@@ -132,21 +149,28 @@ class TestMachineModel(TestCase):
     def test_machine_current_keeps_only_derived_runtime_cache(self):
         """Machine.current() should keep derived cache entries, not runtime config."""
         import archivebox.machine.models as models
+        from archivebox.config.constants import CONSTANTS
 
-        chrome_path = "/tmp/archivebox-test-chromium"
-        node_path = "/tmp/archivebox-test-node"
-        open(chrome_path, "a").close()
-        open(node_path, "a").close()
-        self.addCleanup(lambda: os.path.exists(chrome_path) and os.remove(chrome_path))
-        self.addCleanup(lambda: os.path.exists(node_path) and os.remove(node_path))
+        active_lib_dir = CONSTANTS.DEFAULT_LIB_DIR
+        active_lib_dir.mkdir(parents=True, exist_ok=True)
+        chrome_path = active_lib_dir / "chromium"
+        node_path = active_lib_dir / "node"
+        chrome_path.write_text("#!/bin/sh\n")
+        node_path.write_text("#!/bin/sh\n")
+        external_path = "/tmp/archivebox-test-external-node"
+        open(external_path, "a").close()
+        self.addCleanup(lambda: chrome_path.exists() and chrome_path.unlink())
+        self.addCleanup(lambda: node_path.exists() and node_path.unlink())
+        self.addCleanup(lambda: os.path.exists(external_path) and os.remove(external_path))
         machine = Machine.current()
         machine.config = {
-            "CHROME_BINARY": chrome_path,
-            "NODE_BINARY": node_path,
+            "CHROME_BINARY": str(chrome_path),
+            "NODE_BINARY": str(node_path),
             "ABX_INSTALL_CACHE": {"wget": "2026-03-24T00:00:00+00:00"},
             "CHROME_ISOLATION": "snapshot",
             "CHROME_USER_DATA_DIR": "/tmp/profile",
             "CHROMIUM_VERSION": "123.4.5",
+            "YTDLP_BINARY": external_path,
             "WGET_BINARY": "/tmp/archivebox-test-missing-wget",
         }
         machine.save(update_fields=["config"])
@@ -154,12 +178,13 @@ class TestMachineModel(TestCase):
 
         refreshed = Machine.current()
 
-        self.assertEqual(refreshed.config.get("CHROME_BINARY"), chrome_path)
-        self.assertEqual(refreshed.config.get("NODE_BINARY"), node_path)
-        self.assertEqual(refreshed.config.get("ABX_INSTALL_CACHE"), {"wget": "2026-03-24T00:00:00+00:00"})
+        self.assertEqual(refreshed.config.get("CHROME_BINARY"), str(chrome_path))
+        self.assertEqual(refreshed.config.get("NODE_BINARY"), str(node_path))
+        self.assertNotIn("ABX_INSTALL_CACHE", refreshed.config)
         self.assertNotIn("CHROME_ISOLATION", refreshed.config)
         self.assertNotIn("CHROME_USER_DATA_DIR", refreshed.config)
         self.assertNotIn("CHROMIUM_VERSION", refreshed.config)
+        self.assertNotIn("YTDLP_BINARY", refreshed.config)
         self.assertNotIn("WGET_BINARY", refreshed.config)
 
     def test_get_config_auto_applies_current_machine_config(self):
@@ -167,12 +192,14 @@ class TestMachineModel(TestCase):
         import archivebox.machine.models as models
         from archivebox.config.common import get_config
 
-        chrome_path = "/tmp/archivebox-test-chromium"
-        open(chrome_path, "a").close()
-        self.addCleanup(lambda: os.path.exists(chrome_path) and os.remove(chrome_path))
+        lib_dir = get_config(include_machine=False).LIB_DIR
+        chrome_path = lib_dir / "chromium"
+        chrome_path.parent.mkdir(parents=True, exist_ok=True)
+        chrome_path.write_text("#!/bin/sh\n")
+        self.addCleanup(lambda: chrome_path.exists() and chrome_path.unlink())
         machine = Machine.current()
         machine.config = {
-            "CHROME_BINARY": chrome_path,
+            "CHROME_BINARY": str(chrome_path),
             "ABX_INSTALL_CACHE": {"chrome": "2026-03-24T00:00:00+00:00"},
             "CHROME_ISOLATION": "snapshot",
         }
@@ -181,8 +208,7 @@ class TestMachineModel(TestCase):
 
         config = get_config()
 
-        self.assertEqual(config.CHROME_BINARY, chrome_path)
-        self.assertEqual(config["ABX_INSTALL_CACHE"], {"chrome": "2026-03-24T00:00:00+00:00"})
+        self.assertEqual(config.CHROME_BINARY, str(chrome_path))
         self.assertEqual(config.CHROME_ISOLATION, "crawl")
 
     def test_machine_manager_current(self):
@@ -223,36 +249,6 @@ class TestNetworkInterfaceModel(TestCase):
         interface = NetworkInterface.current()
         self.assertIsNotNone(interface)
 
-    def test_networkinterface_current_refresh_creates_new_interface_when_properties_change(self):
-        """Refreshing should persist a new NetworkInterface row when the host network fingerprint changes."""
-        import archivebox.machine.models as models
-
-        first = {
-            "mac_address": "aa:bb:cc:dd:ee:01",
-            "ip_public": "1.1.1.1",
-            "ip_local": "192.168.1.10",
-            "dns_server": "8.8.8.8",
-            "hostname": "host-a",
-            "iface": "en0",
-            "isp": "ISP A",
-            "city": "City",
-            "region": "Region",
-            "country": "Country",
-        }
-        second = {
-            **first,
-            "ip_public": "2.2.2.2",
-            "ip_local": "10.0.0.5",
-        }
-
-        with patch.object(models, "get_host_network", side_effect=[first, second]):
-            interface1 = NetworkInterface.current(refresh=True)
-            interface2 = NetworkInterface.current(refresh=True)
-
-        self.assertNotEqual(interface1.id, interface2.id)
-        self.assertEqual(interface1.machine_id, interface2.machine_id)
-        self.assertEqual(NetworkInterface.objects.filter(machine=interface1.machine).count(), 2)
-
 
 class TestBinaryModel(TestCase):
     """Test the Binary model."""
@@ -344,6 +340,20 @@ class TestBinaryModel(TestCase):
         assert binary is not None
         self.assertEqual(binary.overrides, overrides)
 
+    def test_binary_from_json_canonicalizes_path_like_names(self):
+        """Binary.from_json() should store command names, not path cache values."""
+        binary = Binary.from_json(
+            {
+                "name": "/tmp/old-lib/pip/venv/bin/trafilatura",
+                "binproviders": "env,pip",
+                "overrides": {"pip": {"install_args": ["trafilatura"]}},
+            },
+        )
+
+        self.assertIsNotNone(binary)
+        assert binary is not None
+        self.assertEqual(binary.name, "trafilatura")
+
     def test_binary_from_json_does_not_coerce_legacy_override_shapes(self):
         """Binary.from_json() should no longer translate legacy non-dict provider overrides."""
         overrides = {
@@ -507,48 +517,82 @@ class TestProcessCurrent(TestCase):
 
     def test_process_detect_type_runner(self):
         """_detect_process_type should detect the background runner command."""
-        with patch("sys.argv", ["archivebox", "run", "--daemon"]):
+        old_argv = sys.argv
+        try:
+            sys.argv = ["archivebox", "run", "--daemon"]
             result = Process._detect_process_type()
             self.assertEqual(result, Process.TypeChoices.ORCHESTRATOR)
+        finally:
+            sys.argv = old_argv
 
     def test_process_detect_type_runner_watch(self):
         """runner_watch should be classified as a worker, not the orchestrator itself."""
-        with patch("sys.argv", ["archivebox", "manage", "runner_watch", "--pidfile=/tmp/runserver.pid"]):
+        old_argv = sys.argv
+        try:
+            sys.argv = ["archivebox", "manage", "runner_watch", "--pidfile=/tmp/runserver.pid"]
             result = Process._detect_process_type()
             self.assertEqual(result, Process.TypeChoices.WORKER)
+        finally:
+            sys.argv = old_argv
 
     def test_process_detect_type_cli(self):
         """_detect_process_type should detect CLI commands."""
-        with patch("sys.argv", ["archivebox", "add", "http://example.com"]):
+        old_argv = sys.argv
+        try:
+            sys.argv = ["archivebox", "add", "http://example.com"]
             result = Process._detect_process_type()
             self.assertEqual(result, Process.TypeChoices.CLI)
+        finally:
+            sys.argv = old_argv
 
     def test_process_detect_type_binary(self):
         """_detect_process_type should detect non-ArchiveBox subprocesses as binary processes."""
-        with patch("sys.argv", ["/usr/bin/wget", "https://example.com"]):
+        old_argv = sys.argv
+        try:
+            sys.argv = ["/usr/bin/wget", "https://example.com"]
             result = Process._detect_process_type()
             self.assertEqual(result, Process.TypeChoices.BINARY)
+        finally:
+            sys.argv = old_argv
 
     def test_process_proc_allows_interpreter_wrapped_script(self):
         """Process.proc should accept a script recorded in DB when wrapped by an interpreter in psutil."""
-        proc = Process.objects.create(
-            machine=Machine.current(),
-            cmd=["/tmp/on_CrawlSetup__90_chrome_launch.daemon.bg.js", "--url=https://example.com/"],
-            pid=12345,
-            status=Process.StatusChoices.RUNNING,
-            started_at=timezone.now(),
+        import psutil
+
+        temp_dir = tempfile.TemporaryDirectory()
+        self.addCleanup(temp_dir.cleanup)
+        script = Path(temp_dir.name) / "on_CrawlSetup__90_chrome_launch.daemon.bg.py"
+        script.write_text("import time\ntime.sleep(30)\n", encoding="utf-8")
+        process = subprocess.Popen(
+            [sys.executable, str(script), "--url=https://example.com/"],
+            stdin=subprocess.DEVNULL,
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
         )
 
-        os_proc = Mock()
-        os_proc.create_time.return_value = proc.started_at.timestamp()
-        os_proc.cmdline.return_value = [
-            "node",
-            "/tmp/on_CrawlSetup__90_chrome_launch.daemon.bg.js",
-            "--url=https://example.com/",
-        ]
+        def cleanup_process():
+            if process.poll() is None:
+                process.terminate()
+                try:
+                    process.wait(timeout=5)
+                except subprocess.TimeoutExpired:
+                    process.kill()
+                    process.wait(timeout=5)
 
-        with patch("archivebox.machine.models.psutil.Process", return_value=os_proc):
-            self.assertIs(proc.proc, os_proc)
+        self.addCleanup(cleanup_process)
+        os_proc = psutil.Process(process.pid)
+        proc = Process.objects.create(
+            machine=Machine.current(),
+            cmd=[str(script), "--url=https://example.com/"],
+            pid=process.pid,
+            status=Process.StatusChoices.RUNNING,
+            started_at=timezone.datetime.fromtimestamp(os_proc.create_time(), tz=timezone.get_current_timezone()),
+        )
+
+        resolved_proc = proc.proc
+        self.assertIsNotNone(resolved_proc)
+        assert resolved_proc is not None
+        self.assertEqual(resolved_proc.pid, process.pid)
 
 
 class TestProcessHierarchy(TestCase):
@@ -789,18 +833,11 @@ class TestProcessClassMethods(TestCase):
             started_at=timezone.now() - PROCESS_TIMEOUT_GRACE - timedelta(seconds=10),
         )
 
-        with (
-            patch.object(Process, "poll", return_value=None),
-            patch.object(Process, "kill_tree") as kill_tree,
-            patch.object(Process, "terminate") as terminate,
-        ):
-            cleaned = Process.cleanup_stale_running()
+        cleaned = Process.cleanup_stale_running()
 
         self.assertGreaterEqual(cleaned, 1)
         stale.refresh_from_db()
         self.assertEqual(stale.status, Process.StatusChoices.EXITED)
-        kill_tree.assert_not_called()
-        terminate.assert_not_called()
 
     def test_cleanup_orphaned_workers_marks_dead_root_children_exited(self):
         """cleanup_orphaned_workers should retire rows whose CLI/orchestrator root is gone."""
@@ -824,14 +861,11 @@ class TestProcessClassMethods(TestCase):
             started_at=started_at,
         )
 
-        with patch.object(Process, "kill_tree") as kill_tree, patch.object(Process, "terminate") as terminate:
-            cleaned = Process.cleanup_orphaned_workers()
+        cleaned = Process.cleanup_orphaned_workers()
 
         self.assertEqual(cleaned, 1)
         child.refresh_from_db()
         self.assertEqual(child.status, Process.StatusChoices.EXITED)
-        kill_tree.assert_not_called()
-        terminate.assert_not_called()
 
     def test_cleanup_orphaned_workers_marks_non_running_children_exited(self):
         """cleanup_orphaned_workers should retire child rows whose OS process is already gone."""

etc/package.json

@@ -1,6 +1,6 @@
 {
   "name": "archivebox",
-  "version": "0.9.32rc1",
+  "version": "0.9.32rc4",
   "repository": "github:ArchiveBox/ArchiveBox",
   "license": "MIT",
   "dependencies": {

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "archivebox"
-version = "0.9.32rc1"
+version = "0.9.32rc4"
 requires-python = ">=3.13"
 description = "Self-hosted internet archiving solution."
 authors = [{name = "Nick Sweeting", email = "pyproject.toml@archivebox.io"}]
@@ -78,10 +78,10 @@ dependencies = [
     "w3lib>=2.2.1",          # used for parsing content-type encoding from http response headers & html tags
     ### Extractor dependencies (optional binaries detected at runtime via shutil.which)
     ### Binary/Package Management
-    "abxbus>=2.5.4",          # EventBus API
-    "abxpkg>=1.11.1",         # for: detecting, versioning, and installing binaries via apt/brew/pip/npm
-    "abx-plugins>=1.11.1",   # shared ArchiveBox plugin package with Chrome/Puppeteer dependency wiring
-    "abx-dl>=1.11.1",        # shared ArchiveBox downloader package with blocking install preflight
+    "abxbus==2.5.7",          # EventBus API
+    "abxpkg>=1.11.8",         # for: detecting, versioning, and installing binaries via apt/brew/pip/npm
+    "abx-plugins>=1.11.9",   # shared ArchiveBox plugin package with Chrome/Puppeteer dependency wiring
+    "abx-dl>=1.11.9",        # shared ArchiveBox downloader package with blocking install preflight
     ### UUID7 backport for Python <3.14
     "uuid7>=0.1.0; python_version < '3.14'",  # provides the uuid_extensions module on Python 3.13
 ]
@@ -167,7 +167,12 @@ build-backend = "pdm.backend"
 includes = ["archivebox/"]
 source-includes = []
 excludes = [
-    "**/.DS_Store",
+    "archivebox/**/.DS_Store",
+    "archivebox/**/__pycache__",
+    "archivebox/**/*.pyc",
+    "archivebox/**/*.pyo",
+    "archivebox/**/*.sqlite3",
+    "archivebox/**/*.sqlite3-*",
     "archivebox/tests/",
     "archivebox/tests/**",
     "tests/",