feat(journeys): prescale run replays journeys at load

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

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`.
 

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:

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]

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']}"