fix(logging): keep agent diagnostics quiet and correlated (ENG-3579)

README.md

@@ -372,3 +372,9 @@ Anything under [`.github/`](.github/) is owned by `@mindsdb/devops` via [CODEOWN
 ### Hotfixes / out-of-band releases
 
 If you genuinely need to release outside the normal flow (e.g. an admin hotfix), coordinate with `@mindsdb/devops` to bypass the tag ruleset. The e2e workflow's version-match guard will still verify the release tag matches `anton.__version__` and fail loudly on mismatch.
+
+## Diagnostics
+
+The CLI, cloud-turn, and resident launcher share safe diagnostic settings.
+See [structured logging](docs/structured-logging.md) for defaults, troubleshooting
+overrides, retained evidence streams and rollout requirements.

anton/cli.py

@@ -664,6 +664,8 @@ def main(
     ),
 ) -> None:
     """Anton — a self-evolving autonomous system."""
+    from anton.diagnostics import configure_logging
+    configure_logging()
     from anton.config.settings import AntonSettings
 
     settings = AntonSettings()

anton/cloud_turn/__main__.py

@@ -23,6 +23,8 @@
 import inspect
 import json
 import logging
+from anton.diagnostics import configure_logging, make_formatter
+from structlog.contextvars import bind_contextvars, bound_contextvars
 import os
 import sys
 import time
@@ -118,6 +120,7 @@ def _stderr_log_handler(*, stream: TextIO) -> logging.Handler:
     sends to it, so no logged traceback prints a provider error body."""
     handler = logging.StreamHandler(stream)
     handler.addFilter(ProviderErrorFilter())
+    handler.setFormatter(make_formatter())
     return handler
 
 
@@ -136,7 +139,8 @@ def _isolated_protocol_stdout():
     sys.stdout = sys.stderr
     # The entrypoint owns the pod's logging. `force` replaces any root handler
     # configured before main(), so the filtered handler is the only one on stderr.
-    logging.basicConfig(handlers=[_stderr_log_handler(stream=sys.stderr)], level=logging.INFO, force=True)
+    configure_logging(stream=sys.stderr)
+    logging.getLogger().handlers[0].addFilter(ProviderErrorFilter())
 
     def emit(event: dict) -> None:
         data = (json.dumps(event) + "\n").encode("utf-8")
@@ -224,6 +228,13 @@ async def _settle_memory(session) -> None:
 
 async def stream_turn(
     raw_line: str, emit, session_builder=None, stdin: BinaryIO | None = None
+) -> None:
+    with bound_contextvars(conversation_id=None):
+        await _stream_turn(raw_line, emit, session_builder, stdin)
+
+
+async def _stream_turn(
+    raw_line: str, emit, session_builder=None, stdin: BinaryIO | None = None
 ) -> None:
     """Parse the request, run one turn, and emit exactly one terminal event.
 
@@ -281,6 +292,7 @@ async def _heartbeat() -> None:
     elicitor = None
     try:
         req = TurnRequestV1.from_json(raw_line)
+        bind_contextvars(conversation_id=req.conversation_id)
         build = functools.partial(builder, req)
         if req.interactive and stdin is not None:
             # Before the build: the build is what registers `ask_user`, and a
@@ -292,7 +304,7 @@ async def _heartbeat() -> None:
         # seconds (a turn-key OAuth token fetch per connector) — run it off
         # the event loop thread so the heartbeat ticker above keeps firing
         # instead of the controller seeing the turn go silent and stall it.
-        session = await asyncio.get_running_loop().run_in_executor(None, build)
+        session = await asyncio.to_thread(build)
         # Fold in any attachments cowork-server staged into the workspace (read
         # off the mount, so image bytes never crossed the stdin wire). Plain
         # string when there are none. Never fail the turn over this — degrade to
@@ -345,7 +357,7 @@ async def _heartbeat() -> None:
                 wire({"kind": "tool_end", "id": event.id,
                       "args": args[:MAX_STEP_CHARS]})
             elif isinstance(event, StreamTaskProgress):
-                logger.info("progress [%s]: %s", event.phase, event.message)
+                logger.info("Turn progress", extra={"event": "turn_progress"})
                 phase = event.phase or ""
                 # A tool_progress line with an id is a step announcement: the
                 # first one opens the renderer's step, every one is a step the
@@ -380,7 +392,7 @@ async def _heartbeat() -> None:
                       "action": event.action,
                       "content": _clip_result_content(event.content or "")})
             elif isinstance(event, StreamContextCompacted):
-                logger.info("context compacted: %s", event.message)
+                logger.debug("Context compacted", extra={"event": "context_compacted"})
                 wire({"kind": "compacted", "message": event.message})
             elif isinstance(event, StreamAskUser):
                 logger.info("ask_user question: %s", event.id)
@@ -451,7 +463,7 @@ async def _heartbeat() -> None:
         terminal_emitted = True
     except Exception as exc:
         # ProviderErrorFilter omits provider bodies from diagnostics; the wire is unchanged.
-        logger.exception("cloud turn failed")
+        logger.exception("cloud turn failed", extra={"event": "turn_failed", "reason": "execution_failed"})
         failed = {"kind": "turn_failed", "error": _scrub(exc)}
         # A billing stop's reset instant rides beside the string, so the host
         # can say when the limit lifts. `reset_at` is already a normalized

anton/cloud_turn/resident.py

@@ -36,6 +36,7 @@
     forbid_same_user_access,
     socket_address,
 )
+from anton.diagnostics import make_formatter
 
 logger = logging.getLogger(__name__)
 
@@ -331,12 +332,14 @@ def _run_cloud_turn() -> int | None:
 
 
 def main() -> int:
+    level = os.environ.get("LOG_LEVEL", "WARNING").upper()
     handler = logging.StreamHandler(sys.stderr)
-    handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s"))
+    handler.setFormatter(make_formatter())
+    handler.setLevel(level)
     # Its own handler rather than the root logger's: a fork must start with the
     # root unconfigured, as the turn configures it.
-    logger.addHandler(handler)
-    logger.setLevel(logging.INFO)
+    logger.handlers = [handler]
+    logger.setLevel(level)
     logger.propagate = False
     return serve(_run_cloud_turn)
 

anton/core/llm/provider.py

@@ -2292,6 +2292,10 @@ def filter(self, record: logging.LogRecord) -> bool:
             provider_error = _provider_error(exc=exc)
             if provider_error is None:
                 continue
+            record._diagnostic_safe_template = True
+            record.error_type = type(exc).__name__
+            record.provider_error = type(provider_error).__name__
+            record.status_code = _log_status(status=getattr(provider_error, "status_code", None))
             record.msg = "Provider operation failed: error_type=%s provider_error=%s status=%s"
             record.args = (
                 type(exc).__name__, type(provider_error).__name__,

anton/diagnostics.py

@@ -0,0 +1,191 @@
+"""Stdlib logging with a shared JSON envelope and explicit diagnostic fields."""
+
+import logging
+import os
+import sys
+import traceback
+from datetime import UTC, datetime
+from math import isfinite
+from pathlib import Path
+from uuid import UUID
+
+import structlog
+
+EVENT_FIELDS = frozenset(
+    [
+        "event",
+        "reason",
+        "outcome",
+        "operation",
+        "provider",
+        "error_type",
+        "error_class",
+        "provider_error",
+        "status_code",
+        "request_id",
+        "request_id_source",
+        "correlation_id",
+        "entry_id",
+        "scratchpad_id",
+        "pod_name",
+        "conversation_id",
+        "project_id",
+        "schedule_id",
+        "turn_id",
+        "job_id",
+        "trace_id",
+        "method",
+        "route",
+        "duration_ms",
+        "user_id",
+        "organization_id",
+        "workspace_id",
+        "attempt",
+        "attempts",
+        "count",
+        "model_alias",
+        "retryable",
+    ]
+)
+
+
+def safe_exception_traceback(exc_info: tuple) -> str | None:
+    """Keep types and stack locations, excluding values, source, notes and locals."""
+    if not exc_info or exc_info[1] is None:
+        return None
+    lines, seen = [], set()
+
+    def visit(error, tb):
+        if error is None or id(error) in seen:
+            return
+        seen.add(id(error))
+        for frame in traceback.extract_tb(tb):
+            lines.append(
+                f'  File "{Path(frame.filename).name}", line {frame.lineno}, in {frame.name}'
+            )
+        lines.append(f"{type(error).__name__}: [error details omitted]")
+        if isinstance(error, BaseExceptionGroup):
+            for child in error.exceptions:
+                visit(child, child.__traceback__)
+        cause = error.__cause__ or (
+            None if error.__suppress_context__ else error.__context__
+        )
+        if cause is not None:
+            visit(cause, cause.__traceback__)
+
+    visit(exc_info[1], exc_info[2])
+    return "\n".join(lines)
+
+
+def _event_schema(logger, method_name, event_dict):
+    record = event_dict["_record"]
+    if event_dict.get("exc_info"):
+        # Exception messages may repeat a request or provider payload.
+        event_dict["message"] = "Operation failed"
+        event_dict.setdefault("event", "operation_failed")
+        event_dict.setdefault("operation", record.funcName)
+    args = event_dict.pop("positional_args", ())
+    # Keep static message templates; formatting positional values can expose
+    # provider replies, generated code, private paths or credential strings.
+    if (
+        args
+        and isinstance(record.msg, str)
+        and not getattr(record, "_diagnostic_safe_template", False)
+    ):
+        event_dict["message"] = record.msg
+    if not record.name.startswith(("anton", "__main__")):
+        event_dict["message"] = "Dependency diagnostic"
+        event_dict.setdefault("event", "dependency_diagnostic")
+    errors = [
+        arg
+        for arg in (args.values() if isinstance(args, dict) else args)
+        if isinstance(arg, BaseException)
+    ]
+    if errors:
+        event_dict["message"] = "Operation failed"
+        event_dict["error_type"] = type(errors[0]).__name__
+        event_dict.setdefault("event", "operation_failed")
+        event_dict.setdefault("operation", record.funcName)
+    if not isinstance(record.msg, str):
+        event_dict["message"] = "Operation diagnostic"
+    event_dict.setdefault(
+        "event",
+        "operation_failed"
+        if record.levelno >= logging.WARNING
+        else "operation_progress",
+    )
+    event_dict.setdefault("operation", record.funcName)
+    payload = {
+        "logged_at": datetime.fromtimestamp(record.created, tz=UTC).isoformat(),
+        "logger": record.name,
+        "log_level": record.levelname,
+        "message": event_dict["message"],
+    }
+    for field in EVENT_FIELDS:
+        value = event_dict.get(field)
+        if isinstance(value, UUID):
+            value = str(value)
+        if type(value) in (str, bool, int) or (
+            type(value) is float and isfinite(value)
+        ):
+            payload[field] = value
+    exc_info = event_dict.get("exc_info")
+    if exc_info:
+        payload["exc_info"] = exc_info
+        if isinstance(exc_info, tuple) and exc_info[1] is not None:
+            payload["error_type"] = type(exc_info[1]).__name__
+    return payload
+
+
+def make_formatter():
+    log_format = os.environ.get("LOG_FORMAT", "json")
+    if log_format not in {"json", "text"}:
+        raise ValueError("LOG_FORMAT must be json or text")
+    renderer = (
+        structlog.processors.JSONRenderer(
+            separators=(",", ":"), ensure_ascii=False, allow_nan=False
+        )
+        if log_format == "json"
+        else structlog.processors.KeyValueRenderer(
+            key_order=["logged_at", "log_level", "logger", "message"]
+        )
+    )
+    return structlog.stdlib.ProcessorFormatter(
+        pass_foreign_args=True,
+        foreign_pre_chain=[
+            structlog.processors.EventRenamer("message"),
+            structlog.contextvars.merge_contextvars,
+            structlog.stdlib.ExtraAdder(allow=EVENT_FIELDS),
+        ],
+        processors=[
+            _event_schema,
+            structlog.processors.ExceptionRenderer(safe_exception_traceback),
+            renderer,
+        ],
+    )
+
+
+def configure_logging(level: str | None = None, *, stream=None) -> None:
+    """Configure process diagnostics without redirecting protocol output."""
+    level = (level or os.environ.get("LOG_LEVEL", "WARNING")).upper()
+    handler = logging.StreamHandler(stream or sys.stderr)
+    handler.setFormatter(make_formatter())
+    handler.setLevel(level)
+    logging.basicConfig(level=level, handlers=[handler], force=True)
+    for name in (
+        "httpx",
+        "httpx2",
+        "httpcore",
+        "httpcore2",
+        "openai",
+        "anthropic",
+        "urllib3",
+        "kubernetes",
+    ):
+        logging.getLogger(name).setLevel(logging.WARNING)
+    usage_handler = logging.StreamHandler(stream or sys.stderr)
+    usage_handler.setFormatter(logging.Formatter("%(message)s"))
+    usage = logging.getLogger("anton.usage")
+    usage.setLevel(logging.INFO)
+    usage.handlers = [usage_handler]
+    usage.propagate = False

docs/structured-logging.md

@@ -0,0 +1,35 @@
+# Runtime diagnostics
+
+`LOG_LEVEL=WARNING` and `LOG_FORMAT=json` are the process defaults. Set
+`LOG_LEVEL=DEBUG` (or `INFO`) temporarily for CLI/cloud-turn diagnostics; unset it
+to restore WARNING. `LOG_FORMAT=text` is a local readability override with the
+same exception privacy policy. Restore `json` for collected deployments.
+
+The packaged `cloud-turn-resident` uses the same settings and safe formatter for
+setup and fork failures. It keeps its own handler so forked turns can configure
+the root logger independently.
+
+Cloud turns bind `conversation_id` for their entire lifetime and reset it when
+the turn ends. Dependency loggers remain at WARNING during troubleshooting.
+`anton.usage` remains an independent plain-text INFO stream for its existing
+usage/evaluation consumers. Agent stdout/protocol events, generated output,
+saved conversations and tracing are separate contracts and retain their current
+content. Arbitrary user-code output is not claimed to be a safe diagnostic.
+
+Publish an Anton release before downstream packages or images claim these
+diagnostics. Rebuild those consumers with that release; changing a source
+checkout does not update installed agents.
+
+Diagnostics use `logged_at`, `logger`, `log_level`, and `message`. Explicit events,
+reasons, source locations and correlation fields identify failures. Exception
+values, source lines, locals, arbitrary extras, HTTP queries and provider bodies
+are excluded. Use the structured fields when querying logs; the previous text
+prefixes are no longer a parsing contract.
+
+Enable extra detail only for an owned troubleshooting session, repeat the failing
+operation, and restore the default afterward. Logging controls do not enable
+framework debug responses. Roll back a release by restoring its previous image
+or package version and its previous logging settings together. Before promotion,
+verify collector parsing, saved searches, alerts and retained operational streams
+against the emitted records. A passing unit suite does not establish deployment
+or collector compatibility.

pyproject.toml

@@ -61,11 +61,12 @@ dependencies = [
     "pyyaml>=6.0",
     "dill==0.3.8",
     "truststore>=0.10.4",
-    # Reads .xlsx artifacts for the structural lint 
+    # Reads .xlsx artifacts for the structural lint
     "openpyxl>=3.1,<4",
     # MCP client (anton/core/mcp/) — v2 renamed FastMCP to MCPServer and
     # replaced the old in-memory test helper; pin the major version.
     "mcp>=2,<3",
+    "structlog==26.1.0",
 ]
 
 [project.optional-dependencies]

tests/test_cloud_turn_entrypoint.py

@@ -312,7 +312,8 @@ async def turn_stream(self, user_input, **kwargs):
     assert {"kind": "delta", "text": "hi"} in events
     assert {"kind": "turn_completed"} in events
     assert "tool call: scratchpad" in caplog.text
-    assert "progress [scratchpad_start]: running cell" in caplog.text
+    assert "running cell" not in caplog.text
+    assert any(record.name == "anton.cloud_turn.__main__" and record.getMessage() == "Turn progress" for record in caplog.records)
     assert "tool result: scratchpad action=exec (20 chars)" in caplog.text
     assert "cloud turn completed" in caplog.text
     # the answer text is emitted, never narrated into the logs