diff --git a/CHANGELOG.md b/CHANGELOG.md
index 8a6497e..7b83a68 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -5,6 +5,32 @@ Format loosely follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ## [Unreleased]
 
+### Added
+- **Use a scrape folder as dataset input**: *Build an AI training dataset*
+  now offers **Pages I scraped earlier** next to the URL options. Pick a
+  folder written by *Scrape pages (no AI)* or `dataforge scrape` (the latest
+  one is the default); its pages become the session's collection without
+  being fetched again, the robots.txt and skip-known questions are skipped,
+  and the run starts at processing. Reads `pages.jsonl`, or the
+  `page_NNN.md` files when only Markdown was saved.
+
+### Fixed
+- **No feedback while scraping several URLs**: *Scrape pages (no AI)* and
+  `dataforge scrape` now show a live bar naming the page being fetched
+  (`Fetching 2/5`) and print each page's result as it finishes, instead of
+  one spinner until the end. With `--json`, the same lines go to stderr when
+  it is a terminal. `scrape_urls` takes a `progress` callback.
+- **Multiple URLs box looked frozen**: it said "blank line to finish", but
+  Enter on a blank line only added another line (questionary's multiline box
+  finishes on Alt+Enter). Enter on an empty line now finishes, an empty box
+  says so instead of submitting nothing, and the count of URLs added is shown.
+- **Back did not go back**: questionary gives a choice with `value=None` its
+  title as the value, so *Back* in the URL method list, the scope question
+  and the adjust-settings menu returned text instead of None. In the wizard
+  this re-showed the same step, and Back on the scope question saved
+  "(b) Back" as the scope. Back now returns to the previous prompt: from a
+  URL box to the method list, from the method list to the main menu.
+
 ## [2.7.0] - 2026-09-24
 
 Easier URL selection: pick pages from the keyboard, choose the language on a
diff --git a/README.md b/README.md
index 54a8271..9a17506 100644
--- a/README.md
+++ b/README.md
@@ -147,7 +147,9 @@ dataforge scrape https://www.husd.us/schools/middle-schools -o schools
 Add `--check` for rule-based checks (empty or very short pages, duplicate
 text), `-f csv` to write only the tables, and `--json` before the command for
 a machine-readable summary. AI agents get the same through the MCP tool
-`scrape_page`.
+`scrape_page`. To turn a scrape folder into training data later, start
+**Build an AI training dataset** in `dataforge` and choose **Pages I scraped
+earlier**: the pages are not fetched again and the run starts at processing.
 
 To update or remove DataForge, run `dataforge update` or `dataforge uninstall`
 yourself, in your own terminal. Both close DataForge first, so nothing is
diff --git a/docs/TECHNICAL.tex b/docs/TECHNICAL.tex
index ce71e22..6d6f960 100644
--- a/docs/TECHNICAL.tex
+++ b/docs/TECHNICAL.tex
@@ -210,6 +210,10 @@ \subsection{Discovery}
 URLs from the keyboard. A recipe replaces this step declaratively with
 \texttt{source.language} / \texttt{source.include} /
 \texttt{source.exclude} / \texttt{source.max\_urls}.
+The wizard also accepts the folder written by a no-AI scrape
+(\texttt{dataforge scrape}) in place of URLs: its pages are stored as the
+session's collection, without being fetched again, and the run starts at
+processing.
 
 \subsection{Collection}
 \label{sec:collection}
diff --git a/src/dataforge/cli/app.py b/src/dataforge/cli/app.py
index a093ede..19e8854 100644
--- a/src/dataforge/cli/app.py
+++ b/src/dataforge/cli/app.py
@@ -92,7 +92,7 @@ def _version_callback(value: bool) -> None:
         raise typer.Exit()
 
 # Type alias for step result sentinels
-StepResult = Literal["next", "back", "back_to_urls", "back_to_config", "home", "exit"]
+StepResult = Literal["next", "back", "retry", "back_to_urls", "back_to_config", "home", "exit"]
 
 
 def _apply_project_file(s, cwd: Path) -> None:
@@ -216,7 +216,14 @@ def scrape(
     logger.add(_sys.stderr, level="WARNING", format="{level}: {message}")
 
     s = get_settings()
-    pages = asyncio.run(qs.scrape_urls(urls, rate_limit=s.rate_limit, tables=tables, check=check))
+    if _JSON_OUTPUT:
+        # stdout carries only the JSON; a person watching still sees each page.
+        progress = _stderr_scrape_progress if _sys.stderr.isatty() else None
+        pages = asyncio.run(qs.scrape_urls(urls, rate_limit=s.rate_limit, tables=tables,
+                                           check=check, progress=progress))
+    else:
+        pages = asyncio.run(_scrape_with_progress(urls, rate_limit=s.rate_limit,
+                                                  tables=tables, check=check))
     from datetime import datetime
 
     out_dir = out or Path("scrape") / datetime.now().strftime("%Y%m%d-%H%M%S")
@@ -230,11 +237,50 @@ def scrape(
             "files": written,
         }, indent=2, ensure_ascii=False))
     else:
-        _print_scrape_results(pages, out_dir)
+        _print_scrape_results(pages, out_dir, per_page=False)
     if not ok:
         raise typer.Exit(code=1)
 
 
+def _stderr_scrape_progress(i: int, total: int, url: str, page) -> None:
+    import sys as _sys
+    if page is not None:
+        _sys.stderr.write(f"[{i}/{total}] {page.status}  {url}\n")
+        _sys.stderr.flush()
+
+
+def _scrape_line(page) -> str:
+    if page.ok:
+        n = len(page.tables)
+        line = (f"[bold green]✓[/] {page.url}  [dim]{page.title[:60]}[/]  {page.word_count} words, "
+                f"{n} table{'s' if n != 1 else ''}")
+        return line + "".join(f"\n  [bold yellow]⚠[/]  {w}" for w in page.warnings)
+    return f"[bold red]✗[/] {page.url}  {page.status}"
+
+
+async def _scrape_with_progress(urls: list[str], **kw) -> list:
+    """scrape_urls with a live bar naming the page being fetched, and a line
+    per page as it finishes, so a long list never looks stalled."""
+    from dataforge import scrape as qs
+
+    total = len(urls)
+    prog = ui.make_progress("Scraping")
+    task = prog.add_task("Scraping", total=total)
+
+    def on_page(i: int, n: int, url: str, page) -> None:
+        if page is None:
+            short = url if len(url) <= 60 else "…" + url[-57:]
+            prog.update(task, description=f"[cyan]Fetching {i}/{n}[/]  [dim]{short}[/]")
+            return
+        prog.console.print(_scrape_line(page))
+        prog.update(task, completed=i)
+
+    ui.info(f"Fetching {total} page{'s' if total != 1 else ''}. "
+            "Pages on the same site are spaced out to respect its rate limit.")
+    with prog:
+        return await qs.scrape_urls(urls, progress=on_page, **kw)
+
+
 def _print_table_preview(table, title: str, max_rows: int = 8) -> None:
     from rich.table import Table as RichTable
 
@@ -1580,17 +1626,24 @@ async def _step_urls(state: dict) -> StepResult:
         if keep:
             return "next"
 
+    # Choosing again replaces an earlier scrape folder choice.
+    state.pop("scrape_dir", None)
+    state.pop("scrape_pages", None)
     try:
-        urls = await _collect_urls()
+        urls = await _collect_urls(state)
     except KeyboardInterrupt:
         return "home"
-    result = None if urls is None else urls
-    if result is None:
-        return "back"
-    if not result:
+    if urls is None:
+        return "home"
+    if not urls:
         ui.error("No valid URLs provided")
-        return "back"
-    state["seed_urls"] = result
+        return "retry"
+    state["seed_urls"] = urls
+    if state.get("scrape_pages"):
+        # Already scraped: no discovery, collection is the imported pages.
+        state["discovery_scope"] = "page"
+        return "next"
+    result = urls
     # A sitemap already lists the pages to choose from; anything else could
     # mean one page or a whole site, so ask rather than always crawling.
     if all(u.lower().endswith(".xml") for u in result):
@@ -1598,7 +1651,7 @@ async def _step_urls(state: dict) -> StepResult:
     else:
         scope = await prompts.ask_discovery_scope(len(result))
         if scope is None:
-            return "back"
+            return "retry"
         state["discovery_scope"] = scope
     return "next"
 
@@ -1651,7 +1704,10 @@ async def _step_config(state: dict) -> StepResult:
             return "back"
         state["n_per_chunk"] = n_per_chunk
 
-        ignore_robots = await prompts.ask_ignore_robots()
+        # Imported pages are not fetched, so robots.txt and "skip pages
+        # seen before" do not apply.
+        imported = bool(state.get("scrape_pages"))
+        ignore_robots = False if imported else await prompts.ask_ignore_robots()
         if ignore_robots:
             ui.warn("robots.txt enforcement disabled — ensure you have permission to scrape this site.")
         state["ignore_robots"] = ignore_robots
@@ -1659,7 +1715,7 @@ async def _step_config(state: dict) -> StepResult:
         from urllib.parse import urlparse as _urlparse
         seed_domain = _urlparse(state.get("seed_urls", [""])[0]).netloc or "this site"
         # Pages named one by one are scraped even if seen before.
-        if state.get("discovery_scope", "site") != "page":
+        if state.get("discovery_scope", "site") != "page" and not imported:
             state["skip_known"] = await prompts.ask_skip_known(seed_domain)
         else:
             state["skip_known"] = False
@@ -1825,7 +1881,15 @@ async def _interactive_pipeline() -> None:
         ui.info(f"Session directory: [dim]{ctx.session_dir()}[/]")
         ui.info(f"Database: [dim]{s.db_path.resolve()}[/]")
 
-        await _run_orchestrator(ctx)
+        start_from = None
+        if state.get("scrape_pages"):
+            from dataforge import scrape as qs
+            qs.import_into_session(ctx, state["scrape_pages"])
+            ui.info(f"Using {len(ctx.scraped_page_ids)} page(s) from "
+                    f"[dim]{state['scrape_dir']}[/]; starting at processing.")
+            start_from = PipelineStage.processing.value
+
+        await _run_orchestrator(ctx, start_from=start_from)
 
         # ── Post-pipeline: offer explore or loop back to menu ─────────────────
         ui.section("Pipeline Complete")
@@ -1859,7 +1923,11 @@ async def _run_wizard(state: dict) -> str:
         if result == "next":
             step_idx += 1
         elif result == "back":
-            step_idx = max(0, step_idx - 1)
+            if step_idx == 0:
+                return "home"
+            step_idx -= 1
+        elif result == "retry":
+            continue
         elif result == "back_to_urls":
             step_idx = STEPS.index("urls")
         elif result == "back_to_config":
@@ -2248,20 +2316,37 @@ async def _ask_export_config(s) -> dict:
     return {"targets": ["local"], "approved_only": True}
 
 
-async def _collect_urls() -> list[str] | None:
+async def _collect_urls(state: dict | None = None) -> list[str] | None:
+    """Ask for URLs. None when the user backs out of the method list.
+
+    Backing out of a later prompt (the URL box, the file path) returns to
+    the method list. With ``state`` (the dataset wizard), a folder from an
+    earlier no-AI scrape may be chosen instead: its pages go in
+    ``state["scrape_pages"]`` and their URLs are returned.
+    """
+    while True:
+        method = await prompts.ask_input_method(allow_scrape_folder=state is not None)
+        if method is None:
+            return None
+        urls = await _urls_by(method, state)
+        if urls is not None:
+            return urls
+
+
+async def _urls_by(method: str, state: dict | None) -> list[str] | None:
+    """URLs for one input method; None when the user backs out of it."""
     from dataforge.utils import sanitise, sanitise_many
 
-    method = await prompts.ask_input_method()
-    if method is None:
-        return None
-    if method == "Single URL":
+    if method == prompts.SCRAPE_FOLDER and state is not None:
+        return await _collect_scrape_folder(state)
+    if method in ("Single URL", "Sitemap URL"):
         url = await prompts.ask_single_url()
         if url is None:
             return None
         clean = sanitise(url)
         if not clean:
             ui.error(f"'{url}' is not a valid URL — skipping.")
-            return None
+            return []
         if clean != url:
             ui.info(f"URL corrected to: [dim]{clean}[/]")
         return [clean]
@@ -2273,6 +2358,7 @@ async def _collect_urls() -> list[str] | None:
         dropped = len(urls) - len(clean)
         if dropped:
             ui.warn(f"{dropped} invalid URL(s) removed.")
+        ui.info(f"{len(clean)} URL{'s' if len(clean) != 1 else ''} added.")
         return clean
     if method == "Text file":
         path = await prompts.ask_file_path()
@@ -2285,20 +2371,33 @@ async def _collect_urls() -> list[str] | None:
             ui.warn(f"{dropped} invalid URL(s) removed from file.")
         ui.info(f"Loaded {len(clean)} URLs from [dim]{path.resolve()}[/]")
         return clean
-    if method == "Sitemap URL":
-        url = await prompts.ask_single_url()
-        if url is None:
-            return None
-        clean = sanitise(url)
-        if not clean:
-            ui.error(f"'{url}' is not a valid URL — skipping.")
-            return None
-        if clean != url:
-            ui.info(f"URL corrected to: [dim]{clean}[/]")
-        return [clean]
     return []
 
 
+def _latest_scrape_dir() -> str:
+    root = Path("scrape")
+    dirs = sorted((d for d in root.glob("*") if d.is_dir()), reverse=True) if root.is_dir() else []
+    return str(dirs[0]) if dirs else ""
+
+
+async def _collect_scrape_folder(state: dict) -> list[str] | None:
+    from dataforge import scrape as qs
+
+    folder = await prompts.ask_scrape_dir(_latest_scrape_dir())
+    if folder is None:
+        return None
+    pages = qs.load_scrape_dir(folder)
+    if not pages:
+        ui.error(f"No pages with text in [dim]{folder.resolve()}[/]")
+        return []
+    words = sum(p.word_count for p in pages)
+    ui.info(f"Loaded {len(pages)} page{'s' if len(pages) != 1 else ''} ({words:,} words) "
+            f"from [dim]{folder.resolve()}[/]. They will not be fetched again.")
+    state["scrape_dir"] = str(folder.resolve())
+    state["scrape_pages"] = pages
+    return [p.url for p in pages]
+
+
 async def _pick_session(question: str) -> PipelineSession | None:
     """Choose any session, newest first. None on back/cancel or when there are none."""
     import questionary
@@ -2374,23 +2473,20 @@ async def _quick_scrape_flow() -> None:
     if out is None:
         return
     s = get_settings()
-    with ui.console.status(f"Fetching {len(urls)} page(s)…"):
-        pages = await qs.scrape_urls(urls, rate_limit=s.rate_limit, tables="csv" in fmts)
+    pages = await _scrape_with_progress(urls, rate_limit=s.rate_limit, tables="csv" in fmts)
     out_dir = Path(out).expanduser()
     qs.write_outputs(pages, out_dir, tuple(fmts))
-    _print_scrape_results(pages, out_dir)
+    _print_scrape_results(pages, out_dir, per_page=False)
+    if any(p.ok for p in pages) and ("md" in fmts or "jsonl" in fmts):
+        ui.info("To turn these pages into training data: Build an AI training dataset, "
+                f"then [bold]{prompts.SCRAPE_FOLDER}[/].")
 
 
-def _print_scrape_results(pages: list, out_dir: Path) -> None:
+def _print_scrape_results(pages: list, out_dir: Path, per_page: bool = True) -> None:
     ok = [p for p in pages if p.ok]
-    for p in pages:
-        if p.ok:
-            ui.success(f"{p.url}  [dim]{p.title[:60]}[/]  {p.word_count} words, "
-                       f"{len(p.tables)} table{'s' if len(p.tables) != 1 else ''}")
-            for w in p.warnings:
-                ui.warn(f"  {w}")
-        else:
-            ui.error(f"{p.url}  {p.status}")
+    if per_page:
+        for p in pages:
+            ui.console.print(_scrape_line(p))
     for p in ok:
         for k, t in enumerate(p.tables[:3], 1):
             _print_table_preview(t, f"{p.title or p.url}: table {k}")
diff --git a/src/dataforge/cli/prompts.py b/src/dataforge/cli/prompts.py
index 7b57024..ea47fe9 100644
--- a/src/dataforge/cli/prompts.py
+++ b/src/dataforge/cli/prompts.py
@@ -28,6 +28,15 @@ def _q(**kw):
     return {**kw, "style": _STYLE}
 
 
+# questionary gives a Choice with value=None its title as the value, so a
+# "Back" entry needs a real value that the caller then turns into None.
+_BACK = "__back__"
+
+
+def _none_if_back(answer):
+    return None if answer == _BACK else answer
+
+
 # ── Ghost-text command prompt ─────────────────────────────────────────────────
 
 class _SuggestFromChoices(AutoSuggest):
@@ -104,26 +113,53 @@ async def ask_command(choices: list[str], aliases: dict[str, str] | None = None)
 
 # ── Input collection ───────────────────────────────────────────────────────────
 
-async def ask_input_method() -> str | None:
+SCRAPE_FOLDER = "Pages I scraped earlier"
+
+
+async def ask_input_method(allow_scrape_folder: bool = False) -> str | None:
+    choices = ["Single URL", "Multiple URLs", "Text file", "Sitemap URL"]
+    if allow_scrape_folder:
+        choices.append(questionary.Choice(
+            SCRAPE_FOLDER, value=SCRAPE_FOLDER,
+            description="Use a folder saved by 'Scrape pages (no AI)': "
+                        "no fetching, goes straight to processing.",
+        ))
     answer = await questionary.select(
         "How would you like to provide URLs?",
-        choices=[
-            "Single URL",
-            "Multiple URLs",
-            "Text file",
-            "Sitemap URL",
-            questionary.Separator(),
-            questionary.Choice("(b) Back to menu", value=None),
-        ],
+        choices=[*choices, questionary.Separator(),
+                 questionary.Choice("(b) Back to menu", value=_BACK)],
         **_q(),
     ).ask_async()
-    return answer
+    return _none_if_back(answer)
+
+
+async def ask_scrape_dir(default: str = "") -> Path | None:
+    """Folder written by a no-AI scrape (pages.jsonl or page_NNN.md files)."""
+    def _ok(v: str) -> bool | str:
+        d = Path(clean_input(v)).expanduser()
+        if not d.is_dir():
+            return f"Folder not found: {clean_input(v)}"
+        if not ((d / "pages.jsonl").is_file() or any(d.glob("page_*.md"))):
+            return "No pages.jsonl or page_NNN.md files in that folder"
+        return True
+
+    path = await questionary.path(
+        "Scrape folder:",
+        default=default,
+        only_directories=True,
+        instruction="(Enter for the latest; drag a folder here or paste its path)",
+        validate=_ok,
+        **_q(),
+    ).ask_async()
+    if path is None:
+        return None
+    return Path(clean_input(path)).expanduser()
 
 
 async def ask_discovery_scope(n_urls: int) -> str | None:
     """How far to go from the URLs given: just them, their links, or the whole site."""
     one = n_urls == 1
-    return await questionary.select(
+    answer = await questionary.select(
         "Scrape just this page, or search for more?" if one
         else "Scrape just these pages, or search for more?",
         choices=[
@@ -144,10 +180,11 @@ async def ask_discovery_scope(n_urls: int) -> str | None:
                             "then pick which pages to keep.",
             ),
             questionary.Separator(),
-            questionary.Choice("(b) Back", value=None),
+            questionary.Choice("(b) Back", value=_BACK),
         ],
         **_q(),
     ).ask_async()
+    return _none_if_back(answer)
 
 
 async def ask_single_url() -> str | None:
@@ -162,15 +199,52 @@ async def ask_single_url() -> str | None:
     return clean_input(url)
 
 
+def _multiline_url_keys():
+    """Enter adds a line; Enter on an empty line finishes (as the prompt says).
+
+    questionary's multiline text only finishes on Alt+Enter or Esc then Enter,
+    so a plain Enter on a blank line looked like the prompt had hung.
+    """
+    from prompt_toolkit.key_binding import KeyBindings
+
+    kb = KeyBindings()
+
+    @kb.add("enter")
+    def _(event) -> None:
+        buf = event.current_buffer
+        if buf.document.current_line.strip():
+            buf.insert_text("\n")
+        else:
+            buf.validate_and_handle()
+
+    return kb
+
+
+def _has_url(text: str) -> bool | str:
+    urls = _split_urls(text)
+    if any(_valid_url(u) for u in urls):
+        return True
+    if urls:
+        return f"'{urls[0]}' is not a valid http(s) URL"
+    return "Enter at least one URL (or Ctrl+C to go back)"
+
+
 async def ask_multiple_urls() -> list[str] | None:
+    """Several URLs, one per line (or separated by spaces or commas).
+
+    Returns every entry given; the caller drops and reports invalid ones.
+    """
     raw = await questionary.text(
-        "Enter URLs (one per line, blank line to finish):",
+        "Enter URLs:",
         multiline=True,
+        instruction="(one per line; press Enter on an empty line to finish, Ctrl+C to go back)",
+        validate=_has_url,
+        key_bindings=_multiline_url_keys(),
         **_q(),
     ).ask_async()
     if raw is None:
         return None
-    return [u for u in _split_urls(raw) if _valid_url(u)]
+    return _split_urls(raw)
 
 
 async def ask_file_path() -> Path | None:
@@ -340,16 +414,17 @@ async def ask_stage_action(stage: str, next_stage: str = "") -> str:
 
 async def ask_adjust_settings_target() -> str | None:
     """Which mid-session setting to change. Returns None on cancel/back."""
-    return await questionary.select(
+    answer = await questionary.select(
         "Adjust which setting?",
         choices=[
             questionary.Choice("Generation model", value="generation_model"),
             questionary.Choice("Quality model", value="quality_model"),
             questionary.Choice("Output directory", value="output_dir"),
-            questionary.Choice("← Back", value=None),
+            questionary.Choice("← Back", value=_BACK),
         ],
         **_q(),
     ).ask_async()
+    return _none_if_back(answer)
 
 
 # ── Export ────────────────────────────────────────────────────────────────────
diff --git a/src/dataforge/cli/ui.py b/src/dataforge/cli/ui.py
index 980e175..aba9dd7 100644
--- a/src/dataforge/cli/ui.py
+++ b/src/dataforge/cli/ui.py
@@ -162,6 +162,8 @@ def review_panel(state: dict) -> None:
         "links": "The URL(s) above and the pages they link to",
         "site": "The whole site",
     }.get(state.get("discovery_scope", "site"), "")
+    if state.get("scrape_dir"):
+        scope = f"Already scraped, not fetched again: {state['scrape_dir']}"
     t.add_row("Scrape", scope)
     t.add_row("Session", state.get("session_name", ""))
     goal_text = state.get("goal", "")[:80]
diff --git a/src/dataforge/scrape.py b/src/dataforge/scrape.py
index ac31772..54e69be 100644
--- a/src/dataforge/scrape.py
+++ b/src/dataforge/scrape.py
@@ -6,12 +6,15 @@
 request DataForge makes (robots.txt, per-domain rate limit, ``Crawl-delay``)
 and no LLM, API key, session or database.
 
-Used by ``dataforge scrape`` and the MCP ``scrape_page`` tool.
+Used by ``dataforge scrape`` and the MCP ``scrape_page`` tool. A folder it
+wrote can later be the input of a dataset run (:func:`load_scrape_dir`,
+:func:`import_into_session`), which then starts at processing.
 """
 from __future__ import annotations
 
 import hashlib
 import json
+from collections.abc import Callable
 from dataclasses import asdict, dataclass, field
 from datetime import UTC, datetime
 from pathlib import Path
@@ -24,6 +27,10 @@
 MIN_WORDS = 50
 FORMATS = ("jsonl", "md", "csv")
 
+# progress(index, total, url, page): called with page=None just before a URL
+# is fetched, then again with its result, so a long list shows activity.
+Progress = Callable[[int, int, str, "ScrapedPage | None"], None]
+
 
 @dataclass
 class ScrapedPage:
@@ -56,50 +63,55 @@ async def scrape_urls(
     rate_limit: float,
     tables: bool = True,
     check: bool = False,
+    progress: Progress | None = None,
 ) -> list[ScrapedPage]:
     """Fetch each URL once, in order, and extract its content (and tables)."""
-    import httpx
-
     from dataforge.collectors import HTTPClient
     from dataforge.utils import RateLimiter
 
     results: list[ScrapedPage] = []
     seen_content: dict[str, str] = {}  # content hash -> first URL with it
     async with HTTPClient(RateLimiter(rate_limit)) as client:
-        for raw in urls:
-            url = sanitise(raw)
-            if not url:
-                results.append(ScrapedPage(url=raw, status="error: not a valid http(s) URL"))
-                continue
-            now = datetime.now(UTC).isoformat(timespec="seconds")
-            try:
-                resp = await client.get(url)
-            except PermissionError:
-                results.append(ScrapedPage(url=url, status="blocked by robots.txt", fetched_at=now))
-                continue
-            except httpx.HTTPStatusError as exc:
-                results.append(ScrapedPage(url=url, status=f"http {exc.response.status_code}",
-                                           fetched_at=now))
-                continue
-            except Exception as exc:
-                results.append(ScrapedPage(url=url, status=f"error: {type(exc).__name__}: {exc}",
-                                           fetched_at=now))
-                continue
-            if "html" not in resp.headers.get("content-type", "html"):
-                results.append(ScrapedPage(url=url, status="not html", fetched_at=now))
-                continue
-            content = extract(resp.text, url)
-            page = ScrapedPage(
-                url=url, status="ok", title=content.title, markdown=content.markdown,
-                text=content.text, word_count=content.word_count,
-                tables=extract_tables(resp.text) if tables else [], fetched_at=now,
-            )
-            if check:
-                page.warnings = _checks(page, seen_content)
+        total = len(urls)
+        for i, raw in enumerate(urls, 1):
+            if progress:
+                progress(i, total, raw, None)
+            page = await _scrape_one(client, raw, tables, check, seen_content)
             results.append(page)
+            if progress:
+                progress(i, total, page.url, page)
     return results
 
 
+async def _scrape_one(client, raw: str, tables: bool, check: bool,
+                      seen_content: dict[str, str]) -> ScrapedPage:
+    import httpx
+
+    url = sanitise(raw)
+    if not url:
+        return ScrapedPage(url=raw, status="error: not a valid http(s) URL")
+    now = datetime.now(UTC).isoformat(timespec="seconds")
+    try:
+        resp = await client.get(url)
+    except PermissionError:
+        return ScrapedPage(url=url, status="blocked by robots.txt", fetched_at=now)
+    except httpx.HTTPStatusError as exc:
+        return ScrapedPage(url=url, status=f"http {exc.response.status_code}", fetched_at=now)
+    except Exception as exc:
+        return ScrapedPage(url=url, status=f"error: {type(exc).__name__}: {exc}", fetched_at=now)
+    if "html" not in resp.headers.get("content-type", "html"):
+        return ScrapedPage(url=url, status="not html", fetched_at=now)
+    content = extract(resp.text, url)
+    page = ScrapedPage(
+        url=url, status="ok", title=content.title, markdown=content.markdown,
+        text=content.text, word_count=content.word_count,
+        tables=extract_tables(resp.text) if tables else [], fetched_at=now,
+    )
+    if check:
+        page.warnings = _checks(page, seen_content)
+    return page
+
+
 def _checks(page: ScrapedPage, seen: dict[str, str]) -> list[str]:
     """Cheap, rule-based quality checks. No LLM."""
     out = []
@@ -160,3 +172,80 @@ def write_outputs(pages: list[ScrapedPage], out_dir: Path,
                 f.write(json.dumps(t, ensure_ascii=False) + "\n")
         written["tables"].append(str(p))
     return written
+
+
+# ── A scrape folder as dataset input ──────────────────────────────────────────
+
+def load_scrape_dir(folder: Path) -> list[ScrapedPage]:
+    """Read back the pages a scrape saved in ``folder``. Only pages with text.
+
+    Uses ``pages.jsonl`` when it is there; otherwise the ``page_NNN.md``
+    files (a scrape saved as Markdown only), whose first line holds the URL.
+    """
+    folder = Path(folder)
+    jsonl = folder / "pages.jsonl"
+    pages: list[ScrapedPage] = []
+    if jsonl.is_file():
+        for line in jsonl.read_text(encoding="utf-8").splitlines():
+            if not line.strip():
+                continue
+            d = json.loads(line)
+            if d.get("status") != "ok" or not (d.get("markdown") or d.get("text")):
+                continue
+            pages.append(ScrapedPage(
+                url=d.get("url", ""), status="ok", title=d.get("title", ""),
+                markdown=d.get("markdown") or d.get("text", ""), text=d.get("text", ""),
+                word_count=d.get("word_count", 0), fetched_at=d.get("fetched_at", ""),
+            ))
+        return pages
+    for md in sorted(folder.glob("page_*.md")):
+        body = md.read_text(encoding="utf-8")
+        first, _, rest = body.partition("\n")
+        url = first.removeprefix("<!--").removesuffix("-->").strip() if first.startswith("<!--") else ""
+        if not url:
+            continue
+        rest = rest.lstrip("\n")
+        title = ""
+        if rest.startswith("# "):
+            title, _, rest = rest.partition("\n")
+            title = title[2:].strip()
+        text = rest.strip()
+        if text:
+            pages.append(ScrapedPage(url=url, status="ok", title=title, markdown=text,
+                                     text=text, word_count=len(text.split())))
+    return pages
+
+
+def import_into_session(ctx, pages: list[ScrapedPage]) -> list[int]:
+    """Store scraped pages as a session's collection, so the run starts at processing.
+
+    Writes each page's Markdown where the scraper would (``sessions/<id>/raw``)
+    and adds the URL and page rows the later stages and ``resume`` read.
+    Returns the new page ids, which are also set on ``ctx``.
+    """
+    from dataforge.storage import DiscoveredURL, URLSource, open_session
+    from dataforge.storage import ScrapedPage as PageRow
+
+    raw_dir = ctx.session_dir() / "raw"
+    raw_dir.mkdir(parents=True, exist_ok=True, mode=0o750)
+    ids: list[int] = []
+    with open_session(ctx.settings.db_path) as db:
+        for i, page in enumerate(pages):
+            raw_path = raw_dir / f"page_{i:05d}.md"
+            raw_path.write_text(page.markdown or page.text, encoding="utf-8")
+            url_row = DiscoveredURL(session_id=ctx.session_id, url=page.url,
+                                    source=URLSource.file, selected=True, scraped=True,
+                                    http_status=200)
+            db.add(url_row)
+            db.flush()
+            row = PageRow(session_id=ctx.session_id, url_id=url_row.id or 0, url=page.url,
+                          title=page.title, raw_path=str(raw_path), word_count=page.word_count)
+            db.add(row)
+            db.flush()
+            ids.append(row.id)
+        db.commit()
+    urls = [p.url for p in pages]
+    ctx.discovered_urls = list(urls)
+    ctx.selected_urls = list(urls)
+    ctx.scraped_page_ids = ids
+    return ids
diff --git a/tests/test_quick_scrape.py b/tests/test_quick_scrape.py
index 8f996e7..b1ea91a 100644
--- a/tests/test_quick_scrape.py
+++ b/tests/test_quick_scrape.py
@@ -138,3 +138,51 @@ async def test_mcp_scrape_page(site):
     assert out["tables"][0]["rows"][0] == ["Ochoa", "(510) 723-3130"]
     assert len(out["markdown"]) == 40 and out["truncated"] is True
     assert "text" not in out
+
+
+async def test_progress_reports_each_url_before_and_after(site):
+    events = []
+    await qs.scrape_urls([f"{site}/schools", f"{site}/nope"], rate_limit=1000,
+                         progress=lambda i, n, url, page: events.append(
+                             (i, n, page.status if page else None)))
+    assert events == [(1, 2, None), (1, 2, "ok"), (2, 2, None), (2, 2, "http 404")]
+
+
+# ── A scrape folder as dataset input ──────────────────────────────────────────
+
+async def test_load_scrape_dir_reads_ok_pages_from_jsonl(site, tmp_path):
+    pages = await qs.scrape_urls([f"{site}/schools", f"{site}/nope"], rate_limit=1000)
+    qs.write_outputs(pages, tmp_path)
+    (page,) = qs.load_scrape_dir(tmp_path)
+    assert page.url == f"{site}/schools" and page.title == "Middle Schools"
+    assert "Emergency kits" in page.markdown and page.word_count > 50
+
+
+async def test_load_scrape_dir_falls_back_to_markdown_files(site, tmp_path):
+    pages = await qs.scrape_urls([f"{site}/schools", f"{site}/copy"], rate_limit=1000)
+    qs.write_outputs(pages, tmp_path, ("md",))
+    loaded = qs.load_scrape_dir(tmp_path)
+    assert [p.url for p in loaded] == [f"{site}/schools", f"{site}/copy"]
+    assert loaded[0].title == "Middle Schools"
+    assert not loaded[0].markdown.startswith("<!--") and "Emergency kits" in loaded[0].markdown
+
+
+async def test_imported_pages_go_through_processing(site, tmp_path, tmp_settings):
+    from sqlmodel import select
+
+    from dataforge.agents import PipelineContext, ProcessorAgent
+    from dataforge.storage import DataFormat, DiscoveredURL, open_session
+
+    pages = await qs.scrape_urls([f"{site}/schools", f"{site}/copy"], rate_limit=1000)
+    qs.write_outputs(pages, tmp_path / "scrape")
+    ctx = PipelineContext(session_id="imp-1", session_name="imp", goal="", format=DataFormat.qa,
+                          seed_urls=[], settings=tmp_settings)
+    ids = qs.import_into_session(ctx, qs.load_scrape_dir(tmp_path / "scrape"))
+    assert len(ids) == 2 and ctx.scraped_page_ids == ids
+    assert ctx.selected_urls == [f"{site}/schools", f"{site}/copy"]
+    with open_session(tmp_settings.db_path) as db:
+        rows = db.exec(select(DiscoveredURL).where(DiscoveredURL.session_id == "imp-1")).all()
+    assert all(r.scraped and r.selected for r in rows)
+
+    ctx = await ProcessorAgent(ctx).run()
+    assert ctx.processed_chunk_ids
diff --git a/tests/test_wizard_navigation.py b/tests/test_wizard_navigation.py
new file mode 100644
index 0000000..a890007
--- /dev/null
+++ b/tests/test_wizard_navigation.py
@@ -0,0 +1,87 @@
+"""Back choices, the multi-URL box and wizard navigation."""
+from __future__ import annotations
+
+import contextlib
+from unittest.mock import AsyncMock, patch
+
+from prompt_toolkit.application import create_app_session
+from prompt_toolkit.input import create_pipe_input
+from prompt_toolkit.output import DummyOutput
+
+from dataforge.cli import app as cli_app
+from dataforge.cli import prompts
+
+DOWN, ENTER = "\x1b[B", "\r"
+
+
+@contextlib.contextmanager
+def keys(text: str):
+    with create_pipe_input() as inp, create_app_session(input=inp, output=DummyOutput()):
+        inp.send_text(text)
+        yield
+
+
+# questionary turns a Choice with value=None into its title, so "Back" used to
+# come back as the string "(b) Back to menu" instead of None.
+
+async def test_input_method_back_returns_none():
+    with keys(DOWN * 4 + ENTER):
+        assert await prompts.ask_input_method() is None
+
+
+async def test_input_method_offers_a_scrape_folder_in_the_wizard():
+    with keys(DOWN * 4 + ENTER):
+        assert await prompts.ask_input_method(allow_scrape_folder=True) == prompts.SCRAPE_FOLDER
+
+
+async def test_discovery_scope_back_returns_none():
+    with keys(DOWN * 3 + ENTER):
+        assert await prompts.ask_discovery_scope(2) is None
+
+
+async def test_adjust_settings_back_returns_none():
+    with keys(DOWN * 3 + ENTER):
+        assert await prompts.ask_adjust_settings_target() is None
+
+
+async def test_multiple_urls_finish_on_an_empty_line():
+    with keys("https://a.example/x" + ENTER + "https://b.example/y" + ENTER + ENTER):
+        assert await prompts.ask_multiple_urls() == ["https://a.example/x", "https://b.example/y"]
+
+
+async def test_multiple_urls_empty_enter_asks_again_instead_of_finishing():
+    with keys(ENTER + "https://a.example" + ENTER + ENTER):
+        assert await prompts.ask_multiple_urls() == ["https://a.example"]
+
+
+async def test_backing_out_of_the_url_box_returns_to_the_method_list():
+    method = AsyncMock(side_effect=["Single URL", None])
+    with patch.object(prompts, "ask_input_method", method), \
+            patch.object(prompts, "ask_single_url", AsyncMock(return_value=None)):
+        assert await cli_app._collect_urls() is None
+    assert method.await_count == 2
+
+
+async def test_back_on_the_first_wizard_step_goes_home():
+    with patch.object(cli_app, "_step_urls", AsyncMock(return_value="back")), \
+            patch.object(cli_app.ui, "screen"):
+        assert await cli_app._run_wizard({}) == "home"
+
+
+async def test_url_step_uses_a_scrape_folder_without_asking_the_scope(tmp_path):
+    from dataforge import scrape as qs
+
+    (tmp_path / "pages.jsonl").write_text(
+        '{"url": "https://a.example/p", "status": "ok", "title": "P", '
+        '"markdown": "Some text here", "text": "Some text here", "word_count": 3}\n',
+        encoding="utf-8")
+    state: dict = {}
+    scope = AsyncMock()
+    with patch.object(prompts, "ask_input_method", AsyncMock(return_value=prompts.SCRAPE_FOLDER)), \
+            patch.object(prompts, "ask_scrape_dir", AsyncMock(return_value=tmp_path)), \
+            patch.object(prompts, "ask_discovery_scope", scope):
+        assert await cli_app._step_urls(state) == "next"
+    assert state["seed_urls"] == ["https://a.example/p"]
+    assert state["discovery_scope"] == "page"
+    assert [p.url for p in state["scrape_pages"]] == [p.url for p in qs.load_scrape_dir(tmp_path)]
+    scope.assert_not_awaited()