diff --git a/nitrostack/__init__.py b/nitrostack/__init__.py
index 9f1c3d5..b48802e 100644
--- a/nitrostack/__init__.py
+++ b/nitrostack/__init__.py
@@ -19,6 +19,7 @@
     AuthContext,
     TaskContext,
 )
+from nitrostack.core.log_emitter import LogRecord, LogEmitter, LogEmitterHandler, log_emitter
 from nitrostack.core.task import (
     TaskManager,
     TaskData,
@@ -206,6 +207,10 @@
     "ConfigService",
     "NitroTestingModule",
     "TaskContext",
+    "LogRecord",
+    "LogEmitter",
+    "LogEmitterHandler",
+    "log_emitter",
     "TaskCancelledError",
     "TaskManager",
     "TaskData",
diff --git a/nitrostack/core/context.py b/nitrostack/core/context.py
index 42cf129..ac76790 100644
--- a/nitrostack/core/context.py
+++ b/nitrostack/core/context.py
@@ -34,8 +34,13 @@ def __init__(self, log_file: Optional[str] = None, name: str = "nitrostack"):
         level = getattr(logging, level_str, logging.DEBUG)
         self.logger.setLevel(level)
         
-        # Avoid adding duplicate handlers if initialized multiple times
-        if not self.logger.handlers:
+        # The log-emitter bridge is not an output handler; keep configuring
+        # the file/stdout destination even though that bridge is installed.
+        has_output_handler = any(
+            not getattr(handler, "_nitrostack_log_emitter", False)
+            for handler in self.logger.handlers
+        )
+        if not has_output_handler:
             formatter = logging.Formatter(
                 '%(asctime)s [%(levelname)s] (%(name)s): %(message)s'
             )
diff --git a/nitrostack/core/log_emitter.py b/nitrostack/core/log_emitter.py
new file mode 100644
index 0000000..bee944a
--- /dev/null
+++ b/nitrostack/core/log_emitter.py
@@ -0,0 +1,193 @@
+"""Process-wide bridge from Python logging to tooling subscribers."""
+from __future__ import annotations
+
+import logging
+import queue
+import threading
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from typing import Any, Callable
+
+
+DEFAULT_BUFFER_SIZE = 256
+_STANDARD_RECORD_FIELDS = frozenset(
+    logging.makeLogRecord({}).__dict__.keys()
+) | {"message", "asctime"}
+
+
+@dataclass(frozen=True)
+class LogRecord:
+    """A structured logging record delivered to tooling subscribers."""
+
+    level: str
+    levelno: int
+    message: str
+    timestamp: str
+    logger_name: str
+    extra: dict[str, Any]
+
+
+class _Subscriber:
+    def __init__(self, callback: Callable[[LogRecord], Any], max_buffer: int):
+        self.callback = callback
+        self.queue: queue.Queue[LogRecord | None] = queue.Queue(maxsize=max_buffer)
+        self.active = True
+        self.condition = threading.Condition()
+        self.callback_thread: int | None = None
+        self.thread = threading.Thread(target=self._deliver, daemon=True, name="nitrostack-log-subscriber")
+        self.thread.start()
+
+    def enqueue(self, record: LogRecord) -> None:
+        with self.condition:
+            if not self.active:
+                return
+            try:
+                self.queue.put_nowait(record)
+            except queue.Full:
+                # Preserve the newest logs when a subscriber cannot keep up.
+                try:
+                    self.queue.get_nowait()
+                except queue.Empty:
+                    pass
+                try:
+                    self.queue.put_nowait(record)
+                except queue.Full:
+                    pass
+
+    def unsubscribe(self) -> None:
+        with self.condition:
+            if not self.active:
+                return
+            self.active = False
+            while True:
+                try:
+                    self.queue.get_nowait()
+                except queue.Empty:
+                    break
+            self.queue.put_nowait(None)
+            if self.callback_thread != threading.get_ident():
+                while self.callback_thread is not None:
+                    self.condition.wait()
+
+    def _deliver(self) -> None:
+        while True:
+            record = self.queue.get()
+            if record is None:
+                return
+            with self.condition:
+                if not self.active:
+                    continue
+                self.callback_thread = threading.get_ident()
+            try:
+                self.callback(record)
+            except Exception:
+                # Subscriber code must not interfere with logging or peers.
+                continue
+            finally:
+                with self.condition:
+                    self.callback_thread = None
+                    self.condition.notify_all()
+
+
+class LogEmitter:
+    """Thread-safe subscription point for records from ``nitrostack`` loggers."""
+
+    def __init__(self, max_buffer: int = DEFAULT_BUFFER_SIZE):
+        if max_buffer < 1:
+            raise ValueError("max_buffer must be at least 1")
+        self._max_buffer = max_buffer
+        self._subscribers: set[_Subscriber] = set()
+        self._lock = threading.Lock()
+        self._installed_handlers: dict[str, logging.Handler] = {}
+        # Loggers to bridge while subscribers exist. A logger with
+        # ``propagate=False`` never reaches ``nitrostack``, so it needs its own.
+        self._bridged_loggers: set[str] = {"nitrostack"}
+
+    def subscribe(self, callback: Callable[[LogRecord], Any]) -> Callable[[], None]:
+        """Subscribe to records; return an idempotent function to stop delivery."""
+        subscriber = _Subscriber(callback, self._max_buffer)
+        with self._lock:
+            self._subscribers.add(subscriber)
+            # Only the process-wide emitter owns NitroStack logger wiring.
+            # Standalone emitters are connected explicitly with
+            # LogEmitterHandler, avoiding duplicate delivery from the shared
+            # ``nitrostack`` logger hierarchy.
+            if self is log_emitter:
+                self._install_handlers_locked()
+
+        def unsubscribe() -> None:
+            with self._lock:
+                self._subscribers.discard(subscriber)
+                if not self._subscribers:
+                    self._remove_handlers_locked()
+            subscriber.unsubscribe()
+
+        return unsubscribe
+
+    def _publish(self, record: LogRecord) -> None:
+        with self._lock:
+            subscribers = tuple(self._subscribers)
+        for subscriber in subscribers:
+            subscriber.enqueue(record)
+
+    def _install_handlers_locked(self) -> None:
+        for logger_name in self._bridged_loggers:
+            self._install_handler_locked(logger_name)
+
+    def _install_handler_locked(self, logger_name: str) -> None:
+        if logger_name in self._installed_handlers:
+            return
+        logger = logging.getLogger(logger_name)
+        handler = LogEmitterHandler(self)
+        handler._nitrostack_log_emitter = True  # type: ignore[attr-defined]
+        logger.addHandler(handler)
+        self._installed_handlers[logger_name] = handler
+
+    def _remove_handlers_locked(self) -> None:
+        for name, handler in self._installed_handlers.items():
+            logging.getLogger(name).removeHandler(handler)
+            handler.close()
+        self._installed_handlers.clear()
+
+
+class LogEmitterHandler(logging.Handler):
+    """Logging handler that publishes structured records to a ``LogEmitter``."""
+
+    def __init__(self, emitter: LogEmitter):
+        super().__init__(logging.NOTSET)
+        self.emitter = emitter
+
+    def emit(self, record: logging.LogRecord) -> None:
+        try:
+            timestamp = datetime.fromtimestamp(record.created, timezone.utc).isoformat()
+            extra = {
+                key: value
+                for key, value in record.__dict__.items()
+                if key not in _STANDARD_RECORD_FIELDS and not key.startswith("_")
+            }
+            self.emitter._publish(
+                LogRecord(
+                    level=record.levelname,
+                    levelno=record.levelno,
+                    message=record.getMessage(),
+                    timestamp=timestamp,
+                    logger_name=record.name,
+                    extra=extra,
+                )
+            )
+        except Exception:
+            self.handleError(record)
+
+
+log_emitter = LogEmitter()
+
+
+def install_log_handler(logger_name: str = "nitrostack") -> None:
+    """Bridge a logger to the global emitter whenever it has subscribers.
+
+    The name is remembered, so a later ``subscribe()`` still attaches to it.
+    """
+    with log_emitter._lock:
+        log_emitter._bridged_loggers.add(logger_name)
+        if log_emitter._subscribers:
+            log_emitter._install_handler_locked(logger_name)
diff --git a/nitrostack/transports/http.py b/nitrostack/transports/http.py
index c915005..c313258 100644
--- a/nitrostack/transports/http.py
+++ b/nitrostack/transports/http.py
@@ -63,6 +63,7 @@
 )
 from nitrostack.transports.subscriptions import subscriptions_listen_endpoint
 from nitrostack.transports.proxy import public_url_for_request
+from nitrostack.core.log_emitter import install_log_handler
 
 if TYPE_CHECKING:
     from nitrostack.core.app import McpApplication
@@ -130,6 +131,7 @@ def _enable_trace_logging() -> None:
     traces are not emitted twice once `FileLogger` does attach its handler.
     stderr keeps them off stdout, which dual mode uses for JSON-RPC.
     """
+    install_log_handler("nitrostack.transports.http")
     if any(getattr(handler, "_nitrostack_trace", False) for handler in logger.handlers):
         return
     handler = logging.StreamHandler(sys.stderr)
diff --git a/tests/test_log_emitter.py b/tests/test_log_emitter.py
new file mode 100644
index 0000000..b1474aa
--- /dev/null
+++ b/tests/test_log_emitter.py
@@ -0,0 +1,275 @@
+from __future__ import annotations
+
+import logging
+import io
+import threading
+
+from nitrostack.core.context import FileLogger
+from nitrostack.core.log_emitter import LogEmitter, LogEmitterHandler, log_emitter
+
+
+def _wait_for(event: threading.Event) -> None:
+    assert event.wait(2), "timed out waiting for log subscriber"
+
+
+def test_nitrostack_warning_reaches_last_resort_stderr_without_subscribers():
+    logger = logging.Logger("nitrostack.test.last-resort")
+    # Detach from pytest's logging capture handlers so Python's fallback is
+    # the only possible destination, as it is before any NitroStack handlers
+    # or subscriptions have been installed.
+    logger.parent = None
+    old_last_resort = logging.lastResort
+    fallback_stream = io.StringIO()
+    fallback = logging.StreamHandler(fallback_stream)
+    try:
+        fallback.setLevel(logging.WARNING)
+        logging.lastResort = fallback
+        logger.setLevel(logging.WARNING)
+        logger.propagate = True
+        logger.warning("warning before subscription")
+        assert not any(getattr(handler, "_nitrostack_log_emitter", False) for handler in logger.handlers)
+        assert "warning before subscription" in fallback_stream.getvalue()
+    finally:
+        logging.lastResort = old_last_resort
+        fallback.close()
+
+
+def test_global_emitter_receives_nitrostack_records_and_extra_fields():
+    received = []
+    done = threading.Event()
+
+    def collect(record):
+        received.append(record)
+        done.set()
+
+    unsubscribe = log_emitter.subscribe(collect)
+    try:
+        logging.getLogger("nitrostack.test.emitter").warning(
+            "hello %s", "tooling", extra={"request_id": "r-1"}
+        )
+        _wait_for(done)
+    finally:
+        unsubscribe()
+
+    record = received[0]
+    assert record.level == "WARNING"
+    assert record.message == "hello tooling"
+    assert record.logger_name == "nitrostack.test.emitter"
+    assert record.extra == {"request_id": "r-1"}
+    assert record.timestamp.endswith("+00:00")
+
+
+def test_unsubscribe_stops_future_delivery():
+    emitter = LogEmitter()
+    received = []
+    ready = threading.Event()
+    unsubscribe = emitter.subscribe(lambda record: (received.append(record), ready.set()))
+    logger = logging.getLogger("nitrostack.test.unsubscribe")
+    handler = LogEmitterHandler(emitter)
+    logger.addHandler(handler)
+    logger.setLevel(logging.INFO)
+    try:
+        logger.info("before")
+        _wait_for(ready)
+        unsubscribe()
+        logger.info("after")
+        assert len(received) == 1
+    finally:
+        logger.removeHandler(handler)
+
+
+def test_unsubscribe_waits_for_in_flight_callback():
+    emitter = LogEmitter()
+    entered = threading.Event()
+    release = threading.Event()
+    drained = threading.Event()
+    unsubscribe_returned = threading.Event()
+    received = []
+
+    def callback(record):
+        received.append(record)
+        entered.set()
+        release.wait(2)
+
+    unsubscribe = emitter.subscribe(callback)
+    logger = logging.getLogger("nitrostack.test.unsubscribe-in-flight")
+    handler = LogEmitterHandler(emitter)
+    logger.addHandler(handler)
+    logger.setLevel(logging.INFO)
+    worker = threading.Thread(target=lambda: logger.info("in flight"))
+    try:
+        worker.start()
+        _wait_for(entered)
+
+        def stop_subscription():
+            unsubscribe()
+            unsubscribe_returned.set()
+
+        stopper = threading.Thread(target=stop_subscription)
+        stopper.start()
+        assert not unsubscribe_returned.wait(0.05)
+        release.set()
+        _wait_for(unsubscribe_returned)
+        stopper.join(2)
+        logger.info("after unsubscribe")
+        assert len(received) == 1
+    finally:
+        release.set()
+        worker.join(2)
+        logger.removeHandler(handler)
+
+
+def test_file_logger_keeps_file_output_and_publishes_to_subscribers(tmp_path):
+    received = []
+    done = threading.Event()
+    unsubscribe = log_emitter.subscribe(lambda record: (received.append(record), done.set()))
+    logger = FileLogger(log_file=str(tmp_path / "sdk.log"), name="nitrostack.test.filelogger")
+    try:
+        logger.info("through FileLogger", meta={"source": "test"})
+        _wait_for(done)
+        assert "through FileLogger" in (tmp_path / "sdk.log").read_text(encoding="utf-8")
+        assert received[0].logger_name == "nitrostack.test.filelogger"
+        assert received[0].extra == {}
+        assert "meta: {'source': 'test'}" in received[0].message
+    finally:
+        unsubscribe()
+        for handler in list(logger.logger.handlers):
+            logger.logger.removeHandler(handler)
+            handler.close()
+
+
+def test_http_logger_with_propagation_disabled_reaches_subscribers():
+    from nitrostack.transports.http import _enable_trace_logging, logger as http_logger
+
+    received = []
+    done = threading.Event()
+    unsubscribe = log_emitter.subscribe(lambda record: (received.append(record), done.set()))
+    old_level = http_logger.level
+    old_propagate = http_logger.propagate
+    old_trace_handlers = {
+        handler
+        for handler in http_logger.handlers
+        if getattr(handler, "_nitrostack_trace", False)
+    }
+    try:
+        _enable_trace_logging()
+        http_logger.info("isolated HTTP logger")
+        _wait_for(done)
+        assert received[0].logger_name == "nitrostack.transports.http"
+        assert received[0].message == "isolated HTTP logger"
+    finally:
+        unsubscribe()
+        http_logger.setLevel(old_level)
+        http_logger.propagate = old_propagate
+        for handler in list(http_logger.handlers):
+            if (
+                getattr(handler, "_nitrostack_trace", False)
+                and handler not in old_trace_handlers
+            ):
+                http_logger.removeHandler(handler)
+                handler.close()
+
+
+def test_http_logger_enabled_before_subscribe_reaches_subscribers():
+    from nitrostack.transports.http import _enable_trace_logging, logger as http_logger
+
+    received = []
+    done = threading.Event()
+    old_level = http_logger.level
+    old_propagate = http_logger.propagate
+    old_trace_handlers = {
+        handler
+        for handler in http_logger.handlers
+        if getattr(handler, "_nitrostack_trace", False)
+    }
+    # Server start order: trace logging is enabled before tooling subscribes.
+    _enable_trace_logging()
+    unsubscribe = log_emitter.subscribe(lambda record: (received.append(record), done.set()))
+    try:
+        http_logger.info("HTTP record after late subscribe")
+        _wait_for(done)
+        assert received[0].logger_name == "nitrostack.transports.http"
+        assert received[0].message == "HTTP record after late subscribe"
+    finally:
+        unsubscribe()
+        assert not any(
+            getattr(handler, "_nitrostack_log_emitter", False)
+            for handler in http_logger.handlers
+        )
+        http_logger.setLevel(old_level)
+        http_logger.propagate = old_propagate
+        for handler in list(http_logger.handlers):
+            if (
+                getattr(handler, "_nitrostack_trace", False)
+                and handler not in old_trace_handlers
+            ):
+                http_logger.removeHandler(handler)
+                handler.close()
+
+
+def test_raising_subscriber_does_not_prevent_another_subscriber():
+    emitter = LogEmitter()
+    received = []
+    done = threading.Event()
+    unsubscribe_raising = emitter.subscribe(
+        lambda record: (_ for _ in ()).throw(RuntimeError("subscriber failure"))
+    )
+    unsubscribe_healthy = emitter.subscribe(lambda record: (received.append(record), done.set()))
+    logger = logging.getLogger("nitrostack.test.failure-isolation")
+    handler = LogEmitterHandler(emitter)
+    logger.addHandler(handler)
+    logger.setLevel(logging.INFO)
+    try:
+        logger.info("still delivered")
+        _wait_for(done)
+        assert received[0].message == "still delivered"
+    finally:
+        logger.removeHandler(handler)
+        unsubscribe_raising()
+        unsubscribe_healthy()
+
+
+def test_slow_subscriber_has_bounded_buffer_and_keeps_recent_records():
+    emitter = LogEmitter(max_buffer=2)
+    entered = threading.Event()
+    release = threading.Event()
+    drained = threading.Event()
+    received = []
+
+    def slow(record):
+        received.append(record.message)
+        if record.message == "0":
+            entered.set()
+            release.wait(2)
+        if record.message == "5":
+            drained.set()
+
+    unsubscribe_slow = emitter.subscribe(slow)
+    logger = logging.getLogger("nitrostack.test.bounded")
+    handler = LogEmitterHandler(emitter)
+    logger.addHandler(handler)
+    logger.setLevel(logging.INFO)
+    try:
+        logger.info("0")
+        _wait_for(entered)
+        for index in range(1, 5):
+            logger.info("%s", index)
+        finished = threading.Event()
+
+        def last_seen(record):
+            if record.message == "5":
+                finished.set()
+
+        # A second subscriber provides an asynchronous completion signal.
+        unsubscribe = emitter.subscribe(last_seen)
+        logger.info("5")
+        _wait_for(finished)
+        release.set()
+        _wait_for(drained)
+        unsubscribe()
+        assert received[0] == "0"
+        assert received[-2:] == ["4", "5"]
+    finally:
+        release.set()
+        logger.removeHandler(handler)
+        unsubscribe_slow()