diff --git a/docs/api-reference.md b/docs/api-reference.md
index 852b514c..5314a695 100644
--- a/docs/api-reference.md
+++ b/docs/api-reference.md
@@ -110,7 +110,7 @@ def your_function(args):
 
 - **`monitoring`** (`MonitoringConfig`, default: `MonitoringConfig()`) - Observability configuration:
   - `collect_stats` (`bool`, default: `True`) - Collect cache hit/miss statistics
-  - `enable_structured_logging` (`bool`, default: `True`) - Enable structured JSON logging
+  - `enable_structured_logging` (`bool`, default: `True`) - Log cache operations at INFO, with their fields on each record ([Structured Logging](#structured-logging))
   - `enable_prometheus_metrics` (`bool`, default: `True`) - Export Prometheus metrics
 
 #### Encryption Parameters
@@ -840,10 +840,11 @@ query examples, and alerting rules.
 
 ### Structured Logging
 
-All operations include structured logging with:
-- Operation context (namespace, cache key, serializer)
-- Performance metrics (duration, cache hit/miss)
-- Error classification and recovery actions
+cachekit writes two kinds of structured record through Python's `logging`, so your handlers and levels decide what is kept.
+
+With `enable_structured_logging=True` (the default), these cache operations are logged at INFO on the `cachekit.decorators.orchestrator` logger: L1 and L2 hits of a sync function that has a backend, failed operations and circuit-breaker rejections. Each record's `structured` attribute holds the operation, the namespace and the redacted cache key, plus the duration, hit, serializer and error type where they apply. The message is plain text; your formatter decides whether those fields are rendered.
+
+SDK diagnostics, such as the CachekitIO backend's debug lines, are one JSON object per record on the `cachekit.logging` logger, with `timestamp`, `level`, `message`, `logger`, `hostname`, `pid` and `thread_id`. Each is written when it is logged, on the calling thread. DEBUG, INFO and WARNING records are sampled at `CACHEKIT_LOG_SAMPLING_RATE` (default `0.1`), and ERROR and CRITICAL records never are. A field value JSON cannot encode is written as its type name, such as `"<dict>"`, never as its text. Nothing is sampled or serialised below that logger's level. A child forked by Python (`os.fork()`, `multiprocessing`, Gunicorn, Celery) writes its own records, with its own `pid`. A child forked from C without Python's at-fork hooks (uWSGI without `--py-call-uwsgi-fork-hooks`) writes none: `logging`'s after-fork lock reset did not run there, so a lock a parent thread held at the fork would hang the call.
 
 ## Best Practices
 
diff --git a/docs/performance.md b/docs/performance.md
index ff5f0969..94685136 100644
--- a/docs/performance.md
+++ b/docs/performance.md
@@ -215,9 +215,9 @@ The regression gate is the instruction budget, run locally with `make perf-ir`.
 `make perf-ir` counts the instructions each hot path executes per call and fails when a path costs **1% or more** above its committed budget. It warns from 0.2%. The orjson round trip fails at 2% and warns from 1% (see Limits). Instruction counts are deterministic where wall clock is not: two full runs agree within 0.08% per path (0.01% on CPython 3.12).
 
 **Method** (`tests/performance/ir_budget.py`; the measured process is `ir_workload.py`):
-- Each path runs under `valgrind --tool=callgrind --separate-threads=yes`, and only the main thread is counted. cachekit's background threads (log writer, L1 cleanup) vary by tens of percent between identical runs.
+- Each path runs under `valgrind --tool=callgrind --separate-threads=yes`, and only the main thread is counted. cachekit's background threads (L1 cleanup) vary by tens of percent between identical runs.
 - Per-call cost is `(Ir[3000 calls] - Ir[1000 calls]) / 2000`, so interpreter startup (~2 billion instructions) and warmup cancel. Interpreter teardown is skipped. The two real-data paths cost about 60 million instructions per call, so they run 10 and 30 calls after 5 warmup calls; startup still cancels, and two runs agreed within 0.03%.
-- The measured process has a fixed environment (`PYTHONHASHSEED=0`, one BLAS/OpenMP thread, a one-day log flush and L1 cleanup interval, nothing inherited), seeded log sampling, and main-thread clocks that advance 1μs per read. Code that records its own duration otherwise executes more instructions when it runs slower. The cleanup sweep reads the real clock, so inside a run it would evict entries stamped with the pinned one.
+- The measured process has a fixed environment (`PYTHONHASHSEED=0`, one BLAS/OpenMP thread, a one-day L1 cleanup interval, nothing inherited), seeded log sampling, and main-thread clocks that advance 1μs per read. Code that records its own duration otherwise executes more instructions when it runs slower. The cleanup sweep reads the real clock, so inside a run it would evict entries stamped with the pinned one.
 - Nothing that runs on real time lands in the measured loop: no background thread wakes, cyclic GC is off, and the GIL switch interval is long enough that the main thread gives up the GIL only where the code releases it.
 - Each path runs at five heap layouts (0 to 880 extra objects held before the workload) and its figure is the median. Any code change moves the layout, and the allocators take shorter or longer paths in some heap states: single layouts of the orjson round trip and the minimal L1 hit sat up to 2.4% from the rest. The median ignores one or two such layouts. The cheapest layout is not used, because a lucky one can sit 1% below the rest, and a budget recorded there makes every later run look like a regression.
 - `ir_workload.py`'s text is compiled into every measured process, so editing it moves the layout the budgets were recorded at: re-record after any change to it. The gate itself (`ir_budget.py`) is kept out of the measured process, so editing it does not move a budget.
diff --git a/src/cachekit/config/nested.py b/src/cachekit/config/nested.py
index 853ccfda..4db5aea2 100644
--- a/src/cachekit/config/nested.py
+++ b/src/cachekit/config/nested.py
@@ -234,7 +234,7 @@ class MonitoringConfig:
 
     Attributes:
         collect_stats: Collect cache hit/miss statistics (default: True)
-        enable_structured_logging: Enable structured JSON logging (default: True)
+        enable_structured_logging: Log cache operations at INFO, with their fields on each record (default: True)
         enable_prometheus_metrics: Export Prometheus metrics (default: True)
 
     Examples:
diff --git a/src/cachekit/config/settings.py b/src/cachekit/config/settings.py
index 6188bcee..82637657 100644
--- a/src/cachekit/config/settings.py
+++ b/src/cachekit/config/settings.py
@@ -196,21 +196,6 @@ class CachekitConfig(RedactingSettings):
         le=1.0,
         description="Log sampling rate (0.0 to 1.0, default 10%)",
     )
-    log_buffer_size: int = Field(
-        default=10000,
-        gt=0,
-        description="Ring buffer size for async logging",
-    )
-    log_batch_size: int = Field(
-        default=100,
-        gt=0,
-        description="Batch size for async log writes",
-    )
-    log_flush_interval: float = Field(
-        default=1.0,
-        gt=0,
-        description="Flush interval for async logging in seconds",
-    )
 
     # Deployment and feature flags
     deployment_uuid: Optional[str] = Field(
diff --git a/src/cachekit/logging.py b/src/cachekit/logging.py
index 554c9efc..a403d8a9 100644
--- a/src/cachekit/logging.py
+++ b/src/cachekit/logging.py
@@ -1,7 +1,12 @@
-"""Structured logging with minimal overhead.
+"""Structured JSON logging with sampling.
 
-This module provides lock-free, sampling-based structured logging
-with asynchronous batch writes.
+Each entry is serialised and written on the caller's thread. There is no background writer: a thread
+started at import is not running in a forked child (Gunicorn ``--preload``, Celery prefork,
+``multiprocessing``'s fork start method), so everything a child buffered for it was lost.
+
+In a child forked from C without at-fork hooks (uWSGI without ``--py-call-uwsgi-fork-hooks``),
+``log()`` and ``error()`` write no entry: ``logging``'s own after-fork lock reset did not run there, so
+a level check or a handler lock a parent thread held at fork would hang the caller.
 """
 
 import json
@@ -11,9 +16,9 @@
 import random
 import threading
 import time
-from dataclasses import dataclass, field
-from typing import Any, Optional
+from typing import Any
 
+from cachekit import l1_cache
 from cachekit.config import get_settings
 from cachekit.hash_utils import redact_error_for_log, redact_key_for_log
 
@@ -21,145 +26,62 @@
 logger = logging.getLogger(__name__)
 
 
-# Global configuration - loaded once at module import from the settings singleton
-_settings = get_settings()
-SAMPLING_RATE = _settings.log_sampling_rate
-RING_BUFFER_SIZE = _settings.log_buffer_size
-BATCH_SIZE = _settings.log_batch_size
-FLUSH_INTERVAL = _settings.log_flush_interval
-del _settings  # keep only the four scalars; don't pin a stale config past reset_settings()
+# Loaded once at module import from the settings singleton
+SAMPLING_RATE = get_settings().log_sampling_rate
 
 # Performance and health thresholds
 HIGH_UTILIZATION_THRESHOLD = 0.9  # When to warn about high utilization
 
+_LEVELS = {
+    "DEBUG": logging.DEBUG,
+    "INFO": logging.INFO,
+    "WARN": logging.WARNING,
+    "WARNING": logging.WARNING,
+    "ERROR": logging.ERROR,
+    "CRITICAL": logging.CRITICAL,
+    "FATAL": logging.CRITICAL,
+}
 
-@dataclass
-class LogEntry:
-    """Lightweight log entry for ring buffer."""
+# What json.dumps raises on a value its default= never sees: a non-str dict key, a cycle, a too-deep nest
+_UNENCODABLE = (TypeError, ValueError, RecursionError)
 
-    timestamp: float
-    level: str
-    message: str
-    extra: dict[str, Any] = field(default_factory=dict)
 
-    def to_dict(self) -> dict[str, Any]:
-        """Convert to dictionary for JSON serialization."""
-        return {
-            "timestamp": self.timestamp,
-            "level": self.level,
-            "message": self.message,
-            **self.extra,
-        }
+def _type_name(value: object) -> str:
+    """Stand in for a value JSON cannot encode: its type only, as its str() may embed a cache key (CWE-532)."""
+    return f"<{type(value).__name__}>"
 
 
-class LockFreeRingBuffer:
-    """Lock-free ring buffer for log entries.
+def _encodable(value: object) -> object:
+    """The value if json can encode it, else its type name."""
+    try:
+        json.dumps(value, default=_type_name)
+    except _UNENCODABLE:
+        return _type_name(value)
+    return value
 
-    Uses atomic operations to minimize contention.
-    """
 
-    def __init__(self, size: int = RING_BUFFER_SIZE):
-        self.size = size
-        self.buffer: list[Optional[LogEntry]] = [None] * size
-        self.write_pos = 0
-        self.read_pos = 0
-
-    def append(self, entry: LogEntry) -> bool:
-        """Append entry to buffer without locks.
-
-        Returns True if successful, False if buffer full.
-        """
-        # Simple atomic increment (in CPython, int ops are atomic)
-        pos = self.write_pos
-        next_pos = (pos + 1) % self.size
-
-        # Check if buffer is full (would overwrite unread data)
-        if next_pos == self.read_pos:
-            return False
-
-        # Write entry
-        self.buffer[pos] = entry
-        self.write_pos = next_pos
-        return True
-
-    def drain(self, max_items: int = BATCH_SIZE) -> list[LogEntry]:
-        """Drain up to max_items from buffer."""
-        items = []
-        count = 0
-
-        while count < max_items and self.read_pos != self.write_pos:
-            entry = self.buffer[self.read_pos]
-            if entry is not None:
-                items.append(entry)
-                self.buffer[self.read_pos] = None  # Clear reference
-            self.read_pos = (self.read_pos + 1) % self.size
-            count += 1
-
-        return items
-
-
-class AsyncLogWriter(threading.Thread):
-    """Background thread for async log writing."""
-
-    def __init__(self, buffer: LockFreeRingBuffer):
-        super().__init__(daemon=True, name="cachekit-LogWriter")
-        self.buffer = buffer
-        self.running = True
-        self._stop_event = threading.Event()
-
-    def run(self):
-        """Main loop - batch write logs periodically."""
-        while self.running:
-            try:
-                # Wait for interval or stop signal
-                if self._stop_event.wait(timeout=FLUSH_INTERVAL):
-                    break
-
-                # Drain and write batch
-                entries = self.buffer.drain()
-                if entries:
-                    self._write_batch(entries)
-
-            except Exception as e:
-                logger.error(f"Error in async log writer: {redact_error_for_log(e)}")
-
-    def stop(self):
-        """Stop the writer thread."""
-        self.running = False
-        self._stop_event.set()
-
-    def _write_batch(self, entries: list[LogEntry]):
-        """Write a batch of entries to the logger."""
-        for entry in entries:
-            try:
-                # Convert to JSON for structured logging
-                msg = json.dumps(entry.to_dict(), separators=(",", ":"))
-                logger.log(getattr(logging, entry.level, logging.INFO), msg)
-            except Exception:
-                # Silently drop malformed entries
-                pass
+def _enabled(levelno: int) -> bool:
+    """Whether to write an entry at levelno: never in a child forked without at-fork hooks (module docstring)."""
+    return not l1_cache._forked_without_hooks() and logger.isEnabledFor(levelno)
 
 
 class StructuredLogger:
     """Structured logger.
 
+    Each entry is one JSON line on the ``cachekit.logging`` logger, written synchronously.
+
     Features:
-    - Lock-free ring buffer
-    - Sampling (10% default)
-    - Async batch writes
+    - Nothing is sampled or serialised below that logger's level
+    - Sampling (10% default) of DEBUG/INFO/WARNING; ERROR and CRITICAL are never sampled
     - PII key-name masking (password/token/secret/key/auth kwargs)
     """
 
     def __init__(self, name: str):
         self.name = name
-        self.buffer = LockFreeRingBuffer()
-        self.writer = AsyncLogWriter(self.buffer)
-        self.writer.start()
 
         # Pre-computed values for performance
         self._sampling_threshold = int(SAMPLING_RATE * 100)
         self._hostname = platform.node()
-        self._pid = os.getpid()
 
         # PII patterns to mask (pre-compiled for speed)
         self._pii_keys = {"password", "token", "secret", "key", "auth"}
@@ -179,30 +101,30 @@ def _mask_pii(self, data: dict[str, Any]) -> dict[str, Any]:
                 data[key] = "****"
         return data
 
+    def _write(self, level: str, levelno: int, message: str, fields: dict[str, Any]) -> None:
+        """Serialise one entry and log it now."""
+        entry: dict[str, Any] = {
+            **self._mask_pii(fields),  # first, so a caller's pid= cannot replace the real one
+            "timestamp": time.time(),
+            "level": level,
+            "message": message,
+            "logger": self.name,
+            "hostname": self._hostname,
+            "pid": os.getpid(),  # read per entry: a forked child reports its own
+            "thread_id": threading.get_ident(),
+        }
+        try:
+            line = json.dumps(entry, separators=(",", ":"), default=_type_name)
+        except _UNENCODABLE:  # raising would replace the exception a caller may be logging
+            line = json.dumps({k: _encodable(v) for k, v in entry.items()}, separators=(",", ":"), default=_type_name)
+        logger.log(levelno, line)
+
     def log(self, level: str, message: str, **kwargs):
         """Main logging method with sampling."""
-        # Fast path - skip if not sampled
-        if not self._should_sample():
-            return
-
-        # Build entry
-        entry = LogEntry(
-            timestamp=time.time(),
-            level=level.upper(),
-            message=message,
-            extra={
-                "logger": self.name,
-                "hostname": self._hostname,
-                "pid": self._pid,
-                "thread_id": threading.get_ident(),
-                **self._mask_pii(kwargs),
-            },
-        )
-
-        # Append to lock-free buffer
-        if not self.buffer.append(entry):
-            # Buffer full - drop entry (non-blocking behavior)
-            pass
+        level = level.upper()
+        levelno = _LEVELS.get(level, logging.INFO)
+        if _enabled(levelno) and (levelno >= logging.ERROR or self._should_sample()):
+            self._write(level, levelno, message, kwargs)
 
     def debug(self, message: str, **kwargs):
         """Debug level logging."""
@@ -217,21 +139,9 @@ def warning(self, message: str, **kwargs):
         self.log("WARNING", message, **kwargs)
 
     def error(self, message: str, **kwargs):
-        """Error level logging - always sampled."""
-        # Errors bypass sampling
-        entry = LogEntry(
-            timestamp=time.time(),
-            level="ERROR",
-            message=message,
-            extra={
-                "logger": self.name,
-                "hostname": self._hostname,
-                "pid": self._pid,
-                "thread_id": threading.get_ident(),
-                **self._mask_pii(kwargs),
-            },
-        )
-        self.buffer.append(entry)
+        """Error level logging - never sampled. Not via log(): a caller field named level or message would collide."""
+        if _enabled(logging.ERROR):
+            self._write("ERROR", logging.ERROR, message, kwargs)
 
     def cache_operation(self, operation: str, cache_key: str, **kwargs):
         """Log cache operation with standard fields."""
@@ -293,21 +203,7 @@ def cache_operation(self, operation: str, cache_key: str, **kwargs):
             # Silently ignore if Prometheus metrics not available
             pass
 
-        # For compatibility with tests that expect standard logging
-        import logging as std_logging
-
-        # Map string level to logging constant
-        level_map = {
-            "DEBUG": std_logging.DEBUG,
-            "INFO": std_logging.INFO,
-            "WARNING": std_logging.WARNING,
-            "ERROR": std_logging.ERROR,
-            "CRITICAL": std_logging.CRITICAL,
-        }
-        log_level = level_map.get(level.upper(), std_logging.INFO)
-
-        logger = std_logging.getLogger(self.name)
-        logger.log(log_level, "cache_operation", extra={"structured": context})
+        self.logger.log(_LEVELS[level], "cache_operation", extra={"structured": context})
 
     def connection_pool_utilization(self, utilization: float, **kwargs):
         """Log connection pool metrics."""
@@ -335,11 +231,6 @@ def _get_context(self) -> dict[str, Any]:
             "thread_id": threading.get_ident(),
         }
 
-    def __del__(self):
-        """Cleanup on deletion."""
-        if hasattr(self, "writer"):
-            self.writer.stop()
-
 
 # Global logger instances cache
 _logger_instances: dict[str, StructuredLogger] = {}
diff --git a/tests/performance/ir_budget.py b/tests/performance/ir_budget.py
index a2e9ff93..1c3bf2b6 100644
--- a/tests/performance/ir_budget.py
+++ b/tests/performance/ir_budget.py
@@ -3,8 +3,8 @@
 Wall-clock benchmarks on a shared host cannot gate a 1% regression: run-to-run noise is several
 percent. Instruction counts can, once these sources of run-to-run noise are handled:
 
-- ``import cachekit`` starts background threads (the log writer, the L1 cleanup worker) whose
-  counts swing by tens of percent between identical runs. Only the MAIN thread is counted
+- cachekit runs background threads (the L1 cleanup worker) whose counts swing by tens of
+  percent between identical runs. Only the MAIN thread is counted
   (callgrind ``--separate-threads=yes``, file ``-01``). Work a library runs on its own threads
   is not counted either: about half of the real-data Arrow round trip runs on pyarrow's threads.
 - Interpreter startup is ~2 billion instructions. Each path runs at two loop sizes and the
@@ -222,11 +222,10 @@ def measure(paths: list[str], jobs: int, child_timeout_s: float = CHILD_TIMEOUT_
         "PYTHONNOUSERSITE": "1",
         "OPENBLAS_NUM_THREADS": "1",
         "OMP_NUM_THREADS": "1",  # pyarrow converts on a thread pool otherwise: +-0.5% per op
-        # cachekit's log writer wakes every 1 s and its L1 sweeper every 30 s of real time. A woken
-        # thread takes the GIL at a load-dependent moment, and the sweeper compares expiry times
-        # stamped by the pinned main-thread clock against the real one, so it evicts live entries
-        # mid-run. A day-long interval keeps both asleep for the whole run.
-        "CACHEKIT_LOG_FLUSH_INTERVAL": "86400",
+        # cachekit's L1 sweeper wakes every 30 s of real time. A woken thread takes the GIL at a
+        # load-dependent moment, and the sweeper compares expiry times stamped by the pinned
+        # main-thread clock against the real one, so it evicts live entries mid-run. A day-long
+        # interval keeps it asleep for the whole run.
         "CACHEKIT_L1_CLEANUP_INTERVAL_SECONDS": "86400",
     }
     global _stopped
diff --git a/tests/unit/test_config_env_fallback.py b/tests/unit/test_config_env_fallback.py
index ca0538d3..7dc9c4e2 100644
--- a/tests/unit/test_config_env_fallback.py
+++ b/tests/unit/test_config_env_fallback.py
@@ -115,6 +115,9 @@ def test_removed_knob_env_vars_are_ignored(self, monkeypatch):
             "L1_ENABLED": "false",
             "ENABLE_PROMETHEUS_METRICS": "false",
             "BACKEND_PROVIDER_CLASS": "cachekit.backends.redis.provider.RedisBackendProvider",
+            "LOG_BUFFER_SIZE": "10000",
+            "LOG_BATCH_SIZE": "100",
+            "LOG_FLUSH_INTERVAL": "86400",
         }
         for name, value in removed.items():
             monkeypatch.setenv(f"CACHEKIT_{name}", value)
diff --git a/tests/unit/test_l1_memory_bounds.py b/tests/unit/test_l1_memory_bounds.py
index 37db34a9..d2668f18 100644
--- a/tests/unit/test_l1_memory_bounds.py
+++ b/tests/unit/test_l1_memory_bounds.py
@@ -330,7 +330,6 @@ def _c_fork_child_outcome() -> object:
     import faulthandler
 
     from cachekit import l1_cache
-    from cachekit import logging as structured_logging
 
     manager = L1CacheManager(default_max_memory_mb=10)
     cache = manager.get_cache("c-fork-ns")  # captured pre-fork, like a decorator's _l1_cache
@@ -340,9 +339,6 @@ def _c_fork_child_outcome() -> object:
     assert cleanup is not None
     manager._stop_cleanup.set()  # ends it, still referenced: the parent ran cleanup
     cleanup.join()
-    for structured in structured_logging._logger_instances.values():  # their writers wake every flush interval
-        structured.writer.stop()
-        structured.writer.join()
     l1_logger = logging.getLogger("cachekit.l1_cache")
     l1_logger.addHandler(handler := logging.StreamHandler(open(os.devnull, "w")))  # noqa: SIM115 - ends with the process
     l1_logger.setLevel(logging.DEBUG)
diff --git a/tests/unit/test_structured_logging.py b/tests/unit/test_structured_logging.py
index fe98aa31..036941dc 100644
--- a/tests/unit/test_structured_logging.py
+++ b/tests/unit/test_structured_logging.py
@@ -1,7 +1,15 @@
 """Unit tests for structured logging module."""
 
+from __future__ import annotations
+
+import ast
+import json
 import logging
-from unittest.mock import patch
+import os
+import signal
+import sys
+import threading
+from unittest.mock import MagicMock, patch
 
 import pytest
 
@@ -9,6 +17,97 @@
     StructuredLogger,
     get_structured_logger,
 )
+from tests.utils.fork_helpers import child_outcome, report, run_fresh
+
+
+class _Capture(logging.Handler):
+    """Collects the JSON entries written to ``cachekit.logging``."""
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.entries: list[dict] = []
+
+    def emit(self, record: logging.LogRecord) -> None:
+        self.entries.append(json.loads(record.getMessage()))
+
+
+@pytest.fixture
+def written():
+    """Entries ``cachekit.logging`` writes while the test runs, with DEBUG enabled."""
+    sink = logging.getLogger("cachekit.logging")
+    capture, level = _Capture(), sink.level
+    sink.addHandler(capture)
+    sink.setLevel(logging.DEBUG)
+    try:
+        yield capture.entries
+    finally:
+        sink.removeHandler(capture)
+        sink.setLevel(level)
+
+
+def _c_fork_child_outcome() -> object:
+    """What a child forked from C reports; run only in a fresh interpreter (TestEntriesAfterFork's C-fork test).
+
+    Importing cachekit must leave this interpreter single-threaded: a thread waiting for the GIL at a
+    fork made from C leaves the child hung at its first GIL release, in no cachekit code. logging's
+    module lock and the handler's lock end held by threads the child lacks, and no level check is
+    cached: in the child, any call that takes either lock hangs, as no after-fork reset ran there.
+    """
+    import ctypes
+    import faulthandler
+
+    import cachekit  # noqa: F401 - everything an application's import starts
+
+    sink = logging.getLogger("cachekit.logging")
+    sink.addHandler(capture := _Capture())
+    sink.setLevel(logging.DEBUG)
+    before = get_structured_logger("test.c_fork.parent")  # built pre-fork, as a module-level _logger is
+    before._sampling_threshold = 100  # never sampled away
+    for lock in (logging._lock, capture.lock):  # held at fork by a thread the child lacks
+        holder = threading.Thread(target=lock.acquire)  # ends holding it
+        holder.start()
+        holder.join()
+    if threading.active_count() != 1:  # not an assert: python -O strips those
+        raise RuntimeError(f"a thread would be alive at fork: {threading.enumerate()}")
+    faulthandler.register(signal.SIGALRM, chain=True)  # the child inherits it: a hang dumps its stack
+    libc_fork = ctypes.PyDLL(None).fork  # PyDLL keeps the GIL through the call
+    r, w = os.pipe()
+    child = libc_fork()
+    if child == 0:
+        try:
+            signal.alarm(5)
+            os.close(r)
+            requested: list[str] = []
+
+            class _RecordingThread(threading.Thread):  # keeps the child single-threaded
+                def start(self) -> None:
+                    requested.append(self.name)
+
+            threading.Thread = _RecordingThread  # the child's copy only; it never returns
+            before.debug("child debug")
+            before.error("child error")
+            outcome: object = {"requested": requested, "written": capture.entries}
+        except BaseException as e:
+            outcome = {"error": repr(e)}
+        try:
+            report(w, outcome)
+        finally:
+            os._exit(1)
+    os.close(w)
+    return child_outcome(child, r)
+
+
+def _cycle() -> list:
+    cycle: list = ["ns:tenant-42:func:app.f:args:x"]
+    cycle.append(cycle)
+    return cycle
+
+
+def _too_deep() -> list:
+    nest: list = ["tenant-42"]
+    for _ in range(sys.getrecursionlimit() + 100):
+        nest = [nest]
+    return nest
 
 
 class TestStructuredLogger:
@@ -86,6 +185,154 @@ def test_cache_operation_error_logging(self, mock_log, logger):
         assert extra["error_type"] == "TimeoutError"
 
 
+class TestSynchronousEntries:
+    """Entries are written on the caller's thread: no writer thread, so none to lose after a fork."""
+
+    def test_error_is_written_before_returning_with_this_pid(self, written):
+        StructuredLogger("test.sync").error("boom", attempt=2)
+
+        assert len(written) == 1
+        entry = written[0]
+        assert (entry["level"], entry["message"], entry["logger"], entry["attempt"]) == ("ERROR", "boom", "test.sync", 2)
+        assert entry["pid"] == os.getpid()
+
+    def test_pii_named_fields_are_masked(self, written):
+        StructuredLogger("test.mask").error("boom", api_token=12345, attempt=1)
+
+        assert written[0]["api_token"] == "****"
+        assert written[0]["attempt"] == 1
+
+    def test_unencodable_value_is_written_as_its_type_only(self, written):
+        """json cannot encode it; its str() would print the key it carries (CWE-532), so only its type is written."""
+        StructuredLogger("test.unencodable").error("boom", failure=ValueError("ns:tenant-42:func:app.f:args:x"))
+
+        assert written[0]["failure"] == "<ValueError>"
+        assert "tenant-42" not in json.dumps(written)
+
+    @pytest.mark.parametrize(
+        ("build", "type_name"),
+        [
+            (lambda: {(1, 2): "ns:tenant-42:func:app.f:args:x"}, "<dict>"),
+            (_cycle, "<list>"),
+        ],
+        ids=["tuple-key", "cycle"],
+    )
+    def test_field_json_raises_on_is_written_as_its_type_only(self, written, build, type_name):
+        """json raises on it (TypeError, ValueError): the entry is still written, its other fields intact.
+
+        Raising instead would replace the exception a caller is logging from inside its except block.
+        """
+        StructuredLogger("test.unwalkable").error("boom", detail=build(), attempt=1)
+
+        assert (written[0]["message"], written[0]["detail"], written[0]["attempt"]) == ("boom", type_name, 1)
+        assert "tenant-42" not in json.dumps(written)
+
+    def test_field_nested_past_the_recursion_limit_is_written_as_its_type_only(self, monkeypatch, written):
+        """RecursionError, forced through json's pure-Python encoder, whose limit is sys.getrecursionlimit() everywhere.
+
+        The C encoder's limit is the thread's stack: on an unlimited stack, 3.14 encodes a 100_000-deep nest.
+        """
+        monkeypatch.setattr(json.encoder, "c_make_encoder", None)
+
+        StructuredLogger("test.too_deep").error("boom", detail=_too_deep(), attempt=1)
+
+        assert (written[0]["message"], written[0]["detail"], written[0]["attempt"]) == ("boom", "<list>", 1)
+
+    def test_generated_fields_win_over_same_named_caller_fields(self, written):
+        """A caller's pid= cannot make the entry report a process other than the one that wrote it."""
+        StructuredLogger("test.reserved").error("boom", pid=1, thread_id=2, logger="spoofed", level="DEBUG")
+
+        entry = written[0]
+        assert (entry["pid"], entry["thread_id"]) == (os.getpid(), threading.get_ident())
+        assert (entry["logger"], entry["level"]) == ("test.reserved", "ERROR")
+
+    @pytest.mark.parametrize(
+        ("level", "sampled", "builds"),
+        [(logging.DEBUG, True, True), (logging.DEBUG, False, False), (logging.INFO, True, False)],
+        ids=["enabled-sampled", "enabled-sampled-out", "disabled"],
+    )
+    def test_debug_is_built_only_when_enabled_and_sampled(self, monkeypatch, caplog, level, sampled, builds):
+        """Below the logger's level nothing is sampled or serialised: debug call sites cost a level check."""
+        caplog.set_level(level, logger="cachekit.logging")
+        structured = StructuredLogger("test.gate")
+        sample = MagicMock(return_value=sampled)
+        write = MagicMock()
+        monkeypatch.setattr(structured, "_should_sample", sample)
+        monkeypatch.setattr(structured, "_write", write)
+
+        structured.debug("detail", attempt=1)
+
+        assert sample.called is (level == logging.DEBUG)
+        assert write.called is builds
+
+    def test_error_and_above_are_never_sampled(self, monkeypatch, written):
+        structured = StructuredLogger("test.unsampled")
+        monkeypatch.setattr(structured, "_should_sample", lambda: False)
+
+        structured.error("boom")
+        structured.log("error", "via log")
+        structured.log("CRITICAL", "critical")
+        structured.warning("dropped")
+
+        assert [e["message"] for e in written] == ["boom", "via log", "critical"]
+
+    @pytest.mark.parametrize(
+        ("name", "levelno"),
+        [("warn", logging.WARNING), ("CRITICAL", logging.CRITICAL), ("FATAL", logging.CRITICAL), ("NOTICE", logging.INFO)],
+    )
+    def test_log_writes_a_level_name_at_its_stdlib_level(self, monkeypatch, caplog, name, levelno):
+        """Each stdlib level name maps to its own level, as the old writer's getattr(logging, name) did; others are INFO."""
+        caplog.set_level(logging.DEBUG, logger="cachekit.logging")
+        structured = StructuredLogger("test.levels")
+        monkeypatch.setattr(structured, "_should_sample", lambda: True)
+
+        structured.log(name, "boom")
+
+        assert [(r.levelno, json.loads(r.getMessage())["level"]) for r in caplog.records if r.name == "cachekit.logging"] == [
+            (levelno, name.upper())
+        ]
+
+    def test_constructing_a_logger_starts_no_thread(self, monkeypatch):
+        started: list[str] = []
+        monkeypatch.setattr(threading.Thread, "start", lambda self: started.append(self.name))
+
+        StructuredLogger("test.threadless").error("boom")
+
+        assert started == []
+
+
+@pytest.mark.skipif(not hasattr(os, "fork"), reason="fork() not available on this platform")
+class TestEntriesAfterFork:
+    """A forked child writes what it logs, with its own PID, and starts no thread doing so."""
+
+    def test_os_fork_child_writes_its_error_with_its_own_pid(self, written):
+        structured = get_structured_logger("test.os_fork")  # built pre-fork, as a module-level _logger is
+        r, w = os.pipe()
+        child = os.fork()
+        if child == 0:
+            try:
+                os.close(r)
+                written.clear()  # the child's copy
+                structured.error("child-error")
+                outcome: object = [(e["message"], e["pid"]) for e in written]
+            except BaseException as e:
+                outcome = repr(e)
+            report(w, outcome)
+        os.close(w)
+
+        assert child_outcome(child, r) == [("child-error", child)]
+
+    def test_child_of_a_c_fork_writes_nothing_and_starts_no_thread(self):
+        """A real fork from C: no at-fork hook runs, so logging's locks may be held by threads the child lacks.
+
+        The child's calls return without taking them: they write nothing and request no thread.
+        """
+        run = run_fresh(_c_fork_child_outcome)
+
+        assert run.returncode == 0, run.stderr
+        assert ast.literal_eval(run.stdout) == {"requested": [], "written": []}, run.stderr  # a hung child's stack
+
+
 class TestFactoryFunction:
     """Test factory function."""
 
diff --git a/tests/unit/test_swr_decorator.py b/tests/unit/test_swr_decorator.py
index b61d4c28..c278e52e 100644
--- a/tests/unit/test_swr_decorator.py
+++ b/tests/unit/test_swr_decorator.py
@@ -342,7 +342,6 @@ def _forked_stale_hit_outcome(fork: str) -> object:
     import types
 
     from cachekit import l1_cache, object_cache
-    from cachekit import logging as structured_logging
     from cachekit.config import L1CacheConfig
 
     backend = FakeSWRBackend()
@@ -363,9 +362,6 @@ def l2_swr() -> int:
     if manager._cleanup_thread is not None:  # ends it, still referenced: the parent ran cleanup
         manager._stop_cleanup.set()
         manager._cleanup_thread.join()
-    for structured in structured_logging._logger_instances.values():  # their writers wake every flush interval
-        structured.writer.stop()
-        structured.writer.join()
     if threading.active_count() != 1:  # not an assert: python -O strips those
         raise RuntimeError(f"a thread would be alive at fork: {threading.enumerate()}")
     faulthandler.register(signal.SIGALRM, chain=True)  # the child inherits it: a hang dumps its stack
diff --git a/tests/utils/tracemalloc_isolation.py b/tests/utils/tracemalloc_isolation.py
index 29eedf01..eb2b80bb 100644
--- a/tests/utils/tracemalloc_isolation.py
+++ b/tests/utils/tracemalloc_isolation.py
@@ -3,8 +3,8 @@
 tracemalloc.start()/stop() swap the process-wide allocator hooks without synchronising with
 threads that are allocating at that moment, so on a free-threaded build a concurrent
 allocation can crash the process: https://github.com/python/cpython/issues/143143. The pytest
-process is never single-threaded (``import cachekit`` starts the log writer; pytest-xdist adds
-its I/O thread), so every tracemalloc start/stop in the test suite goes through here.
+process is never single-threaded (pytest-xdist adds its I/O thread), so every tracemalloc
+start/stop in the test suite goes through here.
 """
 
 from __future__ import annotations
@@ -19,14 +19,9 @@
 
 import pytest
 
-from cachekit import logging as ck_logging
-
 
 def _run_alone(measure: Callable[[], Any]) -> Any:
-    """Subprocess side: stop cachekit's log writers, then measure with no other thread alive."""
-    for structured in ck_logging._logger_instances.values():
-        structured.writer.stop()
-        structured.writer.join()
+    """Subprocess side: measure with no other thread alive."""
     assert threading.active_count() == 1, f"tracemalloc must not start or stop beside {threading.enumerate()}"
     result = measure()
     assert threading.active_count() == 1, f"a thread started while measuring: {threading.enumerate()}"