Recover interrupted hook work by hook identity

archivebox/cli/archivebox_run.py

@@ -243,6 +243,7 @@ def process_stdin_records() -> int:
                     crawl_id,
                     snapshot_ids=None if crawl_id in full_crawl_ids else sorted(snapshot_ids_by_crawl[crawl_id]),
                     selected_plugins=None if crawl_id in run_all_plugins_for_crawl else sorted(plugin_names_by_crawl[crawl_id]),
+                    selected_plugins_are_explicit=False,
                 )
     return 0
 

archivebox/core/models.py

@@ -2737,7 +2737,7 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW
 
         return snapshot
 
-    def create_pending_archiveresults(self) -> list["ArchiveResult"]:
+    def create_pending_archiveresults(self, hooks: Iterable[tuple[str, str]] | None = None) -> list["ArchiveResult"]:
         """
         Create ArchiveResult records for all enabled hooks.
 
@@ -2748,18 +2748,17 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW
         Creates one ArchiveResult per hook (not per plugin), with hook_name set.
         This enables step-based execution where all hooks in a step can run in parallel.
         """
-        from archivebox.plugins.hooks import discover_hooks
-        from archivebox.config.common import get_config
+        if hooks is None:
+            from archivebox.plugins.hooks import discover_hooks
+            from archivebox.config.common import get_config
 
-        # Get merged config with crawl-specific PLUGINS filter
-        config = get_config(crawl=self.crawl, snapshot=self)
-        hooks = discover_hooks("Snapshot", config=config)
+            # Compatibility path for direct model callers. The runner passes its
+            # abx-dl hook inventory explicitly so queued rows match execution.
+            config = get_config(crawl=self.crawl, snapshot=self)
+            hooks = ((hook_path.parent.name, hook_path.stem) for hook_path in discover_hooks("Snapshot", config=config))
         archiveresults = []
 
-        for hook_path in hooks:
-            hook_name = hook_path.stem  # e.g., 'on_Snapshot__50_wget'
-            plugin = hook_path.parent.name  # e.g., 'wget'
-
+        for plugin, hook_name in hooks:
             # 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(

archivebox/core/recovery_util.py

@@ -1,12 +1,19 @@
 from __future__ import annotations
 
+from pathlib import Path
+
 from django.utils import timezone
 from rich.console import Console
 
 
+def _is_signal_interrupted_exit(exit_code: int | None) -> bool:
+    return exit_code is not None and (exit_code < 0 or exit_code >= 128)
+
+
 def recover_orchestrator_state(*, include_chrome: bool = False) -> dict[str, int]:
     from archivebox.crawls.models import Crawl
     from archivebox.core.models import ArchiveResult, Snapshot
+    from archivebox.services.archive_result_service import _collect_output_metadata
     from archivebox.machine.models import Process
     from django.db.models import Exists, OuterRef, Q, Subquery, Value
     from django.db.models.functions import Coalesce
@@ -22,6 +29,7 @@ def recover_orchestrator_state(*, include_chrome: bool = False) -> dict[str, int
         "archiveresults_backoff": 0,
         "snapshots_queued_plugin_rows_waiting_on_stale_lease": 0,
         "archiveresults_started_without_running_process": 0,
+        "archiveresults_missing_for_orphaned_hook_processes": 0,
         "snapshots_started_without_running_results": 0,
         "crawls_started_with_due_snapshots": 0,
         "crawls_started_waiting_on_future_snapshots": 0,
@@ -100,6 +108,70 @@ def recover_orchestrator_state(*, include_chrome: bool = False) -> dict[str, int
         process=None,
         modified_at=now,
     )
+    orphaned_hook_processes = Process.objects.filter(
+        process_type=Process.TypeChoices.HOOK,
+        archiveresult__isnull=True,
+    ).exclude(status=Process.StatusChoices.RUNNING)
+    for process in orphaned_hook_processes.only("id", "pwd", "cmd", "process_type", "status"):
+        hook_script_name = process.hook_script_name
+        if not hook_script_name or not process.pwd:
+            continue
+        plugin_dir = Path(process.pwd)
+        snapshot = Snapshot.objects.filter(id=plugin_dir.parent.name).first()
+        if snapshot is None:
+            continue
+        result, created = ArchiveResult.objects.get_or_create(
+            snapshot=snapshot,
+            plugin=plugin_dir.name,
+            hook_name=Path(hook_script_name).stem,
+            defaults={
+                "status": ArchiveResult.StatusChoices.QUEUED,
+            },
+        )
+        if result.status == ArchiveResult.StatusChoices.QUEUED:
+            requeue_snapshot = False
+            # A runner can die after the hook Process exits but before the
+            # ProcessCompletedEvent projector links/finalizes ArchiveResult.
+            # Reconstruct only that exact hook row from the durable Process row.
+            output_files, output_size, output_mimetypes = _collect_output_metadata(plugin_dir)
+            result.process = process
+            if _is_signal_interrupted_exit(process.exit_code):
+                # The owning runner died or was asked to stop while the hook was
+                # still active. Keep the work item queued so takeover retries the
+                # same hook; treating an unknown signal exit as success would
+                # silently skip unfinished side effects.
+                result.output_files = {}
+                result.output_size = 0
+                result.output_mimetypes = ""
+                result.output_str = ""
+                result.status = ArchiveResult.StatusChoices.QUEUED
+                requeue_snapshot = True
+            else:
+                result.output_files = output_files
+                result.output_size = output_size
+                result.output_mimetypes = output_mimetypes
+                result.output_str = process.stderr if process.exit_code not in (0, None) else ""
+                result.status = (
+                    ArchiveResult.StatusChoices.FAILED
+                    if process.exit_code not in (0, None)
+                    else (ArchiveResult.StatusChoices.SUCCEEDED if output_files else ArchiveResult.StatusChoices.NORESULTS)
+                )
+            result.save(
+                update_fields=[
+                    "process",
+                    "output_files",
+                    "output_size",
+                    "output_mimetypes",
+                    "output_str",
+                    "status",
+                    "modified_at",
+                ],
+            )
+            if requeue_snapshot:
+                Snapshot.objects.filter(id=snapshot.id).update(retry_at=now, modified_at=now)
+        if created:
+            cleaned["archiveresults_missing_for_orphaned_hook_processes"] += 1
+            Snapshot.objects.filter(id=snapshot.id).update(retry_at=now, modified_at=now)
     started_snapshots = Snapshot.objects.filter(status=Snapshot.StatusChoices.STARTED).filter(
         Q(retry_at__isnull=True) | Q(retry_at__gt=now),
     )

archivebox/core/views.py

@@ -329,7 +329,7 @@ class SnapshotView(View):
         hidden_card_plugins = {"archivedotorg", "favicon", "title"}
         outputs = [
             out
-            for out in snapshot.discover_outputs(include_filesystem_fallback=False)
+            for out in snapshot.discover_outputs(include_filesystem_fallback=True)
             if (out.get("size") or 0) > 0 and out.get("name") not in hidden_card_plugins
         ]
         archiveresults = {out["name"]: out for out in outputs}

archivebox/machine/models.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 __package__ = "archivebox.machine"
 
 import os
+import signal
 import sys
 import uuid
 import socket
@@ -55,6 +56,13 @@ PROCESS_TIMEOUT_GRACE = timedelta(seconds=30)  # Extra margin before force-clean
 START_TIME_TOLERANCE = 5.0  # Seconds tolerance for start time matching
 
 
+def _default_exit_code_for_unowned_process(process_type: str) -> int:
+    # Hooks are externally visible work items. If their owning runner disappeared
+    # before recording the real exit code, retrying is safer than converting an
+    # unknown interrupted extraction into a durable success/no-result row.
+    return 128 + signal.SIGTERM if process_type == Process.TypeChoices.HOOK else 0
+
+
 def _find_existing_binary_for_reference(machine: Machine, reference: str) -> Binary | None:
     reference = str(reference or "").strip()
     if not reference:
@@ -1608,7 +1616,9 @@ class Process(ModelWithDeleteAfter, models.Model):
                     is_stale = True  # Process no longer exists
 
             if is_stale:
-                proc.mark_exited(exit_code=proc.exit_code if proc.exit_code is not None else 0)
+                proc.mark_exited(
+                    exit_code=proc.exit_code if proc.exit_code is not None else _default_exit_code_for_unowned_process(proc.process_type),
+                )
                 cleaned += 1
 
         return cleaned
@@ -2138,8 +2148,7 @@ class Process(ModelWithDeleteAfter, models.Model):
                 # TODO: Uncomment to cleanup (keeping for debugging for now)
                 # self.stderr_file.unlink(missing_ok=True)
 
-            # Try to get exit code from proc or default to unknown
-            self.exit_code = self.exit_code if self.exit_code is not None else 0
+            self.exit_code = self.exit_code if self.exit_code is not None else _default_exit_code_for_unowned_process(self.process_type)
             if self.exit_code == -1:
                 self.exit_code = 137
             self.ended_at = timezone.now()
@@ -2503,7 +2512,9 @@ class Process(ModelWithDeleteAfter, models.Model):
         # orphaned Process backlog cannot be materialized in memory at once.
         for proc in running_children.iterator(chunk_size=100):
             if not proc.is_running:
-                proc.mark_exited(exit_code=proc.exit_code if proc.exit_code is not None else 0)
+                proc.mark_exited(
+                    exit_code=proc.exit_code if proc.exit_code is not None else _default_exit_code_for_unowned_process(proc.process_type),
+                )
                 cleaned += 1
                 continue
 
@@ -2527,7 +2538,9 @@ class Process(ModelWithDeleteAfter, models.Model):
             ):
                 continue
 
-            proc.mark_exited(exit_code=proc.exit_code if proc.exit_code is not None else 0)
+            proc.mark_exited(
+                exit_code=proc.exit_code if proc.exit_code is not None else _default_exit_code_for_unowned_process(proc.process_type),
+            )
             cleaned += 1
 
         if cleaned:

archivebox/services/archive_result_service.py

@@ -240,6 +240,10 @@ def _has_content_files(output_files: Any) -> bool:
     return any(Path(path).suffix not in {".log", ".pid", ".sh"} for path in _normalize_output_files(output_files))
 
 
+def _is_signal_interrupted_exit(exit_code: int) -> bool:
+    return exit_code < 0 or (exit_code >= 128 and exit_code != PROCESS_EXIT_SKIPPED)
+
+
 def _iter_archiveresult_records(stdout: str) -> list[dict]:
     records: list[dict] = []
     for raw_line in stdout.splitlines():
@@ -352,6 +356,13 @@ def _save_archiveresult_event_to_db(
         with _perf_span("archivebox.ArchiveResultService.on_ArchiveResultEvent.result_update"):
             result.save(update_fields=[*update_fields, "modified_at"])
 
+    if result.status == ArchiveResult.StatusChoices.QUEUED:
+        # ArchiveResult has no retry_at column. If a shutdown/takeover projects
+        # a killed hook back to QUEUED, wake the parent Snapshot/Crawl so the
+        # next runner retries that exact hook instead of waiting on a stale
+        # active-state lease.
+        snapshot.update_and_requeue(retry_at=timezone.now())
+
     if result.status in (ArchiveResult.StatusChoices.SUCCEEDED, ArchiveResult.StatusChoices.NORESULTS):
         with _perf_span("archivebox.ArchiveResultService.on_ArchiveResultEvent.title_update"):
             title_output_str = result.output_str if result.status == ArchiveResult.StatusChoices.SUCCEEDED else ""
@@ -435,14 +446,21 @@ class ArchiveResultService(BaseService):
         # TODO: consider moving this fallback derivation into abx-dl itself.
         # First try both patterns: if the whole abx-dl process crashes, restarting
         # the snapshot may be enough, but don't guess before validating it.
-        process_failed = event.exit_code not in (0, PROCESS_EXIT_SKIPPED)
+        process_interrupted = _is_signal_interrupted_exit(event.exit_code)
+        process_failed = event.exit_code not in (0, PROCESS_EXIT_SKIPPED) and not process_interrupted
         with _perf_span("archivebox.ArchiveResultService.on_ProcessCompletedEvent.emit_archive_result_fallback"):
             await event.emit(
                 ArchiveResultEvent(
                     snapshot_id=snapshot_event.snapshot_id,
                     plugin=event.plugin_name,
                     hook_name=event.hook_name,
-                    status="failed" if process_failed else ("succeeded" if _has_content_files(event.output_files) else "noresult"),
+                    status=(
+                        "queued"
+                        if process_interrupted
+                        else "failed"
+                        if process_failed
+                        else ("succeeded" if _has_content_files(event.output_files) else "noresult")
+                    ),
                     output_str=event.stderr if process_failed else "",
                     output_files=event.output_files,
                     start_ts=event.start_ts,

archivebox/services/runner.py

@@ -10,7 +10,6 @@ import threading
 import time
 from contextlib import nullcontext
 from datetime import timedelta
-from functools import lru_cache
 from pathlib import Path
 from tempfile import TemporaryDirectory
 from typing import Any
@@ -214,6 +213,7 @@ class CrawlRunner:
         show_progress: bool = True,
         interactive_interrupts: bool = False,
         config_overrides: dict[str, Any] | None = None,
+        selected_plugins_are_explicit: bool = True,
     ):
         self.crawl = crawl
         self.bus = create_bus(name=_bus_name("ArchiveBox", str(crawl.id)), total_timeout=3600.0)
@@ -242,6 +242,7 @@ class CrawlRunner:
         )
         ArchiveResultService(self.bus)
         self.selected_plugins = selected_plugins
+        self.selected_plugins_from_args = selected_plugins is not None and selected_plugins_are_explicit
         self.initial_snapshot_ids = snapshot_ids
         self.snapshot_tasks: dict[str, asyncio.Task[None]] = {}
         self.snapshot_semaphore = asyncio.Semaphore(1)
@@ -1015,6 +1016,7 @@ class CrawlRunner:
                 await sync_to_async(run_snapshot_maintenance, thread_sensitive=True)(snapshot_id)
                 return
             snapshot_selected_plugins = self.selected_plugins
+            selected_hooks_by_plugin = None
             if snapshot["status"] == "started":
                 _reset_count, running_count = await sync_to_async(snapshot["_snapshot"].reset_abandoned_results, thread_sensitive=True)()
                 if running_count:
@@ -1025,10 +1027,24 @@ class CrawlRunner:
                         thread_sensitive=True,
                     )()
                     return
-                snapshot_selected_plugins = snapshot_selected_plugins or await sync_to_async(
-                    queued_plugins_for_snapshot,
-                    thread_sensitive=True,
-                )(snapshot["id"])
+                if await sync_to_async(snapshot["_snapshot"].is_finished_processing, thread_sensitive=True)():
+                    await sync_to_async(finalize_completed_snapshot, thread_sensitive=True)(
+                        snapshot["id"],
+                        output_dir=Path(snapshot["output_dir"]),
+                    )
+                    return
+                if not self.selected_plugins_from_args:
+                    queued_plugins, selected_hooks_by_plugin = await sync_to_async(
+                        queued_plugins_and_hooks_for_snapshot,
+                        thread_sensitive=True,
+                    )(snapshot["id"])
+                    if queued_plugins:
+                        if snapshot_selected_plugins:
+                            queued_plugins = [plugin for plugin in queued_plugins if plugin in snapshot_selected_plugins]
+                            selected_hooks_by_plugin = {
+                                plugin: hooks for plugin, hooks in (selected_hooks_by_plugin or {}).items() if plugin in queued_plugins
+                            }
+                        snapshot_selected_plugins = queued_plugins
             if snapshot["depth"] > 0 and CrawlLimitState.from_config(snapshot["config"]).get_stop_reason() in (
                 "crawl_max_size",
                 "crawl_timeout",
@@ -1043,6 +1059,18 @@ class CrawlRunner:
                 if snapshot_selected_plugins
                 else self.plugins
             )
+            if selected_hooks_by_plugin is not None:
+                filtered_plugins = {}
+                for plugin_name, plugin in plugins.items():
+                    selected_hook_names = selected_hooks_by_plugin.get(plugin_name)
+                    if selected_hook_names is None:
+                        filtered_plugins[plugin_name] = plugin
+                        continue
+                    filtered_hooks = [
+                        hook for hook in plugin.hooks if hook.name in selected_hook_names or Path(hook.name).stem in selected_hook_names
+                    ]
+                    filtered_plugins[plugin_name] = plugin.model_copy(update={"hooks": filtered_hooks})
+                plugins = filtered_plugins
             abx_snapshot = AbxSnapshot(
                 id=snapshot["id"],
                 url=snapshot["url"],
@@ -1062,6 +1090,7 @@ class CrawlRunner:
                 snapshot_cleanup_enabled=True,
                 snapshot_cleanup_phase_timeout=snapshot_phase_timeout,
                 abort_requested=self.crawl_is_cancelled,
+                selected_hooks_by_plugin=selected_hooks_by_plugin,
             )
             try:
                 snapshot_event = SnapshotEvent(
@@ -1137,6 +1166,7 @@ def run_crawl(
     show_progress: bool = True,
     interactive_interrupts: bool = False,
     config_overrides: dict[str, Any] | None = None,
+    selected_plugins_are_explicit: bool = True,
 ) -> None:
     from archivebox.crawls.models import Crawl
     from django.db import close_old_connections
@@ -1154,6 +1184,7 @@ def run_crawl(
                     show_progress=show_progress,
                     interactive_interrupts=interactive_interrupts,
                     config_overrides=config_overrides,
+                    selected_plugins_are_explicit=selected_plugins_are_explicit,
                 ).run(),
             )
         finally:
@@ -1241,14 +1272,7 @@ def run_binary(binary_id: str) -> None:
     asyncio.run(_run_binary(binary_id))
 
 
-@lru_cache(maxsize=1)
-def _snapshot_hook_names_by_plugin() -> dict[str, frozenset[str]]:
-    return {
-        plugin.name: frozenset(hook.name for hook in plugin.filter_hooks("Snapshot")) for plugin in _discover_archivebox_plugins().values()
-    }
-
-
-def queued_plugins_for_snapshot(snapshot_id: str) -> list[str] | None:
+def queued_plugins_and_hooks_for_snapshot(snapshot_id: str) -> tuple[list[str] | None, dict[str, set[str] | None] | None]:
     from archivebox.core.models import ArchiveResult
 
     queued_results = list(
@@ -1259,30 +1283,39 @@ def queued_plugins_for_snapshot(snapshot_id: str) -> list[str] | None:
         .exclude(plugin="")
         .only("id", "plugin", "hook_name"),
     )
-    hooks_by_plugin = _snapshot_hook_names_by_plugin()
-    obsolete_result_ids = [
-        result.id
-        for result in queued_results
-        if result.hook_name and result.hook_name not in hooks_by_plugin.get(result.plugin, frozenset())
-    ]
-    if obsolete_result_ids:
-        # Hook names are the scheduler identity for ArchiveResults. If an old
-        # queued row names a hook that the current plugin model cannot run, hard
-        # fail only that row so the scheduler drains without hiding stale/broken
-        # plugin state as an intentional skip.
-        ArchiveResult.objects.filter(
-            id__in=obsolete_result_ids,
-            status=ArchiveResult.StatusChoices.QUEUED,
-        ).update(
-            status=ArchiveResult.StatusChoices.FAILED,
-            output_str="Hook no longer exists in the current plugin set.",
-            modified_at=timezone.now(),
-        )
 
-    queued_plugins = sorted({result.plugin for result in queued_results if result.id not in obsolete_result_ids})
+    selected_hooks_by_plugin: dict[str, set[str] | None] = {}
+    queued_plugins = sorted({result.plugin for result in queued_results})
+    for result in queued_results:
+        # hook_name is the modern scheduler identity. Empty hook_name rows are
+        # legacy plugin-level work and must keep running the whole plugin.
+        if not result.hook_name:
+            selected_hooks_by_plugin[result.plugin] = None
+        elif result.plugin not in selected_hooks_by_plugin:
+            selected_hooks_by_plugin[result.plugin] = {result.hook_name}
+        elif selected_hooks_by_plugin[result.plugin] is not None:
+            selected_hooks_by_plugin[result.plugin].add(result.hook_name)
     if queued_plugins:
-        return queued_plugins
-    return None
+        return queued_plugins, selected_hooks_by_plugin
+    return None, None
+
+
+def queued_plugins_for_snapshot(snapshot_id: str) -> list[str] | None:
+    queued_plugins, _selected_hooks_by_plugin = queued_plugins_and_hooks_for_snapshot(snapshot_id)
+    return queued_plugins
+
+
+def snapshot_hooks_for_pending_archiveresults(snapshot) -> list[tuple[str, str]]:
+    from archivebox.config.common import get_config
+
+    config = get_config(crawl=snapshot.crawl, snapshot=snapshot)
+    plugin_names = [name.strip() for name in str(config.PLUGINS or "").split(",") if name.strip()]
+    plugins = (
+        filter_plugins(_discover_archivebox_plugins(), plugin_names, include_providers=True)
+        if plugin_names
+        else _discover_archivebox_plugins()
+    )
+    return sorted((plugin.name, hook.name) for plugin in plugins.values() for hook in plugin.filter_hooks("Snapshot"))
 
 
 def run_snapshot_maintenance(snapshot_id: str, *, output_dir: Path | None = None) -> bool:
@@ -1461,6 +1494,7 @@ def run_due_snapshot(snapshot, *, lock_seconds: int, interactive_interrupts: boo
                 selected_plugins=selected_plugins,
                 process_discovered_snapshots_inline=True,
                 interactive_interrupts=interactive_interrupts,
+                selected_plugins_are_explicit=False,
             )
         finally:
             # Targeted plugin rows can complete while the Snapshot remains
@@ -1492,6 +1526,7 @@ def run_due_snapshot(snapshot, *, lock_seconds: int, interactive_interrupts: boo
                 selected_plugins=selected_plugins,
                 process_discovered_snapshots_inline=True,
                 interactive_interrupts=interactive_interrupts,
+                selected_plugins_are_explicit=False,
             )
             if search_only_plugins:
                 from archivebox.core.models import ArchiveResult
@@ -1533,7 +1568,7 @@ def run_due_snapshot(snapshot, *, lock_seconds: int, interactive_interrupts: boo
         # rows before ticking so maintenance-only final rows, e.g. search
         # backfill on a paused snapshot, cannot make queued -> sealed skip the
         # real extraction work after resume.
-        snapshot.create_pending_archiveresults()
+        snapshot.create_pending_archiveresults(hooks=snapshot_hooks_for_pending_archiveresults(snapshot))
         snapshot.sm.tick()
         snapshot.refresh_from_db()
         if snapshot.status == Snapshot.StatusChoices.SEALED:
@@ -1552,7 +1587,15 @@ def run_due_snapshot(snapshot, *, lock_seconds: int, interactive_interrupts: boo
         selected_plugins=queued_plugins_for_snapshot(str(snapshot.id)),
         process_discovered_snapshots_inline=True,
         interactive_interrupts=interactive_interrupts,
+        selected_plugins_are_explicit=False,
     )
+    snapshot.refresh_from_db()
+    if queued_plugins_for_snapshot(str(snapshot.id)):
+        # Hook-level resume work is tracked by queued ArchiveResult rows, not by
+        # the Snapshot lease. If a partial pass returns with rows still queued,
+        # wake the Snapshot immediately so takeover does not wait out a stale
+        # active-state lock before running the remaining hooks.
+        snapshot.update_and_requeue(retry_at=timezone.now())
     return True
 
 
@@ -1826,6 +1869,7 @@ def _run_due_queued_plugin_result(
         config_overrides={
             "CRAWL_MAX_CONCURRENT_SNAPSHOTS": QUEUED_PLUGIN_RESULT_BATCH_SIZE,
         },
+        selected_plugins_are_explicit=False,
     )
     if all(plugin.startswith("search_backend_") for plugin in selected_plugins):
         queued_results = ArchiveResult.objects.filter(

archivebox/tests/test_opencode_agent.py

@@ -48,7 +48,6 @@ def _set_archivebox_config(data_dir: Path, *values: str, env: dict[str, str] | N
 @pytest.fixture
 def opencode_archive_config(initialized_archive):
     port = _free_port()
-    lib_dir = initialized_archive / "lib"
     state_dir = initialized_archive / "opencode"
     env = os.environ.copy()
     env.update(
@@ -57,8 +56,6 @@ def opencode_archive_config(initialized_archive):
             "ABXPKG_MIN_RELEASE_AGE": "0",
             "ABX_RUNTIME": "archivebox",
             "ARCHIVEBOX_ALLOW_NO_UNIX_SOCKETS": "true",
-            "LIB_DIR": str(lib_dir),
-            "ABXPKG_LIB_DIR": str(lib_dir),
             "OPENCODE_ENABLED": "True",
             "OPENCODE_HOST": "127.0.0.1",
             "OPENCODE_PORT": str(port),
@@ -75,10 +72,9 @@ def opencode_archive_config(initialized_archive):
         f"OPENCODE_WORKDIR={initialized_archive}",
         f"OPENCODE_STATE_DIR={state_dir}",
         "OPENCODE_TIMEOUT=60",
-        f"LIB_DIR={lib_dir}",
         env=env,
     )
-    return SimpleNamespace(data_dir=initialized_archive, lib_dir=lib_dir, port=port, state_dir=state_dir, env=env)
+    return SimpleNamespace(data_dir=initialized_archive, port=port, state_dir=state_dir, env=env)
 
 
 @pytest.fixture

archivebox/tests/test_takeover_util.py

@@ -379,7 +379,7 @@ def test_live_server_keeps_http_runtime_while_update_runs_real_sqlite_indexer(tm
         )
         assert "Stopping older ArchiveBox runner process" in update_stdout
 
-        deadline = time.time() + 90
+        deadline = time.time() + 180
         runner_pid_after = runner_pid_before
         while time.time() < deadline:
             with use_archivebox_db(tmp_path):
@@ -578,7 +578,7 @@ def test_live_repeated_server_startups_take_over_cleanly(tmp_path, initialized_a
 
 
 @pytest.mark.timeout(420)
-def test_live_add_update_jobs_survive_server_and_cli_owner_exits(tmp_path, initialized_archive):
+def test_live_add_update_jobs_survive_server_and_cli_owner_exits(tmp_path, initialized_archive, recursive_test_site):
     plugins_root = tmp_path / "runtime_plugins"
     marker_dir = tmp_path / "slow-plugin-markers"
     plugin_dir = plugins_root / "slow_exit"
@@ -658,8 +658,8 @@ def test_live_add_update_jobs_survive_server_and_cli_owner_exits(tmp_path, initi
                 "--max-urls=2",
                 "--crawl-max-size=50mb",
                 "--plugins=wget,parse_html_urls,slow_exit",
-                "https://example.com",
-                "https://blog.sweeting.me",
+                recursive_test_site["root_url"],
+                recursive_test_site["child_urls"][0],
             ],
             cwd=tmp_path,
             env=env,
@@ -696,12 +696,15 @@ def test_live_add_update_jobs_survive_server_and_cli_owner_exits(tmp_path, initi
         assert add_proc.poll() is None, "foreground add should keep owning its crawl after the server exits"
         assert "Got SIGTERM" in server_log.read_text(encoding="utf-8", errors="replace")
 
+        stop_archivebox_process(add_proc, signal.SIGKILL, timeout=30)
+        add_output = add_log.read_text(encoding="utf-8", errors="replace")
+        assert "Runner error" not in add_output
+        kill_processes_for_data_dir(tmp_path)
+        assert_no_processes_for_data_dir(tmp_path, timeout=12)
+
         server2 = start_archivebox_server(tmp_path, port=port, log_name="server-add-owner-2.log", env=env)
         _server2_log = server2.log_path
         (marker_dir / "allow-finish").touch()
-        stop_archivebox_process(add_proc, signal.SIGTERM, timeout=30)
-        add_output = add_log.read_text(encoding="utf-8", errors="replace")
-        assert "Runner error" not in add_output
 
         deadline = time.time() + 90
         crawls = []
@@ -751,17 +754,12 @@ def test_live_add_update_jobs_survive_server_and_cli_owner_exits(tmp_path, initi
         assert len(counter_runs) == len(set(counter_runs))
         assert not (marker_dir / "counter-duplicates.txt").exists()
 
-        # TODO: improve abx-dl's ability to explicitly resume from a given plugin / hook and skip ones before that
-        # current behavior: on retry, earlier sealed results are left untouched; the interrupted result is marked skipped
-        # and may have partial output saved to fs
-        # assertions that enforce the current behavior (uncommented): previous results are not run twice + interrupted
-        # result is marked skipped
-        assert bad_results == [("slow_exit", ArchiveResult.StatusChoices.SKIPPED, "")]
-
-        # desired future behavior: earlier sealed results are left untouched, interrupted result is retried and cleanly
-        # overwrites on top of any previous partial output
-        # assert not bad_results
-        # assert (marker_dir / "hook-finished").exists()
+        # The interrupted hook should be retried directly without rerunning the
+        # previous hook in the same plugin. That keeps plugin-level shell hooks
+        # idempotent across runner takeover instead of depending on each hook to
+        # detect partial prior work itself.
+        assert not bad_results
+        assert (marker_dir / "hook-finished").exists()
     finally:
         for proc in (add_proc, add_proc2, server, server2, server3):
             if proc is not None and proc.poll() is None:

archivebox/tests/test_urls.py

@@ -895,19 +895,19 @@ class TestUrlRouting:
             assert f"http://{public_host}/static/archive.png" in live_html
             assert "?preview=1" in live_html
             assert "function createMainFrame(previousFrame)" in live_html
-            assert "function activateCardPreview(card, link)" in live_html
-            assert "ensureMainFrame(true)" in live_html
+            assert "function activateCardPreview(card, link, updateHash=true)" in live_html
+            assert "ensureMainFrame(currentSrc !== nextSrcAbs)" in live_html
             assert "previousFrame.parentNode.replaceChild(frame, previousFrame)" in live_html
             assert "previousFrame.src = 'about:blank'" in live_html
             assert "event.stopImmediatePropagation()" in live_html
-            assert "const matchingLink = [...document.querySelectorAll('a[target=preview]')].find" in live_html
+            assert "const matchingLink = findPreviewLinkForHash(selectedPreviewHash)" in live_html
             assert "jQuery(link).click()" not in live_html
             assert "searchParams.delete('preview')" in live_html
             assert "doc.body.style.flexDirection = 'column'" in live_html
             assert "doc.body.style.alignItems = 'center'" in live_html
             assert "img.style.margin = '0 auto'" in live_html
             assert "window.location.hash = getPreviewHashValueFromHref(rawTarget)" in live_html
-            assert "const selectedPreviewHash = decodeURIComponent(window.location.hash.slice(1)).toLowerCase()" in live_html
+            assert "const selectedPreviewHash = window.location.hash ? decodeURIComponent(window.location.hash.slice(1)).toLowerCase() : ''" in live_html
             assert "pointer-events: none;" in live_html
             assert "pointer-events: auto;" in live_html
             assert 'class="thumbnail-click-overlay"' in live_html
@@ -921,19 +921,19 @@ class TestUrlRouting:
             assert f"http://{public_host}/static/archive.png" in static_html
             assert "?preview=1" in static_html
             assert "function createMainFrame(previousFrame)" in static_html
-            assert "function activateCardPreview(card, link)" in static_html
-            assert "ensureMainFrame(true)" in static_html
+            assert "function activateCardPreview(card, link, updateHash=true)" in static_html
+            assert "ensureMainFrame(currentSrc !== nextSrcAbs)" in static_html
             assert "previousFrame.parentNode.replaceChild(frame, previousFrame)" in static_html
             assert "previousFrame.src = 'about:blank'" in static_html
             assert "event.stopImmediatePropagation()" in static_html
-            assert "const matchingLink = [...document.querySelectorAll('a[target=preview]')].find" in static_html
+            assert "const matchingLink = findPreviewLinkForHash(selectedPreviewHash)" in static_html
             assert "jQuery(link).click()" not in static_html
             assert "searchParams.delete('preview')" in static_html
             assert "doc.body.style.flexDirection = 'column'" in static_html
             assert "doc.body.style.alignItems = 'center'" in static_html
             assert "img.style.margin = '0 auto'" in static_html
             assert "window.location.hash = getPreviewHashValueFromHref(rawTarget)" in static_html
-            assert "const selectedPreviewHash = decodeURIComponent(window.location.hash.slice(1)).toLowerCase()" in static_html
+            assert "const selectedPreviewHash = window.location.hash ? decodeURIComponent(window.location.hash.slice(1)).toLowerCase() : ''" in static_html
             assert "pointer-events: none;" in static_html
             assert "pointer-events: auto;" in static_html
             assert 'class="thumbnail-click-overlay"' in static_html