Revert "fix: project one archive result per plugin"

archivebox/api/v1_core.py

@@ -528,7 +528,12 @@ def create_archiveresult(
     normalized_status = ArchiveResult.normalize_status(status)
     parsed_output_json = _parse_archiveresult_output_json(output_json)
     hook = hook_name or ARCHIVERESULT_UPLOAD_HOOK_NAME
-    existing_result = ArchiveResult.objects.filter(snapshot=snapshot, plugin=plugin_name).first()
+    result_lookup = {
+        "snapshot": snapshot,
+        "plugin": plugin_name,
+        "hook_name": hook,
+    }
+    existing_result = ArchiveResult.objects.filter(**result_lookup).first()
     existing_output_files = dict(existing_result.output_files or {}) if existing_result else {}
     output_files = _write_archiveresult_files(
         request,
@@ -541,7 +546,7 @@ def create_archiveresult(
 
     with transaction.atomic():
         Snapshot.objects.select_for_update().get(pk=snapshot.pk)
-        existing_result = ArchiveResult.objects.filter(snapshot=snapshot, plugin=plugin_name).first()
+        existing_result = ArchiveResult.objects.filter(**result_lookup).first()
         if existing_result:
             output_files = {
                 **dict(existing_result.output_files or {}),
@@ -565,7 +570,6 @@ def create_archiveresult(
         output_size, output_mimetypes = _summarize_archiveresult_output_files(output_files)
         output_file_paths = list(output_files.keys())
         result.status = normalized_status
-        result.hook_name = hook
         result.output_str = output_str or (output_file_paths[0] if output_file_paths else "")
         result.output_json = parsed_output_json
         result.output_files = output_files

archivebox/cli/archivebox_archiveresult.py

@@ -67,7 +67,7 @@ def create_archiveresults(
     Reads Snapshot records from stdin and emits ArchiveResult request JSONL.
     Pass-through: Non-Snapshot/ArchiveResult records are output unchanged.
     If --plugin is specified, only emits requests for that plugin.
-    Otherwise, emits one request for each enabled Snapshot plugin.
+    Otherwise, emits requests for all enabled snapshot hooks.
 
     Exit codes:
         0: Success
@@ -144,17 +144,18 @@ def create_archiveresults(
 
     created_count = 0
     for snapshot in snapshots:
-        if plugin:
+        config = get_config(crawl=snapshot.crawl, snapshot=snapshot)
+        hooks = [
+            hook
+            for hook in discover_hooks("Snapshot", filter_disabled=not plugin, config=config)
+            if not plugin or hook.parent.name == plugin
+        ]
+        for hook_path in hooks:
+            hook_name = hook_path.stem
+            plugin_name = hook_path.parent.name
             if not is_tty:
-                write_record(build_archiveresult_request(snapshot.id, plugin, status=status))
+                write_record(build_archiveresult_request(snapshot.id, plugin_name, hook_name=hook_name, status=status))
             created_count += 1
-        else:
-            config = get_config(crawl=snapshot.crawl, snapshot=snapshot)
-            hooks = discover_hooks("Snapshot", filter_disabled=True, config=config)
-            for plugin_name in dict.fromkeys(hook_path.parent.name for hook_path in hooks):
-                if not is_tty:
-                    write_record(build_archiveresult_request(snapshot.id, plugin_name, status=status))
-                created_count += 1
 
     rprint(f"[green]Created {created_count} archive result request records[/green]", file=sys.stderr)
     return 0
@@ -254,11 +255,9 @@ def update_archiveresults(
 
             # Apply updates from CLI flags
             if status:
-                if status == ArchiveResult.StatusChoices.QUEUED:
-                    result.reset_for_retry()
-                else:
-                    result.status = status
-                    result.save(update_fields=["status", "modified_at"])
+                result.status = status
+
+            result.save()
             updated_count += 1
 
             if not is_tty:

archivebox/cli/archivebox_extract.py

@@ -70,7 +70,12 @@ def process_archiveresult_by_id(archiveresult_id: str) -> int:
             )
             return 1
 
-        run_crawl(str(snapshot.crawl_id), snapshot_ids=[str(snapshot.id)], selected_plugins=[archiveresult.plugin])
+        run_crawl(
+            str(snapshot.crawl_id),
+            snapshot_ids=[str(snapshot.id)],
+            selected_plugins=[archiveresult.plugin],
+            selected_plugins_are_explicit=False,
+        )
         archiveresult.refresh_from_db()
 
         if archiveresult.status == ArchiveResult.StatusChoices.SUCCEEDED:
@@ -122,6 +127,7 @@ def run_plugins(
     from archivebox.core.models import Snapshot
     from archivebox.core.models import ArchiveResult
     from archivebox.services.runner import run_crawl
+    from abx_dl.models import discover_plugins
 
     is_tty = sys.stdout.isatty()
 
@@ -142,6 +148,8 @@ def run_plugins(
     # Gather snapshot IDs and optional plugin constraints to process
     snapshot_ids = set()
     requested_plugins_by_snapshot: dict[str, set[str]] = defaultdict(set)
+    requested_hooks_by_snapshot: dict[str, set[tuple[str, str]]] = defaultdict(set)
+    plugin_level_requests_by_snapshot: dict[str, set[str]] = defaultdict(set)
     for record in records:
         record_type = record.get("type")
 
@@ -163,7 +171,14 @@ def run_plugins(
                 snapshot_ids.add(str(snapshot_id))
                 plugin_name = record.get("plugin")
                 if plugin_name and not plugins_list:
-                    requested_plugins_by_snapshot[str(snapshot_id)].add(str(plugin_name))
+                    snapshot_key = str(snapshot_id)
+                    plugin_key = str(plugin_name)
+                    requested_plugins_by_snapshot[snapshot_key].add(plugin_key)
+                    hook_name = str(record.get("hook_name") or "")
+                    if hook_name:
+                        requested_hooks_by_snapshot[snapshot_key].add((plugin_key, hook_name))
+                    else:
+                        plugin_level_requests_by_snapshot[snapshot_key].add(plugin_key)
 
         elif "id" in record:
             # Assume it's a snapshot ID
@@ -192,21 +207,38 @@ def run_plugins(
             if snapshot_id in existing_snapshot_ids
             for plugin_name in plugin_names
         )
-    queued_rows: set[tuple[str, str]] = set()
-    if preserve_queued and requested_pairs:
-        queued_rows = {
-            (str(snapshot_id), plugin_name)
-            for snapshot_id, plugin_name in ArchiveResult.objects.filter(
-                snapshot_id__in=existing_snapshot_ids,
-                plugin__in={plugin_name for _snapshot_id, plugin_name in requested_pairs},
-                status=ArchiveResult.StatusChoices.QUEUED,
-            ).values_list("snapshot_id", "plugin")
+    plugins_by_name = discover_plugins(runtime="archivebox")
+    requested_rows: set[tuple[str, str, str]] = set()
+    for snapshot_id, plugin_name in requested_pairs:
+        exact_hook_names = {
+            hook_name
+            for requested_plugin, hook_name in requested_hooks_by_snapshot.get(snapshot_id, set())
+            if requested_plugin == plugin_name
         }
-    rows_to_queue = requested_pairs - queued_rows
+        if exact_hook_names and plugin_name not in plugin_level_requests_by_snapshot.get(snapshot_id, set()):
+            requested_rows.update((snapshot_id, plugin_name, hook_name) for hook_name in exact_hook_names)
+            continue
+        plugin = plugins_by_name.get(plugin_name)
+        hooks = plugin.filter_hooks("Snapshot") if plugin is not None else []
+        if hooks:
+            requested_rows.update((snapshot_id, plugin_name, hook.name) for hook in hooks)
+        else:
+            requested_rows.add((snapshot_id, plugin_name, ""))
+
+    queued_rows: set[tuple[str, str, str]] = set()
+    if preserve_queued and requested_rows:
+        queued_rows = {
+            (str(snapshot_id), plugin_name, hook_name)
+            for snapshot_id, plugin_name, hook_name in ArchiveResult.objects.filter(
+                snapshot_id__in=existing_snapshot_ids,
+                plugin__in={plugin_name for _snapshot_id, plugin_name, _hook_name in requested_rows},
+                status=ArchiveResult.StatusChoices.QUEUED,
+            ).values_list("snapshot_id", "plugin", "hook_name")
+        }
+    rows_to_queue = requested_rows - queued_rows
 
     reset_fields = {
         "status": ArchiveResult.StatusChoices.QUEUED,
-        "hook_name": "",
         "output_str": "",
         "output_json": None,
         "output_files": {},
@@ -217,28 +249,28 @@ def run_plugins(
         "modified_at": timezone.now(),
     }
     if rows_to_queue and plugins_list:
-        rows_to_reset_by_plugin: dict[str, set[str]] = defaultdict(set)
-        for snapshot_id, plugin_name in rows_to_queue:
-            rows_to_reset_by_plugin[plugin_name].add(snapshot_id)
-        for plugin_name, plugin_snapshot_ids in rows_to_reset_by_plugin.items():
-            ArchiveResult.objects.filter(snapshot_id__in=plugin_snapshot_ids, plugin=plugin_name).update(
+        rows_to_reset_by_hook: dict[tuple[str, str], set[str]] = defaultdict(set)
+        for snapshot_id, plugin_name, hook_name in rows_to_queue:
+            rows_to_reset_by_hook[(plugin_name, hook_name)].add(snapshot_id)
+        for (plugin_name, hook_name), plugin_snapshot_ids in rows_to_reset_by_hook.items():
+            ArchiveResult.objects.filter(snapshot_id__in=plugin_snapshot_ids, plugin=plugin_name, hook_name=hook_name).update(
                 **reset_fields,
             )
     elif rows_to_queue and requested_plugins_by_snapshot:
-        snapshot_ids_by_plugin: dict[str, set[str]] = defaultdict(set)
-        for snapshot_id, plugin_name in rows_to_queue:
-            snapshot_ids_by_plugin[plugin_name].add(snapshot_id)
-        for plugin_name, plugin_snapshot_ids in snapshot_ids_by_plugin.items():
-            ArchiveResult.objects.filter(snapshot_id__in=plugin_snapshot_ids, plugin=plugin_name).update(
+        snapshot_ids_by_hook: dict[tuple[str, str], set[str]] = defaultdict(set)
+        for snapshot_id, plugin_name, hook_name in rows_to_queue:
+            snapshot_ids_by_hook[(plugin_name, hook_name)].add(snapshot_id)
+        for (plugin_name, hook_name), plugin_snapshot_ids in snapshot_ids_by_hook.items():
+            ArchiveResult.objects.filter(snapshot_id__in=plugin_snapshot_ids, plugin=plugin_name, hook_name=hook_name).update(
                 **reset_fields,
             )
     existing_rows = (
         {
-            (str(snapshot_id), plugin_name)
-            for snapshot_id, plugin_name in ArchiveResult.objects.filter(
+            (str(snapshot_id), plugin_name, hook_name)
+            for snapshot_id, plugin_name, hook_name in ArchiveResult.objects.filter(
                 snapshot_id__in=existing_snapshot_ids,
-                plugin__in={plugin_name for _snapshot_id, plugin_name in rows_to_queue},
-            ).values_list("snapshot_id", "plugin")
+                plugin__in={plugin_name for _snapshot_id, plugin_name, _hook_name in rows_to_queue},
+            ).values_list("snapshot_id", "plugin", "hook_name")
         }
         if rows_to_queue
         else set()
@@ -250,10 +282,10 @@ def run_plugins(
                 ArchiveResult(
                     snapshot_id=snapshot_id,
                     plugin=plugin_name,
-                    hook_name="",
+                    hook_name=hook_name,
                     status=ArchiveResult.StatusChoices.QUEUED,
                 )
-                for snapshot_id, plugin_name in sorted(missing_rows)
+                for snapshot_id, plugin_name, hook_name in sorted(missing_rows)
             ],
             batch_size=500,
         )
@@ -261,13 +293,13 @@ def run_plugins(
     processed_count = len(existing_snapshot_ids)
     queue_at = timezone.now()
     if existing_snapshot_ids:
-        if requested_pairs:
-            # Search indexing on a sealed Snapshot is the only targeted plugin
+        if requested_rows:
+            # Search indexing on a sealed Snapshot is the only targeted hook
             # allowed to bypass the normal lifecycle. Every other requested
             # plugin requeues its Snapshot through the unified state machine.
-            affected_snapshot_ids = {snapshot_id for snapshot_id, _plugin_name in rows_to_queue}
+            affected_snapshot_ids = {snapshot_id for snapshot_id, _plugin_name, _hook_name in rows_to_queue}
             if preserve_queued and queued_rows:
-                queued_snapshot_ids = {snapshot_id for snapshot_id, _plugin_name in queued_rows}
+                queued_snapshot_ids = {snapshot_id for snapshot_id, _plugin_name, _hook_name in queued_rows}
                 affected_snapshot_ids.update(
                     str(snapshot_id)
                     for snapshot_id in Snapshot.objects.filter(id__in=queued_snapshot_ids)
@@ -282,7 +314,7 @@ def run_plugins(
                     )
                 )
             requested_plugins_by_id: dict[str, set[str]] = defaultdict(set)
-            for snapshot_id, plugin_name in requested_pairs:
+            for snapshot_id, plugin_name, _hook_name in requested_rows:
                 requested_plugins_by_id[snapshot_id].add(plugin_name)
             for snapshot in Snapshot.objects.filter(id__in=affected_snapshot_ids).only("id", "status", "modified_at"):
                 # Guard the read-time status so we never bump retry_at on a
@@ -318,7 +350,7 @@ def run_plugins(
                     refresh=False,
                     extra_filter={"status": snapshot.status},
                 )
-    if existing_crawl_ids and not requested_pairs:
+    if existing_crawl_ids and not requested_rows:
         from archivebox.crawls.models import Crawl
 
         for crawl in Crawl.objects.filter(id__in=existing_crawl_ids).only("id", "status", "retry_at", "modified_at"):
@@ -362,7 +394,7 @@ def run_plugins(
             selected_plugins = (
                 plugins_list
                 or sorted(
-                    {plugin for snapshot_id in crawl_snapshot_ids for plugin in requested_plugins_by_snapshot.get(str(snapshot_id), set())},
+                    {plugin for snapshot_id, plugin, _hook_name in requested_rows if snapshot_id in crawl_snapshot_ids},
                 )
                 or None
             )
@@ -371,6 +403,7 @@ def run_plugins(
                 snapshot_ids=sorted(crawl_snapshot_ids),
                 selected_plugins=selected_plugins,
                 show_progress=show_progress,
+                selected_plugins_are_explicit=bool(plugins_list),
             )
 
     if not emit_results:

archivebox/cli/archivebox_run.py

@@ -151,18 +151,7 @@ def process_stdin_records() -> int:
                     queued_count += 1
 
             elif record_type == TYPE_ARCHIVERESULT:
-                if record_id:
-                    # Existing archiveresult - re-queue
-                    try:
-                        archiveresult = ArchiveResult.objects.get(id=record_id)
-                    except ArchiveResult.DoesNotExist:
-                        archiveresult = None
-                else:
-                    archiveresult = None
-
-                snapshot_id = record.get("snapshot_id")
-                plugin_name = record.get("plugin")
-                snapshot = None
+                archiveresult = ArchiveResult.from_json(record)
                 if archiveresult:
                     if archiveresult.status in [
                         ArchiveResult.StatusChoices.FAILED,
@@ -172,12 +161,10 @@ def process_stdin_records() -> int:
                     ]:
                         archiveresult.reset_for_retry()
                     snapshot = archiveresult.snapshot
-                    plugin_name = plugin_name or archiveresult.plugin
-                elif snapshot_id:
-                    try:
-                        snapshot = Snapshot.objects.get(id=snapshot_id)
-                    except Snapshot.DoesNotExist:
-                        snapshot = None
+                    plugin_name = archiveresult.plugin
+                else:
+                    snapshot = None
+                    plugin_name = None
 
                 if snapshot:
                     snapshot.queue_for_extraction()

archivebox/core/migrations/0052_unique_archiveresult_per_snapshot_plugin.py

@@ -1,71 +0,0 @@
-from django.db import migrations, models
-from django.db.models import Sum
-
-
-def consolidate_archiveresults_per_plugin(apps, schema_editor):
-    ArchiveResult = apps.get_model("core", "ArchiveResult")
-    Snapshot = apps.get_model("core", "Snapshot")
-    duplicate_groups = ArchiveResult.objects.values("snapshot_id", "plugin").annotate(count=models.Count("id")).filter(count__gt=1)
-
-    affected_snapshot_ids = set()
-    for group in duplicate_groups.iterator(chunk_size=200):
-        rows = list(
-            ArchiveResult.objects.filter(
-                snapshot_id=group["snapshot_id"],
-                plugin=group["plugin"],
-            ).order_by("created_at", "id"),
-        )
-        winner = max(
-            rows,
-            key=lambda row: (
-                bool(row.output_files),
-                int(row.output_size or 0),
-                row.modified_at,
-                str(row.id),
-            ),
-        )
-        output_files = {}
-        mimetypes = set()
-        for row in rows:
-            output_files.update(row.output_files or {})
-            mimetypes.update(part.strip() for part in (row.output_mimetypes or "").split(",") if part.strip())
-
-        winner.output_files = output_files
-        winner.output_size = max(
-            sum(int(metadata.get("size") or 0) for metadata in output_files.values() if isinstance(metadata, dict)),
-            *(int(row.output_size or 0) for row in rows),
-        )
-        winner.output_mimetypes = ",".join(sorted(mimetypes))
-        winner.start_ts = min((row.start_ts for row in rows if row.start_ts), default=None)
-        winner.end_ts = max((row.end_ts for row in rows if row.end_ts), default=None)
-        winner.save(update_fields=["output_files", "output_size", "output_mimetypes", "start_ts", "end_ts"])
-        ArchiveResult.objects.filter(id__in=[row.id for row in rows if row.id != winner.id]).delete()
-        affected_snapshot_ids.add(group["snapshot_id"])
-
-    for snapshot_id in affected_snapshot_ids:
-        total = ArchiveResult.objects.filter(snapshot_id=snapshot_id).aggregate(total=Sum("output_size"))["total"] or 0
-        Snapshot.objects.filter(id=snapshot_id).update(output_size=total)
-
-
-class Migration(migrations.Migration):
-    dependencies = [
-        ("core", "0051_postgres_url_pattern_ops_index"),
-    ]
-
-    operations = [
-        migrations.RemoveConstraint(
-            model_name="archiveresult",
-            name="unique_archiveresult_per_snapshot_hook",
-        ),
-        migrations.RunPython(
-            consolidate_archiveresults_per_plugin,
-            reverse_code=migrations.RunPython.noop,
-        ),
-        migrations.AddConstraint(
-            model_name="archiveresult",
-            constraint=models.UniqueConstraint(
-                fields=("snapshot", "plugin"),
-                name="unique_archiveresult_per_snapshot_plugin",
-            ),
-        ),
-    ]

archivebox/core/models.py

@@ -1740,8 +1740,8 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW
                     self.add_tag_ids([tag.pk])
 
     def _merge_archive_results_from_index(self, index_data: dict, update_existing: bool = True):
-        """Merge ArchiveResults one row per plugin; retries update the existing row."""
-        existing = {ar.plugin: ar for ar in ArchiveResult.objects.filter(snapshot=self)}
+        """Merge ArchiveResults one row per hook; retries update the existing row."""
+        existing = {(ar.plugin, ar.hook_name): ar for ar in ArchiveResult.objects.filter(snapshot=self)}
         if update_existing:
             for archiveresult in existing.values():
                 normalized_status = ArchiveResult.normalize_status(archiveresult.status)
@@ -1804,7 +1804,7 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW
         output_mimetypes = result_data.get("output_mimetypes", "")
 
         hook_name = result_data.get("hook_name", "")
-        existing_result = existing.get(plugin)
+        existing_result = existing.get((plugin, hook_name))
         if existing_result:
             if not update_existing:
                 return
@@ -1813,9 +1813,6 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW
             if existing_result.status != status:
                 existing_result.status = status
                 update_fields.append("status")
-            if hook_name and existing_result.hook_name != hook_name:
-                existing_result.hook_name = hook_name
-                update_fields.append("hook_name")
             if output_str and existing_result.output_str != output_str:
                 existing_result.output_str = output_str
                 update_fields.append("output_str")
@@ -1872,7 +1869,7 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW
                 end_ts=end_ts,
                 process=process,
             )
-        existing[plugin] = archiveresult
+        existing[(plugin, hook_name)] = archiveresult
 
     def write_index_json(self):
         """Write index.json in 0.9.x format (deprecated, use write_index_jsonl)."""
@@ -2202,15 +2199,18 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW
             for tag in dup.tags.all():
                 keeper.add_tag_ids([tag.pk])
 
-            # Move ArchiveResults while preserving the one-row-per-plugin invariant.
+            # Move each hook result, merging only an exact identity collision.
             for result in ArchiveResult.objects.filter(snapshot=dup):
-                existing = ArchiveResult.objects.filter(snapshot=keeper, plugin=result.plugin).first()
+                existing = ArchiveResult.objects.filter(
+                    snapshot=keeper,
+                    plugin=result.plugin,
+                    hook_name=result.hook_name,
+                ).first()
                 if existing is None:
                     result.snapshot = keeper
                     result.save(update_fields=["snapshot", "modified_at"])
                     continue
-                prefer_result = result.output_size > existing.output_size
-                prior_output_sizes = (existing.output_size, result.output_size)
+
                 output_files = {**(existing.output_files or {}), **(result.output_files or {})}
                 existing.output_files = output_files
                 existing.output_size = max(
@@ -2219,13 +2219,15 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW
                         for metadata in output_files.values()
                         if isinstance(metadata, dict)
                     ),
-                    *prior_output_sizes,
+                    existing.output_size,
+                    result.output_size,
                 )
-                if prefer_result:
+                if result.modified_at >= existing.modified_at:
                     existing.status = result.status
-                    existing.hook_name = result.hook_name
                     existing.output_str = result.output_str
                     existing.output_json = result.output_json
+                    existing.start_ts = result.start_ts
+                    existing.end_ts = result.end_ts
                 existing.output_mimetypes = ",".join(
                     sorted(
                         {
@@ -2976,8 +2978,8 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW
         - abx_plugins/plugins/*/on_Snapshot__*.{py,sh,js}
         - data/custom_plugins/*/on_Snapshot__*.{py,sh,js}
 
-        Creates one ArchiveResult per plugin. A queued plugin row runs all of that
-        plugin's Snapshot hooks; hook_name records the most recently completed hook.
+        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.
         """
         try:
             self.validate_url_for_archiving()
@@ -2995,21 +2997,17 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW
             hooks = ((hook_path.parent.name, hook_path.stem) for hook_path in discover_hooks("Snapshot", config=config))
         archiveresults = []
 
-        for plugin in dict.fromkeys(plugin for plugin, _hook_name in hooks):
-            # ArchiveResult output is one filesystem directory per plugin, so all
-            # hooks and retries update this row in place instead of creating siblings.
+        for plugin, hook_name in hooks:
+            # Hooks in one plugin share a filesystem directory, but each hook has
+            # its own durable result row and retries update that exact row.
             archiveresult, _created = ArchiveResult.objects.get_or_create(
                 snapshot=self,
                 plugin=plugin,
+                hook_name=hook_name,
                 defaults={
-                    "hook_name": "",
                     "status": ArchiveResult.INITIAL_STATE,
                 },
             )
-            if archiveresult.hook_name == self.BROWSER_EXTENSION_UPLOAD_HOOK_NAME:
-                archiveresult.hook_name = ""
-                archiveresult.status = ArchiveResult.INITIAL_STATE
-                archiveresult.save(update_fields=["hook_name", "status", "modified_at"])
             if archiveresult.status == ArchiveResult.INITIAL_STATE:
                 archiveresults.append(archiveresult)
 
@@ -3098,10 +3096,10 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW
             snapshot=self,
             status=ArchiveResult.StatusChoices.FAILED,
         )
+        legacy_result_count = retryable_results.filter(hook_name="").count()
         now = timezone.now()
-        count = retryable_results.update(
+        count = retryable_results.exclude(hook_name="").update(
             status=ArchiveResult.StatusChoices.QUEUED,
-            hook_name="",
             output_str="",
             output_json=None,
             output_files={},
@@ -3112,11 +3110,11 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW
             modified_at=now,
         )
 
-        if count > 0:
+        if count + legacy_result_count > 0:
             self.refresh_from_db(fields=["modified_at", "retry_at", "status"])
             self.queue_for_extraction(when=now)
 
-        return count
+        return count + legacy_result_count
 
     # =========================================================================
     # URL Helper Properties (migrated from Link schema)
@@ -3966,7 +3964,7 @@ class ArchiveResult(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithNotes):
             models.Index(fields=["-start_ts", "-id"], name="archiveresult_start_idx"),
         ]
         constraints: ClassVar[list[models.BaseConstraint]] = [
-            models.UniqueConstraint(fields=["snapshot", "plugin"], name="unique_archiveresult_per_snapshot_plugin"),
+            models.UniqueConstraint(fields=["snapshot", "plugin", "hook_name"], name="unique_archiveresult_per_snapshot_hook"),
         ]
 
     def __str__(self):
@@ -4077,16 +4075,16 @@ class ArchiveResult(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithNotes):
             except ArchiveResult.DoesNotExist:
                 pass
 
-        # Get or create by snapshot_id + plugin. The filesystem has one output
-        # directory per plugin, so every hook and retry updates that same DB row.
+        # Get or create by the durable scheduler identity. Hooks in one plugin
+        # share an output directory, while retries update only their exact row.
         try:
             snapshot = Snapshot.objects.get(id=snapshot_id)
 
             result, _ = ArchiveResult.objects.get_or_create(
                 snapshot=snapshot,
                 plugin=plugin,
+                hook_name=record.get("hook_name", ""),
                 defaults={
-                    "hook_name": record.get("hook_name", ""),
                     "status": record.get("status", "queued"),
                     "output_str": record.get("output_str", ""),
                 },
@@ -4153,10 +4151,13 @@ class ArchiveResult(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithNotes):
     def schedule_delete_cleanup(self, *, using: str | None = None) -> None:
         """Remove shared plugin output and refresh persisted Snapshot metadata after commit."""
         snapshot_id = self.snapshot_id
+        plugin = self.plugin
         paths = self.validate_output_paths_for_delete(self.output_paths_for_delete())
 
         def cleanup() -> None:
-            type(self).delete_output_paths(paths)
+            results = type(self).objects.using(using) if using else type(self).objects
+            if not results.filter(snapshot_id=snapshot_id, plugin=plugin).exists():
+                type(self).delete_output_paths(paths)
             type(self).refresh_snapshot_output_sizes({snapshot_id})
             snapshot = Snapshot.objects.filter(pk=snapshot_id).first()
             if snapshot:
@@ -4190,7 +4191,6 @@ class ArchiveResult(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithNotes):
 
     def reset_for_retry(self, *, save: bool = True) -> None:
         self.status = self.StatusChoices.QUEUED
-        self.hook_name = ""
         self.retry_at = None
         self.output_str = ""
         self.output_json = None
@@ -4203,7 +4203,6 @@ class ArchiveResult(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithNotes):
             self.save(
                 update_fields=[
                     "status",
-                    "hook_name",
                     "retry_at",
                     "output_str",
                     "output_json",

archivebox/core/recovery_util.py

@@ -10,6 +10,11 @@ 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 _canonical_hook_name(hook_name: str) -> str:
+    hook_name = Path(hook_name).name
+    return Path(hook_name).stem if Path(hook_name).suffix in {".py", ".js", ".sh"} else hook_name
+
+
 def recover_orchestrator_state(*, include_chrome: bool = False, crawl_id: str | None = None) -> dict[str, int]:
     from archivebox.crawls.models import Crawl
     from archivebox.core.models import ArchiveResult, Snapshot
@@ -168,14 +173,23 @@ def recover_orchestrator_state(*, include_chrome: bool = False, crawl_id: str |
             orphaned_hook_processes = orphaned_hook_processes.filter(snapshot_pwd_filter)
         else:
             orphaned_hook_processes = orphaned_hook_processes.none()
-    for process in orphaned_hook_processes.only("id", "pwd", "cmd", "process_type", "status", "started_at", "ended_at").order_by(
-        "-started_at",
-        "-id",
-    ):
+    for process in orphaned_hook_processes.only(
+        "id",
+        "pwd",
+        "cmd",
+        "process_type",
+        "status",
+        "exit_code",
+        "stdout",
+        "stderr",
+        "started_at",
+        "ended_at",
+    ).order_by("-started_at", "-id"):
         hook_script_name = process.hook_script_name
         if not hook_script_name or not process.pwd:
             continue
         plugin_dir = Path(process.pwd)
+        hook_name = _canonical_hook_name(hook_script_name)
         if crawl_snapshot_ids is not None and plugin_dir.parent.name not in crawl_snapshot_ids:
             continue
         try:
@@ -190,26 +204,26 @@ def recover_orchestrator_state(*, include_chrome: bool = False, crawl_id: str |
         result, created = ArchiveResult.objects.get_or_create(
             snapshot=snapshot,
             plugin=plugin_dir.name,
+            hook_name=hook_name,
             defaults={
-                "hook_name": Path(hook_script_name).stem,
                 "status": ArchiveResult.StatusChoices.QUEUED,
             },
         )
-        if created or result.status == ArchiveResult.StatusChoices.QUEUED:
+        process_is_newer = bool(process.started_at and (result.start_ts is None or process.started_at >= result.start_ts))
+        if result.status == ArchiveResult.StatusChoices.QUEUED or process_is_newer:
             requeue_snapshot = False
             # A runner can die after the hook Process exits but before the
             # ProcessCompletedEvent projector links/finalizes ArchiveResult.
-            # Reconstruct the plugin row from its newest durable Process row.
+            # Reconstruct only that exact hook row from the durable Process row.
             output_files, output_size, output_mimetypes = _collect_output_metadata(plugin_dir)
             emitted_records = [
                 record
                 for record in Process.parse_records_from_text(process.stdout or "")
                 if record.get("type") == "ArchiveResult"
                 and (record.get("plugin") or plugin_dir.name) == plugin_dir.name
-                and (record.get("hook_name") or Path(hook_script_name).stem) == Path(hook_script_name).stem
+                and _canonical_hook_name(str(record.get("hook_name") or hook_name)) == hook_name
             ]
             emitted_result = emitted_records[-1] if emitted_records else {}
-            result.hook_name = Path(hook_script_name).stem
             result.process = process
             result.start_ts = process.started_at
             result.end_ts = process.ended_at
@@ -222,6 +236,7 @@ def recover_orchestrator_state(*, include_chrome: bool = False, crawl_id: str |
                 result.output_size = 0
                 result.output_mimetypes = ""
                 result.output_str = ""
+                result.output_json = None
                 result.status = ArchiveResult.StatusChoices.QUEUED
                 requeue_snapshot = True
             else:
@@ -246,7 +261,6 @@ def recover_orchestrator_state(*, include_chrome: bool = False, crawl_id: str |
                 )
             result.save(
                 update_fields=[
-                    "hook_name",
                     "process",
                     "start_ts",
                     "end_ts",

archivebox/core/views.py

@@ -847,26 +847,27 @@ def _resolve_archiveresult_relpath(snapshot: Snapshot, rel_path: str) -> tuple[s
 
     plugin = parts[0]
     plugin_relpath = posixpath.join(*parts[1:])
-    result = (
-        ArchiveResult.objects.filter(snapshot=snapshot, plugin=plugin, status=ArchiveResult.StatusChoices.SUCCEEDED)
-        .only("plugin", "output_files")
-        .first()
+    results = list(
+        ArchiveResult.objects.filter(
+            snapshot=snapshot,
+            plugin=plugin,
+            status=ArchiveResult.StatusChoices.SUCCEEDED,
+        ).only("plugin", "output_files"),
     )
-    if not result:
+    if not results:
         return rel_path, None
-    if not result.output_files:
-        return rel_path, result
 
-    output_files = result.output_files or {}
-    for candidate in (plugin_relpath, rel_path):
-        file_info = output_files.get(candidate)
-        if not isinstance(file_info, dict):
-            continue
-        if file_info.get("root_relative"):
-            return candidate, result
-        return rel_path, result
+    for result in results:
+        output_files = result.output_files or {}
+        for candidate in (plugin_relpath, rel_path):
+            file_info = output_files.get(candidate)
+            if not isinstance(file_info, dict):
+                continue
+            if file_info.get("root_relative"):
+                return candidate, result
+            return rel_path, result
 
-    return rel_path, result
+    return rel_path, results[0]
 
 
 def _plugin_full_preview_response(

archivebox/services/archive_result_service.py

@@ -4,7 +4,6 @@ import asyncio
 import inspect
 import json
 import os
-import re
 import signal
 import sys
 import time
@@ -203,11 +202,6 @@ def _normalize_status(status: str) -> str:
     return status or "failed"
 
 
-def _snapshot_hook_order(hook_name: str) -> int:
-    match = re.match(r"^on_Snapshot__(\d+)", hook_name or "")
-    return int(match.group(1)) if match else -1
-
-
 def _normalize_snapshot_title(candidate: str, *, snapshot_url: str) -> str:
     title = " ".join(line.strip() for line in str(candidate or "").splitlines() if line.strip()).strip()
     if not title:
@@ -325,7 +319,6 @@ def _save_archiveresult_event_to_db(
         start_ts = parse_event_datetime(event.start_ts)
         end_ts = parse_event_datetime(event.end_ts) or timezone.now()
         defaults = {
-            "hook_name": event.hook_name,
             "status": _normalize_status(event.status),
             "output_str": event.output_str,
             "output_json": event.output_json,
@@ -344,6 +337,7 @@ def _save_archiveresult_event_to_db(
         result = ArchiveResult.objects.filter(
             snapshot=snapshot,
             plugin=event.plugin,
+            hook_name=event.hook_name,
         ).first()
     if result is None:
         try:
@@ -351,6 +345,7 @@ def _save_archiveresult_event_to_db(
                 result = ArchiveResult.objects.create(
                     snapshot=snapshot,
                     plugin=event.plugin,
+                    hook_name=event.hook_name,
                     **defaults,
                 )
         except IntegrityError:
@@ -358,30 +353,9 @@ def _save_archiveresult_event_to_db(
                 result = ArchiveResult.objects.get(
                     snapshot=snapshot,
                     plugin=event.plugin,
+                    hook_name=event.hook_name,
                 )
 
-    if result.output_files:
-        merged_output_files = {**result.output_files, **defaults["output_files"]}
-        defaults["output_files"] = merged_output_files
-        defaults["output_size"], defaults["output_mimetypes"] = _summarize_output_files(merged_output_files)
-        defaults["output_size"] = max(defaults["output_size"], int(result.output_size or 0))
-        defaults["output_mimetypes"] = ",".join(
-            dict.fromkeys(
-                mimetype.strip()
-                for value in (defaults["output_mimetypes"], result.output_mimetypes)
-                for mimetype in value.split(",")
-                if mimetype.strip()
-            ),
-        )
-    if result.start_ts and defaults["start_ts"]:
-        defaults["start_ts"] = min(result.start_ts, defaults["start_ts"])
-    if result.end_ts and defaults["end_ts"]:
-        defaults["end_ts"] = max(result.end_ts, defaults["end_ts"])
-    if _snapshot_hook_order(result.hook_name) > _snapshot_hook_order(event.hook_name):
-        for field in ("hook_name", "status", "output_str", "output_json", "process_id", "notes"):
-            if field in result.__dict__:
-                defaults[field] = result.__dict__[field]
-
     with _perf_span("archivebox.ArchiveResultService.on_ArchiveResultEvent.diff_fields"):
         update_fields = []
         for field, value in defaults.items():
@@ -422,7 +396,7 @@ def _save_archiveresult_event_to_db(
 
 
 def mark_archiveresult_started(event: ProcessStartedEvent, *, snapshot_id: str, process_id: str) -> None:
-    """Advance an existing queued plugin row after its OS process is persisted."""
+    """Advance an existing queued hook row after its OS process is persisted."""
     from archivebox.core.models import ArchiveResult
 
     started_at = parse_event_datetime(event.start_ts)
@@ -431,9 +405,9 @@ def mark_archiveresult_started(event: ProcessStartedEvent, *, snapshot_id: str,
     ArchiveResult.objects.filter(
         snapshot_id=snapshot_id,
         plugin=event.plugin_name,
+        hook_name=event.hook_name,
         status=ArchiveResult.StatusChoices.QUEUED,
     ).update(
-        hook_name=event.hook_name,
         status=ArchiveResult.StatusChoices.STARTED,
         start_ts=started_at,
         end_ts=None,
@@ -448,7 +422,7 @@ class ArchiveResultService(BaseService):
 
     def __init__(self, bus):
         self._completed_process_event_ids: set[str] = set()
-        self._save_locks: dict[tuple[str, str], asyncio.Lock] = {}
+        self._save_locks: dict[tuple[str, str, str], asyncio.Lock] = {}
         super().__init__(bus)
         self.bus.on(ArchiveResultEvent, self.on_ArchiveResultEvent__save_to_db)
         self.bus.on(ProcessCompletedEvent, self.on_ProcessCompletedEvent__save_to_db)
@@ -463,7 +437,7 @@ class ArchiveResultService(BaseService):
                 where=lambda candidate: self.bus.event_is_child_of(event, candidate),
             )
 
-        key = (str(event.snapshot_id), event.plugin)
+        key = (str(event.snapshot_id), event.plugin, event.hook_name)
         lock = self._save_locks.setdefault(key, asyncio.Lock())
         async with lock:
             await sync_to_async(_save_archiveresult_event_to_db, thread_sensitive=True)(event, process_started)

archivebox/services/runner.py

@@ -1038,7 +1038,7 @@ class CrawlRunner:
                 )
                 return [plugin for plugin in queued_plugins if plugin in expanded_selected_plugins]
 
-            queued_plugins = None
+            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:
@@ -1056,8 +1056,8 @@ class CrawlRunner:
                     )
                     return
                 if not self.selected_plugins_from_args:
-                    queued_plugins = await sync_to_async(
-                        queued_plugins_for_snapshot,
+                    queued_plugins, selected_hooks_by_plugin = await sync_to_async(
+                        queued_plugins_and_hooks_for_snapshot,
                         thread_sensitive=True,
                     )(snapshot["id"])
                     if queued_plugins:
@@ -1070,10 +1070,13 @@ class CrawlRunner:
                                     snapshot["id"],
                                     disabled_queued_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
             elif not self.selected_plugins_from_args:
-                queued_plugins = await sync_to_async(
-                    queued_plugins_for_snapshot,
+                queued_plugins, selected_hooks_by_plugin = await sync_to_async(
+                    queued_plugins_and_hooks_for_snapshot,
                     thread_sensitive=True,
                 )(snapshot["id"])
                 if queued_plugins:
@@ -1086,6 +1089,9 @@ class CrawlRunner:
                                 snapshot["id"],
                                 disabled_queued_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",
@@ -1100,10 +1106,10 @@ class CrawlRunner:
                 if snapshot_selected_plugins
                 else self.plugins
             )
-            if queued_plugins is not None:
-                await sync_to_async(fail_unavailable_queued_plugins, thread_sensitive=True)(
+            if selected_hooks_by_plugin is not None:
+                await sync_to_async(fail_unavailable_queued_hooks, thread_sensitive=True)(
                     snapshot["id"],
-                    queued_plugins,
+                    selected_hooks_by_plugin,
                     plugins,
                 )
                 remaining_queued_plugins = await sync_to_async(
@@ -1123,6 +1129,7 @@ class CrawlRunner:
                     return
                 snapshot_selected_plugins = remaining_queued_plugins
                 plugins = filter_plugins(self.plugins, snapshot_selected_plugins, include_providers=True)
+                selected_hooks_by_plugin = include_background_prerequisite_hooks(selected_hooks_by_plugin, plugins)
             abx_snapshot = AbxSnapshot(
                 id=snapshot["id"],
                 url=snapshot["url"],
@@ -1146,7 +1153,7 @@ class CrawlRunner:
                 snapshot_cleanup_enabled=True,
                 snapshot_cleanup_phase_timeout=snapshot_phase_timeout,
                 abort_requested=self.crawl_is_cancelled,
-                selected_hooks_by_plugin=None,
+                selected_hooks_by_plugin=selected_hooks_by_plugin,
             )
             try:
                 snapshot_event = SnapshotEvent(
@@ -1368,19 +1375,37 @@ def run_binary(binary_id: str) -> None:
     asyncio.run(_run_binary(binary_id))
 
 
-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_plugins = list(
+    queued_results = list(
         ArchiveResult.objects.filter(
             snapshot_id=snapshot_id,
             status=ArchiveResult.StatusChoices.QUEUED,
         )
         .exclude(plugin="")
-        .order_by("plugin")
-        .values_list("plugin", flat=True),
+        .only("id", "plugin", "hook_name"),
     )
-    return queued_plugins or None
+
+    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, 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 config_overrides_for_queued_plugins(selected_plugins: list[str], **overrides: Any) -> dict[str, Any]:
@@ -1397,27 +1422,40 @@ def config_overrides_for_queued_plugins(selected_plugins: list[str], **overrides
     return config_overrides
 
 
-def fail_unavailable_queued_plugins(
+def fail_unavailable_queued_hooks(
     snapshot_id: str,
-    queued_plugins: list[str],
+    selected_hooks_by_plugin: dict[str, set[str] | None],
     plugins: dict[str, Plugin],
 ) -> None:
     from archivebox.core.models import ArchiveResult
 
-    unavailable_plugins = set(queued_plugins) - set(plugins)
-    if not unavailable_plugins:
-        return
     now = timezone.now()
-    ArchiveResult.objects.filter(
-        snapshot_id=snapshot_id,
-        plugin__in=unavailable_plugins,
-        status=ArchiveResult.StatusChoices.QUEUED,
-    ).update(
-        status=ArchiveResult.StatusChoices.FAILED,
-        start_ts=now,
-        end_ts=now,
-        output_str="Queued plugin is not available in this ArchiveBox installation",
-    )
+    for plugin_name, selected_hook_names in selected_hooks_by_plugin.items():
+        if selected_hook_names is None:
+            continue
+        if plugin_name in plugins:
+            available_hook_names = {
+                name for hook in plugins[plugin_name].filter_hooks("Snapshot") for name in (hook.name, Path(hook.name).stem)
+            }
+        else:
+            available_hook_names = set()
+        missing_hook_names = [hook_name for hook_name in selected_hook_names if hook_name not in available_hook_names]
+        if not missing_hook_names:
+            continue
+        # Hook-level resume rows are durable scheduler state. If a plugin is
+        # installed but no longer exposes a queued hook, mark that row failed so
+        # the snapshot is not retried forever with no hook left to execute.
+        ArchiveResult.objects.filter(
+            snapshot_id=snapshot_id,
+            plugin=plugin_name,
+            hook_name__in=missing_hook_names,
+            status=ArchiveResult.StatusChoices.QUEUED,
+        ).update(
+            status=ArchiveResult.StatusChoices.FAILED,
+            start_ts=now,
+            end_ts=now,
+            output_str="Queued hook is no longer available in the installed plugin",
+        )
 
 
 def skip_disabled_queued_plugins(snapshot_id: str, plugin_names: list[str]) -> None:
@@ -1442,6 +1480,35 @@ def skip_disabled_queued_plugins(snapshot_id: str, plugin_names: list[str]) -> N
     )
 
 
+def include_background_prerequisite_hooks(
+    selected_hooks_by_plugin: dict[str, set[str] | None],
+    plugins: dict[str, Plugin],
+) -> dict[str, set[str] | None]:
+    expanded: dict[str, set[str] | None] = {}
+    for plugin_name, selected_hook_names in selected_hooks_by_plugin.items():
+        if selected_hook_names is None or plugin_name not in plugins:
+            expanded[plugin_name] = selected_hook_names
+            continue
+        plugin_hooks = sorted(plugins[plugin_name].filter_hooks("Snapshot"), key=lambda hook: hook.sort_key)
+        selected_sort_keys = [
+            hook.sort_key for hook in plugin_hooks if hook.name in selected_hook_names or Path(hook.name).stem in selected_hook_names
+        ]
+        if not selected_sort_keys:
+            expanded[plugin_name] = set(selected_hook_names)
+            continue
+        first_selected_sort_key = min(selected_sort_keys)
+        expanded_hook_names = set(selected_hook_names)
+        # Earlier background hooks publish live resources (e.g. Chrome tabs)
+        # needed by later foreground hooks, but completed foreground hooks stay
+        # final and are not rerun during hook-level resume.
+        for hook in plugin_hooks:
+            if hook.is_background and hook.sort_key < first_selected_sort_key:
+                expanded_hook_names.add(hook.name)
+                expanded_hook_names.add(Path(hook.name).stem)
+        expanded[plugin_name] = expanded_hook_names
+    return expanded
+
+
 def snapshot_hooks_for_pending_archiveresults(snapshot) -> list[tuple[str, str]]:
     from archivebox.config.common import get_config
     from archivebox.core.models import Snapshot
@@ -1705,10 +1772,10 @@ def _run_due_snapshot_locked(snapshot, *, lock_seconds: int, interactive_interru
         return False
     snapshot.refresh_from_db()
     if snapshot.status == Snapshot.StatusChoices.QUEUED:
-        has_server_workset = snapshot.archiveresult_set.exclude(
+        has_server_archiveresults = snapshot.archiveresult_set.exclude(
             hook_name=Snapshot.BROWSER_EXTENSION_UPLOAD_HOOK_NAME,
         ).exists()
-        if has_server_workset and snapshot.is_finished_processing():
+        if has_server_archiveresults and snapshot.is_finished_processing():
             finalize_completed_snapshot(str(snapshot.id), output_dir=Path(snapshot.output_dir))
             snapshot.refresh_from_db()
             if snapshot.status == Snapshot.StatusChoices.SEALED:
@@ -1717,10 +1784,10 @@ def _run_due_snapshot_locked(snapshot, *, lock_seconds: int, interactive_interru
                     snapshot.refresh_from_db()
                 _runner_console_line(crawl_id=snapshot.crawl_id, snapshot=snapshot, status="SEALED")
                 return True
-        if not has_server_workset:
-            # Materialize one durable row per configured plugin. Existing
-            # browser-uploaded rows are reused and queued so the server adds its
-            # outputs to that plugin result instead of creating a sibling row.
+        # A Snapshot with no server hook rows is fresh lifecycle work; materialize
+        # its configured hook set. Browser-extension uploads are completed external
+        # outputs, not the durable server workset, and must coexist with these rows.
+        if not has_server_archiveresults:
             snapshot.create_pending_archiveresults(hooks=snapshot_hooks_for_pending_archiveresults(snapshot))
         snapshot.sm.tick()
         snapshot.refresh_from_db()
@@ -1737,11 +1804,11 @@ def _run_due_snapshot_locked(snapshot, *, lock_seconds: int, interactive_interru
             _runner_console_line(crawl_id=snapshot.crawl_id, snapshot=snapshot, status="SEALED")
             return True
     if snapshot.status == Snapshot.StatusChoices.STARTED:
-        queued_plugins = queued_plugins_for_snapshot(str(snapshot.id))
-        if queued_plugins:
-            fail_unavailable_queued_plugins(
+        queued_plugins, selected_hooks_by_plugin = queued_plugins_and_hooks_for_snapshot(str(snapshot.id))
+        if queued_plugins and selected_hooks_by_plugin:
+            fail_unavailable_queued_hooks(
                 str(snapshot.id),
-                queued_plugins,
+                selected_hooks_by_plugin,
                 _discover_archivebox_plugins(),
             )
             if not queued_plugins_for_snapshot(str(snapshot.id)):
@@ -1765,10 +1832,10 @@ def _run_due_snapshot_locked(snapshot, *, lock_seconds: int, interactive_interru
     )
     snapshot.refresh_from_db()
     if queued_plugins_for_snapshot(str(snapshot.id)):
-        # Plugin-level resume work is tracked by queued ArchiveResult rows, not by
+        # 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 plugins.
+        # active-state lock before running the remaining hooks.
         snapshot.update_and_requeue(retry_at=timezone.now())
     return True
 

archivebox/tests/test_api_v1_core_archiveresults.py

@@ -14,7 +14,7 @@ from archivebox.tests.conftest import api_client_request
 pytestmark = pytest.mark.django_db(transaction=True)
 
 
-def test_archiveresult_upload_upserts_one_snapshot_plugin_result(client, api_admin_user, api_headers):
+def test_archiveresult_upload_upserts_by_snapshot_plugin_and_hook(client, api_admin_user, api_headers):
     crawl = Crawl.objects.create(urls="https://example.com", created_by=api_admin_user)
     snapshot = Snapshot.objects.create(url="https://example.com/unified-result", crawl=crawl)
 
@@ -39,13 +39,14 @@ def test_archiveresult_upload_upserts_one_snapshot_plugin_result(client, api_adm
     assert server_response.status_code == 200, server_response.content
     assert extension_update.status_code == 200, extension_update.content
     assert extension_update.json()["id"] == extension_response.json()["id"]
-    assert server_response.json()["id"] == extension_response.json()["id"]
+    assert server_response.json()["id"] != extension_response.json()["id"]
 
-    results = ArchiveResult.objects.filter(snapshot=snapshot, plugin="screenshot")
-    assert results.count() == 1
-    result = results.get()
-    assert result.hook_name == "on_Snapshot__archivebox_browser_extension_upload"
-    assert set(result.output_files) == {"browser.png", "server.png", "browser-2.png"}
+    results = ArchiveResult.objects.filter(snapshot=snapshot, plugin="screenshot").order_by("hook_name")
+    assert results.count() == 2
+    extension_result = results.get(hook_name="on_Snapshot__archivebox_browser_extension_upload")
+    server_result = results.get(hook_name="on_Snapshot__50_screenshot")
+    assert set(extension_result.output_files) == {"browser.png", "browser-2.png"}
+    assert set(server_result.output_files) == {"server.png"}
 
 
 def test_archiveresult_upload_api_queues_snapshot_maintenance_without_finalizing(client, api_admin_user, api_headers):

archivebox/tests/test_archive_result_service.py

@@ -230,7 +230,7 @@ def test_archiveresult_event_retry_updates_existing_hook_row(tmp_path, hermetic_
     _cleanup_machine_process_rows()
 
 
-def test_archiveresult_duplicate_plugin_rows_are_rejected():
+def test_archiveresult_duplicate_hook_rows_are_rejected():
     from django.db import IntegrityError, transaction
     from archivebox.core.models import ArchiveResult
 
@@ -246,47 +246,11 @@ def test_archiveresult_duplicate_plugin_rows_are_rejected():
         ArchiveResult.objects.create(
             snapshot=snapshot,
             plugin="wget",
-            hook_name="on_Snapshot__99_other_wget_hook",
+            hook_name="on_Snapshot__06_wget.finite.bg",
             status=ArchiveResult.StatusChoices.SUCCEEDED,
         )
 
 
-def test_archivewebpage_lifecycle_hooks_project_one_plugin_output():
-    from abx_dl.events import ArchiveResultEvent
-    from archivebox.core.models import ArchiveResult
-    from archivebox.services.archive_result_service import _save_archiveresult_event_to_db
-
-    snapshot = _create_snapshot()
-    _save_archiveresult_event_to_db(
-        ArchiveResultEvent(
-            snapshot_id=str(snapshot.id),
-            plugin="archivewebpage",
-            hook_name="on_Snapshot__16_archivewebpage_start",
-            status="succeeded",
-            output_str="recording started",
-            output_files=[OutputFile(path="recording.json", extension="json", mimetype="application/json", size=175)],
-        ),
-        None,
-    )
-    _save_archiveresult_event_to_db(
-        ArchiveResultEvent(
-            snapshot_id=str(snapshot.id),
-            plugin="archivewebpage",
-            hook_name="on_Snapshot__65_archivewebpage_stop",
-            status="succeeded",
-            output_str="archivewebpage.wacz",
-            output_files=[OutputFile(path="archivewebpage.wacz", extension="wacz", size=2048)],
-        ),
-        None,
-    )
-
-    result = ArchiveResult.objects.get(snapshot=snapshot, plugin="archivewebpage")
-    assert result.hook_name == "on_Snapshot__65_archivewebpage_stop"
-    assert result.output_str == "archivewebpage.wacz"
-    assert set(result.output_files) == {"recording.json", "archivewebpage.wacz"}
-    assert ArchiveResult.objects.filter(snapshot=snapshot, plugin="archivewebpage").count() == 1
-
-
 def test_process_completed_projects_failed_archiveresult_from_shipped_hook(tmp_path, hermetic_lib_dir):
     from archivebox.core.models import ArchiveResult
 
@@ -470,13 +434,12 @@ def test_retry_failed_archiveresults_requeues_snapshot_in_queued_state():
     reset_count = snapshot.retry_failed_archiveresults()
 
     snapshot.refresh_from_db()
-    result = ArchiveResult.objects.get(snapshot=snapshot, plugin="chrome")
+    result = ArchiveResult.objects.get(snapshot=snapshot, plugin="chrome", hook_name="on_Snapshot__11_chrome_wait")
     assert reset_count == 1
     assert snapshot.status == Snapshot.StatusChoices.QUEUED
     assert snapshot.retry_at is not None
     assert snapshot.current_step == 0
     assert result.status == ArchiveResult.StatusChoices.QUEUED
-    assert result.hook_name == ""
     assert result.output_str == ""
     assert result.output_json is None
     assert result.output_files == {}

archivebox/tests/test_cli_extract_input.py

@@ -22,7 +22,7 @@ def create_extract_snapshot(initialized_archive, env, url="https://example.com")
     )
 
 
-def test_extract_archiveresult_record_queues_one_plugin_result(initialized_archive):
+def test_extract_archiveresult_record_queues_only_exact_hook(initialized_archive):
     env = cli_env(PLUGINS="archivewebpage")
     create_extract_snapshot(initialized_archive, env)
 
@@ -52,7 +52,7 @@ def test_extract_archiveresult_record_queues_one_plugin_result(initialized_archi
         rows = list(
             ArchiveResult.objects.filter(snapshot_id=snapshot_id, plugin="archivewebpage").values_list("hook_name", "status"),
         )
-    assert rows == [("", ArchiveResult.StatusChoices.QUEUED)]
+    assert rows == [(hook_name, ArchiveResult.StatusChoices.QUEUED)]
 
 
 def test_extract_runs_on_snapshot_id(initialized_archive):

archivebox/tests/test_cli_run.py

@@ -240,7 +240,7 @@ class TestRunWithArchiveResult:
     """Tests for `archivebox run` with ArchiveResult input."""
 
     @pytest.mark.django_db(transaction=True)
-    def test_run_treats_no_id_archiveresult_request_as_plugin_work(self, initialized_archive):
+    def test_run_creates_and_runs_exact_no_id_archiveresult_request(self, initialized_archive):
         import json
 
         from archivebox.core.models import ArchiveResult
@@ -284,9 +284,13 @@ class TestRunWithArchiveResult:
                     "output_str",
                 ),
             )
-        assert len(rows) == 1
-        assert rows[0][0] != missing_hook
-        assert rows[0][1] in ArchiveResult.FINAL_STATES
+        assert rows == [
+            (
+                missing_hook,
+                ArchiveResult.StatusChoices.FAILED,
+                "Queued hook is no longer available in the installed plugin",
+            ),
+        ]
 
     def test_run_requeues_failed_archiveresult(self, initialized_archive):
         """Run re-queues a failed ArchiveResult."""
@@ -1019,7 +1023,7 @@ class TestRecoverOrchestratorState:
         assert snapshot.retry_at is None
         assert snapshot.downloaded_at is not None
 
-    def test_create_pending_archiveresults_creates_one_plugin_row_not_hook_rows(self):
+    def test_create_pending_archiveresults_uses_canonical_hook_names(self):
         from django.utils import timezone
 
         from archivebox.base_models.models import get_or_create_system_user_pk
@@ -1039,17 +1043,11 @@ class TestRecoverOrchestratorState:
             retry_at=timezone.now(),
         )
 
-        results = snapshot.create_pending_archiveresults(
-            hooks=[
-                ("chrome", "on_Snapshot__00_chrome_launch.daemon.bg"),
-                ("chrome", "on_Snapshot__01_chrome_tab.daemon.bg"),
-                ("chrome", "on_Snapshot__30_chrome_navigate"),
-                ("title", "on_Snapshot__54_title"),
-            ],
-        )
+        snapshot.create_pending_archiveresults()
 
-        assert [(result.plugin, result.hook_name) for result in results] == [("chrome", ""), ("title", "")]
-        assert ArchiveResult.objects.filter(snapshot=snapshot).count() == 2
+        hook_names = list(ArchiveResult.objects.filter(snapshot=snapshot).values_list("hook_name", flat=True))
+        assert hook_names
+        assert all(not hook_name.endswith((".py", ".js", ".sh")) for hook_name in hook_names)
 
     def test_snapshot_hooks_for_pending_archiveresults_respects_disabled_plugins_when_plugins_empty(self):
         from django.utils import timezone
@@ -1665,7 +1663,7 @@ class TestRecoverOrchestratorState:
         assert result.end_ts is not None
 
     @pytest.mark.django_db(transaction=True)
-    def test_run_due_snapshot_runs_plugin_for_obsolete_queued_hook_name(self):
+    def test_run_due_snapshot_fails_obsolete_queued_hook_name(self):
         from django.utils import timezone
 
         from archivebox.base_models.models import get_or_create_system_user_pk
@@ -1696,8 +1694,8 @@ class TestRecoverOrchestratorState:
 
         result.refresh_from_db()
         snapshot.refresh_from_db()
-        assert result.status in ArchiveResult.FINAL_STATES
-        assert result.hook_name != "on_Snapshot__50_singlefile.py"
+        assert result.status == ArchiveResult.StatusChoices.FAILED
+        assert result.output_str == "Queued hook is no longer available in the installed plugin"
         assert snapshot.retry_at is None
 
     @pytest.mark.django_db(transaction=True)
@@ -1776,10 +1774,7 @@ class TestRecoverOrchestratorState:
         recursive_test_site,
         chrome_isolation,
     ):
-        from pathlib import Path
-
-        from archivebox.core.models import ArchiveResult, Snapshot
-        from archivebox.machine.models import Process
+        from archivebox.core.models import ArchiveResult
         from archivebox.tests.test_orm_helpers import use_archivebox_db
 
         env = cli_env(disable_extractors=True)
@@ -1817,19 +1812,13 @@ class TestRecoverOrchestratorState:
         snapshot_id = navigate_record["snapshot_id"]
 
         with use_archivebox_db(initialized_archive):
-            snapshot = Snapshot.objects.get(id=snapshot_id)
-            tab_processes = [
-                process
-                for process in Process.objects.filter(pwd=str(Path(snapshot.output_dir) / "chrome")).order_by("-started_at")
-                if Path(process.hook_script_name or "").stem == "on_Snapshot__01_chrome_tab.daemon.bg"
-            ]
-            assert tab_processes
-            first_tab_process_id = tab_processes[0].id
-            plugin_result = ArchiveResult.objects.get(
+            tab_result = ArchiveResult.objects.get(
                 snapshot_id=snapshot_id,
                 plugin="chrome",
+                hook_name="on_Snapshot__01_chrome_tab.daemon.bg",
             )
-            assert plugin_result.hook_name == "on_Snapshot__30_chrome_navigate"
+            first_tab_process_id = tab_result.process_id
+            assert first_tab_process_id is not None
 
         update_process = run_archivebox_cmd(
             ["archiveresult", "update", "--status=queued"],
@@ -1863,18 +1852,18 @@ class TestRecoverOrchestratorState:
             navigate_result = ArchiveResult.objects.get(
                 snapshot_id=snapshot_id,
                 plugin="chrome",
+                hook_name="on_Snapshot__30_chrome_navigate",
+            )
+            tab_result = ArchiveResult.objects.get(
+                snapshot_id=snapshot_id,
+                plugin="chrome",
+                hook_name="on_Snapshot__01_chrome_tab.daemon.bg",
             )
-            snapshot = Snapshot.objects.get(id=snapshot_id)
-            tab_processes = [
-                process
-                for process in Process.objects.filter(pwd=str(Path(snapshot.output_dir) / "chrome")).order_by("-started_at")
-                if Path(process.hook_script_name or "").stem == "on_Snapshot__01_chrome_tab.daemon.bg"
-            ]
 
         assert run_process.returncode == 0
         assert navigate_result.status == ArchiveResult.StatusChoices.SUCCEEDED
-        assert navigate_result.hook_name == "on_Snapshot__30_chrome_navigate"
-        assert tab_processes[0].id != first_tab_process_id
+        assert tab_result.process_id is not None
+        assert tab_result.process_id != first_tab_process_id
 
     def test_recover_orchestrator_state_ignores_sealed_downloaded_snapshot_without_results(self):
         from django.utils import timezone

archivebox/tests/test_migration_archiveresult_unique_plugin.py

@@ -1,53 +0,0 @@
-import pytest
-from django.db import connection
-from django.db.migrations.executor import MigrationExecutor
-
-
-pytestmark = pytest.mark.django_db(transaction=True)
-
-
-def test_migration_consolidates_plugin_rows_and_enforces_uniqueness():
-    try:
-        executor = MigrationExecutor(connection)
-        executor.migrate([("core", "0051_postgres_url_pattern_ops_index")])
-        old_apps = executor.loader.project_state([("core", "0051_postgres_url_pattern_ops_index")]).apps
-        Crawl = old_apps.get_model("crawls", "Crawl")
-        Snapshot = old_apps.get_model("core", "Snapshot")
-        ArchiveResult = old_apps.get_model("core", "ArchiveResult")
-        crawl = Crawl.objects.create(urls="https://example.com")
-        snapshot = Snapshot.objects.create(url="https://example.com/migration", crawl=crawl, output_size=10)
-        ArchiveResult.objects.create(
-            snapshot=snapshot,
-            plugin="responses",
-            hook_name="browser-upload",
-            status="succeeded",
-            output_files={"browser.png": {"size": 7}},
-            output_size=7,
-            output_mimetypes="image/png",
-        )
-        ArchiveResult.objects.create(
-            snapshot=snapshot,
-            plugin="responses",
-            hook_name="server-capture",
-            status="noresults",
-            output_files={"server.json": {"size": 3}},
-            output_size=3,
-            output_mimetypes="application/json",
-        )
-
-        executor = MigrationExecutor(connection)
-        executor.migrate([("core", "0052_unique_archiveresult_per_snapshot_plugin")])
-        new_apps = executor.loader.project_state([("core", "0052_unique_archiveresult_per_snapshot_plugin")]).apps
-        Snapshot = new_apps.get_model("core", "Snapshot")
-        ArchiveResult = new_apps.get_model("core", "ArchiveResult")
-        snapshot = Snapshot.objects.get(id=snapshot.id)
-        result = ArchiveResult.objects.get(snapshot=snapshot, plugin="responses")
-
-        assert ArchiveResult.objects.filter(snapshot=snapshot, plugin="responses").count() == 1
-        assert result.status == "succeeded"
-        assert set(result.output_files) == {"browser.png", "server.json"}
-        assert result.output_size == 10
-        assert snapshot.output_size == 10
-        assert set(result.output_mimetypes.split(",")) == {"application/json", "image/png"}
-    finally:
-        MigrationExecutor(connection).migrate([("core", "0052_unique_archiveresult_per_snapshot_plugin")])

archivebox/tests/test_migrations_08_to_09.py

@@ -897,7 +897,10 @@ def test_update_preserves_legacy_plugin_directory_without_output_files(migration
         (snapshot["id"],),
     ).fetchall()
     conn.close()
-    assert rows == [("", "{}", "")]
+    assert len(rows) == 2
+    assert all(output_str == "" for output_str, _output_files, _hook_name in rows)
+    assert len({hook_name for _output_str, _output_files, hook_name in rows}) == 2
+    assert sum(not hook_name for _output_str, _output_files, hook_name in rows) == 1
 
 
 def test_07_filesystem_hop_preserves_complete_output_tree(tmp_path):

archivebox/tests/test_snapshot_service.py

@@ -40,7 +40,7 @@ def _snapshot_state(cwd: Path, url: str) -> dict[str, object]:
         }
 
 
-def test_snapshot_merge_consolidates_plugin_output_identity(admin_user):
+def test_snapshot_merge_consolidates_only_exact_hook_identity(admin_user):
     from archivebox.crawls.models import Crawl
 
     keeper_crawl = Crawl.objects.create(urls="https://example.com", created_by=admin_user)
@@ -64,15 +64,24 @@ def test_snapshot_merge_consolidates_plugin_output_identity(admin_user):
         output_files={"archivewebpage.wacz": {"size": 7}},
         output_size=7,
     )
+    ArchiveResult.objects.create(
+        snapshot=duplicate,
+        plugin="archivewebpage",
+        hook_name="on_Snapshot__16_archivewebpage_start",
+        status=ArchiveResult.StatusChoices.SUCCEEDED,
+        output_str="recording started",
+    )
+
     Snapshot._merge_snapshots([keeper, duplicate])
 
     assert not Snapshot.objects.filter(pk=duplicate.pk).exists()
     results = ArchiveResult.objects.filter(snapshot=keeper, plugin="archivewebpage")
-    assert results.count() == 1
-    stop_result = results.get()
+    assert results.count() == 2
+    stop_result = results.get(hook_name="on_Snapshot__65_archivewebpage_stop")
     assert stop_result.status == ArchiveResult.StatusChoices.SUCCEEDED
     assert stop_result.output_str == "archivewebpage.wacz"
     assert set(stop_result.output_files) == {"older.wacz", "archivewebpage.wacz"}
+    assert results.filter(hook_name="on_Snapshot__16_archivewebpage_start").exists()
 
 
 @pytest.mark.timeout(180)

archivebox/tests/test_ui_admin_archiveresult.py

@@ -142,8 +142,7 @@ class TestArchiveResultAdminListView:
         snapshot.refresh_from_db()
         assert snapshot.output_size == 0
 
-    def test_duplicate_plugin_result_is_rejected_without_touching_output(self, snapshot):
-        from django.db import IntegrityError, transaction
+    def test_deleting_sibling_hook_preserves_shared_plugin_output(self, snapshot):
         from archivebox.core.models import ArchiveResult
 
         output_dir = Path(snapshot.output_dir) / "responses"
@@ -157,13 +156,14 @@ class TestArchiveResultAdminListView:
             status=ArchiveResult.StatusChoices.SUCCEEDED,
             output_size=18,
         )
-        with pytest.raises(IntegrityError), transaction.atomic():
-            ArchiveResult.objects.create(
-                snapshot=snapshot,
-                plugin="responses",
-                hook_name="on_Snapshot__24_responses.daemon.bg.replayed",
-                status=ArchiveResult.StatusChoices.NORESULTS,
-            )
+        duplicate = ArchiveResult.objects.create(
+            snapshot=snapshot,
+            plugin="responses",
+            hook_name="on_Snapshot__24_responses.daemon.bg.replayed",
+            status=ArchiveResult.StatusChoices.NORESULTS,
+        )
+
+        duplicate.delete()
 
         assert ArchiveResult.objects.filter(pk=primary.pk).exists()
         assert output_file.read_text() == "captured response\n"

archivebox/tests/test_ui_admin_snapshot.py

@@ -794,10 +794,9 @@ class TestSnapshotOutputDeletion:
         from archivebox.core.models import ArchiveResult
 
         first = self._create_output(snapshot, size=11)
-        second = self._create_output(snapshot, plugin="singlefile", hook_name="on_Snapshot__50_singlefile.py", size=13)
+        second = self._create_output(snapshot, hook_name="on_Snapshot__51_screenshot_retry.py", size=13)
         kept = self._create_output(snapshot, plugin="pdf", hook_name="on_Snapshot__60_pdf.py", size=7)
         deleted_dir = Path(first.output_dir)
-        second_deleted_dir = Path(second.output_dir)
         kept_dir = Path(kept.output_dir)
         hashes_dir = Path(snapshot.output_dir) / "hashes"
         hashes_dir.mkdir(parents=True, exist_ok=True)
@@ -830,7 +829,6 @@ class TestSnapshotOutputDeletion:
         assert not ArchiveResult.objects.filter(pk__in=[first.pk, second.pk]).exists()
         assert ArchiveResult.objects.filter(pk=kept.pk).exists()
         assert not deleted_dir.exists()
-        assert not second_deleted_dir.exists()
         assert kept_dir.exists()
         snapshot.refresh_from_db()
         assert snapshot.output_size == 7

archivebox/tests/test_urls.py

@@ -15,22 +15,26 @@ REPO_ROOT = Path(__file__).resolve().parents[2]
 
 
 @pytest.mark.django_db
-def test_archiveresult_relpath_uses_plugin_result_that_owns_output(admin_user):
+def test_archiveresult_relpath_uses_sibling_hook_that_owns_output(admin_user):
     from archivebox.core.models import ArchiveResult, Snapshot
     from archivebox.core.views import _resolve_archiveresult_relpath
     from archivebox.crawls.models import Crawl
 
     crawl = Crawl.objects.create(urls="https://example.com", created_by=admin_user)
     snapshot = Snapshot.objects.create(url="https://example.com", crawl=crawl)
+    ArchiveResult.objects.create(
+        snapshot=snapshot,
+        plugin="screenshot",
+        hook_name="on_Snapshot__archivebox_browser_extension_upload",
+        status=ArchiveResult.StatusChoices.SUCCEEDED,
+        output_files={"browser.png": {"size": 7}},
+    )
     server_result = ArchiveResult.objects.create(
         snapshot=snapshot,
         plugin="screenshot",
         hook_name="on_Snapshot__50_screenshot",
         status=ArchiveResult.StatusChoices.SUCCEEDED,
-        output_files={
-            "browser.png": {"size": 7},
-            "screenshot.png": {"size": 6, "root_relative": True},
-        },
+        output_files={"screenshot.png": {"size": 6, "root_relative": True}},
     )
 
     resolved_path, result = _resolve_archiveresult_relpath(snapshot, "screenshot/screenshot.png")