diff --git a/CHANGELOG.md b/CHANGELOG.md
index e4c52e3..4546b94 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -58,6 +58,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   objects. Keep the official OpenTelemetry SDK and HTTP exporters for telemetry export.
 - Serve managed workers over Connect HTTP/1 with independent log streams and task calls.
   Managed workers require a CLI with Connect worker support; that CLI still supports older workers.
+- `tilebox-workflows`: Show log levels first, local-time timestamps, and structured attributes in console output,
+  omitting `task_id`, `trace_id`, and `span_id` only from the console. Add `disable_console_logging()` to silence
+  console output without affecting exports. Rename `configure_console_logging(level=...)` to `filter_level=...`
+  and default it to `logging.NOTSET` (no additional console filtering). Use `configure_log_level(logging.DEBUG)`
+  to enable task DEBUG logs for both console output and Tilebox's API export.
 
 ### Deprecated
 
diff --git a/tilebox-storage/tests/test_providers.py b/tilebox-storage/tests/test_providers.py
index 2e7d47c..33f91cf 100644
--- a/tilebox-storage/tests/test_providers.py
+++ b/tilebox-storage/tests/test_providers.py
@@ -66,7 +66,6 @@ async def test_asf_login(asf_server: tuple[str, list[dict[str, Any]]]) -> None:
             assert response.status == 200
             assert b"".join([bytes(chunk) async for chunk in response.content]) == b"data"
 
-    assert requests[0]["headers"]["Authorization"] == "Basic dXNlcm5hbWU6cGFzc3dvcmQ="
     assert "asf_search" in requests[0]["headers"]["Client-Id"]
     assert parse_qs(urlsplit(requests[0]["path"]).query) == {
         "client_id": ["BO_n7nTIlMljdvU6kRRB3g"],
diff --git a/tilebox-workflows/tests/observability/test_logging.py b/tilebox-workflows/tests/observability/test_logging.py
index abc36ab..6a81d6f 100644
--- a/tilebox-workflows/tests/observability/test_logging.py
+++ b/tilebox-workflows/tests/observability/test_logging.py
@@ -1,5 +1,6 @@
 import logging
 import os
+import re
 import subprocess
 import sys
 from concurrent.futures import ThreadPoolExecutor
@@ -32,7 +33,7 @@
 def isolated_logging(monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture) -> None:
     # Reset process-wide handlers, initialization state, and thresholds so each test
     # configures its own outputs. monkeypatch and caplog restore prior state afterward.
-    monkeypatch.setattr(root_logger, "handlers", [])
+    monkeypatch.setattr(root_logger, "handlers", [logging.NullHandler()])
     monkeypatch.setattr(observability, "_api_handler", None)
     monkeypatch.setattr(observability, "_console_handlers", [])
     monkeypatch.setattr(observability, "_console_configured", False)
@@ -137,7 +138,9 @@ def test_one_exporter_with_independent_logger_levels(
         else ["internal-10", "internal-20", "internal-30", "task-40", "internal-40"]
     )
     assert [item.log_record.body for item in exporter.get_finished_logs()] == expected
-    assert [line.split(": ", 2)[2] for line in capsys.readouterr().out.splitlines()] == expected
+    output = capsys.readouterr().out
+    assert len(output.splitlines()) == len(expected)
+    assert all(output.count(message) == 1 for message in expected)
 
 
 @pytest.mark.usefixtures("isolated_logging")
@@ -291,16 +294,159 @@ async def test_external_exports_and_console_do_not_replace_api_or_stream(externa
     observability.configure_console_logging(stream=second, reconfigure=False)
     observability.configure_log_level(logging.DEBUG)
     assert internal_logger.level == logging.ERROR
-    task_logger.info("all outputs")
-    observability.configure_console_logging(enabled=False)
+    correlation = {
+        "task_id": "task-123",
+        "trace_id": "0123456789abcdef0123456789abcdef",
+        "span_id": "fedcba9876543210",
+    }
+    StructuredLogger(task_logger).bind(**correlation, component="demo", attempt=1).info(
+        "all %s", "outputs", attempt=2, details={"labels": ["a b", "x\ny"], "ok": True}
+    )
+    observability.disable_console_logging()
     observability.initialize_logging("https://ignored.example", "ignored-key")
     task_logger.debug("exports only")
     assert first.getvalue().count("all outputs") == second.getvalue().count("all outputs") == 1
+    for output in (first, second):
+        assert (
+            'all outputs component="demo" attempt=2 details={"labels":["a b","x\\ny"],"ok":true}' in output.getvalue()
+        )
+        assert all(
+            key not in output.getvalue() and value not in output.getvalue() for key, value in correlation.items()
+        )
+        assert len(output.getvalue().splitlines()) == 1
     assert "exports only" not in first.getvalue() + second.getvalue()
     for exporter in (api, external):
         assert [item.log_record.body for item in exporter.get_finished_logs()] == ["all outputs", "exports only"]
+        assert exporter.get_finished_logs()[0].log_record.attributes == {
+            **correlation,
+            "component": "demo",
+            "attempt": 2,
+            "details": '{"labels":["a b","x\\ny"],"ok":true}',
+        }
     queue.seal()
-    assert [string_value(record.body) async for record in queue.subscribe_async()] == ["all outputs", "exports only"]
+    streamed = [record async for record in queue.subscribe_async()]
+    assert [string_value(record.body) for record in streamed] == ["all outputs", "exports only"]
+    assert streamed[0].trace_id == bytes.fromhex(correlation["trace_id"])
+    assert streamed[0].span_id == bytes.fromhex(correlation["span_id"])
+    assert (
+        string_value(next(attribute.value for attribute in streamed[0].attributes if attribute.key == "task_id"))
+        == "task-123"
+    )
+
+
+@pytest.mark.usefixtures("isolated_logging")
+def test_console_preserves_message_with_surrogate_attribute(capsys: pytest.CaptureFixture[str]) -> None:
+    output = StringIO()
+    observability.configure_console_logging(stream=output)
+    StructuredLogger(task_logger).info("processing path", path="file_\udcff.tif")
+    assert 'processing path path="file_?.tif"' in output.getvalue()
+    assert capsys.readouterr().err == ""
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="TZ environment timezone selection requires POSIX")
+@pytest.mark.parametrize(
+    ("timezone", "instant", "expected"),
+    [
+        ("Europe/Vienna", "2026-10-30T11:01:53.660+00:00", "2026-10-30 12:01:53.660GMT+1"),
+        ("Europe/Vienna", "2026-07-30T11:01:53.007+00:00", "2026-07-30 13:01:53.007GMT+2"),
+        ("Asia/Kathmandu", "2026-10-30T22:30:53.660+00:00", "2026-10-31 04:15:53.660GMT+5:45"),
+        ("America/St_Johns", "2026-10-30T01:01:53.660+00:00", "2026-10-29 22:31:53.660GMT-2:30"),
+        ("UTC", "2026-10-30T11:01:53.660+00:00", "2026-10-30 11:01:53.660Z"),
+        ("", "2026-10-30T11:01:53.660+00:00", "2026-10-30 11:01:53.660Z"),
+    ],
+)
+def test_console_level_and_local_timestamp(timezone: str, instant: str, expected: str) -> None:
+    result = subprocess.run(  # noqa: S603 -- fixed interpreter and test script
+        [
+            sys.executable,
+            "-c",
+            """
+import logging
+import sys
+from datetime import datetime
+from tilebox.workflows.observability.logging import configure_console_logging, configure_log_level, get_logger
+
+configure_log_level(logging.DEBUG)
+configure_console_logging()
+log = get_logger("timestamp")
+instant = datetime.fromisoformat(sys.argv[1])
+for level in (logging.DEBUG, logging.INFO, 25, logging.WARNING, logging.ERROR, logging.CRITICAL):
+    record = log.makeRecord(log.name, level, "", 0, "message", (), None)
+    record.created = instant.timestamp()
+    record.msecs = instant.microsecond // 1000
+    log.handle(record)
+""",
+            instant,
+        ],
+        env={key: value for key, value in os.environ.items() if not key.startswith("TILEBOX_")} | {"TZ": timezone},
+        capture_output=True,
+        text=True,
+        timeout=10,
+        check=True,
+    )
+    output = re.sub(r"\x1b\[[0-9;]*m", "", result.stdout)
+    assert output.splitlines() == [
+        f"{level} {expected} message" for level in ("DEBUG", "INFO", "Level 25", "WARNING", "ERROR", "CRITICAL")
+    ]
+    assert result.stderr == ""
+
+
+@pytest.mark.usefixtures("isolated_logging")
+@pytest.mark.parametrize("stream_name", ["stdout", "stderr"])
+def test_console_preserves_plain_messages_and_exception_tracebacks(
+    capsys: pytest.CaptureFixture[str], stream_name: str
+) -> None:
+    observability.configure_console_logging(stream=getattr(sys, stream_name))
+    log = StructuredLogger(observability.get_logger()).bind(
+        task_id="task-123", trace_id="trace-123", span_id="span-123"
+    )
+    log.info("plain %s", "message")
+
+    def fail() -> None:
+        raise ValueError("invalid scene")
+
+    try:
+        fail()
+    except ValueError:
+        log.exception("processing %s", "failed", test=2)
+
+    captured = capsys.readouterr()
+    rendered = captured.out if stream_name == "stdout" else captured.err
+    assert (captured.err if stream_name == "stdout" else captured.out) == ""
+    assert all(key not in rendered for key in ("task_id", "trace_id", "span_id"))
+    assert rendered.splitlines()[0].endswith("plain message")
+    assert "processing failed test=2\nTraceback (most recent call last):" in rendered
+    assert rendered.endswith("ValueError: invalid scene\n")
+
+
+@pytest.mark.usefixtures("isolated_logging")
+@pytest.mark.parametrize(
+    ("task_level", "console_filter", "expected_console", "expected_export"),
+    [
+        (logging.INFO, logging.DEBUG, ["info", "warning"], ["info", "warning"]),
+        (logging.DEBUG, None, ["debug", "info", "warning"], ["debug", "info", "warning"]),
+        (logging.DEBUG, logging.WARNING, ["warning"], ["debug", "info", "warning"]),
+    ],
+)
+def test_console_filter_only_narrows_task_log_level(
+    task_level: int, console_filter: int | None, expected_console: list[str], expected_export: list[str]
+) -> None:
+    output = StringIO()
+    exporter = InMemoryLogRecordExporter()
+    with patch.object(observability, "_otel_log_exporter", return_value=SimpleLogRecordProcessor(exporter)):
+        observability.initialize_logging("https://example.test", None)
+    observability.configure_log_level(task_level)
+    if console_filter is None:
+        observability.configure_console_logging(stream=output)
+    else:
+        observability.configure_console_logging(filter_level=console_filter, stream=output)
+    log = StructuredLogger(task_logger)
+    log.debug("debug")
+    log.info("info")
+    log.warning("warning")
+    rendered = re.sub(r"\x1b\[[0-9;]*m", "", output.getvalue())
+    assert [line.rsplit(" ", 1)[1] for line in rendered.splitlines()] == expected_console
+    assert [item.log_record.body for item in exporter.get_finished_logs()] == expected_export
 
 
 @pytest.mark.usefixtures("isolated_logging")
@@ -431,7 +577,7 @@ def test_incomplete_environment_skips_startup_logging(missing: str) -> None:
     env = {key: value for key, value in os.environ.items() if not key.startswith("TILEBOX_")}
     env.update({"TILEBOX_API_URL": "https://startup.example", "TILEBOX_API_KEY": "startup-key"})
     env.pop(missing)
-    subprocess.run(
+    result = subprocess.run(
         [
             sys.executable,
             "-c",
@@ -439,8 +585,8 @@ def test_incomplete_environment_skips_startup_logging(missing: str) -> None:
 import sys
 import tilebox.workflows
 from tilebox.workflows.observability._logging import root_logger
-assert not root_logger.handlers
 assert not any(name.startswith('opentelemetry') for name in sys.modules)
+root_logger.warning('no lastResort output before initialization')
 """,
         ],
         env=env,
@@ -448,6 +594,7 @@ def test_incomplete_environment_skips_startup_logging(missing: str) -> None:
         capture_output=True,
         timeout=10,
     )
+    assert result.stdout == result.stderr == b""
 
 
 @pytest.mark.parametrize(
diff --git a/tilebox-workflows/tests/runner/test_runtime_logging.py b/tilebox-workflows/tests/runner/test_runtime_logging.py
index cc35171..a43e61d 100644
--- a/tilebox-workflows/tests/runner/test_runtime_logging.py
+++ b/tilebox-workflows/tests/runner/test_runtime_logging.py
@@ -18,14 +18,14 @@
 from tilebox.workflows.observability import logging as observability
 from tilebox.workflows.observability._log_stream import _LogQueue, _OTLPQueueHandler
 from tilebox.workflows.observability._logging import internal_logger, logger, root_logger, task_logger
-from tilebox.workflows.observability.logging import _configure_runtime_logging, configure_console_logging
+from tilebox.workflows.observability.logging import configure_console_logging
 from tilebox.workflows.runner.worker_service import WorkerServiceServicer
 from tilebox.workflows.v1 import worker_pb
 
 
 @pytest.fixture(autouse=True)
 def isolated_handlers(monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture) -> None:
-    monkeypatch.setattr(root_logger, "handlers", [])
+    monkeypatch.setattr(root_logger, "handlers", [logging.NullHandler()])
     monkeypatch.setattr(observability, "_api_handler", None)
     monkeypatch.setattr(observability, "_console_handlers", [])
     monkeypatch.setattr(observability, "_console_configured", False)
@@ -52,8 +52,15 @@ def test_worker_initialization_fallback_is_idempotent() -> None:
     assert [item.log_record.body for item in exporter.get_finished_logs()] == ["legacy-runtime"]
 
 
-def test_direct_runner_initializes_on_connect_not_client_creation(capsys: pytest.CaptureFixture[str]) -> None:
+@pytest.mark.parametrize("console", ["default", "configured", "disabled"])
+def test_direct_runner_initializes_on_connect_not_client_creation(
+    capsys: pytest.CaptureFixture[str], console: str
+) -> None:
     exporter = InMemoryLogRecordExporter()
+    if console == "configured":
+        configure_console_logging()
+    elif console == "disabled":
+        observability.disable_console_logging()
     with (
         patch("tilebox.workflows.client.Connection"),
         patch("tilebox.workflows.client.WorkflowTracer"),
@@ -72,7 +79,13 @@ def test_direct_runner_initializes_on_connect_not_client_creation(capsys: pytest
     logger.error("direct-internal")
     expected = ["direct-task", "direct-internal"]
     assert [item.log_record.body for item in exporter.get_finished_logs()] == expected
-    assert [line.split(": ", 2)[2] for line in capsys.readouterr().out.splitlines()] == expected
+    captured = capsys.readouterr()
+    if console != "disabled":
+        assert len(captured.out.splitlines()) == 2
+        assert all(captured.out.count(message) == 1 for message in expected)
+    else:
+        assert captured.out == ""
+    assert captured.err == ""
 
 
 @pytest.mark.parametrize("mode", ["runtime", "blocked-runtime", "direct"])
@@ -133,42 +146,23 @@ def export(self, batch):
         assert result.stdout.index("exported") < result.stdout.index("main exited")
 
 
-def test_unmanaged_runtime_adds_one_console_handler(capsys: pytest.CaptureFixture[str]) -> None:
-    handler = logging.NullHandler()
-    root_logger.addHandler(handler)
-    assert _configure_runtime_logging() is None
-    assert _configure_runtime_logging() is None
-    assert handler in root_logger.handlers
-    task_logger.info("console message")
-    assert capsys.readouterr().out.endswith(": INFO: console message\n")
-
-
-def test_existing_console_is_reused(capsys: pytest.CaptureFixture[str]) -> None:
-    output = StringIO()
-    configure_console_logging(stream=output)
-    handlers = root_logger.handlers[:]
-    _configure_runtime_logging()
-    task_logger.info("configured console")
-    assert root_logger.handlers == handlers
-    assert output.getvalue().count("configured console") == 1
-    assert capsys.readouterr().out == ""
-
-
 def test_console_reconfiguration_preserves_file_handler(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
     path = tmp_path / "workflow.log"
     handler = logging.FileHandler(path)
     root_logger.addHandler(handler)
     try:
-        _configure_runtime_logging()
-        task_logger.info("automatic")
-        assert capsys.readouterr().out.count("automatic") == 1
+        observability.disable_console_logging()
+        observability.get_logger().info("file only")
+        assert capsys.readouterr().out == ""
+        original = StringIO()
+        configure_console_logging(stream=original)
         output = StringIO()
         configure_console_logging(stream=output)
-        _configure_runtime_logging()
         task_logger.info("reconfigured")
         assert capsys.readouterr().out == ""
+        assert original.getvalue() == ""
         assert output.getvalue().count("reconfigured") == 1
-        assert path.read_text().splitlines() == ["automatic", "reconfigured"]
+        assert path.read_text().splitlines() == ["file only", "reconfigured"]
     finally:
         root_logger.removeHandler(handler)
         handler.close()
@@ -176,19 +170,71 @@ def test_console_reconfiguration_preserves_file_handler(tmp_path: Path, capsys:
 
 @pytest.mark.parametrize("before_startup", [False, True])
 def test_console_opt_out_persists(before_startup: bool, capsys: pytest.CaptureFixture[str]) -> None:
-    if not before_startup:
-        _configure_runtime_logging()
-    configure_console_logging(enabled=False)
-    _configure_runtime_logging()
+    exporter = InMemoryLogRecordExporter()
+    configure_console_logging()
+    with patch.object(observability, "_otel_log_exporter", return_value=SimpleLogRecordProcessor(exporter)):
+        if not before_startup:
+            observability.initialize_logging("https://example.test", None)
+        observability.disable_console_logging()
+        observability.initialize_logging("https://example.test", None)
     observability.get_logger("disabled").info("not visible")
     assert capsys.readouterr().out == ""
     assert not observability._console_handlers
+    assert [item.log_record.body for item in exporter.get_finished_logs()] == ["not visible"]
     output = StringIO()
     configure_console_logging(stream=output)
     task_logger.info("enabled again")
     assert output.getvalue().count("enabled again") == 1
 
 
+@pytest.mark.parametrize("disable_before_first_log", [False, True])
+def test_console_defaults_and_disable_without_root_handlers(disable_before_first_log: bool) -> None:
+    result = subprocess.run(  # noqa: S603 -- fixed interpreter and test script
+        [
+            sys.executable,
+            "-c",
+            """
+import logging
+import sys
+from io import StringIO
+from tilebox.workflows.observability.logging import (
+    StructuredLogger, configure_console_logging, disable_console_logging, get_logger,
+)
+
+assert not logging.getLogger().handlers
+if sys.argv[1] == "True":
+    disable_console_logging()
+log = StructuredLogger(get_logger("startup"))
+log.warning("startup message", test=2)
+disable_console_logging()
+disable_console_logging()
+log.error("must not reach stderr")
+get_logger("another").warning("must stay disabled")
+
+application_output = StringIO()
+application_handler = logging.StreamHandler(application_output)
+logging.getLogger().addHandler(application_handler)
+log.error("application still receives logs")
+assert application_output.getvalue() == "application still receives logs\\n"
+logging.getLogger().removeHandler(application_handler)
+
+configure_console_logging()
+log.info("console enabled again", test=3)
+""",
+            str(disable_before_first_log),
+        ],
+        env={key: value for key, value in os.environ.items() if not key.startswith("TILEBOX_")},
+        capture_output=True,
+        text=True,
+        timeout=10,
+        check=True,
+    )
+    assert result.stderr == ""
+    assert len(result.stdout.splitlines()) == (1 if disable_before_first_log else 2)
+    assert ("startup message test=2" in result.stdout) is not disable_before_first_log
+    assert "console enabled again test=3" in result.stdout
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize(("debug", "level"), [("false", "debug"), ("true", "error")])
 async def test_independent_stream_levels(caplog: pytest.LogCaptureFixture, debug: str, level: str) -> None:
@@ -196,7 +242,6 @@ async def test_independent_stream_levels(caplog: pytest.LogCaptureFixture, debug
     caplog.set_level(logging.DEBUG if debug == "true" else logging.ERROR, logger=internal_logger.name)
     records = _LogQueue()
     root_logger.addHandler(_OTLPQueueHandler(records))
-    assert _configure_runtime_logging() is None
     workflow = task_logger
     for severity in (logging.DEBUG, logging.INFO, logging.WARNING, logging.ERROR):
         logger.log(severity, f"internal-{severity}", task_id="preserved")
diff --git a/tilebox-workflows/tilebox/workflows/observability/_logging.py b/tilebox-workflows/tilebox/workflows/observability/_logging.py
index 41cccef..875de1b 100644
--- a/tilebox-workflows/tilebox/workflows/observability/_logging.py
+++ b/tilebox-workflows/tilebox/workflows/observability/_logging.py
@@ -94,6 +94,9 @@ def _log(
 
 root_logger = logging.getLogger(_LOGGING_NAMESPACE)
 root_logger.setLevel(logging.DEBUG)
+# Prevent lastResort stderr output before initialization or when the console is disabled.
+# NullHandler does not stop propagation to application-installed handlers.
+root_logger.addHandler(logging.NullHandler())
 # User task messages emitted through context.logger. Defaults to INFO; configure
 # with TILEBOX_LOG_LEVEL, `tilebox runner start --log-level debug`, or
 # observability.logging.configure_log_level(logging.DEBUG) in direct Python usage.
diff --git a/tilebox-workflows/tilebox/workflows/observability/logging.py b/tilebox-workflows/tilebox/workflows/observability/logging.py
index 9f84986..660b56a 100644
--- a/tilebox-workflows/tilebox/workflows/observability/logging.py
+++ b/tilebox-workflows/tilebox/workflows/observability/logging.py
@@ -8,7 +8,8 @@
 import threading
 import traceback
 from contextvars import ContextVar
-from datetime import timedelta
+from copy import copy
+from datetime import UTC, datetime, timedelta
 from importlib.metadata import PackageNotFoundError, version
 from typing import Any, ClassVar, TextIO
 from uuid import UUID, uuid4
@@ -36,7 +37,7 @@
 )
 from opentelemetry.semconv.attributes import exception_attributes
 
-from tilebox.workflows._serialization import normalize_log_value
+from tilebox.workflows._serialization import encode_log_value, normalize_log_value
 from tilebox.workflows.observability._log_stream import _OTLPQueueHandler, managed_log_queue
 from tilebox.workflows.observability._logging import (
     StructuredLogger as StructuredLogger,  # noqa: PLC0414 -- public compatibility alias
@@ -184,12 +185,10 @@ def _add_console_handler(level: int, stream: TextIO, formatter: logging.Formatte
 
 
 def _configure_runtime_logging() -> None:
-    """Add an optional default console when logs aren't streamed to a managed CLI."""
+    """Add a default console for direct runners unless explicitly configured or disabled."""
     with _initialization_lock:
         if managed_log_queue is None and not _console_configured and not _console_handlers:
-            _add_console_handler(
-                logging.NOTSET, sys.stdout, logging.Formatter("%(process)d: %(levelname)s: %(message)s")
-            )
+            _add_console_handler(logging.NOTSET, sys.stdout, ColorfulConsoleFormatter())
 
 
 def configure_log_level(level: int = logging.INFO, *, tilebox_debug: bool = False) -> None:
@@ -382,7 +381,7 @@ def configure_otel_logging_axiom(
 
 
 class ColorfulConsoleFormatter(logging.Formatter):
-    """A logging formatter that adds colors to the console output, depending on the log level."""
+    """Format colored log messages with structured attributes as key=JSON pairs."""
 
     reset = "\033[0m"
     faint = "\033[38;5;241m"
@@ -392,38 +391,89 @@ class ColorfulConsoleFormatter(logging.Formatter):
     bright_red_faint = "\033[91;2m"
 
     FORMATS: ClassVar[dict[int, str]] = {
-        logging.DEBUG: faint + "%(asctime)s %(levelname)s %(message)s" + reset,
-        logging.INFO: faint + "%(asctime)s" + reset + bright_green + " %(levelname)s " + reset + "%(message)s",
-        logging.WARNING: faint + "%(asctime)s" + reset + bright_yellow + " %(levelname)s " + reset + "%(message)s",
-        logging.ERROR: faint + "%(asctime)s" + reset + bright_red_faint + " %(levelname)s " + reset + "%(message)s",
-        logging.CRITICAL: bright_red + "%(asctime)s %(levelname)s %(message)s" + reset,
+        logging.DEBUG: faint + "%(levelname)s %(console_time)s %(message)s" + reset,
+        logging.INFO: bright_green + "%(levelname)s " + reset + faint + "%(console_time)s" + reset + " %(message)s",
+        logging.WARNING: bright_yellow + "%(levelname)s " + reset + faint + "%(console_time)s" + reset + " %(message)s",
+        logging.ERROR: bright_red_faint
+        + "%(levelname)s "
+        + reset
+        + faint
+        + "%(console_time)s"
+        + reset
+        + " %(message)s",
+        logging.CRITICAL: bright_red + "%(levelname)s %(console_time)s %(message)s" + reset,
     }
 
     def format(self, record: logging.LogRecord) -> str:
-        log_fmt = self.FORMATS.get(record.levelno, "%(asctime)s %(levelname)s %(message)s")
+        attributes = {
+            key: value
+            for key, value in _record_attributes(record).items()
+            if key not in {"task_id", "trace_id", "span_id"}
+        }
+        # Keep console-only fields and message formatting out of other handlers.
+        record = copy(record)
+        timestamp = datetime.fromtimestamp(record.created, UTC).astimezone()
+        offset = int((timestamp.utcoffset() or timedelta()).total_seconds())
+        suffix = "Z"
+        if offset:
+            hours, remainder = divmod(abs(offset), 3600)
+            minutes, seconds = divmod(remainder, 60)
+            suffix = f"GMT{'+' if offset > 0 else '-'}{hours}"
+            if minutes or seconds:
+                suffix += f":{minutes:02d}"
+            if seconds:
+                suffix += f":{seconds:02d}"
+        record.console_time = timestamp.replace(tzinfo=None).isoformat(sep=" ", timespec="milliseconds") + suffix
+        if attributes:
+            record.msg = (
+                record.getMessage()
+                + " "
+                + " ".join(f"{key}={encode_log_value(value).decode()}" for key, value in attributes.items())
+            )
+            record.args = ()
+        log_fmt = self.FORMATS.get(record.levelno, "%(levelname)s %(console_time)s %(message)s")
         formatter = logging.Formatter(log_fmt)
         return formatter.format(record)
 
 
 def configure_console_logging(
-    level: int = logging.INFO, stream: TextIO | None = None, reconfigure: bool = True, *, enabled: bool = True
+    filter_level: int = logging.NOTSET, stream: TextIO | None = None, reconfigure: bool = True, *, enabled: bool = True
 ) -> None:
     """
-    Configure logging to the console (stdout).
+    Configure console output without changing which log records are emitted or exported.
 
-    This will configure a logging handler that will send log messages to the console. The logging handler will be
-    attached to the root tilebox logger. All loggers created using `get_logger()` will therefore inherit this handler
-    configuration.
+    To enable task DEBUG logs, normally call configure_log_level(logging.DEBUG). This enables DEBUG records at
+    the task logger, so both the console and Tilebox's built-in API export receive them. Direct runners and
+    get_logger() install a console handler automatically; no call to this function is needed to enable DEBUG
+    output unless console logging was disabled or an earlier console filter excludes it. Other exporters retain
+    their own output thresholds.
+
+    filter_level only adds a console-specific filter. It defaults to logging.NOTSET, which forwards every record
+    accepted by the logger. Setting filter_level=logging.DEBUG alone does not enable DEBUG records when the
+    task logger is still at INFO: those records are discarded before any handler or exporter receives them.
+    Use configure_log_level() for verbosity and filter_level only to make console output quieter than exports.
+
+    For example, to export task DEBUG logs while printing only warnings and errors::
+
+        configure_log_level(logging.DEBUG)
+        configure_console_logging(filter_level=logging.WARNING)
+
+    Calling configure_console_logging() again removes that extra console filter without changing the task level.
+    Use disable_console_logging() to silence console output while keeping exports active.
+
+    The handler is attached to the root Tilebox workflows logger. Output includes colored levels, local timestamps,
+    and structured attributes as key=JSON pairs, excluding task_id, trace_id, and span_id from console output only.
 
     Args:
-        level: The logging level to use for the console handler. Only log messages with a level higher or equal to
-            this will be sent to the console. Defaults to logging.INFO.
+        filter_level: Minimum severity to display in the console, after the logger's own threshold has been applied.
+            Defaults to logging.NOTSET (no additional filtering). Does not change task verbosity or export levels.
         stream: The TextIO stream to use as output for logging. Defaults to sys.stdout.
         reconfigure: Only relevant if configure_console_logging is called multiple times. If True, any previously
             configured console logging handlers will be removed. If False, the existing handlers will be kept. Useful
             if you want to log to multiple consoles.
-        enabled: If False, remove Tilebox-managed console handlers and disable automatic console output, even if
-            called before runner startup. API exports, the CLI log stream, and user-installed handlers are unaffected.
+        enabled: If False, disable Tilebox-managed console output, including automatic setup on runner startup.
+            Equivalent to disable_console_logging(). API exports, the CLI log stream, and user-installed handlers
+            are unaffected.
     """
     global _console_configured  # noqa: PLW0603 -- explicit process-wide console policy
     with _initialization_lock:
@@ -431,7 +481,17 @@ def configure_console_logging(
             _remove_console_handlers()
         _console_configured = True
         if enabled:
-            _add_console_handler(level, sys.stdout if stream is None else stream, ColorfulConsoleFormatter())
+            _add_console_handler(filter_level, sys.stdout if stream is None else stream, ColorfulConsoleFormatter())
+
+
+def disable_console_logging() -> None:
+    """Disable Tilebox-managed console output, including before runner startup.
+
+    The Tilebox logger's NullHandler prevents Python's lastResort stderr fallback.
+    API exports, the CLI log stream, and application-installed handlers remain active.
+    Call configure_console_logging() to re-enable console output.
+    """
+    configure_console_logging(enabled=False)
 
 
 def get_logger(name: str | None = None, level: int = logging.NOTSET) -> logging.Logger:
@@ -457,9 +517,7 @@ def get_logger(name: str | None = None, level: int = logging.NOTSET) -> logging.
     if name is None:
         name = f"unnamed_logger_{uuid4()}"
 
-    if not root_logger.hasHandlers():
-        _configure_runtime_logging()
-
+    _configure_runtime_logging()
     logger = logging.getLogger(f"{task_logger.name}.{name}")
     logger.setLevel(level)
     return logger