fix: stabilize dockerized crawl runtime

.dockerignore

@@ -19,6 +19,10 @@ venv/
 .docker_venv/
 .docker-venv/
 node_modules/
+abx-dl/
+abxpkg/
+abx-plugins/
+abxbus/
 chrome/
 chromeprofile/
 chrome_profile/

Dockerfile

@@ -145,7 +145,7 @@ RUN --mount=type=cache,target=/var/cache/apt,sharing=locked,id=apt-$TARGETARCH$T
     echo "[+] APT Installing extractor dependencies for $TARGETPLATFORM..." \
     && apt-get update -qq \
     && apt-get install -qq -y --no-install-recommends \
-        git ripgrep \
+        git ripgrep default-jre \
         # Packages we have also needed in the past:
         # youtube-dl wget2 aria2 python3-pyxattr rtmpdump libfribidi-bin mpv \
         # curl wget (already installed above)
@@ -205,8 +205,6 @@ RUN --mount=type=cache,target=/var/cache/apt,sharing=locked,id=apt-$TARGETARCH$T
     && apt-get install -y --no-upgrade \
         nodejs \
     && rm -rf /var/lib/apt/lists/* \
-    # Update NPM to latest version
-    && npm i -g npm --cache /root/.npm \
     # Save version info
     && ( \
         which node && node --version \
@@ -282,8 +280,9 @@ RUN --mount=type=cache,target=/var/cache/apt,sharing=locked,id=apt-$TARGETARCH$T
     && uv pip install "playwright>=1.49.1" \
     && uv run playwright install chromium --no-shell --with-deps \
     && export CHROME_BINARY="$(uv run python -c 'from playwright.sync_api import sync_playwright; print(sync_playwright().start().chromium.executable_path)')" \
-    && ln -s "$CHROME_BINARY" /usr/bin/chromium-browser \
-    && ln -s /browsers/ffmpeg-*/ffmpeg-linux /usr/bin/ffmpeg \
+    && ln -sf "$CHROME_BINARY" /usr/bin/chromium-browser \
+    && ln -sf "$CHROME_BINARY" /usr/bin/chromium \
+    && ln -sf /browsers/ffmpeg-*/ffmpeg-linux /usr/bin/ffmpeg \
     && mkdir -p "/home/${ARCHIVEBOX_USER}/.config/chromium/Crash Reports/pending/" \
     && chown -R "$DEFAULT_PUID:$DEFAULT_PGID" "/home/${ARCHIVEBOX_USER}/.config" \
     && mkdir -p "$PLAYWRIGHT_BROWSERS_PATH" \
@@ -294,13 +293,25 @@ RUN --mount=type=cache,target=/var/cache/apt,sharing=locked,id=apt-$TARGETARCH$T
     && rm -rf /var/lib/apt/lists/* \
     && ( \
         uv pip show playwright \
+        && which chromium && chromium --version \
         && which chromium-browser && /usr/bin/chromium-browser --version || /usr/lib/chromium/chromium --version \
         && which ffmpeg && ffmpeg -version \
         && echo -e '\n\n' \
     ) | tee -a /VERSION.txt
 
 # Install Node extractor dependencies
-ENV PATH="/home/$ARCHIVEBOX_USER/.npm/bin:$PATH"
+ENV PATH="/home/$ARCHIVEBOX_USER/.npm/bin:$PATH" \
+    PERSONAS_DIR=/data/personas \
+    NODE_PATH="/home/$ARCHIVEBOX_USER/.npm/lib/node_modules:/usr/lib/node_modules:/usr/share/archivebox/lib/npm/node_modules:/data/personas/Default/node_modules" \
+    PUPPETEER_EXECUTABLE_PATH=/usr/bin/chromium-browser \
+    CHROME_BIN=/usr/bin/chromium-browser \
+    CHROME_BINARY=/usr/bin/chromium-browser \
+    CHROMIUM_BINARY=/usr/bin/chromium-browser \
+    CHROME_USER_DATA_DIR=/data/personas/Default/chrome_profile \
+    CHROME_HEADLESS=true \
+    CHROME_SANDBOX=false \
+    CHROME_ISOLATION=snapshot \
+    CHROME_ARGS_EXTRA='["--disable-gpu","--disable-features=Translate,OptimizationGuideModelDownloading,MediaRouter"]'
 USER $ARCHIVEBOX_USER
 WORKDIR "/home/$ARCHIVEBOX_USER/.npm"
 RUN --mount=type=cache,target=/home/archivebox/.npm_cache,sharing=locked,id=npm-$TARGETARCH$TARGETVARIANT,uid=$DEFAULT_PUID,gid=$DEFAULT_PGID \
@@ -310,6 +321,7 @@ RUN --mount=type=cache,target=/home/archivebox/.npm_cache,sharing=locked,id=npm-
         "@postlight/parser@^2.2.3" \
         "readability-extractor@github:ArchiveBox/readability-extractor" \
         "single-file-cli@^1.1.54" \
+        "puppeteer-core@^23.5.0" \
         "puppeteer@^23.5.0" \
         "@puppeteer/browsers@^2.4.0" \
     && rm -Rf "/home/$ARCHIVEBOX_USER/.cache/puppeteer"
@@ -331,17 +343,24 @@ RUN ( \
 # Install ArchiveBox Python venv dependencies from uv.lock
 RUN --mount=type=bind,source=pyproject.toml,target=/app/pyproject.toml \
     --mount=type=bind,source=uv.lock,target=/app/uv.lock \
+    --mount=type=cache,target=/var/cache/apt,sharing=locked,id=apt-$TARGETARCH$TARGETVARIANT \
     --mount=type=cache,target=/root/.cache/uv,sharing=locked,id=uv-$TARGETARCH$TARGETVARIANT \
     echo "[+] PIP Installing ArchiveBox dependencies from pyproject.toml and uv.lock..." \
+    && apt-get update -qq \
+    && apt-get install -qq -y --no-install-recommends build-essential gcc python3-dev \
     && uv sync \
         --frozen \
         --inexact \
         --all-extras \
         --no-install-project \
-        --no-install-workspace
+        --no-install-workspace \
+    && apt-get purge -y python3-dev build-essential gcc \
+    && apt-get autoremove -y \
+    && rm -rf /var/lib/apt/lists/*
     # installs the pip packages that archivebox depends on, defined in pyproject.toml and uv.lock dependencies
 
-# Install ArchiveBox Python package + workspace dependencies from source
+# Install ArchiveBox Python package from the checked-out source.
+# Sibling abx-* packages are installed from uv.lock inside the container, not copied from the local checkout.
 COPY --chown=root:root --chmod=755 "." "$CODE_DIR/"
 RUN --mount=type=cache,target=/root/.cache/uv,sharing=locked,id=uv-$TARGETARCH$TARGETVARIANT \
     echo "[*] Installing ArchiveBox Python source code from $CODE_DIR..." \

archivebox/services/archive_result_service.py

@@ -285,7 +285,7 @@ class ArchiveResultService(BaseService):
         records = _iter_archiveresult_records(event.stdout)
         if records:
             for record in records:
-                await self.bus.emit(
+                await event.emit(
                     ArchiveResultEvent(
                         snapshot_id=record.get("snapshot_id") or snapshot_event.snapshot_id,
                         plugin=record.get("plugin") or event.plugin_name,
@@ -298,10 +298,10 @@ class ArchiveResultService(BaseService):
                         end_ts=event.end_ts,
                         error=record.get("error") or (event.stderr if event.exit_code != 0 else ""),
                     ),
-                )
+                ).now()
             return
 
-        await self.bus.emit(
+        await event.emit(
             ArchiveResultEvent(
                 snapshot_id=snapshot_event.snapshot_id,
                 plugin=event.plugin_name,
@@ -313,4 +313,4 @@ class ArchiveResultService(BaseService):
                 end_ts=event.end_ts,
                 error=event.stderr if event.exit_code != 0 else "",
             ),
-        )
+        ).now()

archivebox/services/binary_service.py

@@ -15,7 +15,7 @@ class BinaryService(BaseService):
         self.bus.on(BinaryRequestEvent, self.on_BinaryRequestEvent)
         self.bus.on(BinaryEvent, self.on_BinaryEvent)
 
-    async def on_BinaryRequestEvent(self, event: BinaryRequestEvent) -> None:
+    async def on_BinaryRequestEvent(self, event: BinaryRequestEvent) -> str | None:
         from archivebox.machine.models import Binary, Machine
 
         machine = await sync_to_async(Machine.current, thread_sensitive=True)()
@@ -58,21 +58,22 @@ class BinaryService(BaseService):
                 "overrides": installed.overrides or {},
             }
         if cached is not None:
-            await self.bus.emit(
-                BinaryEvent(
-                    name=event.name,
-                    plugin_name=event.plugin_name,
-                    hook_name=event.hook_name,
-                    abspath=cached["abspath"],
-                    version=cached["version"],
-                    sha256=cached["sha256"],
-                    binproviders=event.binproviders or cached["binproviders"],
-                    binprovider=cached["binprovider"],
-                    overrides=event.overrides or cached["overrides"],
-                    binary_id=event.binary_id,
-                    machine_id=cached["machine_id"],
-                ),
+            binary_event = BinaryEvent(
+                name=event.name,
+                plugin_name=event.plugin_name,
+                hook_name=event.hook_name,
+                abspath=cached["abspath"],
+                version=cached["version"],
+                sha256=cached["sha256"],
+                binproviders=event.binproviders or cached["binproviders"],
+                binprovider=cached["binprovider"],
+                overrides=event.overrides or cached["overrides"],
+                binary_id=event.binary_id,
+                machine_id=cached["machine_id"],
             )
+            await event.emit(binary_event).now()
+            return binary_event.abspath
+        return None
 
     async def on_BinaryEvent(self, event: BinaryEvent) -> None:
         from archivebox.machine.models import Binary, Machine

archivebox/services/process_service.py

@@ -23,6 +23,13 @@ def parse_event_datetime(value: str | None):
     return dt
 
 
+def current_network_interface_with_machine():
+    from archivebox.machine.models import NetworkInterface
+
+    current_iface = NetworkInterface.current(refresh=True)
+    return NetworkInterface.objects.select_related("machine").get(id=current_iface.id)
+
+
 class ProcessService(BaseService):
     LISTENS_TO: ClassVar[list[type[BaseEvent]]] = [ProcessStartedEvent, ProcessCompletedEvent]
     EMITS: ClassVar[list[type[BaseEvent]]] = []
@@ -33,9 +40,9 @@ class ProcessService(BaseService):
         self.bus.on(ProcessCompletedEvent, self.on_ProcessCompletedEvent__save_to_db)
 
     async def on_ProcessStartedEvent__save_to_db(self, event: ProcessStartedEvent) -> None:
-        from archivebox.machine.models import NetworkInterface, Process
+        from archivebox.machine.models import Process
 
-        iface = await sync_to_async(NetworkInterface.current, thread_sensitive=True)(refresh=True)
+        iface = await sync_to_async(current_network_interface_with_machine, thread_sensitive=True)()
         process_type = event.process_type or (
             Process.TypeChoices.BINARY if event.hook_name.startswith("on_BinaryRequest") else Process.TypeChoices.HOOK
         )
@@ -92,9 +99,9 @@ class ProcessService(BaseService):
         await process.asave()
 
     async def on_ProcessCompletedEvent__save_to_db(self, event: ProcessCompletedEvent) -> None:
-        from archivebox.machine.models import NetworkInterface, Process
+        from archivebox.machine.models import Process
 
-        iface = await sync_to_async(NetworkInterface.current, thread_sensitive=True)(refresh=True)
+        iface = await sync_to_async(current_network_interface_with_machine, thread_sensitive=True)()
         process_type = event.process_type or (
             Process.TypeChoices.BINARY if event.hook_name.startswith("on_BinaryRequest") else Process.TypeChoices.HOOK
         )

archivebox/services/runner.py

@@ -16,22 +16,37 @@ from asgiref.sync import sync_to_async
 from django.utils import timezone
 from rich.console import Console
 
-from abx_dl.events import BinaryRequestEvent, MachineEvent
+from abx_dl.events import (
+    BinaryRequestEvent,
+    CrawlCleanupEvent,
+    CrawlEvent,
+    CrawlStartEvent,
+    InstallEvent,
+    MachineEvent,
+    ProcessCompletedEvent,
+    ProcessEvent,
+    SnapshotEvent,
+    slow_warning_timeout,
+)
 from abx_dl.heartbeat import CrawlHeartbeat
 from abx_dl.limits import CrawlLimitState
 from abx_dl.models import Plugin, Snapshot as AbxSnapshot, discover_plugins, filter_plugins
 from abx_dl.orchestrator import (
+    compute_install_phase_timeout,
+    compute_phase_timeout,
     create_bus,
-    download,
+    get_install_plugins,
     install_plugins as abx_install_plugins,
     setup_services as setup_abx_services,
 )
+from abx_dl.services.process_service import ProcessService as HookProcessService
+from abx_dl.services.snapshot_service import SnapshotService as HookSnapshotService
 
 from .archive_result_service import ArchiveResultService
 from .binary_service import BinaryService
 from .crawl_service import CrawlService
 from .machine_service import MachineService
-from .process_service import ProcessService
+from .process_service import ProcessService as PersistedProcessService
 from .snapshot_service import SnapshotService
 from .tag_service import TagService
 from .live_ui import LiveBusUI
@@ -64,14 +79,14 @@ async def _emit_machine_config(
             config=user_config,
             config_type="user",
         ),
-    )
+    ).now()
     if derived_machine_config:
         await bus.emit(
             MachineEvent(
                 config=derived_machine_config,
                 config_type="derived",
             ),
-        )
+        ).now()
 
 
 def ensure_background_runner(*, allow_under_pytest: bool = False) -> bool:
@@ -123,7 +138,8 @@ class CrawlRunner:
         self.crawl = crawl
         self.bus = create_bus(name=_bus_name("ArchiveBox", str(crawl.id)), total_timeout=3600.0)
         self.plugins = discover_plugins()
-        ProcessService(self.bus)
+        HookProcessService(self.bus, emit_jsonl=False, interactive_tty=False)
+        PersistedProcessService(self.bus)
         BinaryService(self.bus)
         TagService(self.bus)
         CrawlService(self.bus, crawl_id=str(crawl.id))
@@ -149,6 +165,10 @@ class CrawlRunner:
         self.primary_url = ""
         self.crawl_output_dir = ""
         self._live_stream = None
+        self.root_crawl_event_id: str | None = None
+
+    def runtime_plugins(self) -> dict[str, Plugin]:
+        return filter_plugins(self.plugins, self.selected_plugins, include_providers=True) if self.selected_plugins else self.plugins
 
     async def run(self) -> None:
         heartbeat = CrawlHeartbeat(
@@ -161,28 +181,6 @@ class CrawlRunner:
             live_ui = self._create_live_ui()
             with live_ui if live_ui is not None else nullcontext():
                 await heartbeat.start()
-                if snapshot_ids:
-                    root_snapshot = await sync_to_async(self.load_snapshot_payload, thread_sensitive=True)(snapshot_ids[0])
-                    setup_abx_services(
-                        self.bus,
-                        plugins=self.plugins,
-                        url=root_snapshot["url"],
-                        snapshot=AbxSnapshot(
-                            id=root_snapshot["id"],
-                            url=root_snapshot["url"],
-                            depth=int(root_snapshot["depth"]),
-                            crawl_id=str(self.crawl.id),
-                        ),
-                        output_dir=Path(root_snapshot["output_dir"]),
-                        install_enabled=False,
-                        crawl_setup_enabled=False,
-                        crawl_start_enabled=False,
-                        snapshot_cleanup_enabled=False,
-                        crawl_cleanup_enabled=False,
-                        persist_derived=False,
-                        auto_install=True,
-                        emit_jsonl=False,
-                    )
                 await _emit_machine_config(
                     self.bus,
                     config={
@@ -200,7 +198,7 @@ class CrawlRunner:
                     await self.run_crawl_cleanup(root_snapshot_id)
         finally:
             await heartbeat.stop()
-            await self.bus.stop()
+            await self.bus.wait_until_idle()
             if self._live_stream is not None:
                 try:
                     self._live_stream.close()
@@ -249,6 +247,7 @@ class CrawlRunner:
         self.derived_config = dict(Machine.current().config)
         self.crawl_output_dir = str(self.crawl.output_dir)
         self.base_config["ABX_RUNTIME"] = "archivebox"
+        self.base_config["CHROME_KEEPALIVE"] = True
         if self.selected_plugins is None:
             raw_plugins = str(self.base_config.get("PLUGINS") or "").strip()
             if raw_plugins:
@@ -268,6 +267,11 @@ class CrawlRunner:
             )
         if self.initial_snapshot_ids:
             return [str(snapshot_id) for snapshot_id in self.initial_snapshot_ids]
+        pending_snapshots = list(
+            self.crawl.snapshot_set.exclude(status="sealed").order_by("depth", "created_at"),
+        )
+        if pending_snapshots:
+            return [str(snapshot.id) for snapshot in pending_snapshots]
         created = self.crawl.create_snapshots_from_urls()
         snapshots = created or list(self.crawl.snapshot_set.filter(depth=0).order_by("created_at"))
         return [str(snapshot.id) for snapshot in snapshots]
@@ -425,49 +429,109 @@ class CrawlRunner:
 
     async def run_crawl_setup(self, snapshot_id: str) -> None:
         snapshot = await sync_to_async(self.load_snapshot_payload, thread_sensitive=True)(snapshot_id)
-        await download(
+        config = _normalize_runtime_config(snapshot["config"])
+        derived_config = _normalize_runtime_config(self.derived_config)
+        output_dir = Path(self.crawl_output_dir)
+        plugins = self.runtime_plugins()
+        abx_snapshot = AbxSnapshot(
+            id=snapshot["id"],
             url=snapshot["url"],
-            plugins=self.plugins,
-            output_dir=Path(snapshot["output_dir"]),
-            selected_plugins=self.selected_plugins,
-            config_overrides=_normalize_runtime_config(snapshot["config"]),
-            derived_config_overrides=_normalize_runtime_config(self.derived_config),
-            bus=self.bus,
-            emit_jsonl=False,
-            install_enabled=True,
+            depth=int(snapshot["depth"]),
+            crawl_id=str(self.crawl.id),
+        )
+        setup_hooks = [(plugin, hook) for plugin in plugins.values() for hook in plugin.filter_hooks("CrawlSetup")]
+        crawl_setup_phase_timeout = compute_phase_timeout(setup_hooks, config)
+        install_phase_timeout = compute_install_phase_timeout(get_install_plugins(plugins), config)
+        await _emit_machine_config(self.bus, config=config, derived_config=derived_config)
+        setup_abx_services(
+            self.bus,
+            plugins=plugins,
+            url=snapshot["url"],
+            snapshot=abx_snapshot,
+            output_dir=output_dir,
+            install_enabled=False,
             crawl_setup_enabled=True,
             crawl_start_enabled=False,
             snapshot_cleanup_enabled=False,
             crawl_cleanup_enabled=False,
+            crawl_setup_phase_timeout=crawl_setup_phase_timeout,
+            snapshot_phase_timeout=0.0,
+            snapshot_cleanup_phase_timeout=0.0,
+            crawl_cleanup_phase_timeout=crawl_setup_phase_timeout,
+            persist_derived=False,
+            auto_install=True,
+            emit_jsonl=False,
             MachineService=None,
             BinaryService=None,
             ProcessService=None,
             ArchiveResultService=None,
             TagService=None,
+            SnapshotService=None,
         )
+        await self.bus.emit(
+            InstallEvent(
+                url=snapshot["url"],
+                snapshot_id=snapshot["id"],
+                output_dir=str(output_dir),
+                event_timeout=install_phase_timeout,
+                event_handler_slow_timeout=slow_warning_timeout(install_phase_timeout),
+            ),
+        ).now()
+        crawl_event = CrawlEvent(
+            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),
+        )
+        self.root_crawl_event_id = crawl_event.event_id
+        await self.bus.emit(crawl_event).now()
+        for plugin, hook in setup_hooks:
+            if hook.is_background:
+                continue
+            process_event = await self.bus.find(
+                ProcessEvent,
+                past=True,
+                future=crawl_setup_phase_timeout,
+                where=lambda candidate, plugin_name=plugin.name, hook_name=hook.name: (
+                    self.bus.event_is_child_of(candidate, crawl_event)
+                    and candidate.plugin_name == plugin_name
+                    and candidate.hook_name == hook_name
+                    and candidate.output_dir == str(output_dir / plugin_name)
+                ),
+            )
+            if process_event is None:
+                raise RuntimeError(f"Crawl setup hook {plugin.name}:{hook.name} did not start")
+            completed_process = await self.bus.find(
+                ProcessCompletedEvent,
+                child_of=process_event,
+                past=True,
+                future=crawl_setup_phase_timeout,
+            )
+            if completed_process is None:
+                raise RuntimeError(f"Crawl setup hook {plugin.name}:{hook.name} did not complete")
+            if completed_process.status == "failed":
+                raise RuntimeError(f"Crawl setup hook {plugin.name}:{hook.name} failed")
 
     async def run_crawl_cleanup(self, snapshot_id: str) -> None:
         snapshot = await sync_to_async(self.load_snapshot_payload, thread_sensitive=True)(snapshot_id)
-        await download(
-            bus=self.bus,
-            url=snapshot["url"],
-            output_dir=Path(snapshot["output_dir"]),
-            plugins=self.plugins,
-            selected_plugins=self.selected_plugins,
-            config_overrides=_normalize_runtime_config(snapshot["config"]),
-            derived_config_overrides=_normalize_runtime_config(self.derived_config),
-            emit_jsonl=False,
-            install_enabled=False,
-            crawl_setup_enabled=False,
-            crawl_start_enabled=False,
-            snapshot_cleanup_enabled=False,
-            crawl_cleanup_enabled=True,
-            MachineService=None,
-            BinaryService=None,
-            ProcessService=None,
-            ArchiveResultService=None,
-            TagService=None,
-        )
+        if self.root_crawl_event_id is None:
+            return
+        config = _normalize_runtime_config(snapshot["config"])
+        output_dir = Path(self.crawl_output_dir)
+        plugins = self.runtime_plugins()
+        setup_hooks = [(plugin, hook) for plugin in plugins.values() for hook in plugin.filter_hooks("CrawlSetup")]
+        crawl_cleanup_phase_timeout = compute_phase_timeout(setup_hooks, config)
+        await self.bus.emit(
+            CrawlCleanupEvent(
+                url=snapshot["url"],
+                snapshot_id=snapshot["id"],
+                output_dir=str(output_dir),
+                event_parent_id=self.root_crawl_event_id,
+                event_timeout=crawl_cleanup_phase_timeout,
+                event_handler_slow_timeout=slow_warning_timeout(crawl_cleanup_phase_timeout),
+            ),
+        ).now()
 
     async def run_snapshot(self, snapshot_id: str) -> None:
         async with self.snapshot_semaphore:
@@ -478,26 +542,48 @@ class CrawlRunner:
                 await sync_to_async(self.seal_snapshot_due_to_limit, thread_sensitive=True)(snapshot_id)
                 return
             try:
-                await download(
+                config = _normalize_runtime_config(snapshot["config"])
+                derived_config = _normalize_runtime_config(self.derived_config)
+                output_dir = Path(snapshot["output_dir"])
+                plugins = self.runtime_plugins()
+                abx_snapshot = AbxSnapshot(
+                    id=snapshot["id"],
                     url=snapshot["url"],
-                    plugins=self.plugins,
-                    output_dir=Path(snapshot["output_dir"]),
-                    selected_plugins=self.selected_plugins,
-                    config_overrides=_normalize_runtime_config(snapshot["config"]),
-                    derived_config_overrides=_normalize_runtime_config(self.derived_config),
-                    bus=self.bus,
-                    emit_jsonl=False,
-                    install_enabled=False,
-                    crawl_setup_enabled=False,
-                    crawl_start_enabled=True,
-                    snapshot_cleanup_enabled=True,
-                    crawl_cleanup_enabled=False,
-                    MachineService=None,
-                    BinaryService=None,
-                    ProcessService=None,
-                    ArchiveResultService=None,
-                    TagService=None,
+                    depth=int(snapshot["depth"]),
+                    crawl_id=str(self.crawl.id),
                 )
+                snapshot_hooks = [(plugin, hook) for plugin in plugins.values() for hook in plugin.filter_hooks("Snapshot")]
+                snapshot_phase_timeout = compute_phase_timeout(snapshot_hooks, config)
+                await _emit_machine_config(self.bus, config=config, derived_config=derived_config)
+                HookSnapshotService(
+                    self.bus,
+                    url=snapshot["url"],
+                    snapshot=abx_snapshot,
+                    output_dir=output_dir,
+                    plugins=plugins,
+                    snapshot_phase_timeout=snapshot_phase_timeout,
+                    snapshot_cleanup_enabled=True,
+                    snapshot_cleanup_phase_timeout=snapshot_phase_timeout,
+                )
+                crawl_start_event = CrawlStartEvent(
+                    url=snapshot["url"],
+                    snapshot_id=snapshot["id"],
+                    output_dir=str(output_dir),
+                    event_timeout=snapshot_phase_timeout,
+                    event_handler_slow_timeout=slow_warning_timeout(snapshot_phase_timeout),
+                )
+                await self.bus.emit(crawl_start_event).now()
+                await self.bus.emit(
+                    SnapshotEvent(
+                        url=snapshot["url"],
+                        snapshot_id=snapshot["id"],
+                        output_dir=str(output_dir),
+                        depth=int(snapshot["depth"]),
+                        event_parent_id=crawl_start_event.event_id,
+                        event_timeout=snapshot_phase_timeout,
+                        event_handler_slow_timeout=slow_warning_timeout(snapshot_phase_timeout),
+                    ),
+                ).now()
                 await self.enqueue_discovered_snapshots_from_outputs(snapshot)
             finally:
                 current_task = asyncio.current_task()
@@ -547,7 +633,7 @@ async def _run_binary(binary_id: str) -> None:
     config["ABX_RUNTIME"] = "archivebox"
     config = _normalize_runtime_config(config)
     bus = create_bus(name=_bus_name("ArchiveBox_binary", str(binary.id)), total_timeout=1800.0)
-    ProcessService(bus)
+    PersistedProcessService(bus)
     BinaryService(bus)
     TagService(bus)
     ArchiveResultService(bus)
@@ -578,9 +664,9 @@ async def _run_binary(binary_id: str) -> None:
                 binproviders=binary.binproviders,
                 overrides=binary.overrides or None,
             ),
-        )
+        ).now(first_result=True)
     finally:
-        await bus.stop()
+        await bus.wait_until_idle()
 
 
 def run_binary(binary_id: str) -> None:
@@ -598,7 +684,7 @@ async def _run_install(plugin_names: list[str] | None = None) -> None:
     config["ABX_RUNTIME"] = "archivebox"
     config = _normalize_runtime_config(config)
     bus = create_bus(name="ArchiveBox_install", total_timeout=3600.0)
-    ProcessService(bus)
+    PersistedProcessService(bus)
     BinaryService(bus)
     TagService(bus)
     ArchiveResultService(bus)
@@ -674,7 +760,7 @@ async def _run_install(plugin_names: list[str] | None = None) -> None:
             if live_ui is not None:
                 live_ui.print_summary(output_dir=output_dir)
     finally:
-        await bus.stop()
+        await bus.wait_until_idle()
         try:
             if live_stream is not None:
                 live_stream.close()

archivebox/tests/test_archive_result_service.py

@@ -385,7 +385,7 @@ def test_process_started_hydrates_binary_and_iface_from_existing_binary_records(
                 timeout=60,
                 url="https://example.com",
             ),
-        )
+        ).now()
         started = await bus.find(
             ProcessStartedEvent,
             past=True,
@@ -451,7 +451,7 @@ def test_process_started_uses_node_binary_for_js_hooks_without_plugin_binary(mon
                 timeout=60,
                 url="https://example.com",
             ),
-        )
+        ).now()
         started = await bus.find(
             ProcessStartedEvent,
             past=True,
@@ -499,7 +499,11 @@ def test_binary_event_reuses_existing_installed_binary_row(monkeypatch):
         binproviders="provider",
     )
 
-    asyncio.run(service.on_BinaryRequestEvent(event))
+    async def run_event():
+        await service.bus.emit(event).now()
+        await service.bus.wait_until_idle()
+
+    asyncio.run(run_event())
 
     binary.refresh_from_db()
     assert Binary.objects.filter(machine=machine, name="wget").count() == 1

archivebox/tests/test_cli_real_flows.py

@@ -0,0 +1,139 @@
+#!/usr/bin/env python3
+"""Real user-facing archive flows against live URLs."""
+
+import json
+import os
+import shutil
+import sqlite3
+import subprocess
+import sys
+
+import pytest
+
+
+@pytest.mark.timeout(180)
+def test_cli_add_real_urls_with_options_writes_inspectable_outputs(tmp_path, process):
+    os.chdir(tmp_path)
+    assert process.returncode == 0, process.stderr
+
+    urls = [
+        "https://example.com",
+        "https://pirate.github.io/stress-tests/challenge.html",
+    ]
+    env = os.environ.copy()
+    env.update(
+        {
+            "USE_COLOR": "false",
+            "SHOW_PROGRESS": "false",
+            "TIMEOUT": "60",
+            "SAVE_WGET": "true",
+            "SAVE_HEADERS": "true",
+            "SAVE_TITLE": "true",
+            "SAVE_READABILITY": "false",
+            "SAVE_SINGLEFILE": "false",
+            "SAVE_MERCURY": "false",
+            "SAVE_SCREENSHOT": "false",
+            "SAVE_PDF": "false",
+            "SAVE_DOM": "false",
+            "SAVE_ARCHIVEDOTORG": "false",
+            "SAVE_GIT": "false",
+            "SAVE_YTDLP": "false",
+            "SAVE_FAVICON": "false",
+            "CHROME_HEADLESS": "true",
+            "CHROME_SANDBOX": "false",
+        },
+    )
+    node_binary = shutil.which("node")
+    chrome_binary = shutil.which("chromium") or shutil.which("google-chrome") or shutil.which("chrome")
+    if node_binary:
+        env["NODE_BINARY"] = node_binary
+    if chrome_binary:
+        env["CHROME_BINARY"] = chrome_binary
+
+    result = subprocess.run(
+        [
+            sys.executable,
+            "-m",
+            "archivebox",
+            "add",
+            "--depth=0",
+            "--max-urls=2",
+            "--max-size=10mb",
+            "--tag=real-flow,challenge",
+            "--parser=url_list",
+            "--plugins=wget,headers,title",
+            *urls,
+        ],
+        cwd=tmp_path,
+        capture_output=True,
+        text=True,
+        env=env,
+        timeout=180,
+    )
+    assert result.returncode == 0, result.stderr or result.stdout
+
+    list_result = subprocess.run(
+        [sys.executable, "-m", "archivebox", "list", "--tag=real-flow"],
+        cwd=tmp_path,
+        capture_output=True,
+        text=True,
+        env=env,
+        timeout=60,
+    )
+    assert list_result.returncode == 0, list_result.stderr or list_result.stdout
+    listed = [json.loads(line) for line in list_result.stdout.splitlines() if line.strip()]
+    assert {item["url"] for item in listed} >= set(urls)
+
+    conn = sqlite3.connect(tmp_path / "index.sqlite3")
+    try:
+        crawl = conn.execute(
+            "SELECT max_depth, max_urls, max_size, tags_str, config FROM crawls_crawl ORDER BY created_at DESC LIMIT 1",
+        ).fetchone()
+        snapshots = conn.execute(
+            "SELECT id, url, depth, status, title FROM core_snapshot ORDER BY url",
+        ).fetchall()
+        archive_results = conn.execute(
+            "SELECT s.url, ar.plugin, ar.status, ar.output_files, ar.output_size "
+            "FROM core_archiveresult ar "
+            "JOIN core_snapshot s ON s.id = ar.snapshot_id "
+            "ORDER BY s.url, ar.plugin",
+        ).fetchall()
+        processes = conn.execute(
+            "SELECT process_type, status, exit_code, pwd, cmd FROM machine_process WHERE process_type = 'hook'",
+        ).fetchall()
+    finally:
+        conn.close()
+
+    assert crawl is not None
+    assert crawl[0] == 0
+    assert crawl[1] == 2
+    assert crawl[2] == 10 * 1024 * 1024
+    assert crawl[3] == "real-flow,challenge"
+    assert "wget,headers,title" in crawl[4]
+
+    snapshot_urls = {url for _id, url, _depth, _status, _title in snapshots}
+    assert snapshot_urls >= set(urls)
+    assert all(depth == 0 for _id, _url, depth, _status, _title in snapshots)
+
+    by_url_plugin = {(url, plugin): status for url, plugin, status, _files, _size in archive_results}
+    for url in urls:
+        assert by_url_plugin[(url, "wget")] == "succeeded"
+        assert by_url_plugin[(url, "headers")] == "succeeded"
+        assert by_url_plugin[(url, "title")] == "succeeded"
+    assert len([status for _url, _plugin, status, _files, _size in archive_results if status == "failed"]) <= 2
+
+    snapshot_root = tmp_path / "users/system/snapshots"
+    html_outputs = [path for path in snapshot_root.rglob("wget/**/*.html") if path.is_file()]
+    header_outputs = [path for path in snapshot_root.rglob("headers/**/headers.json") if path.is_file()]
+    index_outputs = [path for path in snapshot_root.rglob("index.jsonl") if path.is_file()]
+    assert html_outputs
+    assert header_outputs
+    assert len(index_outputs) >= len(urls)
+
+    combined_html = "\n".join(path.read_text(errors="ignore") for path in html_outputs)
+    assert "Example Domain" in combined_html
+    assert "Browser-use Challenge for AI Browser Drivers" in combined_html
+
+    assert processes
+    assert any("wget" in (pwd or "") or "wget" in (cmd or "") for _type, _status, _exit, pwd, cmd in processes)
+    assert any("headers" in (pwd or "") or "headers" in (cmd or "") for _type, _status, _exit, pwd, cmd in processes)

archivebox/tests/test_process_service.py

@@ -2,3 +2,86 @@ import pytest
 
 
 pytestmark = pytest.mark.django_db
+
+
+@pytest.mark.django_db(transaction=True)
+def test_process_completed_persists_with_uncached_network_interface(monkeypatch, tmp_path):
+    import asyncio
+    from uuid import uuid4
+
+    from abx_dl.events import ProcessCompletedEvent
+    from abx_dl.orchestrator import create_bus
+    from archivebox.machine.models import Machine, NetworkInterface, Process
+    from archivebox.services.process_service import ProcessService
+
+    machine = Machine.objects.create(
+        guid=f"test-guid-{uuid4()}",
+        hostname="test-host",
+        hw_in_docker=False,
+        hw_in_vm=False,
+        hw_manufacturer="Test",
+        hw_product="Test Product",
+        hw_uuid=f"test-hw-{uuid4()}",
+        os_arch="arm64",
+        os_family="darwin",
+        os_platform="macOS",
+        os_release="14.0",
+        os_kernel="Darwin",
+        stats={},
+        config={},
+    )
+    iface = NetworkInterface.objects.create(
+        machine=machine,
+        mac_address="00:11:22:33:44:55",
+        ip_public="203.0.113.10",
+        ip_local="10.0.0.10",
+        dns_server="1.1.1.1",
+        hostname="test-host",
+        iface="en0",
+        isp="Test ISP",
+        city="Test City",
+        region="Test Region",
+        country="Test Country",
+    )
+    monkeypatch.setattr(
+        NetworkInterface,
+        "current",
+        classmethod(lambda cls, refresh=False: NetworkInterface.objects.get(id=iface.id)),
+    )
+
+    output_dir = tmp_path / "headers"
+    output_dir.mkdir()
+    bus = create_bus(name="test_process_completed_uncached_iface")
+    ProcessService(bus)
+
+    async def run_event() -> None:
+        event = bus.emit(
+            ProcessCompletedEvent(
+                plugin_name="headers",
+                hook_name="on_Snapshot__27_headers.daemon.bg",
+                hook_path="/bin/echo",
+                hook_args=["--url=https://example.com"],
+                is_background=True,
+                output_dir=str(output_dir),
+                env={},
+                timeout=60,
+                pid=123,
+                stdout="",
+                stderr="",
+                exit_code=0,
+                status="succeeded",
+                output_files=[],
+                start_ts="2026-05-13T07:22:00+00:00",
+                end_ts="2026-05-13T07:22:01+00:00",
+            ),
+        )
+        await event.now()
+        await event.event_results_list()
+
+    asyncio.run(run_event())
+
+    process = Process.objects.get(pwd=str(output_dir), cmd=["/bin/echo", "--url=https://example.com"])
+    assert process.machine_id == machine.id
+    assert process.iface_id == iface.id
+    assert process.process_type == Process.TypeChoices.HOOK
+    assert process.status == Process.StatusChoices.EXITED

archivebox/tests/test_recursive_crawl.py

@@ -314,6 +314,140 @@ def test_recursive_crawl_respects_depth_limit(tmp_path, process, disable_extract
     assert max_depth_found <= 1, f"Max depth should not exceed 1, got {max_depth_found}. Depth distribution: {depth_counts}"
 
 
+def test_recursive_crawl_depth_two_writes_real_outputs_and_process_records(tmp_path, process, recursive_test_site):
+    """Run a real depth=2 crawl and verify DB, output files, and process side effects."""
+    os.chdir(tmp_path)
+
+    env = os.environ.copy()
+    env.update(
+        {
+            "URL_ALLOWLIST": r"127\.0\.0\.1[:/].*",
+            "SAVE_WGET": "true",
+            "SAVE_READABILITY": "false",
+            "SAVE_SINGLEFILE": "false",
+            "SAVE_MERCURY": "false",
+            "SAVE_SCREENSHOT": "false",
+            "SAVE_PDF": "false",
+            "SAVE_HEADERS": "false",
+            "SAVE_ARCHIVEDOTORG": "false",
+            "SAVE_GIT": "false",
+            "SAVE_YTDLP": "false",
+            "SAVE_TITLE": "false",
+            "SAVE_FAVICON": "false",
+            "USE_CHROME": "false",
+            "USE_COLOR": "false",
+            "SHOW_PROGRESS": "false",
+        },
+    )
+
+    stdout, stderr = run_add_until(
+        ["archivebox", "add", "--depth=2", "--plugins=wget,parse_html_urls", recursive_test_site["root_url"]],
+        env=env,
+        timeout=180,
+        condition=lambda c: (
+            c.execute("SELECT COUNT(*) FROM core_snapshot WHERE depth = 0").fetchone()[0] >= 1
+            and c.execute("SELECT COUNT(*) FROM core_snapshot WHERE depth = 1").fetchone()[0] >= len(recursive_test_site["child_urls"])
+            and c.execute("SELECT COUNT(*) FROM core_snapshot WHERE depth = 2").fetchone()[0] >= len(recursive_test_site["deep_urls"])
+            and c.execute(
+                "SELECT COUNT(*) "
+                "FROM core_archiveresult ar "
+                "JOIN core_snapshot s ON s.id = ar.snapshot_id "
+                "WHERE ar.plugin LIKE 'parse_%_urls' "
+                "AND s.depth IN (0, 1) "
+                "AND ar.status IN ('started', 'succeeded', 'failed', 'skipped', 'noresults')",
+            ).fetchone()[0]
+            >= 2
+        ),
+    )
+
+    if stderr:
+        print(f"\n=== STDERR ===\n{stderr}\n=== END STDERR ===\n")
+    if stdout:
+        print(f"\n=== STDOUT (last 2000 chars) ===\n{stdout[-2000:]}\n=== END STDOUT ===\n")
+
+    conn = sqlite3.connect("index.sqlite3")
+    c = conn.cursor()
+
+    depth_counts = dict(c.execute("SELECT depth, COUNT(*) FROM core_snapshot GROUP BY depth ORDER BY depth").fetchall())
+    crawl = c.execute("SELECT id, max_depth FROM crawls_crawl ORDER BY created_at DESC LIMIT 1").fetchone()
+    root_snapshot = c.execute(
+        "SELECT id, url, depth, parent_snapshot_id FROM core_snapshot WHERE depth = 0 ORDER BY created_at LIMIT 1",
+    ).fetchone()
+    child_rows = c.execute(
+        "SELECT id, url, parent_snapshot_id FROM core_snapshot WHERE depth = 1",
+    ).fetchall()
+    deep_rows = c.execute(
+        "SELECT id, url, parent_snapshot_id FROM core_snapshot WHERE depth = 2",
+    ).fetchall()
+    parser_results = c.execute(
+        "SELECT s.url, s.depth, ar.plugin, ar.status, ar.output_files, ar.output_size "
+        "FROM core_archiveresult ar "
+        "JOIN core_snapshot s ON s.id = ar.snapshot_id "
+        "WHERE ar.plugin LIKE 'parse_%_urls' "
+        "ORDER BY s.depth, s.url",
+    ).fetchall()
+    wget_results = c.execute(
+        "SELECT s.url, s.depth, ar.status, ar.output_files, ar.output_size "
+        "FROM core_archiveresult ar "
+        "JOIN core_snapshot s ON s.id = ar.snapshot_id "
+        "WHERE ar.plugin = 'wget' "
+        "ORDER BY s.depth, s.url",
+    ).fetchall()
+    process_rows = c.execute(
+        "SELECT process_type, worker_type, status, exit_code, pwd, cmd "
+        "FROM machine_process "
+        "WHERE process_type = 'hook' "
+        "ORDER BY created_at",
+    ).fetchall()
+    conn.close()
+
+    assert crawl is not None
+    assert crawl[1] == 2
+    assert root_snapshot is not None
+    assert root_snapshot[2] == 0
+    assert root_snapshot[3] is None
+    assert depth_counts.get(0, 0) >= 1
+    assert depth_counts.get(1, 0) >= len(recursive_test_site["child_urls"])
+    assert depth_counts.get(2, 0) >= len(recursive_test_site["deep_urls"])
+    assert max(depth_counts) <= 2
+
+    child_urls = {row[1] for row in child_rows}
+    deep_urls = {row[1] for row in deep_rows}
+    child_ids = {row[0] for row in child_rows}
+    assert set(recursive_test_site["child_urls"]).issubset(child_urls)
+    assert set(recursive_test_site["deep_urls"]).issubset(deep_urls)
+    assert all(parent_id == root_snapshot[0] for _id, _url, parent_id in child_rows)
+    assert all(parent_id in child_ids for _id, _url, parent_id in deep_rows)
+
+    parser_statuses = {status for _url, _depth, _plugin, status, _files, _size in parser_results}
+    wget_statuses = {status for _url, _depth, status, _files, _size in wget_results}
+    assert parser_results
+    assert wget_results
+    assert "succeeded" in parser_statuses
+    assert "succeeded" in wget_statuses
+    assert len([row for row in parser_results if row[3] == "failed"]) <= 2
+    assert len([row for row in wget_results if row[2] == "failed"]) <= 2
+
+    urls_jsonl_files = list(Path("users/system/snapshots").rglob("parse_html_urls/**/urls.jsonl"))
+    assert urls_jsonl_files, "parse_html_urls should write urls.jsonl files"
+    parsed_urls = set()
+    for path in urls_jsonl_files:
+        for line in path.read_text().splitlines():
+            if line.strip():
+                parsed_urls.add(json.loads(line)["url"])
+    assert set(recursive_test_site["child_urls"]).issubset(parsed_urls)
+    assert set(recursive_test_site["deep_urls"]).issubset(parsed_urls)
+
+    snapshot_dirs = [path.parent for path in Path("users/system/snapshots").rglob("index.jsonl")]
+    assert snapshot_dirs
+    for snapshot_dir in snapshot_dirs:
+        assert (snapshot_dir / "index.jsonl").exists()
+
+    assert process_rows
+    assert any("parse_html_urls" in (pwd or "") or "parse_html_urls" in (cmd or "") for *_rest, pwd, cmd in process_rows)
+    assert any("wget" in (pwd or "") or "wget" in (cmd or "") for *_rest, pwd, cmd in process_rows)
+
+
 def test_crawl_snapshot_has_parent_snapshot_field(tmp_path, process, disable_extractors_dict):
     """Test that Snapshot model has parent_snapshot field."""
     os.chdir(tmp_path)

archivebox/tests/test_runner.py

@@ -16,6 +16,7 @@ class _DummyBus:
     def __init__(self, name: str):
         self.name = name
         self.registrations = []
+        self.emitted = []
 
     def on(self, event_pattern, handler):
         registration = SimpleNamespace(event_pattern=event_pattern, handler=handler)
@@ -25,9 +26,21 @@ class _DummyBus:
     def off(self, event_pattern, registration):
         self.registrations = [existing for existing in self.registrations if existing is not registration]
 
+    def emit(self, event):
+        self.emitted.append(event)
+
+        class _Pending:
+            async def now(self, *args, **kwargs):
+                return event
+
+        return _Pending()
+
     async def stop(self):
         return None
 
+    async def wait_until_idle(self):
+        return None
+
 
 class _DummyService:
     def __init__(self, *args, **kwargs):
@@ -64,7 +77,8 @@ def test_run_snapshot_reuses_crawl_bus_for_all_snapshots(monkeypatch):
 
     monkeypatch.setattr(runner_module, "create_bus", fake_create_bus)
     monkeypatch.setattr(runner_module, "discover_plugins", lambda: {})
-    monkeypatch.setattr(runner_module, "ProcessService", _DummyService)
+    monkeypatch.setattr(runner_module, "HookProcessService", _DummyService)
+    monkeypatch.setattr(runner_module, "PersistedProcessService", _DummyService)
     monkeypatch.setattr(runner_module, "BinaryService", _DummyService)
     monkeypatch.setattr(runner_module, "TagService", _DummyService)
     monkeypatch.setattr(runner_module, "CrawlService", _DummyService)
@@ -73,23 +87,6 @@ def test_run_snapshot_reuses_crawl_bus_for_all_snapshots(monkeypatch):
     monkeypatch.setattr(runner_module, "_emit_machine_config", lambda *args, **kwargs: asyncio.sleep(0))
     monkeypatch.setattr(runner_module, "setup_abx_services", lambda *args, **kwargs: None)
 
-    download_calls = []
-
-    async def fake_download(*, url, bus, config_overrides, **kwargs):
-        extra_context = json.loads(config_overrides["EXTRA_CONTEXT"])
-        download_calls.append(
-            {
-                "url": url,
-                "bus": bus,
-                "snapshot_id": extra_context["snapshot_id"],
-                "source_url": url,
-            },
-        )
-        await asyncio.sleep(0)
-        return []
-
-    monkeypatch.setattr(runner_module, "download", fake_download)
-
     crawl_runner = runner_module.CrawlRunner(crawl)
     snapshot_data = {
         str(snapshot_a.id): {
@@ -129,10 +126,13 @@ def test_run_snapshot_reuses_crawl_bus_for_all_snapshots(monkeypatch):
 
     asyncio.run(run_both())
 
-    assert len(download_calls) == 2
-    assert {call["snapshot_id"] for call in download_calls} == {str(snapshot_a.id), str(snapshot_b.id)}
-    assert {call["source_url"] for call in download_calls} == {snapshot_a.url, snapshot_b.url}
-    assert len({id(call["bus"]) for call in download_calls}) == 1
+    from abx_dl.events import SnapshotEvent
+
+    snapshot_events = [event for event in crawl_runner.bus.emitted if isinstance(event, SnapshotEvent)]
+    assert len(snapshot_events) == 2
+    assert {event.snapshot_id for event in snapshot_events} == {str(snapshot_a.id), str(snapshot_b.id)}
+    assert {event.url for event in snapshot_events} == {snapshot_a.url, snapshot_b.url}
+    assert crawl_runner.bus is created_buses[0]
     assert len(created_buses) == 1
 
 
@@ -217,7 +217,8 @@ def test_runner_prepare_refreshes_network_interface_and_attaches_current_process
 
     monkeypatch.setattr(runner_module, "discover_plugins", lambda: {})
     monkeypatch.setattr(runner_module, "create_bus", lambda **kwargs: _DummyBus(kwargs["name"]))
-    monkeypatch.setattr(runner_module, "ProcessService", _DummyService)
+    monkeypatch.setattr(runner_module, "HookProcessService", _DummyService)
+    monkeypatch.setattr(runner_module, "PersistedProcessService", _DummyService)
     monkeypatch.setattr(runner_module, "BinaryService", _DummyService)
     monkeypatch.setattr(runner_module, "TagService", _DummyService)
     monkeypatch.setattr(runner_module, "CrawlService", _DummyService)
@@ -357,18 +358,13 @@ def test_run_snapshot_skips_descendant_when_max_size_already_reached(monkeypatch
 
     monkeypatch.setattr(runner_module, "discover_plugins", lambda: {})
     monkeypatch.setattr(runner_module, "create_bus", lambda **kwargs: _DummyBus(kwargs["name"]))
-    monkeypatch.setattr(runner_module, "ProcessService", _DummyService)
+    monkeypatch.setattr(runner_module, "HookProcessService", _DummyService)
+    monkeypatch.setattr(runner_module, "PersistedProcessService", _DummyService)
     monkeypatch.setattr(runner_module, "BinaryService", _DummyService)
     monkeypatch.setattr(runner_module, "TagService", _DummyService)
     monkeypatch.setattr(runner_module, "CrawlService", _DummyService)
     monkeypatch.setattr(runner_module, "SnapshotService", _DummyService)
     monkeypatch.setattr(runner_module, "ArchiveResultService", _DummyService)
-    monkeypatch.setattr(
-        runner_module,
-        "download",
-        lambda *args, **kwargs: (_ for _ in ()).throw(AssertionError("snapshot download should have been skipped")),
-    )
-
     crawl_runner = runner_module.CrawlRunner(crawl)
     state_dir = tmp_path / ".abx-dl"
     state_dir.mkdir(parents=True, exist_ok=True)
@@ -460,7 +456,7 @@ def test_seal_snapshot_cancels_queued_descendants_after_max_size():
 
         asyncio.run(emit_event())
     finally:
-        asyncio.run(bus.stop())
+        asyncio.run(bus.wait_until_idle())
 
     root.refresh_from_db()
     child.refresh_from_db()
@@ -562,7 +558,8 @@ def test_crawl_runner_calls_load_and_finalize_run_state(monkeypatch):
 
     monkeypatch.setattr(runner_module, "create_bus", lambda *args, **kwargs: _DummyBus("runner"))
     monkeypatch.setattr(runner_module, "discover_plugins", lambda: {})
-    monkeypatch.setattr(runner_module, "ProcessService", _DummyService)
+    monkeypatch.setattr(runner_module, "HookProcessService", _DummyService)
+    monkeypatch.setattr(runner_module, "PersistedProcessService", _DummyService)
     monkeypatch.setattr(runner_module, "BinaryService", _DummyService)
     monkeypatch.setattr(runner_module, "TagService", _DummyService)
     monkeypatch.setattr(runner_module, "CrawlService", _DummyService)
@@ -702,27 +699,20 @@ def test_crawl_runner_calls_crawl_cleanup_after_snapshot_phase(monkeypatch):
 
 def test_abx_process_service_background_process_finishes_after_process_exit(monkeypatch, tmp_path):
     from abx_dl.events import ProcessCompletedEvent, ProcessEvent
+    from abx_dl.orchestrator import create_bus
     from abx_dl.services.process_service import ProcessService
 
-    service = object.__new__(ProcessService)
-    service.emit_jsonl = False
-    service.interactive_tty = False
-    service.pause_requested = asyncio.Event()
-    service.abort_requested = False
+    bus = create_bus(name="test_abx_process_service_background_process_finishes_after_process_exit")
+    service = ProcessService(bus, emit_jsonl=False, interactive_tty=False)
     emitted_events = []
 
-    class FakeBus:
-        async def emit(self, event):
-            emitted_events.append(event)
-            return event
+    async def collect_completed(event):
+        emitted_events.append(event)
 
-    service.bus = FakeBus()
+    bus.on(ProcessCompletedEvent, collect_completed)
 
     async def fake_stream_stdout(**kwargs):
-        try:
-            await asyncio.Event().wait()
-        except asyncio.CancelledError:
-            return ["daemon output\n"]
+        return ["daemon output\n"]
 
     monkeypatch.setattr(service, "_stream_stdout", fake_stream_stdout)
 
@@ -748,10 +738,8 @@ def test_abx_process_service_background_process_finishes_after_process_exit(monk
             process_type="hook",
             worker_type="hook",
         )
-        await asyncio.wait_for(
-            service.on_ProcessEvent(event),
-            timeout=0.5,
-        )
+        await asyncio.wait_for(bus.emit(event).now(), timeout=0.5)
+        await bus.wait_until_idle()
 
     asyncio.run(run_test())
 

archivebox/tests/test_schedule_e2e.py

@@ -2,6 +2,7 @@
 """End-to-end tests for scheduling across CLI, server, API, and web UI."""
 
 import os
+import re
 import socket
 import sqlite3
 import subprocess
@@ -229,6 +230,14 @@ def get_counts(cwd: Path, scheduled_url: str, one_shot_url: str) -> tuple[int, i
         conn.close()
 
 
+def get_depth_counts(cwd: Path) -> dict[int, int]:
+    conn = sqlite3.connect(cwd / "index.sqlite3")
+    try:
+        return dict(conn.execute("SELECT depth, COUNT(*) FROM core_snapshot GROUP BY depth").fetchall())
+    finally:
+        conn.close()
+
+
 def create_admin_and_token(cwd: Path) -> str:
     script = textwrap.dedent(
         """
@@ -421,3 +430,434 @@ def test_schedule_web_ui_post_works_over_running_server(tmp_path, recursive_test
         assert row == ("daily", recursive_test_site["root_url"], "web-ui")
     finally:
         stop_server(tmp_path)
+
+
+@pytest.mark.timeout(240)
+def test_web_ui_add_depth_two_crawls_and_renders_real_outputs_over_running_server(tmp_path, recursive_test_site):
+    os.chdir(tmp_path)
+    init_archive(tmp_path)
+
+    port = get_free_port()
+    env = build_test_env(
+        port,
+        PLUGINS="wget,parse_html_urls",
+        PUBLIC_INDEX="True",
+        PUBLIC_ADD_VIEW="True",
+    )
+    create_admin_and_token(tmp_path)
+
+    try:
+        start_server(tmp_path, env=env, port=port)
+        add_page = wait_for_http(port, host=f"web.archivebox.localhost:{port}", path="/add/")
+        assert add_page.status_code == 200
+        assert 'name="depth"' in add_page.text
+        assert 'name="url"' in add_page.text
+
+        response = requests.post(
+            f"http://127.0.0.1:{port}/add/",
+            headers={"Host": f"web.archivebox.localhost:{port}"},
+            data={
+                "url": recursive_test_site["root_url"],
+                "depth": "2",
+                "max_urls": "20",
+                "max_size": "0",
+                "archiving_plugins": ["wget"],
+                "parsing_plugins": ["parse_html_urls"],
+                "tag": "web-depth-two",
+                "url_filters_allowlist": r"127\.0\.0\.1[:/].*",
+                "url_filters_denylist": "",
+                "schedule": "",
+                "notes": "created from running-server web ui",
+                "persona": "Default",
+                "index_only": "",
+                "config": "{}",
+            },
+            timeout=10,
+            allow_redirects=False,
+        )
+        assert response.status_code in (302, 303), response.text
+
+        deadline = time.time() + 180
+        while time.time() < deadline:
+            depth_counts = get_depth_counts(tmp_path)
+            if (
+                depth_counts.get(0, 0) >= 1
+                and depth_counts.get(1, 0) >= len(recursive_test_site["child_urls"])
+                and depth_counts.get(2, 0) >= len(recursive_test_site["deep_urls"])
+            ):
+                break
+            time.sleep(2)
+        else:
+            raise AssertionError(f"timed out waiting for depth=2 crawl, got depth counts {get_depth_counts(tmp_path)}")
+
+        conn = sqlite3.connect(tmp_path / "index.sqlite3")
+        try:
+            depth_counts = dict(conn.execute("SELECT depth, COUNT(*) FROM core_snapshot GROUP BY depth").fetchall())
+            crawl = conn.execute(
+                "SELECT max_depth, max_urls, tags_str, notes FROM crawls_crawl ORDER BY created_at DESC LIMIT 1",
+            ).fetchone()
+            snapshot_rows = conn.execute(
+                "SELECT url, depth, status, parent_snapshot_id FROM core_snapshot ORDER BY depth, url",
+            ).fetchall()
+            archive_results = conn.execute(
+                "SELECT plugin, status, output_files, output_size FROM core_archiveresult ORDER BY plugin, status",
+            ).fetchall()
+        finally:
+            conn.close()
+
+        assert crawl == (2, 20, "web-depth-two", "created from running-server web ui")
+        assert depth_counts.get(0, 0) >= 1
+        assert depth_counts.get(1, 0) >= len(recursive_test_site["child_urls"])
+        assert depth_counts.get(2, 0) >= len(recursive_test_site["deep_urls"])
+        assert max(depth_counts) <= 2
+        assert set(recursive_test_site["child_urls"]).issubset({url for url, depth, _status, _parent in snapshot_rows if depth == 1})
+        assert set(recursive_test_site["deep_urls"]).issubset({url for url, depth, _status, _parent in snapshot_rows if depth == 2})
+
+        result_statuses = [(plugin, status) for plugin, status, _files, _size in archive_results]
+        assert ("wget", "succeeded") in result_statuses
+        assert any(plugin.endswith("parse_html_urls") and status == "succeeded" for plugin, status in result_statuses)
+        assert len([status for _plugin, status, _files, _size in archive_results if status == "failed"]) <= 2
+        assert list((tmp_path / "users/system/snapshots").rglob("parse_html_urls/**/urls.jsonl"))
+        assert list((tmp_path / "users/system/snapshots").rglob("wget/**/*.html"))
+
+        progress = requests.get(
+            f"http://127.0.0.1:{port}/admin/live-progress/",
+            headers={"Host": f"admin.archivebox.localhost:{port}"},
+            timeout=10,
+        )
+        assert progress.status_code == 200
+        assert "active_crawls" in progress.json()
+
+        index_page = requests.get(
+            f"http://127.0.0.1:{port}/",
+            headers={"Host": f"web.archivebox.localhost:{port}"},
+            timeout=10,
+        )
+        assert index_page.status_code == 200
+        assert recursive_test_site["root_url"] in index_page.text
+
+        session = requests.Session()
+        login_page = session.get(
+            f"http://127.0.0.1:{port}/admin/login/",
+            headers={"Host": f"admin.archivebox.localhost:{port}"},
+            timeout=10,
+        )
+        assert login_page.status_code == 200
+        csrf_match = re.search(r'name="csrfmiddlewaretoken" value="([^"]+)"', login_page.text)
+        assert csrf_match, login_page.text[:500]
+        login_response = session.post(
+            f"http://127.0.0.1:{port}/admin/login/",
+            headers={"Host": f"admin.archivebox.localhost:{port}", "Referer": f"http://admin.archivebox.localhost:{port}/admin/login/"},
+            data={
+                "username": "apitestadmin",
+                "password": "testpass123",
+                "csrfmiddlewaretoken": csrf_match.group(1),
+                "next": "/admin/",
+            },
+            timeout=10,
+            allow_redirects=False,
+        )
+        assert login_response.status_code in (302, 303), login_response.text
+        snapshot_admin = session.get(
+            f"http://127.0.0.1:{port}/admin/core/snapshot/",
+            headers={"Host": f"admin.archivebox.localhost:{port}"},
+            timeout=10,
+        )
+        assert snapshot_admin.status_code == 200
+        assert recursive_test_site["root_url"] in snapshot_admin.text
+    finally:
+        stop_server(tmp_path)
+
+
+@pytest.mark.timeout(180)
+def test_core_rest_api_crud_uses_token_auth_and_persists_side_effects_over_running_server(tmp_path, recursive_test_site):
+    os.chdir(tmp_path)
+    init_archive(tmp_path)
+
+    port = get_free_port()
+    env = build_test_env(port, PUBLIC_INDEX="True")
+    api_token = create_admin_and_token(tmp_path)
+    api_headers = {
+        "Host": f"api.archivebox.localhost:{port}",
+        "X-ArchiveBox-API-Key": api_token,
+    }
+
+    try:
+        start_server(tmp_path, env=env, port=port)
+        docs = wait_for_http(port, host=f"api.archivebox.localhost:{port}", path="/api/v1/docs")
+        assert docs.status_code == 200
+        openapi = wait_for_http(port, host=f"api.archivebox.localhost:{port}", path="/api/v1/openapi.json")
+        assert openapi.status_code == 200
+        paths = openapi.json()["paths"]
+        assert "/api/v1/core/snapshots" in paths
+        assert "/api/v1/crawls/crawls" in paths
+
+        unauth = requests.get(
+            f"http://127.0.0.1:{port}/api/v1/crawls/crawls",
+            headers={"Host": f"api.archivebox.localhost:{port}"},
+            timeout=10,
+        )
+        assert unauth.status_code in (401, 403)
+        bad_auth = requests.get(
+            f"http://127.0.0.1:{port}/api/v1/crawls/crawls",
+            headers={"Host": f"api.archivebox.localhost:{port}", "X-ArchiveBox-API-Key": "bad-token"},
+            timeout=10,
+        )
+        assert bad_auth.status_code in (401, 403)
+
+        crawl_response = requests.post(
+            f"http://127.0.0.1:{port}/api/v1/crawls/crawls",
+            headers=api_headers,
+            json={
+                "urls": [recursive_test_site["root_url"]],
+                "max_depth": 2,
+                "max_urls": 7,
+                "max_size": 0,
+                "tags": ["api-depth-two"],
+                "label": "api crawl",
+                "notes": "created through REST API",
+                "config": {"PLUGINS": "wget,parse_html_urls", "URL_ALLOWLIST": r"127\.0\.0\.1[:/].*"},
+            },
+            timeout=10,
+        )
+        assert crawl_response.status_code == 200, crawl_response.text
+        crawl_payload = crawl_response.json()
+        crawl_id = crawl_payload["id"]
+        assert crawl_payload["max_depth"] == 2
+        assert crawl_payload["max_urls"] == 7
+        assert crawl_payload["tags_str"] == "api-depth-two"
+        assert crawl_payload["config"]["PLUGINS"] == "wget,parse_html_urls"
+
+        snapshot_response = requests.post(
+            f"http://127.0.0.1:{port}/api/v1/core/snapshots",
+            headers=api_headers,
+            json={
+                "url": recursive_test_site["child_urls"][0],
+                "crawl_id": crawl_id,
+                "depth": 1,
+                "title": "API child snapshot",
+                "tags": ["api-child"],
+                "status": "queued",
+            },
+            timeout=10,
+        )
+        assert snapshot_response.status_code == 200, snapshot_response.text
+        snapshot_payload = snapshot_response.json()
+        snapshot_id = snapshot_payload["id"]
+        assert snapshot_payload["url"] == recursive_test_site["child_urls"][0]
+        assert snapshot_payload["tags"] == ["api-child"]
+
+        patch_snapshot = requests.patch(
+            f"http://127.0.0.1:{port}/api/v1/core/snapshot/{snapshot_id}",
+            headers=api_headers,
+            json={"status": "sealed", "tags": ["api-child", "api-patched"]},
+            timeout=10,
+        )
+        assert patch_snapshot.status_code == 200, patch_snapshot.text
+        assert patch_snapshot.json()["status"] == "sealed"
+        assert set(patch_snapshot.json()["tags"]) == {"api-child", "api-patched"}
+
+        tag_create = requests.post(
+            f"http://127.0.0.1:{port}/api/v1/core/tags/create/",
+            headers=api_headers,
+            json={"name": "api-extra"},
+            timeout=10,
+        )
+        assert tag_create.status_code == 200, tag_create.text
+        tag_id = tag_create.json()["tag_id"]
+
+        add_tag = requests.post(
+            f"http://127.0.0.1:{port}/api/v1/core/tags/add-to-snapshot/",
+            headers=api_headers,
+            json={"snapshot_id": snapshot_id, "tag_id": tag_id},
+            timeout=10,
+        )
+        assert add_tag.status_code == 200, add_tag.text
+        remove_tag = requests.post(
+            f"http://127.0.0.1:{port}/api/v1/core/tags/remove-from-snapshot/",
+            headers=api_headers,
+            json={"snapshot_id": snapshot_id, "tag_name": "api-extra"},
+            timeout=10,
+        )
+        assert remove_tag.status_code == 200, remove_tag.text
+
+        crawl_patch = requests.patch(
+            f"http://127.0.0.1:{port}/api/v1/crawls/crawl/{crawl_id}",
+            headers=api_headers,
+            json={"status": "sealed", "tags": ["api-sealed"]},
+            timeout=10,
+        )
+        assert crawl_patch.status_code == 200, crawl_patch.text
+        assert crawl_patch.json()["status"] == "sealed"
+        assert crawl_patch.json()["tags_str"] == "api-sealed"
+
+        snapshots_list = requests.get(
+            f"http://127.0.0.1:{port}/api/v1/core/snapshots?tag=api-patched&with_archiveresults=true",
+            headers=api_headers,
+            timeout=10,
+        )
+        assert snapshots_list.status_code == 200, snapshots_list.text
+        snapshot_items = snapshots_list.json()["items"]
+        assert len(snapshot_items) == 1
+        assert snapshot_items[0]["id"] == snapshot_id
+        assert snapshot_items[0]["archiveresults"] == []
+
+        bearer_response = requests.get(
+            f"http://127.0.0.1:{port}/api/v1/crawls/crawl/{crawl_id}",
+            headers={"Host": f"api.archivebox.localhost:{port}", "Authorization": f"Bearer {api_token}"},
+            timeout=10,
+        )
+        assert bearer_response.status_code == 200, bearer_response.text
+        query_response = requests.get(
+            f"http://127.0.0.1:{port}/api/v1/crawls/crawl/{crawl_id}?api_key={api_token}",
+            headers={"Host": f"api.archivebox.localhost:{port}"},
+            timeout=10,
+        )
+        assert query_response.status_code == 200, query_response.text
+
+        delete_snapshot = requests.delete(
+            f"http://127.0.0.1:{port}/api/v1/core/snapshot/{snapshot_id}",
+            headers=api_headers,
+            timeout=10,
+        )
+        assert delete_snapshot.status_code == 200, delete_snapshot.text
+        assert delete_snapshot.json()["success"] is True
+
+        delete_crawl = requests.delete(
+            f"http://127.0.0.1:{port}/api/v1/crawls/crawl/{crawl_id}",
+            headers=api_headers,
+            timeout=10,
+        )
+        assert delete_crawl.status_code == 200, delete_crawl.text
+        assert delete_crawl.json()["success"] is True
+
+        conn = sqlite3.connect(tmp_path / "index.sqlite3")
+        try:
+            assert conn.execute("SELECT COUNT(*) FROM crawls_crawl WHERE id = ?", (crawl_id,)).fetchone()[0] == 0
+            assert conn.execute("SELECT COUNT(*) FROM core_snapshot WHERE id = ?", (snapshot_id,)).fetchone()[0] == 0
+            assert conn.execute("SELECT COUNT(*) FROM core_tag WHERE name = 'api-extra'").fetchone()[0] == 1
+        finally:
+            conn.close()
+    finally:
+        stop_server(tmp_path)
+
+
+@pytest.mark.timeout(180)
+def test_cli_rest_api_add_search_update_remove_over_running_server(tmp_path, recursive_test_site):
+    os.chdir(tmp_path)
+    init_archive(tmp_path)
+
+    port = get_free_port()
+    env = build_test_env(port, PUBLIC_INDEX="True")
+    api_token = create_admin_and_token(tmp_path)
+    api_headers = {
+        "Host": f"api.archivebox.localhost:{port}",
+        "X-ArchiveBox-API-Key": api_token,
+    }
+
+    try:
+        start_server(tmp_path, env=env, port=port)
+        wait_for_http(port, host=f"api.archivebox.localhost:{port}", path="/api/v1/docs")
+
+        add_response = requests.post(
+            f"http://127.0.0.1:{port}/api/v1/cli/add",
+            headers=api_headers,
+            json={
+                "urls": [recursive_test_site["root_url"]],
+                "tag": "api-cli",
+                "depth": 1,
+                "parser": "url_list",
+                "plugins": "wget",
+                "update": True,
+                "overwrite": False,
+                "index_only": True,
+            },
+            timeout=10,
+        )
+        assert add_response.status_code == 200, add_response.text
+        add_payload = add_response.json()
+        assert add_payload["success"] is True
+        assert add_payload["result_format"] == "json"
+        assert add_payload["result"]["num_snapshots"] == 1
+        crawl_id = add_payload["result"]["crawl_id"]
+        snapshot_id = add_payload["result"]["snapshot_ids"][0]
+
+        search_response = requests.post(
+            f"http://127.0.0.1:{port}/api/v1/cli/search",
+            headers=api_headers,
+            json={
+                "filter_patterns": [recursive_test_site["root_url"]],
+                "filter_type": "exact",
+                "status": "indexed",
+                "sort": "bookmarked_at",
+                "as_json": True,
+                "as_html": False,
+                "as_csv": "",
+                "with_headers": False,
+            },
+            timeout=10,
+        )
+        assert search_response.status_code == 200, search_response.text
+        search_payload = search_response.json()
+        assert search_payload["success"] is True
+        assert search_payload["result_format"] == "json"
+        assert any(item["url"] == recursive_test_site["root_url"] for item in search_payload["result"])
+
+        update_response = requests.post(
+            f"http://127.0.0.1:{port}/api/v1/cli/update",
+            headers=api_headers,
+            json={
+                "resume": None,
+                "after": 0,
+                "before": 4102444800,
+                "filter_type": "exact",
+                "filter_patterns": [recursive_test_site["root_url"]],
+                "batch_size": 1,
+                "continuous": False,
+            },
+            timeout=20,
+        )
+        assert update_response.status_code == 200, update_response.text
+        assert update_response.json()["success"] is True
+
+        conn = sqlite3.connect(tmp_path / "index.sqlite3")
+        try:
+            crawl = conn.execute(
+                "SELECT max_depth, tags_str, config FROM crawls_crawl WHERE id IN (?, ?)",
+                (crawl_id, crawl_id.replace("-", "")),
+            ).fetchone()
+        finally:
+            conn.close()
+
+        assert crawl is not None
+        assert crawl[0] == 1
+        assert crawl[1] == "api-cli"
+        assert '"INDEX_ONLY": true' in crawl[2] or '"INDEX_ONLY":true' in crawl[2]
+
+        remove_response = requests.post(
+            f"http://127.0.0.1:{port}/api/v1/cli/remove",
+            headers=api_headers,
+            json={
+                "delete": True,
+                "after": 0,
+                "before": 4102444800,
+                "filter_type": "exact",
+                "filter_patterns": [recursive_test_site["root_url"]],
+            },
+            timeout=20,
+        )
+        assert remove_response.status_code == 200, remove_response.text
+        remove_payload = remove_response.json()
+        assert remove_payload["success"] is True
+        assert remove_payload["result"]["removed_count"] == 1
+        assert snapshot_id in remove_payload["result"]["removed_snapshot_ids"]
+
+        conn = sqlite3.connect(tmp_path / "index.sqlite3")
+        try:
+            snapshot_count = conn.execute("SELECT COUNT(*) FROM core_snapshot WHERE id = ?", (snapshot_id,)).fetchone()[0]
+        finally:
+            conn.close()
+
+        assert snapshot_count == 0
+    finally:
+        stop_server(tmp_path)

archivebox/tests/test_tag_service.py

@@ -39,7 +39,7 @@ def test_tag_event_projects_tag_to_snapshot():
                 name="example",
                 snapshot_id=str(snapshot.id),
             ),
-        )
+        ).now()
 
     asyncio.run(emit_tag_event())
 

bin/docker_entrypoint.sh

@@ -26,6 +26,8 @@ set -o pipefail
 
 # Load global invariants (set by Dockerfile during image build time, not intended to be customized by users at runtime)
 export DATA_DIR="${DATA_DIR:-/data}"
+export TMP_DIR="${TMP_DIR:-/tmp/archivebox}"
+export LIB_DIR="${LIB_DIR:-/usr/share/archivebox/lib}"
 export ARCHIVEBOX_USER="${ARCHIVEBOX_USER:-archivebox}"
 
 # Global default PUID and PGID if data dir is empty and no intended PUID+PGID is set manually by user
@@ -104,6 +106,18 @@ if ! chown $PUID:$PGID "$DATA_DIR"/* > /dev/null 2>&1; then
     find "$DATA_DIR" -type d -not -path "$DATA_DIR/archive*" -exec chown $PUID:$PGID {} \; > /dev/null 2>&1
     find "$DATA_DIR" -type f -not -path "$DATA_DIR/archive/*" -exec chown $PUID:$PGID {} \; > /dev/null 2>&1
 fi
+
+# Active browser processes do not survive container restarts, but their lock
+# files can. Clear stale browser state before dropping privileges.
+find "$DATA_DIR/personas" -type f \( \
+    -name "SingletonLock" \
+    -o -name "SingletonSocket" \
+    -o -name "SingletonCookie" \
+    -o -name "DevToolsActivePort" \
+    -o -name ".launch.lock" \
+    -o -name ".target.lock" \
+\) -delete >/dev/null 2>&1 || true
+find /tmp "$TMP_DIR" -maxdepth 1 -type d -name "archivebox-chrome-profile.*" -mmin +30 -exec rm -rf {} + >/dev/null 2>&1 || true
     
 
 # also chown BROWSERS_DIR because otherwise 'archivebox setup' wont be able to 'playwright install chromium' at runtime

docker-compose.yml

@@ -11,7 +11,7 @@
 
 services:
     archivebox:
-        image: archivebox/archivebox:latest
+        image: ${ARCHIVEBOX_IMAGE:-archivebox/archivebox:latest}
         ports:
             - 8000:8000
         volumes:
@@ -29,6 +29,17 @@ services:
             - SEARCH_BACKEND_ENGINE=sonic       # tells ArchiveBox to use its built-in Sonic worker for fast full-text search
             # - SEARCH_BACKEND_HOST_NAME=127.0.0.1
             - SEARCH_BACKEND_PASSWORD=SomeSecretPassword
+            - PERSONAS_DIR=/data/personas
+            - NODE_PATH=/home/archivebox/.npm/lib/node_modules:/usr/lib/node_modules:/usr/share/archivebox/lib/npm/node_modules:/data/personas/Default/node_modules
+            - PUPPETEER_EXECUTABLE_PATH=/usr/bin/chromium-browser
+            - CHROME_BIN=/usr/bin/chromium-browser
+            - CHROME_BINARY=/usr/bin/chromium-browser
+            - CHROMIUM_BINARY=/usr/bin/chromium-browser
+            - CHROME_USER_DATA_DIR=/data/personas/Default/chrome_profile
+            - CHROME_HEADLESS=true
+            - CHROME_SANDBOX=false
+            - CHROME_ISOLATION=snapshot
+            - CHROME_ARGS_EXTRA=["--disable-gpu","--disable-features=Translate,OptimizationGuideModelDownloading,MediaRouter"]
             # - PUID=911                        # set to your host user's UID & GID if you encounter permissions issues
             # - PGID=911                        # UID/GIDs lower than 500 may clash with system uids and are not recommended
             # For options below, it's better to set in data/ArchiveBox.conf or use `docker compose run archivebox config --set SOME_KEY=someval` instead of setting here:
@@ -37,6 +48,7 @@ services:
             # - USER_AGENT="..."                # set a custom USER_AGENT to avoid being blocked as a bot
             # ...
             # For more info, see: https://github.com/ArchiveBox/ArchiveBox/wiki/Docker#configuration
+        shm_size: "1gb"
             
         # For ad-blocking during archiving, uncomment this section and the pihole service below
         # networks:
@@ -54,22 +66,11 @@ services:
     # https://github.com/ArchiveBox/ArchiveBox/wiki/Scheduled-Archiving
 
 
-    ### ArchiveBox now starts and uses Sonic automatically when SEARCH_BACKEND_ENGINE=sonic,
-    # so the old standalone docker sidecar below is no longer necessary.
+    ### ArchiveBox now starts and uses Sonic automatically when SEARCH_BACKEND_ENGINE=sonic.
     # If Sonic is ever started after not running for a while, update its full-text index by running:
     #   $ docker compose run archivebox update --index-only
     # https://github.com/ArchiveBox/ArchiveBox/wiki/Setting-up-Search
 
-    # sonic:
-    #     image: archivebox/sonic:latest
-    #     expose:
-    #         - 1491
-    #     environment:
-    #         - SEARCH_BACKEND_PASSWORD=SomeSecretPassword
-    #     volumes:
-    #         #- ./sonic.cfg:/etc/sonic.cfg:ro    # mount to customize: https://raw.githubusercontent.com/ArchiveBox/ArchiveBox/stable/etc/sonic.cfg
-    #         - ./data/sonic:/var/lib/sonic/store
-
 
     ### This optional container runs xvfb+noVNC so you can watch the ArchiveBox browser as it archives things,
     # or remote control it to set up a chrome profile w/ login credentials for sites you want to archive.
@@ -78,6 +79,8 @@ services:
 
     novnc:
         image: theasp/novnc:latest
+        profiles:
+            - novnc
         environment:
             - DISPLAY_WIDTH=1920
             - DISPLAY_HEIGHT=1080

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "archivebox"
-version = "0.9.30rc1"
+version = "0.9.30rc2"
 requires-python = ">=3.13"
 description = "Self-hosted internet archiving solution."
 authors = [{name = "Nick Sweeting", email = "pyproject.toml@archivebox.io"}]
@@ -78,10 +78,10 @@ dependencies = [
     "w3lib>=2.2.1",          # used for parsing content-type encoding from http response headers & html tags
     ### Extractor dependencies (optional binaries detected at runtime via shutil.which)
     ### Binary/Package Management
-    "abxbus>=2.4.9",          # explicit direct dep so local dev env resolves sibling abxbus repo, matching abx-dl EventBus API
+    "abxbus @ git+https://github.com/ArchiveBox/abxbus.git@event-wait-now-results",  # PR #22 EventBus API
     "abxpkg>=1.10.5",         # for: detecting, versioning, and installing binaries via apt/brew/pip/npm
-    "abx-plugins>=1.10.30",   # shared ArchiveBox plugin package with install_args-only overrides
-    "abx-dl>=1.10.30",        # shared ArchiveBox downloader package with install_args-only overrides
+    "abx-plugins>=1.10.31",   # shared ArchiveBox plugin package with install_args-only overrides
+    "abx-dl>=1.10.31",        # shared ArchiveBox downloader package with install_args-only overrides
     ### UUID7 backport for Python <3.14
     "uuid7>=0.1.0; python_version < '3.14'",  # provides the uuid_extensions module on Python 3.13
 ]
@@ -158,6 +158,10 @@ exclude-newer = "5 days"
 exclude-newer-package = { abx-plugins = "1 second", abx-dl = "1 second", abxpkg = "1 second", abxbus = "1 second" }
 # compile-bytecode = true
 
+[tool.uv.sources]
+abx-plugins = { git = "https://github.com/ArchiveBox/abx-plugins.git", branch = "main" }
+abx-dl = { git = "https://github.com/ArchiveBox/abx-dl.git", branch = "main" }
+
 [build-system]
 requires = ["pdm-backend"]
 build-backend = "pdm.backend"