diff --git a/.coverage b/.coverage
deleted file mode 100644
index 89085e7..0000000
Binary files a/.coverage and /dev/null differ
diff --git a/.github/copilot-instructions.md b/.github/copilot-instructions.md
index 3f9ac7b..c91f1c4 100644
--- a/.github/copilot-instructions.md
+++ b/.github/copilot-instructions.md
@@ -18,7 +18,9 @@ Python style
 - Use type aliases for complex types
 - Use built-in generics for type hints
 - Use `logging.getLogger(__name__)`, not print
-- Lazy % formatting in logging: `logger.info("msg %s", var)`
+- Structured logging: constant message + data in `extra`: `logger.info("Message accepted.", extra={"writer": name})`
+- Do not re-log `topic`, `user`, `resource`, `http_method` or the Lambda context; they are bound per request in `src/utils/observability.py`
+- `logger.exception()` only inside an `except` block; outside it pass `exc_info=exc` to `logger.error()`
 - F-strings in exceptions: `raise ValueError(f"Error {var}")`
 - All imports at top of file (never inside functions)
 - Apache 2.0 license header in every .py file (including `__init__.py`)
@@ -27,6 +29,8 @@ Python style
 - Use single backticks in docstrings (`value`), never double backticks (`` ``value`` ``)
 - Do not use `# -----------` separator comments to divide sections
 - End all log messages with a period: `logger.info("Message.")`
+- Every non-2xx response must produce exactly one log line explaining the cause
+- Levels: TRACE payloads, DEBUG steps, INFO request outcomes, WARNING rejected requests, ERROR actionable failures
 
 Patterns
 - `__init__` methods must not raise exceptions; defer validation and connection to first use (lazy init)
diff --git a/DEVELOPER.md b/DEVELOPER.md
index 0eeff70..0616d88 100644
--- a/DEVELOPER.md
+++ b/DEVELOPER.md
@@ -208,6 +208,45 @@ View container logs in pytest output by increasing log level:
 pytest tests/integration/ -v --log-cli-level=DEBUG
 ```
 
+## Logging Conventions
+
+Logging is structured. `src/utils/observability.py` attaches the AWS Lambda Powertools JSON handler to the root logger, so modules keep using the standard library logger and inherit the format, the Lambda execution context and the request correlation id.
+
+```python
+import logging
+
+logger = logging.getLogger(__name__)
+```
+
+Rules:
+
+- The message is a constant sentence ending with a period. Variable data goes into `extra`, never into the sentence.
+  ```python
+  logger.warning("Request rejected: unknown topic.", extra={"known_topics": sorted(known_topics)})
+  ```
+- Do not add `topic`, `user`, `resource`, `http_method`, `correlation_id` or the Lambda context to `extra`; they are bound once per request by `bind_request_context()` and `append_request_keys()`.
+- Every non-2xx response must produce exactly one log line explaining the cause.
+- Levels: `TRACE` payloads, `DEBUG` steps, `INFO` request outcomes, `WARNING` rejected requests and soft failures, `ERROR` failures that need action. See the level table in [README](./README.md#logging--correlation).
+- Never log tokens, passwords or full message payloads outside `TRACE`. `TRACE` payload logging goes through `log_payload_at_trace()`, which redacts and size caps the payload.
+- `logger.exception()` is only valid inside an `except` block. Outside one, pass the captured exception: `logger.error("...", exc_info=exc)`.
+- Durations are logged as milliseconds with an explicit key (`duration_ms`, `writer_duration_ms`, `query_duration_ms`).
+
+Assert on structured fields in tests, not on formatted strings:
+
+```python
+def test_rejects_unknown_topic(caplog):
+    caplog.set_level(logging.WARNING)
+    ...
+    assert "Request rejected: unknown topic." == caplog.records[-1].message
+```
+
+Keys bound with `append_request_keys()` live on the Powertools formatter rather than on the `LogRecord`. To assert on them, render the record:
+
+```python
+payload = json.loads(logger.registered_formatter.format(caplog.records[-1]))
+assert "run-42" == payload["correlation_id"]
+```
+
 ## Run All Quality Gates
 
 Run Black, Pylint, mypy, unit tests (with coverage), and integration tests in a single command:
diff --git a/README.md b/README.md
index f8323a9..e5e83ce 100644
--- a/README.md
+++ b/README.md
@@ -128,11 +128,52 @@ Supporting configs:
 - `topic_schemas/*.json` – each file contains a JSON Schema for a topic. In the current code these are explicitly loaded inside `event_gate_lambda.py`. (Future enhancement: auto-discover or index file.)
 
 Environment variables:
-- `LOG_LEVEL` (optional) – defaults to `INFO`.
+- `LOG_LEVEL` (optional) – defaults to `INFO`. Accepts `TRACE`, `DEBUG`, `INFO`, `WARNING`, `ERROR`. An unknown value falls back to `INFO` and is reported with a warning.
+- `POWERTOOLS_LOG_LEVEL` (optional) – takes precedence over `LOG_LEVEL`.
+- `POWERTOOLS_SERVICE_NAME` (optional) – value of the `service` key in every log line. Defaults to `eventgate`.
+- `TRACE_REDACT_KEYS` (optional) – comma separated message keys redacted from `TRACE` payload logs. Defaults to `password,secret,token,key,apikey,api_key`.
+- `TRACE_MAX_BYTES` (optional) – maximum size of a logged `TRACE` payload. Defaults to `10000`.
 - `CONF_DIR` (optional) – directory containing `config.json` and `access.json`. Defaults to `conf`.
 - `POSTGRES_SECRET_NAME` (optional) – AWS Secrets Manager secret name holding PostgreSQL connection credentials (host, port, database, user, password). Required for Postgres writer and stats reader.
 - `POSTGRES_SECRET_REGION` (optional) – AWS region of the Secrets Manager secret. Must be set together with `POSTGRES_SECRET_NAME`.
 
+## Logging & Correlation
+
+Both lambdas emit structured JSON logs through [AWS Lambda Powertools](https://docs.aws.amazon.com/powertools/python/latest/core/logger/). Every line carries the service name, log level, the Lambda execution context (`function_name`, `function_request_id`, `cold_start`) and a `correlation_id`.
+
+The correlation id is resolved per request in this order:
+1. The `X-Correlation-ID` request header, when it matches `^[A-Za-z0-9._:-]{1,128}$`.
+2. The `X-Request-ID` request header, with the same constraint.
+3. The API Gateway request id (`requestContext.requestId`).
+
+Callers that already have a run or job id should send it as `X-Correlation-ID` so their logs and EventGate logs can be joined. The resolved id is returned in the `X-Correlation-ID` response header of every response, including errors.
+
+Log levels used by the service:
+
+| Level     | Content                                                                                       |
+|-----------|-----------------------------------------------------------------------------------------------|
+| `TRACE`   | Full message payloads, redacted and size capped. Never enable by default.                      |
+| `DEBUG`   | Configuration loading, lazy initialization, per-writer send attempts, connection reuse.        |
+| `INFO`    | One line per request outcome, cold start initialization, accepted messages, stats query results.|
+| `WARNING` | Rejected requests (auth, authorization, validation), degraded health, Kafka flush retries.     |
+| `ERROR`   | Writer failures, partial fan-out failures, failed queries, unhandled request errors.           |
+
+Every non-2xx response has exactly one log line explaining the cause.
+
+Example CloudWatch Logs Insights queries:
+
+```text
+fields @timestamp, level, message, status_code, duration_ms
+| filter correlation_id = "<id>"
+| sort @timestamp asc
+```
+
+```text
+fields @timestamp, topic, user, message
+| filter level = "WARNING" and status_code = 403
+| stats count() by user, topic
+```
+
 ## Local Development & Testing
 
 | Purpose                       | Relative link                                                         |
diff --git a/adr/001-observability/001-observability.md b/adr/001-observability/001-observability.md
new file mode 100644
index 0000000..1e66639
--- /dev/null
+++ b/adr/001-observability/001-observability.md
@@ -0,0 +1,84 @@
+# ADR 001. Observability -- structured logging and request correlation
+
+## Decision
+
+Adopt [AWS Lambda Powertools for Python](https://docs.aws.amazon.com/powertools/python/latest/core/logger/) as the logging backend for both EventGate lambdas. Logs are emitted as JSON, enriched with the Lambda execution context, and carry a request correlation id that is also returned to the caller. AWS X-Ray tracing is deliberately **not** adopted at this point.
+
+### Background
+
+EventGate ran with 79 log statements, of which 52 were `DEBUG` and none were `INFO`. Production runs with `LOG_LEVEL=INFO`, so the service emitted effectively nothing per request: no request line, no outcome, no reason for a rejection. Raising the level to `DEBUG` was the only alternative and produced unusable output, because the level was applied to the root logger and `botocore`, `urllib3` and `s3transfer` flooded the log group.
+
+The concrete operational failures this caused:
+
+- A client receiving `401`, `403` or `400` produced **no log line at all**. Support could not answer "why was I rejected?".
+- A partial fan-out failure (Kafka accepted, Postgres failed) returned `500` without recording which sink had already accepted the message, so reconciliation had to be done by hand.
+- Nothing tied log lines from one invocation together, and nothing tied EventGate's lines to the caller's job or run.
+- No durations were recorded anywhere, so "which writer is slow" was unanswerable.
+
+See issue [#193](https://github.com/AbsaOSS/EventGate/issues/193).
+
+### Considered options
+
+**1. Plain standard library logging with a hand-written JSON formatter.**
+No new dependency, but the Lambda execution context, cold start detection, correlation id plumbing and log sampling would all have to be written and maintained here.
+
+**2. AWS Lambda Powertools (chosen).**
+Purpose-built for this runtime. Provides the JSON formatter, `cold_start`, `function_request_id`, correlation id handling and sampling out of the box. The lambdas are deployed as container images, so it is a single `requirements.txt` entry with no layer ARNs to pin per region. Cost is one production dependency and roughly 2-4 MB of image size, negligible next to the librdkafka build already in the image.
+
+**3. Powertools plus X-Ray tracing.**
+Deferred. X-Ray usage in the target account is not decided, it carries its own cost, and `confluent-kafka` is not patchable by the X-Ray SDK, so Kafka spans would need manual instrumentation. The correlation id covers the immediate need, which is joining log lines across an invocation and across services.
+
+### Integration approach
+
+Powertools' `Logger` is created once in `src/utils/observability.py`, and its handler is attached to the **root** logger by `configure_root_logging()`. Modules keep using `logging.getLogger(__name__)`.
+
+This matters for two reasons:
+
+- Existing call sites did not have to change to gain JSON output, so the change stayed reviewable.
+- Persistent keys (including `correlation_id`) live on the shared `LambdaPowertoolsFormatter` instance, so a key appended once per request appears on every module's log lines without threading a logger through the call stack.
+
+The alternative, `copy_config_to_registered_loggers()`, sets `propagate = False` on each module logger, which breaks `caplog` in the unit tests. Attaching at the root avoids that.
+
+`inject_lambda_context` is **not** used as a decorator. It raises `AttributeError` when the Lambda context is `None`, which is how both unit and integration tests invoke `lambda_handler`. `bind_request_context()` performs the same binding and tolerates a missing context.
+
+### Log level policy
+
+| Level     | Content                                                                                        |
+|-----------|------------------------------------------------------------------------------------------------|
+| `TRACE`   | Full message payloads, redacted and size capped. Never enabled by default.                       |
+| `DEBUG`   | Configuration loading, lazy initialization, per-writer send attempts, connection reuse.          |
+| `INFO`    | One line per request outcome, cold start initialization, accepted messages, stats query results. |
+| `WARNING` | Rejected requests (auth, authorization, validation), degraded health, Kafka flush retries.       |
+| `ERROR`   | Writer failures, partial fan-out failures, failed queries, unhandled request errors.             |
+
+The binding rule is: **every non-2xx response produces exactly one log line explaining the cause.**
+
+Third-party loggers (`boto3`, `botocore`, `urllib3`, `s3transfer`, `aiosql`, `confluent_kafka`) are capped at `WARNING` so that running the service at `DEBUG` or `TRACE` stays readable.
+
+The custom `TRACE` level (5) introduced earlier is retained. It is registered with the standard `logging` module before the Powertools `Logger` is constructed, and the level is passed numerically, so Powertools' level validation resolves it. A unit test pins this behaviour, because an unrecognised level would silently fall back to `INFO` and disable payload logging without any error.
+
+### Correlation id contract
+
+Resolved per request, in order:
+
+1. `X-Correlation-ID` request header.
+2. `X-Request-ID` request header.
+3. API Gateway request id (`requestContext.requestId`).
+
+Header values are accepted only when they match `^[A-Za-z0-9._:-]{1,128}$`. Inbound header values are written into the log stream, so newlines and control characters are rejected to prevent log injection, and the length is capped so the id cannot bloat every line.
+
+The resolved id is returned in the `X-Correlation-ID` response header of every response, success or error, so a client can quote a single id when reporting a problem. It is stamped centrally in `dispatch_request()`, which is the single point every response passes through.
+
+### Consequences
+
+- Log output changes from plain text to JSON. Nothing downstream parses these logs today; they are read by humans and queried in CloudWatch Logs Insights, where JSON is strictly easier to query.
+- `dispatch_request()` now catches `Exception` at the boundary instead of a fixed list of six exception types. Previously a `psycopg2.Error` or `KafkaException` escaping a handler reached the runtime, producing an unstructured traceback and an opaque `502` from API Gateway. `SystemExit` still propagates, so the `/terminate` route is unaffected.
+- A malformed request body on `POST /topics/{topic_name}` now returns `400 validation` instead of `500 internal`. It is a client error and was only reaching the internal error path because `json.loads` was left to the boundary handler.
+- Callers gain a response header they did not have before. This is additive and does not change the response body contract.
+- CloudWatch cost rises slightly. `INFO` stays at roughly one to three lines per invocation; `POWERTOOLS_LOGGER_SAMPLE_RATE` is available if `DEBUG` detail is wanted in production without enabling it for every invocation.
+
+### Not in scope
+
+- **X-Ray tracing.** If it is enabled later, `Tracer` from the same library plugs into `observability.py`; the function needs `Tracing: Active` and the `xray:PutTraceSegments` permission, and Kafka spans need manual subsegments.
+- **EMF metrics** (writer failures, auth failures, cold starts) as alarm sources. Worth a separate issue.
+- **Downstream correlation propagation**: Kafka message headers and the EventBridge `TraceHeader`. Useful, but only once a consumer is ready to read them.
diff --git a/requirements.txt b/requirements.txt
index 8edb61c..52f912e 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,3 +1,4 @@
+aws-lambda-powertools==3.31.1
 cryptography==49.0.0
 jsonschema==4.26.0
 PyJWT==2.13.0
diff --git a/src/event_gate_lambda.py b/src/event_gate_lambda.py
index 3988717..748d213 100644
--- a/src/event_gate_lambda.py
+++ b/src/event_gate_lambda.py
@@ -16,8 +16,9 @@
 
 """AWS Lambda entry point for the EventGate service."""
 
-import os
+import logging
 import sys
+import time
 from typing import Any
 
 import boto3
@@ -27,22 +28,22 @@
 from src.handlers.handler_health import HandlerHealth
 from src.handlers.handler_token import HandlerToken
 from src.handlers.handler_topic import HandlerTopic
-from src.utils.conf_path import CONF_DIR, INVALID_CONF_ENV
+from src.utils.conf_path import CONF_DIR, log_configuration_source
 from src.utils.config_loader import load_config
 from src.utils.constants import SSL_CA_BUNDLE_KEY
-from src.utils.logging_levels import init_root_logger
+from src.utils.observability import configured_log_level, init_lambda_logging
 from src.utils.utils import dispatch_request
 from src.writers.writer_eventbridge import WriterEventBridge
 from src.writers.writer_kafka import WriterKafka
 from src.writers.writer_postgres import WriterPostgres
 
-logger = init_root_logger(__name__)
-logger.debug("Initialized logger with level %s.", os.environ.get("LOG_LEVEL", "INFO"))
+logger = init_lambda_logging(__name__)
+
+_INIT_STARTED_AT = time.perf_counter()
+logger.info("Initializing EventGate lambda.", extra={"log_level": configured_log_level()})
 
 # Load main configuration
-logger.debug("Using CONF_DIR=%s.", CONF_DIR)
-if INVALID_CONF_ENV:
-    logger.warning("CONF_DIR env var set to non-existent path: %s; fell back to %s.", INVALID_CONF_ENV, CONF_DIR)
+log_configuration_source(logger)
 config = load_config(CONF_DIR)
 logger.debug("Loaded main configuration.")
 
@@ -73,6 +74,15 @@
 handler_health = HandlerHealth(writers)
 handler_api = HandlerApi().with_api_definition_loaded()
 
+logger.info(
+    "EventGate lambda initialized.",
+    extra={
+        "init_duration_ms": round((time.perf_counter() - _INIT_STARTED_AT) * 1000, 2),
+        "writers": sorted(writers),
+        "topics": sorted(handler_topic.topics),
+    },
+)
+
 # Route to handler function mapping
 ROUTE_MAP: dict[str, Any] = {
     "/api": lambda _: handler_api.get_api(),
@@ -85,12 +95,12 @@
 }
 
 
-def lambda_handler(event: dict[str, Any], _context: Any = None) -> dict[str, Any]:
+def lambda_handler(event: dict[str, Any], context: Any = None) -> dict[str, Any]:
     """AWS Lambda entry point for EventGate. Dispatches based on API Gateway proxy resource field.
     Args:
         event: The event data from API Gateway.
-        _context: The mandatory context argument for AWS Lambda invocation (unused).
+        context: The AWS Lambda context object, used to enrich logs with the execution context.
     Returns:
         A dictionary compatible with API Gateway Lambda Proxy integration.
     """
-    return dispatch_request(event, ROUTE_MAP, logger)
+    return dispatch_request(event, ROUTE_MAP, logger, context)
diff --git a/src/event_stats_lambda.py b/src/event_stats_lambda.py
index f8bc438..aab4e43 100644
--- a/src/event_stats_lambda.py
+++ b/src/event_stats_lambda.py
@@ -16,24 +16,25 @@
 
 """AWS Lambda entry point for the EventStats service."""
 
-import os
+import logging
+import time
 from typing import Any
 
 from src.handlers.handler_health import HandlerHealth
 from src.handlers.handler_stats import HandlerStats
 from src.readers.reader_postgres import ReaderPostgres
-from src.utils.conf_path import CONF_DIR, INVALID_CONF_ENV
+from src.utils.conf_path import CONF_DIR, log_configuration_source
 from src.utils.config_loader import load_topic_names
-from src.utils.logging_levels import init_root_logger
+from src.utils.observability import configured_log_level, init_lambda_logging
 from src.utils.utils import dispatch_request
 
-logger = init_root_logger(__name__)
-logger.debug("Initialized EventStats logger with level %s.", os.environ.get("LOG_LEVEL", "INFO"))
+logger = init_lambda_logging(__name__)
+
+_INIT_STARTED_AT = time.perf_counter()
+logger.info("Initializing EventStats lambda.", extra={"log_level": configured_log_level()})
 
 # Load main configuration
-logger.debug("Using CONF_DIR=%s.", CONF_DIR)
-if INVALID_CONF_ENV:
-    logger.warning("CONF_DIR env var set to non-existent path: %s; fell back to %s.", INVALID_CONF_ENV, CONF_DIR)
+log_configuration_source(logger)
 
 # Load topic names.
 topic_names = load_topic_names(CONF_DIR)
@@ -46,6 +47,14 @@
 handler_stats = HandlerStats(topics, reader_postgres)
 handler_health = HandlerHealth({"postgres_reader": reader_postgres})
 
+logger.info(
+    "EventStats lambda initialized.",
+    extra={
+        "init_duration_ms": round((time.perf_counter() - _INIT_STARTED_AT) * 1000, 2),
+        "topics": sorted(topics),
+    },
+)
+
 # Route to handler function mapping
 ROUTE_MAP: dict[str, Any] = {
     "/stats/{topic_name}": handler_stats.handle_request,
@@ -53,12 +62,12 @@
 }
 
 
-def lambda_handler(event: dict[str, Any], _context: Any = None) -> dict[str, Any]:
+def lambda_handler(event: dict[str, Any], context: Any = None) -> dict[str, Any]:
     """AWS Lambda entry point for EventStats. Dispatches based on API Gateway proxy resource field.
     Args:
         event: The event data from API Gateway.
-        _context: The mandatory context argument for AWS Lambda invocation (unused).
+        context: The AWS Lambda context object, used to enrich logs with the execution context.
     Returns:
         A dictionary compatible with API Gateway Lambda Proxy integration.
     """
-    return dispatch_request(event, ROUTE_MAP, logger)
+    return dispatch_request(event, ROUTE_MAP, logger, context)
diff --git a/src/handlers/handler_api.py b/src/handlers/handler_api.py
index beda89a..e37af5a 100644
--- a/src/handlers/handler_api.py
+++ b/src/handlers/handler_api.py
@@ -46,10 +46,10 @@ def with_api_definition_loaded(self) -> "HandlerApi":
             if not self.api_spec:
                 raise ValueError("API specification file is empty")
 
-            logger.debug("Loaded API definition from %s.", api_path)
+            logger.debug("Loaded API definition.", extra={"api_path": api_path})
             return self
         except (FileNotFoundError, PermissionError, ValueError) as exc:
-            logger.exception("Failed to load or read API specification from %s.", api_path)
+            logger.exception("Failed to load or read the API specification.", extra={"api_path": api_path})
             raise RuntimeError("API specification initialization failed") from exc
 
     def get_api(self) -> dict[str, Any]:
diff --git a/src/handlers/handler_health.py b/src/handlers/handler_health.py
index d184ea2..eeab801 100644
--- a/src/handlers/handler_health.py
+++ b/src/handlers/handler_health.py
@@ -68,14 +68,14 @@ def get_health(self) -> dict[str, Any]:
         uptime_seconds = int((datetime.now(timezone.utc) - self.start_time).total_seconds())
 
         if not failures:
-            logger.debug("Health check passed.")
+            logger.debug("Health check passed.", extra={"dependencies": statuses})
             return {
                 "statusCode": 200,
                 "headers": {"Content-Type": "application/json"},
                 "body": json.dumps({"status": "ok", "uptime_seconds": uptime_seconds, "dependencies": statuses}),
             }
 
-        logger.debug("Health check degraded: %s.", failures)
+        logger.warning("Health check degraded.", extra={"failures": failures, "dependencies": statuses})
         return {
             "statusCode": 503,
             "headers": {"Content-Type": "application/json"},
diff --git a/src/handlers/handler_stats.py b/src/handlers/handler_stats.py
index 391d02d..7a48bfa 100644
--- a/src/handlers/handler_stats.py
+++ b/src/handlers/handler_stats.py
@@ -18,10 +18,12 @@
 
 import json
 import logging
+import time
 from typing import Any
 
 from src.readers.reader_postgres import ReaderPostgres
 from src.utils.constants import POSTGRES_DEFAULT_LIMIT, SUPPORTED_STATS_TOPICS
+from src.utils.observability import append_request_keys
 from src.utils.utils import build_error_response
 
 logger = logging.getLogger(__name__)
@@ -38,6 +40,14 @@ def __init__(
         self.topics = topics
         self.reader_postgres = reader_postgres
 
+    @staticmethod
+    def _log_field_rejected(field_name: str) -> None:
+        """Log a rejected request body field.
+        Args:
+            field_name: Name of the field that failed validation.
+        """
+        logger.warning("Request rejected: request body field is invalid.", extra={"field": field_name})
+
     def handle_request(self, event: dict[str, Any]) -> dict[str, Any]:
         """Handle POST /stats/{topic_name} requests.
         Args:
@@ -48,13 +58,22 @@ def handle_request(self, event: dict[str, Any]) -> dict[str, Any]:
         path_params = event.get("pathParameters") or {}
         topic_name = path_params.get("topic_name", "")
         if not topic_name:
+            logger.warning("Request rejected: path parameter 'topic_name' is missing.")
             return build_error_response(400, "validation", "Missing path parameter 'topic_name'.")
         topic_name = topic_name.lower()
+        append_request_keys(topic=topic_name)
+
+        logger.debug("Handling POST topic stats.")
 
         if topic_name not in self.topics:
+            logger.warning("Request rejected: unknown topic.", extra={"known_topics": sorted(self.topics)})
             return build_error_response(404, "topic", f"Topic '{topic_name}' not found.")
 
         if topic_name not in SUPPORTED_STATS_TOPICS:
+            logger.warning(
+                "Request rejected: stats are not supported for the topic.",
+                extra={"supported_topics": sorted(SUPPORTED_STATS_TOPICS)},
+            )
             return build_error_response(
                 400, "validation", f"Stats are only supported for topics '{', '.join(SUPPORTED_STATS_TOPICS)}'."
             )
@@ -63,9 +82,11 @@ def handle_request(self, event: dict[str, Any]) -> dict[str, Any]:
         try:
             body = json.loads(event.get("body") or "{}")
         except (json.JSONDecodeError, TypeError):
+            logger.warning("Request rejected: request body is not valid JSON.")
             return build_error_response(400, "validation", "Request body must be valid JSON.")
 
         if not isinstance(body, dict):
+            logger.warning("Request rejected: request body is not a JSON object.")
             return build_error_response(400, "validation", "Request body must be a JSON object.")
 
         timestamp_start = body.get("timestamp_start")
@@ -74,15 +95,20 @@ def handle_request(self, event: dict[str, Any]) -> dict[str, Any]:
         limit: int = body.get("limit", POSTGRES_DEFAULT_LIMIT)
 
         if timestamp_start is not None and (isinstance(timestamp_start, bool) or not isinstance(timestamp_start, int)):
+            self._log_field_rejected("timestamp_start")
             return build_error_response(400, "validation", "Field 'timestamp_start' must be an integer (epoch ms).")
         if timestamp_end is not None and (isinstance(timestamp_end, bool) or not isinstance(timestamp_end, int)):
+            self._log_field_rejected("timestamp_end")
             return build_error_response(400, "validation", "Field 'timestamp_end' must be an integer (epoch ms).")
         if cursor is not None and (isinstance(cursor, bool) or not isinstance(cursor, int)):
+            self._log_field_rejected("cursor")
             return build_error_response(400, "validation", "Field 'cursor' must be an integer (internal_id).")
         if not isinstance(limit, int) or isinstance(limit, bool) or limit < 1:
+            self._log_field_rejected("limit")
             return build_error_response(400, "validation", "Field 'limit' must be a positive integer.")
 
         # Execute query
+        started_at = time.perf_counter()
         try:
             rows, pagination = self.reader_postgres.read_stats(
                 timestamp_start=timestamp_start,
@@ -91,9 +117,22 @@ def handle_request(self, event: dict[str, Any]) -> dict[str, Any]:
                 limit=limit,
             )
         except RuntimeError:
-            logger.exception("Stats query failed for topic %s.", topic_name)
+            logger.exception(
+                "Stats query failed.",
+                extra={"query_duration_ms": round((time.perf_counter() - started_at) * 1000, 2)},
+            )
             return build_error_response(500, "database", "Stats query failed.")
 
+        logger.info(
+            "Stats query completed.",
+            extra={
+                "query_duration_ms": round((time.perf_counter() - started_at) * 1000, 2),
+                "row_count": len(rows),
+                "has_more": pagination.get("has_more"),
+                "limit": pagination.get("limit"),
+            },
+        )
+
         return {
             "statusCode": 200,
             "headers": {"Content-Type": "application/json"},
diff --git a/src/handlers/handler_token.py b/src/handlers/handler_token.py
index 114b8a6..e1e4937 100644
--- a/src/handlers/handler_token.py
+++ b/src/handlers/handler_token.py
@@ -63,7 +63,10 @@ def _refresh_keys_if_needed(self) -> None:
             logger.debug("Token public keys are stale, refreshing now.")
             self.with_public_keys_queried()
         except RuntimeError:
-            logger.warning("Token public key refresh failed, using existing keys.")
+            logger.warning(
+                "Token public key refresh failed, using existing keys.",
+                extra={"public_key_count": len(self.public_keys)},
+            )
 
     def with_public_keys_queried(self) -> "HandlerToken":
         """Load token public keys from the configured URL.
@@ -92,12 +95,15 @@ def with_public_keys_queried(self) -> "HandlerToken":
             self.public_keys = [
                 cast(RSAPublicKey, serialization.load_der_public_key(base64.b64decode(raw_key))) for raw_key in raw_keys
             ]
-            logger.debug("Loaded %d token public keys.", len(self.public_keys))
+            logger.info("Loaded token public keys.", extra={"public_key_count": len(self.public_keys)})
             self._last_loaded_at = datetime.now(timezone.utc)
 
             return self
         except (requests.RequestException, ValueError, KeyError, UnsupportedAlgorithm) as exc:
-            logger.exception("Failed to fetch or deserialize token public key.")
+            logger.exception(
+                "Failed to fetch or deserialize token public key.",
+                extra={"public_keys_url": self.public_keys_url},
+            )
             raise RuntimeError("Token public key initialization failed") from exc
 
     def decode_jwt(self, token_encoded: str) -> dict[str, Any]:
@@ -111,12 +117,17 @@ def decode_jwt(self, token_encoded: str) -> dict[str, Any]:
         """
         self._refresh_keys_if_needed()
 
-        logger.debug("Decoding JWT.")
+        logger.debug("Decoding JWT.", extra={"public_key_count": len(self.public_keys)})
         for public_key in self.public_keys:
             try:
                 return jwt.decode(token_encoded, public_key, algorithms=["RS256"])
             except jwt.PyJWTError:
                 continue
+
+        logger.warning(
+            "JWT verification failed for all public keys.",
+            extra={"public_key_count": len(self.public_keys)},
+        )
         raise jwt.PyJWTError("Verification failed for all public keys")
 
     def get_token_provider_info(self) -> dict[str, Any]:
diff --git a/src/handlers/handler_topic.py b/src/handlers/handler_topic.py
index 09b6661..d241c40 100644
--- a/src/handlers/handler_topic.py
+++ b/src/handlers/handler_topic.py
@@ -19,6 +19,7 @@
 import json
 import logging
 import os
+import time
 from typing import Any
 
 import jwt
@@ -30,6 +31,7 @@
 from src.utils.conf_path import CONF_DIR
 from src.utils.config_loader import TopicAccessMap, TopicKeyMap, load_access_config, load_topic_keys_config
 from src.utils.constants import TOPIC_DLCHANGE, TOPIC_RUNS, TOPIC_STATUS_CHANGE, TOPIC_TEST
+from src.utils.observability import append_request_keys
 from src.utils.utils import build_error_response
 from src.writers.writer import WriteError, Writer
 
@@ -68,7 +70,7 @@ def with_load_topic_schemas(self) -> "HandlerTopic":
             The current instance with loaded topic schemas.
         """
         topic_schemas_dir = os.path.join(CONF_DIR, "topic_schemas")
-        logger.debug("Loading topic schemas from %s.", topic_schemas_dir)
+        logger.debug("Loading topic schemas.", extra={"topic_schemas_dir": topic_schemas_dir})
 
         with open(os.path.join(topic_schemas_dir, "runs.json"), "r", encoding="utf-8") as file:
             self.topics[TOPIC_RUNS] = json.load(file)
@@ -79,7 +81,7 @@ def with_load_topic_schemas(self) -> "HandlerTopic":
         with open(os.path.join(topic_schemas_dir, "status_change.json"), "r", encoding="utf-8") as file:
             self.topics[TOPIC_STATUS_CHANGE] = json.load(file)
 
-        logger.debug("Loaded topic schemas successfully.")
+        logger.debug("Loaded topic schemas.", extra={"topics": sorted(self.topics)})
         return self
 
     def with_load_topic_keys_config(self) -> "HandlerTopic":
@@ -109,17 +111,35 @@ def handle_request(self, event: dict[str, Any]) -> dict[str, Any]:
         Returns:
             API Gateway response.
         """
-        topic_name = event["pathParameters"]["topic_name"].lower()
+        path_parameters = event.get("pathParameters") or {}
+        raw_topic_name = path_parameters.get("topic_name")
+        if not raw_topic_name:
+            logger.warning("Request rejected: path parameter 'topic_name' is missing.")
+            return build_error_response(400, "validation", "Missing path parameter 'topic_name'.")
+
+        topic_name = str(raw_topic_name).lower()
+        append_request_keys(topic=topic_name)
         method = event.get("httpMethod")
 
         if method == "GET":
             return self._get_topic_schema(topic_name)
         if method == "POST":
+            try:
+                topic_message = json.loads(event.get("body") or "")
+            except (json.JSONDecodeError, TypeError):
+                logger.warning("Request rejected: message body is not valid JSON.")
+                return build_error_response(400, "validation", "Request body must be valid JSON.")
+            if not isinstance(topic_message, dict):
+                logger.warning("Request rejected: message body is not a JSON object.")
+                return build_error_response(400, "validation", "Request body must be a JSON object.")
+
             return self._post_topic_message(
                 topic_name,
-                json.loads(event["body"]),
+                topic_message,
                 self.handler_token.extract_token(event.get("headers", {})),
             )
+
+        logger.warning("Request rejected: unsupported HTTP method.")
         return build_error_response(404, "route", "Resource not found")
 
     def _get_topic_schema(self, topic_name: str) -> dict[str, Any]:
@@ -129,9 +149,10 @@ def _get_topic_schema(self, topic_name: str) -> dict[str, Any]:
         Returns:
             API Gateway response with topic schema or error.
         """
-        logger.debug("Handling GET TopicSchema(%s).", topic_name)
+        logger.debug("Handling GET topic schema.")
 
         if topic_name not in self.topics:
+            logger.warning("Request rejected: unknown topic.", extra={"known_topics": sorted(self.topics)})
             return build_error_response(404, "topic", f"Topic '{topic_name}' not found")
 
         return {
@@ -153,7 +174,7 @@ def _post_topic_message(self, topic_name: str, topic_message: dict[str, Any], to
             jwt.PyJWTError: If token decoding fails.
             ValidationError: If message validation fails.
         """
-        logger.debug("Handling POST TopicMessage(%s).", topic_name)
+        logger.debug("Handling POST topic message.")
 
         if not self.access_config:
             logger.error("Access configuration not loaded.")
@@ -161,46 +182,60 @@ def _post_topic_message(self, topic_name: str, topic_message: dict[str, Any], to
 
         try:
             token: dict[str, Any] = self.handler_token.decode_jwt(token_encoded)
-        except jwt.PyJWTError:  # type: ignore[attr-defined]
+        except jwt.PyJWTError as exc:  # type: ignore[attr-defined]
+            logger.warning(
+                "Request rejected: token verification failed.",
+                extra={"auth_error": type(exc).__name__, "token_present": bool(token_encoded)},
+            )
             return build_error_response(401, "auth", "Invalid or missing token")
 
         if topic_name not in self.topics:
+            logger.warning("Request rejected: unknown topic.", extra={"known_topics": sorted(self.topics)})
             return build_error_response(404, "topic", f"Topic '{topic_name}' not found")
 
         user = token.get("sub")
+        append_request_keys(user=user)
+
         authorized_user = self._resolve_authorized_user(topic_name, user)
         if authorized_user is None:
+            logger.warning("Request rejected: user is not authorized for the topic.")
             return build_error_response(403, "auth", f"User '{user}' is not authorized for topic '{topic_name}'")
 
+        # Log under the configured spelling of the user, since the token casing may differ.
+        append_request_keys(user=authorized_user)
+
         allowed, perm_error = self._validate_user_permissions(topic_name, authorized_user, topic_message)
         if not allowed:
+            logger.warning("Request rejected: user permissions do not allow the message.", extra={"reason": perm_error})
             return build_error_response(
                 403,
                 "permission",
                 perm_error or f"Permission denied for user '{authorized_user}' for POST to topic '{topic_name}'",
             )
 
-        logger.debug("Authorized user '%s' for topic '%s'.", authorized_user, topic_name)
+        logger.debug("User authorized for the topic.")
 
         try:
             validate(instance=topic_message, schema=self.topics[topic_name])
         except ValidationError as exc:
+            logger.warning(
+                "Request rejected: message does not match the topic schema.",
+                extra={"validation_path": str(exc.json_path), "validator": str(exc.validator)},
+            )
             return build_error_response(400, "validation", exc.message)
 
         message_key = self._resolve_message_key(topic_name, topic_message)
-        errors = []
-        for writer_name, writer in self.writers.items():
-            try:
-                writer.write(topic_name, topic_message, message_key)
-            except WriteError as exc:
-                errors.append({"type": writer_name, "message": str(exc)})
+        errors, written_by = self._write_to_all(topic_name, topic_message, message_key)
 
         if errors:
             logger.error(
-                "POST to topic '%s' failed: %d of %d writer(s) reported errors.",
-                topic_name,
-                len(errors),
-                len(self.writers),
+                "Message dispatch failed for at least one writer.",
+                extra={
+                    "writers_ok": written_by,
+                    "writers_failed": [error["type"] for error in errors],
+                    "writer_count": len(self.writers),
+                    "message_key": message_key,
+                },
             )
             return {
                 "statusCode": 500,
@@ -208,13 +243,56 @@ def _post_topic_message(self, topic_name: str, topic_message: dict[str, Any], to
                 "body": json.dumps({"success": False, "statusCode": 500, "errors": errors}),
             }
 
-        logger.info("Message accepted for topic '%s' from user '%s'.", topic_name, authorized_user)
+        logger.info("Message accepted.", extra={"writers_ok": written_by, "message_key": message_key})
         return {
             "statusCode": 202,
             "headers": {"Content-Type": "application/json"},
             "body": json.dumps({"success": True, "statusCode": 202}),
         }
 
+    def _write_to_all(
+        self,
+        topic_name: str,
+        topic_message: dict[str, Any],
+        message_key: str,
+    ) -> tuple[list[dict[str, str]], list[str]]:
+        """Dispatch a message to every configured writer, collecting per-writer outcomes.
+        Args:
+            topic_name: Target topic name.
+            topic_message: Message payload.
+            message_key: Resolved transport key.
+        Returns:
+            Tuple of (errors, written_by) where `errors` holds one entry per failing writer and
+            `written_by` lists the writers that accepted the message.
+        """
+        errors: list[dict[str, str]] = []
+        written_by: list[str] = []
+
+        for writer_name, writer in self.writers.items():
+            started_at = time.perf_counter()
+            try:
+                writer.write(topic_name, topic_message, message_key)
+                written_by.append(writer_name)
+                logger.debug(
+                    "Writer accepted the message.",
+                    extra={
+                        "writer": writer_name,
+                        "writer_duration_ms": round((time.perf_counter() - started_at) * 1000, 2),
+                    },
+                )
+            except WriteError as exc:
+                errors.append({"type": writer_name, "message": str(exc)})
+                logger.error(
+                    "Writer failed to publish the message.",
+                    extra={
+                        "writer": writer_name,
+                        "writer_duration_ms": round((time.perf_counter() - started_at) * 1000, 2),
+                        "writer_error": str(exc),
+                    },
+                )
+
+        return errors, written_by
+
     def _resolve_authorized_user(self, topic_name: str, user: str | None) -> str | None:
         """Match a token user to a configured user for a topic, ignoring case.
         Args:
@@ -273,11 +351,11 @@ def _resolve_message_key(self, topic_name: str, message: dict[str, Any]) -> str:
 
         key_value = message.get(key_field)
         if key_value is None:
-            logger.warning("Topic key field '%s' missing for topic '%s'.", key_field, topic_name)
+            logger.warning("Topic key field is missing from the message.", extra={"key_field": key_field})
             return ""
 
         if isinstance(key_value, (dict, list)):
-            logger.warning("Topic key field '%s' for topic '%s' is not scalar.", key_field, topic_name)
+            logger.warning("Topic key field is not a scalar value.", extra={"key_field": key_field})
             return ""
 
         return str(key_value)
diff --git a/src/readers/reader_postgres.py b/src/readers/reader_postgres.py
index 355f0b4..98444f8 100644
--- a/src/readers/reader_postgres.py
+++ b/src/readers/reader_postgres.py
@@ -113,6 +113,11 @@ def read_stats(
         ts_start = timestamp_start if timestamp_start is not None else (now_ms - POSTGRES_DEFAULT_WINDOW_MS)
         ts_end = timestamp_end if timestamp_end is not None else now_ms
 
+        started_at = time.perf_counter()
+        logger.debug(
+            "Running stats query.",
+            extra={"ts_start": ts_start, "ts_end": ts_end, "cursor": cursor, "limit": limit},
+        )
         try:
             col_names, raw_rows = self._execute_with_retry(
                 lambda conn: self._run_stats_query(conn, ts_start, ts_end, cursor, limit)
@@ -139,7 +144,14 @@ def read_stats(
             "limit": limit,
         }
 
-        logger.debug("Stats query returned %d rows.", len(rows))
+        logger.debug(
+            "Stats query returned rows.",
+            extra={
+                "row_count": len(rows),
+                "has_more": has_more,
+                "query_duration_ms": round((time.perf_counter() - started_at) * 1000, 2),
+            },
+        )
         return rows, pagination
 
     def _run_stats_query(
diff --git a/src/utils/conf_path.py b/src/utils/conf_path.py
index 0998987..62cb477 100644
--- a/src/utils/conf_path.py
+++ b/src/utils/conf_path.py
@@ -22,6 +22,7 @@
 4. Fallback to <project_root>/conf even if missing (subsequent file operations will raise)
 """
 
+import logging
 import os
 
 
@@ -72,4 +73,18 @@ def resolve_conf_dir(env_var: str = "CONF_DIR"):
 
 CONF_DIR, INVALID_CONF_ENV = resolve_conf_dir()
 
-__all__ = ["resolve_conf_dir", "CONF_DIR", "INVALID_CONF_ENV"]
+
+def log_configuration_source(target_logger: logging.Logger) -> None:
+    """Report which configuration directory the service resolved.
+    Args:
+        target_logger: Logger of the calling entry point.
+    """
+    target_logger.debug("Using configuration directory.", extra={"conf_dir": CONF_DIR})
+    if INVALID_CONF_ENV:
+        target_logger.warning(
+            "CONF_DIR env var points to a non-existent path; falling back.",
+            extra={"invalid_conf_dir": INVALID_CONF_ENV, "conf_dir": CONF_DIR},
+        )
+
+
+__all__ = ["resolve_conf_dir", "log_configuration_source", "CONF_DIR", "INVALID_CONF_ENV"]
diff --git a/src/utils/config_loader.py b/src/utils/config_loader.py
index 3093da4..fb59fdb 100644
--- a/src/utils/config_loader.py
+++ b/src/utils/config_loader.py
@@ -43,7 +43,7 @@ def load_config(conf_dir: str) -> dict[str, Any]:
     config_path = os.path.join(conf_dir, "config.json")
     with open(config_path, "r", encoding="utf-8") as file:
         config: dict[str, Any] = json.load(file)
-    logger.debug("Loaded main configuration from %s.", config_path)
+    logger.debug("Loaded main configuration.", extra={"config_path": config_path})
     return config
 
 
@@ -139,7 +139,7 @@ def load_access_config(config: dict[str, Any], aws_s3: ServiceResource) -> Topic
         Normalized mapping of topic names to per-user permission constraints.
     """
     access_path: str = config["access_config"]
-    logger.debug("Loading access configuration from %s.", access_path)
+    logger.debug("Loading access configuration.", extra={"access_path": access_path})
     access_data = _load_json_from_path(access_path, aws_s3)
 
     logger.debug("Loaded access configuration.")
@@ -174,7 +174,7 @@ def load_topic_keys_config(config: dict[str, Any], aws_s3: ServiceResource) -> T
         logger.debug("No topic_keys_config configured. Using empty topic key mapping.")
         return {}
 
-    logger.debug("Loading topic key configuration from %s.", topic_keys_path)
+    logger.debug("Loading topic key configuration.", extra={"topic_keys_path": topic_keys_path})
     topic_key_data = _load_json_from_path(topic_keys_path, aws_s3)
 
     logger.debug("Loaded topic key configuration.")
@@ -203,5 +203,5 @@ def load_topic_names(conf_dir: str) -> list[str]:
         if os.path.isfile(schema_path):
             topics.append(topic_name)
 
-    logger.debug("Discovered %d topic(s) from %s.", len(topics), schemas_dir)
+    logger.debug("Discovered topics.", extra={"topic_count": len(topics), "schemas_dir": schemas_dir})
     return topics
diff --git a/src/utils/logging_levels.py b/src/utils/logging_levels.py
index b21062f..957fb35 100644
--- a/src/utils/logging_levels.py
+++ b/src/utils/logging_levels.py
@@ -14,9 +14,12 @@
 # limitations under the License.
 #
 
-"""Custom logging levels and shared logging setup.
+"""Custom logging levels and log level resolution.
 
-Adds a TRACE level below DEBUG for very verbose payload logging.
+Adds a TRACE level below DEBUG for very verbose payload logging and resolves the
+configured level from the environment. The level is registered with the standard
+`logging` module, which is what makes `LOG_LEVEL=TRACE` resolvable by AWS Lambda
+Powertools as well.
 """
 
 from __future__ import annotations
@@ -44,24 +47,30 @@ def trace(self: logging.Logger, message: str, *args, **kws):  # type: ignore[ove
     logging.Logger.trace = trace  # type: ignore[attr-defined]
 
 
-def init_root_logger(module_name: str) -> logging.Logger:
-    """Initialize the root logger and return a module-level logger.
-    Args:
-        module_name: Typically `__name__` of the calling module.
+def configured_log_level() -> str:
+    """Read the log level requested through the environment.
     Returns:
-        A logger instance for the calling module.
+        The upper-cased level name, defaulting to `INFO`.
     """
-    root_logger = logging.getLogger()
-    if not root_logger.handlers:
-        root_logger.addHandler(logging.StreamHandler())
+    return (os.environ.get("POWERTOOLS_LOG_LEVEL") or os.environ.get("LOG_LEVEL") or "INFO").strip().upper()
 
-    log_level = os.environ.get("LOG_LEVEL", "INFO").upper()
-    try:
-        root_logger.setLevel(log_level)
-    except ValueError:
-        root_logger.setLevel("INFO")
-        root_logger.warning("Invalid LOG_LEVEL %s; defaulting to INFO.", log_level)
-    return logging.getLogger(module_name)
+
+def resolve_log_level() -> int:
+    """Resolve the configured log level to a numeric logging level.
+    Returns:
+        The numeric level for the configured name, or `logging.INFO` when it is unknown.
+    """
+    level = logging.getLevelName(configured_log_level())
+    return level if isinstance(level, int) else logging.INFO
+
+
+def invalid_log_level() -> str | None:
+    """Report a configured log level that could not be resolved.
+    Returns:
+        The rejected level name, or `None` when the configured level is valid.
+    """
+    configured = configured_log_level()
+    return None if isinstance(logging.getLevelName(configured), int) else configured
 
 
-__all__ = ["TRACE_LEVEL", "init_root_logger"]
+__all__ = ["TRACE_LEVEL", "configured_log_level", "invalid_log_level", "resolve_log_level"]
diff --git a/src/utils/observability.py b/src/utils/observability.py
new file mode 100644
index 0000000..8a161ec
--- /dev/null
+++ b/src/utils/observability.py
@@ -0,0 +1,188 @@
+#
+# Copyright 2026 ABSA Group Limited
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""Structured logging and request correlation for the EventGate lambdas.
+
+Wraps AWS Lambda Powertools `Logger` so that every log line is emitted as JSON
+enriched with the Lambda execution context and a request correlation id.
+Module loggers keep using `logging.getLogger(__name__)`; the Powertools handler
+is attached to the root logger, so their records are formatted and enriched too.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+import re
+from typing import Any
+
+from aws_lambda_powertools import Logger
+
+from src.utils.logging_levels import TRACE_LEVEL, configured_log_level, invalid_log_level, resolve_log_level
+
+DEFAULT_SERVICE_NAME = "eventgate"
+
+# Header accepted from callers so that an upstream request id can be carried through EventGate.
+CORRELATION_ID_HEADER = "x-correlation-id"
+REQUEST_ID_HEADER = "x-request-id"
+# Header returned to the caller on every response, so a client can quote a single id in a ticket.
+CORRELATION_ID_RESPONSE_HEADER = "X-Correlation-ID"
+
+# Inbound correlation ids end up in the log stream, so they must not carry newlines or control
+# characters (log injection) and must stay small enough not to bloat every log line.
+CORRELATION_ID_PATTERN = re.compile(r"^[A-Za-z0-9._:-]{1,128}$")
+
+# Third-party loggers that flood the log group when the service runs at DEBUG or TRACE.
+NOISY_LOGGERS = ("boto3", "botocore", "urllib3", "s3transfer", "aiosql", "confluent_kafka")
+
+logger: Logger = Logger(
+    service=os.environ.get("POWERTOOLS_SERVICE_NAME", DEFAULT_SERVICE_NAME),
+    level=resolve_log_level(),
+    use_rfc3339=True,
+    utc=True,
+    log_uncaught_exceptions=True,
+)
+
+# Keys appended for the duration of a single invocation; cleared at the start of the next one.
+_request_scoped_keys: set[str] = set()
+
+# Mutable container so the flag can be flipped without a module level `global` statement.
+_container_state: dict[str, bool] = {"cold_start": True}
+
+
+def configure_root_logging() -> None:
+    """Route every logger in the process through the Powertools JSON handler.
+
+    Replaces handlers installed by the Lambda runtime so that records are not emitted twice,
+    and caps noisy third-party loggers at WARNING so `LOG_LEVEL=DEBUG` stays readable.
+    """
+    log_level = resolve_log_level()
+
+    root_logger = logging.getLogger()
+    root_logger.handlers.clear()
+    root_logger.addHandler(logger.registered_handler)
+    root_logger.setLevel(log_level)
+
+    for noisy_logger in NOISY_LOGGERS:
+        logging.getLogger(noisy_logger).setLevel(max(log_level, logging.WARNING))
+
+
+def init_lambda_logging(module_name: str) -> logging.Logger:
+    """Configure logging for a lambda entry point.
+    Args:
+        module_name: Typically `__name__` of the calling entry point module.
+    Returns:
+        The module logger to use for the rest of the module.
+    """
+    configure_root_logging()
+    module_logger = logging.getLogger(module_name)
+
+    rejected_level = invalid_log_level()
+    if rejected_level:
+        module_logger.warning(
+            "Invalid log level configured; defaulting to INFO.",
+            extra={"log_level": rejected_level},
+        )
+
+    return module_logger
+
+
+def resolve_correlation_id(event: dict[str, Any]) -> str:
+    """Resolve the correlation id for an invocation.
+    Args:
+        event: API Gateway proxy event.
+    Returns:
+        The caller supplied correlation id when present and well-formed, otherwise the
+        API Gateway request id, otherwise an empty string.
+    """
+    headers = event.get("headers") or {}
+    lowered = {str(key).lower(): value for key, value in headers.items()}
+
+    for header in (CORRELATION_ID_HEADER, REQUEST_ID_HEADER):
+        candidate = lowered.get(header)
+        if isinstance(candidate, str) and CORRELATION_ID_PATTERN.fullmatch(candidate.strip()):
+            return candidate.strip()
+
+    request_context = event.get("requestContext") or {}
+    api_request_id = request_context.get("requestId")
+    return str(api_request_id) if api_request_id else ""
+
+
+def append_request_keys(**keys: Any) -> None:
+    """Append structured keys for the current invocation only.
+    Args:
+        **keys: Key/value pairs added to every subsequent log line of this invocation.
+    """
+    if not keys:
+        return
+    _request_scoped_keys.update(keys)
+    logger.append_keys(**keys)
+
+
+def _clear_request_keys() -> None:
+    """Drop keys appended during the previous invocation of a warm container."""
+    if _request_scoped_keys:
+        logger.remove_keys(list(_request_scoped_keys))
+        _request_scoped_keys.clear()
+
+
+def bind_request_context(event: dict[str, Any], context: Any = None) -> str:
+    """Bind Lambda and request context to the logger for the current invocation.
+    Args:
+        event: API Gateway proxy event.
+        context: AWS Lambda context object. May be `None` when invoked locally or from tests.
+    Returns:
+        The resolved correlation id (possibly an empty string).
+    """
+    _clear_request_keys()
+
+    correlation_id = resolve_correlation_id(event)
+    logger.set_correlation_id(correlation_id or None)
+
+    append_request_keys(
+        cold_start=_container_state["cold_start"],
+        resource=event.get("resource", ""),
+        http_method=event.get("httpMethod", ""),
+    )
+    _container_state["cold_start"] = False
+
+    if context is not None:
+        append_request_keys(
+            function_name=getattr(context, "function_name", None),
+            function_request_id=getattr(context, "aws_request_id", None),
+            function_memory_size=getattr(context, "memory_limit_in_mb", None),
+            function_arn=getattr(context, "invoked_function_arn", None),
+        )
+
+    return correlation_id
+
+
+__all__ = [
+    "CORRELATION_ID_HEADER",
+    "CORRELATION_ID_PATTERN",
+    "CORRELATION_ID_RESPONSE_HEADER",
+    "DEFAULT_SERVICE_NAME",
+    "NOISY_LOGGERS",
+    "REQUEST_ID_HEADER",
+    "TRACE_LEVEL",
+    "append_request_keys",
+    "bind_request_context",
+    "configure_root_logging",
+    "configured_log_level",
+    "init_lambda_logging",
+    "logger",
+    "resolve_correlation_id",
+]
diff --git a/src/utils/postgres_base.py b/src/utils/postgres_base.py
index b474026..9b1ae5a 100644
--- a/src/utils/postgres_base.py
+++ b/src/utils/postgres_base.py
@@ -127,7 +127,10 @@ def _get_connection(self) -> Any:
         if options:
             connect_kwargs["options"] = options
         self._connection = psycopg2.connect(**connect_kwargs)
-        logger.debug("New PostgreSQL connection established.")
+        logger.info(
+            "New PostgreSQL connection established.",
+            extra={"database": pg_config["database"], "host": pg_config["host"], "port": pg_config["port"]},
+        )
         return self._connection
 
     def _close_connection(self) -> None:
@@ -138,7 +141,7 @@ def _close_connection(self) -> None:
             try:
                 conn_to_close.close()
             except (PsycopgError, OSError):
-                logger.debug("Failed to close PostgreSQL connection.")
+                logger.debug("Failed to close PostgreSQL connection.", exc_info=True)
 
     def _execute_with_retry[T](self, operation: Callable[..., T], *, retry: bool = True) -> T:
         """Run `operation(connection)` with one retry on `OperationalError`.
@@ -161,5 +164,8 @@ def _execute_with_retry[T](self, operation: Callable[..., T], *, retry: bool = T
                 last_exc = exc
                 self._close_connection()
                 if attempt < max_attempts - 1:
-                    logger.warning("PostgreSQL connection lost, reconnecting.")
+                    logger.warning(
+                        "PostgreSQL connection lost, reconnecting.",
+                        extra={"attempt": attempt + 1, "max_attempts": max_attempts},
+                    )
         raise RuntimeError(f"Database connection failed after {max_attempts} attempts: {last_exc}") from last_exc
diff --git a/src/utils/utils.py b/src/utils/utils.py
index e19d494..09e22e0 100644
--- a/src/utils/utils.py
+++ b/src/utils/utils.py
@@ -18,11 +18,14 @@
 
 import json
 import logging
+import time
 from collections.abc import Callable
 from typing import Any
 
 import boto3
 
+from src.utils.observability import CORRELATION_ID_RESPONSE_HEADER, bind_request_context
+
 logger = logging.getLogger(__name__)
 
 
@@ -48,37 +51,63 @@ def build_error_response(status: int, err_type: str, message: str) -> dict[str,
     }
 
 
+def with_correlation_header(response: dict[str, Any], correlation_id: str) -> dict[str, Any]:
+    """Add the correlation id header to a response so callers can quote it when reporting issues.
+    Args:
+        response: API Gateway response dictionary.
+        correlation_id: Correlation id of the current request. Ignored when empty.
+    Returns:
+        The response with the correlation header set.
+    """
+    if not correlation_id or not isinstance(response, dict):
+        return response
+
+    headers = dict(response.get("headers") or {})
+    headers[CORRELATION_ID_RESPONSE_HEADER] = correlation_id
+    response["headers"] = headers
+    return response
+
+
 def dispatch_request(
     event: dict[str, Any],
     route_map: dict[str, Callable[..., dict[str, Any]]],
     request_logger: logging.Logger,
+    context: Any = None,
 ) -> dict[str, Any]:
     """Dispatch an API Gateway event to the matching route handler.
+    Binds the request context to the logger, logs the request outcome and stamps the response
+    with the correlation id.
     Args:
         event: API Gateway proxy event.
         route_map: Mapping of resource paths to handler callables.
         request_logger: Logger instance for error reporting.
+        context: AWS Lambda context object, when available.
     Returns:
         API Gateway response dictionary.
     """
+    started_at = time.perf_counter()
+    correlation_id = bind_request_context(event, context)
+    resource = str(event.get("resource", "")).lower()
+
     try:
-        resource = event.get("resource", "").lower()
         route_function = route_map.get(resource)
+        if route_function is None:
+            request_logger.warning("No route matched the requested resource.")
+            response = build_error_response(404, "route", "Resource not found.")
+        else:
+            request_logger.debug("Dispatching request to the route handler.")
+            response = route_function(event)
+    except Exception:  # pylint: disable=broad-exception-caught
+        # Boundary handler: any escaping exception must become a stable 500 with a logged cause,
+        # otherwise API Gateway returns an opaque 502 and the traceback is unstructured.
+        request_logger.exception("Unhandled error while processing the request.")
+        response = build_error_response(500, "internal", "Unexpected server error.")
+
+    duration_ms = round((time.perf_counter() - started_at) * 1000, 2)
+    status_code = response.get("statusCode") if isinstance(response, dict) else None
+    request_logger.info("Request completed.", extra={"status_code": status_code, "duration_ms": duration_ms})
 
-        if route_function:
-            return route_function(event)
-
-        return build_error_response(404, "route", "Resource not found.")
-    except (
-        KeyError,
-        json.JSONDecodeError,
-        ValueError,
-        AttributeError,
-        TypeError,
-        RuntimeError,
-    ) as request_exc:
-        request_logger.exception("Request processing error: %s.", request_exc)
-        return build_error_response(500, "internal", "Unexpected server error.")
+    return with_correlation_header(response, correlation_id)
 
 
 def load_postgres_config(secret_name: str, secret_region: str) -> dict[str, Any]:
diff --git a/src/writers/writer_eventbridge.py b/src/writers/writer_eventbridge.py
index 076f007..8fb8d9b 100644
--- a/src/writers/writer_eventbridge.py
+++ b/src/writers/writer_eventbridge.py
@@ -18,6 +18,7 @@
 
 import json
 import logging
+import time
 from typing import Any, Optional
 
 import boto3
@@ -68,8 +69,9 @@ def write(self, topic_name: str, message: dict[str, Any], message_key: str = "")
 
         log_payload_at_trace(logger, "EventBridge", topic_name, message)
 
+        started_at = time.perf_counter()
         try:
-            logger.debug("Sending to EventBridge %s.", topic_name)
+            logger.debug("Sending message to EventBridge.", extra={"topic": topic_name})
             response = self._client.put_events(
                 Entries=[
                     {
@@ -80,15 +82,40 @@ def write(self, topic_name: str, message: dict[str, Any], message_key: str = "")
                     }
                 ]
             )
+            duration_ms = round((time.perf_counter() - started_at) * 1000, 2)
             failed_count = response.get("FailedEntryCount", 0)
             if failed_count > 0:
                 self._entries = response.get("Entries", [])
                 failed_repr = self._format_failed_entries()
                 msg = f"{failed_count} EventBridge entries failed: {failed_repr}"
-                logger.error(msg)
+                logger.error(
+                    "EventBridge rejected entries.",
+                    extra={
+                        "topic": topic_name,
+                        "writer_duration_ms": duration_ms,
+                        "failed_entry_count": failed_count,
+                        "failed_entries": failed_repr,
+                    },
+                )
                 raise WriteError(msg)
+
+            entries = response.get("Entries") or [{}]
+            logger.debug(
+                "EventBridge accepted the message.",
+                extra={
+                    "topic": topic_name,
+                    "writer_duration_ms": duration_ms,
+                    "event_id": entries[0].get("EventId"),
+                },
+            )
         except (BotoCoreError, ClientError) as err:  # explicit AWS client-related errors
-            logger.exception("EventBridge put_events call failed.")
+            logger.exception(
+                "EventBridge put_events call failed.",
+                extra={
+                    "topic": topic_name,
+                    "writer_duration_ms": round((time.perf_counter() - started_at) * 1000, 2),
+                },
+            )
             raise WriteError(str(err)) from err
 
     def check_health(self) -> str | None:
diff --git a/src/writers/writer_kafka.py b/src/writers/writer_kafka.py
index 58a9ad3..f5ac563 100644
--- a/src/writers/writer_kafka.py
+++ b/src/writers/writer_kafka.py
@@ -109,20 +109,32 @@ def write(self, topic_name: str, message: dict[str, Any], message_key: str = "")
         log_payload_at_trace(logger, "Kafka", topic_name, message)
 
         errors: list[str] = []
-        has_exception = False
+        captured_exception: KafkaException | None = None
+        delivery: dict[str, Any] = {}
+        started_at = time.perf_counter()
+
+        def delivery_report(err: Any, msg: Any) -> None:
+            """Collect the Kafka delivery outcome for logging and error reporting."""
+            if err is not None:
+                errors.append(str(err))
+                return
+            try:
+                delivery.update({"kafka_partition": msg.partition(), "kafka_offset": msg.offset()})
+            except (AttributeError, TypeError):  # producer stubs may not expose message metadata
+                pass
 
         # Produce step
         try:
-            logger.debug("Sending to Kafka %s.", topic_name)
+            logger.debug("Sending message to Kafka.", extra={"topic": topic_name, "message_key": message_key})
             self._producer.produce(
                 topic_name,
                 key=message_key,
                 value=json.dumps(message).encode("utf-8"),
-                callback=lambda err, msg: (errors.append(str(err)) if err is not None else None),
+                callback=delivery_report,
             )
         except KafkaException as e:
             errors.append(f"Produce exception: {e}")
-            has_exception = True
+            captured_exception = e
 
         # Flush step (always attempted)
         remaining: int | None = None
@@ -131,28 +143,52 @@ def write(self, topic_name: str, message: dict[str, Any], message_key: str = "")
                 remaining = self._flush_with_timeout(_KAFKA_FLUSH_TIMEOUT_SEC)
             except KafkaException as e:
                 errors.append(f"Flush exception: {e}")
-                has_exception = True
+                captured_exception = e
 
             # Treat None (flush returns None in some stubs) as success equivalent to 0 pending
-            if (remaining is None or remaining == 0) and not errors:
+            if remaining is None or remaining == 0:
                 break
             if attempt < _MAX_RETRIES:
                 logger.warning(
-                    "Kafka flush pending (%s message(s) remain) attempt %d/%d.", remaining, attempt, _MAX_RETRIES
+                    "Kafka flush pending, retrying.",
+                    extra={
+                        "topic": topic_name,
+                        "pending_messages": remaining,
+                        "attempt": attempt,
+                        "max_attempts": _MAX_RETRIES,
+                    },
                 )
                 time.sleep(_RETRY_BACKOFF_SEC)
 
         # Warn if messages still pending after retries
         if isinstance(remaining, int) and remaining > 0:
             logger.warning(
-                "Kafka flush timeout after %ss: %d message(s) still pending.", _KAFKA_FLUSH_TIMEOUT_SEC, remaining
+                "Kafka flush timed out with messages still pending.",
+                extra={
+                    "topic": topic_name,
+                    "pending_messages": remaining,
+                    "flush_timeout_sec": _KAFKA_FLUSH_TIMEOUT_SEC,
+                },
             )
 
+        duration_ms = round((time.perf_counter() - started_at) * 1000, 2)
+
         if errors:
             failure_text = "Kafka writer failed: " + "; ".join(errors)
-            (logger.exception if has_exception else logger.error)(failure_text)
+            # logger.exception() is only valid inside an except block; outside it the traceback is
+            # taken from sys.exc_info() and would be empty, so pass the captured exception instead.
+            logger.error(
+                "Kafka writer failed.",
+                exc_info=captured_exception,
+                extra={"topic": topic_name, "writer_duration_ms": duration_ms, "writer_errors": errors},
+            )
             raise WriteError(failure_text)
 
+        logger.debug(
+            "Kafka accepted the message.",
+            extra={"topic": topic_name, "writer_duration_ms": duration_ms, **delivery},
+        )
+
     def check_health(self) -> str | None:
         """Check Kafka writer health.
         Returns:
diff --git a/src/writers/writer_postgres.py b/src/writers/writer_postgres.py
index 5eae61d..72a86fb 100644
--- a/src/writers/writer_postgres.py
+++ b/src/writers/writer_postgres.py
@@ -18,6 +18,7 @@
 
 import json
 import logging
+import time
 from dataclasses import dataclass
 from functools import cached_property
 from pathlib import Path
@@ -80,7 +81,6 @@ def _insert_dlchange(self, cursor: Any, message: dict[str, Any]) -> None:
             cursor: Database cursor.
             message: Event payload.
         """
-        logger.debug("Sending to Postgres - dlchange.")
         cursor.execute(
             self._queries.insert_dlchange,
             {
@@ -108,7 +108,6 @@ def _insert_run(self, cursor: Any, message: dict[str, Any]) -> None:
             cursor: Database cursor.
             message: Event payload (includes jobs array).
         """
-        logger.debug("Sending to Postgres - runs.")
         cursor.execute(
             self._queries.insert_run,
             {
@@ -122,6 +121,7 @@ def _insert_run(self, cursor: Any, message: dict[str, Any]) -> None:
                 "timestamp_end": message["timestamp_end"],
             },
         )
+        logger.debug("Inserting run jobs.", extra={"job_count": len(message["jobs"])})
         for job in message["jobs"]:
             cursor.execute(
                 self._queries.insert_run_job,
@@ -143,7 +143,6 @@ def _insert_test(self, cursor: Any, message: dict[str, Any]) -> None:
             cursor: Database cursor.
             message: Event payload.
         """
-        logger.debug("Sending to Postgres - test.")
         cursor.execute(
             self._queries.insert_test,
             {
@@ -167,21 +166,30 @@ def write(self, topic_name: str, message: dict[str, Any], message_key: str = "")
         Raises:
             WriteError: If publishing fails.
         """
+        # Checked first: a topic this writer does not persist must never fail the request because
+        # of a Postgres configuration problem it is not involved in.
+        if topic_name not in POSTGRES_WRITE_TOPICS:
+            logger.debug(
+                "Topic is not persisted by the Postgres writer - skipping.",
+                extra={"topic": topic_name, "postgres_topics": sorted(POSTGRES_WRITE_TOPICS)},
+            )
+            return
+
         try:
             pg_config = self._pg_config
         except (RuntimeError, BotoCoreError, ClientError, ValueError, KeyError) as e:
             err_msg = f"The Postgres writer failed with unknown error: {e!s}"
-            logger.exception(err_msg)
+            logger.exception("Postgres writer failed to load its configuration.", extra={"topic": topic_name})
             raise WriteError(err_msg) from e
 
         if not pg_config.get("database"):
-            logger.debug("No Postgres - skipping Postgres writer.")
+            logger.debug("No Postgres database configured - skipping Postgres writer.")
             return
 
         missing = [field for field in REQUIRED_CONNECTION_FIELDS if not pg_config.get(field)]
         if missing:
             msg = f"PostgreSQL connection field '{missing[0]}' not configured."
-            logger.error(msg)
+            logger.error("Postgres connection is not fully configured.", extra={"missing_fields": missing})
             raise WriteError(msg)
 
         if not self._is_psycopg2_available():
@@ -189,21 +197,19 @@ def write(self, topic_name: str, message: dict[str, Any], message_key: str = "")
 
         log_payload_at_trace(logger, "Postgres", topic_name, message)
 
-        if topic_name not in POSTGRES_WRITE_TOPICS:
-            msg = f"Unknown topic for Postgres/{topic_name}"
-            logger.debug(msg)
-            return  # no need to pollute the logs and no write should happen for these
-
         try:
             self._execute_with_retry(lambda conn: self._write_topic(conn, topic_name, message), retry=False)
         except (RuntimeError, PsycopgError, ValueError, KeyError) as e:
             self._close_connection()
             err_msg = f"The Postgres writer failed with unknown error: {e!s}"
-            logger.exception(err_msg)
+            logger.exception("Postgres writer failed while inserting the message.", extra={"topic": topic_name})
             raise WriteError(err_msg) from e
 
     def _write_topic(self, connection: Any, topic_name: str, message: dict[str, Any]) -> None:
         """Execute the insert for the given topic inside a transaction."""
+        started_at = time.perf_counter()
+        logger.debug("Inserting message into Postgres.", extra={"topic": topic_name})
+
         with connection.cursor() as cursor:
             if topic_name == TOPIC_DLCHANGE:
                 self._insert_dlchange(cursor, message)
@@ -213,6 +219,11 @@ def _write_topic(self, connection: Any, topic_name: str, message: dict[str, Any]
                 self._insert_test(cursor, message)
         connection.commit()
 
+        logger.debug(
+            "Postgres accepted the message.",
+            extra={"topic": topic_name, "writer_duration_ms": round((time.perf_counter() - started_at) * 1000, 2)},
+        )
+
     @staticmethod
     def _is_psycopg2_available() -> bool:
         """Check whether psycopg2 is importable."""
diff --git a/tests/integration/test_health_endpoint.py b/tests/integration/test_health_endpoint.py
index 2229454..d743b59 100644
--- a/tests/integration/test_health_endpoint.py
+++ b/tests/integration/test_health_endpoint.py
@@ -17,6 +17,7 @@
 
 import json
 
+from src.utils.observability import CORRELATION_ID_RESPONSE_HEADER
 from tests.integration.conftest import EventGateTestClient
 
 
@@ -44,3 +45,26 @@ def test_get_health_includes_uptime(self, eventgate_client: EventGateTestClient)
         assert "uptime_seconds" in body
         assert isinstance(body["uptime_seconds"], (int, float))
         assert body["uptime_seconds"] >= 0
+
+
+class TestCorrelationId:
+    """Tests for the request correlation id contract."""
+
+    def test_caller_correlation_id_is_returned(self, eventgate_client: EventGateTestClient) -> None:
+        """Test a caller supplied correlation id is echoed back in the response headers."""
+        response = eventgate_client.invoke("/health", "GET", headers={"X-Correlation-ID": "run-42"})
+
+        assert "run-42" == response["headers"][CORRELATION_ID_RESPONSE_HEADER]
+
+    def test_malformed_correlation_id_is_not_echoed(self, eventgate_client: EventGateTestClient) -> None:
+        """Test a malformed correlation id is rejected instead of being written back."""
+        response = eventgate_client.invoke("/health", "GET", headers={"X-Correlation-ID": "bad value\ninjected"})
+
+        assert CORRELATION_ID_RESPONSE_HEADER not in response["headers"]
+
+    def test_error_responses_carry_the_correlation_id(self, eventgate_client: EventGateTestClient) -> None:
+        """Test the correlation id is present on error responses too."""
+        response = eventgate_client.invoke("/unknown", "GET", headers={"X-Correlation-ID": "run-43"})
+
+        assert 404 == response["statusCode"]
+        assert "run-43" == response["headers"][CORRELATION_ID_RESPONSE_HEADER]
diff --git a/tests/unit/handlers/test_handler_topic.py b/tests/unit/handlers/test_handler_topic.py
index f82035f..2a0ce59 100644
--- a/tests/unit/handlers/test_handler_topic.py
+++ b/tests/unit/handlers/test_handler_topic.py
@@ -15,6 +15,7 @@
 #
 
 import json
+import logging
 import re
 from unittest.mock import patch, mock_open, MagicMock
 
@@ -434,3 +435,133 @@ def test_post_permission_denied(
         body = json.loads(resp["body"])
         assert "permission" == body["errors"][0]["type"]
         assert expected_fragment in body["errors"][0]["message"]
+
+
+## request rejection logging
+def logged_messages(caplog, level):
+    """Collect log messages captured at a single level."""
+    return [record.message for record in caplog.records if record.levelno == level]
+
+
+def test_post_missing_topic_path_parameter_is_rejected(event_gate_module, make_event, caplog):
+    caplog.set_level(logging.WARNING)
+    event = make_event("/topics/{topic_name}", method="POST", body={"a": 1})
+
+    resp = event_gate_module.lambda_handler(event)
+
+    assert 400 == resp["statusCode"]
+    assert "Request rejected: path parameter 'topic_name' is missing." in logged_messages(caplog, logging.WARNING)
+
+
+def test_post_non_object_body_is_rejected(event_gate_module, make_event, caplog):
+    caplog.set_level(logging.WARNING)
+    event = make_event("/topics/{topic_name}", method="POST", topic="public.cps.za.test", body="[1, 2]")
+
+    resp = event_gate_module.lambda_handler(event)
+
+    assert 400 == resp["statusCode"]
+    assert "Request rejected: message body is not a JSON object." in logged_messages(caplog, logging.WARNING)
+
+
+def test_unsupported_method_is_rejected(event_gate_module, make_event, caplog):
+    caplog.set_level(logging.WARNING)
+    event = make_event("/topics/{topic_name}", method="PUT", topic="public.cps.za.test")
+
+    resp = event_gate_module.lambda_handler(event)
+
+    assert 404 == resp["statusCode"]
+    assert "Request rejected: unsupported HTTP method." in logged_messages(caplog, logging.WARNING)
+
+
+def test_invalid_token_is_logged(event_gate_module, make_event, valid_payload, caplog):
+    caplog.set_level(logging.WARNING)
+    with patch.object(event_gate_module.handler_token, "decode_jwt", side_effect=jwt.PyJWTError("nope")):
+        event = make_event(
+            "/topics/{topic_name}",
+            method="POST",
+            topic="public.cps.za.test",
+            body=valid_payload,
+            headers={"Authorization": "Bearer token"},
+        )
+        resp = event_gate_module.lambda_handler(event)
+
+    assert 401 == resp["statusCode"]
+    assert "Request rejected: token verification failed." in logged_messages(caplog, logging.WARNING)
+
+
+def test_unauthorized_user_is_logged(event_gate_module, make_event, valid_payload, caplog):
+    caplog.set_level(logging.WARNING)
+    with patch.object(event_gate_module.handler_token, "decode_jwt", return_value={"sub": "UnknownUser"}):
+        event = make_event(
+            "/topics/{topic_name}",
+            method="POST",
+            topic="public.cps.za.test",
+            body=valid_payload,
+            headers={"Authorization": "Bearer token"},
+        )
+        resp = event_gate_module.lambda_handler(event)
+
+    assert 403 == resp["statusCode"]
+    assert "Request rejected: user is not authorized for the topic." in logged_messages(caplog, logging.WARNING)
+
+
+def test_schema_validation_failure_is_logged(event_gate_module, make_event, caplog):
+    caplog.set_level(logging.WARNING)
+    with patch.object(event_gate_module.handler_token, "decode_jwt", return_value={"sub": "TestUser"}):
+        event_gate_module.handler_topic.access_config["public.cps.za.test"] = {"TestUser": {}}
+        event = make_event(
+            "/topics/{topic_name}",
+            method="POST",
+            topic="public.cps.za.test",
+            body={"event_id": "e1"},
+            headers={"Authorization": "Bearer token"},
+        )
+        resp = event_gate_module.lambda_handler(event)
+
+    assert 400 == resp["statusCode"]
+    assert "Request rejected: message does not match the topic schema." in logged_messages(caplog, logging.WARNING)
+
+
+def test_partial_writer_failure_reports_both_sides(event_gate_module, make_event, valid_payload, caplog):
+    caplog.set_level(logging.ERROR)
+    with patch.object(event_gate_module.handler_token, "decode_jwt", return_value={"sub": "TestUser"}):
+        event_gate_module.handler_topic.access_config["public.cps.za.test"] = {"TestUser": {}}
+        for name, writer in event_gate_module.handler_topic.writers.items():
+            writer.write = MagicMock(side_effect=WriteError("down") if name == "postgres" else None)
+
+        event = make_event(
+            "/topics/{topic_name}",
+            method="POST",
+            topic="public.cps.za.test",
+            body=valid_payload,
+            headers={"Authorization": "Bearer token"},
+        )
+        resp = event_gate_module.lambda_handler(event)
+
+    dispatch_failures = [r for r in caplog.records if r.message == "Message dispatch failed for at least one writer."]
+    assert 500 == resp["statusCode"]
+    assert 1 == len(dispatch_failures)
+    assert ["postgres"] == dispatch_failures[0].writers_failed
+    assert ["eventbridge", "kafka"] == sorted(dispatch_failures[0].writers_ok)
+
+
+def test_accepted_message_is_logged(event_gate_module, make_event, valid_payload, caplog):
+    caplog.set_level(logging.INFO)
+    with patch.object(event_gate_module.handler_token, "decode_jwt", return_value={"sub": "TestUser"}):
+        event_gate_module.handler_topic.access_config["public.cps.za.test"] = {"TestUser": {}}
+        for writer in event_gate_module.handler_topic.writers.values():
+            writer.write = MagicMock(return_value=None)
+
+        event = make_event(
+            "/topics/{topic_name}",
+            method="POST",
+            topic="public.cps.za.test",
+            body=valid_payload,
+            headers={"Authorization": "Bearer token"},
+        )
+        resp = event_gate_module.lambda_handler(event)
+
+    accepted = [r for r in caplog.records if r.message == "Message accepted."]
+    assert 202 == resp["statusCode"]
+    assert 1 == len(accepted)
+    assert ["eventbridge", "kafka", "postgres"] == sorted(accepted[0].writers_ok)
diff --git a/tests/unit/test_event_gate_lambda.py b/tests/unit/test_event_gate_lambda.py
index 142f18f..14a5056 100644
--- a/tests/unit/test_event_gate_lambda.py
+++ b/tests/unit/test_event_gate_lambda.py
@@ -54,7 +54,7 @@ def test_internal_error_path(event_gate_module, make_event):
 
 
 def test_post_invalid_json_body(event_gate_module, make_event):
-    # Invalid JSON triggers json.loads exception and returns 500 internal error
+    # Invalid JSON is rejected as a client error, not swallowed as an internal error
     event = make_event(
         "/topics/{topic_name}",
         method="POST",
@@ -63,9 +63,9 @@ def test_post_invalid_json_body(event_gate_module, make_event):
         headers={"Authorization": "Bearer token"},
     )
     resp = event_gate_module.lambda_handler(event)
-    assert 500 == resp["statusCode"]
+    assert 400 == resp["statusCode"]
     body = json.loads(resp["body"])
-    assert any(e["type"] == "internal" for e in body["errors"])  # internal error path
+    assert any(e["type"] == "validation" for e in body["errors"])  # validation error path
 
 
 def test_boto3_s3_client_default_ssl_verification():
diff --git a/tests/unit/utils/test_conf_path.py b/tests/unit/utils/test_conf_path.py
index 7098b0c..183e614 100644
--- a/tests/unit/utils/test_conf_path.py
+++ b/tests/unit/utils/test_conf_path.py
@@ -39,7 +39,7 @@ def test_env_var_invalid_directory_falls_back_parent(monkeypatch):
     conf_dir, invalid = conf_path_module.resolve_conf_dir()
     # Should fall back to repository conf directory /conf
     expected_root_conf = (Path(conf_path_module.__file__).resolve().parent.parent.parent / "conf").resolve()
-    assert Path(conf_dir).resolve() == expected_root_conf
+    assert expected_root_conf == Path(conf_dir).resolve()
     assert invalid == os.path.abspath(missing_path)
 
 
@@ -78,7 +78,8 @@ def build(base: Path, code: str):
     mod = _load_isolated_conf_path(build)
     try:
         conf_dir, invalid = mod.resolve_conf_dir()
-        assert conf_dir.endswith("pkg/conf")  # current directory conf chosen
+        expected_current_conf = (Path(mod.__file__).resolve().parent / "conf").resolve()
+        assert expected_current_conf == Path(conf_dir).resolve()  # current directory conf chosen
         assert invalid is None
     finally:
         mod._tmp.cleanup()  # type: ignore[attr-defined]
@@ -97,7 +98,7 @@ def build(base: Path, code: str):
         conf_dir, invalid = mod.resolve_conf_dir()
         # Parent conf path returned even though it does not exist
         expected_root_conf = (Path(mod.__file__).resolve().parent.parent.parent / "conf").resolve()
-        assert Path(conf_dir).resolve() == expected_root_conf
+        assert expected_root_conf == Path(conf_dir).resolve()
         assert invalid is None
     finally:
         mod._tmp.cleanup()  # type: ignore[attr-defined]
@@ -118,7 +119,8 @@ def build(base: Path, code: str):
         bad_path = "/definitely/not/there/abc123"
         monkeypatch.setenv("CONF_DIR", bad_path)
         conf_dir, invalid = mod.resolve_conf_dir()
-        assert conf_dir.endswith("pkg_invalid_current/conf")
+        expected_current_conf = (Path(mod.__file__).resolve().parent / "conf").resolve()
+        assert expected_current_conf == Path(conf_dir).resolve()
         assert invalid == os.path.abspath(bad_path)
     finally:
         mod._tmp.cleanup()  # type: ignore[attr-defined]
@@ -139,7 +141,7 @@ def build(base: Path, code: str):
         monkeypatch.setenv("CONF_DIR", bad_path)
         conf_dir, invalid = mod.resolve_conf_dir()
         expected_root_conf = (Path(mod.__file__).resolve().parent.parent.parent / "conf").resolve()
-        assert Path(conf_dir).resolve() == expected_root_conf
+        assert expected_root_conf == Path(conf_dir).resolve()
         assert invalid == os.path.abspath(bad_path)
     finally:
         mod._tmp.cleanup()  # type: ignore[attr-defined]
diff --git a/tests/unit/utils/test_logging_levels.py b/tests/unit/utils/test_logging_levels.py
new file mode 100644
index 0000000..416cce9
--- /dev/null
+++ b/tests/unit/utils/test_logging_levels.py
@@ -0,0 +1,71 @@
+#
+# Copyright 2026 ABSA Group Limited
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+import logging
+
+import pytest
+from aws_lambda_powertools import Logger
+
+from src.utils.logging_levels import TRACE_LEVEL, configured_log_level, invalid_log_level, resolve_log_level
+
+
+@pytest.fixture(autouse=True)
+def clear_log_level_env(monkeypatch):
+    """Start every test from an unset log level configuration."""
+    monkeypatch.delenv("LOG_LEVEL", raising=False)
+    monkeypatch.delenv("POWERTOOLS_LOG_LEVEL", raising=False)
+
+
+## configured_log_level()
+def test_configured_log_level_defaults_to_info():
+    assert "INFO" == configured_log_level()
+
+
+def test_powertools_log_level_takes_precedence(monkeypatch):
+    monkeypatch.setenv("LOG_LEVEL", "DEBUG")
+    monkeypatch.setenv("POWERTOOLS_LOG_LEVEL", "warning")
+
+    assert "WARNING" == configured_log_level()
+
+
+## resolve_log_level()
+@pytest.mark.parametrize(
+    "configured,expected",
+    [("TRACE", TRACE_LEVEL), ("debug", logging.DEBUG), ("INFO", logging.INFO), ("ERROR", logging.ERROR)],
+)
+def test_resolve_log_level_known_levels(monkeypatch, configured, expected):
+    monkeypatch.setenv("LOG_LEVEL", configured)
+
+    assert expected == resolve_log_level()
+    assert invalid_log_level() is None
+
+
+def test_resolve_log_level_falls_back_to_info(monkeypatch):
+    monkeypatch.setenv("LOG_LEVEL", "NOT_A_LEVEL")
+
+    assert logging.INFO == resolve_log_level()
+    assert "NOT_A_LEVEL" == invalid_log_level()
+
+
+## Powertools compatibility
+def test_powertools_logger_accepts_the_custom_trace_level(monkeypatch, capsys):
+    monkeypatch.setenv("LOG_LEVEL", "TRACE")
+
+    powertools_logger = Logger(service="eventgate-trace-check", level=resolve_log_level())
+    powertools_logger.trace("Payload.")
+
+    assert TRACE_LEVEL == powertools_logger.log_level
+    assert '"level":"TRACE"' in capsys.readouterr().out
diff --git a/tests/unit/utils/test_observability.py b/tests/unit/utils/test_observability.py
new file mode 100644
index 0000000..0af76bf
--- /dev/null
+++ b/tests/unit/utils/test_observability.py
@@ -0,0 +1,168 @@
+#
+# Copyright 2026 ABSA Group Limited
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+import json
+import logging
+
+import pytest
+
+from src.utils import observability
+from src.utils.logging_levels import TRACE_LEVEL
+from src.utils.observability import (
+    append_request_keys,
+    bind_request_context,
+    configure_root_logging,
+    logger,
+    resolve_correlation_id,
+)
+
+
+class FakeLambdaContext:
+    function_name = "eventgate"
+    aws_request_id = "aws-request-id-1"
+    memory_limit_in_mb = 512
+    invoked_function_arn = "arn:aws:lambda:eu-west-1:000000000000:function:eventgate"
+
+
+def formatted(record):
+    """Render a captured record the way the lambda emits it, so persistent keys are visible."""
+    return json.loads(logger.registered_formatter.format(record))
+
+
+@pytest.fixture(autouse=True)
+def clean_logger_state():
+    """Keep the module level logger state from leaking between tests."""
+    observability._container_state["cold_start"] = True
+    yield
+    bind_request_context({})
+    logger.set_correlation_id(None)
+
+
+## resolve_correlation_id()
+def test_correlation_id_from_correlation_header():
+    event = {"headers": {"X-Correlation-ID": "run-42_a.b:c"}}
+
+    assert "run-42_a.b:c" == resolve_correlation_id(event)
+
+
+def test_correlation_id_header_lookup_is_case_insensitive():
+    event = {"headers": {"x-CORRELATION-id": " run-42 "}}
+
+    assert "run-42" == resolve_correlation_id(event)
+
+
+def test_correlation_id_from_request_id_header():
+    event = {"headers": {"X-Request-ID": "req-7"}}
+
+    assert "req-7" == resolve_correlation_id(event)
+
+
+@pytest.mark.parametrize(
+    "value",
+    ["with space", "line\nbreak", "", "!@#$", "x" * 129],
+)
+def test_correlation_id_rejects_malformed_header(value):
+    event = {"headers": {"X-Correlation-ID": value}, "requestContext": {"requestId": "apigw-1"}}
+
+    assert "apigw-1" == resolve_correlation_id(event)
+
+
+def test_correlation_id_falls_back_to_api_gateway_request_id():
+    event = {"requestContext": {"requestId": "apigw-1"}}
+
+    assert "apigw-1" == resolve_correlation_id(event)
+
+
+def test_correlation_id_empty_when_unavailable():
+    assert "" == resolve_correlation_id({})
+
+
+## bind_request_context()
+def test_bind_request_context_binds_correlation_id_and_request_keys(caplog):
+    caplog.set_level(logging.INFO)
+    event = {"headers": {"X-Correlation-ID": "run-42"}, "resource": "/topics/{topic_name}", "httpMethod": "POST"}
+
+    correlation_id = bind_request_context(event, FakeLambdaContext())
+    logging.getLogger("src.handlers.handler_topic").info("Message accepted.")
+
+    payload = formatted(caplog.records[-1])
+    assert "run-42" == correlation_id
+    assert "run-42" == payload["correlation_id"]
+    assert "/topics/{topic_name}" == payload["resource"]
+    assert "POST" == payload["http_method"]
+    assert "aws-request-id-1" == payload["function_request_id"]
+    assert payload["cold_start"] is True
+
+
+def test_bind_request_context_reports_cold_start_once(caplog):
+    caplog.set_level(logging.INFO)
+
+    bind_request_context({}, FakeLambdaContext())
+    logging.getLogger("src.utils.utils").info("Request completed.")
+    first = formatted(caplog.records[-1])
+
+    bind_request_context({}, FakeLambdaContext())
+    logging.getLogger("src.utils.utils").info("Request completed.")
+    second = formatted(caplog.records[-1])
+
+    assert first["cold_start"] is True
+    assert second["cold_start"] is False
+
+
+def test_bind_request_context_tolerates_missing_lambda_context(caplog):
+    caplog.set_level(logging.INFO)
+
+    bind_request_context({"resource": "/health"}, None)
+    logging.getLogger("src.utils.utils").info("Request completed.")
+
+    payload = formatted(caplog.records[-1])
+    assert "/health" == payload["resource"]
+    assert "function_request_id" not in payload
+
+
+def test_request_keys_do_not_leak_into_the_next_invocation(caplog):
+    caplog.set_level(logging.INFO)
+
+    bind_request_context({"resource": "/topics/{topic_name}"}, None)
+    append_request_keys(topic="test", user="someone")
+    logging.getLogger("src.handlers.handler_topic").info("Message accepted.")
+    assert "test" == formatted(caplog.records[-1])["topic"]
+
+    bind_request_context({"resource": "/health"}, None)
+    logging.getLogger("src.handlers.handler_health").info("Health check passed.")
+
+    payload = formatted(caplog.records[-1])
+    assert "topic" not in payload
+    assert "user" not in payload
+
+
+## configure_root_logging()
+def test_configure_root_logging_routes_root_through_powertools_handler():
+    configure_root_logging()
+
+    root_logger = logging.getLogger()
+    assert [logger.registered_handler] == root_logger.handlers
+
+
+def test_configure_root_logging_caps_noisy_loggers(monkeypatch):
+    monkeypatch.setenv("LOG_LEVEL", "TRACE")
+    monkeypatch.delenv("POWERTOOLS_LOG_LEVEL", raising=False)
+
+    configure_root_logging()
+
+    assert TRACE_LEVEL == logging.getLogger().level
+    for noisy_logger in observability.NOISY_LOGGERS:
+        assert logging.WARNING == logging.getLogger(noisy_logger).level
diff --git a/tests/unit/utils/test_utils.py b/tests/unit/utils/test_utils.py
index cc59bf4..eaed1b0 100644
--- a/tests/unit/utils/test_utils.py
+++ b/tests/unit/utils/test_utils.py
@@ -15,8 +15,23 @@
 #
 
 import json
+import logging
 
-from src.utils.utils import build_error_response
+import pytest
+
+from src.utils.observability import CORRELATION_ID_RESPONSE_HEADER
+from src.utils.utils import build_error_response, dispatch_request
+
+logger = logging.getLogger(__name__)
+
+
+def make_event(resource="/health", method="GET", correlation_id="run-42"):
+    """Build a minimal API Gateway proxy event."""
+    return {
+        "resource": resource,
+        "httpMethod": method,
+        "headers": {"X-Correlation-ID": correlation_id},
+    }
 
 
 ## build_error_response()
@@ -33,3 +48,65 @@ def test_build_error_response_structure():
     assert 1 == len(body["errors"])
     assert "topic" == body["errors"][0]["type"]
     assert "Topic not found" == body["errors"][0]["message"]
+
+
+## dispatch_request()
+def test_dispatch_request_routes_and_stamps_the_correlation_id():
+    route_map = {"/health": lambda _: {"statusCode": 200, "headers": {"Content-Type": "application/json"}}}
+
+    resp = dispatch_request(make_event(), route_map, logger)
+
+    assert 200 == resp["statusCode"]
+    assert "run-42" == resp["headers"][CORRELATION_ID_RESPONSE_HEADER]
+
+
+def test_dispatch_request_overwrites_a_handler_supplied_correlation_id():
+    """The id bound to the request wins over one a handler put on its own response."""
+    route_map = {"/health": lambda _: {"statusCode": 200, "headers": {CORRELATION_ID_RESPONSE_HEADER: "stale-id"}}}
+
+    resp = dispatch_request(make_event(), route_map, logger)
+
+    assert "run-42" == resp["headers"][CORRELATION_ID_RESPONSE_HEADER]
+
+
+def test_dispatch_request_logs_the_request_outcome(caplog):
+    caplog.set_level(logging.INFO)
+    route_map = {"/health": lambda _: {"statusCode": 503, "headers": {}}}
+
+    dispatch_request(make_event(), route_map, logger)
+
+    completed = [record for record in caplog.records if record.message == "Request completed."]
+    assert 1 == len(completed)
+    assert 503 == completed[0].status_code
+    assert completed[0].duration_ms >= 0
+
+
+def test_dispatch_request_warns_on_unknown_route(caplog):
+    caplog.set_level(logging.WARNING)
+
+    resp = dispatch_request(make_event(resource="/unknown"), {}, logger)
+
+    assert 404 == resp["statusCode"]
+    assert any(record.message == "No route matched the requested resource." for record in caplog.records)
+
+
+@pytest.mark.parametrize("error", [RuntimeError("boom"), OSError("connection reset"), IndexError("out of range")])
+def test_dispatch_request_converts_any_handler_error_into_a_logged_500(caplog, error):
+    caplog.set_level(logging.ERROR)
+
+    def failing_route(_event):
+        raise error
+
+    resp = dispatch_request(make_event(), {"/health": failing_route}, logger)
+
+    assert 500 == resp["statusCode"]
+    assert "internal" == json.loads(resp["body"])["errors"][0]["type"]
+    assert "run-42" == resp["headers"][CORRELATION_ID_RESPONSE_HEADER]
+    assert any(record.message == "Unhandled error while processing the request." for record in caplog.records)
+
+
+def test_dispatch_request_does_not_swallow_system_exit():
+    route_map = {"/terminate": lambda _: (_ for _ in ()).throw(SystemExit("TERMINATING"))}
+
+    with pytest.raises(SystemExit):
+        dispatch_request(make_event(resource="/terminate"), route_map, logger)
diff --git a/tests/unit/writers/test_writer_kafka.py b/tests/unit/writers/test_writer_kafka.py
index bce2baf..7a95d60 100644
--- a/tests/unit/writers/test_writer_kafka.py
+++ b/tests/unit/writers/test_writer_kafka.py
@@ -140,8 +140,8 @@ def test_write_flush_retries_until_success(monkeypatch, caplog):
     # It should break as soon as remaining == 0 (after flush call returning 0)
     assert producer.flush_calls == 5  # sequence consumed until 0
     # Warnings logged for attempts before success (flush_calls -1) because last attempt didn't warn
-    warn_messages = [r.message for r in caplog.records if r.levelno == logging.WARNING]
-    assert any("attempt 1" in m or "attempt 2" in m for m in warn_messages)
+    warn_records = [r for r in caplog.records if r.levelno == logging.WARNING]
+    assert [1, 2, 3, 4] == [r.attempt for r in warn_records]
 
 
 def test_write_timeout_warning_when_remaining_after_retries(monkeypatch, caplog):
@@ -152,9 +152,10 @@ def test_write_timeout_warning_when_remaining_after_retries(monkeypatch, caplog)
     writer._producer = producer
     writer.write("topic", {"f": 6})
     timeout_warnings = [
-        r.message for r in caplog.records if "timeout" in r.message
-    ]  # final warning should mention timeout
+        r for r in caplog.records if r.message == "Kafka flush timed out with messages still pending."
+    ]  # final warning reports the pending messages
     assert timeout_warnings, "Expected timeout warning logged"
+    assert 2 == timeout_warnings[0].pending_messages
     assert producer.flush_calls == 3  # retried 3 times
 
 
diff --git a/tests/unit/writers/test_writer_postgres.py b/tests/unit/writers/test_writer_postgres.py
index c7ae19d..dc6aee2 100644
--- a/tests/unit/writers/test_writer_postgres.py
+++ b/tests/unit/writers/test_writer_postgres.py
@@ -228,6 +228,19 @@ def test_write_unknown_topic_skips_silently(reset_env, monkeypatch):
     assert 0 == len(store)
 
 
+def test_write_unknown_topic_ignores_broken_postgres_configuration(reset_env, monkeypatch):
+    """A topic Postgres does not persist must not fail because Postgres itself is misconfigured."""
+    writer = WriterPostgres({})
+    monkeypatch.setattr(
+        type(writer),
+        "_pg_config",
+        property(lambda self: (_ for _ in ()).throw(RuntimeError("secret unavailable"))),
+    )
+    monkeypatch.setattr(pb, "psycopg2", None)
+
+    writer.write("public.cps.za.unknown", {})
+
+
 def test_write_success_known_topic(reset_env, monkeypatch):
     store = []
     monkeypatch.setattr(pb, "psycopg2", DummyPsycopg(store))