From 8a16d3accba1e8750d2e4e3c5a606b1b3df08b31 Mon Sep 17 00:00:00 2001 From: Lucas Koontz Date: Tue, 20 Oct 2026 23:19:04 -0700 Subject: [PATCH 1/2] fix(logging): keep agent diagnostics quiet and correlated Refs: ENG-3579 --- README.md | 5 + anton/cli.py | 2 + anton/cloud_turn/__main__.py | 22 ++- anton/core/llm/provider.py | 4 + anton/diagnostics.py | 191 ++++++++++++++++++++++ docs/structured-logging.md | 31 ++++ pyproject.toml | 3 +- tests/test_cloud_turn_entrypoint.py | 3 +- tests/test_cloud_turn_process.py | 15 +- tests/test_cloud_turn_provider_logging.py | 3 +- tests/test_diagnostics.py | 68 ++++++++ uv.lock | 19 ++- 12 files changed, 350 insertions(+), 16 deletions(-) create mode 100644 anton/diagnostics.py create mode 100644 docs/structured-logging.md create mode 100644 tests/test_diagnostics.py diff --git a/README.md b/README.md index 00ec37e71..c8c411755 100644 --- a/README.md +++ b/README.md @@ -372,3 +372,8 @@ 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 + +See [structured logging](docs/structured-logging.md) for defaults, troubleshooting +overrides, retained evidence streams and rollout requirements. diff --git a/anton/cli.py b/anton/cli.py index 65c866293..2af68d665 100644 --- a/anton/cli.py +++ b/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() diff --git a/anton/cloud_turn/__main__.py b/anton/cloud_turn/__main__.py index fdeac44b8..fe2b0096a 100644 --- a/anton/cloud_turn/__main__.py +++ b/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 diff --git a/anton/core/llm/provider.py b/anton/core/llm/provider.py index 07d68a645..726eb6c6b 100644 --- a/anton/core/llm/provider.py +++ b/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__, diff --git a/anton/diagnostics.py b/anton/diagnostics.py new file mode 100644 index 000000000..668cce6b0 --- /dev/null +++ b/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 diff --git a/docs/structured-logging.md b/docs/structured-logging.md new file mode 100644 index 000000000..070b525e6 --- /dev/null +++ b/docs/structured-logging.md @@ -0,0 +1,31 @@ +# 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. + +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. diff --git a/pyproject.toml b/pyproject.toml index cf70c7cff..9a3bcfaca 100644 --- a/pyproject.toml +++ b/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] diff --git a/tests/test_cloud_turn_entrypoint.py b/tests/test_cloud_turn_entrypoint.py index 8c52fc811..c42fd1f28 100644 --- a/tests/test_cloud_turn_entrypoint.py +++ b/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 diff --git a/tests/test_cloud_turn_process.py b/tests/test_cloud_turn_process.py index c75b005d8..a23f1b2bc 100644 --- a/tests/test_cloud_turn_process.py +++ b/tests/test_cloud_turn_process.py @@ -78,10 +78,17 @@ def test_provider_bodies_stay_off_stderr_for_every_logger(tmp_path, mode): assert events == [{"kind": "turn_failed", "error": "RuntimeError: could not finish the turn"}] assert PROVIDER_PRIVATE not in stdout assert PROVIDER_PRIVATE not in stderr - replaced = "Provider operation failed: error_type={} provider_error=RateLimitError status=429" - assert "WARNING:anton.core.session:" + replaced.format("RuntimeError") in stderr - assert "ERROR:anton.cloud_turn.__main__.child:" + replaced.format("RateLimitError") in stderr - assert "ERROR:anton.cloud_turn.__main__:" + replaced.format("RuntimeError") in stderr + records = [json.loads(line) for line in stderr.splitlines()] + expected = {"anton.core.session": ("WARNING", "RuntimeError"), + "anton.cloud_turn.__main__.child": ("ERROR", "RateLimitError"), + "anton.cloud_turn.__main__": ("ERROR", "RuntimeError")} + for name, (level, error_type) in expected.items(): + record = next(record for record in records if record["logger"] == name) + assert record["log_level"] == level + assert record["error_type"] == error_type + assert record["provider_error"] == "RateLimitError" + assert record["status_code"] == 429 + def test_malformed_input_clean_protocol(tmp_path): diff --git a/tests/test_cloud_turn_provider_logging.py b/tests/test_cloud_turn_provider_logging.py index da38dc8b7..745b38a49 100644 --- a/tests/test_cloud_turn_provider_logging.py +++ b/tests/test_cloud_turn_provider_logging.py @@ -180,7 +180,8 @@ async def turn_stream(self, *args, **kwargs): assert record.getMessage() == "cloud turn failed" assert record.exc_info[1] is error assert "ordinary runtime failure" in logging.Formatter().format(record) - assert "RuntimeError: ordinary runtime failure" in pod_stderr.getvalue() + assert "RuntimeError: [error details omitted]" in pod_stderr.getvalue() + assert "ordinary runtime failure" not in pod_stderr.getvalue() assert events[-1] == {"kind": "turn_failed", "error": _scrub(error)} diff --git a/tests/test_diagnostics.py b/tests/test_diagnostics.py new file mode 100644 index 000000000..b5ab7a578 --- /dev/null +++ b/tests/test_diagnostics.py @@ -0,0 +1,68 @@ +"""The stdlib adapter keeps diagnostic values out of every output format.""" + +import json +import logging +import sys + +import pytest +from structlog.contextvars import bound_contextvars + +from anton.diagnostics import make_formatter + + +@pytest.mark.parametrize("format", ["json", "text"]) +def test_exception_chain_and_args_are_private(monkeypatch, format): + monkeypatch.setenv("LOG_FORMAT", format) + try: + try: + raise ValueError("private-inner") + except ValueError: + raise RuntimeError("private-outer") + except RuntimeError as exc: + record = logging.LogRecord( + "runtime", logging.ERROR, __file__, 1, "Failure: %s", (exc,), sys.exc_info() + ) + record.password = "private-extra" + with bound_contextvars(conversation_id="conversation-1"): + output = make_formatter().format(record) + assert "private-" not in output + assert "conversation-1" in output + assert "RuntimeError" in output and "ValueError" in output + assert "test_exception_chain_and_args" in output + if format == "json": + event = json.loads(output) + assert event["event"] == "operation_failed" + assert event["error_type"] == "RuntimeError" + assert event["log_level"] == "ERROR" + + +def test_exception_as_argument_without_exc_info_is_private(): + record = logging.LogRecord( + "runtime", + logging.WARNING, + __file__, + 1, + "Failure: %s", + (RuntimeError("private-argument"),), + None, + ) + output = make_formatter().format(record) + assert "private-argument" not in output + assert "RuntimeError" in output + + +async def test_concurrent_contexts_reset(): + import asyncio + + from structlog.contextvars import get_contextvars + + async def operation(value): + with bound_contextvars(conversation_id=value): + await asyncio.sleep(0) + record = logging.LogRecord( + "runtime", logging.WARNING, __file__, 1, "Operation failed", (), None + ) + return json.loads(make_formatter().format(record))["conversation_id"] + + assert await asyncio.gather(operation("one"), operation("two")) == ["one", "two"] + assert "conversation_id" not in get_contextvars() diff --git a/uv.lock b/uv.lock index 84112695c..63a5a03c3 100644 --- a/uv.lock +++ b/uv.lock @@ -63,6 +63,7 @@ dependencies = [ { name = "python-dotenv" }, { name = "pyyaml" }, { name = "rich" }, + { name = "structlog" }, { name = "truststore" }, { name = "typer" }, ] @@ -96,6 +97,7 @@ requires-dist = [ { name = "python-dotenv", specifier = ">=0.21" }, { name = "pyyaml", specifier = ">=6.0" }, { name = "rich", specifier = ">=13.0,<15" }, + { name = "structlog", specifier = "==26.1.0" }, { name = "truststore", specifier = ">=0.10.4" }, { name = "typer", specifier = ">=0.12.4" }, ] @@ -342,8 +344,8 @@ name = "httpcore2" version = "2.12.0" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "h11", marker = "python_full_version < '3.12' or sys_platform != 'emscripten'" }, - { name = "truststore", marker = "python_full_version < '3.12' or sys_platform != 'emscripten'" }, + { name = "h11" }, + { name = "truststore" }, ] sdist = { url = "https://files.pythonhosted.org/packages/be/ad/f4f0e57345f1870f3e8cb624e058d7eca6e5a27d33bcc3311d9b618734cd/httpcore2-2.12.0.tar.gz", hash = "sha256:9293522bba0aa7c4c8e9e3f040c16575bd8868e155a77fa30c7a9085a5eae648", size = 67548, upload-time = "2026-08-18T13:22:08.211Z" } wheels = [ @@ -1198,6 +1200,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/c8/cb/6a6a47d5b464bd08695d254f3da6e7986cc70c9fa5d778eda57538edfe56/starlette-1.6.0-py3-none-any.whl", hash = "sha256:a86dd39d14bb45f85a3d18525215a9ef0cfd1f192ac793220e72598c90335f0c", size = 75969, upload-time = "2026-08-08T18:27:56.196Z" }, ] +[[package]] +name = "structlog" +version = "26.1.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/5e/89/b4a0bcfdf4f71a3dea31379f095929613d7e4528a0996bca6aa964cd0dca/structlog-26.1.0.tar.gz", hash = "sha256:f63a716cbd1b1291cf7661de7794b455acfa4c43c5bcf1630e6ad5ddc1adb3b7", size = 1459881, upload-time = "2026-06-06T07:33:39.348Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/a9/18/489c97b834dfff9cf2fc2507cede4bcd4b11e67f84bc462acd1992496f86/structlog-26.1.0-py3-none-any.whl", hash = "sha256:e081a26d6c373e6d201eca24eede26d8ffab07f88f477822e679183428d3d91e", size = 73764, upload-time = "2026-06-06T07:33:38.046Z" }, +] + [[package]] name = "truststore" version = "0.10.4" @@ -1248,8 +1259,8 @@ name = "uvicorn" version = "0.53.0" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "click", marker = "python_full_version < '3.12' or sys_platform != 'emscripten'" }, - { name = "h11", marker = "python_full_version < '3.12' or sys_platform != 'emscripten'" }, + { name = "click" }, + { name = "h11" }, ] sdist = { url = "https://files.pythonhosted.org/packages/5d/ad/04bbb797c84fc1f26cb171f7394716f4865ffb8d8c5e1eef42565c2dfa6b/uvicorn-0.53.0.tar.gz", hash = "sha256:a9356f0cb89b3b8621529c5d5eebd69bfe154f4c3f68b4cf2de47e45fa855c2e", size = 110881, upload-time = "2026-09-14T07:44:23.815Z" } wheels = [ From b73859e9fab97bfd430bf5829d012b9f96b473db Mon Sep 17 00:00:00 2001 From: Lucas Koontz Date: Wed, 21 Oct 2026 01:13:16 -0700 Subject: [PATCH 2/2] fix(logging): apply safe diagnostics to resident startup Refs: ENG-3579 --- README.md | 1 + anton/cloud_turn/resident.py | 9 ++++-- docs/structured-logging.md | 4 +++ tests/test_diagnostics.py | 53 ++++++++++++++++++++++++++++++++++++ 4 files changed, 64 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index c8c411755..e165399f4 100644 --- a/README.md +++ b/README.md @@ -375,5 +375,6 @@ If you genuinely need to release outside the normal flow (e.g. an admin hotfix), ## 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. diff --git a/anton/cloud_turn/resident.py b/anton/cloud_turn/resident.py index db3c23e91..34ca42f51 100644 --- a/anton/cloud_turn/resident.py +++ b/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) diff --git a/docs/structured-logging.md b/docs/structured-logging.md index 070b525e6..1d3ba967d 100644 --- a/docs/structured-logging.md +++ b/docs/structured-logging.md @@ -5,6 +5,10 @@ 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 diff --git a/tests/test_diagnostics.py b/tests/test_diagnostics.py index b5ab7a578..1c65a6651 100644 --- a/tests/test_diagnostics.py +++ b/tests/test_diagnostics.py @@ -66,3 +66,56 @@ async def operation(value): assert await asyncio.gather(operation("one"), operation("two")) == ["one", "two"] assert "conversation_id" not in get_contextvars() + + +@pytest.mark.parametrize("log_format", ["json", "text"]) +@pytest.mark.parametrize("level", [None, "DEBUG"]) +def test_resident_entrypoint_keeps_setup_and_fork_failures_safe(monkeypatch, capsys, log_format, level): + from types import SimpleNamespace + + from anton.cloud_turn import resident + + monkeypatch.setenv("LOG_FORMAT", log_format) + if level is None: + monkeypatch.delenv("LOG_LEVEL", raising=False) + else: + monkeypatch.setenv("LOG_LEVEL", level) + monkeypatch.setattr(resident.logger, "handlers", []) + monkeypatch.setattr(resident.logger, "level", resident.logger.level) + monkeypatch.setattr(resident.logger, "propagate", resident.logger.propagate) + monkeypatch.setattr(resident, "forbid_same_user_access", lambda: None) + root = logging.getLogger() + original_root = (root.handlers[:], root.level) + + def fail_preload(): + raise RuntimeError("private-preload-value") + + def fail_fork(): + raise OSError("private-fork-value") + + def serve(entry): + assert resident._prepare(fail_preload) is False + closed = [] + worker = object.__new__(resident._Resident) + worker._fork(SimpleNamespace(close=lambda: closed.append(True)), []) + assert closed == [True] + resident.logger.info("Resident progress") + return 0 + + monkeypatch.setattr(resident.os, "fork", fail_fork) + monkeypatch.setattr(resident, "serve", serve) + assert resident.main() == 0 + assert (root.handlers, root.level) == original_root + output = capsys.readouterr() + assert output.out == "" + assert "private-" not in output.err + assert ("Resident progress" in output.err) is (level == "DEBUG") + assert "RuntimeError" in output.err and "OSError" in output.err + if log_format == "json": + events = [json.loads(line) for line in output.err.splitlines()] + failures = [event for event in events if event["log_level"] == "ERROR"] + assert [event["error_type"] for event in failures] == ["RuntimeError", "OSError"] + assert all(event["logger"] == resident.__name__ for event in events) + assert all(event["event"] == "operation_failed" for event in failures) + assert "fail_preload" in failures[0]["exception"] + assert "fail_fork" in failures[1]["exception"]