From 2bfb3ad4eb119e5de54dd4bf98db1881db997d1a Mon Sep 17 00:00:00 2001
From: Nick Sweeting <git@sweeting.me>
Date: Tue, 9 Jun 2026 22:09:28 -0700
Subject: [PATCH] release: archivebox 0.9.35rc5

---
 AGENTS.md                                    |  14 ++
 archivebox/cli/archivebox_add.py             |  90 +++--------
 archivebox/core/forms.py                     |   7 +-
 archivebox/core/models.py                    |  25 ++-
 archivebox/crawls/models.py                  | 112 +++++++++++---
 archivebox/services/runner.py                |   4 +-
 archivebox/tests/test_api_v1_cli_add.py      |  34 +++--
 archivebox/tests/test_cli_add.py             | 152 +++++++++++--------
 archivebox/tests/test_ui_add_view.py         |  38 +++--
 archivebox/tests/test_ui_add_view_runtime.py |  29 ++--
 etc/package.json                             |   2 +-
 pyproject.toml                               |   8 +-
 12 files changed, 297 insertions(+), 218 deletions(-)
 create mode 100644 AGENTS.md

diff --git a/AGENTS.md b/AGENTS.md
new file mode 100644
index 00000000..6077e39c
--- /dev/null
+++ b/AGENTS.md
@@ -0,0 +1,14 @@
+I found 31 unique `no mocking` prompts in recent Codex history, across 21 session transcripts. Only one transcript literally started with `no mocking`; most had it later as testing guidance. Consolidated advice:
+
+- Tests must hit real user-facing code paths: CLI commands, REST/API calls, browser UI, real hooks, real ArchiveBox data dirs, real pytest fixtures, and real subprocess/binary behavior.
+- No mocking, faking, simulating, monkey patching, handwritten fake objects, fake buses, fake hooks, fake binaries, fake handlers, or direct-post shortcuts when the user path is through UI/extension/CLI.
+- No skipped, xfailed, flaky, or “works around platform” tests. Flakiness is treated as a bug, especially on macOS/browser flows.
+- Prefer live integration tests over narrow unit tests when behavior depends on browsers, binaries, ArchiveBox crawls, plugins, LLMs, or server state.
+- Assertions must validate real correctness: returned values, exit codes, DB rows, filesystem contents, field values, uploaded files, rendered output, and side effects. “No error occurred” or “attribute exists” is not enough.
+- Start fixes with failing red tests that reproduce the missing behavior or regression, then implement the minimal fix and confirm the test passes.
+- Use realistic setup patterns “like a user would”: events + bus + handlers, real browser pages/CDP sessions, real URLs or `pytest-httpserver`, real rows, real snapshots, real installs, real local browser/server state.
+- For ArchiveBox/API tests, use existing `conftest.py` fixtures and test harnesses, real test DB rows/data dirs, and user-facing commands/APIs rather than bespoke helpers.
+- For browser/extension tests, trigger behavior through the real extension UI or actual browser session, not direct posting or mocked browser/session objects.
+- For binary/provider tests, use real binaries and real installs; verify constraints and final installed package metadata, not just install success.
+- For coverage quality, keep tests strict, deterministic, grouped consistently, and use a few larger realistic tests when that gives better surface coverage than many tiny fake unit tests.
+- Avoid weakening test coverage, adding compatibility/shim/fallback layers, or guessing from code shape. Trace root causes, verify assumptions with tests/scripts, and let real type/parse errors surface normally.
diff --git a/archivebox/cli/archivebox_add.py b/archivebox/cli/archivebox_add.py
index 2c83832e..218abb10 100644
--- a/archivebox/cli/archivebox_add.py
+++ b/archivebox/cli/archivebox_add.py
@@ -5,7 +5,6 @@ __package__ = "archivebox.cli"
 __command__ = "archivebox add"
 
 import sys
-import json
 import os
 from pathlib import Path
 
@@ -13,6 +12,7 @@ from typing import Any, TYPE_CHECKING
 
 import rich_click as click
 
+from archivebox.config import CONSTANTS
 from archivebox.misc.util import enforce_types, docstring
 from archivebox.misc.util import parse_filesize_to_bytes
 
@@ -49,23 +49,6 @@ def _collect_input_urls(args: tuple[str, ...], *, parser: str = "auto") -> list[
     return urls
 
 
-def _direct_url_lines(raw_urls: str | list[str]) -> list[str]:
-    from archivebox.misc.util import validate_url
-
-    lines = [line.strip() for line in raw_urls.splitlines() if line.strip()] if isinstance(raw_urls, str) else [str(url).strip() for url in raw_urls if str(url).strip()]
-    direct_urls = []
-    for line in lines:
-        try:
-            direct_urls.append(validate_url(line))
-        except ValueError:
-            return []
-    return direct_urls
-
-
-def _should_import_as_source(raw_urls: str | list[str]) -> bool:
-    return not bool(_direct_url_lines(raw_urls))
-
-
 @enforce_types
 def add(
     urls: str | list[str],
@@ -106,7 +89,6 @@ def add(
     crawl_max_size = parse_filesize_to_bytes(crawl_max_size)
     crawl_timeout = int(crawl_timeout or 0)
     snapshot_max_size = parse_filesize_to_bytes(snapshot_max_size)
-    from archivebox import CONSTANTS
     from archivebox.config.permissions import USER, HOSTNAME
     from archivebox.config.common import get_config
 
@@ -143,25 +125,7 @@ def add(
     created_by_id = created_by_id or get_or_create_system_user_pk()
     started_at = timezone.now()
 
-    import_as_source = _should_import_as_source(urls)
     source_text = urls if isinstance(urls, str) else "\n".join(str(url) for url in urls)
-    if import_as_source:
-        url_list = []
-    elif isinstance(urls, str):
-        url_list = [line.strip() for line in urls.splitlines() if line.strip()]
-    else:
-        url_list = [str(url).strip() for url in urls if str(url).strip()]
-    if snapshot_ids and len(snapshot_ids) != len(url_list):
-        raise ValueError("snapshot_ids length must match urls length")
-    admitted_urls: list[str] = []
-    admitted_snapshot_ids: list[str] | None = [] if snapshot_ids else None
-    for index, url in enumerate(url_list):
-        if Snapshot.is_archivebox_internal_url(url, config=runtime_config):
-            print(f"[yellow][!] Skipping internal ArchiveBox URL: {url}[/yellow]")
-            continue
-        admitted_urls.append(url)
-        if admitted_snapshot_ids is not None and snapshot_ids:
-            admitted_snapshot_ids.append(snapshot_ids[index])
 
     # 2. Create a new Crawl with inline URLs
     cli_args = [*sys.argv]
@@ -200,26 +164,16 @@ def add(
     # stripped by Crawl.save() and rederived when hooks run.
     crawl_config.update(config_overrides)
 
-    if import_as_source:
-        urls_content = ""
-    else:
-        # 1. Save the provided URLs to sources/2024-11-05__23-59-59__cli_add.txt
-        sources_file = CONSTANTS.SOURCES_DIR / f"{timezone.now().strftime('%Y-%m-%d__%H-%M-%S')}__cli_add.txt"
-        sources_file.parent.mkdir(parents=True, exist_ok=True)
-        if admitted_snapshot_ids is not None:
-            sources_file.write_text(
-                "\n".join(
-                    json.dumps({"url": url, "id": admitted_snapshot_ids[index], "tags": tag, "depth": 0})
-                    for index, url in enumerate(admitted_urls)
-                ),
-            )
-        else:
-            sources_file.write_text("\n".join(admitted_urls))
-        urls_content = sources_file.read_text()
-
+    # Crawl.urls is the user's submitted source, not a derived work queue for
+    # this add path. Keeping it byte-for-byte readable is what lets API/UI/CLI
+    # callers audit or resume imports without losing RSS/Netscape/JSON metadata
+    # that is not representable as one plain URL per line.
     crawl = Crawl.objects.create(
-        urls=urls_content,
-        max_depth=depth + 1 if import_as_source else depth,
+        urls=source_text,
+        # The internal root snapshot occupies depth 0. URLs discovered from the
+        # submitted source become normal child snapshots at depth 1, so the
+        # effective crawl limit must be one hop deeper than the user requested.
+        max_depth=depth + 1,
         tags_str=tag,
         persona_id=persona_obj.id,
         label=f"{USER}@{HOSTNAME} $ {cmd_str} [{timestamp}]",
@@ -229,18 +183,18 @@ def add(
         config=crawl_config,
     )
 
-    if import_as_source:
-        sources_file = CONSTANTS.SOURCES_DIR / f"{crawl.id.hex}_{timezone.now().strftime('%Y-%m-%d__%H-%M-%S')}__cli_add.txt"
-        sources_file.parent.mkdir(parents=True, exist_ok=True)
-        sources_file.write_text(source_text)
-        root_snapshot = Snapshot.objects.create(
-            url=sources_file.resolve().as_uri(),
-            crawl=crawl,
-            depth=0,
-            title=sources_file.name,
-        )
-        crawl.urls = json.dumps({"type": "Snapshot", "url": root_snapshot.url, "id": str(root_snapshot.id), "depth": 0})
-        crawl.save(update_fields=["urls", "modified_at"])
+    # Parser plugins consume this root snapshot through the normal Snapshot
+    # hook lifecycle. ArchiveBox does not select parser plugins or call them
+    # directly; non-parser plugins cheaply no-result unsupported internal input.
+    root_snapshot = Snapshot.objects.create(
+        url=Snapshot.INTERNAL_INPUT_URL,
+        crawl=crawl,
+        depth=0,
+        title="stdin.txt",
+    )
+    staticfile_dir = root_snapshot.output_dir / "staticfile"
+    staticfile_dir.mkdir(parents=True, exist_ok=True)
+    (staticfile_dir / "stdin.txt").write_text(source_text, encoding="utf-8")
 
     print(f"[green]\\[+] Created Crawl {crawl.id} with max_depth={depth}[/green]")
     first_url = crawl.get_urls_list()[0] if crawl.get_urls_list() else ""
diff --git a/archivebox/core/forms.py b/archivebox/core/forms.py
index 7a936853..283dc21f 100644
--- a/archivebox/core/forms.py
+++ b/archivebox/core/forms.py
@@ -315,12 +315,9 @@ class AddLinkForm(PluginConfigFormMixin, forms.Form):
 
     def clean_url(self):
         value = self.cleaned_data.get("url") or ""
-        valid_urls = []
-        for url in find_all_urls(value):
-            valid_urls.append(url)
-        if not valid_urls:
+        if not list(find_all_urls(value)):
             raise forms.ValidationError("Enter at least one valid URL.")
-        return "\n".join(valid_urls)
+        return value
 
     def clean_url_filters(self):
         from archivebox.crawls.models import Crawl
diff --git a/archivebox/core/models.py b/archivebox/core/models.py
index f43e3326..f9876586 100755
--- a/archivebox/core/models.py
+++ b/archivebox/core/models.py
@@ -9,7 +9,7 @@ from datetime import datetime, timedelta
 import os
 import json
 from pathlib import Path
-from urllib.parse import unquote, urlparse
+from urllib.parse import urlparse
 
 from statemachine import State, registry
 
@@ -489,6 +489,8 @@ class SnapshotManager(models.Manager.from_queryset(SnapshotQuerySet)):  # ty: ig
 
 
 class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelWithNotes, ModelWithHealthStats, ModelWithStateMachine):
+    INTERNAL_INPUT_URL = "archivebox://internal"
+
     id = CompactUUIDField(primary_key=True, default=uuid7, editable=False, unique=True)
     created_at = models.DateTimeField(default=timezone.now, db_index=True)
     modified_at = models.DateTimeField(auto_now=True)
@@ -897,7 +899,7 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW
         return False
 
     def validate_url_for_archiving(self, *, config: Mapping[str, Any] | Any | None = None) -> None:
-        if self.is_crawl_source_file_url():
+        if self.is_internal_input_url():
             return
 
         try:
@@ -908,17 +910,8 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW
         if self.is_archivebox_internal_url(self.url, config=config):
             raise ValidationError({"url": "ArchiveBox cannot archive its own admin, web, api, or snapshot URLs."})
 
-    def is_crawl_source_file_url(self) -> bool:
-        parsed = urlparse((self.url or "").strip())
-        if parsed.scheme != "file" or self.depth != 0 or not self.crawl_id:
-            return False
-        try:
-            source_path = Path(unquote(parsed.path)).resolve()
-            sources_dir = CONSTANTS.SOURCES_DIR.resolve()
-            source_path.relative_to(sources_dir)
-        except (OSError, ValueError):
-            return False
-        return source_path.is_file() and source_path.name.startswith(f"{self.crawl_id.hex}_")
+    def is_internal_input_url(self) -> bool:
+        return (self.url or "").strip() == self.INTERNAL_INPUT_URL and self.depth == 0 and bool(self.crawl_id)
 
     def save(self, *args, **kwargs):
         update_fields = kwargs.get("update_fields")
@@ -969,6 +962,12 @@ class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelW
             crawl = Crawl.objects.filter(pk=self.crawl_id).first()
             if crawl is None:
                 return
+            # Snapshot.save() normally appends newly created URLs to Crawl.urls
+            # so legacy/direct crawls can keep their queue text in sync. For
+            # internal-input crawls that would corrupt the original submitted
+            # import text; parsed URLs are represented by child Snapshot rows.
+            if crawl.has_internal_input_root():
+                return
             if not crawl.url_passes_filters(self.url, snapshot=self, use_effective_config=False):
                 return
             # Best-effort skip if our URL is already recorded on the crawl;
diff --git a/archivebox/crawls/models.py b/archivebox/crawls/models.py
index a87ea99b..2ffb11fa 100755
--- a/archivebox/crawls/models.py
+++ b/archivebox/crawls/models.py
@@ -35,7 +35,7 @@ from archivebox.base_models.models import (
 )
 from archivebox.workers.models import RETRY_AT_MAX, ModelWithStateMachine, BaseStateMachine
 from archivebox.crawls.schedule_util import next_run_for_schedule, validate_schedule
-from archivebox.misc.util import validate_url, validate_url_length
+from archivebox.misc.util import parse_date, validate_url, validate_url_length
 
 if TYPE_CHECKING:
     from archivebox.core.models import Snapshot
@@ -521,6 +521,18 @@ class Crawl(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelWith
             return []
         return [url.strip() for url in self.urls.split("\n") if url.strip() and not url.strip().startswith("#")]
 
+    def has_internal_input_root(self) -> bool:
+        """Return True when Crawl.urls is preserved source text, not the work queue.
+
+        `archivebox add` creates a synthetic root snapshot to run parser hooks
+        through the same Snapshot lifecycle as every other extractor. In that
+        mode the raw submitted import text must remain in Crawl.urls forever;
+        parsed URLs live as child Snapshot rows and should not be appended back.
+        """
+        from archivebox.core.models import Snapshot
+
+        return self.snapshot_set.filter(url=Snapshot.INTERNAL_INPUT_URL, depth=0).exists()
+
     @staticmethod
     def normalize_domain(value: str) -> str:
         candidate = (value or "").strip().lower()
@@ -955,6 +967,13 @@ class Crawl(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelWith
 
         if self.status == self.StatusChoices.SEALED:
             return []
+        # Internal-input crawls preserve the submitted text verbatim in
+        # Crawl.urls. The root snapshot's parser hooks are the only supported
+        # path for turning that text into child snapshots, otherwise a later
+        # runner pass could reinterpret plain URL-looking lines as direct
+        # depth-0 work and bypass format-specific metadata parsing.
+        if self.has_internal_input_root():
+            return []
 
         created_snapshots = []
         crawl_tag_names = self.current_tag_names()
@@ -1116,10 +1135,16 @@ class Crawl(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelWith
         allowlist = self.split_filter_patterns(config.get("URL_ALLOWLIST", ""))
         denylist = self.split_filter_patterns(config.get("URL_DENYLIST", ""))
 
+        def metadata_score(record: Mapping[str, Any]) -> int:
+            # Multiple parsers can discover the same URL from one import root.
+            # Keep the record with the richest user-facing metadata so generic
+            # text/HTML extraction does not erase RSS/Netscape/JSON fields.
+            return sum(bool(record.get(field)) for field in ("title", "bookmarked_at", "timestamp", "tags"))
+
         deduped_records: dict[str, Mapping[str, Any]] = {}
         for record in records:
             url = sanitize_extracted_url(fix_url_from_markdown(str(record.get("url") or "").strip()))
-            if not url or url in deduped_records:
+            if not url:
                 continue
             try:
                 validate_url(url)
@@ -1130,11 +1155,43 @@ class Crawl(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelWith
                 print(f"[yellow][!] Skipping internal ArchiveBox discovered snapshot URL: {url}[/yellow]")
                 continue
             if self.url_passes_compiled_filters(url, allowlist=allowlist, denylist=denylist):
-                deduped_records[url] = record
+                existing_record = deduped_records.get(url)
+                if existing_record is None or metadata_score(record) > metadata_score(existing_record):
+                    deduped_records[url] = record
 
         if not deduped_records:
             return []
 
+        existing_in_crawl = {
+            snapshot.url: snapshot for snapshot in self.snapshot_set.prefetch_related("tags").filter(url__in=deduped_records.keys())
+        }
+        for url, snapshot in existing_in_crawl.items():
+            record = deduped_records[url]
+            update_fields = []
+            title = Snapshot._normalize_title_candidate(str(record.get("title") or "").strip()[:512], snapshot_url=url)
+            if title and (not snapshot.title or len(title) > len(snapshot.title or "")):
+                snapshot.title = title
+                update_fields.append("title")
+            bookmarked_at = None
+            try:
+                bookmarked_at = parse_date(record.get("bookmarked_at") or record.get("timestamp"))
+            except (TypeError, ValueError, OSError):
+                pass
+            if bookmarked_at and snapshot.bookmarked_at != bookmarked_at:
+                snapshot.bookmarked_at = bookmarked_at
+                update_fields.append("bookmarked_at")
+            if update_fields:
+                snapshot.save(update_fields=[*update_fields, "modified_at"])
+            tag_names = {
+                *self.parse_tag_names(
+                    str(record.get("tags") or ""),
+                    pattern=self._config_value(config, "TAG_SEPARATOR_PATTERN", r"[,]"),
+                ),
+            }
+            if tag_names:
+                tag_ids = [Tag.objects.get_or_create(name=tag_name)[0].pk for tag_name in tag_names]
+                snapshot.tags.add(*tag_ids)
+
         existing_scope = Snapshot.objects if bool(self._config_value(config, "ONLY_NEW", True)) else self.snapshot_set
         existing_urls = set(existing_scope.filter(url__in=deduped_records.keys()).values_list("url", flat=True))
         urls = [url for url in deduped_records.keys() if url not in existing_urls]
@@ -1145,25 +1202,32 @@ class Crawl(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelWith
             return []
 
         now = timezone.now()
-        snapshots = [
-            Snapshot(
-                url=url,
-                timestamp=str((now + timedelta(microseconds=index)).timestamp()),
-                title=Snapshot._normalize_title_candidate(
-                    str(deduped_records[url].get("title") or "").strip()[:512],
-                    snapshot_url=url,
-                )
-                or None,
-                crawl=self,
-                parent_snapshot=parent_snapshot,
-                depth=depth,
-                status=Snapshot.StatusChoices.QUEUED,
-                retry_at=now,
-                bookmarked_at=now,
-                created_at=now,
+        snapshots = []
+        for index, url in enumerate(urls):
+            record = deduped_records[url]
+            bookmarked_at = now
+            try:
+                bookmarked_at = parse_date(record.get("bookmarked_at") or record.get("timestamp")) or now
+            except (TypeError, ValueError, OSError):
+                pass
+            snapshots.append(
+                Snapshot(
+                    url=url,
+                    timestamp=str((now + timedelta(microseconds=index)).timestamp()),
+                    title=Snapshot._normalize_title_candidate(
+                        str(record.get("title") or "").strip()[:512],
+                        snapshot_url=url,
+                    )
+                    or None,
+                    crawl=self,
+                    parent_snapshot=parent_snapshot,
+                    depth=depth,
+                    status=Snapshot.StatusChoices.QUEUED,
+                    retry_at=now,
+                    bookmarked_at=bookmarked_at,
+                    created_at=now,
+                ),
             )
-            for index, url in enumerate(urls)
-        ]
         for snapshot in snapshots:
             snapshot.set_delete_at_from_config(self._config_value(config, "DELETE_AFTER", "0"))
 
@@ -1184,7 +1248,11 @@ class Crawl(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelWith
 
         crawl_urls = {url for _raw_line, url in self._iter_url_lines() if url}
         new_url_lines = [snapshot.url for snapshot in created_snapshots if snapshot.url not in crawl_urls]
-        if new_url_lines:
+        # For internal-input crawls, Crawl.urls is the immutable source text.
+        # Child snapshots are the parsed/indexed representation, so appending
+        # discovered URLs here would both duplicate state and destroy the exact
+        # import artifact users submitted through CLI/API/UI.
+        if new_url_lines and not self.has_internal_input_root():
             self.urls = (self.urls.rstrip() + "\n" + "\n".join(new_url_lines)).lstrip("\n")
             self.save(update_fields=["urls", "modified_at"])
 
diff --git a/archivebox/services/runner.py b/archivebox/services/runner.py
index 9353ad5a..4b8218ad 100644
--- a/archivebox/services/runner.py
+++ b/archivebox/services/runner.py
@@ -1024,7 +1024,9 @@ class CrawlRunner:
                 return
             config = normalize_runtime_config(snapshot["config"])
             snapshot_config_plugins = [name.strip() for name in str(config.get("PLUGINS") or "").split(",") if name.strip()]
-            snapshot_selected_plugins = self.selected_plugins if self.selected_plugins_from_args else (snapshot_config_plugins or self.selected_plugins)
+            snapshot_selected_plugins = (
+                self.selected_plugins if self.selected_plugins_from_args else (snapshot_config_plugins or 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)()
diff --git a/archivebox/tests/test_api_v1_cli_add.py b/archivebox/tests/test_api_v1_cli_add.py
index 8a6bf052..b8623e63 100644
--- a/archivebox/tests/test_api_v1_cli_add.py
+++ b/archivebox/tests/test_api_v1_cli_add.py
@@ -176,7 +176,9 @@ def wait_for_expected_import_snapshots(cwd: Path, expected_urls: set[str], *, ti
         if all(count == 1 for count in counts.values()) and not bad_statuses:
             return
         time.sleep(1)
-    raise AssertionError(f"timed out waiting for one queued/started/sealed snapshot per URL, got counts={counts}, bad_statuses={bad_statuses}")
+    raise AssertionError(
+        f"timed out waiting for one queued/started/sealed snapshot per URL, got counts={counts}, bad_statuses={bad_statuses}",
+    )
 
 
 def malicious_add_inputs(tmp_path: Path, *, safe_url: str) -> tuple[list[str], Path]:
@@ -215,24 +217,21 @@ def assert_no_file_or_shell_payload_snapshots(cwd: Path, *, canary: Path) -> Non
     with use_archivebox_db(cwd):
         snapshots = list(Snapshot.objects.all())
     assert not canary.exists()
-    assert not [
-        snapshot.url
-        for snapshot in snapshots
-        if str(snapshot.url).startswith("file:") and not snapshot.is_crawl_source_file_url()
-    ]
+    assert not [snapshot.url for snapshot in snapshots if str(snapshot.url).startswith("file:")]
     for forbidden in ("/etc/hosts", "/etc/passwd", "other_crawl_source", "archivebox_shell_injection_canary"):
         assert not [snapshot.url for snapshot in snapshots if forbidden in str(snapshot.url)]
 
 
 def test_basic_success_case_request(client, tmp_path, api_headers):
     init_archive(tmp_path)
+    submitted_url = "https://example.com/api-cli-add-basic"
 
     response = api_client_request(
         client,
         "post",
         "/api/v1/cli/add",
         payload={
-            "urls": ["https://example.com/api-cli-add-basic"],
+            "urls": [submitted_url],
             "depth": 0,
             "parser": "url_list",
             "plugins": "__archivebox_test_no_plugins__",
@@ -243,6 +242,11 @@ def test_basic_success_case_request(client, tmp_path, api_headers):
 
     assert response.status_code == 200, response.content
     assert response.json()["success"] is True
+    crawl = Crawl.objects.get()
+    root_snapshot = Snapshot.objects.get()
+    assert crawl.urls == submitted_url
+    assert root_snapshot.url == Snapshot.INTERNAL_INPUT_URL
+    assert (root_snapshot.output_dir / "staticfile" / "stdin.txt").read_text(encoding="utf-8") == submitted_url
 
 
 @pytest.mark.timeout(360)
@@ -275,6 +279,13 @@ def test_api_cli_add_import_text_formats_preserve_metadata_and_crawl_inner_urls(
             body = response.json()
             assert body["success"] is True
             assert body["result"]["crawl_id"]
+            with use_archivebox_db(tmp_path):
+                crawl = Crawl.objects.get(id=body["result"]["crawl_id"])
+                root_snapshot = crawl.snapshot_set.get(url=Snapshot.INTERNAL_INPUT_URL)
+                root_input = (root_snapshot.output_dir / "staticfile" / "stdin.txt").read_text(encoding="utf-8")
+            source_text = import_path.read_text(encoding="utf-8")
+            assert crawl.urls == source_text
+            assert root_input == source_text
 
         wait_for_import_processing(tmp_path, expected_urls)
         stop_server(tmp_path)
@@ -300,12 +311,11 @@ def test_api_cli_add_import_text_formats_preserve_metadata_and_crawl_inner_urls(
 
     with use_archivebox_db(tmp_path):
         crawls = list(Crawl.objects.order_by("created_at"))
-        snapshots_by_url = {
-            snapshot.url: snapshot
-            for snapshot in Snapshot.objects.prefetch_related("tags").filter(url__in=expected_urls)
-        }
+        snapshots_by_url = {snapshot.url: snapshot for snapshot in Snapshot.objects.prefetch_related("tags").filter(url__in=expected_urls)}
+        tags_by_url = {snapshot.url: set(snapshot.tags.values_list("name", flat=True)) for snapshot in snapshots_by_url.values()}
 
     assert len(crawls) == len(import_files)
+    assert [crawl.urls for crawl in crawls] == [path.read_text(encoding="utf-8") for path in import_files.values()]
     assert all(crawl.tags_str == "api-import" for crawl in crawls)
     assert all(crawl.status in {Crawl.StatusChoices.STARTED, Crawl.StatusChoices.SEALED} for crawl in crawls)
     assert len(snapshots_by_url) == len(expected_urls)
@@ -319,7 +329,7 @@ def test_api_cli_add_import_text_formats_preserve_metadata_and_crawl_inner_urls(
         if expected.get("date"):
             assert snapshot.bookmarked_at.date().isoformat() == expected["date"]
         if expected.get("tags"):
-            assert expected["tags"] | {"api-import"} <= set(snapshot.tags.values_list("name", flat=True))
+            assert expected["tags"] | {"api-import"} <= tags_by_url[snapshot.url]
 
 
 @pytest.mark.timeout(240)
diff --git a/archivebox/tests/test_cli_add.py b/archivebox/tests/test_cli_add.py
index 658e3974..6de356f4 100644
--- a/archivebox/tests/test_cli_add.py
+++ b/archivebox/tests/test_cli_add.py
@@ -188,7 +188,9 @@ def wait_for_expected_import_snapshots(cwd: Path, expected_urls: set[str], *, ti
         if all(count == 1 for count in counts.values()) and not bad_statuses:
             return
         time.sleep(1)
-    raise AssertionError(f"timed out waiting for one queued/started/sealed snapshot per URL, got counts={counts}, bad_statuses={bad_statuses}")
+    raise AssertionError(
+        f"timed out waiting for one queued/started/sealed snapshot per URL, got counts={counts}, bad_statuses={bad_statuses}",
+    )
 
 
 def malicious_add_inputs(tmp_path: Path, *, safe_url: str) -> tuple[list[str], Path]:
@@ -227,11 +229,7 @@ def assert_no_file_or_shell_payload_snapshots(cwd: Path, *, canary: Path) -> Non
     with use_archivebox_db(cwd):
         snapshots = list(Snapshot.objects.all())
     assert not canary.exists()
-    assert not [
-        snapshot.url
-        for snapshot in snapshots
-        if str(snapshot.url).startswith("file:") and not snapshot.is_crawl_source_file_url()
-    ]
+    assert not [snapshot.url for snapshot in snapshots if str(snapshot.url).startswith("file:")]
     for forbidden in ("/etc/hosts", "/etc/passwd", "other_crawl_source", "archivebox_shell_injection_canary"):
         assert not [snapshot.url for snapshot in snapshots if forbidden in str(snapshot.url)]
 
@@ -249,8 +247,14 @@ def test_add_single_url_records_url_in_crawl(initialized_archive):
 
     with use_archivebox_db(initialized_archive):
         crawl = Crawl.objects.get()
+        root_snapshot = Snapshot.objects.get()
+        root_input = (root_snapshot.output_dir / "staticfile" / "stdin.txt").read_text(encoding="utf-8")
 
+    assert crawl.urls == "https://example.com"
     assert crawl.get_urls_list() == ["https://example.com"]
+    assert root_snapshot.url == Snapshot.INTERNAL_INPUT_URL
+    assert root_snapshot.depth == 0
+    assert root_input == "https://example.com"
 
 
 @pytest.mark.timeout(360)
@@ -262,14 +266,22 @@ def test_add_stdin_import_formats_preserve_metadata_and_crawl_inner_urls(initial
     env = cli_env(port=port, server=True, **IMPORT_FORMAT_ENV)
 
     for import_path in import_files.values():
+        source_text = import_path.read_text(encoding="utf-8")
         result = run_archivebox_cmd(
             ["add", "--bg", "--depth=0", "--tag=cli-stdin-import"],
             cwd=initialized_archive,
             env=env,
-            stdin=import_path.read_text(encoding="utf-8"),
+            stdin=source_text,
             timeout=360,
         )
         assert result.returncode == 0, result.stderr or result.stdout
+        with use_archivebox_db(initialized_archive):
+            crawl = Crawl.objects.order_by("-created_at").first()
+            assert crawl is not None
+            root_snapshot = crawl.snapshot_set.get(url=Snapshot.INTERNAL_INPUT_URL)
+            root_input = (root_snapshot.output_dir / "staticfile" / "stdin.txt").read_text(encoding="utf-8")
+        assert crawl.urls == source_text
+        assert root_input == source_text
 
     try:
         start_archivebox_server(initialized_archive, env=env, port=port)
@@ -292,12 +304,11 @@ def test_add_stdin_import_formats_preserve_metadata_and_crawl_inner_urls(initial
 
     with use_archivebox_db(initialized_archive):
         crawls = list(Crawl.objects.order_by("created_at"))
-        snapshots_by_url = {
-            snapshot.url: snapshot
-            for snapshot in Snapshot.objects.prefetch_related("tags").filter(url__in=expected_urls)
-        }
+        snapshots_by_url = {snapshot.url: snapshot for snapshot in Snapshot.objects.prefetch_related("tags").filter(url__in=expected_urls)}
+        tags_by_url = {snapshot.url: set(snapshot.tags.values_list("name", flat=True)) for snapshot in snapshots_by_url.values()}
 
     assert len(crawls) == len(import_files)
+    assert [crawl.urls for crawl in crawls] == [path.read_text(encoding="utf-8") for path in import_files.values()]
     assert all(crawl.tags_str == "cli-stdin-import" for crawl in crawls)
     assert all(crawl.status in {Crawl.StatusChoices.STARTED, Crawl.StatusChoices.SEALED} for crawl in crawls)
     assert len(snapshots_by_url) == len(expected_urls)
@@ -311,7 +322,7 @@ def test_add_stdin_import_formats_preserve_metadata_and_crawl_inner_urls(initial
         if expected.get("date"):
             assert snapshot.bookmarked_at.date().isoformat() == expected["date"]
         if expected.get("tags"):
-            assert expected["tags"] | {"cli-stdin-import"} <= set(snapshot.tags.values_list("name", flat=True))
+            assert expected["tags"] | {"cli-stdin-import"} <= tags_by_url[snapshot.url]
 
 
 @pytest.mark.timeout(240)
@@ -343,7 +354,10 @@ def test_add_rejects_file_path_and_shell_injection_payloads(initialized_archive)
         crawl = Crawl.objects.get()
     assert crawl.status in {Crawl.StatusChoices.STARTED, Crawl.StatusChoices.SEALED}
     assert len({url for url, _status, _tag in snapshots}) == 1
-    assert all(status in {Snapshot.StatusChoices.QUEUED, Snapshot.StatusChoices.STARTED, Snapshot.StatusChoices.SEALED} for _url, status, _tag in snapshots)
+    assert all(
+        status in {Snapshot.StatusChoices.QUEUED, Snapshot.StatusChoices.STARTED, Snapshot.StatusChoices.SEALED}
+        for _url, status, _tag in snapshots
+    )
     assert "cli-security" in {tag for _url, _status, tag in snapshots}
 
 
@@ -437,17 +451,19 @@ def test_run_rejects_depth_two_file_url_snapshot_injected_directly_with_sql(init
     with use_archivebox_db(initialized_archive):
         injected_snapshot.refresh_from_db()
         snapshot_urls = set(Snapshot.objects.values_list("url", flat=True))
-        file_results = list(ArchiveResult.objects.filter(
-            snapshot=injected_snapshot,
-        ).values_list("plugin", "status"))
+        file_results = list(
+            ArchiveResult.objects.filter(
+                snapshot=injected_snapshot,
+            ).values_list("plugin", "status"),
+        )
 
     assert injected_snapshot.url == file_url
     assert secret_url not in snapshot_urls
     assert file_results == []
 
 
-def test_add_bg_queues_crawl_without_creating_snapshots(initialized_archive):
-    """Background add should leave root snapshot creation to the runner."""
+def test_add_bg_queues_internal_input_root_snapshot(initialized_archive):
+    """Background add stores submitted input on an internal root snapshot for the runner."""
     env = cli_env(disable_extractors=True)
     result = run_archivebox_cmd(
         ["add", "--bg", "--depth=0", "https://example.com"],
@@ -459,11 +475,15 @@ def test_add_bg_queues_crawl_without_creating_snapshots(initialized_archive):
 
     with use_archivebox_db(initialized_archive):
         crawl = Crawl.objects.get()
-        snapshot_count = Snapshot.objects.count()
+        root_snapshot = Snapshot.objects.get()
+        root_input = (root_snapshot.output_dir / "staticfile" / "stdin.txt").read_text(encoding="utf-8")
 
     assert crawl.status == Crawl.StatusChoices.QUEUED
     assert crawl.retry_at is not None
-    assert snapshot_count == 0
+    assert crawl.urls == "https://example.com"
+    assert root_snapshot.url == Snapshot.INTERNAL_INPUT_URL
+    assert root_snapshot.depth == 0
+    assert root_input == "https://example.com"
 
 
 def test_add_index_only_rejected_urls_leave_empty_crawl_for_runner_to_seal(initialized_archive):
@@ -485,21 +505,23 @@ def test_add_index_only_rejected_urls_leave_empty_crawl_for_runner_to_seal(initi
 
     with use_archivebox_db(initialized_archive):
         crawl = Crawl.objects.get()
-        snapshot_count = Snapshot.objects.count()
+        root_snapshot = Snapshot.objects.get()
 
     assert crawl.status == Crawl.StatusChoices.QUEUED
     assert crawl.retry_at is None
-    assert snapshot_count == 0
+    assert crawl.urls == "https://example.com"
+    assert root_snapshot.url == Snapshot.INTERNAL_INPUT_URL
 
     run_queued_crawls(initialized_archive, env)
 
     with use_archivebox_db(initialized_archive):
         crawl = Crawl.objects.get()
-        snapshot_count = Snapshot.objects.count()
+        snapshot_urls = set(Snapshot.objects.values_list("url", flat=True))
 
     assert crawl.status == Crawl.StatusChoices.SEALED
     assert crawl.retry_at is None
-    assert snapshot_count == 0
+    assert crawl.urls == "https://example.com"
+    assert snapshot_urls == {Snapshot.INTERNAL_INPUT_URL}
 
 
 def test_add_index_only_rejects_archivebox_internal_urls(initialized_archive):
@@ -521,12 +543,12 @@ def test_add_index_only_rejects_archivebox_internal_urls(initialized_archive):
 
     with use_archivebox_db(initialized_archive):
         crawl = Crawl.objects.get()
-        snapshot_count = Snapshot.objects.count()
+        snapshot_urls = set(Snapshot.objects.values_list("url", flat=True))
 
-    assert crawl.get_urls_list() == []
+    assert crawl.urls == "\n".join(internal_urls)
     assert crawl.status == Crawl.StatusChoices.QUEUED
     assert crawl.retry_at is None
-    assert snapshot_count == 0
+    assert snapshot_urls == {Snapshot.INTERNAL_INPUT_URL}
 
 
 def test_add_creates_crawl_record(initialized_archive):
@@ -544,8 +566,8 @@ def test_add_creates_crawl_record(initialized_archive):
     assert crawl_count == 1
 
 
-def test_add_creates_source_file(initialized_archive):
-    """Test that add creates a source file with the URL."""
+def test_add_creates_internal_input_file(initialized_archive):
+    """Test that add stores submitted text under the root snapshot staticfile output."""
     env = cli_env(disable_extractors=True)
     run_archivebox_cmd(
         ["add", "--index-only", "--depth=0", "https://example.com"],
@@ -553,14 +575,10 @@ def test_add_creates_source_file(initialized_archive):
         env=env,
     )
 
-    sources_dir = initialized_archive / "sources"
-    assert sources_dir.exists()
-
-    source_files = list(sources_dir.glob("*cli_add.txt"))
-    assert len(source_files) >= 1
-
-    source_content = source_files[0].read_text()
-    assert "https://example.com" in source_content
+    with use_archivebox_db(initialized_archive):
+        root_snapshot = Snapshot.objects.get()
+        source_content = (root_snapshot.output_dir / "staticfile" / "stdin.txt").read_text(encoding="utf-8")
+    assert source_content == "https://example.com"
 
 
 def test_add_multiple_urls_single_command(initialized_archive):
@@ -576,19 +594,16 @@ def test_add_multiple_urls_single_command(initialized_archive):
 
     with use_archivebox_db(initialized_archive):
         crawl = Crawl.objects.get()
+        root_snapshot = Snapshot.objects.get()
+        root_input = (root_snapshot.output_dir / "staticfile" / "stdin.txt").read_text(encoding="utf-8")
 
-    assert crawl.get_urls_list() == ["https://example.com", "https://example.org"]
+    assert crawl.urls == "https://example.com\nhttps://example.org"
+    assert root_input == "https://example.com\nhttps://example.org"
 
 
-def test_add_from_file(initialized_archive):
-    """Test adding URLs from a file.
-
-    The add command should treat a file argument as URL input and queue
-    a crawl containing each URL.
-    """
-
+def test_add_rejects_file_path_argument(initialized_archive):
+    """Local files must be piped through stdin, not passed as archiveable path arguments."""
     env = cli_env(disable_extractors=True)
-    # Create a file with URLs
     urls_file = initialized_archive / "urls.txt"
     urls_file.write_text("https://example.com\nhttps://example.org\n")
 
@@ -598,15 +613,13 @@ def test_add_from_file(initialized_archive):
         env=env,
     )
 
-    assert result.returncode == 0
+    assert result.returncode != 0
 
     with use_archivebox_db(initialized_archive):
-        crawl_count = Crawl.objects.count()
-        urls = Crawl.objects.get().get_urls_list()
+        assert Crawl.objects.count() == 0
+        assert Snapshot.objects.count() == 0
 
-    # The file is parsed into two input URLs.
-    assert crawl_count == 1
-    assert urls == ["https://example.com", "https://example.org"]
+    assert "No URLs provided" in (result.stderr or result.stdout)
 
 
 def test_add_with_depth_0_flag(initialized_archive):
@@ -738,11 +751,14 @@ def test_add_duplicate_url_creates_separate_crawls(initialized_archive):
 
     with use_archivebox_db(initialized_archive):
         crawl_count = Crawl.objects.count()
-        crawl_urls = list(Crawl.objects.order_by("created_at").values_list("urls", flat=True))
+        root_inputs = [
+            snapshot.output_dir.joinpath("staticfile", "stdin.txt").read_text(encoding="utf-8")
+            for snapshot in Snapshot.objects.order_by("created_at")
+        ]
 
     # Each add creates a new crawl with its own queued work.
     assert crawl_count == 2
-    assert crawl_urls == ["https://example.com", "https://example.com"]
+    assert root_inputs == ["https://example.com", "https://example.com"]
 
 
 def test_add_with_overwrite_flag(initialized_archive):
@@ -857,15 +873,16 @@ def test_add_index_only_queues_crawl_without_starting_runner(initialized_archive
 
     with use_archivebox_db(initialized_archive):
         crawl = Crawl.objects.get()
-        snapshot_count = Snapshot.objects.count()
+        root_snapshot = Snapshot.objects.get()
 
     assert crawl.status == Crawl.StatusChoices.QUEUED
     assert crawl.retry_at is None
-    assert snapshot_count == 0
+    assert crawl.urls == "https://example.com"
+    assert root_snapshot.url == Snapshot.INTERNAL_INPUT_URL
 
 
-def test_add_index_only_leaves_snapshot_creation_to_runner(initialized_archive):
-    """Test that index-only add does not create snapshots before the runner."""
+def test_add_index_only_creates_only_internal_root_snapshot(initialized_archive):
+    """Test that index-only add creates the input root but not parsed child snapshots."""
     env = cli_env(disable_extractors=True)
     run_archivebox_cmd(
         ["add", "--index-only", "--depth=0", "https://example.com"],
@@ -874,11 +891,11 @@ def test_add_index_only_leaves_snapshot_creation_to_runner(initialized_archive):
     )
 
     with use_archivebox_db(initialized_archive):
-        crawl_id = Crawl.objects.values_list("id", flat=True).get()
-        snapshot_count = Snapshot.objects.count()
+        crawl = Crawl.objects.get()
+        root_snapshot = Snapshot.objects.get()
 
-    assert crawl_id
-    assert snapshot_count == 0
+    assert crawl.urls == "https://example.com"
+    assert root_snapshot.url == Snapshot.INTERNAL_INPUT_URL
 
 
 def test_snapshot_create_sets_snapshot_timestamp(initialized_archive):
@@ -1006,7 +1023,7 @@ def test_cli_add_real_urls_with_options_writes_inspectable_outputs(initialized_a
         processes = list(Process.objects.filter(process_type="hook").values_list("process_type", "status", "exit_code", "pwd", "cmd"))
 
     assert real_flow_crawl is not None
-    assert real_flow_crawl[0] == 0
+    assert real_flow_crawl[0] == 1
     assert real_flow_crawl[1] == "real-flow,challenge"
     real_flow_config = real_flow_crawl[2] or {}
     assert real_flow_config["CRAWL_MAX_URLS"] == 2
@@ -1019,7 +1036,7 @@ def test_cli_add_real_urls_with_options_writes_inspectable_outputs(initialized_a
 
     snapshot_urls = {url for _id, url, _depth, _status, _title in snapshots}
     assert snapshot_urls >= {*wget_urls, chrome_url}
-    assert all(depth == 0 for _id, _url, depth, _status, _title in snapshots)
+    assert all(depth == (0 if url == Snapshot.INTERNAL_INPUT_URL else 1) for _id, url, depth, _status, _title in snapshots)
 
     by_url_plugin = {(url, plugin): status for url, plugin, status, _files, _size, _output in archive_results}
     assert by_url_plugin[("https://example.com", "wget")] == "succeeded"
@@ -1109,14 +1126,15 @@ def test_cli_recursive_crawl_processes_discovered_html_urls(initialized_archive,
             .values_list("snapshot__url", "plugin", "status", "output_files"),
         )
 
-    assert crawl[0] == 2
+    assert crawl[0] == 3
     assert crawl[1] == "recursive-flow"
     crawl_config = crawl[2] or {}
     assert crawl_config["CRAWL_MAX_URLS"] == 2
     assert crawl_config["CRAWL_MAX_SIZE"] == 50 * 1024 * 1024
     assert crawl_config.get("SNAPSHOT_MAX_SIZE", 0) == 0
-    assert (root_url, 0, "sealed") in snapshots
-    assert any(url == child_url and depth == 1 and status == "sealed" for url, depth, status in snapshots)
+    assert (Snapshot.INTERNAL_INPUT_URL, 0, "sealed") in snapshots
+    assert (root_url, 1, "sealed") in snapshots
+    assert any(url == child_url and depth == 2 and status == "sealed" for url, depth, status in snapshots)
 
     by_url_plugin = {(url, plugin): status for url, plugin, status, _files in archive_results}
     assert by_url_plugin[(root_url, "wget")] == "succeeded"
diff --git a/archivebox/tests/test_ui_add_view.py b/archivebox/tests/test_ui_add_view.py
index 692d7822..793f06d3 100644
--- a/archivebox/tests/test_ui_add_view.py
+++ b/archivebox/tests/test_ui_add_view.py
@@ -418,7 +418,10 @@ def test_add_view_queues_crawl_for_background_runner(client, admin_user, monkeyp
     assert crawl is not None
     assert crawl.status == Crawl.StatusChoices.QUEUED
     assert crawl.retry_at is not None
-    assert crawl.snapshot_set.count() == 0
+    assert crawl.urls == "https://example.com"
+    root_snapshot = crawl.snapshot_set.get()
+    assert root_snapshot.url == Snapshot.INTERNAL_INPUT_URL
+    assert (root_snapshot.output_dir / "staticfile" / "stdin.txt").read_text(encoding="utf-8") == "https://example.com"
 
 
 def test_add_view_start_paused_creates_paused_crawl_without_snapshots(client, admin_user, monkeypatch):
@@ -452,7 +455,10 @@ def test_add_view_start_paused_creates_paused_crawl_without_snapshots(client, ad
     assert crawl is not None
     assert crawl.status == Crawl.StatusChoices.PAUSED
     assert crawl.retry_at == RETRY_AT_MAX
-    assert crawl.snapshot_set.count() == 0
+    assert crawl.urls == "https://example.com/paused"
+    root_snapshot = crawl.snapshot_set.get()
+    assert root_snapshot.url == Snapshot.INTERNAL_INPUT_URL
+    assert (root_snapshot.output_dir / "staticfile" / "stdin.txt").read_text(encoding="utf-8") == "https://example.com/paused"
     assert crawl.config.get("INDEX_ONLY") is not True
 
 
@@ -494,17 +500,19 @@ def test_add_view_extracts_urls_from_mixed_text_input(client, admin_user, monkey
 
     crawl = Crawl.objects.order_by("-created_at").first()
     assert crawl is not None
-    assert crawl.urls == "\n".join(
+    expected_input = "\n".join(
         [
-            "https://sweeting.me",
-            "https://google.com",
-            "https://github.com/ArchiveBox/ArchiveBox",
-            "https://news.ycombinator.com",
-            "https://en.wikipedia.org/wiki/Classification_(machine_learning)",
-            "https://example.com/three",
-            "https://example.com/four",
+            "https://sweeting.me,https://google.com",
+            "Notes: [ArchiveBox](https://github.com/ArchiveBox/ArchiveBox), https://news.ycombinator.com",
+            "[Wiki](https://en.wikipedia.org/wiki/Classification_(machine_learning))",
+            '{"items":["https://example.com/three"]}',
+            "csv,https://example.com/four",
         ],
     )
+    assert crawl.urls == expected_input
+    root_snapshot = crawl.snapshot_set.get()
+    assert root_snapshot.url == Snapshot.INTERNAL_INPUT_URL
+    assert (root_snapshot.output_dir / "staticfile" / "stdin.txt").read_text(encoding="utf-8") == expected_input
 
 
 def test_add_view_trims_trailing_punctuation_from_markdown_urls(client, admin_user, monkeypatch):
@@ -542,12 +550,16 @@ def test_add_view_trims_trailing_punctuation_from_markdown_urls(client, admin_us
 
     crawl = Crawl.objects.order_by("-created_at").first()
     assert crawl is not None
-    assert crawl.urls == "\n".join(
+    expected_input = "\n".join(
         [
-            "https://github.com/ArchiveBox/ArchiveBox",
-            "https://github.com/abc?abc#234234",
+            "Docs: https://github.com/ArchiveBox/ArchiveBox.",
+            "Issue: https://github.com/abc?abc#234234?.",
         ],
     )
+    assert crawl.urls == expected_input
+    root_snapshot = crawl.snapshot_set.get()
+    assert root_snapshot.url == Snapshot.INTERNAL_INPUT_URL
+    assert (root_snapshot.output_dir / "staticfile" / "stdin.txt").read_text(encoding="utf-8") == expected_input
 
 
 def test_add_view_exposes_api_token_for_tag_widget_autocomplete(client, admin_user, monkeypatch):
diff --git a/archivebox/tests/test_ui_add_view_runtime.py b/archivebox/tests/test_ui_add_view_runtime.py
index 5a4bf09f..68dc9be1 100644
--- a/archivebox/tests/test_ui_add_view_runtime.py
+++ b/archivebox/tests/test_ui_add_view_runtime.py
@@ -166,7 +166,9 @@ def wait_for_expected_import_snapshots(cwd: Path, expected_urls: set[str], *, ti
         if all(count == 1 for count in counts.values()) and not bad_statuses:
             return
         time.sleep(1)
-    raise AssertionError(f"timed out waiting for one queued/started/sealed snapshot per URL, got counts={counts}, bad_statuses={bad_statuses}")
+    raise AssertionError(
+        f"timed out waiting for one queued/started/sealed snapshot per URL, got counts={counts}, bad_statuses={bad_statuses}",
+    )
 
 
 def malicious_add_inputs(tmp_path: Path, *, safe_url: str) -> tuple[list[str], Path]:
@@ -205,11 +207,7 @@ def assert_no_file_or_shell_payload_snapshots(cwd: Path, *, canary: Path) -> Non
     with use_archivebox_db(cwd):
         snapshots = list(Snapshot.objects.all())
     assert not canary.exists()
-    assert not [
-        snapshot.url
-        for snapshot in snapshots
-        if str(snapshot.url).startswith("file:") and not snapshot.is_crawl_source_file_url()
-    ]
+    assert not [snapshot.url for snapshot in snapshots if str(snapshot.url).startswith("file:")]
     for forbidden in ("/etc/hosts", "/etc/passwd", "other_crawl_source", "archivebox_shell_injection_canary"):
         assert not [snapshot.url for snapshot in snapshots if forbidden in str(snapshot.url)]
 
@@ -369,11 +367,12 @@ def test_public_add_view_import_text_formats_preserve_metadata_and_resume_withou
         assert 'name="url"' in add_page.text
 
         for import_path in import_files.values():
+            source_text = import_path.read_text(encoding="utf-8")
             response = requests.post(
                 f"http://127.0.0.1:{port}/add/",
                 headers={"Host": f"web.archivebox.localhost:{port}", "Referer": f"http://web.archivebox.localhost:{port}/add/"},
                 data={
-                    "url": import_path.read_text(encoding="utf-8"),
+                    "url": source_text,
                     "depth": "0",
                     "max_urls": "0",
                     "crawl_max_size": "0",
@@ -392,6 +391,13 @@ def test_public_add_view_import_text_formats_preserve_metadata_and_resume_withou
                 allow_redirects=False,
             )
             assert response.status_code in (302, 303), response.text
+            with use_archivebox_db(tmp_path):
+                crawl = Crawl.objects.order_by("-created_at").first()
+                assert crawl is not None
+                root_snapshot = crawl.snapshot_set.get(url=Snapshot.INTERNAL_INPUT_URL)
+                root_input = (root_snapshot.output_dir / "staticfile" / "stdin.txt").read_text(encoding="utf-8")
+            assert crawl.urls == source_text
+            assert root_input == source_text
 
         wait_for_import_processing(tmp_path, expected_urls)
         stop_server(tmp_path)
@@ -411,12 +417,11 @@ def test_public_add_view_import_text_formats_preserve_metadata_and_resume_withou
 
     with use_archivebox_db(tmp_path):
         crawls = list(Crawl.objects.order_by("created_at"))
-        snapshots_by_url = {
-            snapshot.url: snapshot
-            for snapshot in Snapshot.objects.prefetch_related("tags").filter(url__in=expected_urls)
-        }
+        snapshots_by_url = {snapshot.url: snapshot for snapshot in Snapshot.objects.prefetch_related("tags").filter(url__in=expected_urls)}
+        tags_by_url = {snapshot.url: set(snapshot.tags.values_list("name", flat=True)) for snapshot in snapshots_by_url.values()}
 
     assert len(crawls) == len(import_files)
+    assert [crawl.urls for crawl in crawls] == [path.read_text(encoding="utf-8") for path in import_files.values()]
     assert all(crawl.tags_str == "public-ui-import" for crawl in crawls)
     assert all(crawl.status in {Crawl.StatusChoices.STARTED, Crawl.StatusChoices.SEALED} for crawl in crawls)
     assert len(snapshots_by_url) == len(expected_urls)
@@ -430,7 +435,7 @@ def test_public_add_view_import_text_formats_preserve_metadata_and_resume_withou
         if expected.get("date"):
             assert snapshot.bookmarked_at.date().isoformat() == expected["date"]
         if expected.get("tags"):
-            assert expected["tags"] | {"public-ui-import"} <= set(snapshot.tags.values_list("name", flat=True))
+            assert expected["tags"] | {"public-ui-import"} <= tags_by_url[snapshot.url]
 
 
 @pytest.mark.timeout(240)
diff --git a/etc/package.json b/etc/package.json
index 2e2f647a..28d55ef0 100644
--- a/etc/package.json
+++ b/etc/package.json
@@ -1,6 +1,6 @@
 {
   "name": "archivebox",
-  "version": "0.9.35rc4",
+  "version": "0.9.35rc5",
   "repository": "github:ArchiveBox/ArchiveBox",
   "license": "MIT",
   "dependencies": {
diff --git a/pyproject.toml b/pyproject.toml
index 9fa04ed5..702914af 100755
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -1,6 +1,6 @@
 [project]
 name = "archivebox"
-version = "0.9.35rc4"
+version = "0.9.35rc5"
 requires-python = ">=3.13"
 description = "Self-hosted internet archiving solution."
 authors = [{name = "Nick Sweeting", email = "pyproject.toml@archivebox.io"}]
@@ -79,9 +79,9 @@ dependencies = [
     ### Extractor dependencies (optional binaries detected at runtime via shutil.which)
     ### Binary/Package Management
     "abxbus==2.5.10",         # EventBus API
-    "abxpkg>=1.11.189",        # for: detecting, versioning, and installing binaries via apt/brew/pip/npm
-    "abx-plugins>=1.11.193",  # shared ArchiveBox plugin package with Chrome/Puppeteer dependency wiring
-    "abx-dl>=1.11.193",       # shared ArchiveBox downloader package with blocking install preflight
+    "abxpkg>=1.11.190",        # for: detecting, versioning, and installing binaries via apt/brew/pip/npm
+    "abx-plugins>=1.11.194",  # shared ArchiveBox plugin package with Chrome/Puppeteer dependency wiring
+    "abx-dl>=1.11.194",       # shared ArchiveBox downloader package with blocking install preflight
     ### UUID7 backport for Python <3.14
     "uuid7>=0.1.0; python_version < '3.14'",  # provides the uuid_extensions module on Python 3.13
 ]