fix: LOW/nit clearance — retier twin dedupe + same-second counter ordering, warn on unknown rotate

- logsetup-2: retier drops a plain file whose .gz twin exists (crash-between-write-and-remove
  residue) up front, so the phantom twin never occupies a retention slot / evicts a real roll.
- logsetup-5: retier breaks mtime ties by the roll counter parsed from the name, so a
  same-second on_start burst tiers newest-first instead of arbitrary listdir order.
- logsetup-3: tiered daily rotator disambiguates an existing dated dest with a counter (no
  clobber on a second same-interval roll).
- logsetup-6: unknown rotate value warns (matching the unknown-output convention) instead of
  silently degrading to a non-rotating file.
- README: size-mode backup naming clarified (numbered flat vs timestamped tiered).
verified stdlib-only; v0.4.0/v0.4.1/size suites still pass. bump v0.4.2 -> v0.4.3

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.3.0
+log_setup @ git+ssh://git@git.rethinkstudios.io/rethink-public/log_setup.git@v0.4.3
 ```
 
 No dependencies — stdlib only.
 
+Drop the `@v0.4.3` suffix from the line above to install the latest unpinned.
+
 ## Quick start
 
 ```python
@@ -43,15 +45,55 @@ emits; the records land in the configured root.
 - **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`.
+  - `"size"` — rolls at `max_bytes`; numbered backups (`run.log.1`, `.2`, …) in `log_dir`
+    for the default flat retention, or timestamped names when tiered retention is on (below).
   - `"on_start"` — on startup, moves an existing `run.log` into `log_dir`
     (`run.<timestamp>.log[.gz]`) 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.
+- **Retention** = `backup_count` (default 14) for every mode — unless tiered retention is
+  enabled (below).
 - **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`).
+
+## 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
+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 in `log_dir` (newest → oldest):
+
+```
+latest.log                       <- live "latest" (stable, tail -f)
+latest.<t1>.log   latest.<t2>.log   latest.<t3>.log       <- 3 newest: plain
+latest.<t4>.log.gz ... latest.<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
@@ -92,7 +134,9 @@ setup_logging(
     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)
+    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)
@@ -177,4 +221,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.3.0"
+version = "0.4.3"
 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

@@ -19,4 +19,4 @@ from .setup import setup_logging
 
 __all__ = ["setup_logging"]
 
-__version__ = "0.3.0"
+__version__ = "0.4.3"

src/log_setup/formats.py

@@ -15,6 +15,11 @@ 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` are NOT
+# LogRecord attrs, so a caller's extra={"time":...}/{"level":...} would otherwise
+# overwrite the UTC timestamp / levelname. guard them explicitly
+_OUTPUT_KEYS = frozenset({"time", "level", "module", "message"})
 
 
 class JsonLinesFormatter(logging.Formatter):
@@ -38,10 +43,14 @@ class JsonLinesFormatter(logging.Formatter):
             "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 (as stdlib Formatter does) so a
+            # second handler/format() of the same record doesn't re-render it
+            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:

src/log_setup/rotation.py

@@ -13,6 +13,60 @@ import time
 from typing import Callable, Optional, Tuple
 
 
+def _move(source: str, dest: str) -> None:
+    """rename source to dest, falling back to copy+unlink across filesystems
+
+    os.replace is atomic but raises OSError(EXDEV) when source and dest are on
+    different filesystems — exactly the container bind-mount / separate-logs-volume
+    case this lib targets. fall back to shutil.move (copy+unlink) so the roll still
+    lands instead of failing every rotation via the handler's silent handleError.
+
+    precondition: `dest` is a free, non-directory path (all call sites generate a unique
+    timestamped/dated dest). os.replace and shutil.move differ on a dest that already
+    exists as a directory, so this helper is not safe for arbitrary dests — only the
+    rotation paths that guarantee a fresh file dest.
+    """
+    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 roll landing on the same dated/stamped name
+    (e.g. two daily rolls in one day) doesn't clobber the earlier file. the suffix goes
+    before nothing here (dest is already the plain path) — 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:
+    """gzip source into dest then remove source (the rolled-file compression idiom)
+
+    the source mtime is carried onto dest so a file keeps its position when it crosses
+    the plain->gz tier boundary — retier ranks by mtime, and a fresh write would
+    otherwise make a just-compressed file look like the newest one and reshuffle tiers.
+    """
+    mtime = _safe_mtime(source)
+    with open(source, "rb") as src, gzip.open(dest, "wb") as dst:
+        shutil.copyfileobj(src, dst)
+    os.remove(source)
+    try:
+        os.utime(dest, (mtime, mtime))
+    except OSError:
+        pass
+
+
 def make_namer(log_dir: str, compress: bool) -> Callable[[str], str]:
     """namer: redirect a rolled filename into log_dir, adding .gz when compressing
 
@@ -26,50 +80,204 @@ def make_namer(log_dir: str, compress: bool) -> Callable[[str], str]:
     return namer
 
 
+def make_size_namer(
+    stem: str, log_dir: str, clock=time.localtime,
+) -> Callable[[str], str]:
+    """namer for tiered SIZE mode: a unique timestamped dest per roll, plain (no .gz)
+
+    stdlib RotatingFileHandler names rolls `<live>.1`, `<live>.2`, ... and shifts them —
+    a scheme that breaks once files are redirected into log_dir (the shift can't find
+    them, so every roll reuses slot 1). tiered retention wants unique per-roll names it
+    can rank + tier like the daily/on_start paths, so ignore the handler's `.N` suffix
+    entirely and mint `<stem>.<Y-m-d_H-M-S>.log`, disambiguating a same-second collision
+    (against both the .log and .log.gz forms) with a counter. always plain — retier
+    decides compression.
+    """
+    def namer(default_name: str) -> str:
+        stamp = time.strftime("%Y-%m-%d_%H-%M-%S", clock())
+        dest = os.path.join(log_dir, f"{stem}.{stamp}.log")
+        counter = 1
+        while os.path.exists(dest) or os.path.exists(dest + ".gz"):
+            dest = os.path.join(log_dir, f"{stem}.{stamp}.{counter}.log")
+            counter += 1
+        return dest
+    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,
 ) -> 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
+    legacy mode (default): gzip on roll when `compress`, then prune `log_dir` to
+    `backup_count` newest rolled files. 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.
+    rolled files we redirect into `log_dir` — pruning here is what bounds retention for
+    the daily and size rolling modes.
+
+    tiered mode (when `keep_uncompressed`/`keep_compressed` are given): land the rolled
+    file PLAIN and re-tier `log_dir` — newest `keep_uncompressed` stay uncompressed, the
+    next `keep_compressed` are gzipped, the rest deleted. `compress`/`backup_count` are
+    ignored in this mode (the tier counts bound retention instead).
     """
+    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:
+            # dest carries the namer's .gz suffix in compress mode; strip it so the
+            # freshly-rolled file lands plain and retier decides its tier. disambiguate a
+            # dest that already exists (a second same-interval daily roll reuses the same
+            # dated name) with a counter, checking both .log and .log.gz forms, so the
+            # earlier roll isn't clobbered.
+            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)
+            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)
+            _move(source, dest)
         if log_dir is not None and prune_stem is not None:
             prune(log_dir, prune_stem, backup_count)
     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,
+) -> None:
     """move an existing live file into log_dir with a timestamp, gzipped if asked
 
     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.
+
+    tiered mode (when `keep_uncompressed`/`keep_compressed` are 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 for the
+    just-rolled file.
     """
     if not os.path.exists(live_path):
         return
+    tiered = keep_uncompressed is not None or keep_compressed is not None
     stem = os.path.splitext(os.path.basename(live_path))[0]
     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"
+    # the stamp is 1-second resolution; two starts in the same second would collide
+    # and the second clobber the first. disambiguate with a numeric counter so a rapid
+    # crash-restart loop doesn't lose the earlier rolled file. check BOTH the .log and
+    # .log.gz forms of each candidate: in tiered mode an earlier same-stamp roll may have
+    # already been compressed to .log.gz, and reusing its bare stem would create a second
+    # file for the same logical roll and break the tier counts
+
+    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)
+
+
+def retier(log_dir: str, stem: str, keep_uncompressed: int, keep_compressed: int) -> 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 (a still-plain file in that band is compressed
+    to <name>.gz and the plain source removed), and 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.
+
+    `stem` is reduced to its basename: rolled files land in log_dir under the basename
+    (the namer/rotate_on_start basename them), so 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 correctly rather than falling back to arbitrary listdir order.
+    """
+    stem = os.path.basename(stem)
+    try:
+        names = [
+            name for name in os.listdir(log_dir)
+            if name.startswith(f"{stem}.") and name != f"{stem}.log"
+        ]
+    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 write and its os.remove can
+    # leave <x>.log beside <x>.log.gz). drop the redundant plain copy up front so the
+    # phantom twin never occupies a retention slot and evicts a distinct older roll.
+    present = set(entries)
+    kept = []
+    for p in entries:
+        if not p.endswith(".gz") and (p + ".gz") in present:
+            try:
+                os.remove(p)
+            except OSError:
+                kept.append(p)  # couldn't remove — keep it in the accounting
+            continue
+        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, and within a tied second the higher roll counter
+    # (a later same-second roll) is newer
+    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 crash/power-loss between _gzip_file's write and its os.remove can leave
+                # a plain source beside a fresh .gz. don't keep both (they'd double-count
+                # toward retention and evict a distinct older roll) — drop the redundant
+                # plain twin, keeping the compressed copy.
+                try:
+                    os.remove(path)
+                except OSError:
+                    pass
+                continue
+            try:
+                _gzip_file(path, dest)
+            except OSError:
+                pass
+
+
+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`), where a colliding same-second roll gets `.1`, `.2`, ... and a higher
+    counter is the later (newer) roll. the first roll of a second has no counter (0).
+
+    daily's dated names (`<stem>.log.<Y-m-d>`) do NOT end in `.log` and are second+-granular
+    (distinct mtimes), so they never need the counter tie-break — return 0 for them rather
+    than misparsing the trailing date component 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) -> None:
@@ -77,9 +285,14 @@ def prune(log_dir: str, stem: str, backup_count: int) -> None:
 
     matches files beginning with `<stem>.` (e.g. run.*), sorted by mtime, deleting the
     oldest beyond the count. used for on_start, which the handlers don't auto-prune.
+
+    `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).
     """
     if backup_count <= 0:
         return
+    stem = os.path.basename(stem)
     try:
         entries = [
             os.path.join(log_dir, name)
@@ -108,14 +321,28 @@ 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,
+    size_tiered: bool = False,
 ) -> Tuple[Callable, Callable]:
     """wire the custom namer + rotator onto a rotating handler; return them
 
     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 rolled files). pass
+    `keep_uncompressed`/`keep_compressed` instead to use tiered retention (newest plain,
+    next gzipped, rest deleted) — see make_rotator.
+
+    `size_tiered` uses a timestamped per-roll namer (make_size_namer) instead of the
+    default one, for a tiered RotatingFileHandler (size mode): stdlib's `.1/.2` numbered
+    shift can't manage files redirected into log_dir, so each roll gets a unique dated
+    name that retier ranks/tiers like the daily/on_start paths.
     """
-    namer = make_namer(log_dir, compress)
-    rotator = make_rotator(compress, log_dir, prune_stem, backup_count)
+    if size_tiered:
+        namer = make_size_namer(os.path.basename(prune_stem or ""), log_dir)
+    else:
+        namer = make_namer(log_dir, compress)
+    rotator = make_rotator(
+        compress, log_dir, prune_stem, backup_count, keep_uncompressed, keep_compressed,
+    )
     handler.namer = namer
     handler.rotator = rotator
     return namer, rotator

src/log_setup/setup.py

@@ -12,7 +12,7 @@ import atexit
 import logging
 import logging.handlers
 import os
-import queue
+import queue as _queue
 from typing import Dict, Optional, Union
 
 from .formats import build_formatter
@@ -22,10 +22,15 @@ log = logging.getLogger(__name__)
 
 _MARKER = "_log_setup_owned"
 _listener = None
+_atexit_registered = False
 
 
 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 is never a real level —
+        # reject it consistently with the per-module path rather than set level 1
+        return logging.INFO
     if isinstance(level, int):
         return level
     if not isinstance(level, str):
@@ -76,6 +81,14 @@ def _clear_owned(root: logging.Logger) -> None:
     global _listener
     if _listener is not None:
         _listener.stop()
+        # the listener owns the real file/console handlers (only the QueueHandler is
+        # root-attached + marked); stopping it doesn't close them, so close them here
+        # to avoid relying on GC finalizers across a re-setup
+        for wrapped in getattr(_listener, "handlers", ()):
+            try:
+                wrapped.close()
+            except Exception:
+                log.warning("log_setup: failed to close queued handler %r", wrapped, exc_info=True)
         _listener = None
     for handler in list(root.handlers):
         if getattr(handler, _MARKER, False):
@@ -83,7 +96,9 @@ def _clear_owned(root: logging.Logger) -> None:
             try:
                 handler.close()
             except Exception:
-                pass
+                # a handler failing to close must not abort re-setup, but log it
+                # rather than swallow silently (consistent with the lib's warn pattern)
+                log.warning("log_setup: failed to close handler %r during re-setup", handler, exc_info=True)
 
 
 def _tag(handler: logging.Handler) -> logging.Handler:
@@ -92,25 +107,66 @@ 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
+
+    `name` is allowed to be passed with or without the extension — "latest" and
+    "latest.log" both yield stem "latest" (live file latest.log), never latest.log.log.
+    only one level is stripped: "app.log.log" -> "app.log" so a legit ".log" inside a
+    name survives.
+    """
+    if name.lower().endswith(".log"):
+        return name[:-4]
+    return name
+
+
 def _file_handler(
     name: 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],
 ) -> logging.Handler:
     """build the configured file handler with custom rolling into log_dir"""
+    tiered = keep_uncompressed is not None or keep_compressed is not None
     if rotate == "size":
+        # stdlib doRollover is a no-op when backupCount == 0, and its numbered .1/.2 shift
+        # can't manage files redirected into log_dir. in tiered size mode force a nonzero
+        # backupCount so the roll always fires (retier bounds retention, not backupCount)
+        # and use the timestamped size-namer (size_tiered) instead of the .N shift.
+        size_backup = backup_count if not tiered else 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=name, backup_count=backup_count,
+            keep_uncompressed=keep_uncompressed, keep_compressed=keep_compressed,
+            size_tiered=tiered,
         )
-        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=name, backup_count=backup_count,
+            keep_uncompressed=keep_uncompressed, keep_compressed=keep_compressed,
+        )
     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,
+                    keep_uncompressed=keep_uncompressed, keep_compressed=keep_compressed,
+                )
+            else:
+                rotate_on_start(live_path, log_dir, compress)
+                prune(log_dir, name, backup_count)
+        elif rotate is not None:
+            # a typo'd rotate value (e.g. "hourly") would otherwise silently fall through
+            # to a non-rotating FileHandler and grow forever — warn, matching the
+            # unknown-`output` convention, rather than degrade silently
+            log.warning(
+                "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
 
@@ -122,6 +178,8 @@ def setup_logging(
     module_levels: Optional[Dict[str, Union[int, str]]] = None,
     rotate: Optional[str] = "daily",
     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,
@@ -132,7 +190,16 @@ 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`.
+    `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 the
+    live file latest.log (never latest.log.log).
+    `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, and the rest are deleted (total retained =
+    sum). this applies to "on_start", "daily", and "size". `backup_count` and the
+    gzip-on-roll behavior of `compress` are IGNORED in tiered mode (the tier counts bound
+    retention). pass NEITHER knob and rotation behaves exactly as before (backup_count +
+    compress) — existing callers are unaffected.
     `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", "aiohttp": "WARNING"}) from
@@ -154,7 +221,7 @@ def setup_logging(
     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.
     """
-    global _listener
+    global _listener, _atexit_registered
 
     root = logging.getLogger()
     root.setLevel(_level_value(level))
@@ -162,7 +229,8 @@ def setup_logging(
     _clear_owned(root)
 
     formatter = build_formatter(output, fmt, datefmt)
-    live_path = f"{name}.log"
+    stem = _normalize_name(name)
+    live_path = f"{stem}.log"
 
     handlers = []
 
@@ -174,7 +242,10 @@ def setup_logging(
 
     if file_ok:
         try:
-            fh = _file_handler(name, live_path, log_dir, rotate, backup_count, max_bytes, compress)
+            fh = _file_handler(
+                stem, live_path, log_dir, rotate, backup_count, max_bytes, compress,
+                keep_uncompressed, keep_compressed,
+            )
             fh.setFormatter(formatter)
             handlers.append(fh)
         except OSError:
@@ -186,12 +257,16 @@ def setup_logging(
         handlers.append(sh)
 
     if queue:
-        record_queue: "queue.Queue" = _make_queue()
+        record_queue: "_queue.Queue" = _make_queue()
         qh = _tag(logging.handlers.QueueHandler(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, so repeated queue re-setups would
+            # otherwise stack identical callbacks (harmless but unbounded)
+            atexit.register(_stop_listener)
+            _atexit_registered = True
     else:
         for handler in handlers:
             root.addHandler(_tag(handler))
@@ -202,9 +277,9 @@ def setup_logging(
     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: