release: archivebox 0.9.33rc29

archivebox/api/urls.py

@@ -8,7 +8,6 @@ from django.urls import path
 from django.views.generic.base import RedirectView
 
 from archivebox.core.host_utils import build_web_url
-
 from .v1_api import urls as v1_api_urls
 
 

archivebox/api/v1_crawls.py

@@ -1,8 +1,9 @@
 __package__ = "archivebox.api"
 
+from pathlib import Path
 from uuid import UUID
 from datetime import datetime
-from django.http import HttpRequest
+from django.http import FileResponse, HttpRequest
 from django.shortcuts import redirect
 from django.utils import timezone
 
@@ -15,7 +16,7 @@ from ninja.errors import HttpError
 from archivebox.core.models import Snapshot
 from archivebox.crawls.models import Crawl
 
-from .auth import API_AUTH_METHODS
+from .auth import API_AUTH_METHODS, auth_using_token
 
 router = Router(tags=["Crawl Models"], auth=API_AUTH_METHODS)
 
@@ -117,6 +118,8 @@ def create_crawl(request: HttpRequest, data: CrawlCreateSchema):
         created_by=request.user if isinstance(request.user, User) else None,
     )
     crawl.create_snapshots_from_urls()
+    if not crawl.snapshot_set.exists():
+        crawl.sm.seal()
     return crawl
 
 
@@ -136,6 +139,50 @@ def get_crawl(request: HttpRequest, crawl_id: str, as_rss: bool = False, with_sn
     return crawl
 
 
+def crawl_file(request: HttpRequest, crawl_id: str, path: str):
+    user = getattr(request, "user", None)
+    is_superuser = bool(
+        getattr(user, "is_authenticated", False) and getattr(user, "is_active", False) and getattr(user, "is_superuser", False),
+    )
+    if not is_superuser:
+        token = request.GET.get("api_key") or request.headers.get("X-ArchiveBox-API-Key")
+        auth_header = request.headers.get("Authorization", "")
+        if not token and auth_header.lower().startswith("bearer "):
+            token = auth_header.split(None, 1)[1].strip()
+        token_user = auth_using_token(token=token, request=request) if token else None
+        is_superuser = bool(token_user and token_user.is_active and token_user.is_superuser)
+    if not is_superuser:
+        raise HttpError(403, "Permission denied")
+
+    crawl = Crawl.objects.get(id__icontains=crawl_id)
+    crawl_root = Path(crawl.output_dir).resolve()
+    file_path = (crawl_root / path).resolve()
+    if not file_path.is_file() or crawl_root not in file_path.parents:
+        raise HttpError(404, "Crawl file not found")
+
+    response = FileResponse(file_path.open("rb"))
+    response["Cache-Control"] = "no-store, no-cache, max-age=0, must-revalidate"
+    response["Pragma"] = "no-cache"
+    response["Expires"] = "0"
+    response["X-Content-Type-Options"] = "nosniff"
+    return response
+
+
+@router.get("/crawl/{crawl_id}/files/{filename}", auth=None, url_name="crawl_file_root")
+def crawl_file_root(request: HttpRequest, crawl_id: str, filename: str):
+    return crawl_file(request, crawl_id, filename)
+
+
+@router.get("/crawl/{crawl_id}/files/{folder}/{filename}", auth=None, url_name="crawl_file_nested_1")
+def crawl_file_nested_1(request: HttpRequest, crawl_id: str, folder: str, filename: str):
+    return crawl_file(request, crawl_id, f"{folder}/{filename}")
+
+
+@router.get("/crawl/{crawl_id}/files/{folder}/{subfolder}/{filename}", auth=None, url_name="crawl_file_nested_2")
+def crawl_file_nested_2(request: HttpRequest, crawl_id: str, folder: str, subfolder: str, filename: str):
+    return crawl_file(request, crawl_id, f"{folder}/{subfolder}/{filename}")
+
+
 @router.patch("/crawl/{crawl_id}", response=CrawlSchema, url_name="patch_crawl")
 def patch_crawl(request: HttpRequest, crawl_id: str, data: CrawlUpdateSchema):
     """Update a crawl (e.g., set status=sealed to cancel queued work)."""

archivebox/cli/archivebox_add.py

@@ -197,10 +197,14 @@ def add(
         # Just create the crawl but don't start processing
         print("[yellow]\\[*] Index-only mode - crawl created but not started[/yellow]")
         crawl.create_snapshots_from_urls()
+        if not crawl.snapshot_set.exists():
+            crawl.sm.seal()
         return crawl, crawl.snapshot_set.all()
 
     if bg:
         crawl.create_snapshots_from_urls()
+        if not crawl.snapshot_set.exists():
+            crawl.sm.seal()
 
     # 5. Start the crawl runner to process the queue
     #    The runner will:
@@ -225,10 +229,14 @@ def add(
         exit_code = 0
         try:
             try:
-                with foreground_shutdown_signals(), foreground_parent_watchdog():
+                with foreground_shutdown_signals(first_signal_message=None), foreground_parent_watchdog():
                     while True:
                         standby_until_runtime_stack_needed(command, data_dir=CONSTANTS.DATA_DIR)
-                        exit_code = run_runner_worker(["--crawl-id", str(crawl.id)], name=f"worker_runner_add_{os.getpid()}")
+                        exit_code = run_runner_worker(
+                            ["--crawl-id", str(crawl.id)],
+                            name=f"worker_runner_add_{os.getpid()}",
+                            interactive_interrupts=True,
+                        )
                         if exit_code == 0:
                             break
                         if not command_owns_runtime_stack(command, data_dir=CONSTANTS.DATA_DIR):

archivebox/cli/archivebox_run.py

@@ -263,9 +263,15 @@ def run_runner(daemon: bool = False, crawl_id: str | None = None, maintenance_on
         rprint(f"[green][*] Existing ArchiveBox orchestrator pid={existing_pid} is already running.[/green]", file=sys.stderr)
         return 0
     current.mark_running(process_type=Process.TypeChoices.ORCHESTRATOR, pwd=str(CONSTANTS.DATA_DIR), timeout=0)
+    interactive_interrupts = current.root.process_type == Process.TypeChoices.ADD
     try:
         with foreground_shutdown_signals(), foreground_parent_watchdog(enabled=not daemon):
-            run_pending_crawls(daemon=daemon, crawl_id=crawl_id, maintenance_only=maintenance_only)
+            run_pending_crawls(
+                daemon=daemon,
+                crawl_id=crawl_id,
+                maintenance_only=maintenance_only,
+                interactive_interrupts=interactive_interrupts,
+            )
         return 0
     except (KeyboardInterrupt, asyncio.CancelledError):
         return 0

archivebox/cli/archivebox_server.py

@@ -4,6 +4,9 @@ __package__ = "archivebox.cli"
 
 import sys
 import os
+import socket
+import subprocess
+import time
 from collections.abc import Iterable
 
 import rich_click as click
@@ -22,7 +25,7 @@ def server(
     nothreading: bool = False,
 ) -> None:
     """Run the ArchiveBox HTTP server"""
-    from archivebox.config.common import get_config, rprint
+    from archivebox.config.common import get_config
 
     config = get_config()
     runserver_args = list(runserver_args or (config.BIND_ADDR,))
@@ -62,12 +65,53 @@ def server(
     except IndexError:
         pass
 
+    if daemonize and os.environ.get("ARCHIVEBOX_SERVER_DAEMON_CHILD") != "1":
+        from archivebox.config import CONSTANTS
+
+        log_path = CONSTANTS.LOGS_DIR / "server.log"
+        log_path.parent.mkdir(parents=True, exist_ok=True)
+        daemon_env = os.environ.copy()
+        daemon_env["ARCHIVEBOX_SERVER_DAEMON_CHILD"] = "1"
+        daemon_cmd = [sys.executable, "-m", "archivebox", "server"]
+        if debug:
+            daemon_cmd.append("--debug")
+        if reload:
+            daemon_cmd.append("--reload")
+        if nothreading:
+            daemon_cmd.append("--nothreading")
+        daemon_cmd.extend(runserver_args)
+        with log_path.open("a", encoding="utf-8") as log_file:
+            proc = subprocess.Popen(
+                daemon_cmd,
+                cwd=os.getcwd(),
+                env=daemon_env,
+                stdin=subprocess.DEVNULL,
+                stdout=log_file,
+                stderr=log_file,
+                start_new_session=True,
+            )
+        deadline = time.monotonic() + 30
+        while time.monotonic() < deadline:
+            if proc.poll() is not None:
+                print(f"[red][X] ArchiveBox daemon server exited early with code {proc.returncode}. See {log_path}[/red]")
+                sys.exit(proc.returncode or 1)
+            try:
+                with socket.create_connection((host, int(port)), timeout=0.25):
+                    break
+            except OSError:
+                time.sleep(0.1)
+        else:
+            print(f"[yellow][!] ArchiveBox daemon server pid={proc.pid} is still starting. See {log_path}[/yellow]")
+        return
+
     os.environ["BIND_ADDR"] = f"{host}:{port}"
     from archivebox.core.host_utils import build_admin_url
 
     admin_url = build_admin_url("/admin/")
 
     from archivebox.workers.supervisord_util import (
+        active_supervisord_runtime_components,
+        format_runtime_components,
         start_server_workers,
         stop_existing_supervisord_process,
         is_port_in_use,
@@ -105,11 +149,16 @@ def server(
             connections.close_all()
 
     try:
-        with foreground_shutdown_signals(), foreground_parent_watchdog():
+        with foreground_shutdown_signals(), foreground_parent_watchdog(enabled=os.environ.get("ARCHIVEBOX_SERVER_DAEMON_CHILD") != "1"):
             while True:
-                standby_until_runtime_stack_needed(command, data_dir=config.DATA_DIR)
-                sys.stdout.write(f"[*] ArchiveBox server parent pid={os.getpid()} is now running the orchestrator and server...\n")
-                sys.stdout.flush()
+                standby_result = standby_until_runtime_stack_needed(command, data_dir=config.DATA_DIR)
+                older_owner = runtime_stack_owner(data_dir=config.DATA_DIR, exclude_id=command.id)
+                takeover_components = active_supervisord_runtime_components(config=config)
+                if older_owner and takeover_components:
+                    print(
+                        "[yellow][*] Taking over "
+                        f"{format_runtime_components(takeover_components)} from older existing archivebox process (pid={older_owner.pid}).[/yellow]",
+                    )
                 stop_existing_supervisord_process()
                 if is_port_in_use(host, int(port)):
                     print(f"[red][X] Error: Port {port} is already in use[/red]")
@@ -119,23 +168,17 @@ def server(
                 result = start_server_workers(
                     host=host,
                     port=port,
-                    daemonize=daemonize,
+                    daemonize=False,
                     debug=run_in_debug,
                     reload=reload,
                     nothreading=nothreading,
                     keep_running=still_owns_runtime_stack,
                     should_stop_supervisord=still_owns_runtime_stack,
+                    resumed_from_pid=standby_result.get("previous_owner_pid") if standby_result.get("resumed") else None,
                 )
                 if result == "interrupted":
                     break
                 if not still_owns_runtime_stack():
-                    owner = runtime_stack_owner(data_dir=config.DATA_DIR)
-                    owner_pid = owner.pid if owner else "unknown"
-                    rprint(
-                        "[yellow][*] A newer archivebox process took over the runner "
-                        f"(pid={owner_pid}). Work will continue there, and will continue here if the other process is stopped and work still remains.[/yellow]",
-                        file=sys.stderr,
-                    )
                     continue
                 if result == "exited":
                     print("[yellow][*] Runtime stack exited while this parent is still leader; restarting...[/yellow]")

archivebox/cli/archivebox_update.py

@@ -533,11 +533,9 @@ def drain_old_archive_dirs(resume_from: str | None = None, batch_size: int = 500
             try:
                 snapshot.status = Snapshot.StatusChoices.SEALED
                 snapshot.retry_at = timezone.now()
-                Snapshot.objects.bulk_create([snapshot])
-                Snapshot.objects.filter(pk=snapshot.pk).update(
-                    status=Snapshot.StatusChoices.SEALED,
-                    retry_at=snapshot.retry_at,
-                )
+                # Snapshot.save() owns URL validation and filesystem/index side
+                # effects. Do not use bulk_create() here; it bypasses save().
+                snapshot.save()
 
                 crawl = _get_snapshot_crawl(snapshot)
                 if crawl is not None:

archivebox/core/migrations/0045_archiveresult_unique_hook.py

@@ -0,0 +1,17 @@
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("core", "0044_alter_archiveresult_status_alter_snapshot_status"),
+    ]
+
+    operations = [
+        migrations.AddConstraint(
+            model_name="archiveresult",
+            constraint=models.UniqueConstraint(
+                fields=("snapshot", "plugin", "hook_name"),
+                name="unique_archiveresult_per_snapshot_hook",
+            ),
+        ),
+    ]

archivebox/core/models.py

@@ -1342,8 +1342,8 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW
                     self.tags.add(tag)
 
     def _merge_archive_results_from_index(self, index_data: dict, update_existing: bool = True):
-        """Merge ArchiveResults - keep both (by plugin+start_ts)."""
-        existing = {(ar.plugin, ar.start_ts): ar for ar in ArchiveResult.objects.filter(snapshot=self)}
+        """Merge ArchiveResults one row per hook; retries update the existing row."""
+        existing = {(ar.plugin, ar.hook_name): ar for ar in ArchiveResult.objects.filter(snapshot=self)}
         if update_existing:
             for archiveresult in existing.values():
                 normalized_status = ArchiveResult.normalize_status(archiveresult.status)
@@ -1404,7 +1404,8 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW
         output_json = result_data.get("output_json")
         output_mimetypes = result_data.get("output_mimetypes", "")
 
-        existing_result = existing.get((plugin, start_ts))
+        hook_name = result_data.get("hook_name", "")
+        existing_result = existing.get((plugin, hook_name))
         if existing_result:
             if not update_existing:
                 return
@@ -1413,22 +1414,25 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW
             if existing_result.status != status:
                 existing_result.status = status
                 update_fields.append("status")
-            if output_str and not existing_result.output_str:
+            if output_str and existing_result.output_str != output_str:
                 existing_result.output_str = output_str
                 update_fields.append("output_str")
-            if output_json and not existing_result.output_json:
+            if output_json and existing_result.output_json != output_json:
                 existing_result.output_json = output_json
                 update_fields.append("output_json")
-            if output_files and not existing_result.output_files:
+            if output_files and existing_result.output_files != output_files:
                 existing_result.output_files = output_files
                 update_fields.append("output_files")
-            if output_size and not existing_result.output_size:
+            if "output_size" in result_data and existing_result.output_size != output_size:
                 existing_result.output_size = output_size
                 update_fields.append("output_size")
-            if output_mimetypes and not existing_result.output_mimetypes:
+            if output_mimetypes and existing_result.output_mimetypes != output_mimetypes:
                 existing_result.output_mimetypes = output_mimetypes
                 update_fields.append("output_mimetypes")
-            if end_ts and not existing_result.end_ts:
+            if start_ts and existing_result.start_ts != start_ts:
+                existing_result.start_ts = start_ts
+                update_fields.append("start_ts")
+            if end_ts and existing_result.end_ts != end_ts:
                 existing_result.end_ts = end_ts
                 update_fields.append("end_ts")
             if update_fields:
@@ -1455,7 +1459,7 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW
             archiveresult = ArchiveResult.objects.create(
                 snapshot=self,
                 plugin=plugin,
-                hook_name=result_data.get("hook_name", ""),
+                hook_name=hook_name,
                 status=status,
                 output_str=output_str,
                 output_json=output_json,
@@ -1466,7 +1470,7 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW
                 end_ts=end_ts,
                 process=process,
             )
-        existing[(plugin, start_ts)] = archiveresult
+        existing[(plugin, hook_name)] = archiveresult
 
     def write_index_json(self):
         """Write index.json in 0.9.x format (deprecated, use write_index_jsonl)."""
@@ -2447,15 +2451,13 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW
             hook_name = hook_path.name  # e.g., 'on_Snapshot__50_wget.py'
             plugin = hook_path.parent.name  # e.g., 'wget'
 
-            # Check if AR already exists for this specific hook
-            if ArchiveResult.objects.filter(snapshot=self, hook_name=hook_name).exists():
-                continue
-
-            archiveresult, created = ArchiveResult.objects.get_or_create(
+            # ArchiveResult output is one filesystem directory per plugin hook, so
+            # retries must update this row in place instead of creating siblings.
+            archiveresult, _created = ArchiveResult.objects.get_or_create(
                 snapshot=self,
+                plugin=plugin,
                 hook_name=hook_name,
                 defaults={
-                    "plugin": plugin,
                     "status": ArchiveResult.INITIAL_STATE,
                 },
             )
@@ -3277,6 +3279,9 @@ class ArchiveResult(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, M
             models.Index(fields=["snapshot", "status"], name="archiveresult_snap_status_idx"),
             models.Index(fields=["-start_ts", "-id"], name="archiveresult_start_idx"),
         ]
+        constraints = [
+            models.UniqueConstraint(fields=["snapshot", "plugin", "hook_name"], name="unique_archiveresult_per_snapshot_hook"),
+        ]
 
     def __str__(self):
         return f"[{self.id}] {self.snapshot.url[:64]} -> {self.plugin}"
@@ -3381,15 +3386,16 @@ class ArchiveResult(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, M
             except ArchiveResult.DoesNotExist:
                 pass
 
-        # Get or create by snapshot_id + plugin
+        # Get or create by snapshot_id + plugin + hook_name. The filesystem has a
+        # single output dir for each hook, so retries update that same DB row.
         try:
             snapshot = Snapshot.objects.get(id=snapshot_id)
 
             result, _ = ArchiveResult.objects.get_or_create(
                 snapshot=snapshot,
                 plugin=plugin,
+                hook_name=record.get("hook_name", ""),
                 defaults={
-                    "hook_name": record.get("hook_name", ""),
                     "status": record.get("status", "queued"),
                     "output_str": record.get("output_str", ""),
                 },

archivebox/core/shutdown_util.py

@@ -98,6 +98,8 @@ def kill_remaining_processes(processes: list[psutil.Process], *, timeout: float
 @contextmanager
 def foreground_shutdown_signals(
     handled_signals: tuple[signal.Signals, ...] = (signal.SIGHUP, signal.SIGINT, signal.SIGTERM),
+    *,
+    first_signal_message: str | None = "\n[🛑] Got {signal_name}, stopping gracefully...\n",
 ) -> Iterator[ShutdownSignalState]:
     """Install foreground signal handlers that print an immediate exit notice.
 
@@ -115,7 +117,8 @@ def foreground_shutdown_signals(
     def raise_keyboard_interrupt(signum, _frame):
         if state.signal_name is None:
             state.signal_name = signal.Signals(signum).name
-            os.write(sys.stdout.fileno(), f"\n[🛑] Got {state.signal_name}, stopping gracefully...\n".encode())
+            if first_signal_message is not None:
+                os.write(sys.stdout.fileno(), first_signal_message.format(signal_name=state.signal_name).encode())
         raise KeyboardInterrupt
 
     try:

archivebox/core/urls.py

@@ -24,7 +24,6 @@ from archivebox.core.views import (
     AddView,
     WebAddView,
     HealthCheckView,
-    live_progress_screencast_frame_view,
     live_progress_view,
 )
 
@@ -70,11 +69,6 @@ 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")),
-    re_path(
-        r"^admin/live-progress/screencast/(?P<object_id>[0-9a-fA-F-]{8,36})\.jpg$",
-        archivebox_admin.admin_view(live_progress_screencast_frame_view),
-        name="live_progress_screencast_frame",
-    ),
     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"),

archivebox/core/views.py

@@ -12,7 +12,7 @@ from pathlib import Path
 from urllib.parse import quote, urlparse
 
 from django.shortcuts import render, redirect
-from django.http import FileResponse, JsonResponse, HttpRequest, HttpResponse, Http404, HttpResponseForbidden, QueryDict
+from django.http import JsonResponse, HttpRequest, HttpResponse, Http404, HttpResponseForbidden, QueryDict
 from django.utils.html import format_html
 from django.utils.safestring import mark_safe
 from django.views import View
@@ -22,7 +22,6 @@ from django.db.models import CharField, Count, Q, Prefetch, Sum
 from django.db.models.functions import Cast
 from django.contrib import messages
 from django.contrib.auth.mixins import UserPassesTestMixin
-from django.core.signing import BadSignature, SignatureExpired, TimestampSigner
 from django.views.decorators.csrf import csrf_exempt
 from django.views.decorators.gzip import gzip_page
 from django.utils.decorators import method_decorator
@@ -79,7 +78,6 @@ from archivebox.hooks import (
 
 ABX_PLUGINS_GITHUB_BASE_URL = "https://github.com/ArchiveBox/abx-plugins/tree/main/abx_plugins/plugins/"
 LIVE_PLUGIN_BASE_URL = "/admin/environment/plugins/"
-SCREENCAST_SIGNER = TimestampSigner(salt="archivebox.live-progress.screencast")
 
 
 def _get_request_config(request: HttpRequest, *, resolve_plugins: bool = False):
@@ -1257,9 +1255,12 @@ class AddView(UserPassesTestMixin, FormView):
             crawl.save(update_fields=["schedule"])
 
         crawl.create_snapshots_from_urls()
-        from archivebox.services.runner import ensure_background_runner
+        if crawl.snapshot_set.exists():
+            from archivebox.services.runner import ensure_background_runner
 
-        ensure_background_runner()
+            ensure_background_runner()
+        else:
+            crawl.sm.seal()
 
         return crawl
 
@@ -1386,33 +1387,6 @@ class HealthCheckView(View):
         return HttpResponse("OK", content_type="text/plain", status=200)
 
 
-def live_progress_screencast_frame_view(request, object_id: str):
-    """Serve cache-only Chrome screencast frames through the admin app."""
-    if not is_admin_user(request):
-        return HttpResponseForbidden("Permission denied")
-
-    token = request.GET.get("token", "")
-    try:
-        if SCREENCAST_SIGNER.unsign(token, max_age=60) != str(object_id):
-            return HttpResponseForbidden("Permission denied")
-    except (BadSignature, SignatureExpired):
-        return HttpResponseForbidden("Permission denied")
-
-    live_root = (CONSTANTS.CACHE_DIR / "chrome_screencast").resolve()
-    frame_path = live_root / str(object_id) / "latest.jpg"
-    try:
-        resolved_frame_path = frame_path.resolve(strict=True)
-    except FileNotFoundError:
-        raise Http404 from None
-    if not resolved_frame_path.is_file() or live_root not in resolved_frame_path.parents:
-        raise Http404
-
-    response = FileResponse(resolved_frame_path.open("rb"), content_type="image/jpeg")
-    response["Cache-Control"] = "no-store, max-age=0"
-    response["X-Content-Type-Options"] = "nosniff"
-    return response
-
-
 @gzip_page
 def live_progress_view(request):
     """Simple JSON endpoint for live progress status - used by admin progress monitor."""
@@ -1560,16 +1534,17 @@ def live_progress_view(request):
             url = str(url or "")
             return url if len(url) <= 96 else f"{url[:93]}..."
 
-        def screencast_frame_url(object_id: str) -> str:
-            frame_path = CONSTANTS.CACHE_DIR / "chrome_screencast" / object_id / "latest.jpg"
+        def screencast_frame_url(crawl_id: str, crawl_dir: Path) -> str:
+            frame_path = crawl_dir / "chrome_screencast" / "latest.jpg"
             try:
                 frame_stat = frame_path.stat()
             except OSError:
                 return ""
             if frame_stat.st_size <= 0:
                 return ""
-            token = SCREENCAST_SIGNER.sign(object_id)
-            return f"/admin/live-progress/screencast/{object_id}.jpg?v={frame_stat.st_mtime_ns}&token={quote(token)}"
+            if now.timestamp() - frame_stat.st_mtime > 15:
+                return ""
+            return f"/api/v1/crawls/crawl/{crawl_id}/files/chrome_screencast/latest.jpg"
 
         machine_id = Machine.current().id
         orchestrator_proc = (
@@ -1982,7 +1957,7 @@ def live_progress_view(request):
             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")
             crawl_setup_pending = sum(1 for item in crawl_setup_plugins if item.get("status") == "queued")
-            crawl_screencast_url = screencast_frame_url(crawl_id)
+            crawl_screencast_url = screencast_frame_url(crawl_id, active_crawl_objects[crawl_id].output_dir)
             crawl_screencast_link = f"/admin/crawls/crawl/{crawl_id}/change/" if crawl_screencast_url else ""
 
             # Get active snapshots for this crawl (already prefetched)
@@ -2034,7 +2009,7 @@ def live_progress_view(request):
                     snapshot_preview_url = snapshot_favicon_url
 
                 if snapshot["status"] == Snapshot.StatusChoices.STARTED:
-                    snapshot_screencast_url = screencast_frame_url(str(snapshot["id"]))
+                    snapshot_screencast_url = screencast_frame_url(crawl_id, active_crawl_objects[crawl_id].output_dir)
                     snapshot_screencast_link = snapshot_view_url(snapshot) if snapshot_screencast_url else ""
 
                 def plugin_sort_key(ar):

archivebox/crawls/models.py

@@ -887,7 +887,7 @@ class Crawl(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelWith
                         snapshot.save(update_fields=["depth", "title", "timestamp", "status", "retry_at", "url", "crawl", "modified_at"])
                     else:
                         snapshot = Snapshot(id=snapshot_id, url=url, crawl=self, **defaults)
-                        snapshot.save(force_insert=True)
+                        snapshot.save()
                         created = True
                 else:
                     snapshot = Snapshot.objects.filter(url=url, crawl=self).first()
@@ -896,7 +896,7 @@ class Crawl(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelWith
                     else:
                         try:
                             snapshot = Snapshot(url=url, crawl=self, **defaults)
-                            snapshot.save(force_insert=True)
+                            snapshot.save()
                             created = True
                         except IntegrityError:
                             snapshot = Snapshot.objects.get(url=url, crawl=self)
@@ -1052,10 +1052,19 @@ class Crawl(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelWith
         for snapshot in snapshots:
             snapshot.set_delete_at_from_config(config.DELETE_AFTER)
 
-        try:
-            created_snapshots = list(Snapshot.objects.bulk_create(snapshots))
-        except ValidationError as err:
-            print(f"[yellow][!] Skipping blocked discovered snapshots: {err}[/yellow]")
+        created_snapshots = []
+        for snapshot in snapshots:
+            try:
+                # Snapshot.save() owns URL validation and filesystem/index side
+                # effects. Do not use bulk_create() here; it bypasses save().
+                snapshot.save()
+            except IntegrityError:
+                continue
+            except ValidationError as err:
+                print(f"[yellow][!] Skipping blocked discovered snapshot URL: {snapshot.url} ({err})[/yellow]")
+                continue
+            created_snapshots.append(snapshot)
+        if not created_snapshots:
             return []
 
         crawl_urls = {url for _raw_line, url in self._iter_url_lines() if url}
@@ -1072,8 +1081,9 @@ class Crawl(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelWith
             }
             if tag_names:
                 tag_names_by_url[snapshot.url] = tag_names
-            # Same transaction rule as create_snapshots_from_urls(): bulk_create()
-            # only writes DB rows; symlink creation waits until after commit.
+            # Snapshot.save() handles model-level validation. The crawl symlink
+            # can still wait until after commit so SQLite does not hold a write
+            # lock while touching the filesystem.
             transaction.on_commit(lambda snapshot=snapshot: snapshot.ensure_crawl_symlink())
 
         tag_names = {tag for tags in tag_names_by_url.values() for tag in tags}
@@ -1469,8 +1479,10 @@ class CrawlMachine(BaseStateMachine):
         | paused.to.itself()
     )
 
-    # Manual event (triggered by last Snapshot sealing)
-    seal = started.to(sealed)
+    # Manual event (triggered by last Snapshot sealing, or by direct
+    # index-only/bg creation when every requested URL is rejected before any
+    # Snapshot rows exist).
+    seal = queued.to(sealed) | started.to(sealed)
     pause_requested = queued.to(paused) | started.to(paused)
     resume_requested = paused.to(queued)
 

archivebox/services/archive_result_service.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import asyncio
 import json
 from collections import defaultdict
 from collections.abc import Iterable
@@ -209,6 +210,8 @@ class ArchiveResultService(BaseService):
     EMITS = []
 
     def __init__(self, bus):
+        self._completed_process_event_ids: set[str] = set()
+        self._save_locks: dict[tuple[str, str, str], asyncio.Lock] = {}
         super().__init__(bus)
         self.bus.on(ArchiveResultEvent, self.on_ArchiveResultEvent__save_to_db)
         self.bus.on(ProcessCompletedEvent, self.on_ProcessCompletedEvent__save_to_db)
@@ -259,12 +262,15 @@ class ArchiveResultService(BaseService):
         if event.error:
             defaults["notes"] = event.error
 
-        result, _created = await ArchiveResult.objects.aupdate_or_create(
-            snapshot=snapshot,
-            plugin=event.plugin,
-            hook_name=event.hook_name,
-            defaults=defaults,
-        )
+        key = (str(snapshot.id), event.plugin, event.hook_name)
+        lock = self._save_locks.setdefault(key, asyncio.Lock())
+        async with lock:
+            result, _created = await ArchiveResult.objects.aupdate_or_create(
+                snapshot=snapshot,
+                plugin=event.plugin,
+                hook_name=event.hook_name,
+                defaults=defaults,
+            )
 
         if result.status in (ArchiveResult.StatusChoices.SUCCEEDED, ArchiveResult.StatusChoices.NORESULTS):
             next_title = _extract_snapshot_title(str(snapshot.output_dir), event.plugin, result.output_str, snapshot_url=snapshot.url)
@@ -273,6 +279,10 @@ class ArchiveResultService(BaseService):
                 await snapshot.asave(update_fields=["title", "modified_at"])
 
     async def on_ProcessCompletedEvent__save_to_db(self, event: ProcessCompletedEvent) -> None:
+        if event.event_id in self._completed_process_event_ids:
+            return
+        self._completed_process_event_ids.add(event.event_id)
+
         if not event.hook_name.startswith("on_Snapshot"):
             return
         snapshot_event = await self.bus.find(
@@ -286,6 +296,10 @@ class ArchiveResultService(BaseService):
 
         records = _iter_archiveresult_records(event.stdout)
         if records:
+            if len(records) > 1:
+                raise RuntimeError(
+                    f"Hook {event.plugin_name}:{event.hook_name} emitted {len(records)} ArchiveResult records; expected exactly one",
+                )
             for record in records:
                 record_status = _normalize_status(record.get("status") or "")
                 record_failed = record_status == "failed" or (not record_status and event.exit_code not in (0, PROCESS_EXIT_SKIPPED))

archivebox/services/runner.py

@@ -20,6 +20,7 @@ from typing import Any
 from asgiref.sync import sync_to_async
 from django.utils import timezone
 from rich.console import Console
+from rich.text import Text
 
 from abx_dl.events import (
     BinaryRequestEvent,
@@ -75,6 +76,43 @@ def _bus_name(prefix: str, identifier: str) -> str:
     return f"{prefix}_{normalized}"
 
 
+def _runner_short_id(identifier) -> str:
+    return str(identifier).replace("-", "")[-8:]
+
+
+def _runner_label(value: str, *, reserve: int) -> str:
+    width = max(24, shutil.get_terminal_size(fallback=(120, 40)).columns - reserve)
+    value = " ".join(str(value or "").split())
+    if len(value) <= width:
+        return value
+    return f"{value[: max(0, width - 3)]}..."
+
+
+def _runner_console_line(*, crawl, snapshot=None, status: str = "STARTED") -> None:
+    line = Text()
+    line.append(f"[Crawl#{_runner_short_id(crawl.id)}]", style="cyan bold")
+    line.append(" ")
+    if snapshot is not None:
+        line.append(f"[Snapshot#{_runner_short_id(snapshot.id)}]", style="magenta bold")
+        line.append(" ")
+    status_styles = {
+        "STARTED": "green bold",
+        "SEALED": "blue bold",
+        "PAUSED": "yellow bold",
+    }
+    line.append(f"[{status}]", style=status_styles.get(status, "white bold"))
+    line.append(" ")
+    prefix_width = len(line.plain)
+    if snapshot is not None:
+        label = snapshot.url
+    else:
+        label = (getattr(crawl, "label", "") or "").strip()
+        if not label:
+            label = (getattr(crawl, "urls", "") or "").partition("\n")[0].strip() or str(crawl.id)
+    line.append(_runner_label(label, reserve=prefix_width))
+    Console(highlight=False).print(line)
+
+
 def _count_selected_hooks(plugins: dict[str, Plugin], selected_plugins: list[str] | None) -> int:
     selected = filter_plugins(plugins, selected_plugins) if selected_plugins else plugins
     return sum(1 for plugin in selected.values() for hook in plugin.hooks if "CrawlSetup" in hook.name or "Snapshot" in hook.name)
@@ -137,6 +175,12 @@ async def _run_event_now(event, timeout: float | None = None):
     return event
 
 
+async def _emit_event_now(bus, event, timeout: float | None = None, parent_event=None):
+    if parent_event is not None:
+        event.event_parent_id = parent_event.event_id
+    return await _run_event_now(bus.emit(event), timeout)
+
+
 def ensure_background_runner(*, allow_under_pytest: bool = False) -> bool:
     if os.environ.get("PYTEST_CURRENT_TEST") and not allow_under_pytest:
         return False
@@ -186,11 +230,12 @@ class CrawlRunner:
         selected_plugins: list[str] | None = None,
         process_discovered_snapshots_inline: bool = True,
         show_progress: bool = True,
+        interactive_interrupts: bool = False,
     ):
         self.crawl = crawl
         self.bus = create_bus(name=_bus_name("ArchiveBox", str(crawl.id)), total_timeout=3600.0)
         self.plugins = discover_plugins()
-        HookProcessService(self.bus, emit_jsonl=False, interactive_tty=False)
+        HookProcessService(self.bus, emit_jsonl=False, interactive_tty=interactive_interrupts)
         register_sonic_daemon_event_handler(self.bus)
         PersistedProcessService(self.bus)
         BinaryService(self.bus)
@@ -199,6 +244,7 @@ class CrawlRunner:
         MachineService(self.bus)
         self.process_discovered_snapshots_inline = process_discovered_snapshots_inline
         self.show_progress = show_progress
+        self.interactive_interrupts = interactive_interrupts
 
         async def ignore_snapshot(_snapshot_id: str) -> None:
             return None
@@ -236,8 +282,17 @@ class CrawlRunner:
             previous = signal.getsignal(sig)
 
             def request_abort(_signum, _frame, sig=sig) -> None:
-                os.write(sys.stdout.fileno(), f"\n[🛑] Got {sig.name}, stopping gracefully...\n".encode())
+                already_requested = self._signal_abort_requested
+                if not already_requested:
+                    message = (
+                        f"\n[🛑] Got {sig.name}, aborting the active hook...\n"
+                        if self.interactive_interrupts
+                        else f"\n[🛑] Got {sig.name}, stopping gracefully...\n"
+                    )
+                    os.write(sys.stdout.fileno(), message.encode())
                 self._request_abort_from_signal(sig)
+                if not already_requested:
+                    return
                 raise KeyboardInterrupt
 
             signal.signal(sig, request_abort)
@@ -251,9 +306,10 @@ class CrawlRunner:
             signal.signal(sig, previous)
 
     def _request_abort_from_signal(self, _sig: signal.Signals) -> None:
+        already_requested = self._signal_abort_requested
         self._signal_abort_requested = True
         self._skip_wait_until_idle = True
-        if self._run_task is not None and not self._run_task.done():
+        if (not self.interactive_interrupts or already_requested) and self._run_task is not None and not self._run_task.done():
             self._run_task.cancel()
 
     async def crawl_is_cancelled(self) -> bool:
@@ -846,30 +902,40 @@ class CrawlRunner:
                 finally:
                     if self.snapshot_tasks:
                         await self.drain_snapshot_tasks()
-                    await _run_event_now(
-                        event.emit(
-                            CrawlCleanupEvent(
-                                url=snapshot["url"],
-                                snapshot_id=snapshot["id"],
-                                output_dir=str(output_dir),
-                                event_timeout=crawl_setup_phase_timeout,
-                                event_handler_slow_timeout=slow_warning_timeout(crawl_setup_phase_timeout),
+                    await asyncio.shield(
+                        asyncio.create_task(
+                            _emit_event_now(
+                                self.bus,
+                                CrawlCleanupEvent(
+                                    url=snapshot["url"],
+                                    snapshot_id=snapshot["id"],
+                                    output_dir=str(output_dir),
+                                    event_timeout=crawl_setup_phase_timeout,
+                                    event_handler_slow_timeout=slow_warning_timeout(crawl_setup_phase_timeout),
+                                ),
+                                crawl_setup_phase_timeout,
+                                parent_event=event,
                             ),
+                            context=contextvars.Context(),
                         ),
-                        crawl_setup_phase_timeout,
                     )
             finally:
                 cancel_watcher.cancel()
                 await asyncio.gather(cancel_watcher, return_exceptions=True)
-            await _run_event_now(
-                event.emit(
-                    CrawlCompletedEvent(
-                        url=snapshot["url"],
-                        snapshot_id=snapshot["id"],
-                        output_dir=str(output_dir),
+            await asyncio.shield(
+                asyncio.create_task(
+                    _emit_event_now(
+                        self.bus,
+                        CrawlCompletedEvent(
+                            url=snapshot["url"],
+                            snapshot_id=snapshot["id"],
+                            output_dir=str(output_dir),
+                        ),
+                        CrawlCompletedEvent.model_fields["event_timeout"].default,
+                        parent_event=event,
                     ),
+                    context=contextvars.Context(),
                 ),
-                CrawlCompletedEvent.model_fields["event_timeout"].default,
             )
 
         on_archivebox_CrawlStartEvent.__name__ = "on_archivebox_CrawlStartEvent__run_snapshots"
@@ -1046,6 +1112,7 @@ def run_crawl(
     selected_plugins: list[str] | None = None,
     process_discovered_snapshots_inline: bool = True,
     show_progress: bool = True,
+    interactive_interrupts: bool = False,
 ) -> None:
     from archivebox.crawls.models import Crawl
     from django.db import close_old_connections
@@ -1061,6 +1128,7 @@ def run_crawl(
                     selected_plugins=selected_plugins,
                     process_discovered_snapshots_inline=process_discovered_snapshots_inline,
                     show_progress=show_progress,
+                    interactive_interrupts=interactive_interrupts,
                 ).run(),
             )
         finally:
@@ -1197,9 +1265,9 @@ def run_snapshot_maintenance(snapshot_id: str) -> bool:
     return True
 
 
-def run_due_crawl(crawl, *, lock_seconds: int) -> bool:
+def run_due_crawl(crawl, *, lock_seconds: int, interactive_interrupts: bool = False) -> bool:
     if crawl.is_paused:
-        print(f"[runner] Crawl {str(crawl.id)[-12:]} paused; skipping until resumed", flush=True)
+        _runner_console_line(crawl=crawl, status="PAUSED")
         return True
     if crawl.status in (crawl.StatusChoices.QUEUED, crawl.StatusChoices.STARTED):
         from archivebox.core.models import Snapshot
@@ -1255,29 +1323,28 @@ def run_due_crawl(crawl, *, lock_seconds: int) -> bool:
             return True
         if not crawl.claim_processing_lock(lock_seconds=lock_seconds):
             return False
-        print(f"[runner] Crawl {str(crawl.id)[-12:]} running status={crawl.status} snapshots={snapshot_count}", flush=True)
-        run_crawl(str(crawl.id), process_discovered_snapshots_inline=True)
+        _runner_console_line(crawl=crawl)
+        run_crawl(str(crawl.id), process_discovered_snapshots_inline=True, interactive_interrupts=interactive_interrupts)
         return True
 
     if crawl.status == crawl.StatusChoices.SEALED:
-        print(f"[runner] Crawl {str(crawl.id)[-12:]} sealed; clearing retry tick", flush=True)
+        _runner_console_line(crawl=crawl, status="SEALED")
         crawl.retry_at = None
         crawl.save(update_fields=["retry_at", "modified_at"])
         return True
 
-    print(f"[runner] Crawl {str(crawl.id)[-12:]} status={crawl.status}; clearing retry tick", flush=True)
     crawl.retry_at = None
     crawl.save(update_fields=["retry_at", "modified_at"])
     return True
 
 
-def run_due_snapshot(snapshot, *, lock_seconds: int) -> bool:
+def run_due_snapshot(snapshot, *, lock_seconds: int, interactive_interrupts: bool = False) -> bool:
     from archivebox.core.models import Snapshot
 
     if snapshot.is_paused:
         selected_plugins = queued_plugins_for_snapshot(str(snapshot.id))
         if snapshot.fs_migration_needed and Snapshot.claim_for_worker(snapshot, lock_seconds=lock_seconds):
-            print(f"[runner] Snapshot {str(snapshot.id)[-12:]} paused maintenance fs_version={snapshot.fs_version}", flush=True)
+            _runner_console_line(crawl=snapshot.crawl, snapshot=snapshot)
             run_snapshot_maintenance(str(snapshot.id))
             if not selected_plugins:
                 # No targeted plugin rows remain, so put paused snapshots back
@@ -1298,10 +1365,7 @@ def run_due_snapshot(snapshot, *, lock_seconds: int) -> bool:
         if not Snapshot.claim_for_worker(snapshot, lock_seconds=lock_seconds):
             return False
         try:
-            print(
-                f"[runner] Snapshot {str(snapshot.id)[-12:]} paused targeted plugins={','.join(selected_plugins)} url={snapshot.url}",
-                flush=True,
-            )
+            _runner_console_line(crawl=snapshot.crawl, snapshot=snapshot)
             # Explicit maintenance, e.g. `archivebox update --index-only`, may
             # need to run search/index hooks for a paused snapshot. That should
             # not resume the crawl or make unrelated queued work runnable, so
@@ -1312,6 +1376,7 @@ def run_due_snapshot(snapshot, *, lock_seconds: int) -> bool:
                 snapshot_ids=[str(snapshot.id)],
                 selected_plugins=selected_plugins,
                 process_discovered_snapshots_inline=True,
+                interactive_interrupts=interactive_interrupts,
             )
         finally:
             # Targeted plugin rows can complete while the Snapshot remains
@@ -1328,22 +1393,17 @@ def run_due_snapshot(snapshot, *, lock_seconds: int) -> bool:
             # migration. Run the filesystem/json save path before queued search
             # backfill rows so both maintenance streams stay ordered without
             # changing Snapshot.status away from SEALED.
-            print(
-                f"[runner] Snapshot {str(snapshot.id)[-12:]} sealed maintenance fs_version={snapshot.fs_version} url={snapshot.url}",
-                flush=True,
-            )
+            _runner_console_line(crawl=snapshot.crawl, snapshot=snapshot)
             return run_snapshot_maintenance(str(snapshot.id))
         selected_plugins = queued_plugins_for_snapshot(str(snapshot.id))
         if selected_plugins:
-            print(
-                f"[runner] Snapshot {str(snapshot.id)[-12:]} sealed targeted plugins={','.join(selected_plugins)} url={snapshot.url}",
-                flush=True,
-            )
+            _runner_console_line(crawl=snapshot.crawl, snapshot=snapshot)
             run_crawl(
                 str(snapshot.crawl_id),
                 snapshot_ids=[str(snapshot.id)],
                 selected_plugins=selected_plugins,
                 process_discovered_snapshots_inline=True,
+                interactive_interrupts=interactive_interrupts,
             )
             return True
         return run_snapshot_maintenance(str(snapshot.id))
@@ -1351,18 +1411,18 @@ def run_due_snapshot(snapshot, *, lock_seconds: int) -> bool:
     if snapshot.status == Snapshot.StatusChoices.STARTED:
         _reset_count, running_count = reset_abandoned_snapshot_results(snapshot)
         if running_count:
-            print(f"[runner] Snapshot {str(snapshot.id)[-12:]} still has {running_count} running ArchiveResults", flush=True)
             snapshot.update_and_requeue(retry_at=timezone.now() + timedelta(seconds=ACTIVE_STATE_LEASE_SECONDS))
             return True
 
     if not snapshot.claim_processing_lock(lock_seconds=lock_seconds):
         return False
-    print(f"[runner] Snapshot {str(snapshot.id)[-12:]} running status={snapshot.status} url={snapshot.url}", flush=True)
+    _runner_console_line(crawl=snapshot.crawl, snapshot=snapshot)
     run_crawl(
         str(snapshot.crawl_id),
         snapshot_ids=[str(snapshot.id)],
         selected_plugins=queued_plugins_for_snapshot(str(snapshot.id)),
         process_discovered_snapshots_inline=True,
+        interactive_interrupts=interactive_interrupts,
     )
     return True
 
@@ -1488,7 +1548,13 @@ def run_install(*, plugin_names: list[str] | None = None) -> None:
     asyncio.run(_run_install(plugin_names=plugin_names))
 
 
-def run_pending_crawls(*, daemon: bool = False, crawl_id: str | None = None, maintenance_only: bool = False) -> int:
+def run_pending_crawls(
+    *,
+    daemon: bool = False,
+    crawl_id: str | None = None,
+    maintenance_only: bool = False,
+    interactive_interrupts: bool = False,
+) -> int:
     from archivebox.crawls.models import Crawl, CrawlSchedule
     from archivebox.core.models import ArchiveResult, Snapshot
     from archivebox.machine.models import Binary, Process
@@ -1515,7 +1581,11 @@ def run_pending_crawls(*, daemon: bool = False, crawl_id: str | None = None, mai
                 due_crawls = due_crawls.filter(id=crawl_id)
             due_crawl = due_crawls.order_by("retry_at", "created_at").first()
             if due_crawl is not None:
-                if not run_due_crawl(due_crawl, lock_seconds=crawl_claim_lock_seconds):
+                if not run_due_crawl(
+                    due_crawl,
+                    lock_seconds=crawl_claim_lock_seconds,
+                    interactive_interrupts=interactive_interrupts,
+                ):
                     continue
                 continue
 
@@ -1526,7 +1596,7 @@ def run_pending_crawls(*, daemon: bool = False, crawl_id: str | None = None, mai
             due_snapshots = due_snapshots.filter(crawl_id=crawl_id)
         due_snapshot = due_snapshots.order_by("retry_at", "created_at").first()
         if due_snapshot is not None:
-            if not run_due_snapshot(due_snapshot, lock_seconds=60):
+            if not run_due_snapshot(due_snapshot, lock_seconds=60, interactive_interrupts=interactive_interrupts):
                 continue
             continue
 

archivebox/services/snapshot_service.py

@@ -1,7 +1,11 @@
 from __future__ import annotations
 
+import sys
+
 from asgiref.sync import sync_to_async
+from django.core.exceptions import ValidationError
 from django.utils import timezone
+from rich import print as rprint
 from abx_dl.events import SnapshotCompletedEvent, SnapshotEvent
 from abx_dl.limits import CrawlLimitState
 from abx_dl.services.base import BaseService
@@ -27,7 +31,21 @@ class SnapshotService(BaseService):
             if snapshot.is_paused:
                 return
             if snapshot.status == Snapshot.StatusChoices.QUEUED:
-                await sync_to_async(snapshot.sm.tick, thread_sensitive=True)()
+                try:
+                    await sync_to_async(snapshot.sm.tick, thread_sensitive=True)()
+                except ValidationError as err:
+                    if "ArchiveBox cannot archive its own admin, web, api, or snapshot URLs." not in str(err):
+                        raise
+                    await Snapshot.objects.filter(id=snapshot.id).aupdate(
+                        status=Snapshot.StatusChoices.SEALED,
+                        retry_at=None,
+                        modified_at=timezone.now(),
+                    )
+                    rprint(
+                        f"[red][X] Refusing to archive ArchiveBox internal URL for security: {snapshot.url}[/red]",
+                        file=sys.stderr,
+                    )
+                    return
                 await sync_to_async(snapshot.refresh_from_db, thread_sensitive=True)()
             elif snapshot.status != Snapshot.StatusChoices.STARTED:
                 return

archivebox/services/supervision_service.py

@@ -51,7 +51,7 @@ def command_is_newest(command, *, process_type: str, data_dir: str | Path, url:
     return bool(leader and leader.id == command.id)
 
 
-def runtime_stack_owner(*, data_dir: str | Path):
+def runtime_stack_owner(*, data_dir: str | Path, exclude_id=None):
     from archivebox.machine.models import Machine, Process
 
     base_qs = Process.objects.filter(
@@ -60,6 +60,8 @@ def runtime_stack_owner(*, data_dir: str | Path):
         pwd=str(data_dir),
         process_type__in=runtime_stack_owner_types(),
     )
+    if exclude_id is not None:
+        base_qs = base_qs.exclude(id=exclude_id)
     for process_types in (
         (Process.TypeChoices.UPDATE,),
         (Process.TypeChoices.SERVER, Process.TypeChoices.ADD),
@@ -141,28 +143,31 @@ def standby_until_leader_needed(command, *, process_type: str, data_dir: str | P
         if not announced:
             leader = newest_live_process(process_type=process_type, data_dir=data_dir, url=url)
             leader_pid = leader.pid if leader else "unknown"
-            rprint(f"[yellow][*] Standing by; newer ArchiveBox parent pid={leader_pid} is running the orchestrator and server.[/yellow]")
+            rprint(f"[yellow][*] Standing by; newer ArchiveBox process pid={leader_pid} is running the orchestrator and server.[/yellow]")
             announced = True
         time.sleep(interval)
     command.modified_at = timezone.now()
     command.save(update_fields=["modified_at"])
 
 
-def standby_until_runtime_stack_needed(command, *, data_dir: str | Path, interval: float = 2.0) -> None:
+def standby_until_runtime_stack_needed(command, *, data_dir: str | Path, interval: float = 2.0) -> dict[str, object]:
     from archivebox.workers.supervisord_util import reap_foreground_supervisord_process
 
     announced = False
+    previous_owner_pid = None
     while not command_owns_runtime_stack(command, data_dir=data_dir):
         reap_foreground_supervisord_process()
         if not announced:
             owner = runtime_stack_owner(data_dir=data_dir)
             owner_pid = owner.pid if owner else "unknown"
+            previous_owner_pid = owner_pid
             rprint(
-                "[yellow][*] A newer archivebox process took over the runner "
-                f"(pid={owner_pid}). Work will continue there, and will continue here if the other process is stopped and work still remains.[/yellow]",
+                "[yellow][*] A newer archivebox process took over the runtime stack "
+                f"(pid={owner_pid}). Work will continue there, and will resume here if that process exits and work still remains.[/yellow]",
                 file=sys.stderr,
             )
             announced = True
         time.sleep(interval)
     command.modified_at = timezone.now()
     command.save(update_fields=["modified_at"])
+    return {"resumed": announced, "previous_owner_pid": previous_owner_pid}

archivebox/tests/test_archive_result_service.py

@@ -70,6 +70,81 @@ def test_process_completed_projects_inline_archiveresult():
     _cleanup_machine_process_rows()
 
 
+def test_archiveresult_event_retry_updates_existing_hook_row():
+    from archivebox.core.models import ArchiveResult
+    from archivebox.services.archive_result_service import ArchiveResultService
+    import asyncio
+
+    snapshot = _create_snapshot()
+    plugin_dir = Path(snapshot.output_dir) / "wget"
+    plugin_dir.mkdir(parents=True, exist_ok=True)
+    (plugin_dir / "index.html").write_text("<html>ok</html>")
+
+    service = ArchiveResultService(create_bus(name="test_archiveresult_retry_updates_existing_hook_row"))
+    first_event = ArchiveResultEvent(
+        snapshot_id=str(snapshot.id),
+        plugin="wget",
+        hook_name="on_Snapshot__06_wget.finite.bg",
+        status="failed",
+        output_str="timed out",
+        start_ts="2026-03-22T12:00:00+00:00",
+        end_ts="2026-03-22T12:00:01+00:00",
+    )
+    retry_event = ArchiveResultEvent(
+        snapshot_id=str(snapshot.id),
+        plugin="wget",
+        hook_name="on_Snapshot__06_wget.finite.bg",
+        status="succeeded",
+        output_str="wget/index.html",
+        output_files=[OutputFile(path="index.html", extension="html", mimetype="text/html", size=15)],
+        start_ts="2026-03-22T12:01:00+00:00",
+        end_ts="2026-03-22T12:01:01+00:00",
+    )
+
+    async def emit_events() -> None:
+        await service.on_ArchiveResultEvent__save_to_db(first_event)
+        first_result_id = await ArchiveResult.objects.values_list("id", flat=True).aget(
+            snapshot=snapshot,
+            plugin="wget",
+            hook_name="on_Snapshot__06_wget.finite.bg",
+        )
+        await service.on_ArchiveResultEvent__save_to_db(retry_event)
+        retry_result = await ArchiveResult.objects.aget(
+            snapshot=snapshot,
+            plugin="wget",
+            hook_name="on_Snapshot__06_wget.finite.bg",
+        )
+        assert retry_result.id == first_result_id
+        assert retry_result.status == ArchiveResult.StatusChoices.SUCCEEDED
+        assert retry_result.output_str == "wget/index.html"
+
+    asyncio.run(emit_events())
+
+    assert ArchiveResult.objects.filter(snapshot=snapshot, plugin="wget", hook_name="on_Snapshot__06_wget.finite.bg").count() == 1
+    _cleanup_machine_process_rows()
+
+
+def test_archiveresult_duplicate_hook_rows_are_rejected():
+    from django.db import IntegrityError, transaction
+    from archivebox.core.models import ArchiveResult
+
+    snapshot = _create_snapshot()
+    ArchiveResult.objects.create(
+        snapshot=snapshot,
+        plugin="wget",
+        hook_name="on_Snapshot__06_wget.finite.bg",
+        status=ArchiveResult.StatusChoices.FAILED,
+    )
+
+    with pytest.raises(IntegrityError), transaction.atomic():
+        ArchiveResult.objects.create(
+            snapshot=snapshot,
+            plugin="wget",
+            hook_name="on_Snapshot__06_wget.finite.bg",
+            status=ArchiveResult.StatusChoices.SUCCEEDED,
+        )
+
+
 def test_process_completed_projects_synthetic_failed_archiveresult():
     from archivebox.core.models import ArchiveResult
     from archivebox.services.archive_result_service import ArchiveResultService

archivebox/tests/test_cli_real_flows.py

@@ -11,6 +11,7 @@ import sys
 import time
 from pathlib import Path
 
+import psutil
 import pytest
 
 from archivebox.core.models import ArchiveResult, Snapshot
@@ -145,6 +146,24 @@ def _wait_for_pid_to_disappear(pid: int, *, timeout: float = 20.0) -> None:
     raise AssertionError(f"PID {pid} is still running")
 
 
+def _wait_for_process(predicate, *, timeout: float = 20.0):
+    deadline = time.time() + timeout
+    last_seen = []
+    while time.time() < deadline:
+        last_seen = []
+        for proc in psutil.process_iter(["pid", "ppid", "cmdline"]):
+            try:
+                cmdline = proc.info.get("cmdline") or []
+                command = " ".join(cmdline)
+                last_seen.append(f"{proc.info.get('pid')} {proc.info.get('ppid')} {command}")
+                if predicate(proc, command):
+                    return proc
+            except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess):
+                continue
+        time.sleep(0.2)
+    raise AssertionError("No matching live process found. Last seen:\n" + "\n".join(last_seen[-50:]))
+
+
 def _supervisor_pid_from_log(log_path: Path) -> int:
     content = log_path.read_text(encoding="utf-8", errors="replace")
     matches = re.findall(r"Supervisord connected \(pid=(\d+)\)", content)
@@ -161,7 +180,37 @@ def _worker_pid_from_log(log_path: Path, worker_name: str) -> int:
 
 def _pgrep_data_dir(data_dir) -> list[str]:
     result = subprocess.run(["pgrep", "-af", str(data_dir)], capture_output=True, text=True, timeout=5)
-    return [line for line in result.stdout.splitlines() if "pgrep -af" not in line]
+    lines = [line for line in result.stdout.splitlines() if "pgrep -af" not in line]
+
+    # A foreground ArchiveBox process can be killed with SIGKILL before Python
+    # cleanup runs. Supervisord's command line only points at its generated
+    # config file, so catch orphaned supervisors by resolving pidfiles whose
+    # configs still reference this real test DATA_DIR.
+    for runtime_root in (Path("/tmp/archivebox"), Path(data_dir) / "tmp"):
+        for config_path in runtime_root.glob("*/supervisord.conf"):
+            try:
+                config_text = config_path.read_text(encoding="utf-8", errors="replace")
+            except OSError:
+                continue
+            if str(data_dir) not in config_text:
+                continue
+            pid_path = config_path.with_name("supervisord.pid")
+            try:
+                pid = int(pid_path.read_text(encoding="utf-8").strip())
+            except (OSError, ValueError):
+                continue
+            if not _pid_is_alive(pid):
+                continue
+            ps_line = subprocess.run(
+                ["ps", "-p", str(pid), "-o", "pid=,ppid=,command="],
+                capture_output=True,
+                text=True,
+                timeout=5,
+            ).stdout.strip()
+            if ps_line:
+                lines.append(ps_line)
+
+    return sorted(set(lines))
 
 
 def _assert_no_processes_for_data_dir(data_dir, *, timeout: float = 10.0) -> None:
@@ -499,8 +548,47 @@ def test_live_server_signal_exit_and_resume_uses_existing_supervisor_state(tmp_p
         _kill_processes_for_data_dir(tmp_path)
 
 
+@pytest.mark.timeout(180)
+def test_live_daemonized_server_keeps_supervisord_owned_by_archivebox_parent(tmp_path, process):
+    os.chdir(tmp_path)
+    assert process.returncode == 0, process.stderr
+
+    env = _live_exit_env(tmp_path)
+    port = _free_port()
+    bind_url = f"http://127.0.0.1:{port}"
+    try:
+        result = subprocess.run(
+            [sys.executable, "-m", "archivebox", "server", "--daemonize", f"127.0.0.1:{port}"],
+            cwd=tmp_path,
+            env=env,
+            capture_output=True,
+            text=True,
+            timeout=90,
+        )
+        assert result.returncode == 0, result.stderr or result.stdout
+        _wait_for_port("127.0.0.1", port, timeout=30)
+
+        server_process = _wait_for_process(
+            lambda _proc, command: "archivebox" in command and " server " in f" {command} " and bind_url.replace("http://", "") in command,
+        )
+        supervisord = _wait_for_process(
+            lambda proc, command: proc.ppid() == server_process.pid and "supervisord" in command,
+        )
+        _wait_for_process(
+            lambda proc, command: proc.ppid() == supervisord.pid and "supervisord_watchdog" in command,
+        )
+
+        os.kill(server_process.pid, signal.SIGKILL)
+        _wait_for_pid_to_disappear(server_process.pid, timeout=10)
+        _wait_for_pid_to_disappear(supervisord.pid, timeout=20)
+        _assert_no_processes_for_data_dir(tmp_path, timeout=12)
+    finally:
+        _kill_processes_for_data_dir(tmp_path)
+        _assert_no_processes_for_data_dir(tmp_path, timeout=12)
+
+
 @pytest.mark.timeout(240)
-def test_live_second_server_takes_over_existing_server_parent(tmp_path, process):
+def test_live_second_server_takes_over_existing_server_process(tmp_path, process):
     os.chdir(tmp_path)
     assert process.returncode == 0, process.stderr
 
@@ -515,8 +603,8 @@ def test_live_second_server_takes_over_existing_server_parent(tmp_path, process)
         assert first.poll() is None
         first_text = first_log.read_text(encoding="utf-8", errors="replace")
         second_text = second_log.read_text(encoding="utf-8", errors="replace")
-        assert "Newer ArchiveBox server parent took over; standing by." in first_text
-        assert "is now running the orchestrator and server" in second_text
+        assert "A newer archivebox process took over the runtime stack" in first_text
+        assert "Starting orchestrator, server" in second_text
 
         status = subprocess.run(
             [sys.executable, "-m", "archivebox", "status"],
@@ -528,10 +616,10 @@ def test_live_second_server_takes_over_existing_server_parent(tmp_path, process)
         )
         assert status.returncode == 0, status.stderr or status.stdout
 
-        first_takeovers = first_log.read_text(encoding="utf-8", errors="replace").count("is now running the orchestrator and server")
+        first_resumes = first_log.read_text(encoding="utf-8", errors="replace").count("Other newer archivebox process")
         _stop_process(second, signal.SIGTERM)
         second = None
-        _wait_for_log_count(first_log, "is now running the orchestrator and server", first_takeovers + 1, timeout=35)
+        _wait_for_log_count(first_log, "Other newer archivebox process", first_resumes + 1, timeout=35)
         assert first.poll() is None
     finally:
         if second is not None and second.poll() is None:
@@ -567,8 +655,8 @@ def test_live_repeated_server_startups_take_over_cleanly(tmp_path, process):
                 current_log = log_path.read_text(encoding="utf-8", errors="replace")
                 assert previous_server.poll() is None
                 assert _pid_is_alive(server_pids[index - 1])
-                assert "Newer ArchiveBox server parent took over; standing by." in previous_log
-                assert "is now running the orchestrator and server" in current_log
+                assert "A newer archivebox process took over the runtime stack" in previous_log
+                assert "Starting orchestrator, server" in current_log
                 _wait_for_pid_to_disappear(daphne_pids[index - 1], timeout=15)
                 _wait_for_pid_to_disappear(runner_pids[index - 1], timeout=15)
 
@@ -596,10 +684,10 @@ def test_live_repeated_server_startups_take_over_cleanly(tmp_path, process):
 
         previous_log_path = tmp_path / "server-chaos-3.log"
         previous_takeovers = previous_log_path.read_text(encoding="utf-8", errors="replace").count(
-            "is now running the orchestrator and server",
+            "Other newer archivebox process",
         )
         _stop_process(servers[-1], signal.SIGTERM)
-        _wait_for_log_count(previous_log_path, "is now running the orchestrator and server", previous_takeovers + 1, timeout=35)
+        _wait_for_log_count(previous_log_path, "Other newer archivebox process", previous_takeovers + 1, timeout=35)
         assert servers[3].poll() is None
     finally:
         for server in reversed(servers):

archivebox/tests/test_cli_update_reindex_snapshots.py

@@ -290,6 +290,62 @@ def test_reconcile_with_index_json_imports_legacy_archive_results_and_process(tm
     assert '"type": "Process"' in jsonl_text
 
 
+@pytest.mark.django_db
+def test_reconcile_with_index_json_merges_retried_archive_results(tmp_path):
+    from archivebox.base_models.models import get_or_create_system_user_pk
+    from archivebox.core.models import ArchiveResult, Snapshot
+    from archivebox.crawls.models import Crawl
+
+    crawl = Crawl.objects.create(
+        urls="https://example.com",
+        created_by_id=get_or_create_system_user_pk(),
+    )
+    snapshot = Snapshot.objects.create(
+        url="https://example.com",
+        crawl=crawl,
+        status=Snapshot.StatusChoices.SEALED,
+    )
+    output_dir = snapshot.output_dir
+    output_dir.mkdir(parents=True, exist_ok=True)
+    (output_dir / "index.json").write_text(
+        json.dumps(
+            {
+                "url": snapshot.url,
+                "timestamp": snapshot.timestamp,
+                "title": "Example Domain",
+                "archive_results": [
+                    {
+                        "plugin": "dom",
+                        "hook_name": "on_Snapshot__12_dom.js",
+                        "status": "failed",
+                        "output": "first attempt failed",
+                        "start_ts": "2024-01-01T00:00:00+00:00",
+                        "end_ts": "2024-01-01T00:00:01+00:00",
+                    },
+                    {
+                        "plugin": "dom",
+                        "hook_name": "on_Snapshot__12_dom.js",
+                        "status": "succeeded",
+                        "output": "dom/output.html",
+                        "output_files": {"output.html": {"size": 42}},
+                        "output_size": 42,
+                        "start_ts": "2024-01-01T00:01:00+00:00",
+                        "end_ts": "2024-01-01T00:01:01+00:00",
+                    },
+                ],
+            },
+        ),
+    )
+
+    snapshot.reconcile_with_index_json()
+
+    result = ArchiveResult.objects.get(snapshot=snapshot, plugin="dom", hook_name="on_Snapshot__12_dom.js")
+    assert ArchiveResult.objects.filter(snapshot=snapshot, plugin="dom", hook_name="on_Snapshot__12_dom.js").count() == 1
+    assert result.status == ArchiveResult.StatusChoices.SUCCEEDED
+    assert result.output_str == "dom/output.html"
+    assert result.output_size == 42
+
+
 @pytest.mark.django_db
 def test_reconcile_with_index_json_trusts_legacy_archive_results(tmp_path):
     from archivebox.base_models.models import get_or_create_system_user_pk

archivebox/tests/test_crawl_runner.py

@@ -420,38 +420,29 @@ def test_crawl_runner_empty_plugin_selection_emits_lifecycle_and_seals_crawl(tmp
     )
     runner = CrawlRunner(crawl)
 
+    seen_events = {
+        CrawlEvent: [],
+        CrawlSetupEvent: [],
+        CrawlStartEvent: [],
+        SnapshotEvent: [],
+        SnapshotCompletedEvent: [],
+        CrawlCleanupEvent: [],
+        CrawlCompletedEvent: [],
+        MachineEvent: [],
+    }
+    for event_type, events in seen_events.items():
+        runner.bus.on(event_type, lambda event, events=events: events.append(event))
+
     asyncio.run(runner.run())
 
-    async def collect_events():
-        crawl_events = await runner.bus.filter(CrawlEvent, past=True)
-        setup_events = await runner.bus.filter(CrawlSetupEvent, past=True)
-        start_events = await runner.bus.filter(CrawlStartEvent, past=True)
-        snapshot_events = await runner.bus.filter(SnapshotEvent, past=True)
-        snapshot_completed_events = await runner.bus.filter(SnapshotCompletedEvent, past=True)
-        cleanup_events = await runner.bus.filter(CrawlCleanupEvent, past=True)
-        completed_events = await runner.bus.filter(CrawlCompletedEvent, past=True)
-        machine_events = await runner.bus.filter(MachineEvent, past=True)
-        return (
-            crawl_events,
-            setup_events,
-            start_events,
-            snapshot_events,
-            snapshot_completed_events,
-            cleanup_events,
-            completed_events,
-            machine_events,
-        )
-
-    (
-        crawl_events,
-        setup_events,
-        start_events,
-        snapshot_events,
-        snapshot_completed_events,
-        cleanup_events,
-        completed_events,
-        machine_events,
-    ) = asyncio.run(collect_events())
+    crawl_events = seen_events[CrawlEvent]
+    setup_events = seen_events[CrawlSetupEvent]
+    start_events = seen_events[CrawlStartEvent]
+    snapshot_events = seen_events[SnapshotEvent]
+    snapshot_completed_events = seen_events[SnapshotCompletedEvent]
+    cleanup_events = seen_events[CrawlCleanupEvent]
+    completed_events = seen_events[CrawlCompletedEvent]
+    machine_events = seen_events[MachineEvent]
 
     assert len(crawl_events) == 1
     assert len(setup_events) == 1
@@ -474,6 +465,7 @@ def test_crawl_runner_empty_plugin_selection_emits_lifecycle_and_seals_crawl(tmp
     assert crawl.retry_at is None
     assert snapshot.status == Snapshot.StatusChoices.SEALED
     assert snapshot.retry_at is None
+    assert snapshot.archiveresult_set.count() == 0
 
 
 @pytest.mark.django_db(transaction=True)
@@ -789,6 +781,7 @@ def test_create_crawl_api_queues_crawl_without_spawning_runner():
     assert str(crawl.id)
     assert crawl.status == "queued"
     assert crawl.retry_at is not None
+    assert crawl.snapshot_set.filter(url="https://example.com").count() == 1
 
 
 def test_wait_for_snapshot_tasks_surfaces_already_failed_task():

archivebox/workers/management/commands/runner_watch.py

@@ -25,7 +25,7 @@ class Command(BaseCommand):
         from archivebox.machine.models import Machine, Process
         from archivebox.workers.supervisord_util import (
             RUNNER_WORKER,
-            get_existing_supervisord_process,
+            SupervisordConnectionCache,
             get_worker,
             start_worker,
             stop_worker,
@@ -43,6 +43,7 @@ class Command(BaseCommand):
 
         interval = max(0.2, float(kwargs.get("interval", 1.0)))
         last_runserver_id = None
+        supervisor_cache = SupervisordConnectionCache()
 
         def stop_duplicate_watchers() -> None:
             machine = Machine.current()
@@ -58,7 +59,7 @@ class Command(BaseCommand):
                     proc.terminate(graceful_timeout=2.0)
 
         def get_supervisor():
-            supervisor = get_existing_supervisord_process()
+            supervisor = supervisor_cache.get()
             if supervisor is None:
                 raise RuntimeError("runner_watch requires a running supervisord process")
             return supervisor
@@ -114,6 +115,7 @@ class Command(BaseCommand):
                     restart_runner()
                 current.heartbeat()
             except Exception:
+                supervisor_cache.clear()
                 pass
 
             time.sleep(interval)

archivebox/workers/management/commands/supervisord_watchdog.py

@@ -0,0 +1,54 @@
+import time
+
+import psutil
+from django.core.management.base import BaseCommand
+
+
+class Command(BaseCommand):
+    help = "Stop a foreground-owned supervisord if its exact ArchiveBox owner Process exits."
+
+    def add_arguments(self, parser):
+        parser.add_argument("--supervisord-process-id", required=True)
+        parser.add_argument("--interval", type=float, default=1.0)
+
+    def handle(self, *args, **kwargs):
+        from archivebox.core.shutdown_util import wait_psutil_and_kill_children
+        from archivebox.machine.models import Process
+
+        supervisord_process_id = kwargs["supervisord_process_id"]
+        interval = max(0.2, float(kwargs["interval"]))
+
+        while True:
+            try:
+                supervisord_process = Process.objects.select_related("parent").get(id=supervisord_process_id)
+            except Process.DoesNotExist:
+                return
+
+            if supervisord_process.status != Process.StatusChoices.RUNNING:
+                return
+
+            supervisord = supervisord_process.proc
+            if supervisord is None:
+                supervisord_process.mark_exited(exit_code=0)
+                return
+
+            owner = supervisord_process.parent
+            if owner is not None and owner.is_running:
+                time.sleep(interval)
+                continue
+
+            try:
+                children = supervisord.children(recursive=True)
+                supervisord.terminate()
+                for child in children:
+                    try:
+                        child.terminate()
+                    except psutil.NoSuchProcess:
+                        pass
+                wait_psutil_and_kill_children(supervisord, children, timeout=5)
+                supervisord_process.mark_exited(exit_code=0)
+            except psutil.NoSuchProcess:
+                supervisord_process.mark_exited(exit_code=0)
+            except (BrokenPipeError, OSError, psutil.TimeoutExpired):
+                pass
+            return

archivebox/workers/supervisord_util.py

@@ -8,6 +8,7 @@ import psutil
 import shutil
 import subprocess
 import shlex
+import signal
 
 from typing import cast
 from pathlib import Path
@@ -37,6 +38,8 @@ WORKERS_DIR_NAME = "workers"
 # Global reference to supervisord process for cleanup
 _supervisord_proc = None
 _desired_supervisord_workers: dict[str, dict[str, str]] = {}
+_ACTIVE_WORKER_STATES = {"STARTING", "RUNNING", "BACKOFF"}
+_RUNTIME_COMPONENT_ORDER = ("orchestrator", "server", "sonic")
 
 
 def _shell_join(args: list[str]) -> str:
@@ -182,6 +185,27 @@ RUNNER_WATCH_WORKER = lambda bind_url: {
     "redirect_stderr": "true",
 }
 
+SUPERVISORD_PARENT_WATCHDOG_WORKER = lambda supervisord_process_id: {
+    "name": "worker_supervisord_parent_watchdog",
+    "command": _shell_join(
+        [
+            sys.executable,
+            "-m",
+            "archivebox",
+            "manage",
+            "supervisord_watchdog",
+            f"--supervisord-process-id={supervisord_process_id}",
+        ],
+    ),
+    "autostart": "false",
+    "autorestart": "false",
+    "stopasgroup": "true",
+    "killasgroup": "true",
+    "stopwaitsecs": "1",
+    "stdout_logfile": "logs/worker_supervisord_parent_watchdog.log",
+    "redirect_stderr": "true",
+}
+
 SERVER_WORKER = lambda host, port: {
     "name": "worker_daphne",
     "command": _shell_join(
@@ -191,7 +215,7 @@ SERVER_WORKER = lambda host, port: {
             "daphne",
             f"--bind={host}",
             f"--port={port}",
-            "--application-close-timeout=15",
+            "--application-close-timeout=0",
             "archivebox.core.asgi:application",
         ],
     ),
@@ -199,7 +223,7 @@ SERVER_WORKER = lambda host, port: {
     "autorestart": "true",
     "stopasgroup": "true",
     "killasgroup": "true",
-    "stopwaitsecs": "20",
+    "stopwaitsecs": "1",
     "stdout_logfile": "logs/worker_daphne.log",
     "redirect_stderr": "true",
 }
@@ -227,6 +251,9 @@ def RUNSERVER_WORKER(host: str, port: str, *, reload: bool, nothreading: bool =
         "environment": ",".join(environment),
         "autostart": "false",
         "autorestart": "true",
+        "stopasgroup": "true",
+        "killasgroup": "true",
+        "stopwaitsecs": "1",
         "stdout_logfile": "logs/worker_runserver.log",
         "redirect_stderr": "true",
     }
@@ -316,6 +343,29 @@ def create_worker_config(daemon):
     worker_conf.write_text(worker_str)
 
 
+def _current_foreground_supervisord_process_id():
+    if not _supervisord_proc or _supervisord_proc.poll() is not None:
+        return None
+
+    try:
+        from archivebox.machine.models import Machine, Process
+
+        current = Process.current()
+        for process in Process.objects.filter(
+            machine=Machine.current(),
+            process_type=Process.TypeChoices.SUPERVISORD,
+            status=Process.StatusChoices.RUNNING,
+            pwd=str(CONSTANTS.DATA_DIR),
+            pid=_supervisord_proc.pid,
+            parent=current,
+        ).iterator(chunk_size=10):
+            if process.is_running:
+                return process.id
+    except Exception:
+        return None
+    return None
+
+
 def sync_supervisord_workers(supervisor, workers: list[tuple[dict[str, str], bool]], *, prune: bool = True):
     """Project desired workers into supervisord from ArchiveBox-owned state.
 
@@ -331,6 +381,12 @@ def sync_supervisord_workers(supervisor, workers: list[tuple[dict[str, str], boo
 
     global _desired_supervisord_workers
 
+    supervisord_process_id = _current_foreground_supervisord_process_id()
+    if supervisord_process_id is not None:
+        watchdog = SUPERVISORD_PARENT_WATCHDOG_WORKER(supervisord_process_id)
+        if all(worker["name"] != watchdog["name"] for worker, _lazy in workers):
+            workers = [*workers, (watchdog, False)]
+
     desired = {worker["name"]: (worker, lazy) for worker, lazy in workers}
     if prune:
         _desired_supervisord_workers = {name: worker for name, (worker, _lazy) in desired.items()}
@@ -416,11 +472,62 @@ def get_existing_supervisord_process(*, quiet: bool = False):
         return None
 
 
+class SupervisordConnectionCache:
+    """Reuse one XML-RPC proxy until it fails, avoiding hot-loop reconnects."""
+
+    def __init__(self, *, quiet: bool = False):
+        self.quiet = quiet
+        self.supervisor = None
+
+    def clear(self) -> None:
+        self.supervisor = None
+
+    def get(self):
+        if self.supervisor is not None:
+            try:
+                self.supervisor.getPID()
+                return self.supervisor
+            except Exception:
+                self.supervisor = None
+
+        supervisor = get_existing_supervisord_process(quiet=self.quiet)
+        if supervisor is None:
+            return None
+
+        self.supervisor = supervisor
+        return supervisor
+
+
 def stop_existing_supervisord_process():
     global _supervisord_proc
     SOCK_FILE = get_sock_file()
     PID_FILE = SOCK_FILE.parent / PID_FILE_NAME
     stop_grace_seconds = configured_stopwaitsecs(tuple(_desired_supervisord_workers.values()))
+    live_supervisord = _live_supervisord_processes_from_db()
+
+    for process, _proc in live_supervisord:
+        if process is None or process.parent_id is None:
+            continue
+        owner = process.parent
+        if owner.pid == os.getpid() or not owner.is_running:
+            continue
+        owner_proc = owner.proc
+        if owner_proc is None:
+            owner.mark_exited(exit_code=0)
+            continue
+        try:
+            print(f"[🦸‍♂️] Stopping older ArchiveBox runtime owner (pid={owner_proc.pid})...")
+            owner_proc.terminate()
+            try:
+                owner_proc.wait(timeout=min(stop_grace_seconds, 5))
+            except psutil.TimeoutExpired:
+                owner_proc.kill()
+                owner_proc.wait(timeout=2)
+            owner.mark_exited(exit_code=0)
+        except psutil.NoSuchProcess:
+            owner.mark_exited(exit_code=0)
+        except (BrokenPipeError, OSError, psutil.TimeoutExpired):
+            pass
 
     supervisor = get_existing_supervisord_process(quiet=True)
     supervisor_pid = None
@@ -691,7 +798,7 @@ def start_worker(supervisor, daemon, lazy=False):
     return sync_supervisord_workers(supervisor, [(daemon, lazy)], prune=False).get(daemon["name"])
 
 
-def run_runner_worker(args: list[str], *, name: str = "worker_runner_once") -> int:
+def run_runner_worker(args: list[str], *, name: str = "worker_runner_once", interactive_interrupts: bool = False) -> int:
     supervisor = get_or_create_supervisord_process(daemonize=False)
     worker = RUNNER_ONCE_WORKER(args, name=name)
     log_path = Path(worker["stdout_logfile"])
@@ -703,26 +810,38 @@ def run_runner_worker(args: list[str], *, name: str = "worker_runner_once") -> i
     log_handle.seek(0, 2)
     sync_supervisord_workers(supervisor, [(worker, False)], prune=False)
     final_states = {"STOPPED", "EXITED", "FATAL", "UNKNOWN"}
+    forwarded_interrupt = False
     try:
         while True:
-            while True:
-                line = log_handle.readline()
-                if not line:
-                    break
-                sys.stdout.write(line)
-                sys.stdout.flush()
-            proc = get_worker(supervisor, name)
-            if proc is None:
-                return 1
-            if proc["statename"] in final_states:
+            try:
                 while True:
                     line = log_handle.readline()
                     if not line:
                         break
                     sys.stdout.write(line)
                     sys.stdout.flush()
-                return int(proc.get("exitstatus") or 0) if proc["statename"] == "EXITED" else 1
-            time.sleep(0.5)
+                proc = get_worker(supervisor, name)
+                if proc is None:
+                    return 1
+                if proc["statename"] in final_states:
+                    while True:
+                        line = log_handle.readline()
+                        if not line:
+                            break
+                        sys.stdout.write(line)
+                        sys.stdout.flush()
+                    return int(proc.get("exitstatus") or 0) if proc["statename"] == "EXITED" else 1
+                time.sleep(0.5)
+            except KeyboardInterrupt:
+                if not interactive_interrupts or forwarded_interrupt:
+                    raise
+                forwarded_interrupt = True
+                proc = get_worker(supervisor, name)
+                pid = int(proc.get("pid") or 0) if proc else 0
+                if pid <= 0:
+                    raise
+                print("[yellow][*] Forwarding Ctrl+C to the active crawl hook...[/yellow]")
+                os.kill(pid, signal.SIGINT)
     finally:
         log_handle.close()
 
@@ -735,6 +854,82 @@ def get_worker(supervisor, daemon_name):
     return None
 
 
+def format_runtime_components(components: list[str] | tuple[str, ...]) -> str:
+    return ", ".join(component for component in components if component)
+
+
+def worker_runtime_component(worker_name: str, *, config=None) -> str | None:
+    if worker_name in {RUNNER_WORKER["name"], RUNNER_WATCH_WORKER("")["name"]}:
+        return "orchestrator"
+    if worker_name in {SERVER_WORKER("", "")["name"], RUNSERVER_WORKER("", "", reload=False)["name"]}:
+        return "server"
+    if config is not None:
+        sonic_worker = get_sonic_supervisord_worker_from_plugin(config)
+        if sonic_worker and worker_name == sonic_worker.get("name"):
+            return "sonic"
+    return None
+
+
+def runtime_components_for_worker_names(worker_names: set[str] | list[str] | tuple[str, ...], *, config=None) -> list[str]:
+    components = {worker_runtime_component(worker_name, config=config) for worker_name in worker_names}
+    return [component for component in _RUNTIME_COMPONENT_ORDER if component in components]
+
+
+def active_supervisord_runtime_components(*, config=None, supervisor=None) -> list[str]:
+    supervisor = supervisor or get_existing_supervisord_process(quiet=True)
+    if supervisor is None:
+        return []
+    try:
+        worker_names = {proc.get("name") for proc in supervisor.getAllProcessInfo() if proc.get("statename") in _ACTIVE_WORKER_STATES}
+    except Exception:
+        return []
+    return runtime_components_for_worker_names({str(name) for name in worker_names if name}, config=config)
+
+
+def build_server_worker_plan(*, config, host: str, port: str, debug: bool, reload: bool, nothreading: bool, supervisor=None):
+    bind_url = f"http://{host}:{port}"
+
+    if debug:
+        server_worker = RUNSERVER_WORKER(host=host, port=port, reload=reload, nothreading=nothreading)
+        bg_workers: list[tuple[dict[str, str], bool]] = (
+            [(RUNNER_WORKER, True), (RUNNER_WATCH_WORKER(bind_url), False)] if reload else [(RUNNER_WORKER, False)]
+        )
+        log_files = ["logs/worker_runserver.log", "logs/worker_runner.log"]
+        if reload:
+            log_files.insert(1, "logs/worker_runner_watch.log")
+    else:
+        server_worker = SERVER_WORKER(host=host, port=port)
+        bg_workers = [(RUNNER_WORKER, False)]
+        log_files = ["logs/worker_daphne.log", "logs/worker_runner.log"]
+
+    sonic_worker = get_sonic_supervisord_worker_from_plugin(config)
+    if sonic_worker is not None:
+        try:
+            current_sonic = get_worker(supervisor, sonic_worker["name"]) if supervisor is not None else None
+            supervisor_pid = supervisor.getPID() if supervisor is not None else None
+        except Exception:
+            current_sonic = None
+            supervisor_pid = None
+        sonic_host = str(getattr(config, "SEARCH_BACKEND_SONIC_HOST_NAME", "127.0.0.1") or "127.0.0.1")
+        if sonic_host.strip().lower() == "localhost":
+            sonic_host = "127.0.0.1"
+        sonic_port = int(getattr(config, "SEARCH_BACKEND_SONIC_PORT"))
+        if not (isinstance(current_sonic, dict) and current_sonic.get("statename") in ("STARTING", "RUNNING")):
+            stop_stale_sonic_processes(sonic_worker, supervisor_pid=supervisor_pid, host=sonic_host, port=sonic_port)
+        if not (isinstance(current_sonic, dict) and current_sonic.get("statename") in ("STARTING", "RUNNING")) and is_port_in_use(
+            sonic_host,
+            sonic_port,
+        ):
+            print(f"[yellow][*] Sonic is already listening on {sonic_host}:{sonic_port}; not starting a duplicate worker.[/yellow]")
+        else:
+            bg_workers.insert(0, (sonic_worker, False))
+            log_files.append(str(sonic_worker["stdout_logfile"]))
+
+    workers = [(server_worker, False), *bg_workers]
+    components = runtime_components_for_worker_names([worker["name"] for worker, _lazy in workers], config=config)
+    return workers, log_files, components
+
+
 def stop_worker(supervisor, daemon_name):
     proc = get_worker(supervisor, daemon_name)
 
@@ -806,7 +1001,7 @@ def tail_multiple_worker_logs(log_files: list[str], follow=True, proc=None, keep
     try:
         while follow:
             if keep_running is not None and not keep_running():
-                print("\n[newer ArchiveBox parent is now running the orchestrator and server]")
+                print("\n[newer ArchiveBox process is now running the orchestrator and server]")
                 return "transferred"
 
             # Check if the monitored process has exited
@@ -857,11 +1052,87 @@ def get_sonic_supervisord_worker_from_plugin(config) -> dict[str, str] | None:
     return cast(dict[str, str] | None, worker)
 
 
-def stop_stale_sonic_processes(sonic_worker: dict[str, str], *, supervisor_pid: int | None) -> None:
+def _proc_cmdline(proc: psutil.Process) -> list[str]:
+    try:
+        return proc.cmdline()
+    except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess):
+        return []
+
+
+def _is_sonic_process(proc: psutil.Process) -> bool:
+    cmdline = _proc_cmdline(proc)
+    return bool(cmdline and Path(cmdline[0]).name == "sonic")
+
+
+def _is_supervisord_process(proc: psutil.Process | None) -> bool:
+    if proc is None:
+        return False
+    cmdline = _proc_cmdline(proc)
+    return any(Path(part).name == "supervisord" for part in cmdline)
+
+
+def _has_live_archivebox_parent(proc: psutil.Process | None) -> bool:
+    try:
+        parent = proc.parent() if proc else None
+    except (psutil.NoSuchProcess, psutil.AccessDenied):
+        return False
+    if parent is None or parent.pid <= 1:
+        return False
+    cmdline = _proc_cmdline(parent)
+    return any("archivebox" in part for part in cmdline)
+
+
+def _terminate_process_tree(root: psutil.Process, *, timeout: float = 2.0) -> None:
+    try:
+        children = root.children(recursive=True)
+    except psutil.NoSuchProcess:
+        return
+    try:
+        root.terminate()
+    except psutil.NoSuchProcess:
+        return
+    for child in children:
+        try:
+            child.terminate()
+        except psutil.NoSuchProcess:
+            pass
+    _gone, alive = psutil.wait_procs([root, *children], timeout=timeout)
+    for proc in alive:
+        try:
+            proc.kill()
+        except psutil.NoSuchProcess:
+            pass
+    psutil.wait_procs(alive, timeout=timeout)
+
+
+def _sonic_listeners(host: str, port: int) -> list[psutil.Process]:
+    listeners = []
+    for proc in psutil.process_iter(["pid", "name", "cmdline"]):
+        if not _is_sonic_process(proc):
+            continue
+        try:
+            connections = proc.net_connections(kind="tcp")
+        except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess):
+            continue
+        for conn in connections:
+            if conn.status != psutil.CONN_LISTEN or not conn.laddr or conn.laddr.port != port:
+                continue
+            addr = str(conn.laddr.ip)
+            if host in {"0.0.0.0", "::", addr} or addr in {"0.0.0.0", "::"}:
+                listeners.append(proc)
+                break
+    return listeners
+
+
+def stop_stale_sonic_processes(
+    sonic_worker: dict[str, str],
+    *,
+    supervisor_pid: int | None,
+    host: str | None = None,
+    port: int | None = None,
+) -> None:
     command = shlex.split(sonic_worker.get("command") or "")
     config_path = Path(command[command.index("-c") + 1]).resolve() if "-c" in command and command.index("-c") + 1 < len(command) else None
-    if config_path is None:
-        return
 
     stale = []
     for proc in psutil.process_iter(["pid", "ppid", "name", "cmdline"]):
@@ -869,28 +1140,37 @@ def stop_stale_sonic_processes(sonic_worker: dict[str, str], *, supervisor_pid:
             cmdline = proc.info.get("cmdline") or []
             if proc.info["pid"] == os.getpid() or proc.info["ppid"] == supervisor_pid:
                 continue
-            if Path(cmdline[0]).name != "sonic" or str(config_path) not in cmdline:
+            if config_path is None or Path(cmdline[0]).name != "sonic" or str(config_path) not in cmdline:
                 continue
             stale.append(proc)
         except (IndexError, psutil.NoSuchProcess, psutil.AccessDenied):
             continue
 
+    if host is not None and port is not None:
+        for proc in _sonic_listeners(host, port):
+            try:
+                proc_ppid = proc.ppid()
+            except (psutil.NoSuchProcess, psutil.AccessDenied):
+                continue
+            if proc.pid == os.getpid() or proc_ppid == supervisor_pid:
+                continue
+            try:
+                supervisor = proc.parent()
+            except (psutil.NoSuchProcess, psutil.AccessDenied):
+                supervisor = None
+            if _is_supervisord_process(supervisor) and not _has_live_archivebox_parent(supervisor):
+                stale.append(supervisor)
+            elif proc_ppid <= 1:
+                stale.append(proc)
+
     if not stale:
         return
 
-    print(f"[yellow][*] Taking over stale Sonic daemon(s) using {pretty_path(config_path)}...[/yellow]")
-    for proc in stale:
-        try:
-            proc.terminate()
-        except psutil.NoSuchProcess:
-            pass
-    _gone, alive = psutil.wait_procs(stale, timeout=2.0)
-    for proc in alive:
-        try:
-            proc.kill()
-        except psutil.NoSuchProcess:
-            pass
-    psutil.wait_procs(alive, timeout=2.0)
+    unique_stale = {proc.pid: proc for proc in stale}.values()
+    target = f"{host}:{port}" if host and port else pretty_path(config_path) if config_path else "unknown Sonic target"
+    print(f"[yellow][*] Taking over stale Sonic daemon(s) using {target}...[/yellow]")
+    for proc in unique_stale:
+        _terminate_process_tree(proc)
 
 
 def start_server_workers(
@@ -902,6 +1182,7 @@ def start_server_workers(
     nothreading=False,
     keep_running=None,
     should_stop_supervisord=None,
+    resumed_from_pid=None,
 ):
     from archivebox.config.common import get_config
 
@@ -910,46 +1191,26 @@ def start_server_workers(
     tail_result = "stopped"
     try:
         supervisor = get_or_create_supervisord_process(daemonize=daemonize)
-        bind_url = f"http://{host}:{port}"
-
-        if debug:
-            server_worker = RUNSERVER_WORKER(host=host, port=port, reload=reload, nothreading=nothreading)
-            bg_workers: list[tuple[dict[str, str], bool]] = (
-                [(RUNNER_WORKER, True), (RUNNER_WATCH_WORKER(bind_url), False)] if reload else [(RUNNER_WORKER, False)]
+        workers, log_files, components = build_server_worker_plan(
+            config=config,
+            host=host,
+            port=port,
+            debug=debug,
+            reload=reload,
+            nothreading=nothreading,
+            supervisor=supervisor,
+        )
+        component_list = format_runtime_components(components)
+        if resumed_from_pid:
+            print(
+                "[yellow][*] Other newer archivebox process "
+                f"(pid={resumed_from_pid}) exited, taking over {component_list} in this process again...[/yellow]",
             )
-            log_files = ["logs/worker_runserver.log", "logs/worker_runner.log"]
-            if reload:
-                log_files.insert(1, "logs/worker_runner_watch.log")
         else:
-            server_worker = SERVER_WORKER(host=host, port=port)
-            bg_workers = [(RUNNER_WORKER, False)]
-            log_files = ["logs/worker_daphne.log", "logs/worker_runner.log"]
-
-        sonic_worker = get_sonic_supervisord_worker_from_plugin(config)
-        if sonic_worker is not None:
-            try:
-                current_sonic = get_worker(supervisor, sonic_worker["name"])
-                supervisor_pid = supervisor.getPID()
-            except Exception:
-                current_sonic = None
-                supervisor_pid = None
-            if not (isinstance(current_sonic, dict) and current_sonic.get("statename") in ("STARTING", "RUNNING")):
-                stop_stale_sonic_processes(sonic_worker, supervisor_pid=supervisor_pid)
-            sonic_host = str(getattr(config, "SEARCH_BACKEND_SONIC_HOST_NAME", "127.0.0.1") or "127.0.0.1")
-            if sonic_host.strip().lower() == "localhost":
-                sonic_host = "127.0.0.1"
-            sonic_port = int(getattr(config, "SEARCH_BACKEND_SONIC_PORT"))
-            if not (isinstance(current_sonic, dict) and current_sonic.get("statename") in ("STARTING", "RUNNING")) and is_port_in_use(
-                sonic_host,
-                sonic_port,
-            ):
-                print(f"[yellow][*] Sonic is already listening on {sonic_host}:{sonic_port}; not starting a duplicate worker.[/yellow]")
-            else:
-                bg_workers.insert(0, (sonic_worker, False))
-                log_files.append(str(sonic_worker["stdout_logfile"]))
+            print(f"[*] Starting {component_list} in this process (pid={os.getpid()})...")
 
         print()
-        sync_supervisord_workers(supervisor, [(server_worker, False), *bg_workers], prune=True)
+        sync_supervisord_workers(supervisor, workers, prune=True)
         print()
 
         if daemonize:
@@ -987,7 +1248,7 @@ def start_server_workers(
             # Ensure supervisord and all children are stopped only while this
             # foreground parent is still the active server parent. Standby
             # parents must not tear down a newer leader's services.
-            stop_existing_supervisord_process()
+            stop_own_supervisord_process()
     return tail_result
 
 

bin/release.sh

@@ -221,6 +221,31 @@ else:
 PY
 }
 
+github_release_enabled() {
+    local version="$1"
+    if [[ "${version}" == *rc* && "${CREATE_GITHUB_RC_RELEASES:-0}" != "1" ]]; then
+        return 1
+    fi
+    if [[ "${CREATE_GITHUB_RELEASES:-1}" == "0" ]]; then
+        return 1
+    fi
+    return 0
+}
+
+create_git_tag() {
+    local version="$1"
+    local tag="${TAG_PREFIX}${version}"
+    if git rev-parse -q --verify "refs/tags/${tag}" >/dev/null; then
+        if [[ "$(git rev-list -n1 "${tag}")" != "$(git rev-parse HEAD)" ]]; then
+            echo "Tag ${tag} already exists but does not point at HEAD" >&2
+            return 1
+        fi
+    else
+        git tag -a "${tag}" -m "release: ${tag}"
+    fi
+    git push origin "refs/tags/${tag}"
+}
+
 wait_for_runs() {
     local slug="$1"
     local event="$2"
@@ -337,6 +362,10 @@ create_release() {
     local slug="$1"
     local version="$2"
     local prerelease_args=()
+    if ! github_release_enabled "${version}"; then
+        echo "Skipping GitHub release object for ${TAG_PREFIX}${version}; git tag will still be pushed."
+        return 0
+    fi
     if [[ "${version}" == *rc* ]]; then
         prerelease_args+=(--prerelease)
     fi
@@ -437,9 +466,10 @@ main() {
     fi
 
     publish_artifacts "${version}"
+    create_git_tag "${version}"
     create_release "${slug}" "${version}"
 
-    if ! gh release view "${TAG_PREFIX}${version}" --repo "${slug}" >/dev/null 2>&1; then
+    if github_release_enabled "${version}" && ! gh release view "${TAG_PREFIX}${version}" --repo "${slug}" >/dev/null 2>&1; then
         echo "GitHub release ${TAG_PREFIX}${version} was not found after creation" >&2
         return 1
     fi

bin/release_dev_stack.sh

@@ -133,6 +133,7 @@ commit_push_publish() {
     local branch="$2"
     local package="$3"
     local version="$4"
+    local tag="v${version}"
 
     (
         cd "$repo"
@@ -146,6 +147,15 @@ commit_push_publish() {
             echo "[*] No staged changes in ${package}; reusing existing commit."
         fi
         git push origin "$branch"
+        if git rev-parse -q --verify "refs/tags/${tag}" >/dev/null; then
+            if [[ "$(git rev-list -n1 "${tag}")" != "$(git rev-parse HEAD)" ]]; then
+                echo "[X] Tag ${tag} already exists but does not point at HEAD in ${package}" >&2
+                exit 1
+            fi
+        else
+            git tag -a "${tag}" -m "release: ${package} ${version}"
+        fi
+        git push origin "refs/tags/${tag}"
         uv --no-cache publish --username="${PYPI_USERNAME}" dist/*
     )
 }

etc/package.json

@@ -1,6 +1,6 @@
 {
   "name": "archivebox",
-  "version": "0.9.33rc26",
+  "version": "0.9.33rc29",
   "repository": "github:ArchiveBox/ArchiveBox",
   "license": "MIT",
   "dependencies": {

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "archivebox"
-version = "0.9.33rc26"
+version = "0.9.33rc29"
 requires-python = ">=3.13"
 description = "Self-hosted internet archiving solution."
 authors = [{name = "Nick Sweeting", email = "pyproject.toml@archivebox.io"}]
@@ -79,9 +79,9 @@ dependencies = [
     ### Extractor dependencies (optional binaries detected at runtime via shutil.which)
     ### Binary/Package Management
     "abxbus==2.5.7",          # EventBus API
-    "abxpkg>=1.11.51",        # for: detecting, versioning, and installing binaries via apt/brew/pip/npm
-    "abx-plugins>=1.11.58",  # shared ArchiveBox plugin package with Chrome/Puppeteer dependency wiring
-    "abx-dl>=1.11.58",       # shared ArchiveBox downloader package with blocking install preflight
+    "abxpkg>=1.11.53",        # for: detecting, versioning, and installing binaries via apt/brew/pip/npm
+    "abx-plugins>=1.11.60",  # shared ArchiveBox plugin package with Chrome/Puppeteer dependency wiring
+    "abx-dl>=1.11.60",       # 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
 ]