feat(journeys): investigate diagnoses the journey step that breaks first

CHANGELOG.md

@@ -39,6 +39,16 @@
   pinned older `prescale-version` keeps working). Until the dashboard compares journey runs only
   with journey baselines, a journey run's PR comment shows no change instead of a possibly
   misleading one, and never fails on it.
+- **`investigate` diagnoses the journey step that breaks first.** With journeys in
+  `prescale/journeys/`, `prescale investigate` ramps them like `run`, then probes the step that
+  broke first with the journey's own state (its 1-user latency comes from replaying the journey
+  from the start), compares it with the journey's other steps, and names the bottleneck and the
+  fix: `culprit checkout, step 3 place order · connection_pool (high confidence)`. If the
+  journeys ramped together starved that journey, the evidence says so; if every journey held alone
+  but they broke together, the step that broke together is diagnosed. A step that answers an
+  unexpected 4xx under load gets a new class, `unexpected_answer` (users conflicting over shared
+  data, or sessions failing under load). The `investigation` field gains `culprit_journey`,
+  `culprit_step` and `interference`. `--urls` investigates URLs as before.
 - **`--expect N`** on `run` says whether you'd hold N users at peak. In a journey run, N is split
   across journeys by weight: each must hold its share alone, and all together must hold N.
 

README.md

@@ -167,6 +167,8 @@ Try this
   → Add a pooler (e.g. pgbouncer) and check the pool checkout timeout.
 ```
 
+With journeys in `prescale/journeys/`, `investigate` diagnoses the journey step that breaks first instead, reaching it the way a user does (logged in, cart filled): `culprit checkout, step 3 place order`. The journey's other steps play the part of the static assets. `--urls` investigates URLs.
+
 ### `journeys` — your users' real paths
 
 A journey is a path a user takes through your app (log in, add to cart, place an order), saved as the requests it makes. Journeys live in your repo as `prescale/journeys/<name>.yaml`:

cli/pyproject.toml

@@ -40,6 +40,7 @@ mcp = [
 dev = [
     "pytest>=7.0.0",
     "pytest-cov>=4.0.0",
+    "jsonschema>=4.0",
 ]
 
 [project.scripts]

cli/src/prescale_cli/commands/investigate.py

@@ -4,6 +4,7 @@
 
 import asyncio
 import json
+from pathlib import Path
 from urllib.parse import urlparse
 
 import click
@@ -13,15 +14,18 @@
 from prescale_cli.cloud import CloudError, require_credentials
 from prescale_cli.commands.audit import _ICONS as _AUDIT_ICONS
 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.run import _journeys_to_run, run_journeys
 from prescale_cli.commands.upload import header_option, upload_after_run, upload_option
-from prescale_cli.investigate import _LATENCY_WALL_S
+from prescale_cli.investigate import _ERROR_THRESHOLD, _LATENCY_WALL_S, audit_block
 from prescale_cli.investigate import investigate as run_investigate
+from prescale_cli.journeys.load import NoJourneyReplaysError
 from prescale_cli.live import LiveRamp, ramp_progress
 from prescale_cli.loadtest import LoadError
-from prescale_cli.profiles import PROFILES, lookup
+from prescale_cli.profiles import PROFILES, lookup, scenario_block
 from prescale_cli.render import render_investigation, render_terminal
-from prescale_cli.result import latest_id, load_result
+from prescale_cli.result import latest_id, load_result, write_result
 
 console = Console()
 
@@ -47,7 +51,18 @@
               help="Test against a launch scenario (see `prescale profiles`); ramps to 1.5x "
                    "its peak to show headroom.")
 @click.option("--audit", "with_audit", is_flag=True,
-              help="Also run the load-free HTTP hygiene audit (what Launch Guard shows).")
+              help="Also run the load-free HTTP hygiene audit.")
+@click.option("--journey", "only", multiple=True,
+              help="Investigate only this journey (repeatable). Default: every journey "
+                   "in prescale/journeys/.")
+@click.option("--no-combined", "no_combined", is_flag=True,
+              help="Skip ramping the journeys together (no interference check).")
+@click.option("--urls", "urls_only", is_flag=True,
+              help="Investigate URLs, even when prescale/journeys/ has journeys.")
+@click.option("--allow-writes", "allow_writes", multiple=True, metavar="HOST",
+              help="Let journeys that change data run against this non-local host, "
+                   "e.g. staging.myapp.com (repeatable).")
+@dir_option
 @app_option
 @app_timeout_option
 @host_permission_option
@@ -56,12 +71,16 @@
 def investigate(url: str | None, paths: tuple[str, ...], max_users: int,
                 stage_seconds: float, max_rps: float | None, yes: bool,
                 as_json: bool, store: str | None, fail_under: int | None,
-                profile_name: str | None, with_audit: bool, app: str | None,
-                app_timeout: float, host_permission: bool, headers: dict[str, str],
-                upload: bool) -> None:
+                profile_name: str | None, with_audit: bool, only: tuple[str, ...],
+                no_combined: bool, 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:
     """Find what breaks first, then probe it to explain WHY - and how to fix it.
 
-    With no URL, re-investigates the latest saved run's target.
+    When prescale/journeys/ has journeys, the journey and step that break first
+    are probed (with the journey's state, e.g. logged in, cart filled). Use
+    --urls to investigate URLs instead. With no URL, re-investigates the latest
+    saved run's target.
 
     \b
     Examples:
@@ -86,7 +105,8 @@ def investigate(url: str | None, paths: tuple[str, ...], max_users: int,
                           "[bold]prescale investigate <url>[/bold].[/dim]")
             raise SystemExit(0)
         url = load_result(last, store=store)["target"]["url"]
-        console.print(f"[dim]Re-investigating latest run: {url}[/dim]")
+        (Console(stderr=True) if as_json else console).print(
+            f"[dim]Re-investigating latest run: {url}[/dim]")
 
     prof = lookup(profile_name) if profile_name else None
     if profile_name and prof is None:
@@ -103,6 +123,12 @@ def investigate(url: str | None, paths: tuple[str, ...], max_users: int,
                       "(expected e.g. http://localhost:8000).")
         raise SystemExit(1)
 
+    journeys = _journeys_to_run(directory, only, urls_only=urls_only, url_mode=bool(paths))
+    if journeys:
+        host = (parsed.hostname or "").lower()
+        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)
 
@@ -123,6 +149,15 @@ def investigate(url: str | None, paths: tuple[str, ...], max_users: int,
         confirm_host(url, console=console, permitted=host_permission, as_json=as_json,
                      headers=headers)
 
+    if journeys:
+        result, live_mode = _investigate_journeys(
+            journeys, url, max_users=max_users, stage_seconds=stage_seconds,
+            max_rps=max_rps, prof=prof, headers=headers, replica=bool(app),
+            as_json=as_json, combined=not no_combined, with_audit=with_audit, store=store)
+        _show(result, as_json=as_json, live_mode=live_mode, upload=upload,
+              fail_under=fail_under)
+        return
+
     progress = ramp_progress(console, latency_wall=_LATENCY_WALL_S, as_json=as_json)
     live_mode = isinstance(progress, LiveRamp)
     try:
@@ -137,7 +172,43 @@ def investigate(url: str | None, paths: tuple[str, ...], max_users: int,
     except LoadError as exc:
         console.print(f"[red]Error:[/red] {exc}")
         raise SystemExit(1)
+    _show(result, as_json=as_json, live_mode=live_mode, upload=upload, fail_under=fail_under)
+
+
+def _investigate_journeys(journeys, url: str, *, max_users: int, stage_seconds: float,
+                          max_rps: float | None, prof, headers: dict[str, str],
+                          replica: bool, as_json: bool, combined: bool, with_audit: bool,
+                          store: str | None) -> tuple[dict, bool]:
+    """Ramp the journeys (as `run` does), then probe the step that broke first."""
+    config = {"method": "GET", "max_users": max_users, "stage_seconds": stage_seconds,
+              "latency_wall_s": _LATENCY_WALL_S, "error_threshold": _ERROR_THRESHOLD,
+              "max_rps": max_rps, "warmup": True, "repeat": 1,
+              "think_time_s": prof.think_time_s if prof else 0.0,
+              "profile": prof.name if prof else None}
+    err = Console(stderr=True) if as_json else console
+    try:
+        report, result, live_mode = run_journeys(
+            journeys, url, config, latency_wall=_LATENCY_WALL_S,
+            error_threshold=_ERROR_THRESHOLD, timeout=10.0, headers=headers,
+            replica=replica, as_json=as_json, combined=combined, diagnose=True)
+    except LoadError as exc:
+        err.print(f"[red]Error:[/red] {exc}")
+        raise SystemExit(1)
+    except NoJourneyReplaysError:
+        err.print("[red]Error:[/red] no journey replays against this app, so there's "
+                  "nothing to investigate. Run `prescale journeys verify` for details.")
+        raise SystemExit(1)
+    if prof is not None:
+        result["profile"] = scenario_block(prof, report.survives_users)
+    if with_audit:
+        result["audit"] = asyncio.run(audit_block(url, headers=headers))
+    write_result(result, store=store)
+    return result, live_mode
+
 
+def _show(result: dict, *, as_json: bool, live_mode: bool, upload: bool,
+          fail_under: int | None) -> None:
+    """Print, upload and gate an investigated Result (URLs or journeys alike)."""
     if as_json:
         click.echo(json.dumps(result, indent=2))
     else:
@@ -149,13 +220,25 @@ def investigate(url: str | None, paths: tuple[str, ...], max_users: int,
                 icon = _AUDIT_ICONS.get(finding["status"], "•")
                 console.print(f"{icon} [bold]{finding['name']}[/bold]  {finding['detail']}")
         if result.get("investigation") is None:
-            console.print("\n[green]✓[/green] Held up — nothing to diagnose.")
+            if (result.get("combined") or {}).get("onset_users") is not None \
+                    or result["verdict"]["onset_users"] is not None:
+                console.print("\n[yellow]⚠[/yellow] It broke, but PreScale couldn't probe the "
+                              "step that broke (see above).")
+            else:
+                console.print("\n[green]✓[/green] Held up — nothing to diagnose.")
         console.print(f"[dim]saved as {result['id']}[/dim]")
 
     if upload:
         upload_after_run(result, as_json=as_json)
 
     survives = result["verdict"]["survives_users"]
+    together = result.get("combined") or {}
+    if together.get("onset_users") is not None and together["survives_users"] < survives:
+        # The journeys broke sooner together than any did alone: that's the number to gate on.
+        survives = together["survives_users"]
+        if not as_json:
+            console.print(f"[dim]Together, the journeys survive ~{survives} users in total; "
+                          "--fail-under uses that.[/dim]")
     if fail_under is not None and survives < fail_under:
         Console(stderr=True).print(
             f"[red]✗ fail-under:[/red] survives ~{survives} < {fail_under}")

cli/src/prescale_cli/commands/run.py

@@ -21,6 +21,7 @@
 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.diagnose import diagnose_run
 from prescale_cli.journeys.load import (
     LABELS,
     JourneyRun,
@@ -386,12 +387,15 @@ def _journeys_to_run(directory: Path, only: tuple[str, ...], *, urls_only: bool,
 
 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, combined: bool = True) -> tuple:
+                  replica: bool, as_json: bool, combined: bool = True,
+                  diagnose: bool = False) -> tuple:
     """The isolation pass: check each journey once, then ramp each one that
     replays on its own; then, with two or more, the combined pass. Returns
     (report, result, live_mode) for the first journey to break, with every
-    journey's verdict in result["journeys"]. Raises NoJourneyReplaysError if no
-    journey replays, and LoadError if a ramp can't run. Shared by `run` and `ci`."""
+    journey's verdict in result["journeys"]. With DIAGNOSE (`investigate`), the
+    step that broke first is probed and result["investigation"] says why. Raises
+    NoJourneyReplaysError if no journey replays, and LoadError if a ramp can't
+    run. Shared by `run`, `investigate` and `ci`."""
     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]")
@@ -439,6 +443,7 @@ def run_journeys(journeys: list[Journey], url: str, config: dict, *, latency_wal
     result["journeys"] = journeys_block(runs)
 
     ramped = [r for r in runs if r.report is not None]
+    together = None
     if combined and len(ramped) > 1:
         if not as_json:
             mix = ", ".join(f"{escape(r.journey.name)} {r.journey.weight:g}" for r in ramped)
@@ -458,6 +463,17 @@ def run_journeys(journeys: list[Journey], url: str, config: dict, *, latency_wal
         result["combined"] = combined_block(together, runs)
         if together_warning and together_warning not in (result["warning"] or ""):
             result["warning"] = " ".join(filter(None, [result["warning"], together_warning]))
+
+    broke_together = together is not None and together.onset_users is not None
+    if diagnose and (worst.report.onset_users is not None or broke_together):
+        if not as_json:
+            console.print("\n[dim]Diagnosing the step that broke first…[/dim]")
+        investigation = asyncio.run(diagnose_run(
+            worst, runs, url, together=together, combined=result.get("combined"),
+            latency_wall=latency_wall, error_threshold=error_threshold, timeout=timeout,
+            headers=headers, max_rps=config["max_rps"]))
+        if investigation is not None:
+            result["investigation"] = investigation
     return worst.report, result, live_mode