diff --git a/CHANGELOG.md b/CHANGELOG.md
index e2909bb..6204b14 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -14,6 +14,12 @@
   - Journeys that change data only run against local hosts or hosts you allow with
     `--allow-writes`; journeys that send messages or delete data are never replayed.
   - The example shop has a JSON API and two example journeys in `examples/journeys/`.
+- **`prescale run` replays journeys at load.** When `prescale/journeys/` has journeys, virtual
+  users walk whole journeys (with each step's think time and their own cookies) instead of hitting
+  single URLs. Each journey is ramped on its own; the report names the journey and the step that
+  break first, with per-step timings, and saved runs carry a `journeys` field (plus
+  `verdict.culprit_journey`). Stale journeys are skipped with the reason. `--journey` picks
+  journeys; `--urls` tests URLs as before.
 
 ### Fixed
 - **`prescale mcp` works on a fresh install again.** `pip install 'prescale[mcp]'` picked up the new
diff --git a/README.md b/README.md
index 7be10e9..16371a8 100644
--- a/README.md
+++ b/README.md
@@ -94,7 +94,7 @@ From source: `git clone https://github.com/pyjeebz/PreScale.git && pip install .
 | [`run <url>`](#run--what-breaks-first) | Ramp traffic; report what breaks first and at what load |
 | [`investigate <url>`](#investigate--and-why) | …and *why* it breaks, plus how to fix it (`--app` to start your app first: [replica mode](#replica-mode--test-your-app-not-your-host)) |
 | [`audit <url>`](#audit--hygiene-no-load) | Load-free HTTP hygiene check (compression, caching, CDN, HTTP/2) |
-| [`journeys`](#journeys--your-users-real-paths) | List your journey files; `journeys verify <url>` checks they still replay |
+| [`journeys`](#journeys--your-users-real-paths) | List your journey files; `journeys verify <url>` checks they still replay. `run` replays them at load |
 | [`compare`](#ci--gate-on-capacity) | Capacity diff of two runs; regression gating for CI |
 | [`profiles`](#frame-it-as-a-launch) | Launch scenarios for `--profile` |
 | [`history` / `show`](#saved-runs-and-the-json-contract) | List and re-open saved runs |
@@ -198,10 +198,23 @@ prescale journeys verify http://localhost:3000     # replay each once, as one us
 
 `verify` stops at the first step that doesn't answer as expected and calls the journey **stale**: the app changed, or it's broken. It exits non-zero, so it works as a CI check.
 
+When `prescale/journeys/` has journeys, **`prescale run` replays them at load**: each virtual user walks a whole journey (logs in, adds to cart, orders), pauses for each step's `think` time, then starts again as a new user. Each journey is ramped on its own, so its verdict doesn't depend on any traffic mix, and the report names the journey and the step that break first:
+
+```
+                    Journeys (each ramped on its own)
+ Journey     Survives   Breaks at step                         Why
+ browse   400+ (held)   —
+ checkout ~20 (20–30)   3. place order  POST /api/carts/{{cart_id}}/order
+                                                    Server returned 5xx under load…
+```
+
+A stale journey isn't load-tested (a failure with no load says nothing about capacity); it's listed with the reason. `--journey NAME` picks journeys, and `--urls` (or `--path`, `--from-sitemap`) tests URLs instead.
+
 - `{{name}}` is a value an earlier step pulled out with `extract` (`body.<field>` or `header.<name>`); `{{unique.email}}` and `{{unique.id}}` are fresh each run; `{{env.NAME}}` reads an environment variable.
 - **Secrets never go in the file.** A literal password, token, API key or `Authorization` value is refused; use `{{env.NAME}}`.
 - Paths start with `/` and are relative to the URL you give (`http://host/app` + `/api/cart` is `http://host/app/api/cart`). A journey only calls that app: a request or redirect to another host fails the step.
-- Journeys that change data (POST, PUT, PATCH, DELETE) run against local hosts, or a host you name with `--allow-writes staging.myapp.com`. Types `side_effect` (emails, SMS, invites) and `destructive` are never replayed.
+- Journeys that change data (POST, PUT, PATCH, DELETE) run against local hosts, or a host you name with `--allow-writes staging.myapp.com` (on `verify` and `run`). Types `side_effect` (emails, SMS, invites) and `destructive` are never replayed.
+- Each run of a journey has its own cookies and extracted values, so one user's login never leaks to another.
 
 Try it on the example shop: `python examples/fragile-shop.py 8400`, then `SHOP_PASSWORD=demo prescale journeys verify http://localhost:8400 --dir examples/journeys`.
 
diff --git a/cli/src/prescale_cli/commands/journeys.py b/cli/src/prescale_cli/commands/journeys.py
index 23f112a..5918f59 100644
--- a/cli/src/prescale_cli/commands/journeys.py
+++ b/cli/src/prescale_cli/commands/journeys.py
@@ -87,12 +87,8 @@ def verify(url: str, directory: Path, only: tuple[str, ...], allow_writes: tuple
         found = [j for j in found if j.name in only]
 
     host = (parsed.hostname or "").lower()
-    _refuse_unallowed_writes(found, host, {h.lower() for h in allow_writes})
-    unset = sorted({name for j in found for name in j.env_names()} - set(os.environ))
-    if unset:
-        console.print(f"[red]Error:[/red] the journeys read environment variable(s) that "
-                      f"aren't set: {', '.join(unset)}.")
-        raise SystemExit(2)
+    refuse_unallowed_writes(found, host, {h.lower() for h in allow_writes})
+    refuse_missing_env(found)
 
     results = asyncio.run(_check_all(found, url, timeout=timeout, headers=headers))
     for result in results:
@@ -125,7 +121,17 @@ def _load_or_exit(directory: Path) -> list[Journey]:
     return found
 
 
-def _refuse_unallowed_writes(found: list[Journey], host: str, allowed: set[str]) -> None:
+def refuse_missing_env(found: list[Journey]) -> None:
+    """Stop before any request if a journey reads an environment variable that
+    isn't set, instead of failing a step halfway through."""
+    unset = sorted({name for j in found for name in j.env_names()} - set(os.environ))
+    if unset:
+        console.print(f"[red]Error:[/red] the journeys read environment variable(s) that "
+                      f"aren't set: {', '.join(unset)}.")
+        raise SystemExit(2)
+
+
+def refuse_unallowed_writes(found: list[Journey], host: str, allowed: set[str]) -> None:
     """Journeys that change data run on local hosts, or on hosts named with
     --allow-writes. Anywhere else they could change real data."""
     if host in _LOCAL_HOSTS or host in allowed:
diff --git a/cli/src/prescale_cli/commands/run.py b/cli/src/prescale_cli/commands/run.py
index f260d62..078b832 100644
--- a/cli/src/prescale_cli/commands/run.py
+++ b/cli/src/prescale_cli/commands/run.py
@@ -13,12 +13,17 @@
 
 import click
 from rich.console import Console
+from rich.markup import escape
 from rich.panel import Panel
 
 from prescale_cli.cloud import CloudError, require_credentials
 from prescale_cli.commands.hostcheck import confirm_host, host_permission_option
+from prescale_cli.commands.journeys import dir_option, refuse_missing_env, refuse_unallowed_writes
 from prescale_cli.commands.replicaopts import app_option, app_timeout_option, start_replica
 from prescale_cli.commands.upload import header_option, upload_after_run, upload_option
+from prescale_cli.journeys.load import JourneyRun, first_to_break, journeys_block, ramp
+from prescale_cli.journeys.model import Journey, load_dir
+from prescale_cli.journeys.replay import check
 from prescale_cli.live import LiveRamp, ramp_progress
 from prescale_cli.loadtest import (
     LoadError,
@@ -82,6 +87,14 @@
               help="Exit non-zero if it survives fewer than N users (a CI gate).")
 @click.option("--profile", "profile_name", default=None,
               help="Frame the run as a launch scenario (see `prescale profiles`).")
+@click.option("--journey", "only", multiple=True,
+              help="Replay only this journey (repeatable). See `prescale journeys`.")
+@click.option("--urls", "urls_only", is_flag=True,
+              help="Test URLs, not journeys, even when prescale/journeys/ has some.")
+@click.option("--allow-writes", "allow_writes", multiple=True, metavar="HOST",
+              help="Let journeys that change data run against this non-local host "
+                   "(repeatable).")
+@dir_option
 @app_option
 @app_timeout_option
 @host_permission_option
@@ -93,10 +106,16 @@ def run(url: str, paths: tuple[str, ...], from_sitemap: bool, max_users: int,
         repeat: int, think_time: float,
         ignore_robots: bool, yes: bool, as_json: bool, html_path: str | None,
         store: str | None, no_save: bool, fail_under: int | None,
-        profile_name: str | None, app: str | None, app_timeout: float,
-        host_permission: bool, headers: dict[str, str], upload: bool) -> None:
+        profile_name: str | None, only: tuple[str, ...], urls_only: bool,
+        allow_writes: tuple[str, ...], directory: Path, app: str | None,
+        app_timeout: float, host_permission: bool, headers: dict[str, str],
+        upload: bool) -> None:
     """Load test URL and report what breaks first.
 
+    When prescale/journeys/ has journeys, each one is replayed at load (users
+    walk the whole journey: log in, add to cart, order) and the report names the
+    journey and step that break first. Use --urls to test URLs instead.
+
     \b
     Examples:
         prescale run http://localhost:8000
@@ -128,10 +147,16 @@ def run(url: str, paths: tuple[str, ...], from_sitemap: bool, max_users: int,
         max_users = prof.peak_users
         think_time = prof.think_time_s
 
+    journeys = _journeys_to_run(directory, only, urls_only=urls_only,
+                                url_mode=bool(paths or from_sitemap or method.upper() != "GET"))
+    host = (parsed.hostname or "").lower()
+    if journeys:
+        refuse_unallowed_writes(journeys, host, {h.lower() for h in allow_writes})
+        refuse_missing_env(journeys)
+
     if app:
         start_replica(app, url, timeout=app_timeout, console=console, quiet=as_json)
 
-    host = (parsed.hostname or "").lower()
     is_local = host in _LOCAL_HOSTS
     if not is_local and not yes:
         if as_json:
@@ -151,6 +176,30 @@ def run(url: str, paths: tuple[str, ...], from_sitemap: bool, max_users: int,
         confirm_host(url, console=console, permitted=host_permission, as_json=as_json,
                      timeout=timeout, headers=headers)
 
+    config = {
+        "method": method,
+        "max_users": max_users,
+        "stage_seconds": stage_seconds,
+        "latency_wall_s": latency_wall,
+        "error_threshold": error_threshold,
+        "max_rps": max_rps,
+        "warmup": warmup,
+        "repeat": repeat,
+        "think_time_s": think_time,
+        "fail_under": fail_under,
+        "profile": prof.name if prof else None,
+    }
+    finish = dict(prof=prof, no_save=no_save, store=store, html_path=html_path,
+                  as_json=as_json, upload=upload, fail_under=fail_under)
+
+    if journeys:
+        report, result, live_mode = _run_journeys(
+            journeys, url, config, latency_wall=latency_wall,
+            error_threshold=error_threshold, timeout=timeout, headers=headers,
+            replica=bool(app), as_json=as_json)
+        _finish(result, report, live_mode=live_mode, **finish)
+        return
+
     extra: list[str] = []
     if from_sitemap:
         if not as_json:
@@ -203,22 +252,15 @@ def run(url: str, paths: tuple[str, ...], from_sitemap: bool, max_users: int,
 
     report = analyze(stages, latency_wall=latency_wall, error_threshold=error_threshold,
                      rate_capped=max_rps is not None)
-
-    config = {
-        "method": method,
-        "max_users": max_users,
-        "stage_seconds": stage_seconds,
-        "latency_wall_s": latency_wall,
-        "error_threshold": error_threshold,
-        "max_rps": max_rps,
-        "warmup": warmup,
-        "repeat": repeat,
-        "think_time_s": think_time,
-        "fail_under": fail_under,
-        "profile": prof.name if prof else None,
-    }
     result = build_result(report, url=url, targets=targets, config=config, warning=warning,
                           replica=bool(app))
+    _finish(result, report, live_mode=live_mode, **finish)
+
+
+def _finish(result: dict, report, *, prof, no_save: bool, store: str | None,
+            html_path: str | None, as_json: bool, upload: bool, fail_under: int | None,
+            live_mode: bool) -> None:
+    """Save, render, upload and gate a finished run (URLs or journeys alike)."""
     if prof is not None:
         result["profile"] = scenario_block(prof, report.survives_users)
     saved_path = None if no_save else write_result(result, store=store)
@@ -244,3 +286,95 @@ def run(url: str, paths: tuple[str, ...], from_sitemap: bool, max_users: int,
         Console(stderr=True).print(
             f"[red]✗ fail-under:[/red] survives ~{report.survives_users} < {fail_under}")
         raise SystemExit(1)
+
+
+def _journeys_to_run(directory: Path, only: tuple[str, ...], *, urls_only: bool,
+                     url_mode: bool) -> list[Journey]:
+    """The journeys this run replays: none in URL mode (--urls, --path,
+    --from-sitemap or a non-GET --method), else every valid file in DIRECTORY,
+    narrowed by --journey. A broken journey file stops the run rather than
+    silently falling back to testing URLs."""
+    if urls_only or url_mode:
+        if only:
+            console.print("[red]Error:[/red] --journey can't be combined with --urls, "
+                          "--path, --from-sitemap or --method.")
+            raise SystemExit(2)
+        return []
+    found, errors = load_dir(directory)
+    for error in errors:
+        console.print(f"[red]Error:[/red] {escape(error)}", highlight=False)
+    if errors:
+        raise SystemExit(2)
+    if only:
+        missing = sorted(set(only) - {j.name for j in found})
+        if missing:
+            console.print(f"[red]Error:[/red] no journey named {', '.join(missing)} "
+                          f"in {escape(str(directory))}.", highlight=False)
+            raise SystemExit(2)
+        found = [j for j in found if j.name in only]
+    return found
+
+
+def _run_journeys(journeys: list[Journey], url: str, config: dict, *, latency_wall: float,
+                  error_threshold: float, timeout: float, headers: dict[str, str],
+                  replica: bool, as_json: bool) -> tuple:
+    """The isolation pass: check each journey once, then ramp each one that
+    replays on its own. Returns (report, result, live_mode) for the first
+    journey to break, with every journey's verdict in result["journeys"]."""
+    if not as_json:
+        console.print(f"\n[bold]PreScale[/bold] — replaying {len(journeys)} "
+                      f"journey{'s' if len(journeys) != 1 else ''} against [cyan]{url}[/cyan]")
+    # --json keeps stdout for the Result alone; notes about journeys go to stderr.
+    out = Console(stderr=True) if as_json else console
+    checks = asyncio.run(_check_all(journeys, url, timeout=timeout, headers=headers))
+    runs = [JourneyRun(j, c) for j, c in zip(journeys, checks)]
+    levels = default_levels(config["max_users"])
+    live_mode = False
+    for run in runs:
+        name = escape(run.journey.name)
+        if not run.check.ok:
+            failed = run.check.failed
+            what = "stale" if run.check.stale else "couldn't run"
+            out.print(f"[yellow]⚠ {name}: {what}, not load-tested[/yellow] "
+                          f"[dim]({escape(failed.step.name)}: {escape(failed.problem or '')})"
+                          "[/dim]", highlight=False)
+            continue
+        if not as_json:
+            console.print(f"\n[bold]Journey {name}[/bold] [dim]({run.journey.type}, "
+                          f"{len(run.journey.steps)} steps)[/dim]", highlight=False)
+        progress = ramp_progress(console, latency_wall=latency_wall, as_json=as_json)
+        live_mode = isinstance(progress, LiveRamp)
+        try:
+            with progress:
+                stages, run.warning = asyncio.run(ramp(
+                    run.journey, url, levels=levels, stage_seconds=config["stage_seconds"],
+                    timeout=timeout, max_rps=config["max_rps"], warmup=config["warmup"],
+                    repeat=config["repeat"], headers=headers, think=config["think_time_s"],
+                    progress_cb=progress.starting, on_stage=progress.finished))
+        except LoadError as exc:
+            out.print(f"[red]Error:[/red] {exc}")
+            raise SystemExit(1)
+        run.report = analyze(stages, latency_wall=latency_wall,
+                             error_threshold=error_threshold,
+                             rate_capped=config["max_rps"] is not None)
+
+    worst = first_to_break(runs)
+    if worst is None:
+        out.print("[red]Error:[/red] no journey replays against this app, so there's "
+                      "nothing to load-test. Run `prescale journeys verify` for details.")
+        raise SystemExit(1)
+    warnings = list(dict.fromkeys(r.warning for r in runs if r.warning))
+    result = build_result(worst.report, url=url,
+                          targets=[s.label for s in worst.journey.steps],
+                          config={**config, "mode": "journeys"},
+                          warning=" ".join(warnings) or None, replica=replica)
+    # Only a journey that broke is a culprit; if all held, there is none.
+    result["verdict"]["culprit_journey"] = (worst.journey.name
+                                            if worst.report.onset_users is not None else None)
+    result["journeys"] = journeys_block(runs)
+    return worst.report, result, live_mode
+
+
+async def _check_all(journeys: list[Journey], url: str, *, timeout: float,
+                     headers: dict[str, str]):
+    return [await check(j, url, timeout=timeout, headers=headers) for j in journeys]
diff --git a/cli/src/prescale_cli/compare.py b/cli/src/prescale_cli/compare.py
index d2d1a45..dea8b2e 100644
--- a/cli/src/prescale_cli/compare.py
+++ b/cli/src/prescale_cli/compare.py
@@ -27,6 +27,8 @@ def _facts(result: dict) -> dict:
         "branch": env.get("git_branch"),
         "created_at": result.get("created_at"),
         "target": result.get("target", {}).get("url"),
+        "mode": (result.get("config") or {}).get("mode") or "urls",
+        "culprit_journey": v.get("culprit_journey"),
     }
 
 
@@ -35,6 +37,9 @@ def compare_results(old: dict, new: dict, *, threshold: float = _DEFAULT_THRESHO
     survives_users; it's 'confident' when the M1 bands don't overlap. A new run
     that held to its max below the old level is 'inconclusive', not a drop."""
     o, n = _facts(old), _facts(new)
+    if o["mode"] != n["mode"]:
+        # URL capacity and journey capacity measure different things: never a regression.
+        return _not_comparable(o, n)
     o_surv, n_surv = o["survives_users"], n["survives_users"]
     pct = (n_surv - o_surv) / o_surv if o_surv else 0.0
     inconclusive = (n["onset_users"] is None and n["max_tested"] is not None
@@ -57,10 +62,28 @@ def compare_results(old: dict, new: dict, *, threshold: float = _DEFAULT_THRESHO
         "improved": improved,
         "inconclusive": inconclusive,
         "confident": confident,
-        "summary": _summary(o, n, pct, regressed, improved, inconclusive),
+        "summary": _summary(o, n, pct, regressed, improved, inconclusive) + _journey_note(o, n),
     }
 
 
+def _not_comparable(o: dict, n: dict) -> dict:
+    kinds = {"urls": "tested URLs", "journeys": "replayed journeys"}
+    return {
+        "old": o, "new": n, "survives_delta": 0, "survives_pct": 0.0, "peak_rps_pct": 0.0,
+        "regressed": False, "improved": False, "inconclusive": True, "confident": False,
+        "summary": (f"Not comparable: the old run {kinds.get(o['mode'], o['mode'])}, the new "
+                    f"run {kinds.get(n['mode'], n['mode'])}. Compare runs of the same kind."),
+    }
+
+
+def _journey_note(o: dict, n: dict) -> str:
+    if o["culprit_journey"] and n["culprit_journey"] \
+            and o["culprit_journey"] != n["culprit_journey"]:
+        return (f" The journey that breaks first changed: {o['culprit_journey']} → "
+                f"{n['culprit_journey']}.")
+    return ""
+
+
 def _summary(o: dict, n: dict, pct: float, regressed: bool, improved: bool,
              inconclusive: bool = False) -> str:
     arrow = f"{o['survives_users']} → {n['survives_users']}"
diff --git a/cli/src/prescale_cli/journeys/load.py b/cli/src/prescale_cli/journeys/load.py
new file mode 100644
index 0000000..d25554a
--- /dev/null
+++ b/cli/src/prescale_cli/journeys/load.py
@@ -0,0 +1,124 @@
+"""Load-test journeys: the isolation pass.
+
+Each journey is first replayed once (`check`). A journey that fails there is
+stale or unreachable and isn't ramped: a failure with no load says nothing about
+capacity. Every journey that replays is then ramped on its own, so its verdict
+(how many users it survives, the step that breaks first, why) doesn't depend on
+any traffic mix. The run's overall verdict is the first journey to break.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+import httpx
+
+from prescale_cli.journeys.model import Journey
+from prescale_cli.journeys.replay import JourneyCheck, client_options, journey_vu
+from prescale_cli.loadtest import RunReport, StageResult, run_loadtest
+
+
+@dataclass
+class JourneyRun:
+    journey: Journey
+    check: JourneyCheck
+    report: RunReport | None = None
+    warning: str | None = None
+
+    @property
+    def status(self) -> str:
+        if self.report is not None:
+            return "ok"
+        return "stale" if self.check.stale else "unreachable"
+
+
+async def ramp(journey: Journey, base_url: str, *, levels: list[int], stage_seconds: float,
+               timeout: float = 10.0, max_rps: float | None = None, warmup: bool = True,
+               repeat: int = 1, headers: dict[str, str] | None = None,
+               env: dict[str, str] | None = None, think: float = 0.0, progress_cb=None,
+               on_stage=None, transport: httpx.AsyncBaseTransport | None = None,
+               ) -> tuple[list[StageResult], str | None]:
+    """Ramp virtual users who each walk JOURNEY, through LEVELS. THINK is the
+    pause for steps without their own `think`."""
+    return await run_loadtest(
+        [base_url], levels=levels, stage_seconds=stage_seconds, timeout=timeout,
+        max_rps=max_rps, warmup=warmup, repeat=repeat, headers=headers,
+        host_permission=True,  # the caller has already checked the host's rules
+        progress_cb=progress_cb, on_stage=on_stage, transport=transport,
+        vu=journey_vu(journey, base_url, env, think), client_options=client_options(base_url),
+        preflight=False,  # `check` already replayed the journey once
+    )
+
+
+def first_to_break(runs: list[JourneyRun]) -> JourneyRun | None:
+    """The ramped journey with the lowest capacity (ties: the one that broke)."""
+    ramped = [r for r in runs if r.report is not None]
+    if not ramped:
+        return None
+    return min(ramped, key=lambda r: (r.report.survives_users,
+                                      r.report.onset_users is None))
+
+
+def journeys_block(runs: list[JourneyRun]) -> list[dict]:
+    """The Result's `journeys` field: one entry per journey, ramped or not."""
+    return [_entry(run) for run in runs]
+
+
+def _entry(run: JourneyRun) -> dict:
+    journey = run.journey
+    entry: dict = {"name": journey.name, "type": journey.type, "weight": journey.weight,
+                   "status": run.status, "reason": None}
+    if run.report is None:
+        failed = run.check.failed
+        index = run.check.steps.index(failed) + 1 if failed else None
+        entry["reason"] = f"step {index} ({failed.step.name}): {failed.problem}" if failed \
+            else None
+        return entry
+
+    report = run.report
+    stage = _stage_at(report)
+    culprit = None
+    if report.culprit_route:
+        index = next((i for i, s in enumerate(journey.steps, 1)
+                      if s.label == report.culprit_route), None)
+        if index is not None:
+            culprit = {"index": index, "name": journey.steps[index - 1].name,
+                       "route": report.culprit_route}
+    entry.update({
+        "survives_users": report.survives_users,
+        "max_tested": report.max_tested,
+        "onset_users": report.onset_users,
+        "onset_reason": report.onset_reason,
+        "confidence": {"survives_low": report.survives_low,
+                       "survives_high": report.survives_high, "stable": report.stable},
+        "culprit_step": culprit,
+        "bottleneck": report.bottleneck,
+        "steps_at_users": stage.users if stage else None,
+        "steps": [_step_stats(i, s, stage) for i, s in enumerate(journey.steps, 1)],
+    })
+    return entry
+
+
+def _stage_at(report: RunReport) -> StageResult | None:
+    """The stage the step timings come from: where it broke, else the last one."""
+    if not report.stages:
+        return None
+    if report.onset_users is not None:
+        for stage in report.stages:
+            if stage.users == report.onset_users:
+                return stage
+    return report.stages[-1]
+
+
+def _step_stats(index: int, step, stage: StageResult | None) -> dict:
+    stat = stage.routes.get(step.label) if stage else None
+    if stat is None:
+        return {"index": index, "name": step.name, "route": step.label, "requests": 0,
+                "p50_ms": None, "p95_ms": None, "error_rate": None}
+    has_latency = bool(stat.latencies)
+    return {
+        "index": index, "name": step.name, "route": step.label, "requests": stat.total,
+        "p50_ms": round(stat.pct(0.50) * 1000) if has_latency else None,
+        "p95_ms": round(stat.pct(0.95) * 1000) if has_latency else None,
+        "error_rate": round(stat.error_rate, 4),
+    }
diff --git a/cli/src/prescale_cli/journeys/replay.py b/cli/src/prescale_cli/journeys/replay.py
index a5222e0..4760ff3 100644
--- a/cli/src/prescale_cli/journeys/replay.py
+++ b/cli/src/prescale_cli/journeys/replay.py
@@ -7,15 +7,20 @@
 `check` replays a journey once, as a single user, to see whether it still works
 against the app as it is now. A step that fails its expectation here, with no
 load at all, means the journey is stale (the app changed), not that the app
-broke under load. Each run gets its own client, so cookies and extracted values
-never leak between users.
+broke under load. `journey_vu` is the same journey as a virtual user for a load
+test: it walks the steps over and over until the stage ends.
+
+Every run of a journey has its own cookies and extracted values, so a login or
+a cart never leaks from one user to another, even on a shared client.
 """
 
 from __future__ import annotations
 
+import asyncio
 import os
 import time
 from dataclasses import dataclass, field
+from http.cookiejar import CookieJar, DefaultCookiePolicy
 from typing import Any
 
 import httpx
@@ -24,6 +29,7 @@
 from prescale_cli.loadtest import client_headers
 
 _SNIPPET = 160
+_MAX_REDIRECTS = 10
 
 
 class OffOriginError(Exception):
@@ -46,6 +52,22 @@ def _origin(url: httpx.URL) -> tuple[str, str, int | None]:
     return url.scheme, url.host, url.port
 
 
+def client_options(base_url: str) -> dict:
+    """Settings for any client that runs journeys: requests (redirect hops too)
+    must stay on BASE_URL's origin, and the client itself keeps no cookies, since
+    each run of a journey carries its own."""
+    origin = _origin(httpx.URL(base_url))
+
+    async def same_origin(request: httpx.Request) -> None:
+        if _origin(request.url) != origin:
+            raise OffOriginError(f"redirected to {request.url.host}, outside the app")
+
+    return {
+        "event_hooks": {"request": [same_origin]},
+        "cookies": CookieJar(policy=DefaultCookiePolicy(allowed_domains=[])),
+    }
+
+
 @dataclass
 class StepCheck:
     step: Step
@@ -53,6 +75,8 @@ class StepCheck:
     latency: float = 0.0
     ok: bool = False
     problem: str | None = None   # why it failed, in plain words
+    kind: str | None = None      # a short failure kind for load reports
+    headers: httpx.Headers | None = None
 
 
 @dataclass
@@ -82,28 +106,73 @@ async def check(journey: Journey, base_url: str, *, timeout: float = 10.0,
     env = dict(os.environ) if env is None else env
     result = JourneyCheck(journey)
     variables: dict[str, Any] = fresh_uniques()
-    origin = _origin(httpx.URL(base_url))
-
-    async def same_origin(request: httpx.Request) -> None:
-        # Runs for redirect hops too, so an app can't bounce a write elsewhere.
-        if _origin(request.url) != origin:
-            raise OffOriginError(f"redirected to {request.url.host}, outside the app")
-
+    jar = httpx.Cookies()
     async with httpx.AsyncClient(timeout=timeout, follow_redirects=True,
                                  headers=client_headers(headers), transport=transport,
-                                 event_hooks={"request": [same_origin]}) as client:
+                                 **client_options(base_url)) as client:
         for step in journey.steps:
-            outcome = await run_step(client, step, base_url, variables, env)
+            outcome = await run_step(client, step, base_url, variables, env, jar)
             result.steps.append(outcome)
             if not outcome.ok:
                 break
     return result
 
 
+def journey_vu(journey: Journey, base_url: str, env: dict[str, str] | None = None,
+               think: float = 0.0):
+    """JOURNEY as a load-test virtual user: walk its steps, pausing for each
+    step's think time, and start again with a fresh identity until the stage
+    deadline. A failed step ends that walk (after its pause), like a real user
+    giving up. Each step is recorded under its label, so the report can name the
+    step that broke. THINK is the pause for steps that don't set their own
+    (`--think-time`, or a launch profile's)."""
+    env = dict(os.environ) if env is None else env
+
+    async def vu(client: httpx.AsyncClient, deadline: float, sink, gate=None) -> None:
+        loop = asyncio.get_running_loop()
+        while loop.time() < deadline:
+            variables: dict[str, Any] = fresh_uniques()
+            jar = httpx.Cookies()
+            for step in journey.steps:
+                if gate is not None and not await gate.wait(deadline):
+                    return
+                if loop.time() >= deadline:
+                    return
+                outcome = await run_step(client, step, base_url, variables, env, jar)
+                _record(sink, outcome)
+                # A user pauses after a step either way; restarting the instant a step
+                # fails would hammer the app harder than real users who hit an error.
+                pause = step.think or think
+                if pause:
+                    remaining = deadline - loop.time()
+                    if remaining <= 0:
+                        return
+                    await asyncio.sleep(min(pause, remaining))
+                if not outcome.ok:
+                    break
+            # Never spin: a walk that failed without awaiting must still yield.
+            await asyncio.sleep(0)
+
+    return vu
+
+
+def _record(sink, outcome: StepCheck) -> None:
+    """One step's outcome into the stage's numbers. An answer the journey didn't
+    expect counts as a failure, even a 4xx: under load that's a step going wrong."""
+    label = outcome.step.label
+    if outcome.status is None:
+        sink.record_error(label, outcome.kind or "network")
+    elif outcome.ok or outcome.status >= 500 or outcome.status == 429:
+        sink.record_response(label, outcome.status, outcome.latency, outcome.headers)
+    else:
+        sink.record_error(label, outcome.kind or f"unexpected {outcome.status}")
+
+
 async def run_step(client: httpx.AsyncClient, step: Step, base_url: str,
-                   variables: dict[str, Any], env: dict[str, str]) -> StepCheck:
-    """Send one step, judge it against `expect`, and pull its `extract` values
-    into VARIABLES for the steps after it."""
+                   variables: dict[str, Any], env: dict[str, str],
+                   jar: httpx.Cookies) -> StepCheck:
+    """Send one step with this journey run's cookies (JAR), judge it against
+    `expect`, and pull its `extract` values into VARIABLES for the steps after it."""
     outcome = StepCheck(step)
     try:
         url = step_url(base_url, str(render(step.path, variables, env)))
@@ -111,24 +180,31 @@ async def run_step(client: httpx.AsyncClient, step: Step, base_url: str,
         payload = render(step.json, variables, env) if step.json is not None else None
         body = render(step.body, variables, env) if step.body is not None else None
     except (TemplateError, OffOriginError) as exc:
-        outcome.problem = str(exc)
+        outcome.problem, outcome.kind = str(exc), "journey error"
         return outcome
 
+    request = client.build_request(step.method, url, headers=req_headers, json=payload,
+                                   content=body)
     start = time.perf_counter()
     try:
-        resp = await client.request(step.method, url, headers=req_headers, json=payload,
-                                    content=body)
+        resp = await _send(client, request, jar)
     except OffOriginError as exc:
-        outcome.problem = str(exc)
+        outcome.problem, outcome.kind = str(exc), "left the app"
         return outcome
     except httpx.TimeoutException:
-        outcome.problem = "timed out"
+        outcome.problem, outcome.kind = "timed out", "timeout"
+        return outcome
+    except httpx.ConnectError:
+        outcome.problem, outcome.kind = "couldn't reach the app (ConnectError)", \
+            "connection refused"
         return outcome
     except httpx.HTTPError as exc:
         outcome.problem = f"couldn't reach the app ({exc.__class__.__name__})"
+        outcome.kind = "network"
         return outcome
     outcome.latency = time.perf_counter() - start
     outcome.status = resp.status_code
+    outcome.headers = resp.headers
 
     if not step.accepts(resp.status_code):
         wanted = (" or ".join(map(str, step.expect)) if step.expect else "a status below 400")
@@ -142,12 +218,32 @@ async def run_step(client: httpx.AsyncClient, step: Step, base_url: str,
         value = extract(resp, source)
         if value is None:
             outcome.problem = f"couldn't find {source} in the response to extract {name}"
+            outcome.kind = "missing extract"
             return outcome
         variables[name] = value
     outcome.ok = True
     return outcome
 
 
+async def _send(client: httpx.AsyncClient, request: httpx.Request,
+                jar: httpx.Cookies) -> httpx.Response:
+    """Send REQUEST with this run's cookies, following redirects one hop at a time
+    so every hop carries the cookies earlier hops set (a login that redirects to
+    /account stays logged in). The client keeps no cookies of its own, and its
+    same-origin hook checks each hop."""
+    for _ in range(_MAX_REDIRECTS + 1):
+        if "cookie" in request.headers:
+            del request.headers["cookie"]
+        jar.set_cookie_header(request)
+        resp = await client.send(request, follow_redirects=False)
+        jar.extract_cookies(resp)
+        if resp.next_request is None:
+            return resp
+        await resp.aclose()
+        request = resp.next_request
+    raise httpx.TooManyRedirects("too many redirects", request=request)
+
+
 def extract(resp: httpx.Response, source: str) -> Any:
     """Read `body.a.b.0` (JSON) or `header.Name` from a response; None if missing."""
     kind, _, where = source.partition(".")
diff --git a/cli/src/prescale_cli/loadtest.py b/cli/src/prescale_cli/loadtest.py
index f45ed1d..131d9f7 100644
--- a/cli/src/prescale_cli/loadtest.py
+++ b/cli/src/prescale_cli/loadtest.py
@@ -284,13 +284,15 @@ class SaturationInfo:
 
 class _Sink:
     """Per-stage accumulator, keyed by route label. A 5xx/429 is a failure;
-    everything else with a status code counts toward latency."""
+    everything else with a status code counts toward latency. Targets are URLs,
+    labelled by path; journeys pass their step labels as they are (`labels=False`)."""
 
-    def __init__(self) -> None:
+    def __init__(self, labels: bool = True) -> None:
         self.routes: dict[str, RouteStat] = {}
+        self._labels = labels
 
     def _stat(self, target: str) -> RouteStat:
-        label = route_label(target)
+        label = route_label(target) if self._labels else target
         stat = self.routes.get(label)
         if stat is None:
             stat = RouteStat()
@@ -389,8 +391,11 @@ async def _worker(client: httpx.AsyncClient, method: str, deadline: float,
 
 async def _run_stage(client: httpx.AsyncClient, targets: list[str], method: str,
                      users: int, duration: float, gate: _RateGate | None = None,
-                     think_time: float = 0.0) -> StageResult:
-    sink = _Sink()
+                     think_time: float = 0.0, vu=None) -> StageResult:
+    """Hold USERS virtual users for DURATION. By default each fires requests at
+    TARGETS; VU, when given, is the virtual user itself: an async callable
+    (client, deadline, sink, gate) that records its own requests (journeys)."""
+    sink = _Sink(labels=vu is None)
     cycle = itertools.cycle(targets)  # round-robin spreads load evenly across routes
     loop = asyncio.get_running_loop()
     start = loop.time()
@@ -399,10 +404,13 @@ async def _run_stage(client: httpx.AsyncClient, targets: list[str], method: str,
     lags: list[float] = []
     watcher = asyncio.create_task(_watch_loop_lag(stop, lags))
     try:
-        await asyncio.gather(
-            *(_worker(client, method, deadline, sink, lambda: next(cycle), gate, think_time)
-              for _ in range(users))
-        )
+        if vu is not None:
+            await asyncio.gather(*(vu(client, deadline, sink, gate) for _ in range(users)))
+        else:
+            await asyncio.gather(
+                *(_worker(client, method, deadline, sink, lambda: next(cycle), gate, think_time)
+                  for _ in range(users))
+            )
     finally:
         stop.set()
         await watcher
@@ -485,13 +493,21 @@ async def run_loadtest(
     transport: httpx.AsyncBaseTransport | None = None,
     headers: dict[str, str] | None = None,
     host_permission: bool = False,
+    vu=None,
+    client_options: dict | None = None,
+    preflight: bool = True,
 ) -> tuple[list[StageResult], str | None]:
     """Preflight the target, then ramp through `levels`, spreading each stage's
     load across `targets`. Stops early once a stage is more than `hard_stop_rate`
     failed. Returns (stages, warning).
 
     A host that restricts load testing (Vercel, Netlify, Cloudflare) stops the run
-    after the preflight unless `host_permission` says the user has permission."""
+    after the preflight unless `host_permission` says the user has permission.
+
+    `vu` replaces the default virtual user (see `_run_stage`), and
+    `client_options` adds settings to the HTTP client. Journeys use both, and skip
+    the preflight (`preflight=False`): each journey was already replayed once, which
+    checks the app better than a request to targets[0]."""
     if not targets:
         raise LoadError("No targets to test.")
     max_conns = max(levels) + 50
@@ -501,35 +517,17 @@ async def run_loadtest(
     gate = _RateGate(max_rps) if max_rps else None
     async with httpx.AsyncClient(
         timeout=timeout, limits=limits, follow_redirects=True,
-        headers=client_headers(headers), transport=transport,
+        headers=client_headers(headers), transport=transport, **(client_options or {}),
     ) as client:
-        try:
-            preflight = await client.request(method, targets[0])
-        except httpx.HTTPError as exc:
-            raise LoadError(f"Couldn't reach {targets[0]}: {exc}") from exc
-        blocker = edge_blocker(preflight.headers)
-        if blocker:
-            # Load now would only measure the firewall, and deepen the block.
-            raise LoadError(
-                f"{_capitalize(blocker)} is blocking requests from this machine (it answered "
-                f"the first request with HTTP {preflight.status_code}), so a load test would "
-                f"only measure the firewall. {_INSTEAD}"
-            )
-        rule = detect_host(preflight.headers)
-        if rule and not host_permission:
-            raise HostPolicyError(rule)
-        if preflight.status_code >= 400:
-            warning = (
-                f"First request returned HTTP {preflight.status_code} — "
-                "results may reflect a broken endpoint, not a load limit."
-            )
+        if preflight:
+            warning = await _preflight(client, method, targets[0], host_permission)
 
         if warmup:
             # Discarded: warms caches/JIT/connection pools so the first measured
             # level isn't cold. Still rate-gated, since it's real traffic.
             warmup_users, warmup_seconds = _warmup_plan(levels, stage_seconds)
             await _run_stage(client, targets, method, warmup_users, warmup_seconds,
-                             gate, think_time)
+                             gate, think_time, vu)
 
         # --repeat pools several ramps per level, so run-to-run variance (cache
         # state, GC, noisy neighbours) widens the confidence band honestly.
@@ -539,7 +537,7 @@ async def run_loadtest(
                 if progress_cb:
                     progress_cb(users)
                 stage = await _run_stage(client, targets, method, users, stage_seconds,
-                                         gate, think_time)
+                                         gate, think_time, vu)
                 by_level.setdefault(users, []).append(stage)
                 if on_stage:
                     on_stage(stage)
@@ -550,6 +548,31 @@ async def run_loadtest(
     return stages, warning
 
 
+async def _preflight(client: httpx.AsyncClient, method: str, target: str,
+                     host_permission: bool) -> str | None:
+    """One request before any load: stop if a firewall already blocks us or the
+    host forbids load tests; return a warning if the target answers with an error."""
+    try:
+        first = await client.request(method, target)
+    except httpx.HTTPError as exc:
+        raise LoadError(f"Couldn't reach {target}: {exc}") from exc
+    blocker = edge_blocker(first.headers)
+    if blocker:
+        # Load now would only measure the firewall, and deepen the block.
+        raise LoadError(
+            f"{_capitalize(blocker)} is blocking requests from this machine (it answered "
+            f"the first request with HTTP {first.status_code}), so a load test would "
+            f"only measure the firewall. {_INSTEAD}"
+        )
+    rule = detect_host(first.headers)
+    if rule and not host_permission:
+        raise HostPolicyError(rule)
+    if first.status_code >= 400:
+        return (f"First request returned HTTP {first.status_code} — "
+                "results may reflect a broken endpoint, not a load limit.")
+    return None
+
+
 def open_client(*, timeout: float = 10.0, max_conns: int = 64,
                 transport: httpx.AsyncBaseTransport | None = None,
                 headers: dict[str, str] | None = None) -> httpx.AsyncClient:
diff --git a/cli/src/prescale_cli/render.py b/cli/src/prescale_cli/render.py
index 49ae8a8..b7dafb3 100644
--- a/cli/src/prescale_cli/render.py
+++ b/cli/src/prescale_cli/render.py
@@ -4,6 +4,7 @@
 from __future__ import annotations
 
 from rich.console import Console
+from rich.markup import escape
 from rich.panel import Panel
 from rich.table import Table
 
@@ -93,6 +94,8 @@ def render_terminal(result: dict, *, show_ramp: bool = True) -> None:
         lines.append(f"Traffic  {traffic} [dim]({VISITOR_NOTE})[/dim]")
     if onset_users is not None:
         culprit = f"{verdict['culprit_route']}  " if (multi and verdict["culprit_route"]) else ""
+        if verdict.get("culprit_journey"):
+            culprit = f"{escape(verdict['culprit_journey'])} › {escape(culprit)}"
         if verdict["onset_reason"] == "latency":
             lines.append(f"Latency wall  {culprit}p95 crosses "
                          f"{latency_wall:g}s at ~{onset_users} {_u(onset_users)}.")
@@ -130,6 +133,41 @@ def render_terminal(result: dict, *, show_ramp: bool = True) -> None:
 
     if multi and stages:
         _render_routes(result)
+    if result.get("journeys"):
+        render_journeys(result)
+
+
+def render_journeys(result: dict) -> None:
+    """One row per journey: how many users it survives and the step that breaks
+    first. Each was ramped on its own, so no row depends on the others."""
+    table = Table(show_header=True, header_style="bold magenta",
+                  title="Journeys (each ramped on its own)")
+    for col, justify in (("Journey", "left"), ("Survives", "right"),
+                         ("Breaks at step", "left"), ("Why", "left")):
+        table.add_column(col, justify=justify)
+    first = result["verdict"].get("culprit_journey")
+    for j in result["journeys"]:
+        name = escape(j["name"])
+        if j["status"] != "ok":
+            label = "stale" if j["status"] == "stale" else "couldn't run"
+            table.add_row(name, f"[yellow]{label}[/yellow]", "—",
+                          f"[dim]{escape(j.get('reason') or '')}[/dim]")
+            continue
+        conf = j.get("confidence") or {}
+        band = ""
+        if conf.get("survives_low") is not None and conf.get("survives_high") is not None \
+                and conf["survives_low"] != conf["survives_high"]:
+            band = f" ({conf['survives_low']}–{conf['survives_high']})"
+        if j["onset_users"] is None:
+            survives = f"{j['max_tested']}+ [dim](held)[/dim]"
+        else:
+            survives = f"~{j['survives_users']}{band}"
+        step = j.get("culprit_step")
+        where = f"{step['index']}. {escape(step['name'])}  [dim]{escape(step['route'])}[/dim]" \
+            if step else "—"
+        style = "bold red" if j["name"] == first and j["onset_users"] is not None else None
+        table.add_row(name, survives, where, escape(j.get("bottleneck") or ""), style=style)
+    console.print(table)
 
 
 def _render_routes(result: dict) -> None:
diff --git a/cli/src/prescale_cli/schema/result.schema.json b/cli/src/prescale_cli/schema/result.schema.json
index bbc89b4..e2e406e 100644
--- a/cli/src/prescale_cli/schema/result.schema.json
+++ b/cli/src/prescale_cli/schema/result.schema.json
@@ -45,6 +45,11 @@
         "replica": {
           "type": "boolean",
           "description": "True when PreScale started the app's own build locally (--app) and tested that, not a deployed host."
+        },
+        "mode": {
+          "type": "string",
+          "enum": ["journeys"],
+          "description": "\"journeys\" when the run replayed journeys (prescale/journeys/); absent for a URL run."
         }
       }
     },
@@ -61,6 +66,10 @@
         "onset_users": { "type": ["integer", "null"] },
         "onset_reason": { "type": ["string", "null"], "enum": ["errors", "latency", null] },
         "culprit_route": { "type": ["string", "null"] },
+        "culprit_journey": {
+          "type": ["string", "null"],
+          "description": "Journey runs: the journey that breaks first. The verdict, stages and culprit_route (a step's method and path) are that journey's."
+        },
         "bottleneck": { "type": ["string", "null"] },
         "saturated": { "type": "boolean" },
         "saturation_users": { "type": ["integer", "null"] },
@@ -84,6 +93,11 @@
         "git_branch": { "type": ["string", "null"] }
       }
     },
+    "journeys": {
+      "type": "array",
+      "description": "Present on journey runs: one entry per journey. Each journey that replays is ramped on its own; a stale or unreachable one isn't load-tested and carries the reason.",
+      "items": { "$ref": "#/$defs/journey" }
+    },
     "investigation": {
       "type": "object",
       "description": "Present when `prescale investigate` diagnosed a failure.",
@@ -123,6 +137,55 @@
     }
   },
   "$defs": {
+    "journey": {
+      "type": "object",
+      "required": ["name", "type", "weight", "status", "reason"],
+      "properties": {
+        "name": { "type": "string" },
+        "type": { "type": "string" },
+        "weight": { "type": "number" },
+        "status": { "enum": ["ok", "stale", "unreachable"] },
+        "reason": { "type": ["string", "null"], "description": "Why a stale or unreachable journey wasn't load-tested." },
+        "survives_users": { "type": "integer" },
+        "max_tested": { "type": "integer" },
+        "onset_users": { "type": ["integer", "null"] },
+        "onset_reason": { "type": ["string", "null"], "enum": ["errors", "latency", null] },
+        "confidence": {
+          "type": "object",
+          "properties": {
+            "survives_low": { "type": ["integer", "null"] },
+            "survives_high": { "type": ["integer", "null"] },
+            "stable": { "type": "boolean" }
+          }
+        },
+        "culprit_step": {
+          "type": ["object", "null"],
+          "properties": {
+            "index": { "type": "integer", "description": "1-based step number." },
+            "name": { "type": "string" },
+            "route": { "type": "string" }
+          }
+        },
+        "bottleneck": { "type": ["string", "null"] },
+        "steps_at_users": { "type": ["integer", "null"], "description": "The level the step timings come from: where it broke, else the highest tested." },
+        "steps": {
+          "type": "array",
+          "items": {
+            "type": "object",
+            "required": ["index", "name", "route", "requests"],
+            "properties": {
+              "index": { "type": "integer" },
+              "name": { "type": "string" },
+              "route": { "type": "string" },
+              "requests": { "type": "integer" },
+              "p50_ms": { "type": ["integer", "null"] },
+              "p95_ms": { "type": ["integer", "null"] },
+              "error_rate": { "type": ["number", "null"] }
+            }
+          }
+        }
+      }
+    },
     "stage": {
       "type": "object",
       "required": [
diff --git a/cli/tests/test_compare.py b/cli/tests/test_compare.py
index 2f4a97c..4dd404a 100644
--- a/cli/tests/test_compare.py
+++ b/cli/tests/test_compare.py
@@ -103,3 +103,25 @@ def test_held_run_below_the_old_level_is_inconclusive_not_a_regression():
 def test_held_run_that_reached_the_old_level_still_compares():
     c = compare_results(_result(200, low=180, high=220), _result(190, held_at=190))
     assert not c["inconclusive"] and not c["regressed"]
+
+
+def _mode(result, mode=None, culprit=None):
+    r = json.loads(json.dumps(result))
+    if mode:
+        r["config"]["mode"] = mode
+    r["verdict"]["culprit_journey"] = culprit
+    return r
+
+
+def test_url_run_and_journey_run_are_not_comparable():
+    old = _result(200)
+    new = _mode(_result(20), mode="journeys", culprit="checkout")
+    cmp = compare_results(old, new)
+    assert not cmp["regressed"] and cmp["inconclusive"]
+    assert cmp["summary"].startswith("Not comparable")
+
+
+def test_journey_runs_note_a_new_first_to_break():
+    old = _mode(_result(100), mode="journeys", culprit="search")
+    new = _mode(_result(100), mode="journeys", culprit="checkout")
+    assert "search → checkout" in compare_results(old, new)["summary"]
diff --git a/cli/tests/test_journeys_load.py b/cli/tests/test_journeys_load.py
new file mode 100644
index 0000000..6d90c15
--- /dev/null
+++ b/cli/tests/test_journeys_load.py
@@ -0,0 +1,260 @@
+"""Journeys at load: virtual users walk whole journeys, each ramped on its own."""
+
+import asyncio
+import json
+
+import httpx
+from click.testing import CliRunner
+
+from prescale_cli.journeys.load import JourneyRun, first_to_break, journeys_block, ramp
+from prescale_cli.journeys.model import parse
+from prescale_cli.journeys.replay import check
+from prescale_cli.loadtest import analyze
+from prescale_cli.main import cli
+
+SHOP = {
+    "version": 1, "name": "checkout", "type": "checkout",
+    "steps": [
+        {"name": "log in", "request": {"method": "POST", "path": "/api/login"},
+         "expect": {"status": 200}, "extract": {"token": "body.token"}},
+        {"name": "place order",
+         "request": {"method": "POST", "path": "/api/orders",
+                     "headers": {"Authorization": "Bearer {{token}}"}},
+         "expect": {"status": 201}},
+    ],
+}
+BROWSE = {"version": 1, "name": "browse", "steps": [{"request": {"path": "/"}}]}
+
+
+def _shop(pool: int):
+    """A mock shop whose order step fails once more than POOL orders are in
+    flight at once, and which checks every order carries its own login's token
+    and cookie."""
+    state = {"in_flight": 0, "logins": 0, "mixed": 0}
+
+    async def handler(request: httpx.Request) -> httpx.Response:
+        if request.url.path == "/api/login":
+            state["logins"] += 1
+            token = f"t{state['logins']}"
+            return httpx.Response(200, json={"token": token},
+                                  headers={"set-cookie": f"sid={token}; Path=/"})
+        if request.url.path == "/api/orders":
+            token = request.headers["authorization"].removeprefix("Bearer ")
+            if request.headers.get("cookie") != f"sid={token}":
+                state["mixed"] += 1
+            if state["in_flight"] >= pool:
+                return httpx.Response(500, text="pool exhausted")
+            state["in_flight"] += 1
+            try:
+                await asyncio.sleep(0.02)
+            finally:
+                state["in_flight"] -= 1
+            return httpx.Response(201)
+        return httpx.Response(200)
+
+    return httpx.MockTransport(handler), state
+
+
+def _ramp(journey, transport, levels=(1, 2, 5, 10, 20)):
+    return asyncio.run(ramp(journey, "http://shop.test", levels=list(levels),
+                            stage_seconds=0.2, warmup=False, transport=transport))
+
+
+def test_ramp_records_each_step_and_finds_the_step_that_breaks():
+    transport, state = _shop(pool=4)
+    stages, _ = _ramp(parse(SHOP), transport)
+    assert set(stages[0].routes) == {"POST /api/login", "POST /api/orders"}
+    report = analyze(stages, latency_wall=2.0, error_threshold=0.02)
+    assert report.onset_reason == "errors"
+    assert report.culprit_route == "POST /api/orders"
+    assert 1 <= report.survives_users < 20
+    # every order went out with its own user's token and cookie
+    assert state["mixed"] == 0 and state["logins"] > 20
+
+
+def test_journeys_block_names_the_step_and_skips_stale_journeys():
+    transport, _ = _shop(pool=4)
+    checkout, browse = parse(SHOP), parse(BROWSE)
+    stages, _ = _ramp(checkout, transport)
+    ok = JourneyRun(checkout, asyncio.run(check(checkout, "http://shop.test", env={},
+                                                transport=transport)),
+                    report=analyze(stages, latency_wall=2.0, error_threshold=0.02))
+
+    def gone(request):
+        return httpx.Response(404, text="moved")
+    stale = JourneyRun(browse, asyncio.run(check(browse, "http://shop.test", env={},
+                                                 transport=httpx.MockTransport(gone))))
+
+    block = journeys_block([ok, stale])
+    assert block[0]["status"] == "ok"
+    assert block[0]["culprit_step"] == {"index": 2, "name": "place order",
+                                        "route": "POST /api/orders"}
+    assert [s["route"] for s in block[0]["steps"]] == ["POST /api/login", "POST /api/orders"]
+    assert block[0]["steps"][1]["error_rate"] > 0
+    assert block[1] == {"name": "browse", "type": "browse", "weight": 60.0,
+                        "status": "stale",
+                        "reason": "step 1 (step 1): returned 404, expected a status "
+                                  "below 400: moved"}
+    assert first_to_break([ok, stale]) is ok
+    json.dumps(block)  # it goes into the saved Result
+
+
+def test_first_to_break_prefers_the_lowest_capacity():
+    transport, _ = _shop(pool=4)
+    weak = JourneyRun(parse(SHOP), None,
+                      report=analyze(_ramp(parse(SHOP), transport)[0],
+                                     latency_wall=2.0, error_threshold=0.02))
+    strong = JourneyRun(parse(BROWSE), None,
+                        report=analyze(_ramp(parse(BROWSE), transport)[0],
+                                       latency_wall=2.0, error_threshold=0.02))
+    assert strong.report.onset_users is None
+    assert first_to_break([strong, weak]) is weak
+    assert first_to_break([JourneyRun(parse(BROWSE), None)]) is None
+
+
+# --- `prescale run` picks journeys ---------------------------------------------
+
+def _dir(tmp_path, *docs):
+    for doc in docs:
+        (tmp_path / f"{doc['name']}.yaml").write_text(json.dumps(doc))
+    return str(tmp_path)
+
+
+def _out(res):
+    return " ".join(res.output.split())
+
+
+def test_run_refuses_journey_writes_on_a_non_local_host(tmp_path):
+    res = CliRunner().invoke(cli, ["run", "https://shop.example.com", "--i-own-this",
+                                   "--dir", _dir(tmp_path, SHOP)])
+    assert res.exit_code == 2
+    assert "--allow-writes shop.example.com" in _out(res)
+
+
+def test_run_stops_on_a_broken_journey_file(tmp_path):
+    res = CliRunner().invoke(cli, ["run", "http://localhost:1",
+                                   "--dir", _dir(tmp_path, {**BROWSE, "version": 9})])
+    assert res.exit_code == 2
+    assert "version must be 1" in _out(res)
+
+
+def test_run_journey_flag_conflicts_with_url_mode(tmp_path):
+    res = CliRunner().invoke(cli, ["run", "http://localhost:1", "--urls", "--journey", "x",
+                                   "--dir", _dir(tmp_path, BROWSE)])
+    assert res.exit_code == 2
+    assert "--journey can't be combined" in _out(res)
+
+
+def test_run_unknown_journey(tmp_path):
+    res = CliRunner().invoke(cli, ["run", "http://localhost:1", "--journey", "nope",
+                                   "--dir", _dir(tmp_path, BROWSE)])
+    assert res.exit_code == 2
+    assert "no journey named nope" in _out(res)
+
+
+# --- review fixes -------------------------------------------------------------
+
+LOGIN_REDIRECT = {
+    "version": 1, "name": "account", "type": "account",
+    "steps": [
+        {"name": "log in", "request": {"method": "POST", "path": "/login"},
+         "expect": {"status": 200}, "extract": {"user": "body.user"}},
+        {"name": "update cart", "request": {"method": "POST", "path": "/cart"},
+         "expect": {"status": 200}, "extract": {"again": "body.user"}},
+    ],
+}
+
+
+def _login_redirects():
+    """POST /login sets a cookie on a 302; POST /cart answers 303. The pages they
+    redirect to only know the user if the cookie came along."""
+    logins = iter(range(10_000))
+
+    def handler(request):
+        cookie = request.headers.get("cookie")
+        if request.url.path == "/login":
+            return httpx.Response(302, headers={"location": "/me",
+                                                "set-cookie": f"sid=u{next(logins)}; Path=/"})
+        if request.url.path == "/cart":
+            return httpx.Response(303, headers={"location": "/cartview"})
+        if request.url.path in ("/me", "/cartview"):
+            return httpx.Response(200, json={"user": cookie} if cookie else {})
+        if request.url.path == "/":
+            return httpx.Response(302, headers={"location": "https://sso.example.com/"})
+        return httpx.Response(404)
+
+    return httpx.MockTransport(handler)
+
+
+def test_cookies_survive_redirects_in_verify():
+    result = asyncio.run(check(parse(LOGIN_REDIRECT), "http://shop.test", env={},
+                               transport=_login_redirects()))
+    assert result.ok, result.failed and result.failed.problem
+
+
+def test_cookies_survive_redirects_at_load_and_root_redirect_is_fine():
+    # "/" redirects to an SSO host: journeys don't probe it, so no OffOriginError.
+    stages, warning = _ramp(parse(LOGIN_REDIRECT), _login_redirects(), levels=(1, 5))
+    assert warning is None
+    for stage in stages:
+        assert all(stat.errors == 0 for stat in stage.routes.values()), stage.routes
+
+
+def _fake_checks(monkeypatch, stale_names=()):
+    """Replace the network in `run`'s journey mode: CHECK says which journeys are
+    stale; RAMP returns a run that holds at every level."""
+    from prescale_cli.commands import run as run_cmd
+    from prescale_cli.journeys.replay import JourneyCheck, StepCheck
+    from prescale_cli.loadtest import RouteStat, StageResult
+
+    async def fake_check_all(journeys, url, *, timeout, headers):
+        out = []
+        for j in journeys:
+            c = JourneyCheck(j)
+            ok = j.name not in stale_names
+            c.steps = [StepCheck(s, status=200 if ok else 404, ok=ok,
+                                 problem=None if ok else "returned 404, expected 200")
+                       for s in j.steps[:1]] if not ok else \
+                [StepCheck(s, status=200, ok=True) for s in j.steps]
+            out.append(c)
+        return out
+
+    async def fake_ramp(journey, url, *, levels, **kwargs):
+        routes = {s.label: RouteStat(total=10, latencies=[0.01] * 10) for s in journey.steps}
+        return [StageResult(users=u, duration=1.0, routes=dict(routes)) for u in levels], None
+
+    monkeypatch.setattr(run_cmd, "_check_all", fake_check_all)
+    monkeypatch.setattr(run_cmd, "ramp", fake_ramp)
+
+
+def test_run_json_stays_pure_when_a_journey_is_stale(tmp_path, monkeypatch):
+    _fake_checks(monkeypatch, stale_names={"browse"})
+    res = CliRunner().invoke(cli, ["run", "http://localhost:1", "--json", "--no-save",
+                                   "-u", "5", "--dir", _dir(tmp_path, SHOP, BROWSE)])
+    assert res.exit_code == 0, res.output
+    result = json.loads(res.stdout)  # nothing but the Result on stdout
+    assert "stale, not load-tested" in res.stderr
+    statuses = {j["name"]: j["status"] for j in result["journeys"]}
+    assert statuses == {"browse": "stale", "checkout": "ok"}
+    # nothing broke, so there's no culprit journey
+    assert result["verdict"]["onset_users"] is None
+    assert result["verdict"]["culprit_journey"] is None
+    assert result["config"]["mode"] == "journeys"
+
+
+def test_run_applies_think_time_to_steps_without_their_own(tmp_path, monkeypatch):
+    seen = {}
+    _fake_checks(monkeypatch)
+    from prescale_cli.commands import run as run_cmd
+    inner = run_cmd.ramp
+
+    async def spy(journey, url, **kwargs):
+        seen[journey.name] = kwargs.get("think")
+        return await inner(journey, url, **kwargs)
+
+    monkeypatch.setattr(run_cmd, "ramp", spy)
+    res = CliRunner().invoke(cli, ["run", "http://localhost:1", "--json", "--no-save",
+                                   "-u", "5", "--think-time", "1.5",
+                                   "--dir", _dir(tmp_path, BROWSE)])
+    assert res.exit_code == 0, res.output
+    assert seen == {"browse": 1.5}