feat(cli): write a log in the project when a command has a problem

CHANGELOG.md

@@ -13,10 +13,12 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 - Local metadata store in SQLite, the first implementation of the metadata port. Its schema is versioned and upgrades itself when a store is opened, after making a backup.
 - `miniops init` creates a project in the current directory: it writes `miniops.toml` and adds `.miniops/` to `.gitignore`.
 - `miniops config show` prints the configuration of the project that the current directory belongs to.
+- A command that has a problem inside a project leaves its log in `.miniops/logs/miniops.log`: what it did and the full traceback of the error. A command without warnings or errors writes nothing there.
 
 ### Changed
 
 - An error that miniOPs reports ends with a short message and exit code 1 instead of a traceback.
+- An unexpected error inside a project shows a short message with the path of the log instead of a traceback. Outside a project, where there is no log, the traceback is still shown.
 - pydantic is now a dependency. It validates the configuration files.
 
 ## [0.1.0a1] - 2026-11-11

src/miniops/__init__.py

@@ -1,7 +1,11 @@
 """miniOPs: a lightweight, polars-native MLOps CLI."""
 
+import logging
 from importlib.metadata import version
 
 __version__ = version("miniops-ml")
 
 __all__ = ["__version__"]
+
+# Without any handler, Python prints the warnings and errors of a library to stderr itself.
+logging.getLogger(__name__).addHandler(logging.NullHandler())

src/miniops/adapters/local/schema.py

@@ -1,9 +1,12 @@
+import logging
 import sqlite3
 from contextlib import closing
 from pathlib import Path
 
 from miniops.core.errors import IncompatibleStoreError
 
+logger = logging.getLogger(__name__)
+
 # A released script never changes: a change to the schema is a new script at the end.
 MIGRATIONS: tuple[str, ...] = (
     """
@@ -102,9 +105,11 @@ def migrate(connection: sqlite3.Connection, path: Path) -> None:
     except BaseException:
         connection.execute("ROLLBACK")
         raise
+    logger.info("The store %s now has schema version %d", path, newest)
 
 
 def _back_up(connection: sqlite3.Connection, path: Path, version: int) -> None:
     backup = path.with_name(f"{path.name}.v{version}.bak")
     with closing(sqlite3.connect(backup)) as copy:
         connection.backup(copy)
+    logger.info("Copied the store %s to %s before upgrading its schema", path, backup)

src/miniops/cli/__init__.py

@@ -1,15 +1,41 @@
+import logging
+import shlex
+import sys
+
 import typer
 
+from miniops import __version__
 from miniops.cli.app import app
+from miniops.cli.log import project_log
 from miniops.core.errors import MiniopsError
 
 __all__ = ["app", "main"]
 
+logger = logging.getLogger(__name__)
+
 
 def main() -> None:
-    """Run the command line. An error that miniOPs reports ends with a message, not a traceback."""
-    try:
-        app(prog_name="miniops")
-    except MiniopsError as error:
-        typer.secho(f"Error: {error}", fg=typer.colors.RED, err=True)
-        raise SystemExit(1) from None
+    """Run the command line.
+
+    An error ends with a short message, and its traceback goes to the log of the project.
+    """
+    with project_log() as command_log:
+        logger.info("miniops %s: %s", __version__, shlex.join(sys.argv[1:]))
+        try:
+            app(prog_name="miniops")
+        except MiniopsError as error:
+            logger.error("Command failed: %s", error, exc_info=True)
+            typer.secho(f"Error: {error}", fg=typer.colors.RED, err=True)
+            raise SystemExit(1) from None
+        except Exception as error:
+            logger.critical("Unexpected error", exc_info=True)
+            if command_log is None or not command_log.saved:
+                raise
+            typer.secho(
+                f"Error: miniOPs hit an unexpected error ({type(error).__name__}: {error}). "
+                "This is a bug.",
+                fg=typer.colors.RED,
+                err=True,
+            )
+            typer.echo(f"Details: {command_log.file}", err=True)
+            raise SystemExit(1) from None

src/miniops/cli/log.py

@@ -0,0 +1,95 @@
+import logging
+from collections import deque
+from collections.abc import Generator
+from contextlib import contextmanager, suppress
+from pathlib import Path
+from typing import override
+
+from miniops.config.project import STATE_DIR, find_root
+
+MAX_BYTES = 1_000_000
+KEPT_RECORDS = 1000
+
+_FORMAT = "%(asctime)s %(levelname)s [%(process)d] %(name)s: %(message)s"
+
+
+class CommandLog(logging.Handler):
+    """The log of one command. It reaches the log file only if something goes wrong.
+
+    Records wait in memory. The first warning or error writes them to the file, and after
+    that every record goes there directly. A command without problems leaves the file alone.
+    """
+
+    def __init__(self, file: Path) -> None:
+        """Create a log that writes to ``file`` once it has a reason to."""
+        super().__init__()
+        self.setFormatter(logging.Formatter(_FORMAT))
+        self.file = file
+        self.saved = False
+        self._waiting: deque[logging.LogRecord] = deque(maxlen=KEPT_RECORDS)
+        self._writing = False
+
+    @override
+    def emit(self, record: logging.LogRecord) -> None:
+        self._waiting.append(record)
+        if not self._writing and record.levelno >= logging.WARNING:
+            self._writing = True
+            self._set_aside_a_full_file()
+        if self._writing:
+            self._write()
+
+    def _set_aside_a_full_file(self) -> None:
+        # The rename fails when there is no file yet, or when another miniOPs process is
+        # writing to it (Windows). This command then adds to the file that is there.
+        with suppress(OSError):
+            if self.file.stat().st_size > MAX_BYTES:
+                self.file.replace(self.file.with_name(f"{self.file.name}.1"))
+
+    def _write(self) -> None:
+        lines = []
+        while self._waiting:
+            record = self._waiting.popleft()
+            try:
+                lines.append(self.format(record))
+            except Exception:
+                self.handleError(record)
+        if not lines:
+            return
+        # A command never fails because of its log.
+        with suppress(OSError):
+            self.file.parent.mkdir(parents=True, exist_ok=True)
+            with self.file.open("a", encoding="utf-8", errors="backslashreplace") as file:
+                file.writelines(f"{line}\n" for line in lines)
+            self.saved = True
+
+
+@contextmanager
+def project_log(start: Path | None = None) -> Generator[CommandLog | None]:
+    """Collect what miniOPs logs during a command, for the project that ``start`` is in.
+
+    ``start`` is the current directory unless given. Yields None when there is no project.
+    """
+    file = _log_file(start)
+    if file is None:
+        yield None
+        return
+    command_log = CommandLog(file)
+    logger = logging.getLogger("miniops")
+    level = logger.level
+    logger.addHandler(command_log)
+    logger.setLevel(logging.DEBUG)
+    try:
+        yield command_log
+    finally:
+        logger.setLevel(level)
+        logger.removeHandler(command_log)
+        command_log.close()
+
+
+def _log_file(start: Path | None) -> Path | None:
+    # Path.cwd() fails when the current directory was removed.
+    with suppress(OSError):
+        root = find_root(start or Path.cwd())
+        if root is not None:
+            return root / STATE_DIR / "logs" / "miniops.log"
+    return None

src/miniops/config/project.py

@@ -50,7 +50,7 @@ def find_project(start: Path) -> Project:
     Raises:
         ConfigError: If there is no project, or its ``miniops.toml`` is not valid.
     """
-    root = _find_root(start)
+    root = find_root(start)
     if root is None:
         raise ConfigError(
             f"No miniOPs project in {start} or above it. Create one with `miniops init`."
@@ -67,7 +67,7 @@ def init_project(root: Path, name: str | None = None) -> Project:
         AlreadyExistsError: If the directory is a project already, or lies inside one.
         ConfigError: If the project would have no name.
     """
-    existing = _find_root(root)
+    existing = find_root(root)
     if existing is not None:
         raise AlreadyExistsError(f"There is already a miniOPs project in {existing}.")
     name = (name or root.name).strip()
@@ -84,7 +84,11 @@ def init_project(root: Path, name: str | None = None) -> Project:
     return Project(root=root, config=_load(config_file))
 
 
-def _find_root(start: Path) -> Path | None:
+def find_root(start: Path) -> Path | None:
+    """Return the directory of the project that ``start`` is in, or None.
+
+    It only looks for ``miniops.toml`` and does not read it.
+    """
     start = start.resolve()
     for directory in (start, *start.parents):
         if (directory / CONFIG_FILE).is_file():

tests/adapters/local/test_schema.py

@@ -1,3 +1,4 @@
+import logging
 import sqlite3
 from contextlib import closing
 from datetime import UTC, datetime
@@ -42,7 +43,7 @@ def test_a_new_store_gets_the_newest_schema(tmp_path: Path) -> None:
 
 
 def test_an_older_store_is_upgraded_after_a_backup(
-    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
 ) -> None:
     path = tmp_path / "metadata.db"
     with SqliteMetadataStore(path) as store:
@@ -52,10 +53,14 @@ def test_an_older_store_is_upgraded_after_a_backup(
         schema, "MIGRATIONS", (*schema.MIGRATIONS, "CREATE TABLE deployments (id TEXT);")
     )
 
-    with SqliteMetadataStore(path) as store:
+    with caplog.at_level(logging.INFO, logger="miniops"), SqliteMetadataStore(path) as store:
         assert store.list_models() == [MODEL]
 
     backup = tmp_path / f"metadata.db.v{older}.bak"
+    assert caplog.messages == [
+        f"Copied the store {path} to {backup} before upgrading its schema",
+        f"The store {path} now has schema version {older + 1}",
+    ]
     assert version_of(path) == older + 1
     assert "deployments" in tables_of(path)
     assert version_of(backup) == older