@@ -68,6 +68,9 @@ python run.py --fresh-cache      delete the task cache of this run, then compute
 python run.py --serial           one process (easier to debug)
 python run.py --verify           also run the test suite and feed the release gate
 python run.py --view [FOLDER]    open existing results as report.html (+ tables.xlsx); runs nothing
+python run.py --status [FOLDER]  what a running (or the last) study is doing; runs nothing
+python run.py --log-level quiet|normal|verbose    how much a run prints (default normal)
+python run.py --heartbeat SECONDS                 seconds between heartbeats (default 60)
 ```
 
 `--view` is a viewer: it reads a results folder that already exists (default: `output_dir/run_name` of
@@ -77,6 +80,48 @@ python run.py --view [FOLDER]    open existing results as report.html (+ tables.
 Results are cached per task and keyed by a hash of every result-relevant input, so an unchanged re-run
 only rebuilds the report and changing one value only re-runs what depends on it.
 
+## Watching a run
+
+A full study takes hours, so it reports what it is doing. Every line is plain ASCII, `HH:MM:SS  message`
+(no colour, no `\r` progress bars, flushed), so it reads the same in PyCharm, in Git Bash and in a file
+(`python run.py > study_console.txt 2>&1`).
+
+| line | meaning |
+|---|---|
+| banner | git commit (`dirty` = tracked files differ), Python and library versions, mode, workers, the cache folder, tasks to run and cached, the time estimate, and the wave plan (which tasks wait for which) |
+| `wave 2 starts: ...` / `wave 2 ends ...` | a wave is the set of tasks that run together; the next one is planned from the results of the last |
+| `start`, `done`, `cached`, `FAILED  <task>` | one line per task, with its seconds; `start` is when the worker opened its state file |
+| `FT_grid:12000  seed 3/3 done  J=0.7741  96 evals  18.4 min` | one finished CMA-ES seed (`search done` adds the seed spread) |
+| `B1:S-03  9/17  S-03 solve attached ...: verified E=25.5 kJ (41 s)` | progress of a task that works in steps (B1 solves, sensitivity samples), printed at each quarter; `verbose` prints every one |
+| `heartbeat  1.2 h elapsed  tasks: 9 done, 22 running, 3 waiting` | followed by one line per running task: phase, seed `k/n`, iteration, best `J`, the task's elapsed time and a rough `~N min left` (from the pace so far, so only an estimate), or `no update for N min` when it has been silent for 5 min |
+| summary | every task with its seconds, the slowest task and the critical path (the slowest task of each wave, added), wall time against the estimate, failed tasks with the path of their log, and the next command |
+
+"Silent" means no progress event of any kind: a search reports every cost evaluation (the
+state file takes it at once, the log at most every 5 s), so a long first generation no longer
+looks stuck. The banner's time estimate is built from the times observed in the last full run
+(`TASK_COST` in `io/stages.py`): the longest task of the first wave plus the follow-up waves.
+
+The terminal shows a heartbeat block every 15 minutes at `normal` (every heartbeat at `verbose`, none at
+`quiet`), so a full clean study prints a few hundred lines. This is a deliberate limit: a line every
+60 s for 3.5 h would be 210 lines on its own. The heartbeat itself still runs every `--heartbeat` seconds
+and goes to the files below, and `--status` shows the same thing at any moment.
+
+Where things are (`<run>` is `output_dir/run_name`, `results/study` by default):
+
+* `<run>/_logs/run_<date>_<time>.log` the parent's whole timestamped log, everything the terminal got and
+  more (every heartbeat, in full); its path is the first line printed;
+* `<run>/_logs/<task>.log` one task's events with times, every search iteration included;
+* `<run>/_logs/<task>.state.json` the latest state of a task (replaced atomically about every 5 s);
+* `<run>/_logs/status.json` the whole run, written at every heartbeat.
+
+`python run.py --status [FOLDER]` reads these from any other terminal and prints what is going on now;
+with no logs in the folder it prints `no status in <folder>`, exits 2 and writes nothing. It cannot be
+combined with a run (`--only`, `--force`, ...), like `--view`. None of `--log-level`, `--heartbeat` and
+`--status` is part of any hash: another level is the same cache key.
+
+A task that fails prints the last lines of its traceback and the path of its log. If a worker process is
+killed (usually out of memory) the run prints one line saying so; use fewer `--workers`.
+
 ## Acceptance run
 
 A full study from nothing, then the viewer. From a fresh clone (the commands are the same in the
 
@@ -63,7 +63,7 @@ ignore_missing_imports = true
 root_package = "sagetrans"
 
 [[tool.importlinter.contracts]]
-name = "Layering: analysis > control > dynamics > physics > atmosphere > config"
+name = "Layering: analysis > control > dynamics > physics > atmosphere > config > progress"
 type = "layers"
 layers = [
     "sagetrans.io",
@@ -73,6 +73,7 @@ layers = [
     "(sagetrans.physics)",
     "sagetrans.atmosphere",
     "sagetrans.config",
+    "sagetrans.progress",  # imports nothing; every layer may report progress through it
 ]
 
 # docs/OUTPUTS_spec.md 1: the viewer reads a results folder; it imports no analysis code.
 
@@ -25,7 +25,7 @@
 
 import numpy as np
 
-from sagetrans import atmosphere
+from sagetrans import atmosphere, progress
 from sagetrans.analysis import restricted, runs
 from sagetrans.analysis.common import RunRecord, Scenario, make_record
 from sagetrans.analysis.ensemble import (
@@ -436,17 +436,18 @@ def lattice_descent(
     `cost` scores a full parameter dict; each distinct point is evaluated once.
     """
     seen: dict[tuple[float, ...], float] = {}
+    rounds = 0
 
     def score(p: dict[str, float]) -> float:
         key = tuple(round(p[n], 10) for n in LATTICE_NAMES)
         if key not in seen:
             seen[key] = float(cost(p))
+            progress.phase(f"lattice descent: round {rounds + 1}, {len(seen)} point(s) scored")
         return seen[key]
 
     cur = snap_ft(start, bounds)
     first = cur
     f_cur = f_first = score(cur)
-    rounds = 0
     while rounds < max_rounds:
         scored = [(score(c), c) for c in lattice_neighbours(cur, bounds)]
         if not scored:
@@ -497,7 +498,7 @@ def cost(u: np.ndarray) -> float:
         return ensemble_cost(metrics(bounds.to_physical(u)), weights)
 
     u0 = bounds.to_unit({n: start[n] for n in bounds.names})
-    res = cma_minimise(cost, u0, seeds, popsize=popsize, maxiter=maxiter)
+    res = cma_minimise(cost, u0, seeds, popsize=popsize, maxiter=maxiter, name="ft")
     best = full(bounds.to_physical(res.best_x))
     record: RunRecord = make_record(
         vp, scens, "ft", cfg, weights, bounds, pin, seeds, popsize, maxiter, start
 
@@ -222,7 +222,7 @@ def cost(u: np.ndarray) -> float:
         return ensemble_cost(metrics(bounds.to_physical(u)), weights)
 
     u0 = bounds.to_unit({**DEFAULT_PARAMS, **(start or {})})
-    res = cma_minimise(cost, u0, seeds, popsize=popsize, maxiter=maxiter)
+    res = cma_minimise(cost, u0, seeds, popsize=popsize, maxiter=maxiter, name=scope)
     best = bounds.to_physical(res.best_x)
     return RestrictedResult(
         scope, best, res.best_f, res.f_start, metrics(best), res,
@@ -410,7 +410,7 @@ def cost(u: np.ndarray) -> float:
         return ensemble_cost(metrics(bounds.to_physical(u)), weights)
 
     u0 = bounds.to_unit({"q": C_Q_START, **start})
-    res = cma_minimise(cost, u0, seeds, popsize=popsize, maxiter=maxiter)
+    res = cma_minimise(cost, u0, seeds, popsize=popsize, maxiter=maxiter, name="c")
     best = bounds.to_physical(res.best_x)
     return RestrictedResult(
         "c", best, res.best_f, res.f_start, metrics(best), res,
 
@@ -6,12 +6,15 @@
 
 from __future__ import annotations
 
+import time
 from collections.abc import Callable
 from dataclasses import dataclass
 
 import cma
 import numpy as np
 
+from sagetrans import progress
+
 
 @dataclass(frozen=True)
 class SeedResult:
@@ -60,25 +63,33 @@ def cma_minimise(
     popsize: int = 8,
     maxiter: int = 30,
     tolfun: float = 1e-6,
+    name: str = "search",
 ) -> SearchResult:
-    """Minimise fn over [0, 1]^n starting at u0; results are cached per rounded vector."""
+    """Minimise fn over [0, 1]^n starting at u0; results are cached per rounded vector.
+
+    `name` only labels the progress events (`sagetrans.progress`); it does not change the search.
+    """
     cache: dict[tuple[float, ...], float] = {}
 
     def cached(u: np.ndarray) -> float:
         key = tuple(np.round(np.clip(u, 0.0, 1.0), 9).tolist())
         if key not in cache:
             cache[key] = float(fn(np.clip(u, 0.0, 1.0)))
+            progress.search_eval(len(cache))
         return cache[key]
 
+    t_search = time.perf_counter()
+    progress.search_start(name, len(seeds), maxiter)
     f_start = cached(u0)
     out: list[SeedResult] = []
-    for seed in seeds:
+    for k, seed in enumerate(seeds, start=1):
+        t_seed = time.perf_counter()
         opts = {
             "seed": seed, "popsize": popsize, "maxiter": maxiter, "bounds": [0.0, 1.0],
             "verbose": -9, "tolfun": tolfun, "tolx": 1e-6,
         }  # fmt: skip
         es = cma.CMAEvolutionStrategy(list(u0), sigma0, opts)
-        best_u, best_f, n = np.array(u0), f_start, 0
+        best_u, best_f, n, it = np.array(u0), f_start, 0, 0
         while not es.stop():
             xs = es.ask()
             fs = [cached(np.array(x)) for x in xs]
@@ -87,6 +98,11 @@ def cached(u: np.ndarray) -> float:
             i = int(np.argmin(fs))
             if fs[i] < best_f:
                 best_f, best_u = fs[i], np.clip(np.array(xs[i]), 0.0, 1.0)
+            it += 1
+            progress.search_iter(k, it, n, best_f)
         out.append(SeedResult(seed, best_u, best_f, n))
+        progress.search_seed_done(k, best_f, n, time.perf_counter() - t_seed)
     best = min(out, key=lambda s: s.f)
-    return SearchResult(out, best.x, best.f, f_start)
+    res = SearchResult(out, best.x, best.f, f_start)
+    progress.search_done(res.best_f, res.spread, time.perf_counter() - t_search)
+    return res
@@ -24,7 +24,7 @@
 from SALib.sample import sobol as sobol_sample
 from scipy.stats import spearmanr
 
-from sagetrans import atmosphere
+from sagetrans import atmosphere, progress
 from sagetrans.analysis import braking, energy, restricted
 from sagetrans.analysis.common import RunRecord, Scenario, make_record
 from sagetrans.analysis.descent import DescentLimits
@@ -280,7 +280,8 @@ def oat(model: HeadlineModel) -> list[OATRow]:
     """Each input alone at +-50 % and +-100 % of its half-range (nominal elsewhere)."""
     base = model({})
     rows: list[OATRow] = []
-    for inp in model.inputs:
+    progress.phase("sensitivity: one at a time")
+    for n_in, inp in enumerate(model.inputs, start=1):
         pts = {
             "lo": inp.nominal - inp.half_range, "hi": inp.nominal + inp.half_range,
             "lo_half": inp.nominal - 0.5 * inp.half_range,
@@ -292,6 +293,7 @@ def oat(model: HeadlineModel) -> list[OATRow]:
                 OATRow(inp.name, out, inp.status, base[out], res["lo"][out], res["hi"][out],
                        res["lo_half"][out], res["hi_half"][out])
             )  # fmt: skip
+        progress.step(f"one at a time: {inp.name}", n_in, len(model.inputs))
     return rows
 
 
@@ -322,13 +324,18 @@ def _problem(inputs: Sequence[UncertainInput]) -> dict[str, Any]:
     }
 
 
-def _evaluate(model: HeadlineModel, x: np.ndarray) -> tuple[dict[str, np.ndarray], int]:
+def _evaluate(
+    model: HeadlineModel, x: np.ndarray, label: str = "sampling"
+) -> tuple[dict[str, np.ndarray], int]:
     names = [i.name for i in model.inputs]
     ys: dict[str, list[float]] = {o: [] for o in model.outputs}
-    for row in x:
+    every = max(1, len(x) // 30)  # a progress step per few percent of the sample
+    for i, row in enumerate(x, start=1):
         out = model(dict(zip(names, row.tolist(), strict=True)))
         for o in model.outputs:
             ys[o].append(out[o])
+        if i % every == 0 or i == len(x):
+            progress.step(f"{label}: {i} of {len(x)} samples", i, len(x))
     arr = {o: np.array(v) for o, v in ys.items()}
     n_failed = int(np.sum(np.isnan(np.column_stack(list(arr.values()))).any(axis=1)))
     better_high: frozenset[str] = getattr(model, "higher_is_better", frozenset())
@@ -343,7 +350,7 @@ def global_morris(model: HeadlineModel, r: int, seed: int = 0, levels: int = 4)
     """Morris elementary effects with `r` trajectories: index = mu_star."""
     prob = _problem(model.inputs)
     x = morris_sample.sample(prob, N=r, num_levels=levels, seed=seed)
-    ys, n_failed = _evaluate(model, x)
+    ys, n_failed = _evaluate(model, x, f"Morris r={r}")
     index, detail = {}, {}
     for o, y in ys.items():
         res = morris_analyze.analyze(prob, x, y, num_levels=levels, seed=seed)
@@ -356,7 +363,7 @@ def global_sobol(model: HeadlineModel, n: int, seed: int = 0) -> GlobalResult:
     """Sobol first- and total-order indices from a Saltelli sample of base size `n` (power of 2)."""
     prob = _problem(model.inputs)
     x = sobol_sample.sample(prob, n, calc_second_order=False, seed=seed)
-    ys, n_failed = _evaluate(model, x)
+    ys, n_failed = _evaluate(model, x, f"Sobol n={n}")
     index, detail = {}, {}
     for o, y in ys.items():
         res = sobol_analyze.analyze(prob, y, calc_second_order=False, seed=seed)
 
@@ -17,6 +17,7 @@ absolute imports mean it does not shadow the standard library.)
 | `recommended.py` | `settings_recommended.csv`: the ArduPilot parameters, the register, the recommended and the default values |
 | `assemble.py` | turns task results into tables, figures, `report.md`, `inputs_used.md`, the run record and the gate verdict |
 | `study.py` | `run_study`: load, check, run tasks in a spawn process pool with a per-task disk cache, assemble |
+| `runlog.py` | run logging: `WorkerLog` (a worker's progress sink: task log and state file), `RunLog` (the terminal lines, heartbeat, `status.json`, summary), `read_status` (`--status`) |
 | `cli.py` | `python run.py` / `python -m sagetrans`: `--check --mode --only --force --serial --verify`, and `--view [FOLDER] --no-open` |
 | `view/` | the results viewer (below): `report.html` and `tables.xlsx` from a results folder |
 | `ci_check.py` | `python -m sagetrans.io.ci_check junit.xml`: fails CI if any of T-01…T-15 is missing or failing |
 
@@ -10,6 +10,9 @@
     python run.py --serial                 # one process (easier to debug)
     python run.py --verify                 # also run the test suite and feed the release gate
     python run.py --view [FOLDER]          # open the results as report.html (+ tables.xlsx); no run
+    python run.py --status [FOLDER]        # what a running (or the last) study is doing; no run
+    python run.py --log-level verbose      # quiet | normal (default) | verbose
+    python run.py --heartbeat 30           # seconds between heartbeats (default 60)
 
 Exit codes: 0 ok, 1 a task failed, 2 the inputs are unusable (or release mode refused),
 3 release mode finished but the release gate blocked the result.
@@ -22,6 +25,7 @@
 from collections.abc import Sequence
 from pathlib import Path
 
+from sagetrans.io import runlog
 from sagetrans.io.inputs import InputError, load_inputs
 from sagetrans.io.settings import ALL_ANALYSES
 from sagetrans.io.study import EXIT_INPUTS, EXIT_OK, preflight, run_study
@@ -54,13 +58,43 @@ def build_parser() -> argparse.ArgumentParser:
                         "study.yaml): writes report.html and tables.xlsx there and opens the page; "
                         "runs nothing")
     p.add_argument("--no-open", action="store_true", help="with --view: do not open the browser")
+    p.add_argument("--log-level", choices=runlog.LEVELS, default="normal",
+                   help="how much a run prints: quiet (banner, failures, summary), normal, or "
+                        "verbose (every search iteration); the files always have everything")
+    p.add_argument("--heartbeat", type=float, default=60.0, metavar="SECONDS",
+                   help="seconds between heartbeats (status.json and the run log; the terminal "
+                        "block at `normal` comes every 15 min, at `verbose` every heartbeat)")
+    p.add_argument("--status", nargs="?", const="", default=None, metavar="FOLDER",
+                   help="print what a running (or the last) study in FOLDER is doing, from "
+                        "another terminal (default: output_dir/run_name of study.yaml); runs and "
+                        "writes nothing; exit 2 when there is no status")
     return p
 
 
 def main(argv: Sequence[str] | None = None) -> int:
     parser = build_parser()
     args = parser.parse_args(argv)
     config = Path(args.config)
+    if args.heartbeat <= 0:
+        parser.error("--heartbeat must be a positive number of seconds")
+    if args.status is not None:
+        clash = [f for f, on in (("--only", args.only), ("--force", args.force),
+                                 ("--verify", args.verify), ("--check", args.check),
+                                 ("--fresh-cache", args.fresh_cache),
+                                 ("--view", args.view is not None), ("--serial", args.serial),
+                                 ("--workers", args.workers)) if on]
+        if clash:
+            parser.error(f"--status only reads a running or finished study; it cannot be "
+                         f"combined with {', '.join(clash)}")
+        from sagetrans.io.view import default_folder
+
+        folder = Path(args.status) if args.status else default_folder(config, args.out)
+        lines = runlog.read_status(folder)
+        if lines is None:
+            print(f"no status in {folder}")
+            return EXIT_INPUTS
+        print("\n".join(lines))
+        return EXIT_OK
     if args.view is not None:
         clash = [f for f, on in (("--only", args.only), ("--force", args.force),
                                  ("--verify", args.verify), ("--check", args.check),
@@ -95,17 +129,19 @@ def main(argv: Sequence[str] | None = None) -> int:
     outcome = run_study(
         config, mode=args.mode, only=only, force=args.force, fresh_cache=args.fresh_cache,
         workers=1 if args.serial else args.workers, out_dir=args.out, junit=args.junit,
-        verify=args.verify,
+        verify=args.verify, log_level=args.log_level, heartbeat=args.heartbeat,
     )  # fmt: skip
     for m in outcome.messages:
-        print(m, file=sys.stderr)
+        print(runlog.ascii_line(m), file=sys.stderr)
     if outcome.folder is not None:
-        print(f"\nresults: {outcome.folder}")
-        print(f"  report:       {outcome.folder / 'report.md'}")
-        print(f"  run record:   {outcome.folder / 'run_record.json'}  [{outcome.label}]")
+        out = [f"\nresults: {outcome.folder}",
+               f"  report:       {outcome.folder / 'report.md'}",
+               f"  run record:   {outcome.folder / 'run_record.json'}  [{outcome.label}]"]
         if outcome.gate is not None:
-            print(outcome.gate.summary())
-        print(f"done in {outcome.seconds / 60:.1f} min")
+            out.append(outcome.gate.summary())
+        out.append(f"done in {outcome.seconds / 60:.1f} min")
+        for line in out:
+            print(runlog.ascii_line(line))
     return outcome.exit_code