diff --git a/archivebox/api/v1_core.py b/archivebox/api/v1_core.py
index c426d41d..cc7ec6f6 100644
--- a/archivebox/api/v1_core.py
+++ b/archivebox/api/v1_core.py
@@ -944,6 +944,9 @@ def create_snapshot(request: HttpRequest, data: SnapshotCreateSchema):
             created_by=request.user if isinstance(request.user, User) else None,
         )
 
+    # Browser uploads must not wait behind the runner's crawl-wide lifecycle
+    # lock. The unique insert recovery and CAS update below are the request-side
+    # coordination boundary for this idempotent metadata sync.
     snapshot = Snapshot.objects.filter(url=data.url, crawl=crawl).first()
     if snapshot is None:
         try:
@@ -973,7 +976,10 @@ def create_snapshot(request: HttpRequest, data: SnapshotCreateSchema):
         raise HttpError(409, "Snapshot changed while metadata was being updated")
 
     if tags:
-        snapshot.save_tags(tags)
+        snapshot.save_tags(
+            tags,
+            created_by=request.user if isinstance(request.user, User) else None,
+        )
 
     try:
         snapshot.ensure_crawl_symlink()
@@ -1026,7 +1032,10 @@ def patch_snapshot(request: HttpRequest, snapshot_id: str, data: SnapshotUpdateS
             update_fields.append("retry_at")
 
         if tags is not None:
-            snapshot.save_tags(normalize_tag_list(tags))
+            snapshot.save_tags(
+                normalize_tag_list(tags),
+                created_by=request.user if isinstance(request.user, User) else None,
+            )
 
         if payload.get("status") == Snapshot.StatusChoices.SEALED:
             snapshot.cancel()
diff --git a/archivebox/core/admin_snapshots.py b/archivebox/core/admin_snapshots.py
index 6e00575b..341daf78 100644
--- a/archivebox/core/admin_snapshots.py
+++ b/archivebox/core/admin_snapshots.py
@@ -356,6 +356,16 @@ class SnapshotAdmin(SearchResultsAdminMixin, ConfigEditorMixin, BaseModelAdmin):
     form = SnapshotAdminForm
     raw_id_fields = ("crawl", "parent_snapshot")
     list_select_related = ()
+
+    def save_related(self, request, form, formsets, change):
+        super().save_related(request, form, formsets, change)
+        tags_str = form.cleaned_data.get("tags_editor", "")
+        tag_names = [name.strip() for name in tags_str.split(",") if name.strip()]
+        form.instance.save_tags(
+            tag_names,
+            created_by=request.user if request.user.is_authenticated else None,
+        )
+
     list_display = (
         "permissions_badge",
         "created_at",
diff --git a/archivebox/core/models.py b/archivebox/core/models.py
index 8855670e..d5065992 100644
--- a/archivebox/core/models.py
+++ b/archivebox/core/models.py
@@ -609,9 +609,11 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW
             try:
                 SnapshotTag(snapshot_id=self.pk, tag_id=tag_id).save(force_insert=True)
             except IntegrityError:
-                # The unique (snapshot, tag) row already exists. In autocommit
-                # mode the failed INSERT is fully rolled back before continuing.
-                continue
+                # Only the unique (snapshot, tag) conflict is idempotent. Do
+                # not hide foreign-key or other integrity failures.
+                if SnapshotTag.objects.filter(snapshot_id=self.pk, tag_id=tag_id).exists():
+                    continue
+                raise
 
     def remove_tag_ids(self, tag_ids: Iterable[int | str]) -> int:
         tag_ids = [tag_id for tag_id in dict.fromkeys(tag_ids) if tag_id]
@@ -2726,10 +2728,10 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW
     def archive_size(self):
         return int(self.output_size or 0)
 
-    def save_tags(self, tags: Iterable[str] = ()) -> None:
+    def save_tags(self, tags: Iterable[str] = (), *, created_by: Any = None) -> None:
         from archivebox.core.tag_util import get_or_create_tag
 
-        tag_ids = {get_or_create_tag(tag)[0].pk for tag in tags if tag.strip()}
+        tag_ids = {get_or_create_tag(tag, created_by=created_by)[0].pk for tag in tags if tag.strip()}
         existing_tag_ids = set(SnapshotTag.objects.filter(snapshot_id=self.pk).values_list("tag_id", flat=True))
         self.remove_tag_ids(existing_tag_ids - tag_ids)
         self.add_tag_ids(tag_ids - existing_tag_ids)
diff --git a/archivebox/core/tag_util.py b/archivebox/core/tag_util.py
index 2336b1f7..4bbbf107 100644
--- a/archivebox/core/tag_util.py
+++ b/archivebox/core/tag_util.py
@@ -152,7 +152,8 @@ def get_or_create_tag(name: str, created_by: User | None = None) -> tuple[Tag, b
     if not normalized_name:
         raise ValueError("Tag name is required")
 
-    return Tag.get_or_create_by_name(normalized_name, defaults={"created_by": created_by})
+    defaults = {"created_by": created_by} if created_by is not None else None
+    return Tag.get_or_create_by_name(normalized_name, defaults=defaults)
 
 
 def rename_tag(tag: Tag, name: str) -> Tag:
diff --git a/archivebox/services/archive_result_service.py b/archivebox/services/archive_result_service.py
index 6689fa6e..b71417be 100644
--- a/archivebox/services/archive_result_service.py
+++ b/archivebox/services/archive_result_service.py
@@ -332,20 +332,13 @@ def _save_archiveresult_event_to_db(
         if event.error:
             defaults["notes"] = event.error
 
-    with _perf_span("archivebox.ArchiveResultService.on_ArchiveResultEvent.result_lookup"):
-        result = ArchiveResult.objects.filter(
-            snapshot=snapshot,
-            plugin=event.plugin,
-            hook_name=event.hook_name,
-        ).first()
-    if result is None:
-        with _perf_span("archivebox.ArchiveResultService.on_ArchiveResultEvent.result_create"):
-            result, _created = ArchiveResult.get_or_create_by_hook(
-                snapshot,
-                event.plugin,
-                event.hook_name,
-                defaults=defaults,
-            )
+    with _perf_span("archivebox.ArchiveResultService.on_ArchiveResultEvent.result_get_or_create"):
+        result, _created = ArchiveResult.get_or_create_by_hook(
+            snapshot,
+            event.plugin,
+            event.hook_name,
+            defaults=defaults,
+        )
 
     with _perf_span("archivebox.ArchiveResultService.on_ArchiveResultEvent.diff_fields"):
         update_fields = []
diff --git a/archivebox/tests/test_api_v1_core_archiveresults.py b/archivebox/tests/test_api_v1_core_archiveresults.py
index 01a5e687..7380dbe7 100644
--- a/archivebox/tests/test_api_v1_core_archiveresults.py
+++ b/archivebox/tests/test_api_v1_core_archiveresults.py
@@ -69,8 +69,9 @@ def test_archiveresult_create_does_not_open_a_database_transaction(client, api_a
         )
 
     assert response.status_code == 200, response.content
-    transaction_queries = [query["sql"] for query in queries if query["sql"].strip().upper() in {"BEGIN", "COMMIT"}]
-    assert transaction_queries == []
+    if connection.vendor == "sqlite":
+        transaction_queries = [query["sql"] for query in queries if query["sql"].strip().upper() in {"BEGIN", "COMMIT"}]
+        assert transaction_queries == []
 
 
 def test_intermediate_archiveresult_chunks_only_write_to_disk(client, api_admin_user, api_headers):
diff --git a/archivebox/tests/test_api_v1_core_snapshots.py b/archivebox/tests/test_api_v1_core_snapshots.py
index 4c01b820..771ca85e 100644
--- a/archivebox/tests/test_api_v1_core_snapshots.py
+++ b/archivebox/tests/test_api_v1_core_snapshots.py
@@ -110,5 +110,43 @@ def test_new_snapshot_creation_does_not_open_a_database_transaction(client, api_
     assert response.status_code == 200, response.content
     assert Snapshot.objects.filter(url=url, crawl=crawl).count() == 1
     assert Snapshot.objects.get(url=url, crawl=crawl).tags.filter(name="browser-extension-upload").exists()
-    transaction_queries = [query["sql"] for query in queries if query["sql"].strip().upper() in {"BEGIN", "COMMIT"}]
-    assert transaction_queries == []
+    if connection.vendor == "sqlite":
+        transaction_queries = [query["sql"] for query in queries if query["sql"].strip().upper() in {"BEGIN", "COMMIT"}]
+        assert transaction_queries == []
+
+
+def test_new_snapshot_creation_does_not_wait_for_active_crawl(client, api_admin_user, api_headers):
+    url = "https://example.com/browser-extension-new-snapshot-active-crawl"
+    crawl = Crawl.objects.create(urls=url, created_by=api_admin_user)
+    lock_acquired = threading.Event()
+    release_lock = threading.Event()
+
+    def hold_active_crawl_lock():
+        with crawl_lifecycle_lock(str(crawl.id)):
+            lock_acquired.set()
+            release_lock.wait(timeout=3)
+
+    holder = threading.Thread(target=hold_active_crawl_lock)
+    holder.start()
+    assert lock_acquired.wait(timeout=1)
+
+    started_at = time.monotonic()
+    response = client.post(
+        "/api/v1/core/snapshots",
+        data={
+            "url": url,
+            "crawl_id": str(crawl.id),
+            "depth": 0,
+            "status": Snapshot.StatusChoices.QUEUED,
+            "tags": ["browser-extension-upload"],
+        },
+        content_type="application/json",
+        **api_headers,
+    )
+    elapsed = time.monotonic() - started_at
+    release_lock.set()
+    holder.join(timeout=3)
+
+    assert response.status_code == 200, response.content
+    assert Snapshot.objects.filter(url=url, crawl=crawl).count() == 1
+    assert elapsed < 1
diff --git a/archivebox/tests/test_api_v1_core_tags_add_to_snapshot.py b/archivebox/tests/test_api_v1_core_tags_add_to_snapshot.py
index 2f2f5221..6f8a1ff0 100644
--- a/archivebox/tests/test_api_v1_core_tags_add_to_snapshot.py
+++ b/archivebox/tests/test_api_v1_core_tags_add_to_snapshot.py
@@ -1,5 +1,5 @@
 import pytest
-from django.db import connection
+from django.db import IntegrityError, connection
 from django.test.utils import CaptureQueriesContext
 
 from archivebox.core.models import Snapshot, Tag
@@ -27,5 +27,27 @@ def test_basic_success_case_request(client, api_admin_user, api_headers):
     assert response.status_code == 200, response.content
     assert response.json()["success"] is True
     assert snapshot.tags.filter(pk=tag.pk).exists()
-    transaction_queries = [query["sql"] for query in queries if query["sql"].strip().upper() in {"BEGIN", "COMMIT"}]
-    assert transaction_queries == []
+    if connection.vendor == "sqlite":
+        transaction_queries = [query["sql"] for query in queries if query["sql"].strip().upper() in {"BEGIN", "COMMIT"}]
+        assert transaction_queries == []
+
+
+def test_add_tag_ids_reraises_non_duplicate_integrity_errors(snapshot):
+    with pytest.raises(IntegrityError):
+        snapshot.add_tag_ids([2**31 - 1])
+
+
+def test_add_tag_ids_treats_existing_snapshot_tag_as_idempotent(snapshot, admin_user):
+    tag = Tag.objects.create(name="already-attached", created_by=admin_user)
+
+    snapshot.add_tag_ids([tag.pk])
+    snapshot.add_tag_ids([tag.pk])
+
+    assert snapshot.tags.filter(pk=tag.pk).count() == 1
+
+
+def test_save_tags_without_creator_uses_tag_model_default(snapshot):
+    snapshot.save_tags(["model-default-creator"])
+
+    tag = Tag.objects.get(name="model-default-creator")
+    assert tag.created_by_id is not None
diff --git a/archivebox/tests/test_api_v1_core_tags_remove_from_snapshot.py b/archivebox/tests/test_api_v1_core_tags_remove_from_snapshot.py
index 54ed29b4..94614fe0 100644
--- a/archivebox/tests/test_api_v1_core_tags_remove_from_snapshot.py
+++ b/archivebox/tests/test_api_v1_core_tags_remove_from_snapshot.py
@@ -28,5 +28,6 @@ def test_basic_success_case_request(client, api_admin_user, api_headers):
     assert response.status_code == 200, response.content
     assert response.json()["success"] is True
     assert not snapshot.tags.filter(pk=tag.pk).exists()
-    transaction_queries = [query["sql"] for query in queries if query["sql"].strip().upper() in {"BEGIN", "COMMIT"}]
-    assert transaction_queries == []
+    if connection.vendor == "sqlite":
+        transaction_queries = [query["sql"] for query in queries if query["sql"].strip().upper() in {"BEGIN", "COMMIT"}]
+        assert transaction_queries == []
diff --git a/archivebox/tests/test_archive_result_service.py b/archivebox/tests/test_archive_result_service.py
index 774b2ca1..6209d020 100644
--- a/archivebox/tests/test_archive_result_service.py
+++ b/archivebox/tests/test_archive_result_service.py
@@ -251,6 +251,32 @@ def test_archiveresult_duplicate_hook_rows_are_rejected():
         )
 
 
+def test_archiveresult_event_create_uses_one_result_lookup():
+    from abx_dl.events import ArchiveResultEvent
+    from django.db import connection
+    from django.test.utils import CaptureQueriesContext
+
+    from archivebox.services.archive_result_service import _save_archiveresult_event_to_db
+
+    snapshot = _create_snapshot()
+    event = ArchiveResultEvent(
+        snapshot_id=str(snapshot.id),
+        plugin="review-query-count",
+        hook_name="on_Snapshot__99_review.py",
+        status="failed",
+    )
+
+    with CaptureQueriesContext(connection) as queries:
+        _save_archiveresult_event_to_db(event, None)
+
+    result_lookups = [
+        query["sql"]
+        for query in queries
+        if query["sql"].lstrip().upper().startswith("SELECT") and 'FROM "core_archiveresult"' in query["sql"]
+    ]
+    assert len(result_lookups) == 1
+
+
 def test_process_completed_projects_failed_archiveresult_from_shipped_hook(tmp_path, hermetic_lib_dir):
     from archivebox.core.models import ArchiveResult
 
diff --git a/archivebox/tests/test_ui_admin_snapshot.py b/archivebox/tests/test_ui_admin_snapshot.py
index b53962db..215ed4d8 100644
--- a/archivebox/tests/test_ui_admin_snapshot.py
+++ b/archivebox/tests/test_ui_admin_snapshot.py
@@ -311,6 +311,35 @@ def test_snapshot_admin_preview_uses_extension_screenshot_when_standard_screensh
     assert "chrome_extension_screenshot/screenshot-2.png" not in preview["fallback_list"]
 
 
+def test_snapshot_admin_attributes_new_tags_to_authenticated_user(client, snapshot, admin_user):
+    from archivebox.core.models import Tag
+
+    client.force_login(admin_user)
+    response = client.post(
+        reverse("admin:core_snapshot_change", args=[snapshot.pk]),
+        {
+            "url": snapshot.url,
+            "title": snapshot.title or "",
+            "tags_editor": "admin-created-tag",
+            "permissions_config": "private",
+            "status": snapshot.status,
+            "retry_at": "",
+            "bookmarked_at_0": snapshot.bookmarked_at.date().isoformat(),
+            "bookmarked_at_1": snapshot.bookmarked_at.time().isoformat(),
+            "crawl": str(snapshot.crawl_id),
+            "config": '{"SAVE_ARCHIVE_DOT_ORG": "false"}',
+            "notes": "",
+            "_save": "Save",
+        },
+        HTTP_HOST=ADMIN_TEST_HOST,
+    )
+
+    assert response.status_code == 302, response.context and response.context["adminform"].form.errors
+    tag = Tag.objects.get(name="admin-created-tag")
+    assert tag.created_by == admin_user
+    assert snapshot.tags.filter(pk=tag.pk).exists()
+
+
 class TestSnapshotProgressStats:
     """Tests for Snapshot.get_progress_stats() method."""