release: 1.0.0

first stable release. pre-1.0.0 verification complete: all surviving MED regressions and
gaps resolved and independently re-fired, tree audited clean across the suite.

Signed-off-by: disqualifier <dev@disqualifier.me>

.gitignore

@@ -1,5 +1,5 @@
 # claude
-CLAUDE.md
+.claude/
 
 # python
 __pycache__/

README.md

@@ -13,11 +13,13 @@ and emit; their records flow into the handlers `log_setup` wired.
 ## Install
 
 ```
-log_setup @ git+ssh://git@git.rethinkstudios.io/rethink-public/log_setup.git@v0.2.0
+log_setup @ git+ssh://git@git.rethinkstudios.io/rethink-public/log_setup.git@v0.6.3
 ```
 
 No dependencies — stdlib only.
 
+Drop the `@v0.6.3` suffix from the line above to install the latest unpinned.
+
 ## Quick start
 
 ```python
@@ -41,17 +43,86 @@ emits; the records land in the configured root.
 - **Format:** `2026-06-27 19:55:05 | module.name | INFO | message`. `%(name)s` is the
   `getLogger` name each module used, so you see which lib/module logged.
 - **Rotation** (`rotate=`):
-  - `"daily"` (default) — rolls at midnight, dated name into `log_dir`, keeps
-    `backup_count` days.
-  - `"size"` — rolls at `max_bytes`, numbered backups in `log_dir`.
-  - `"on_start"` — on startup, moves an existing `run.log` into `log_dir`
-    (`run.<timestamp>.log[.gz]`) and starts fresh; prunes to `backup_count`.
+  - `"daily"` (default) — rolls at midnight into `log_dir`, keeps `backup_count` days.
+  - `"size"` — rolls at `max_bytes` into `log_dir`, keeps `backup_count`. `backup_count=0`
+    means **keep no rolled history**: the live file still rolls at `max_bytes` (size is
+    always bounded), each rolled file is deleted immediately after landing — it does not
+    disable rotation (see **Retention** below).
+  - `"on_start"` — on startup, moves an existing live file into `log_dir` and starts fresh;
+    prunes to `backup_count`.
   - `None` — single file, no rotation.
-- **compress=True** (default) gzips each rolled file (`run.log.2026-06-27.gz`).
-- **Retention** = `backup_count` (default 14) for every mode.
+- **Historic files are named off the project** — see below. Every rolled file is
+  `<project>.<timestamp>.log[.gz]`; the live file keeps its own name.
+- **compress=True** (default) gzips each rolled file.
+- **Retention** = `backup_count` (default 14) for every mode — unless tiered retention is
+  enabled (below). For `rotate="size"`, `backup_count=0` is "keep none" (not "disable
+  rotation") — see the `size` bullet above and the note at the bottom of this section.
 - **console=True** (off by default) also logs to stdout in the same format — opt in when
   you want live terminal output alongside the file.
 
+The `name` you pass is normalized so it produces exactly one `.log`: `name="latest"` and
+`name="latest.log"` both yield the live file `latest.log` (never `latest.log.log`).
+
+## Historic files are named off the project (`history_name`)
+
+The **live** file keeps its defined `name` (`latest.log`). The **historic** (rolled/gz)
+files are named off the **project namespace** — by default the current directory's basename
+— so you can tell at a glance which service a log came from:
+
+```python
+# app run from  bestbuy/run.py , with name="latest":
+setup_logging(name="latest", rotate="daily")
+# logs/
+#   latest.log                          <- live (the tail -f target)
+#   bestbuy.2026-07-01_02-00-00.log     <- historic, named off the project dir
+#   bestbuy.2026-06-30_02-00-00.log.gz
+```
+
+- **Default** = `os.path.basename(os.getcwd())` (the project directory). Zero config.
+- Override with **`history_name="foo"`** → historic files become `foo.<timestamp>.log[.gz]`.
+- This changed in **v0.5.0**: historic files used to reuse the live `name`. To keep the old
+  behavior, pass `history_name=name`.
+- Retention (tier counts / `backup_count`) is unchanged — it's just keyed to the project
+  stem now.
+
+## Tiered retention (`keep_uncompressed` / `keep_compressed`)
+
+The default is a flat `backup_count`: every rolled file is gzipped on roll and the oldest
+are deleted past the count. If instead you want the recent logs **uncompressed** (read them
+without `zcat`) and older ones **gzipped**, pass the two tier knobs:
+
+```python
+# app run from  bestbuy/run.py , with name="latest":
+setup_logging(
+    name="latest",
+    rotate="on_start",        # works for on_start, daily, and size
+    keep_uncompressed=3,      # newest 3 rolled logs kept PLAIN
+    keep_compressed=7,        # next 7 kept GZIPPED; total retained = 10
+)
+```
+
+Result — the live file stays at its stable path in cwd; `log_dir` (default `logs/`) holds
+the tiered historic files, named off the project stem (newest → oldest):
+
+```
+./latest.log                                              <- live (stable, tail -f, in cwd)
+logs/
+    bestbuy.<t1>.log  bestbuy.<t2>.log  bestbuy.<t3>.log   <- 3 newest: plain
+    bestbuy.<t4>.log.gz ... bestbuy.<t10>.log.gz           <- next 7: gzipped
+    (anything past 10 deleted)
+```
+
+- Each restart (`on_start`) or roll (`daily`/`size`) moves the live file into `log_dir`,
+  then re-tiers: newest `keep_uncompressed` stay plain, the next `keep_compressed` are
+  gzipped in place, the rest deleted. Total kept = `keep_uncompressed + keep_compressed`.
+- **Opt-in by presence** — pass either knob to enable tiering. Pass **neither** and
+  rotation behaves exactly as before (`backup_count` + gzip-on-roll), so existing callers
+  are unaffected.
+- In tiered mode `backup_count` and the gzip-on-roll behavior of `compress` are **ignored**
+  — the tier counts bound retention instead.
+- `keep_uncompressed=0` → everything gzipped; `keep_compressed=0` → only the plain tier.
+  Retention is count-based (not time-based).
+
 ## Output format (`output=`)
 
 Two formats, two needs. Default is `"text"`; the live-file name is the same either way
@@ -68,17 +139,22 @@ Promtail glob, bind-mount path, or your `tail` command.
 ```python
 setup_logging(name="run", output="json")
 logging.getLogger("bot.core").info("ready", extra={"monitor": "heartbeat"})
-# -> {"time": "2026-06-28T14:03:11Z", "level": "INFO", "module": "bot.core",
-#     "message": "ready", "monitor": "heartbeat"}
+# -> {"time": "2026-06-28T14:03:11Z", "ts": 1782151391, "level": "INFO",
+#     "module": "bot.core", "message": "ready", "monitor": "heartbeat"}
 ```
 
-- **Fields:** `time`, `level`, `module`, `message` always; any `extra={...}` keys land
-  as **top-level** fields (stamp `monitor`/`service`/request-id for Loki labels — the lib
-  stays domain-agnostic); error records carry the traceback in `exc_info` (never dropped).
+- **Fields:** `time`, `ts`, `level`, `module`, `message` always; any `extra={...}` keys
+  land as **top-level** fields (stamp `monitor`/`service`/request-id for Loki labels —
+  the lib stays domain-agnostic); error records carry the traceback in `exc_info` (never
+  dropped).
 - **Time is UTC ISO-8601 with a `Z`** (`2026-06-28T14:03:11Z`), not local. json is the
   aggregation path — logs from many servers/containers sort unambiguously only in UTC;
   Grafana converts to local for display. (Text mode stays local — that's a human on one
   box.)
+- **`ts` (added v0.6.0)** is the same instant as a unix epoch integer
+  (`int(record.created)`, second resolution) alongside `time` — for a consumer that wants
+  a sortable number instead of parsing the ISO string. Additive: existing `time` is
+  unchanged, and a consumer that ignores unknown JSON keys is unaffected.
 - Both file and console use the chosen format. `fmt`/`datefmt` apply to text only (json
   builds fields, not a format string). An unknown `output` falls back to text + warns,
   never crashes. **Zero new deps** — stdlib `json` only.
@@ -89,9 +165,13 @@ logging.getLogger("bot.core").info("ready", extra={"monitor": "heartbeat"})
 setup_logging(
     name="run",               # base -> run.log (the live file at cwd)
     log_dir="logs",           # rotated/compressed copies live here (created if absent)
-    level="INFO",             # root level (str name or logging constant)
+    level="INFO",             # root level everything inherits (str name or logging constant)
+    module_levels=None,       # {logger_name: level} per-logger overrides (exact name match)
     rotate="daily",           # "daily" | "size" | "on_start" | None
-    backup_count=14,          # rotated files to keep (older auto-deleted)
+    history_name=None,        # stem for rolled/historic files; None -> cwd basename (project)
+    backup_count=14,          # rotated files to keep (flat retention; ignored if tiered)
+    keep_uncompressed=None,   # tiered: newest N rolled logs kept PLAIN (opt-in)
+    keep_compressed=None,     # tiered: next M rolled logs kept GZIPPED (opt-in)
     max_bytes=10_000_000,     # only for rotate="size"
     compress=True,            # gzip rolled files
     console=False,            # also log to stdout (off by default; opt in)
@@ -102,6 +182,47 @@ setup_logging(
 ) -> logging.Logger           # returns the configured root logger
 ```
 
+## Quieting noisy dependencies (`module_levels`)
+
+`level` is the **root default** — every logger inherits it. `module_levels` is an
+optional `{logger_name: level}` map of **per-logger overrides** applied at setup, the
+standard "turn down the chatty dependency while my own code stays at INFO" case:
+
+```python
+setup_logging(
+    name="run",
+    level="INFO",                 # our code logs at INFO
+    module_levels={
+        "motor": "WARNING",       # quiet the driver
+        "pymongo": "WARNING",
+        "aiohttp": "WARNING",     # also quiets aiohttp.client / aiohttp.access (hierarchy)
+    },
+)
+```
+
+- **Exact-name match — names are NOT discovered.** It calls
+  `logging.getLogger(name).setLevel(level)` for exactly the name you give. There's no
+  smart find of noisy modules; you name the loggers. A typo (`"moter"`) silently
+  configures a logger nothing uses — no error, no effect. Get the names right.
+- **Hierarchy applies** (the one "smart" part, and it's just stdlib): naming a **parent**
+  quiets its whole subtree. `"aiohttp"` also quiets `aiohttp.client`, `aiohttp.access`,
+  etc. — the way to catch sub-loggers without listing each.
+- **str or int** per entry (`"WARNING"` or `logging.WARNING`) — same normalization as the
+  root `level`.
+- **Never crashes:** a bad level for one entry is **skipped with a warning**; the other
+  entries and the rest of setup still apply. Consistent with the never-crash-over-logging
+  rule.
+- `None`/`{}` (default) → no overrides; existing callers are unaffected.
+
+Common noisy library logger names: `motor`, `pymongo`, `aiohttp` (parent quiets
+`aiohttp.client`/`aiohttp.access`), `discord` / `discord.*`, `asyncio`, `urllib3`. Check
+a lib's *actual* logger name — some log under a name different from their package.
+
+This already works without the lib (`logging.getLogger("motor").setLevel(WARNING)` after
+setup does the same via stdlib hierarchy). The param's value is ergonomic: it keeps the
+overrides in the **one** `setup_logging` call at the entry point instead of scattering
+`setLevel` calls afterward — which is the whole point of `log_setup`.
+
 ## Async-friendly (`queue=True`)
 
 For async-heavy apps, `queue=True` routes records through a stdlib `QueueHandler` to a
@@ -119,6 +240,42 @@ setup_logging(name="run", queue=True)
   duplicate lines) and leaves handlers your app added itself alone.
 - **Never crashes the app over logging:** if `log_dir` isn't writable, it falls back to
   console-only with a warning instead of raising.
+- **`rotate="size"` always bounds the live file (v0.5.1+).** Previously, `backup_count=0`
+  with `rotate="size"` silently disabled rotation entirely (the live file grew forever,
+  ignoring `max_bytes`). As of v0.5.1, the live file always rolls at `max_bytes`
+  regardless of `backup_count`; `backup_count=0` means "keep zero rolled files" (each roll
+  is deleted right after it lands) rather than "never roll." `backup_count>=1` behaves as
+  documented (keeps that many rolled files). This does not change `"daily"`/`"on_start"`,
+  where `backup_count=0` still means "roll, but don't prune the rolled files" (unbounded
+  `log_dir` growth) — that is a separate, pre-existing knob, not this fix's scope.
+- **`console=True` writes to stdout (v0.6.3).** the console handler now binds `sys.stdout`
+  explicitly instead of the stdlib `StreamHandler` default of `sys.stderr`, so a supervisor
+  capturing stdout sees console log lines. `console` stays off by default — the file is the
+  primary sink.
+- **`"daily"` regression fixed (v0.6.2).** v0.5.1's `rotate="size"` fix above shared its
+  rotator with `"daily"`, so a `rotate="daily", backup_count=0` roll was incorrectly
+  deleted at every midnight rollover instead of just landing unpruned. The rotator is now
+  rotate-mode aware: the zero-retention delete only ever fires for `"size"`, matching the
+  contract in the bullet above — `"daily"`/`"on_start"` with `backup_count=0` were always
+  meant to roll without pruning and now do again.
+- **Gzip is atomic compress-or-skip (v0.5.1+).** `_gzip_file` writes to a `.tmp` sibling,
+  verifies it decompresses cleanly, and only then `os.replace`s it onto the final `.gz` and
+  removes the plain source — a crash/OOM/interrupt at any point leaves the plain `.log`
+  intact with no `.gz` (or the `.tmp` cleaned up), so this code never produces a truncated
+  `.gz` at the path retention trusts. Because a corrupt `.gz` can't arise from this path, the
+  tiered retention dedupe drops a plain twin unconditionally when its `.gz` exists (no
+  runtime `_gz_intact` reconciliation — that was removed once compression became atomic).
+- **Setup-time warnings reach the log file (v0.6.0+).** Previously, a warning raised
+  during `setup_logging` itself (an invalid `module_levels` entry, a handler failing to
+  close on re-setup, an unknown `rotate` value) was emitted *before* any handler was
+  attached, so it only reached stderr via logging's `lastResort` fallback and never
+  `run.log`. As of v0.6.0 these are buffered and flushed once the handlers are attached,
+  so they land in the configured log like any other record.
+- **JSON output gained a `ts` field (v0.6.0, additive).** Alongside the existing `time`
+  (UTC ISO-8601, unchanged), each JSON line now also carries `ts`: the same instant as a
+  unix epoch integer (`int(record.created)`, second resolution) — for a consumer that
+  wants a sortable number instead of parsing the ISO string. Purely additive: `time` is
+  byte-for-byte unchanged, and a consumer that ignores unknown JSON keys is unaffected.
 
 ## Scope — what this is NOT
 
@@ -135,4 +292,4 @@ color formatting, per-logger filters, remote handlers.
 
 ## Versioning
 
-Tagged `vX.Y.Z`. Pin the tag.
+Releases are tagged `vX.Y.Z`. The install line above pins a release; drop the `@vX.Y.Z` suffix to install the latest unpinned. Pin deliberately for reproducible installs.

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "log_setup"
-version = "0.2.0"
+version = "1.0.0"
 description = "stdlib app-entry-point logging setup: live run.log, rotation, gzip, retention, consistent format"
 requires-python = ">=3.10"
 dependencies = []

src/log_setup/__init__.py

@@ -1,22 +1,21 @@
-"""log_setup — app-entry-point logging configuration (sync, stdlib only).
-
-call once at an application's entry point to configure the whole process: a live
-run.log, rotation (daily/size/on_start), gzip of rolled files, retention, console
-output, and a consistent `time | module | level | message` format.
+"""log_setup - app-entry-point logging configuration (sync, stdlib only). see README.
 
     from log_setup import setup_logging
 
     setup_logging(name="run", level="INFO")        # daily rotation, logs/ dir, gzip
     log = logging.getLogger(__name__)
-    log.info("up")                                  # -> run.log + console
+    log.info("up")                                  # -> run.log (add console=True for stdout too)
 
-reusable libraries do NOT call this — they only `logging.getLogger(__name__)` and
-emit; the application owns this setup. shipping logs to a backend is out of scope
-(that's Promtail's job against the produced files).
+reusable libraries do NOT call this; the application owns setup, libraries only emit.
 """
 
+from importlib.metadata import PackageNotFoundError, version
+
 from .setup import setup_logging
 
 __all__ = ["setup_logging"]
 
-__version__ = "0.2.0"
+try:
+    __version__ = version("log_setup")
+except PackageNotFoundError:
+    __version__ = "0.0.0+unknown"

src/log_setup/formats.py

@@ -1,10 +1,8 @@
 """log formats for the app-wide setup: human-readable text + structured JSON lines.
 
-two output formats, two proven needs. `text` (default) is the human `tail -f` format
-(`time | module | level | message`, local time). `json` is the Grafana/Loki path —
-one JSON object per line (JSON Lines), fields parsed into labels natively, UTC
-timestamps so logs aggregated from many machines/containers sort unambiguously.
-`%(name)s` is the getLogger name the emitting module used, so each module shows.
+`text` (default) is the human `tail -f` format (`time | module | level | message`,
+local time). `json` is the Grafana/Loki path - one JSON object per line, UTC
+timestamps so logs aggregated from many machines sort unambiguously. see README.
 """
 
 import datetime
@@ -15,33 +13,40 @@ DEFAULT_FORMAT = "%(asctime)s | %(name)s | %(levelname)s | %(message)s"
 DEFAULT_DATEFMT = "%Y-%m-%d %H:%M:%S"
 
 _RESERVED = frozenset(vars(logging.makeLogRecord({})).keys()) | {"message", "asctime"}
+# this formatter's own canonical output keys - stdlib's LogRecord rejects `extra` keys
+# colliding with real attribute names (e.g. `module`), but `time`/`level`/`ts` are NOT
+# LogRecord attrs, so a caller's extra={"time":...}/{"ts":...} would otherwise overwrite
+# the UTC timestamp / epoch. guard them explicitly
+_OUTPUT_KEYS = frozenset({"time", "ts", "level", "module", "message"})
 
 
 class JsonLinesFormatter(logging.Formatter):
     """format each record as a single-line JSON object (JSON Lines / .jsonl)
 
-    emits at minimum time/level/module/message. time is UTC ISO-8601 with a `Z`
-    suffix (e.g. 2026-06-28T14:03:11Z) so logs aggregated across machines and
-    containers sort unambiguously — Grafana converts to local for display. any
-    field passed via logging `extra={...}` lands as a top-level JSON field, which
-    is how a caller stamps monitor/service/request-id for Loki labels without the
-    lib knowing those domain concepts. a traceback (exc_info) is rendered into an
-    `exc_info` string field rather than dropped.
+    emits at minimum time/ts/level/module/message. `time` is UTC ISO-8601 with a `Z`
+    suffix; `ts` is the same instant as a unix epoch int, both sortable across
+    machines/containers. any field passed via logging `extra={...}` lands as a
+    top-level JSON field. a traceback (exc_info) is rendered into an `exc_info`
+    string field rather than dropped.
     """
 
     def format(self, record: logging.LogRecord) -> str:
         when = datetime.datetime.fromtimestamp(record.created, datetime.timezone.utc)
         payload = {
             "time": when.strftime("%Y-%m-%dT%H:%M:%SZ"),
+            "ts": int(record.created),
             "level": record.levelname,
             "module": record.name,
             "message": record.getMessage(),
         }
         for key, value in record.__dict__.items():
-            if key not in _RESERVED and not key.startswith("_"):
+            if key not in _RESERVED and key not in _OUTPUT_KEYS and not key.startswith("_"):
                 payload[key] = value
         if record.exc_info:
-            payload["exc_info"] = self.formatException(record.exc_info)
+            # cache the rendered traceback on the record, like stdlib Formatter does
+            if not record.exc_text:
+                record.exc_text = self.formatException(record.exc_info)
+            payload["exc_info"] = record.exc_text
         elif record.exc_text:
             payload["exc_info"] = record.exc_text
         if record.stack_info:
@@ -49,19 +54,27 @@ class JsonLinesFormatter(logging.Formatter):
         return json.dumps(payload, default=str)
 
 
-def build_formatter(output: str = "text", fmt=None, datefmt=None) -> logging.Formatter:
+def build_formatter(output: str = "text", fmt=None, datefmt=None, warnings: list = None) -> logging.Formatter:
     """build the formatter for the chosen output format
 
     `output="text"` (default) returns the human-readable text formatter, honoring
     the raw `fmt`/`datefmt` format-string overrides. `output="json"` returns the
-    structured `JsonLinesFormatter` (which ignores `fmt`/`datefmt` — it builds
+    structured `JsonLinesFormatter` (which ignores `fmt`/`datefmt` - it builds
     fields, not a format string). an unrecognized `output` falls back to text and
-    warns, never raising — a bad format arg must not take the app down.
+    warns, never raising - a bad format arg must not take the app down.
+
+    `warnings` is the setup-time buffer (see `setup_logging`): passing it appends
+    the unknown-output warning there instead of emitting immediately, so it lands
+    in the configured log like the other setup-time warnings rather than only
+    stderr. `None` (default) preserves the old immediate-emit behavior for direct
+    callers outside `setup_logging`.
     """
     if output == "json":
         return JsonLinesFormatter()
     if output != "text":
-        logging.getLogger(__name__).warning(
-            "log_setup: unknown output %r; falling back to 'text'", output
-        )
+        message = "log_setup: unknown output %r; falling back to 'text'"
+        if warnings is None:
+            logging.getLogger(__name__).warning(message, output)
+        else:
+            warnings.append((message, output))
     return logging.Formatter(fmt or DEFAULT_FORMAT, datefmt or DEFAULT_DATEFMT)

src/log_setup/rotation.py

@@ -1,96 +1,377 @@
 """custom namer/rotator + on-start rotation + retention pruning (stdlib only).
 
-the stdlib rotating handlers roll a file next to the live file; these helpers
-override the namer/rotator so rolled files land in `log_dir` and are gzipped when
-asked, keep the live file at its stable path, and handle the on-start and prune
-paths the handlers don't manage themselves.
+the stdlib rotating handlers roll a file next to the live file; these helpers override
+the namer/rotator so rolled files land in `log_dir` and are gzipped when asked, keep
+the live file at its stable path, and handle the on-start and prune paths the handlers
+don't manage themselves.
+
+FOOTGUN: compression is atomic - `_gzip_file` writes to a `.tmp` sibling, verifies it
+decompresses, and only then `os.replace`s it onto the final `.gz` and removes the plain.
+a crash/OOM/power-loss or failed verify at any point leaves the plain `.log` intact and
+no `.gz`, so a corrupt/partial `.gz` is never produced and retention never has to choose
+between an intact plain and a corrupt archive.
 """
 
 import gzip
 import os
+import re
 import shutil
 import time
 from typing import Callable, Optional, Tuple
 
 
-def make_namer(log_dir: str, compress: bool) -> Callable[[str], str]:
-    """namer: redirect a rolled filename into log_dir, adding .gz when compressing
+def _move(source: str, dest: str) -> None:
+    """rename source to dest, falling back to copy+unlink across filesystems
 
-    the handler hands us the default rolled path (next to the live file); we keep its
-    basename but place it under log_dir, and append .gz so the gzipped name matches.
+    FOOTGUN: os.replace is atomic but raises OSError(EXDEV) across filesystems (the
+    container bind-mount / separate-logs-volume case) - falls back to shutil.move so
+    the roll still lands instead of failing rotation silently. precondition: `dest` is
+    a free, non-directory path.
+    """
+    try:
+        os.replace(source, dest)
+    except OSError:
+        shutil.move(source, dest)
+
+
+def _free_dest(dest: str) -> str:
+    """return `dest`, or a `.N`-suffixed variant if it (or its .gz twin) already exists
+
+    used by the tiered rotator so a second same-stamp roll doesn't clobber the earlier
+    file; checks both the plain and .gz forms of each candidate.
+    """
+    if not os.path.exists(dest) and not os.path.exists(dest + ".gz"):
+        return dest
+    counter = 1
+    while True:
+        candidate = f"{dest}.{counter}"
+        if not os.path.exists(candidate) and not os.path.exists(candidate + ".gz"):
+            return candidate
+        counter += 1
+
+
+def _gzip_file(source: str, dest: str) -> None:
+    """atomically gzip source into dest, then remove source - or leave source untouched
+
+    compression is all-or-nothing: writes to `dest + ".tmp"`, verifies that tmp
+    decompresses cleanly, and only then `os.replace`s it onto `dest` and removes source.
+    a crash/OOM/power-loss or a failed verify at ANY point removes the tmp and raises with
+    source intact and no `.gz` at `dest` - so a corrupt/partial `.gz` is never produced and
+    nothing downstream ever has to choose between an intact plain and a corrupt archive.
+
+    the source mtime is carried onto dest so a file keeps its tier position when it
+    crosses the plain->gz boundary - a fresh write would otherwise make a
+    just-compressed file look newest and reshuffle tiers.
+    """
+    mtime = _safe_mtime(source)
+    tmp_dest = dest + ".tmp"
+    try:
+        with open(source, "rb") as src, gzip.open(tmp_dest, "wb") as dst:
+            shutil.copyfileobj(src, dst)
+        if not _gz_intact(tmp_dest):
+            raise OSError(f"gzip of {source!r} did not verify")
+    except BaseException:
+        try:
+            os.remove(tmp_dest)
+        except OSError:
+            pass
+        raise
+    os.replace(tmp_dest, dest)
+    os.remove(source)
+    try:
+        os.utime(dest, (mtime, mtime))
+    except OSError:
+        pass
+
+
+def _gz_intact(path: str) -> bool:
+    """return True if the gzip file at path decompresses cleanly end to end
+
+    used by `_gzip_file` to verify a freshly-written `.gz` before it replaces the plain
+    source; reads the whole stream, any failure means "not intact".
+    """
+    try:
+        with gzip.open(path, "rb") as handle:
+            while handle.read(1 << 20):
+                pass
+        return True
+    except Exception:
+        return False
+
+
+def make_history_namer(
+    stem: str, log_dir: str, compress: bool = False, plain: bool = False,
+    clock=time.localtime,
+) -> Callable[[str], str]:
+    """namer minting historic rolled files `<stem>.<Y-m-d_H-M-S>.log[.gz]` in log_dir
+
+    used by size and daily (and their tiered variants). `stem` is the HISTORY stem (the
+    project namespace), independent of the live file's name. FOOTGUN: prune/retier's
+    `stem` must match this one or nothing matches and retention silently never fires.
+
+    ignores the stdlib handler's own rolled name (`.N` for size, `.log.<date>` for
+    daily) in favor of a uniform timestamped name so all modes converge on one shape
+    retier can rank/tier. `plain=True` (tiered mode) always lands `.log`, letting retier
+    decide compression; same-second collisions disambiguate with a counter, checking
+    both .log and .log.gz forms.
     """
     def namer(default_name: str) -> str:
-        base = os.path.basename(default_name)
-        target = os.path.join(log_dir, base)
-        return target + ".gz" if compress else target
+        stamp = time.strftime("%Y-%m-%d_%H-%M-%S", clock())
+        base = os.path.join(log_dir, f"{stem}.{stamp}")
+        candidate = base
+        counter = 1
+        while os.path.exists(candidate + ".log") or os.path.exists(candidate + ".log.gz"):
+            candidate = f"{base}.{counter}"
+            counter += 1
+        suffix = ".log.gz" if (compress and not plain) else ".log"
+        return candidate + suffix
     return namer
 
 
 def make_rotator(
     compress: bool, log_dir: Optional[str] = None,
     prune_stem: Optional[str] = None, backup_count: int = 0,
+    keep_uncompressed: Optional[int] = None, keep_compressed: Optional[int] = None,
+    rotate_mode: Optional[str] = None, live_stem: Optional[str] = None,
 ) -> Callable[[str, str], None]:
     """rotator: move (or gzip) the source live file to the destination rolled path
 
-    prunes `log_dir` to `backup_count` newest rolled files after each roll when
-    `log_dir`/`prune_stem` are given. the stdlib handler's own retention
-    (`getFilesToDelete`) only scans the live file's directory, so it never sees the
-    rolled files we redirect into `log_dir` — pruning here is what actually bounds
-    retention for the daily and size rolling modes.
+    legacy mode (default): gzip on roll when `compress`, then prune `log_dir` to
+    `backup_count` newest rolled files - the stdlib handler's own retention only scans
+    the live file's directory, so it never sees files redirected into `log_dir`; pruning
+    here is what bounds retention for daily/size. FOOTGUN: for `rotate_mode="size"`,
+    `backup_count <= 0` means "keep no rolled history", but `prune()` itself no-ops at
+    `<= 0` (its own sentinel for "don't touch history") - so a zero-retention roll is
+    deleted by the rotator directly right after landing, rather than relying on prune to
+    do it. `rotate_mode` gates this delete-on-land branch to `"size"` only - `"daily"`
+    (and any other non-size mode) with `backup_count <= 0` still rolls without pruning,
+    matching the documented contract that only `size` always bounds the live file.
+
+    tiered mode (when `keep_uncompressed`/`keep_compressed` are given): land the rolled
+    file PLAIN and re-tier `log_dir` - newest `keep_uncompressed` stay uncompressed, next
+    `keep_compressed` gzipped, rest deleted. `compress`/`backup_count` are ignored.
     """
+    tiered = keep_uncompressed is not None or keep_compressed is not None
+
     def rotator(source: str, dest: str) -> None:
         if not os.path.exists(source):
             return
+        if tiered:
+            # strip the namer's .gz suffix so the roll lands plain and retier decides
+            # its tier; disambiguate a same-interval collision via _free_dest
+            plain_dest = _free_dest(dest[:-3] if dest.endswith(".gz") else dest)
+            _move(source, plain_dest)
+            if log_dir is not None and prune_stem is not None:
+                retier(log_dir, prune_stem, keep_uncompressed or 0, keep_compressed or 0, live_stem)
+            return
         if compress:
-            with open(source, "rb") as src, gzip.open(dest, "wb") as dst:
-                shutil.copyfileobj(src, dst)
-            os.remove(source)
+            _gzip_file(source, dest)
         else:
-            os.replace(source, dest)
-        if log_dir is not None and prune_stem is not None:
-            prune(log_dir, prune_stem, backup_count)
+            _move(source, dest)
+        if backup_count <= 0:
+            if rotate_mode == "size":
+                try:
+                    os.remove(dest)
+                except OSError:
+                    pass
+        elif log_dir is not None and prune_stem is not None:
+            prune(log_dir, prune_stem, backup_count, live_stem)
     return rotator
 
 
-def rotate_on_start(live_path: str, log_dir: str, compress: bool, clock=time.localtime) -> None:
+def rotate_on_start(
+    live_path: str, log_dir: str, compress: bool, clock=time.localtime,
+    keep_uncompressed: Optional[int] = None, keep_compressed: Optional[int] = None,
+    history_stem: Optional[str] = None,
+) -> None:
     """move an existing live file into log_dir with a timestamp, gzipped if asked
 
+    named off `history_stem` (the project namespace) when given, so historic files
+    carry the project name independent of the live file's stem; falls back to the live
+    file's own stem when history_stem is None/empty.
+
     no-op if the live file doesn't exist. used by rotate="on_start" before the fresh
-    handler opens a new live file. the timestamp form is run.<%Y-%m-%d_%H-%M-%S>.log.
+    handler opens a new live file. timestamp form is run.<%Y-%m-%d_%H-%M-%S>.log.
+
+    tiered mode (`keep_uncompressed`/`keep_compressed` given): the rolled file always
+    lands PLAIN (so it can occupy the newest uncompressed tier) and `retier` decides
+    compression/deletion across the whole stem - `compress` is ignored here.
     """
     if not os.path.exists(live_path):
         return
-    stem = os.path.splitext(os.path.basename(live_path))[0]
+    tiered = keep_uncompressed is not None or keep_compressed is not None
+    live_stem = os.path.splitext(os.path.basename(live_path))[0]
+    stem = os.path.basename(history_stem) if history_stem else live_stem
     stamp = time.strftime("%Y-%m-%d_%H-%M-%S", clock())
-    dest = os.path.join(log_dir, f"{stem}.{stamp}.log")
-    if compress:
-        dest += ".gz"
-        with open(live_path, "rb") as src, gzip.open(dest, "wb") as dst:
-            shutil.copyfileobj(src, dst)
-        os.remove(live_path)
+    suffix = ".log.gz" if (compress and not tiered) else ".log"
+
+    # a 1-second stamp collision (rapid crash-restart) is disambiguated with a
+    # counter; check BOTH .log and .log.gz since a tiered same-stamp roll may
+    # already be compressed
+    def _taken(path: str) -> bool:
+        base = path[:-3] if path.endswith(".gz") else path
+        return os.path.exists(base) or os.path.exists(base + ".gz")
+
+    dest = os.path.join(log_dir, f"{stem}.{stamp}{suffix}")
+    counter = 1
+    while _taken(dest):
+        dest = os.path.join(log_dir, f"{stem}.{stamp}.{counter}{suffix}")
+        counter += 1
+    if compress and not tiered:
+        _gzip_file(live_path, dest)
     else:
-        os.replace(live_path, dest)
+        _move(live_path, dest)
+    if tiered:
+        retier(log_dir, stem, keep_uncompressed or 0, keep_compressed or 0, live_stem)
 
 
-def prune(log_dir: str, stem: str, backup_count: int) -> None:
+def retier(log_dir: str, stem: str, keep_uncompressed: int, keep_compressed: int,
+           live_stem: Optional[str] = None) -> None:
+    """re-tier rolled files for stem: newest plain, next gzipped, rest deleted
+
+    newest-first by mtime: the first `keep_uncompressed` stay uncompressed, the next
+    `keep_compressed` are gzipped in place, everything beyond
+    keep_uncompressed+keep_compressed is deleted. the live <stem>.log is never touched.
+    fail-soft per file (skip on OSError) so retention never crashes setup.
+
+    FOOTGUN: `stem` is reduced to its basename to match how rolled files land in
+    log_dir (namer/rotate_on_start basename them) - a `name` containing a directory
+    (e.g. "sub/run") must be matched by "run." here or nothing matches and retention
+    silently never fires (unbounded pileup).
+
+    ordering is by mtime, then by the roll counter parsed from the name, so a
+    same-second burst (tied mtimes, counter-disambiguated stamps like run.<t>.log /
+    run.<t>.1.log) still tiers newest-first rather than falling back to listdir order.
+
+    candidates are also checked against this lib's own rolled-file shape
+    (`<stem>.<stamp>[.N].log[.gz]`, see `_is_rolled_name`) - a foreign file that only
+    shares the `<stem>.` prefix (e.g. a project's own `<stem>.audit.log` dropped into
+    the same log_dir) is left alone rather than tiered/gzipped/deleted.
+    """
+    stem = os.path.basename(stem)
+    live_stem = os.path.basename(live_stem) if live_stem else None
+    try:
+        names = [
+            name for name in os.listdir(log_dir)
+            if name != f"{stem}.log" and _is_rolled_name(stem, name, live_stem)
+        ]
+    except OSError:
+        return
+    entries = [os.path.join(log_dir, name) for name in names]
+    # dedupe plain/gz twins FIRST: a crash between _gzip_file's os.replace and its
+    # os.remove(source) can leave <x>.log beside <x>.log.gz. the .gz is guaranteed intact
+    # (_gzip_file verifies before it replaces, and never produces a partial .gz), so the
+    # plain twin is redundant and dropped unconditionally.
+    present = set(entries)
+    kept = []
+    for p in entries:
+        if not p.endswith(".gz") and (p + ".gz") in present:
+            try:
+                os.remove(p)
+                continue
+            except OSError:
+                pass  # couldn't remove - keep it in the accounting
+        kept.append(p)
+    files = [(p, _safe_mtime(p), _roll_counter(p)) for p in kept if os.path.isfile(p)]
+    # newest-first: higher mtime first, ties broken by higher (later) roll counter
+    files.sort(key=lambda t: (t[1], t[2]), reverse=True)
+
+    keep = keep_uncompressed + keep_compressed
+    for index, (path, _, _) in enumerate(files):
+        if index >= keep:
+            try:
+                os.remove(path)
+            except OSError:
+                pass
+        elif index >= keep_uncompressed and not path.endswith(".gz"):
+            dest = path + ".gz"
+            if os.path.exists(dest):
+                # a verified .gz twin already exists (a prior crashed roll) - drop the
+                # redundant plain rather than re-gzip over it
+                try:
+                    os.remove(path)
+                except OSError:
+                    pass
+                continue
+            try:
+                _gzip_file(path, dest)
+            except OSError:
+                pass
+
+
+def _rolled_name_pattern(stem: str, live_stem: Optional[str] = None) -> "re.Pattern":
+    """compiled regex matching this lib's own rolled-file shapes for stem
+
+    all forms are date-bearing so a foreign same-stem file (e.g. `proj.audit.log`) is
+    never mistaken for a roll and pruned/retiered/deleted:
+    - current uniform namer: `<stem>.<Y-m-d_H-M-S>[.N].log[.gz]`
+      (make_history_namer/rotate_on_start, see attach_rolling)
+    - legacy pre-v0.5.0 daily: `<stem>.log.<Y-m-d>[.gz]` (the stdlib TimedRotatingFileHandler
+      shape) - matched so an upgraded deployment's existing history is still pruned/retired
+      instead of piling up forever. legacy rolls were named off the LIVE file's name, not the
+      history stem, so when `live_stem` is given (and differs) its legacy shape is matched too
+    """
+    stems = [stem] if live_stem is None or live_stem == stem else [stem, live_stem]
+    forms = []
+    for s in stems:
+        esc = re.escape(s)
+        forms.append(rf"{esc}\.\d{{4}}-\d{{2}}-\d{{2}}_\d{{2}}-\d{{2}}-\d{{2}}(?:\.\d+)?\.log(?:\.gz)?")
+        forms.append(rf"{esc}\.log\.\d{{4}}-\d{{2}}-\d{{2}}(?:\.gz)?")
+    return re.compile(rf"^(?:{'|'.join(forms)})$")
+
+
+def _is_rolled_name(stem: str, name: str, live_stem: Optional[str] = None) -> bool:
+    """return whether name has this lib's own rolled-log shape for stem (or the live stem)"""
+    return _rolled_name_pattern(stem, live_stem).match(name) is not None
+
+
+def _roll_counter(path: str) -> int:
+    """parse the same-second disambiguation counter out of a rolled filename
+
+    only the on_start / size-namer shape carries a counter: `<stem>.<stamp>[.<counter>].log`
+    (optionally `.gz`); a higher counter is the later roll, first roll of a second is 0.
+    daily's dated names (`<stem>.log.<Y-m-d>`) don't end in `.log`, so they never need
+    the tie-break - return 0 rather than misparsing the trailing date as a counter.
+    """
+    base = path[:-3] if path.endswith(".gz") else path
+    if not base.endswith(".log"):
+        return 0
+    base = base[:-4]
+    tail = base.rsplit(".", 1)[-1]
+    return int(tail) if tail.isdigit() else 0
+
+
+def prune(log_dir: str, stem: str, backup_count: int, live_stem: Optional[str] = None) -> None:
     """keep only the newest `backup_count` rolled files for a given stem in log_dir
 
-    matches files beginning with `<stem>.` (e.g. run.*), sorted by mtime, deleting the
+    matches files beginning with `<stem>.` (e.g. run.*), sorted newest-first by mtime
+    then roll counter (mirrors retier's ordering, see _roll_counter), deleting the
     oldest beyond the count. used for on_start, which the handlers don't auto-prune.
+
+    FOOTGUN: `stem` is reduced to its basename so a `name` containing a directory (e.g.
+    "sub/run") still matches the basenamed rolled files in log_dir - else nothing
+    matches and old files pile up forever.
+
+    candidates are also checked against this lib's own rolled-file shape
+    (`<stem>.<stamp>[.N].log[.gz]`, see `_is_rolled_name`) - a foreign file that only
+    shares the `<stem>.` prefix (e.g. a project's own `<stem>.audit.log` dropped into
+    the same log_dir) is left alone rather than counted and deleted as a roll.
     """
     if backup_count <= 0:
         return
+    stem = os.path.basename(stem)
+    live_stem = os.path.basename(live_stem) if live_stem else None
     try:
         entries = [
             os.path.join(log_dir, name)
             for name in os.listdir(log_dir)
-            if name.startswith(f"{stem}.") and name != f"{stem}.log"
+            if name != f"{stem}.log" and _is_rolled_name(stem, name, live_stem)
         ]
     except OSError:
         return
-    files = [(p, _safe_mtime(p)) for p in entries if os.path.isfile(p)]
-    files.sort(key=lambda pair: pair[1], reverse=True)
-    for path, _ in files[backup_count:]:
+    files = [(p, _safe_mtime(p), _roll_counter(p)) for p in entries if os.path.isfile(p)]
+    files.sort(key=lambda t: (t[1], t[2]), reverse=True)
+    for path, _, _ in files[backup_count:]:
         try:
             os.remove(path)
         except OSError:
@@ -108,14 +389,35 @@ def _safe_mtime(path: str) -> float:
 def attach_rolling(
     handler, log_dir: str, compress: bool,
     prune_stem: Optional[str] = None, backup_count: int = 0,
+    keep_uncompressed: Optional[int] = None, keep_compressed: Optional[int] = None,
+    tiered: bool = False, rotate_mode: Optional[str] = None,
 ) -> Tuple[Callable, Callable]:
     """wire the custom namer + rotator onto a rotating handler; return them
 
+    rolled files are named off `prune_stem` (the HISTORY stem - the project namespace),
+    independent of the live file name, via make_history_namer: `<stem>.<stamp>.log[.gz]`
+    uniform across size and daily - replacing the stdlib handler's own rolled-name
+    scheme (`.N` for size, `.log.<date>` for daily), which can't inject a project stem
+    and (for size) can't be managed once files are redirected into log_dir.
+
     pass `prune_stem`/`backup_count` so the rotator prunes `log_dir` after each roll
-    (the handler's own retention can't see the redirected rolled files).
+    (the handler's own retention can't see the redirected files). pass
+    `keep_uncompressed`/`keep_compressed` for tiered retention instead (see
+    make_rotator); `tiered=True` lands rolls plain (retier compresses). `rotate_mode`
+    ("size"/"daily") is forwarded to `make_rotator` so the zero-retention delete-on-land
+    branch only ever fires for `"size"`.
     """
-    namer = make_namer(log_dir, compress)
-    rotator = make_rotator(compress, log_dir, prune_stem, backup_count)
+    namer = make_history_namer(
+        os.path.basename(prune_stem or ""), log_dir, compress, plain=tiered,
+    )
+    # the live file name (stem of the handler's baseFilename, minus one .log) - so retention
+    # also recognizes pre-v0.5.0 legacy rolls, which were named off the LIVE name not the stem
+    live_base = os.path.basename(getattr(handler, "baseFilename", "") or "")
+    live_stem = live_base[:-4] if live_base.endswith(".log") else live_base
+    rotator = make_rotator(
+        compress, log_dir, prune_stem, backup_count, keep_uncompressed, keep_compressed,
+        rotate_mode=rotate_mode, live_stem=live_stem or None,
+    )
     handler.namer = namer
     handler.rotator = rotator
     return namer, rotator

src/log_setup/setup.py

@@ -1,19 +1,23 @@
-"""app-entry-point logging setup (sync, stdlib only).
+"""app-entry-point logging setup (sync, stdlib only). see README.
 
-`setup_logging` configures the root logger once for the whole process: a live
-run.log at a stable path, rotation (daily/size/on_start/none) into a logs/ dir, gzip
-of rolled files, retention, console output, and a consistent format. it is called by
-the APPLICATION, not by reusable libraries (those stay emit-only). it is idempotent
-(no duplicate handlers on repeat calls), never crashes the app over logging, and can
-route through a background queue so an async event loop doesn't block on file I/O.
+`setup_logging` configures the root logger once for the whole process. called by the
+APPLICATION, not by reusable libraries (those stay emit-only). idempotent, never
+crashes the app over logging, and can route through a background queue so an async
+event loop doesn't block on file I/O.
+
+`rotate="size"` always bounds the live file: the roll fires at `max_bytes` regardless
+of `backup_count`, including `backup_count=0` (means "keep zero rolled files", not
+"never roll" - each roll is deleted right after landing).
 """
 
 import atexit
 import logging
 import logging.handlers
 import os
-import queue
-from typing import Optional, Union
+import queue as _queue
+import sys
+import traceback
+from typing import Dict, Optional, Union
 
 from .formats import build_formatter
 from .rotation import attach_rolling, prune, rotate_on_start
@@ -22,25 +26,86 @@ log = logging.getLogger(__name__)
 
 _MARKER = "_log_setup_owned"
 _listener = None
+_atexit_registered = False
+
+
+class _PreservingQueueHandler(logging.handlers.QueueHandler):
+    """QueueHandler that hands the record to the listener untouched
+
+    stdlib's QueueHandler.prepare() calls self.format(record) with the queue
+    handler's OWN (default) formatter, then nulls exc_info/exc_text/stack_info -
+    losing structured fields (e.g. JsonLinesFormatter's exc_info key) before the
+    listener's real formatter ever sees the record. that behavior exists to keep a
+    record picklable across a multiprocessing.Queue; this lib only ever uses an
+    in-process queue.Queue, so there is nothing to pickle and nothing to strip -
+    the listener's handler formats the untouched record exactly once.
+    """
+
+    def prepare(self, record: logging.LogRecord) -> logging.LogRecord:
+        return record
+
+
+def _exc_text() -> str:
+    """render sys.exc_info() as text, for capturing a traceback into a buffered warning"""
+    return "".join(traceback.format_exception(*sys.exc_info())).strip()
+
+
+def _flush_warnings(warnings: list) -> None:
+    """emit buffered setup-time warnings now that handlers are attached"""
+    for message, *args in warnings:
+        log.warning(message, *args)
 
 
 def _level_value(level: Union[int, str]) -> int:
     """coerce a level name or int to a logging level int (defaults to INFO)"""
+    if isinstance(level, bool):
+        # bool is an int subclass (True==1, below DEBUG) but never a real level
+        return logging.INFO
     if isinstance(level, int):
         return level
     if not isinstance(level, str):
         return logging.INFO
     resolved = logging.getLevelName(level.upper())
-    # getLevelName returns the string "Level XXX" for an unknown name, which
-    # setLevel then rejects — never crash the app over a bad level, fall back to INFO
+    # getLevelName returns "Level XXX" for an unknown name; fall back to INFO
     return resolved if isinstance(resolved, int) else logging.INFO
 
 
-def _clear_owned(root: logging.Logger) -> None:
+def _strict_level_value(level: Union[int, str]) -> Optional[int]:
+    """like _level_value but reports invalid as None so the caller can skip + warn"""
+    if isinstance(level, bool):
+        return None
+    if isinstance(level, int):
+        return level
+    if not isinstance(level, str):
+        return None
+    resolved = logging.getLevelName(level.upper())
+    return resolved if isinstance(resolved, int) else None
+
+
+def _apply_module_levels(module_levels: Optional[Dict[str, Union[int, str]]], warnings: list) -> None:
+    """set per-logger level overrides by exact logger name, never crashing"""
+    if not module_levels:
+        return
+    for mod_name, raw_level in module_levels.items():
+        value = _strict_level_value(raw_level)
+        if value is None:
+            warnings.append(("log_setup: invalid level %r for logger %r; skipping", raw_level, mod_name))
+            continue
+        logging.getLogger(mod_name).setLevel(value)
+
+
+def _clear_owned(root: logging.Logger, warnings: list) -> None:
     """remove only the handlers this lib previously added; leave app handlers alone"""
     global _listener
     if _listener is not None:
         _listener.stop()
+        # listener owns the real file/console handlers; stopping it doesn't close
+        # them, so close here rather than rely on GC finalizers across a re-setup
+        for wrapped in getattr(_listener, "handlers", ()):
+            try:
+                wrapped.close()
+            except Exception:
+                warnings.append(("log_setup: failed to close queued handler %r: %s", wrapped, _exc_text()))
         _listener = None
     for handler in list(root.handlers):
         if getattr(handler, _MARKER, False):
@@ -48,7 +113,9 @@ def _clear_owned(root: logging.Logger) -> None:
             try:
                 handler.close()
             except Exception:
-                pass
+                warnings.append(
+                    ("log_setup: failed to close handler %r during re-setup: %s", handler, _exc_text())
+                )
 
 
 def _tag(handler: logging.Handler) -> logging.Handler:
@@ -57,25 +124,77 @@ def _tag(handler: logging.Handler) -> logging.Handler:
     return handler
 
 
+def _normalize_name(name: str) -> str:
+    """strip one trailing '.log' (case-insensitive) so the stem is extension-free
+
+    "latest" and "latest.log" both yield stem "latest" (never latest.log.log); only
+    one level is stripped, so "app.log.log" -> "app.log".
+    """
+    if name.lower().endswith(".log"):
+        return name[:-4]
+    return name
+
+
+def _history_stem() -> str:
+    """the project namespace for historic files: the cwd basename, "" for a degenerate cwd"""
+    try:
+        return os.path.basename(os.getcwd().rstrip(os.sep))
+    except OSError:
+        return ""
+
+
 def _file_handler(
-    name: str, live_path: str, log_dir: str, rotate: Optional[str],
+    history_stem: str, live_path: str, log_dir: str, rotate: Optional[str],
     backup_count: int, max_bytes: int, compress: bool,
+    keep_uncompressed: Optional[int], keep_compressed: Optional[int], warnings: list,
 ) -> logging.Handler:
-    """build the configured file handler with custom rolling into log_dir"""
+    """build the configured file handler with custom rolling into log_dir
+
+    `history_stem` is the PROJECT stem that rolled/historic files are named off,
+    decoupled from the live file's own name (`live_path`).
+    """
+    tiered = keep_uncompressed is not None or keep_compressed is not None
     if rotate == "size":
+        # stdlib doRollover no-ops at backupCount==0 - force nonzero so the roll
+        # always fires; the REAL backup_count still flows to attach_rolling, whose
+        # make_rotator treats <=0 as "keep no rolled history" and deletes each roll
+        size_backup = max(backup_count, 1)
         handler = logging.handlers.RotatingFileHandler(
-            live_path, maxBytes=max_bytes, backupCount=backup_count, encoding="utf-8",
+            live_path, maxBytes=max_bytes, backupCount=size_backup, encoding="utf-8",
+        )
+        attach_rolling(
+            handler, log_dir, compress, prune_stem=history_stem, backup_count=backup_count,
+            keep_uncompressed=keep_uncompressed, keep_compressed=keep_compressed,
+            tiered=tiered, rotate_mode=rotate,
         )
-        attach_rolling(handler, log_dir, compress, prune_stem=name, backup_count=backup_count)
     elif rotate == "daily":
         handler = logging.handlers.TimedRotatingFileHandler(
             live_path, when="midnight", backupCount=backup_count, encoding="utf-8",
         )
-        attach_rolling(handler, log_dir, compress, prune_stem=name, backup_count=backup_count)
+        attach_rolling(
+            handler, log_dir, compress, prune_stem=history_stem, backup_count=backup_count,
+            keep_uncompressed=keep_uncompressed, keep_compressed=keep_compressed,
+            tiered=tiered, rotate_mode=rotate,
+        )
     else:
         if rotate == "on_start":
-            rotate_on_start(live_path, log_dir, compress)
-            prune(log_dir, name, backup_count)
+            if tiered:
+                rotate_on_start(
+                    live_path, log_dir, compress, history_stem=history_stem,
+                    keep_uncompressed=keep_uncompressed, keep_compressed=keep_compressed,
+                )
+            else:
+                rotate_on_start(live_path, log_dir, compress, history_stem=history_stem)
+                live_base = os.path.basename(live_path)
+                live_stem = live_base[:-4] if live_base.endswith(".log") else live_base
+                prune(log_dir, history_stem, backup_count, live_stem or None)
+        elif rotate is not None:
+            # a typo'd value would otherwise silently fall through to a non-rotating
+            # FileHandler and grow forever - warn instead of degrade silently
+            warnings.append((
+                "log_setup: unknown rotate %r; expected 'daily'/'size'/'on_start'/None - "
+                "no rotation applied (single growing file)", rotate,
+            ))
         handler = logging.FileHandler(live_path, encoding="utf-8")
     return handler
 
@@ -84,8 +203,12 @@ def setup_logging(
     name: str = "run",
     log_dir: str = "logs",
     level: Union[int, str] = "INFO",
+    module_levels: Optional[Dict[str, Union[int, str]]] = None,
     rotate: Optional[str] = "daily",
+    history_name: Optional[str] = None,
     backup_count: int = 14,
+    keep_uncompressed: Optional[int] = None,
+    keep_compressed: Optional[int] = None,
     max_bytes: int = 10_000_000,
     compress: bool = True,
     console: bool = False,
@@ -96,27 +219,62 @@ def setup_logging(
 ) -> logging.Logger:
     """configure the root logger for the whole process and return it
 
-    `name` -> <name>.log live file at cwd; rolled/compressed copies go to `log_dir`.
-    `rotate` is "daily" (default), "size", "on_start", or None. `console=True` adds a
-    stdout handler (off by default — the file is the output). `queue=True` routes records
-    through a background QueueListener so file I/O never blocks the caller (the listener
-    is stopped at exit). `output` is "text" (default, human `time | module | level |
-    message`, local time) or "json" (structured one-JSON-object-per-line for the
-    Grafana/Loki path, UTC timestamps, `extra=` fields surfaced as top-level keys); both
-    file and console use the chosen format and the live-file name is the same regardless.
-    the raw `fmt`/`datefmt` overrides apply to text output only. idempotent: a repeat call
-    clears only the handlers this function added. never raises over logging — an
-    unwritable `log_dir` falls back to console-only with a warning even when `console` is
-    off, so output is never silently lost; an unknown `output` falls back to text.
+    `name` -> <name>.log live file at cwd; rolled/compressed copies go to `log_dir`. a
+    trailing ".log" in `name` is stripped so "latest" and "latest.log" both produce
+    latest.log (never latest.log.log).
+
+    `history_name` names the rolled/historic files (`<history_name>.<timestamp>.log[.gz]`),
+    independent of the live file: defaults to the PROJECT namespace = the cwd basename
+    (run from bestbuy/ -> historic files bestbuy.<stamp>...). the live file always keeps
+    `name`; only historic files carry the project name.
+
+    `keep_uncompressed`/`keep_compressed` (default None) enable TIERED retention: when
+    either is given, rolled files are kept as the newest `keep_uncompressed` uncompressed
+    + the next `keep_compressed` gzipped, rest deleted (total retained = sum). applies to
+    "on_start", "daily", and "size". `backup_count` and gzip-on-roll `compress` are
+    IGNORED in tiered mode. pass NEITHER knob and rotation behaves exactly as before.
+
+    `level` is the root default every logger inherits. `module_levels` is an optional
+    map of exact logger name -> level applied after the root is set, the ergonomic way
+    to quiet noisy dependencies (e.g. {"motor": "WARNING"}) from one call instead of
+    scattering `getLogger(...).setLevel(...)` calls. names match EXACTLY (no discovery:
+    a typo'd name silently configures an unused logger), but hierarchy applies, so
+    naming a parent ("aiohttp") quiets its whole subtree. str or int per entry; a bad
+    value is skipped with a warning and never aborts the others or the setup.
+
+    `rotate` is "daily" (default), "size", "on_start", or None. for `rotate="size"`, the
+    live file always rolls at `max_bytes` regardless of `backup_count`: `backup_count=0`
+    means "keep zero rolled files" (each roll lands then is deleted immediately), NOT
+    "disable rotation". `backup_count>=1` keeps that many rolled files as before.
+
+    `console=True` adds a stdout handler (off by default - the file is the output).
+    `queue=True` routes records through a background QueueListener so file I/O never
+    blocks the caller (stopped at exit). `output` is "text" (default, human `time |
+    module | level | message`, local time) or "json" (structured JSON Lines, UTC
+    timestamps + a unix-epoch `ts`, `extra=` fields surfaced as top-level keys); file
+    and console use the same format, live-file name unaffected. `fmt`/`datefmt` apply
+    to text output only.
+
+    idempotent: a repeat call clears only the handlers this function added. never
+    raises over logging - an unwritable `log_dir` falls back to console-only with a
+    warning even when `console` is off; an unknown `output` falls back to text.
     """
-    global _listener
+    global _listener, _atexit_registered
+
+    warnings: list = []
 
     root = logging.getLogger()
     root.setLevel(_level_value(level))
-    _clear_owned(root)
+    _apply_module_levels(module_levels, warnings)
+    _clear_owned(root, warnings)
 
-    formatter = build_formatter(output, fmt, datefmt)
-    live_path = f"{name}.log"
+    formatter = build_formatter(output, fmt, datefmt, warnings)
+    stem = _normalize_name(name)
+    live_path = f"{stem}.log"
+    # history_name if given, else the cwd basename; falls back to the live stem for a
+    # degenerate cwd so naming/retention never break
+    history_source = history_name if history_name is not None else _history_stem()
+    history_stem = os.path.basename(_normalize_name(history_source)) or stem
 
     handlers = []
 
@@ -128,37 +286,46 @@ def setup_logging(
 
     if file_ok:
         try:
-            fh = _file_handler(name, live_path, log_dir, rotate, backup_count, max_bytes, compress)
+            fh = _file_handler(
+                history_stem, live_path, log_dir, rotate, backup_count, max_bytes, compress,
+                keep_uncompressed, keep_compressed, warnings,
+            )
             fh.setFormatter(formatter)
             handlers.append(fh)
         except OSError:
             file_ok = False
 
     if console or not file_ok:
-        sh = logging.StreamHandler()
+        sh = logging.StreamHandler(sys.stdout)
         sh.setFormatter(formatter)
         handlers.append(sh)
 
     if queue:
-        record_queue: "queue.Queue" = _make_queue()
-        qh = _tag(logging.handlers.QueueHandler(record_queue))
+        record_queue: "_queue.Queue" = _make_queue()
+        qh = _tag(_PreservingQueueHandler(record_queue))
         root.addHandler(qh)
         _listener = logging.handlers.QueueListener(record_queue, *handlers, respect_handler_level=True)
         _listener.start()
-        atexit.register(_stop_listener)
+        if not _atexit_registered:
+            # register once - atexit doesn't dedupe, repeated re-setups would stack
+            atexit.register(_stop_listener)
+            _atexit_registered = True
     else:
         for handler in handlers:
             root.addHandler(_tag(handler))
 
     if not file_ok:
-        log.warning("log_setup: log_dir %r not writable; logging to console only", log_dir)
+        warnings.append(("log_setup: log_dir %r not writable; logging to console only", log_dir))
+
+    # flush now that handlers are attached, so warnings land in the configured log
+    _flush_warnings(warnings)
 
     return root
 
 
-def _make_queue() -> "queue.Queue":
+def _make_queue() -> "_queue.Queue":
     """unbounded in-memory queue for the QueueHandler -> QueueListener path"""
-    return queue.Queue(-1)
+    return _queue.Queue(-1)
 
 
 def _stop_listener() -> None: