Keep Logfire and OpenTelemetry export warnings out of the interactive clai2 terminal (#9796)

src/pydantic_clai2/AGENTS.md

@@ -321,6 +321,7 @@ Keep documented plugin-author paths (`pydantic_clai2.plugins` and
 | `builtin_plugins/logfire_session.py` | the `observability` plugin's `CLAI session` roots: `SessionTracing` and the `git_email` lookup for `user_tag: git-email` |
 | `builtin_plugins/logfire_setup.py` | the `observability` plugin's setup menu: which Logfire, Logfire's device sign-in (not the MCP OAuth in `logfire_oauth.py`), account email from `/v1/account/me`, project pick, write token saved in `/keys` |
 | `builtin_plugins/logfire_destination.py` | "which Logfire", shared by `observability` setup and `logfire_mcp`: regions from harness, a host, UI URL, or MCP URL parsed to one origin, the picker, and the last pick remembered for the other plugin |
+| `ui/telemetry_log.py` | sends unhandled `logfire` and `opentelemetry` logger warnings to `telemetry.log` during the interactive session, so `logging.lastResort` never writes over the editor |
 | `ui/telemetry.py` | UI telemetry sinks, `record`/`span`, the menu naming, and `conversation_selected`, which calls every sink's session `root` (a sink subscribed with `ui_events=False` gets only that) once startup picks the conversation; instrument shared chokepoints (`run_worker`, `Commands.execute_async`, `FieldMenu`, the loader, `/keys`, the prompt), never one menu at a time, and record names, not content; typed text goes only through `prompt_text`, which the subscriber's `include_content` gates |
 | `builtin_plugins/compaction.py` | the built-in `compaction` plugin: harness `FallbackCompaction([SummarizingCompaction, SlidingWindowCompaction])`, `/compact`, the context alert, its settings menu |
 | `commands.py` | `Command`, the registry, completion |

src/pydantic_clai2/PLUGINS.md

@@ -350,6 +350,15 @@ such as explicit OTLP exporters, still applies. The SDKs' own log messages, such
 failed or retried exports, never print to the terminal; a logging handler that a
 plugin configures still receives them.
 
+Export problems, such as a request to Logfire timing out, are logged as
+warnings and errors by the `logfire` and `opentelemetry` loggers. Logfire retries
+failed span exports from disk; OpenTelemetry drops a metrics batch it could not
+send. In the interactive shell these records go to `telemetry.log` next to
+`config.db` (owner-only, rotated at 1 MB, one previous file kept) instead of over the editor,
+and CLAI names that file on exit when the session wrote to it. A logger your own
+application already handles, when embedding `chat()`, is left alone. Headless
+`clai2 -p` still writes them to stderr, separate from the answer on stdout.
+
 Previously named `logfire`, this plugin keeps existing enabled/disabled choices,
 settings, and saved token references. No reconfiguration is needed. Old commands
 and project or drop-in declarations using `logfire` refer to the same plugin,

src/pydantic_clai2/README.md

@@ -2206,6 +2206,9 @@ cannot choose the telemetry destination. Relative `XDG_CONFIG_HOME` values fall
 back to `~/.config`. Export uses `send_to_logfire='if-token-present'`: no credentials
 means no Logfire export and no interactive project setup. Logfire's
 terminal console output is disabled so it does not interfere with the editor.
+Warnings and errors Logfire and OpenTelemetry log about failed exports go to
+`telemetry.log` next to `config.db` instead of over the editor; CLAI names the
+file on exit when a session wrote to it. Headless `clai2 -p` keeps them on stderr.
 Standard SDK configuration, including explicitly configured OTLP exporters, still
 applies; disable the plugin to stop its instrumentation altogether.
 

src/pydantic_clai2/pydantic_clai2/_app.py

@@ -105,6 +105,7 @@ from pydantic_clai2.ui.rendering.spinners import Spinner, Spinners
 from pydantic_clai2.ui.rendering.status import Status, StatusLine
 from pydantic_clai2.ui.rendering.tool_output import terminal_text
 from pydantic_clai2.ui.rendering.usage_report import cost_line, session_usage
+from pydantic_clai2.ui.telemetry_log import telemetry_log
 
 if TYPE_CHECKING:
     from pydantic_clai2.auth import CodexAuth
@@ -302,87 +303,90 @@ async def chat(
         )
     fresh = False
     warming: Thread | None = None
-    async with agent:
-        while True:
-            reason: SessionEndReason = 'error'
-            with theme.use(lambda: shell.context.settings.theme, output=console.file if console.is_terminal else None):
+    with telemetry_log(shell.context.store.path.with_name('telemetry.log'), console=console):
+        async with agent:
+            while True:
+                reason: SessionEndReason = 'error'
+                with theme.use(
+                    lambda: shell.context.settings.theme, output=console.file if console.is_terminal else None
+                ):
+                    try:
+                        async with create_task_group() as workers:
+                            workers.start_soon(shell.sessions.namer.run)
+                            try:
+                                with (
+                                    transcript.capture(console),
+                                    shell.defer_identity() if resume is not None else nullcontext(),
+                                ):
+                                    await shell.loader.load_all(fresh=fresh)
+                                    _report_project_plugins(shell.loader, console)
+                                    if resume is not None:
+                                        source = [resume_from] if resume_from else []
+                                        console.print(
+                                            await shell.sessions.command([*source, resume] if resume else source),
+                                            markup=False,
+                                        )
+                                        resume = None
+                                warming = warming or warm_imports.start()
+                                reason = await shell.run()
+                            finally:
+                                workers.cancel_scope.cancel()
+                    except BaseExceptionGroup as exc:
+                        if len(exc.exceptions) == 1:
+                            raise exc.exceptions[0] from None
+                        raise
+                    finally:
+                        with transcript.capture(console):
+                            await shell.loader.close(reason)
+                if not shell.reload_requested:
+                    if (executable := shell.updates.relaunch) is not None:
+                        summary = shell.session.summary
+                        raise Relaunch(executable=executable, session_id=summary.id if summary.revision else None)
+                    return
+                shell.reload_requested = False
+                if warming is not None:  # pragma: no branch -- a reload follows a run, which started warming
+                    warming.join()
                 try:
-                    async with create_task_group() as workers:
-                        workers.start_soon(shell.sessions.namer.run)
-                        try:
-                            with (
-                                transcript.capture(console),
-                                shell.defer_identity() if resume is not None else nullcontext(),
-                            ):
-                                await shell.loader.load_all(fresh=fresh)
-                                _report_project_plugins(shell.loader, console)
-                                if resume is not None:
-                                    source = [resume_from] if resume_from else []
-                                    console.print(
-                                        await shell.sessions.command([*source, resume] if resume else source),
-                                        markup=False,
-                                    )
-                                    resume = None
-                            warming = warming or warm_imports.start()
-                            reason = await shell.run()
-                        finally:
-                            workers.cancel_scope.cancel()
-                except BaseExceptionGroup as exc:
-                    if len(exc.exceptions) == 1:
-                        raise exc.exceptions[0] from None
-                    raise
-                finally:
-                    with transcript.capture(console):
-                        await shell.loader.close(reason)
-            if not shell.reload_requested:
-                if (executable := shell.updates.relaunch) is not None:
-                    summary = shell.session.summary
-                    raise Relaunch(executable=executable, session_id=summary.id if summary.revision else None)
-                return
-            shell.reload_requested = False
-            if warming is not None:  # pragma: no branch -- a reload follows a run, which started warming
-                warming.join()
-            try:
-                shell = reload_clai(
-                    lambda shell=shell: create_shell(
-                        rebuild_stock(()) if rebuild_stock is not None else agent,
-                        deps=deps,
-                        plugins=plugins,
-                        usage_limits=shell.session.usage_limits,
-                        console=console,
-                        settings=shell.context.settings,
-                        store=SettingsStore(shell.context.store.path),
-                        builtin_plugins=(
-                            STOCK_PLUGINS
-                            if use_stock_defaults
-                            else DEFAULT_PLUGINS
-                            if use_defaults
-                            else builtin_plugins
-                        ),
-                        project=project,
-                        message_history=shell.session.messages,
-                        summary=shell.session.summary,
-                        transcript=shell.transcript,
-                        load_plugins=load_plugins,
-                    )
-                )
-            except Exception as exc:  # noqa: BLE001 -- development edits must not discard the conversation.
-                with transcript.capture(console):
-                    console.print(
-                        f'Reload failed: {type(exc).__name__}: {exc}', style=theme.color(theme.ERROR), markup=False
-                    )
-                    if isinstance(exc, ImportError) and exc.name and exc.name.startswith('pydantic_ai_harness'):
-                        console.print(
-                            'Harness is not refreshed by /reload. Restart CLAI2 with the same launch options and '
-                            '--resume to continue this session. Keep the worktree if asked to remove it.',
-                            style=theme.color(theme.INFO),
-                            markup=False,
+                    shell = reload_clai(
+                        lambda shell=shell: create_shell(
+                            rebuild_stock(()) if rebuild_stock is not None else agent,
+                            deps=deps,
+                            plugins=plugins,
+                            usage_limits=shell.session.usage_limits,
+                            console=console,
+                            settings=shell.context.settings,
+                            store=SettingsStore(shell.context.store.path),
+                            builtin_plugins=(
+                                STOCK_PLUGINS
+                                if use_stock_defaults
+                                else DEFAULT_PLUGINS
+                                if use_defaults
+                                else builtin_plugins
+                            ),
+                            project=project,
+                            message_history=shell.session.messages,
+                            summary=shell.session.summary,
+                            transcript=shell.transcript,
+                            load_plugins=load_plugins,
                         )
-                fresh = False
-            else:
-                with transcript.capture(console):
-                    console.print('CLAI2 reloaded. Conversation preserved.', style=theme.color(theme.INFO))
-                fresh = True
+                    )
+                except Exception as exc:  # noqa: BLE001 -- development edits must not discard the conversation.
+                    with transcript.capture(console):
+                        console.print(
+                            f'Reload failed: {type(exc).__name__}: {exc}', style=theme.color(theme.ERROR), markup=False
+                        )
+                        if isinstance(exc, ImportError) and exc.name and exc.name.startswith('pydantic_ai_harness'):
+                            console.print(
+                                'Harness is not refreshed by /reload. Restart CLAI2 with the same launch options and '
+                                '--resume to continue this session. Keep the worktree if asked to remove it.',
+                                style=theme.color(theme.INFO),
+                                markup=False,
+                            )
+                    fresh = False
+                else:
+                    with transcript.capture(console):
+                        console.print('CLAI2 reloaded. Conversation preserved.', style=theme.color(theme.INFO))
+                    fresh = True
 
 
 def _parse(name: str) -> ModelRef:

src/pydantic_clai2/pydantic_clai2/ui/telemetry_log.py

@@ -0,0 +1,84 @@
+"""Keep telemetry SDK log records out of the interactive terminal.
+
+Logfire and OpenTelemetry report export problems, such as a request to Logfire timing out, through `logging`.
+With no handler configured, Python's last-resort handler writes them to stderr, over the editor. While the
+interactive session runs they go to a small rotating file instead.
+"""
+
+import logging
+import os
+from collections.abc import Generator
+from contextlib import contextmanager
+from io import TextIOWrapper
+from logging.handlers import RotatingFileHandler
+from pathlib import Path
+
+from rich.console import Console
+
+from pydantic_clai2.ui.rendering import theme
+
+_LOGGERS = ('logfire', 'opentelemetry')
+_MAX_BYTES = 1_000_000
+
+
+class _LogFile(RotatingFileHandler):
+    """A rotating log file that remembers whether it wrote anything and never reports its own failures."""
+
+    wrote = False
+    _failed = False
+
+    def emit(self, record: logging.LogRecord) -> None:
+        self._failed = False
+        super().emit(record)
+        self.wrote = self.wrote or not self._failed
+
+    def _open(self) -> TextIOWrapper:
+        # Exporter errors can quote endpoint details, so the file (and each rotation's new one) is owner-only.
+        os.close(os.open(self.baseFilename, os.O_CREAT | os.O_APPEND | os.O_WRONLY, 0o600))
+        os.chmod(self.baseFilename, 0o600)
+        return super()._open()
+
+    def handleError(self, record: logging.LogRecord) -> None:
+        # The default prints a traceback to stderr, over the editor; losing the record is the lesser harm.
+        self._failed = True
+
+
+def _unhandled_below(record: logging.LogRecord) -> bool:
+    """Whether no logger between the record's own and the top-level one it propagated to has a handler."""
+    parts = record.name.split('.')
+    loggers = logging.root.manager.loggerDict
+    return not any(
+        isinstance(logger := loggers.get('.'.join(parts[:depth])), logging.Logger) and logger.handlers
+        for depth in range(2, len(parts) + 1)
+    )
+
+
+@contextmanager
+def telemetry_log(path: Path, *, console: Console) -> Generator[None]:
+    """Send warnings and errors from the Logfire and OpenTelemetry loggers to `path`, and name it on exit if written.
+
+    A logger that already reaches a handler, one the embedding application configured on it, an ancestor, or a
+    descendant, is left alone: only records that would otherwise fall through to `logging.lastResort` are redirected.
+    """
+    # `delay` opens the file on the first record, so a session without problems creates nothing.
+    handler = _LogFile(path, maxBytes=_MAX_BYTES, backupCount=1, encoding='utf-8', delay=True)
+    handler.setLevel(logging.WARNING)
+    handler.setFormatter(logging.Formatter('%(asctime)s %(levelname)s %(name)s: %(message)s'))
+    handler.addFilter(_unhandled_below)
+    # `lastResort` only fires when no handler at all is on a record's path, so a handler on these loggers is enough;
+    # records still propagate, so a handler the application adds to the root later sees them too.
+    loggers = [logger for name in _LOGGERS if not (logger := logging.getLogger(name)).hasHandlers()]
+    for logger in loggers:
+        logger.addHandler(handler)
+    try:
+        yield
+    finally:
+        for logger in loggers:
+            logger.removeHandler(handler)
+        handler.close()
+        if handler.wrote:
+            console.print(
+                f'Logfire or OpenTelemetry reported problems, such as failed exports. See {path}.',
+                style=theme.color(theme.MUTED),
+                markup=False,
+            )

tests/clai2/test_telemetry_log.py

@@ -0,0 +1,195 @@
+"""Telemetry SDK warnings, such as failed Logfire exports, go to a file instead of over the interactive editor."""
+
+import io
+import logging
+import os
+from collections.abc import Generator, Sequence
+from contextlib import contextmanager
+from pathlib import Path
+
+import pytest
+from rich.console import Console
+
+from pydantic_ai import Agent
+from pydantic_ai.capabilities import AbstractCapability
+from pydantic_ai.models.test import TestModel
+from pydantic_clai2 import chat
+from pydantic_clai2.cli.command_context import CommandContext
+from pydantic_clai2.commands import Command
+from pydantic_clai2.config import Settings
+from pydantic_clai2.config.settings_store import SettingsStore
+from tests.clai2.test_app_edges import inputs
+
+_METRICS = 'opentelemetry.exporter.otlp.proto.http.metric_exporter'
+_NOTICE = 'Logfire or OpenTelemetry reported problems, such as failed exports. See {}.\n'
+
+
+class _Exports(AbstractCapability[None]):
+    """`/fail` logs what Logfire's and OpenTelemetry's exporters log when a request to Logfire times out."""
+
+    def get_commands(self, context: CommandContext) -> Sequence[Command]:
+        def fail(args: list[str]) -> str:
+            logging.getLogger('logfire').warning('Currently retrying %s failed export(s) (%s bytes)', 1, 955)
+            logging.getLogger(_METRICS).error('Failed to export metrics batch code: %s, reason: %s', None, 'timed out')
+            logging.getLogger('logfire').info('Not a problem')
+            return 'Failed.'
+
+        return [Command(name='fail', description='Log export failures', handler=fail)]
+
+
+@contextmanager
+def unconfigured_logging() -> Generator[None]:
+    """Logging as the CLI leaves it, with no handler, inside the test body.
+
+    pytest's own root handlers would hide `logging.lastResort`, and it adds them again for the call phase, so a
+    fixture cannot remove them; restoring the same list before the call phase ends lets pytest remove its own.
+    """
+    root = logging.getLogger()
+    handlers = root.handlers
+    root.handlers = []
+    try:
+        yield
+    finally:
+        root.handlers = handlers
+
+
+async def _chat(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, commands: list[str]) -> str:
+    inputs(monkeypatch, [*commands, '/exit'])
+    output = io.StringIO()
+    await chat(
+        Agent(TestModel()),
+        deps=None,
+        plugins=[_Exports()],
+        settings=Settings(model='test'),
+        console=Console(file=output, width=1000),
+        store=SettingsStore(tmp_path / 'config.db'),
+    )
+    return output.getvalue()
+
+
+async def test_export_problems_go_to_a_file_not_the_terminal(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
+) -> None:
+    loggers = [logging.getLogger(name) for name in ('logfire', 'opentelemetry')]
+    before = [(logger.propagate, list(logger.handlers)) for logger in loggers]
+
+    # Even with INFO enabled on the logger, only warnings and errors reach the file, as only they reach `lastResort`.
+    logging.getLogger('logfire').setLevel(logging.INFO)
+    try:
+        with unconfigured_logging():
+            text = await _chat(tmp_path, monkeypatch, ['/fail'])
+    finally:
+        logging.getLogger('logfire').setLevel(logging.NOTSET)
+    with unconfigured_logging():
+        # Nothing fell through to `logging.lastResort`, which writes to stderr, over the editor.
+        assert capsys.readouterr().err == ''
+        # The session leaves logging as it found it, so later records reach stderr again.
+        assert [(logger.propagate, list(logger.handlers)) for logger in loggers] == before
+        logging.getLogger('logfire').warning('After the session')
+        assert capsys.readouterr().err == 'After the session\n'
+
+    log = tmp_path / 'telemetry.log'
+    if os.name != 'nt':  # pragma: no branch
+        assert log.stat().st_mode & 0o777 == 0o600
+    assert [line.split(' ', 2)[2] for line in log.read_text().splitlines()] == [
+        'WARNING logfire: Currently retrying 1 failed export(s) (955 bytes)',
+        f'ERROR {_METRICS}: Failed to export metrics batch code: None, reason: timed out',
+    ]
+    assert 'Currently retrying' not in text
+    assert text.endswith(_NOTICE.format(log))
+
+
+async def test_handlers_the_application_configured_still_get_their_records(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """An application embedding `chat()` that handles `logfire` itself keeps those records; the rest are redirected."""
+    stream = io.StringIO()
+    handler = logging.StreamHandler(stream)
+    monkeypatch.setattr(logging.getLogger('logfire'), 'handlers', [handler])
+
+    with unconfigured_logging():
+        text = await _chat(tmp_path, monkeypatch, ['/fail'])
+
+    assert stream.getvalue() == 'Currently retrying 1 failed export(s) (955 bytes)\n'
+    log = tmp_path / 'telemetry.log'
+    assert [line.split(' ', 2)[2] for line in log.read_text().splitlines()] == [
+        f'ERROR {_METRICS}: Failed to export metrics batch code: None, reason: timed out',
+    ]
+    assert text.endswith(_NOTICE.format(log))
+
+
+async def test_handlers_the_application_configured_on_a_descendant_still_get_their_records(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """A record an application handles on a more specific logger is not also written to the file."""
+    stream = io.StringIO()
+    handler = logging.StreamHandler(stream)
+    monkeypatch.setattr(logging.getLogger(_METRICS), 'handlers', [handler])
+
+    with unconfigured_logging():
+        text = await _chat(tmp_path, monkeypatch, ['/fail'])
+
+    assert stream.getvalue() == 'Failed to export metrics batch code: None, reason: timed out\n'
+    log = tmp_path / 'telemetry.log'
+    assert [line.split(' ', 2)[2] for line in log.read_text().splitlines()] == [
+        'WARNING logfire: Currently retrying 1 failed export(s) (955 bytes)',
+    ]
+    assert text.endswith(_NOTICE.format(log))
+
+
+async def test_the_log_rotates_at_its_limit_keeping_one_previous_file(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    log = tmp_path / 'telemetry.log'
+    log.write_bytes(b'x' * 1_000_000)
+    with unconfigured_logging():
+        await _chat(tmp_path, monkeypatch, ['/fail'])
+    assert (tmp_path / 'telemetry.log.1').read_bytes() == b'x' * 1_000_000
+    assert 'Currently retrying' in log.read_text()
+    if os.name != 'nt':  # pragma: no branch
+        assert log.stat().st_mode & 0o777 == 0o600
+
+
+async def test_a_root_handler_added_during_the_session_still_gets_records(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Records keep propagating, so an application handler added on the root mid-session is not cut off."""
+    stream = io.StringIO()
+
+    class _Late(AbstractCapability[None]):
+        def get_commands(self, context: CommandContext) -> Sequence[Command]:
+            def attach(args: list[str]) -> str:
+                logging.getLogger().addHandler(logging.StreamHandler(stream))
+                return 'Attached.'
+
+            return [Command(name='attach', description='Add a root handler', handler=attach)]
+
+    inputs(monkeypatch, ['/attach', '/fail', '/exit'])
+    with unconfigured_logging():
+        await chat(
+            Agent(TestModel()),
+            deps=None,
+            plugins=[_Exports(), _Late()],
+            settings=Settings(model='test'),
+            console=Console(file=io.StringIO(), width=1000),
+            store=SettingsStore(tmp_path / 'config.db'),
+        )
+    assert 'Currently retrying 1 failed export(s) (955 bytes)' in stream.getvalue()
+
+
+async def test_a_session_without_problems_creates_no_file(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    with unconfigured_logging():
+        text = await _chat(tmp_path, monkeypatch, [])
+    assert not (tmp_path / 'telemetry.log').exists()
+    assert 'Logfire or OpenTelemetry' not in text
+
+
+async def test_an_unwritable_file_stays_quiet(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
+) -> None:
+    """A record the file cannot take is dropped: a logging traceback on stderr would land over the editor."""
+    (tmp_path / 'telemetry.log').mkdir()
+    with unconfigured_logging():
+        text = await _chat(tmp_path, monkeypatch, ['/fail'])
+        assert capsys.readouterr().err == ''
+    assert 'Logfire or OpenTelemetry' not in text