diff --git a/faststream_outbox/broker.py b/faststream_outbox/broker.py
index 1441f80..e660171 100644
--- a/faststream_outbox/broker.py
+++ b/faststream_outbox/broker.py
@@ -10,6 +10,7 @@
 import datetime as _dt
 import logging
 import typing
+import warnings
 from collections.abc import Iterable, Sequence
 from types import TracebackType
 
@@ -42,6 +43,8 @@
 
 
 if typing.TYPE_CHECKING:
+    from weakref import WeakSet
+
     from fast_depends.dependencies import Dependant
     from fast_depends.library.serializer import SerializerProto
     from faststream._internal.context.repository import ContextRepo
@@ -121,7 +124,9 @@ class OutboxBroker(
 ):
     """FastStream broker backed by a Postgres outbox table."""
 
-    _subscribers: list["OutboxSubscriber"]
+    # P25: the runtime container is a WeakSet (set by the upstream Registrator), not a
+    # list — annotate it accurately so the _subscribers/subscribers distinction is clear.
+    _subscribers: "WeakSet[OutboxSubscriber]"
 
     def __init__(  # noqa: PLR0913
         self,
@@ -212,6 +217,11 @@ async def _connect(self) -> "AsyncEngine":
 
     @typing.override
     async def __aenter__(self) -> typing.Self:
+        # Upstream equivalent (replaced):
+        #   BrokerUsecase.__aenter__ -> faststream/_internal/broker/broker.py
+        # Upstream's __aenter__ only connects; we upgrade to a full start() so the
+        # subscriber loops spin up under `async with broker:`. Re-check this if upstream
+        # changes __aenter__/__aexit__ pairing (P26).
         await self.start()
         return self
 
@@ -220,6 +230,32 @@ async def start(self) -> None:
         await self.connect()
         await super().start()
         self._warn_on_unstarted_foreign_publishers()
+        self._warn_on_duplicate_queues()
+
+    def _warn_on_duplicate_queues(self) -> None:
+        # P22: the registration-time overlap warning only sees the broker's own
+        # _subscribers, so duplicates introduced via include_router slip through. Re-check
+        # at start() over the full ``subscribers`` property — but warn only for queues a
+        # router contributed to, since same-broker overlaps were already flagged at
+        # registration time (avoids double-warning one mistake).
+        direct = set(self._subscribers)
+        counts: dict[str, int] = {}
+        router_queues: set[str] = set()
+        for sub in self.subscribers:
+            from_router = sub not in direct
+            for q in getattr(sub, "_queues", []):
+                counts[q] = counts.get(q, 0) + 1
+                if from_router:
+                    router_queues.add(q)
+        duplicated = sorted(q for q, n in counts.items() if n > 1 and q in router_queues)
+        if duplicated:
+            warnings.warn(
+                f"Multiple subscribers serve queue(s) {duplicated}: their workers compete "
+                f"for the same rows (nondeterministic via SKIP LOCKED). Use one subscriber "
+                f"per queue, or attach multiple handlers to a single subscriber.",
+                UserWarning,
+                stacklevel=2,
+            )
 
     def _warn_on_unstarted_foreign_publishers(self) -> None:
         """
@@ -251,7 +287,9 @@ def _warn_on_unstarted_foreign_publishers(self) -> None:
                     if key in self._warned_foreign_config_ids:
                         continue
                     self._warned_foreign_config_ids.add(key)
-                    queues = sorted({q for s in self.subscribers for q in getattr(s, "_queues", [])})
+                    # P21: name the queue(s) of the subscriber actually decorated by this
+                    # foreign publisher, not every subscriber on the broker.
+                    queues = sorted(getattr(sub, "_queues", []))
                     _logger.warning(
                         "Foreign publisher %r is decorated on outbox subscriber(s) for "
                         "queue(s) %s, but its broker has not been started yet. The first "
@@ -276,11 +314,15 @@ async def stop(self, *_args: object, **_kwargs: object) -> None:
         #
         # Upstream equivalent (replaced):
         #   BrokerUsecase.stop -> faststream/_internal/broker/broker.py
+        # P20: snapshot once — re-evaluating the ``subscribers`` property after the await
+        # (e.g. a mid-shutdown include_router) would desync the strict zip and raise out of
+        # stop(), defeating its never-raise contract.
+        subs = list(self.subscribers)
         results = await asyncio.gather(
-            *(sub.stop() for sub in self.subscribers),
+            *(sub.stop() for sub in subs),
             return_exceptions=True,
         )
-        for sub, result in zip(self.subscribers, results, strict=True):
+        for sub, result in zip(subs, results, strict=True):
             if isinstance(result, BaseException):
                 self._log_subscriber_stop_error(sub, result)
         self.running = False
@@ -384,6 +426,11 @@ async def publish_batch(  # ty: ignore[invalid-method-override]
         every row in the batch identically — per-row timer dedup is not supported,
         use :meth:`publish` for that.
         """
+        # P1: validate the session type up front so an empty batch fails the same way a
+        # non-empty one does (the command constructor that checks it is only built below).
+        if not isinstance(session, AsyncSession):
+            msg = "broker.publish_batch requires an sqlalchemy.ext.asyncio.AsyncSession"
+            raise TypeError(msg)
         if not bodies:
             # Validate the activate args even when there's no work so callers
             # get the same misuse error on empty batches as on real ones.
diff --git a/faststream_outbox/client.py b/faststream_outbox/client.py
index 77d8a87..6fe2d44 100644
--- a/faststream_outbox/client.py
+++ b/faststream_outbox/client.py
@@ -21,9 +21,12 @@
 from typing import TYPE_CHECKING
 
 from sqlalchemy import (
+    ARRAY,
     Float,
     MetaData,
+    String,
     and_,
+    any_,
     bindparam,
     delete,
     func,
@@ -182,7 +185,11 @@ async def fetch(
             select(t.c.id)
             .where(
                 t.c.next_attempt_at <= func.now(),
-                or_(*(t.c.queue == q for q in queues)),
+                # P11: ``queue = ANY(:queues)`` binds a single array param, so the SQL
+                # text is stable regardless of the number of queues — asyncpg can reuse
+                # the prepared statement across ticks instead of recompiling an
+                # OR-of-N-equalities whose text changes with the queue count.
+                t.c.queue == any_(bindparam("queues", list(queues), type_=ARRAY(String))),
                 # The OR is split into two index-implying disjuncts so Postgres'
                 # partial-index inference picks up both `_pending_idx` (Branch A,
                 # `acquired_token IS NULL`) and `_lease_idx` (Branch B,
@@ -260,7 +267,16 @@ async def delete_with_lease(
         if conn is None:
             msg = "OutboxClient.delete_with_lease requires a live AsyncConnection (got None)"
             raise TypeError(msg)
-        if dlq_payload is not None and self._dlq_table is not None:
+        if dlq_payload is not None:
+            if self._dlq_table is None:
+                # P10: a dlq_payload with no dlq_table would silently degrade to a plain
+                # DELETE — losing the audit row. If the broker/client wiring ever desyncs,
+                # fail loudly rather than drop forensics.
+                msg = (
+                    "delete_with_lease received a dlq_payload but the client has no dlq_table; "
+                    "refusing to drop the audit row silently"
+                )
+                raise RuntimeError(msg)
             stmt, params = self._build_dlq_cte_stmt(message_id, acquired_token, dlq_payload)
             result = await conn.execute(stmt, params)
             return (result.rowcount or 0) > 0
@@ -303,14 +319,14 @@ def _build_dlq_cte_stmt(
             f"WITH deleted AS ("  # noqa: S608
             f"DELETE FROM {outbox_name} "
             f"WHERE id = :message_id AND acquired_token = :acquired_token "
-            f"RETURNING id, queue, payload, headers, deliveries_count, created_at"
+            f"RETURNING id, queue, payload, headers, deliveries_count, created_at, timer_id"
             f") "
             f"INSERT INTO {dlq_name} ("
             f"original_id, queue, payload, headers, deliveries_count, created_at, "
-            f"failure_reason, last_exception"
+            f"failure_reason, last_exception, timer_id"
             f") "
             f"SELECT id, queue, payload, headers, deliveries_count, created_at, "
-            f":failure_reason, :last_exception "
+            f":failure_reason, :last_exception, timer_id "
             f"FROM deleted"
         )
         sql = text(cte_sql)
@@ -405,6 +421,7 @@ def _row_to_message(row: dict) -> OutboxInnerMessage:
         last_attempt_at=row["last_attempt_at"],
         acquired_at=row["acquired_at"],
         acquired_token=row["acquired_token"],
+        timer_id=row["timer_id"],
     )
 
 
diff --git a/faststream_outbox/configs.py b/faststream_outbox/configs.py
index 9802571..348427d 100644
--- a/faststream_outbox/configs.py
+++ b/faststream_outbox/configs.py
@@ -30,11 +30,6 @@ class OutboxBrokerConfig(BrokerConfig):
     # copy audit data into this table in the same statement as the outbox DELETE.
     # See ``OutboxClient.delete_with_lease`` for the CTE shape.
     dlq_table: "Table | None" = None
-
-    async def connect(self) -> None:
-        # Engine and client are wired up by the broker's constructor; nothing to do here.
-        pass
-
-    async def disconnect(self) -> None:
-        # Caller owns the engine — never dispose it here.
-        pass
+    # P24: the former connect()/disconnect() overrides were dead code — BrokerConfig has
+    # no such hooks and nothing in the package or upstream called them. The broker's own
+    # start()/connect() wire the engine/client; the caller owns the engine lifecycle.
diff --git a/faststream_outbox/envelope.py b/faststream_outbox/envelope.py
index cbfd38a..42f9d37 100644
--- a/faststream_outbox/envelope.py
+++ b/faststream_outbox/envelope.py
@@ -38,5 +38,16 @@ def _encode_payload(
         raise ValueError(msg)
     if content_type:
         out_headers["content-type"] = content_type
-    out_headers.setdefault("correlation_id", correlation_id or gen_cor_id())
+    # An explicit ``correlation_id`` kwarg used to lose silently to a
+    # ``headers["correlation_id"]`` of a different value (the kwarg was dropped).
+    # Treat a genuine mismatch as a conflict (like content-type above); otherwise the
+    # kwarg wins when set, falling back to the header, then a fresh id (P2).
+    header_cid = out_headers.get("correlation_id")
+    if correlation_id is not None and header_cid is not None and correlation_id != header_cid:
+        msg = (
+            f"correlation_id={correlation_id!r} conflicts with headers['correlation_id']="
+            f"{header_cid!r}. Pass correlation_id one way, not both."
+        )
+        raise ValueError(msg)
+    out_headers["correlation_id"] = correlation_id or header_cid or gen_cor_id()
     return payload, out_headers
diff --git a/faststream_outbox/message.py b/faststream_outbox/message.py
index 9295b99..fea342c 100644
--- a/faststream_outbox/message.py
+++ b/faststream_outbox/message.py
@@ -62,6 +62,10 @@ class OutboxInnerMessage:
     acquired_at: _dt.datetime | None
     acquired_token: uuid.UUID | None
 
+    # P9: the originating timer_id (single-publish dedup key), so a terminally-failed
+    # timer keeps it in the DLQ audit trail. None for non-timer rows.
+    timer_id: str | None = None
+
     retry_strategy: "RetryStrategyProto | None" = None
     last_exception: BaseException | None = None
 
@@ -109,7 +113,9 @@ async def _nack(self) -> None:
         if self.retry_strategy is not None:
             try:
                 delay = self.retry_strategy.get_next_attempt_delay(
-                    first_attempt_at=self.first_attempt_at or self.last_attempt_at,  # ty: ignore[invalid-argument-type]
+                    # _record_attempt() above always sets first_attempt_at, so the
+                    # former ``or self.last_attempt_at`` fallback was dead code (P19).
+                    first_attempt_at=self.first_attempt_at,  # ty: ignore[invalid-argument-type]
                     last_attempt_at=self.last_attempt_at,  # ty: ignore[invalid-argument-type]
                     attempts_count=self.attempts_count,
                     exception=self.last_exception,
diff --git a/faststream_outbox/metrics/__init__.py b/faststream_outbox/metrics/__init__.py
index de7bf19..3358712 100644
--- a/faststream_outbox/metrics/__init__.py
+++ b/faststream_outbox/metrics/__init__.py
@@ -21,6 +21,10 @@
   ran). ``exception_type`` is present when ``last_exception`` was set (post-handler raises;
   manual ``msg.reject()`` may omit it).
 * ``lease_lost`` — terminal flush found a foreign lease. Tags include ``phase`` (``terminal`` | ``retry``).
+  ``acked`` / ``nacked_retried`` / ``nacked_terminal`` are emitted only **after** the
+  terminal/retry write lands (P17); a row whose lease was reclaimed emits ``lease_lost``
+  *instead* (never a paired acked/nacked), so the row isn't counted twice when it
+  redelivers.
 * ``dlq_written`` — emitted from ``_flush_terminal`` after the DELETE+INSERT CTE commits.
   Fires only when ``OutboxBroker`` was constructed with ``dlq_table=...`` AND the row was
   terminal-by-failure (any ``nacked_terminal`` reason). Skipped on lease-lost. Tags:
@@ -31,7 +35,10 @@
 * ``published`` — producer-side insert. Tags include ``status`` (``success`` | ``error``),
   ``count, size_bytes, duration_seconds``. No ``subscriber`` tag.
   ``count`` is **messages landed**, not publish attempts — errors and ``timer_id``
-  no-ops carry ``count=0``. Counter-style adapters should `inc(count)` so totals
+  no-ops both carry ``count=0`` (P6). ``count=0`` alone does not distinguish them:
+  a successful ``timer_id`` conflict has ``status="success"`` with no ``exception_type``,
+  while a failed publish has ``status="error"`` with an ``exception_type``.
+  Counter-style adapters should `inc(count)` so totals
   reflect messages-on-the-wire; duration histograms record every attempt
   (including failures) so failed-publish latency stays observable.
 
diff --git a/faststream_outbox/metrics/prometheus.py b/faststream_outbox/metrics/prometheus.py
index 17f8f38..feb4204 100644
--- a/faststream_outbox/metrics/prometheus.py
+++ b/faststream_outbox/metrics/prometheus.py
@@ -292,7 +292,12 @@ def __call__(self, event: str, tags: Mapping[str, typing.Any]) -> None:  # noqa:
             # every attempt (with the status label) so failed-publish latency stays
             # observable.
             count = tags.get("count", 1)
-            if count > 0:
+            if status == "error":
+                # P28: an error lands 0 messages but is one failed publish. Increment the
+                # status="error" series (the exact label dashboards alert on) by 1, rather
+                # than gating on count > 0 — which left it permanently at zero.
+                self._published_total.labels(*publish_base, status).inc()
+            elif count > 0:
                 self._published_total.labels(*publish_base, status).inc(count)
             duration = tags.get("duration_seconds")
             if duration is not None:
diff --git a/faststream_outbox/publisher/producer.py b/faststream_outbox/publisher/producer.py
index 5b6a652..65e4868 100644
--- a/faststream_outbox/publisher/producer.py
+++ b/faststream_outbox/publisher/producer.py
@@ -79,15 +79,18 @@ def disconnect(self) -> None:
         pass
 
     async def publish(self, cmd: OutboxPublishCommand) -> int | None:
-        payload, hdrs = _encode_payload(
-            cmd.body,
-            headers=cmd.headers,
-            correlation_id=cmd.correlation_id,
-            serializer=self.serializer,
-        )
         start_perf = time.perf_counter()
-        size_bytes = len(payload)
+        size_bytes = 0
         try:
+            # P3: encode inside the try so a serialization / content-type / correlation_id
+            # failure still emits the ``published`` error metric instead of bypassing it.
+            payload, hdrs = _encode_payload(
+                cmd.body,
+                headers=cmd.headers,
+                correlation_id=cmd.correlation_id,
+                serializer=self.serializer,
+            )
+            size_bytes = len(payload)
             row_id = await self._do_publish(cmd, payload, hdrs)
         except Exception as exc:
             self._emit_metric(
@@ -173,15 +176,17 @@ async def publish_batch(self, cmd: OutboxPublishCommand) -> None:
             next_at = cmd.activate_at
         rows: list[dict[str, typing.Any]] = []
         total_size = 0
-        for body in bodies:
-            payload, hdrs = _encode_payload(body, headers=cmd.headers, serializer=self.serializer)
-            total_size += len(payload)
-            row: dict[str, typing.Any] = {"queue": cmd.queue, "payload": payload, "headers": hdrs}
-            if next_at is not None:
-                row["next_attempt_at"] = next_at
-            rows.append(row)
         start_perf = time.perf_counter()
         try:
+            # P3: encode inside the try so a serialization failure on any body still
+            # emits the ``published`` error metric.
+            for body in bodies:
+                payload, hdrs = _encode_payload(body, headers=cmd.headers, serializer=self.serializer)
+                total_size += len(payload)
+                row: dict[str, typing.Any] = {"queue": cmd.queue, "payload": payload, "headers": hdrs}
+                if next_at is not None:
+                    row["next_attempt_at"] = next_at
+                rows.append(row)
             await cmd.session.execute(insert(self._table), rows)
             # Skip NOTIFY only when genuinely future-dated; past times are eligible.
             if next_at is None or next_at <= now:
diff --git a/faststream_outbox/response.py b/faststream_outbox/response.py
index 42e401c..d33cdcd 100644
--- a/faststream_outbox/response.py
+++ b/faststream_outbox/response.py
@@ -20,6 +20,10 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 
 
+# Matches the ``queue`` column width in ``make_outbox_table`` (``String(255)``).
+_MAX_QUEUE_LENGTH = 255
+
+
 class OutboxPublishCommand(BatchPublishCommand):
     """Outbox-specific publish command: carries session + scheduling fields end-to-end."""
 
@@ -46,6 +50,24 @@ def __init__(
         if activate_at is not None and activate_at.tzinfo is None:
             msg = "OutboxPublishCommand requires activate_at to be timezone-aware"
             raise ValueError(msg)
+        # P5: ``queue`` reaches SQL unvalidated — empty / non-str / over the
+        # ``String(255)`` column would surface as an opaque DB error (or silent
+        # truncation). Reject at the single-source-of-truth constructor instead.
+        if not isinstance(queue, str):
+            msg = f"queue must be a str, got {type(queue).__name__}"
+            raise TypeError(msg)
+        if not queue:
+            msg = "queue must be a non-empty string"
+            raise ValueError(msg)
+        if len(queue) > _MAX_QUEUE_LENGTH:
+            msg = f"queue must be at most {_MAX_QUEUE_LENGTH} characters (got {len(queue)})"
+            raise ValueError(msg)
+        # P4: timer_id / correlation_id are per-row single-publish concepts the batch
+        # path silently drops (each batched row gets its own auto correlation_id and
+        # no dedup key). Reject them on a batch command rather than accept-and-ignore.
+        if bodies and (timer_id is not None or correlation_id is not None):
+            msg = "timer_id / correlation_id are not supported for batch publishes (multiple bodies)"
+            raise ValueError(msg)
         super().__init__(
             body,
             *bodies,
diff --git a/faststream_outbox/retry.py b/faststream_outbox/retry.py
index 8c51181..0cbcded 100644
--- a/faststream_outbox/retry.py
+++ b/faststream_outbox/retry.py
@@ -13,6 +13,16 @@
 # retry horizon, so this never changes behavior for a real configuration.
 _MAX_DELAY_SECONDS = 100.0 * 365.0 * 24.0 * 60.0 * 60.0
 
+# Jitter is applied as ``delay *= 1 + U(-j/2, +j/2)``; j > 2 makes the lower bound
+# negative, so a jittered delay can go negative → an immediate (hot) retry. Bound it.
+_MAX_JITTER_FACTOR = 2.0
+
+
+def _validate_jitter_factor(jitter_factor: float) -> None:
+    if not (0.0 <= jitter_factor <= _MAX_JITTER_FACTOR):
+        msg = f"jitter_factor must be between 0 and {_MAX_JITTER_FACTOR}, got {jitter_factor}"
+        raise ValueError(msg)
+
 
 class RetryStrategyProto(Protocol):
     """
@@ -41,6 +51,15 @@ class _RetryStrategyTemplate(ABC, RetryStrategyProto):
     max_attempts: int | None = None
     max_total_delay_seconds: float | None = None
 
+    def __post_init__(self) -> None:
+        # P23: reject knobs that produce nonsense (a hot-retry loop, an unreachable cap).
+        if self.max_attempts is not None and self.max_attempts < 1:
+            msg = f"max_attempts must be >= 1 if set, got {self.max_attempts}"
+            raise ValueError(msg)
+        if self.max_total_delay_seconds is not None and self.max_total_delay_seconds <= 0:
+            msg = f"max_total_delay_seconds must be > 0 if set, got {self.max_total_delay_seconds}"
+            raise ValueError(msg)
+
     @abstractmethod
     def _delay_seconds(self, *, attempts_count: int) -> float: ...
 
@@ -83,6 +102,13 @@ class ConstantRetry(_RetryStrategyTemplate):
     jitter_factor: float = 0.0
     _random: random.Random = field(default_factory=random.Random)
 
+    def __post_init__(self) -> None:
+        super().__post_init__()
+        if self.delay_seconds <= 0:
+            msg = f"delay_seconds must be > 0, got {self.delay_seconds}"
+            raise ValueError(msg)
+        _validate_jitter_factor(self.jitter_factor)
+
     def _delay_seconds(self, *, attempts_count: int) -> float:  # noqa: ARG002
         delay = self.delay_seconds
         if self.jitter_factor:
@@ -97,6 +123,16 @@ class LinearRetry(_RetryStrategyTemplate):
     jitter_factor: float = 0.0
     _random: random.Random = field(default_factory=random.Random)
 
+    def __post_init__(self) -> None:
+        super().__post_init__()
+        if self.initial_delay_seconds <= 0:
+            msg = f"initial_delay_seconds must be > 0, got {self.initial_delay_seconds}"
+            raise ValueError(msg)
+        if self.step_seconds < 0:
+            msg = f"step_seconds must be >= 0, got {self.step_seconds}"
+            raise ValueError(msg)
+        _validate_jitter_factor(self.jitter_factor)
+
     def _delay_seconds(self, *, attempts_count: int) -> float:
         delay = self.initial_delay_seconds + self.step_seconds * max(0, attempts_count - 1)
         if self.jitter_factor:
@@ -112,6 +148,19 @@ class ExponentialRetry(_RetryStrategyTemplate):
     jitter_factor: float = 0.0
     _random: random.Random = field(default_factory=random.Random)
 
+    def __post_init__(self) -> None:
+        super().__post_init__()
+        if self.initial_delay_seconds <= 0:
+            msg = f"initial_delay_seconds must be > 0, got {self.initial_delay_seconds}"
+            raise ValueError(msg)
+        if self.multiplier <= 0:
+            msg = f"multiplier must be > 0, got {self.multiplier}"
+            raise ValueError(msg)
+        if self.max_delay_seconds is not None and self.max_delay_seconds <= 0:
+            msg = f"max_delay_seconds must be > 0 if set, got {self.max_delay_seconds}"
+            raise ValueError(msg)
+        _validate_jitter_factor(self.jitter_factor)
+
     def _delay_seconds(self, *, attempts_count: int) -> float:
         try:
             delay = self.initial_delay_seconds * (self.multiplier ** max(0, attempts_count - 1))
diff --git a/faststream_outbox/schema.py b/faststream_outbox/schema.py
index ea1f6f6..b8b01c8 100644
--- a/faststream_outbox/schema.py
+++ b/faststream_outbox/schema.py
@@ -15,6 +15,7 @@
 
 from sqlalchemy import (
     BigInteger,
+    CheckConstraint,
     Column,
     DateTime,
     Index,
@@ -46,13 +47,26 @@ def make_outbox_table(metadata: "MetaData", table_name: str = "outbox") -> Table
     ``outbox_<table_name>`` past Postgres' 63-byte identifier limit (NAMEDATALEN-1).
     """
     # Byte length, not char count — UTF-8 multibyte chars expand and would silently
-    # truncate the channel identifier on one side of LISTEN/NOTIFY.
-    encoded_channel = b"outbox_" + table_name.encode("utf-8")
-    if len(encoded_channel) > _POSTGRES_IDENT_MAX_BYTES:
+    # truncate identifiers. Guard on the LONGEST identifier derived from table_name:
+    # the NOTIFY channel "outbox_<t>" (7-byte prefix) AND the index / constraint names
+    # ("<t>_pending_idx", "<t>_timer_id_uq", "<t>_lease_idx", "<t>_lease_ck"; suffixes
+    # up to ~12 bytes). The index suffixes are longer than the channel prefix, so a
+    # table_name that fits the channel can still overflow an index name and fail at
+    # CREATE INDEX time (P7).
+    name_bytes = table_name.encode("utf-8")
+    derived = (
+        b"outbox_" + name_bytes,
+        name_bytes + b"_pending_idx",
+        name_bytes + b"_timer_id_uq",
+        name_bytes + b"_lease_idx",
+        name_bytes + b"_lease_ck",
+    )
+    longest = max(derived, key=len)
+    if len(longest) > _POSTGRES_IDENT_MAX_BYTES:
         msg = (
-            f"table_name {table_name!r} too long for NOTIFY channel "
-            f"'outbox_<table_name>': must fit in {_POSTGRES_IDENT_MAX_BYTES} bytes "
-            f"(got {len(encoded_channel)})"
+            f"table_name {table_name!r} too long: the derived identifier "
+            f"{longest.decode('utf-8', 'replace')!r} must fit in {_POSTGRES_IDENT_MAX_BYTES} bytes "
+            f"(got {len(longest)})"
         )
         raise ValueError(msg)
     table = Table(
@@ -71,6 +85,14 @@ def make_outbox_table(metadata: "MetaData", table_name: str = "outbox") -> Table
         Column("acquired_at", DateTime(timezone=True), nullable=True),
         Column("acquired_token", Uuid, nullable=True),
         Column("timer_id", String(255), nullable=True),
+        # P8: a lease is either fully set or fully unset. A half-set lease (e.g. a
+        # manual UPDATE that sets acquired_at but not acquired_token) would be
+        # permanently invisible to fetch, cancel_timer, and the metrics — this
+        # CHECK makes that state unrepresentable.
+        CheckConstraint(
+            "(acquired_token IS NULL) = (acquired_at IS NULL)",
+            name=f"{table_name}_lease_ck",
+        ),
     )
     # Partial index that backs the fetch query's hot branch
     # (`WHERE acquired_token IS NULL AND queue = ? AND next_attempt_at <= now()`).
@@ -138,6 +160,9 @@ def make_dlq_table(metadata: "MetaData", table_name: str = "outbox_dlq") -> Tabl
         # canonical set can grow without a migration that rewrites every audit row.
         Column("failure_reason", String(64), nullable=False),
         Column("last_exception", String, nullable=True),
+        # P9: carry the originating timer_id so a terminally-failed timer keeps its
+        # business dedup key in the audit trail. Nullable — most rows have none.
+        Column("timer_id", String(255), nullable=True),
     )
     Index(f"{table_name}_queue_failed_idx", table.c.queue, table.c.failed_at)
     return table
diff --git a/faststream_outbox/subscriber/factory.py b/faststream_outbox/subscriber/factory.py
index b9e31cb..2d9c951 100644
--- a/faststream_outbox/subscriber/factory.py
+++ b/faststream_outbox/subscriber/factory.py
@@ -74,7 +74,7 @@ def create_subscriber(
     )
 
 
-def _validate_subscriber_config(
+def _validate_subscriber_config(  # noqa: C901  # flat sequence of independent knob checks
     *,
     max_workers: int,
     fetch_batch_size: int,
@@ -99,11 +99,22 @@ def _validate_subscriber_config(
     if fetch_batch_size <= 0:
         msg = f"fetch_batch_size must be >= 1, got {fetch_batch_size}"
         raise ValueError(msg)
+    # P12: non-positive intervals/TTL turn the adaptive backoff into a busy-poll (or an
+    # instantly-expiring lease). Reject up front rather than spin a hot loop at runtime.
+    if min_fetch_interval <= 0:
+        msg = f"min_fetch_interval must be > 0, got {min_fetch_interval}"
+        raise ValueError(msg)
+    if max_fetch_interval <= 0:
+        msg = f"max_fetch_interval must be > 0, got {max_fetch_interval}"
+        raise ValueError(msg)
+    if lease_ttl_seconds <= 0:
+        msg = f"lease_ttl_seconds must be > 0, got {lease_ttl_seconds}"
+        raise ValueError(msg)
     if min_fetch_interval > max_fetch_interval:
         msg = (
             f"min_fetch_interval ({min_fetch_interval}) must be <= max_fetch_interval "
-            f"({max_fetch_interval}); the adaptive backoff treats min as a floor and "
-            f"max as the ceiling."
+            f"({max_fetch_interval}); the adaptive idle backoff grows from ~min_fetch_interval "
+            f"(the base interval, with ±50% jitter) up to max_fetch_interval (the ceiling)."
         )
         raise ValueError(msg)
     is_no_retry = isinstance(retry_strategy, NoRetry)
diff --git a/faststream_outbox/subscriber/usecase.py b/faststream_outbox/subscriber/usecase.py
index 018bcbf..0d3d6c5 100644
--- a/faststream_outbox/subscriber/usecase.py
+++ b/faststream_outbox/subscriber/usecase.py
@@ -161,6 +161,9 @@ def __init__(
         # Set by the LISTEN callback to wake _fetch_loop early; cleared after each
         # wakeup. When LISTEN is unavailable the event simply never fires and the
         # loop sleeps the full adaptive interval.
+        # Frozen set of served queues for the O(1) NOTIFY-payload filter (P14); the
+        # queue list is fixed per subscriber, so caching it here clarifies intent.
+        self._served_queues: frozenset[str] = frozenset(config.queues)
         self._notify_event: asyncio.Event = asyncio.Event()
         # Set by stop() to halt _fetch_inner before tasks are cancelled. Distinct
         # from self.running: running must stay True during drain so FastStream's
@@ -237,9 +240,14 @@ async def stop(self) -> None:
         with anyio.move_on_after(self._outer_config.graceful_timeout):
             await self._inflight.join()
         self.running = False
-        for task in self.tasks:
+        tasks = list(self.tasks)
+        for task in tasks:
             if not task.done():
                 task.cancel()
+        # P16: await the cancellations so the loops have actually unwound (and released
+        # their connections) before stop() returns — otherwise a caller's immediate
+        # engine.dispose() races the teardown. return_exceptions swallows CancelledError.
+        await asyncio.gather(*tasks, return_exceptions=True)
         self.tasks.clear()
 
     @property
@@ -356,9 +364,11 @@ async def _open_listen_connection(self, engine: "AsyncEngine") -> "_asyncpg.Conn
         non-asyncpg driver, permission error, network problem). The fetch loop falls back
         to polling-only behavior in that case.
 
-        A separate connection is required because asyncpg's ``add_listener`` makes the
-        connection's reader task monopolize it — interleaving normal queries breaks
-        notification delivery.
+        A separate connection is used so application fetch traffic and the LISTEN reader
+        task don't contend on one connection. The only query this loop runs on the LISTEN
+        connection is the bounded ``SELECT 1`` liveness probe in ``_fetch_inner`` — a
+        deliberate, infrequent exception that surfaces a silently-dropped socket; it does
+        not interfere with notification delivery (P15).
         """
         if _asyncpg is None or "asyncpg" not in (engine.url.drivername or ""):
             return None
@@ -391,14 +401,20 @@ async def _open_listen_connection(self, engine: "AsyncEngine") -> "_asyncpg.Conn
                     await conn.close()
         return conn if listening else None
 
-    def _on_notify(self, *_args: object) -> None:
+    def _on_notify(self, *args: object) -> None:
         """
         Asyncpg notification callback: ``(connection, pid, channel, payload)``.
 
-        We only need the wake-up signal; payload is ignored. Setting an ``asyncio.Event``
-        from the asyncpg reader task is safe — it runs on the same event loop.
+        The payload is the publisher's queue name (``pg_notify('outbox_<table>', queue)``).
+        We only wake for queues this subscriber serves — on a busy multi-queue table that
+        avoids a cross-queue wakeup storm (every queue's NOTIFY waking every subscriber).
+        If the payload shape is unexpected, wake conservatively rather than risk a missed
+        delivery (P14). Setting an ``asyncio.Event`` from the asyncpg reader task is safe —
+        it runs on the same event loop.
         """
-        self._notify_event.set()
+        payload = args[3] if len(args) >= 4 else None  # noqa: PLR2004
+        if payload is None or payload in self._served_queues:
+            self._notify_event.set()
 
     async def _worker_loop(self) -> None:
         """Thin wrapper around :meth:`_run_with_reconnect` for the worker path."""
@@ -491,10 +507,21 @@ async def _worker_inner(self, *, writer_conn: "AsyncConnection | None") -> None:
             row = await self._inflight.get()
             try:
                 await self.dispatch_one(row, writer_conn=writer_conn)
+            except _OutboxConfigError as e:
+                # P18: a config error (e.g. OutboxResponse + foreign publisher) is not a
+                # connection failure. Letting it propagate to _run_with_reconnect would tear
+                # down the writer connection and back off (up to 30s), throttling unrelated
+                # rows. Log it and continue; the row's lease expires and it is reclaimed.
+                # Fix the configuration to stop the error.
+                self._log(
+                    log_level=logging.ERROR,
+                    message=f"Outbox configuration error (fix required; row left to lease-expiry retry): {e!r}",
+                    exc_info=e,
+                )
             finally:
                 self._inflight.task_done()
 
-    async def dispatch_one(
+    async def dispatch_one(  # noqa: C901  # linear pipeline: guard, consume, branch on outcome, flush
         self,
         row: OutboxInnerMessage,
         *,
@@ -519,11 +546,14 @@ async def dispatch_one(
         row.retry_strategy = self._config.retry_strategy
         base = self._base_tags(row.queue)
         if not row.allow_delivery(max_deliveries=self._config.max_deliveries, logger=logger):
-            self._emit_metric(
-                "nacked_terminal",
-                {**base, "deliveries_count": row.deliveries_count, "reason": "max_deliveries"},
-            )
-            await self._safe_flush(row, terminal=True, writer_conn=writer_conn)
+            # P17: flush first; emit the terminal metric only if the DELETE landed. A
+            # lease-lost delete (rowcount 0 → redelivered) emits ``lease_lost`` from the
+            # flush instead, so emitting nacked_terminal here too would double-count.
+            if await self._safe_flush(row, terminal=True, writer_conn=writer_conn):
+                self._emit_metric(
+                    "nacked_terminal",
+                    {**base, "deliveries_count": row.deliveries_count, "reason": "max_deliveries"},
+                )
             return
         # AckPolicy middleware catches handler exceptions; _CaptureExceptionMiddleware
         # stashes exc onto row.last_exception before nack runs, so retry strategies
@@ -549,7 +579,11 @@ async def dispatch_one(
             # terminal/retry, so its state is undefined — flushing or emitting an
             # ack/nack would lie. The lease will expire and the row will be
             # reclaimed; the ERROR log is the operator signal.
-            self._log(log_level=logging.ERROR, message=f"Outbox worker error: {e!r}", exc_info=e)
+            self._log(
+                log_level=logging.ERROR,
+                message=f"Outbox handler error escaped consume(); row left to lease-expiry retry: {e!r}",
+                exc_info=e,
+            )
             return
         # Shutdown race: SubscriberUsecase.consume() returns None without invoking
         # process_message when self.running has been flipped to False by stop().
@@ -571,15 +605,19 @@ async def dispatch_one(
             terminal_tags: dict[str, typing.Any] = {**common, "reason": row.terminal_failure_reason}
             if row.last_exception is not None:
                 terminal_tags["exception_type"] = type(row.last_exception).__name__
-            self._emit_metric("nacked_terminal", terminal_tags)
+            outcome: tuple[str, dict[str, typing.Any]] = ("nacked_terminal", terminal_tags)
         elif row.pending_delay_seconds is not None:
             retry_tags: dict[str, typing.Any] = {**common, "next_delay_seconds": row.pending_delay_seconds}
             if row.last_exception is not None:
                 retry_tags["exception_type"] = type(row.last_exception).__name__
-            self._emit_metric("nacked_retried", retry_tags)
+            outcome = ("nacked_retried", retry_tags)
         else:
-            self._emit_metric("acked", common)
-        await self._safe_flush(row, terminal=row.to_delete, writer_conn=writer_conn)
+            outcome = ("acked", common)
+        # P17: emit the outcome metric only after the flush actually lands. A lease-lost
+        # flush (rowcount 0 → the row gets redelivered) emits ``lease_lost`` from the flush
+        # method instead; emitting acked/nacked here too would count the row twice.
+        if await self._safe_flush(row, terminal=row.to_delete, writer_conn=writer_conn):
+            self._emit_metric(*outcome)
 
     async def _safe_flush(
         self,
@@ -587,10 +625,14 @@ async def _safe_flush(
         *,
         terminal: bool,
         writer_conn: "AsyncConnection | None",
-    ) -> None:
+    ) -> bool:
         """
         Run the terminal/retry write, propagating errors only when ``writer_conn`` is set.
 
+        Returns True iff the write landed (rowcount > 0). False means the lease was lost
+        (a newer fetch reclaimed the row) or — on the test-broker path — the flush raised
+        and was swallowed; either way the caller must NOT emit an acked/nacked metric (P17).
+
         When ``writer_conn`` is None (test broker / sync dispatch), flush errors are logged
         and swallowed — there's no shared connection to rebuild and the legacy test-broker
         contract is "publish never raises from a flush failure". When ``writer_conn`` is
@@ -600,20 +642,24 @@ async def _safe_flush(
         flush = self._flush_terminal if terminal else self._flush_retry
         if writer_conn is None:
             try:
-                await flush(row, writer_conn=None)
+                return await flush(row, writer_conn=None)
             except Exception as e:  # noqa: BLE001
-                self._log(log_level=logging.ERROR, message=f"Outbox worker error: {e!r}", exc_info=e)
-            return
-        await flush(row, writer_conn=writer_conn)
+                self._log(
+                    log_level=logging.ERROR,
+                    message=f"Outbox terminal-flush error (swallowed; no writer connection): {e!r}",
+                    exc_info=e,
+                )
+                return False
+        return await flush(row, writer_conn=writer_conn)
 
     async def _flush_terminal(
         self,
         row: OutboxInnerMessage,
         *,
         writer_conn: "AsyncConnection | None",
-    ) -> None:
+    ) -> bool:
         if row.acquired_token is None:
-            return
+            return False
         # Build the DLQ payload only when this row is terminal-by-failure AND the
         # broker is configured with a DLQ table. Success-by-ack rows reach this
         # method too (terminal=True via to_delete) but carry
@@ -651,26 +697,31 @@ async def _flush_terminal(
                     "deliveries_count": row.deliveries_count,
                 },
             )
-            return
+            return False
         if dlq_payload is not None:
-            self._emit_metric(
-                "dlq_written",
-                {
-                    **self._base_tags(row.queue),
-                    "deliveries_count": row.deliveries_count,
-                    "failure_reason": row.terminal_failure_reason,
-                    "exception_type": (type(row.last_exception).__name__ if row.last_exception is not None else None),
-                },
-            )
+            # P34: omit exception_type when there's no exception (e.g. max_deliveries)
+            # rather than emitting it as None, matching the nacked_terminal convention.
+            dlq_tags: dict[str, typing.Any] = {
+                **self._base_tags(row.queue),
+                "deliveries_count": row.deliveries_count,
+                "failure_reason": row.terminal_failure_reason,
+            }
+            if row.last_exception is not None:
+                dlq_tags["exception_type"] = type(row.last_exception).__name__
+            self._emit_metric("dlq_written", dlq_tags)
+        return True
 
     async def _flush_retry(
         self,
         row: OutboxInnerMessage,
         *,
         writer_conn: "AsyncConnection | None",
-    ) -> None:
+    ) -> bool:
         if row.acquired_token is None or row.pending_delay_seconds is None:
-            return
+            return False
+        # P19: first_attempt_at / last_attempt_at are the worker's clock; next_attempt_at
+        # is computed server-side (now() + delay) inside mark_pending_with_lease, so the
+        # reschedule time stays clock-skew-immune even though these audit timestamps don't.
         updated = await self._client.mark_pending_with_lease(
             writer_conn,
             row.id,
@@ -701,6 +752,8 @@ async def _flush_retry(
                     "deliveries_count": row.deliveries_count,
                 },
             )
+            return False
+        return True
 
     @typing.override
     async def get_one(self, *, timeout: float = 5.0) -> typing.NoReturn:
diff --git a/faststream_outbox/testing.py b/faststream_outbox/testing.py
index acd630f..cfca37a 100644
--- a/faststream_outbox/testing.py
+++ b/faststream_outbox/testing.py
@@ -80,6 +80,12 @@ def feed(
         next_attempt_at: _dt.datetime | None = None,
         timer_id: str | None = None,
     ) -> int | None:
+        # P31: reject naive datetimes up front — the publish path is tz-strict, and a
+        # naive next_attempt_at otherwise blows up deep in a patched-away logger with no
+        # diagnostic. Match the production contract here.
+        if next_attempt_at is not None and next_attempt_at.tzinfo is None:
+            msg = "feed() requires next_attempt_at to be timezone-aware"
+            raise ValueError(msg)
         # Mirror the real client's partial-unique-on-(queue, timer_id) behavior:
         # re-feeding a timer that already exists is a no-op.
         if timer_id is not None and any(r.queue == queue and r.timer_id == timer_id for r in self._rows):
@@ -88,7 +94,10 @@ def feed(
             id=self._next_id,
             queue=queue,
             payload=payload,
-            headers=headers,
+            # P32: store a copy so a handler mutating msg.headers can't corrupt the
+            # "persisted" row (the real client round-trips through the DB; the fake must
+            # not share the dict by reference).
+            headers=dict(headers) if headers is not None else None,
             next_attempt_at=next_attempt_at or _utcnow(),
             timer_id=timer_id,
         )
@@ -169,6 +178,7 @@ async def delete_with_lease(
                             "failed_at": _utcnow(),
                             "failure_reason": dlq_payload["failure_reason"],
                             "last_exception": dlq_payload["last_exception"],
+                            "timer_id": row.timer_id,  # P9 parity with the real DLQ CTE
                         },
                     )
                 del self._rows[i]
@@ -227,7 +237,7 @@ def _to_inner(row: _FakeRow) -> OutboxInnerMessage:
         id=row.id,
         queue=row.queue,
         payload=row.payload,
-        headers=row.headers,
+        headers=dict(row.headers) if row.headers is not None else None,  # P32: don't share the dict by reference
         attempts_count=row.attempts_count,
         deliveries_count=row.deliveries_count,
         created_at=row.created_at,
@@ -236,11 +246,19 @@ def _to_inner(row: _FakeRow) -> OutboxInnerMessage:
         last_attempt_at=row.last_attempt_at,
         acquired_at=row.acquired_at,
         acquired_token=row.acquired_token,
+        timer_id=row.timer_id,
     )
 
 
 def _find_subscriber_for_queue(broker: OutboxBroker, queue: str) -> "OutboxSubscriber | None":
-    """First matching subscriber wins — mirrors production fetch behavior for overlapping subscribers."""
+    """
+    First matching subscriber wins (deterministic).
+
+    NB: this does NOT mirror production for *overlapping* subscribers — there, multiple
+    subscribers on the same queue compete via ``FOR UPDATE SKIP LOCKED``, so which one
+    claims a given row is nondeterministic. The fake picks the first match for test
+    repeatability (P35).
+    """
     for raw_subscriber in broker.subscribers:
         sub = typing.cast("OutboxSubscriber", raw_subscriber)
         if not sub.calls:
@@ -391,7 +409,10 @@ async def fake_publish(
         activate_at: _dt.datetime | None = None,
         timer_id: str | None = None,
     ) -> int | None:
-        # session is ignored in test mode — the fake client has no transaction.
+        # P33: test mode is intentionally lenient about ``session`` — the fake client has
+        # no transaction, so any value (including None) is accepted here. This DIVERGES from
+        # production and from ``publisher.publish()`` / ``OutboxResponse``, which require a
+        # real AsyncSession; tests that assert that contract must use those paths.
         del session
         _validate_activate_args("broker.publish", activate_in, activate_at)
         payload, hdrs = _encode_payload(
diff --git a/tests/test_fake.py b/tests/test_fake.py
index 96b4659..85f85df 100644
--- a/tests/test_fake.py
+++ b/tests/test_fake.py
@@ -856,7 +856,7 @@ async def test_subscriber_with_no_handler_skips_loop_setup() -> None:
         max_deliveries=None,
         config=broker.config.broker_config,  # type: ignore[arg-type]
     )
-    broker._subscribers.add(sub)  # noqa: SLF001  # ty: ignore[unresolved-attribute]
+    broker._subscribers.add(sub)  # noqa: SLF001
     async with TestOutboxBroker(broker, run_loops=True):
         # Inside the test broker the logger is wired; call start() directly so the
         # ``if not self.calls: return`` branch fires (no add_call() was performed).
@@ -980,7 +980,7 @@ async def test_sync_publish_skips_callless_subscriber() -> None:
         max_deliveries=None,
         config=broker.config.broker_config,  # type: ignore[arg-type]
     )
-    broker._subscribers.add(sub)  # noqa: SLF001  # ty: ignore[unresolved-attribute]
+    broker._subscribers.add(sub)  # noqa: SLF001
 
     test_broker = TestOutboxBroker(broker)
     async with test_broker:
@@ -1047,6 +1047,27 @@ async def test_fake_client_cancel_timer_removes_unleased_row() -> None:
     assert fake.rows == []
 
 
+def test_fake_client_feed_rejects_naive_next_attempt_at() -> None:
+    """P31: feed() rejects a naive datetime up front, matching the tz-strict publish path."""
+    fake = FakeOutboxClient()
+    with pytest.raises(ValueError, match="timezone-aware"):
+        fake.feed(queue="q", payload=b"x", next_attempt_at=_dt.datetime.now())  # noqa: DTZ005  # naive on purpose
+
+
+async def test_fake_headers_not_shared_by_reference() -> None:
+    """P32: the fake must copy headers at its boundaries so handler/caller mutation can't corrupt rows."""
+    fake = FakeOutboxClient()
+    src = {"k": "v"}
+    fake.feed(queue="q", payload=b"x", headers=src)
+    src["k"] = "mutated-by-caller"
+    assert fake.rows[0].headers == {"k": "v"}  # stored copy is independent of the caller's dict
+
+    inner = _to_inner(fake.rows[0])
+    assert inner.headers is not None
+    inner.headers["k"] = "mutated-by-handler"
+    assert fake.rows[0].headers == {"k": "v"}  # persisted row unaffected by inner-message mutation
+
+
 async def test_fake_client_cancel_timer_unknown_returns_false() -> None:
     fake = FakeOutboxClient()
     assert await fake.cancel_timer(queue="q", timer_id="never-existed") is False
@@ -1620,6 +1641,31 @@ def recorder(event: str, tags: typing.Any) -> None:
     assert test_broker.fake_client.rows == []
 
 
+async def test_fake_dlq_written_omits_exception_type_when_no_exception() -> None:
+    """P34: dlq_written must omit exception_type (not emit None) for a max_deliveries terminal."""
+    events: list[tuple[str, dict]] = []
+
+    def recorder(event: str, tags: typing.Any) -> None:
+        events.append((event, dict(tags)))
+
+    broker, _received, _raised = _make_broker_with_dlq(max_deliveries=1, recorder=recorder)
+    test_broker = TestOutboxBroker(broker)
+
+    async with test_broker:
+        sub = next(iter(broker._subscribers))  # noqa: SLF001
+        test_broker.feed("orders", b"x")
+        row = test_broker.fake_client.rows[0]
+        row.deliveries_count = 5  # over max_deliveries=1 → terminal without ever running the handler
+        row.acquired_token = uuid.uuid4()
+        row.acquired_at = _dt.datetime.now(tz=_dt.UTC)
+        await sub.dispatch_one(_to_inner(row), writer_conn=None)
+
+    dlq_events = [t for e, t in events if e == "dlq_written"]
+    assert len(dlq_events) == 1
+    assert dlq_events[0]["failure_reason"] == "max_deliveries"
+    assert "exception_type" not in dlq_events[0]  # omitted, not None
+
+
 async def test_test_broker_aenter_returns_single_outbox_broker() -> None:
     """
     0.7.1's EnterType binding means TestOutboxBroker yields a single OutboxBroker, not a list/tuple.
diff --git a/tests/test_integration.py b/tests/test_integration.py
index b74c653..5ff1073 100644
--- a/tests/test_integration.py
+++ b/tests/test_integration.py
@@ -11,6 +11,7 @@
 import pytest
 from faststream.kafka import KafkaBroker, TestKafkaBroker
 from sqlalchemy import MetaData, Table, event, insert, select, text
+from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncEngine, async_sessionmaker, create_async_engine
 
 from faststream_outbox import (
@@ -1364,6 +1365,46 @@ async def handle(body: dict) -> None:
             await conn.execute(text(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE'))
 
 
+async def test_lease_pairing_check_rejects_half_set_lease(pg_engine: AsyncEngine, outbox_table: Table) -> None:
+    """P8: a half-set lease (acquired_at set, acquired_token NULL) is unrepresentable via the CHECK."""
+    with pytest.raises(IntegrityError):
+        async with pg_engine.begin() as conn:
+            await conn.execute(
+                insert(outbox_table).values(
+                    queue="orders",
+                    payload=b"x",
+                    acquired_at=_dt.datetime.now(tz=_dt.UTC),  # acquired_token left NULL
+                ),
+            )
+
+
+async def test_dlq_preserves_timer_id(
+    pg_engine: AsyncEngine,
+    outbox_table: Table,
+    dlq_table: Table,
+) -> None:
+    """P9: a terminally-failed timer keeps its timer_id in the DLQ audit trail."""
+    broker = OutboxBroker(pg_engine, outbox_table=outbox_table, dlq_table=dlq_table)
+    handled = asyncio.Event()
+
+    @broker.subscriber("orders", retry_strategy=NoRetry(), min_fetch_interval=0.02)
+    async def handle(body: dict) -> None:
+        del body
+        handled.set()
+        boom = "always fails"
+        raise RuntimeError(boom)
+
+    session_factory = async_sessionmaker(pg_engine, expire_on_commit=False)
+    async with broker:
+        async with session_factory() as session, session.begin():
+            await broker.publish({"x": 1}, queue="orders", session=session, timer_id="email-42")
+        await asyncio.wait_for(handled.wait(), timeout=5.0)
+
+    rows = await _dlq_rows(pg_engine, dlq_table)
+    assert len(rows) == 1
+    assert rows[0]["timer_id"] == "email-42"
+
+
 async def test_dlq_insert_failure_rolls_back_delete(
     pg_engine: AsyncEngine,
     outbox_table: Table,
diff --git a/tests/test_metrics_prometheus.py b/tests/test_metrics_prometheus.py
index 2e7ba02..0fa6b6a 100644
--- a/tests/test_metrics_prometheus.py
+++ b/tests/test_metrics_prometheus.py
@@ -132,6 +132,35 @@ def test_prometheus_published_event_uses_destination_label() -> None:
     )
 
 
+def test_prometheus_published_error_status_total_fires_at_count_zero() -> None:
+    """
+    P28: a status="error" published event (count=0) must increment the error-status total.
+
+    The old ``if count > 0`` gate left ``published_messages_total{status="error"}`` — the
+    exact series dashboards alert on — permanently at zero.
+    """
+    reg, rec = _make_recorder()
+    rec(
+        "published",
+        {
+            "queue": "q",
+            "status": "error",
+            "count": 0,
+            "size_bytes": 0,
+            "duration_seconds": 0.001,
+            "exception_type": "ValueError",
+        },
+    )
+    assert (
+        _sample(
+            reg,
+            "faststream_published_messages_total",
+            {"app_name": "", "broker": "outbox", "destination": "q", "status": "error"},
+        )
+        == 1.0
+    )
+
+
 def test_prometheus_published_duration_uses_destination_label() -> None:
     reg, rec = _make_recorder()
     rec(
diff --git a/tests/test_relay.py b/tests/test_relay.py
index 0fe7708..fd7da83 100644
--- a/tests/test_relay.py
+++ b/tests/test_relay.py
@@ -286,3 +286,29 @@ async def relay(body: dict[str, Any]) -> dict[str, Any]:
         f"Expected exactly one WARNING referencing relay_queue, got {len(matching)}: "
         f"{[r.getMessage() for r in caplog.records]}"
     )
+
+
+async def test_foreign_warning_names_only_decorated_subscriber_queues(caplog: pytest.LogCaptureFixture) -> None:
+    """P21: the unstarted-foreign-broker warning names the decorated subscriber's queue, not every queue."""
+    metadata = MetaData()
+    outbox_table = make_outbox_table(metadata)
+    broker_outbox = OutboxBroker(outbox_table=outbox_table)
+    broker_kafka = KafkaBroker("kafka://test:9092")
+    publisher_kafka = broker_kafka.publisher("relay_topic")
+
+    @publisher_kafka
+    @broker_outbox.subscriber("relay_queue")
+    async def relay(body: dict[str, Any]) -> dict[str, Any]:
+        return body  # pragma: no cover  # never invoked — test only exercises start()
+
+    @broker_outbox.subscriber("unrelated_queue")
+    async def plain(body: dict[str, Any]) -> None: ...  # a second, foreign-publisher-free subscriber
+
+    with caplog.at_level(logging.WARNING, logger="faststream_outbox"):
+        async with TestOutboxBroker(broker_outbox, run_loops=False):
+            pass
+
+    foreign = [r.getMessage() for r in caplog.records if "has not been started" in r.getMessage()]
+    assert len(foreign) == 1
+    assert "relay_queue" in foreign[0]
+    assert "unrelated_queue" not in foreign[0]  # the old code listed every subscriber's queues
diff --git a/tests/test_unit.py b/tests/test_unit.py
index e020b79..0f49291 100644
--- a/tests/test_unit.py
+++ b/tests/test_unit.py
@@ -50,6 +50,7 @@
     _TRUNCATION_SUFFIX,
     OutboxSubscriber,
     _compute_backoff,
+    _OutboxConfigError,
 )
 from faststream_outbox.testing import FakeOutboxClient, FakeOutboxProducer
 
@@ -136,13 +137,22 @@ def test_make_outbox_table_attaches_to_metadata() -> None:
 
 
 def test_make_outbox_table_accepts_max_length_name() -> None:
-    # 56 ASCII bytes + "outbox_" (7) = 63 — exactly the Postgres limit.
+    # P7: the binding identifier is the longest derived name — "<t>_pending_idx" /
+    # "<t>_timer_id_uq" (12-byte suffix), not the "outbox_" channel prefix (7). So the
+    # longest valid table_name is 63 - 12 = 51 bytes.
     metadata = MetaData()
-    name = "a" * 56
+    name = "a" * 51
     t = make_outbox_table(metadata, table_name=name)
     assert t.name == name
 
 
+def test_make_outbox_table_rejects_name_that_fits_channel_but_overflows_index() -> None:
+    """P7: a 52-byte name fits the NOTIFY channel (7+52=59) but overflows the index name (52+12=64)."""
+    metadata = MetaData()
+    with pytest.raises(ValueError, match="63 bytes"):
+        make_outbox_table(metadata, table_name="a" * 52)
+
+
 def test_make_outbox_table_rejects_oversize_name() -> None:
     metadata = MetaData()
     with pytest.raises(ValueError, match="63 bytes"):
@@ -205,6 +215,18 @@ def test_encode_payload_allows_user_content_type_for_bytes_body() -> None:
     assert headers["content-type"] == "application/octet-stream"
 
 
+def test_encode_payload_raises_on_correlation_id_conflict() -> None:
+    """P2: an explicit correlation_id that mismatches headers['correlation_id'] is a conflict (was silently dropped)."""
+    with pytest.raises(ValueError, match="correlation_id"):
+        _encode_payload({"x": 1}, correlation_id="kwarg-id", headers={"correlation_id": "header-id"})
+
+
+def test_encode_payload_correlation_id_matching_header_is_ok() -> None:
+    """P2: kwarg == header is not a conflict."""
+    _, headers = _encode_payload({"x": 1}, correlation_id="same", headers={"correlation_id": "same"})
+    assert headers["correlation_id"] == "same"
+
+
 class _PydanticBody(BaseModel):
     order_id: int
     name: str
@@ -340,6 +362,77 @@ async def test_inner_message_nack_with_no_strategy_is_terminal() -> None:
     assert msg.to_delete
 
 
+def test_constant_retry_rejects_oversize_jitter() -> None:
+    """P23: jitter_factor > 2 would make a jittered delay negative (a hot retry)."""
+    with pytest.raises(ValueError, match="jitter_factor"):
+        ConstantRetry(delay_seconds=1.0, jitter_factor=2.5)
+
+
+def test_constant_retry_rejects_non_positive_delay() -> None:
+    """P23: a non-positive delay is a hot retry."""
+    with pytest.raises(ValueError, match="delay_seconds"):
+        ConstantRetry(delay_seconds=0.0)
+
+
+def test_exponential_retry_rejects_non_positive_initial_delay() -> None:
+    """P23: ExponentialRetry needs a positive initial delay."""
+    with pytest.raises(ValueError, match="initial_delay_seconds"):
+        ExponentialRetry(initial_delay_seconds=0.0)
+
+
+def test_retry_rejects_zero_max_attempts() -> None:
+    """P23: max_attempts < 1 means 'never retry' expressed confusingly — reject it."""
+    with pytest.raises(ValueError, match="max_attempts"):
+        ConstantRetry(delay_seconds=1.0, max_attempts=0)
+
+
+def test_retry_rejects_non_positive_max_total_delay() -> None:
+    """P23: max_total_delay_seconds must be > 0 if set."""
+    with pytest.raises(ValueError, match="max_total_delay_seconds"):
+        ConstantRetry(delay_seconds=1.0, max_total_delay_seconds=0.0)
+
+
+def test_linear_retry_rejects_non_positive_initial_delay() -> None:
+    """P23: LinearRetry needs a positive initial delay."""
+    with pytest.raises(ValueError, match="initial_delay_seconds"):
+        LinearRetry(initial_delay_seconds=0.0, step_seconds=1.0)
+
+
+def test_linear_retry_rejects_negative_step() -> None:
+    """P23: a negative step would shrink delays toward zero (hot retry)."""
+    with pytest.raises(ValueError, match="step_seconds"):
+        LinearRetry(initial_delay_seconds=1.0, step_seconds=-1.0)
+
+
+def test_exponential_retry_rejects_non_positive_multiplier() -> None:
+    """P23: a non-positive multiplier is nonsensical for exponential backoff."""
+    with pytest.raises(ValueError, match="multiplier"):
+        ExponentialRetry(initial_delay_seconds=1.0, multiplier=0.0)
+
+
+def test_exponential_retry_rejects_non_positive_max_delay() -> None:
+    """P23: max_delay_seconds must be > 0 if set."""
+    with pytest.raises(ValueError, match="max_delay_seconds"):
+        ExponentialRetry(initial_delay_seconds=1.0, max_delay_seconds=0.0)
+
+
+def test_warn_on_duplicate_queues_across_routers() -> None:
+    """P22: a duplicate queue introduced via include_router is caught at start()-time."""
+    broker = _make_broker()
+
+    @broker.subscriber("orders")
+    async def h1(body: dict) -> None: ...
+
+    router = OutboxRouter()
+
+    @router.subscriber("orders")  # same queue, via a router → invisible to registration-time check
+    async def h2(body: dict) -> None: ...
+
+    broker.include_router(router)
+    with pytest.warns(UserWarning, match="compete for the same rows"):
+        broker._warn_on_duplicate_queues()  # noqa: SLF001
+
+
 async def test_inner_message_nack_with_strategy_schedules_retry() -> None:
     msg = _make_msg(retry_strategy=ConstantRetry(delay_seconds=60))
     await msg.nack()
@@ -659,6 +752,60 @@ async def test_publish_command_batch_bodies_preserves_none_body() -> None:
     assert OutboxPublishCommand(b"a", b"b", queue="q", session=session).batch_bodies == (b"a", b"b")
 
 
+def test_publish_command_rejects_empty_queue() -> None:
+    """P5: queue validation in the single-source-of-truth constructor."""
+    with pytest.raises(ValueError, match="non-empty"):
+        OutboxPublishCommand(b"x", queue="", session=_make_session_mock())
+
+
+def test_publish_command_rejects_non_str_queue() -> None:
+    """P5: a non-str queue is a TypeError, not an opaque SQL error."""
+    with pytest.raises(TypeError, match="queue must be a str"):
+        OutboxPublishCommand(b"x", queue=123, session=_make_session_mock())  # ty: ignore[invalid-argument-type]
+
+
+def test_publish_command_rejects_oversize_queue() -> None:
+    """P5: queue over the String(255) column limit is rejected up front."""
+    with pytest.raises(ValueError, match="255"):
+        OutboxPublishCommand(b"x", queue="q" * 256, session=_make_session_mock())
+
+
+def test_publish_command_rejects_timer_id_for_batch() -> None:
+    """P4: timer_id is meaningless for a batch (multiple bodies) — reject, don't silently drop."""
+    with pytest.raises(ValueError, match="batch"):
+        OutboxPublishCommand(b"a", b"b", queue="q", session=_make_session_mock(), timer_id="t")
+
+
+def test_publish_command_rejects_correlation_id_for_batch() -> None:
+    """P4: correlation_id is per-row single-publish only — reject on a batch command."""
+    with pytest.raises(ValueError, match="batch"):
+        OutboxPublishCommand(b"a", b"b", queue="q", session=_make_session_mock(), correlation_id="c")
+
+
+async def test_broker_publish_batch_empty_still_rejects_non_async_session() -> None:
+    """P1: the session-type check fires even on an empty batch (no command is built there)."""
+    broker = _make_broker()
+    with pytest.raises(TypeError, match="AsyncSession"):
+        await broker.publish_batch(queue="orders", session=object())  # ty: ignore[invalid-argument-type]  # no bodies
+
+
+async def test_producer_publish_emits_error_metric_on_encode_failure() -> None:
+    """P3: an encode failure (content-type conflict) still emits the ``published`` error metric."""
+    events, recorder = _events_recorder()
+    metadata = MetaData()
+    table = make_outbox_table(metadata)
+    producer = OutboxProducer(table=table, parser=None, decoder=None, metrics_recorder=recorder)
+    cmd = OutboxPublishCommand(
+        {"x": 1},
+        queue="orders",
+        session=_make_session_mock(),
+        headers={"content-type": "text/plain"},
+    )
+    with pytest.raises(ValueError, match="content-type"):
+        await producer.publish(cmd)
+    assert any(ev == "published" and tags.get("status") == "error" for ev, tags in events)
+
+
 async def test_publish_command_rejects_non_async_session() -> None:
     with pytest.raises(TypeError, match="AsyncSession"):
         OutboxPublishCommand(b"x", queue="orders", session=object())  # ty: ignore[invalid-argument-type]
@@ -1440,12 +1587,6 @@ async def test_fake_client_validate_schema_raises_and_ping_passes() -> None:
 # --- OutboxBrokerConfig connect/disconnect (no-op stubs) ---
 
 
-async def test_outbox_broker_config_connect_disconnect_noop() -> None:
-    cfg = OutboxBrokerConfig()
-    await cfg.connect()  # must not raise
-    await cfg.disconnect()  # must not raise
-
-
 # --- subscriber get_one + _make_response_publisher ---
 
 
@@ -1503,6 +1644,42 @@ async def handle(body: dict) -> None: ...
     return next(iter(broker._subscribers))  # noqa: SLF001
 
 
+def test_on_notify_wakes_only_for_served_queues() -> None:
+    """P14: NOTIFY payload is the queue name; only wake for queues this subscriber serves."""
+    sub = _make_subscriber_for_listener_test()  # serves "orders"
+    sub._notify_event.clear()  # noqa: SLF001
+    sub._on_notify(None, 123, "outbox_outbox", "orders")  # noqa: SLF001  # served
+    assert sub._notify_event.is_set()  # noqa: SLF001
+
+    sub._notify_event.clear()  # noqa: SLF001
+    sub._on_notify(None, 123, "outbox_outbox", "other-queue")  # noqa: SLF001  # not served
+    assert not sub._notify_event.is_set()  # noqa: SLF001
+
+
+def test_on_notify_wakes_conservatively_on_unexpected_payload() -> None:
+    """P14: an unexpected callback shape wakes (don't risk a missed delivery)."""
+    sub = _make_subscriber_for_listener_test()
+    sub._notify_event.clear()  # noqa: SLF001
+    sub._on_notify()  # noqa: SLF001  # no args
+    assert sub._notify_event.is_set()  # noqa: SLF001
+
+
+async def test_stop_awaits_cancelled_tasks() -> None:
+    """P16: stop() awaits the cancelled loop tasks so they've unwound before it returns."""
+    sub = _make_subscriber_for_listener_test()
+    sub.running = True
+
+    async def _forever() -> None:
+        await asyncio.sleep(3600)
+
+    sub.add_task(_forever)
+    await asyncio.sleep(0)  # let the task start (reach its await) before we cancel it
+    tasks = list(sub.tasks)
+    assert tasks
+    await sub.stop()
+    assert all(t.done() for t in tasks)  # gather() awaited the cancellations
+
+
 async def test_open_listen_connection_returns_none_for_non_asyncpg_driver() -> None:
     sub = _make_subscriber_for_listener_test()
     engine = MagicMock()
@@ -1848,6 +2025,65 @@ async def handle(body: dict) -> None: ...
     assert not any(e.startswith("nacked") for e, _ in events)
 
 
+async def test_dispatch_one_lease_lost_emits_only_lease_lost_not_acked() -> None:
+    """
+    P17: a lease-lost terminal flush emits ``lease_lost`` only — not a false ``acked`` that double-counts.
+
+    Before P17 the acked/nacked metric fired before the flush, so a row whose lease was
+    reclaimed (flush rowcount 0 → redelivered) was counted once here and again on redelivery.
+    """
+    events, recorder = _events_recorder()
+    metadata = MetaData()
+    table = make_outbox_table(metadata)
+    broker = OutboxBroker(outbox_table=table, metrics_recorder=recorder)
+
+    @broker.subscriber("orders")
+    async def handle(body: dict) -> None: ...
+
+    fake = FakeOutboxClient()
+    test_broker = TestOutboxBroker(broker)
+    test_broker.fake_client = fake
+    msg = _make_msg(queue="orders")
+
+    async with test_broker:
+        sub = next(iter(broker._subscribers))  # noqa: SLF001
+        # Lease lost: the delete finds no matching (id, token) row → rowcount 0.
+        with patch.object(fake, "delete_with_lease", new=AsyncMock(return_value=False)):
+            await sub.dispatch_one(msg, writer_conn=None)
+
+    names = [e for e, _ in events]
+    assert "lease_lost" in names
+    assert "acked" not in names  # the false ack that double-counted is gone
+
+
+async def test_worker_inner_swallows_config_error_without_reconnect() -> None:
+    """
+    P18: an _OutboxConfigError in the worker loop is logged and swallowed, not propagated.
+
+    Letting it reach _run_with_reconnect would tear down the writer connection and back
+    off (throttling unrelated rows). The worker must continue; the row's lease expires.
+    """
+    fake = FakeOutboxClient()
+    broker, test_broker = _make_broker_for_dispatch(fake)
+    async with test_broker:
+        sub = next(iter(broker._subscribers))  # noqa: SLF001
+        sub.running = True
+        sub._inflight.put_nowait(_make_msg(queue="orders"))  # noqa: SLF001
+        calls = {"n": 0}
+
+        async def _raise_then_stop(row: object, *, writer_conn: object) -> None:
+            del row, writer_conn
+            calls["n"] += 1
+            sub.running = False  # let the worker loop exit after this row
+            msg = "bad relay chain"
+            raise _OutboxConfigError(msg)
+
+        with patch.object(sub, "dispatch_one", new=_raise_then_stop):
+            await sub._worker_inner(writer_conn=None)  # noqa: SLF001  # must NOT raise
+
+    assert calls["n"] == 1
+
+
 async def test_dispatch_one_max_deliveries_emits_terminal_without_dispatched() -> None:
     """
     T7: the max_deliveries terminal emits nacked_terminal(reason=max_deliveries) with NO preceding 'dispatched'.
@@ -2399,6 +2635,27 @@ def test_subscriber_rejects_min_above_max_fetch_interval() -> None:
         _register_subscriber(broker, min_fetch_interval=10.0, max_fetch_interval=1.0)
 
 
+def test_subscriber_rejects_non_positive_min_fetch_interval() -> None:
+    """P12: a non-positive min_fetch_interval would busy-poll."""
+    broker = _make_broker()
+    with pytest.raises(ValueError, match="min_fetch_interval must be > 0"):
+        _register_subscriber(broker, min_fetch_interval=0.0)
+
+
+def test_subscriber_rejects_non_positive_max_fetch_interval() -> None:
+    """P12: a non-positive max_fetch_interval would busy-poll."""
+    broker = _make_broker()
+    with pytest.raises(ValueError, match="max_fetch_interval must be > 0"):
+        _register_subscriber(broker, max_fetch_interval=0.0)
+
+
+def test_subscriber_rejects_non_positive_lease_ttl() -> None:
+    """P12: a non-positive lease_ttl_seconds means an instantly-expiring lease."""
+    broker = _make_broker()
+    with pytest.raises(ValueError, match="lease_ttl_seconds must be > 0"):
+        _register_subscriber(broker, lease_ttl_seconds=0.0)
+
+
 def test_subscriber_rejects_ack_first() -> None:
     # ACK_FIRST has no legitimate outbox use — deletes before the handler runs, so a
     # handler crash silently drops the row. Better to refuse than warn-and-ship.
@@ -2618,7 +2875,10 @@ async def test_metrics_max_deliveries_emits_terminal_reason() -> None:
 
     async with test_broker:
         sub = next(iter(broker._subscribers))  # noqa: SLF001
-        await sub.dispatch_one(msg, writer_conn=None)
+        # P17: the outcome metric now emits only after a successful flush, so give the
+        # delete something to land on (the synthetic row isn't in the fake store).
+        with patch.object(fake, "delete_with_lease", new=AsyncMock(return_value=True)):
+            await sub.dispatch_one(msg, writer_conn=None)
 
     terminals = [t for e, t in events if e == "nacked_terminal"]
     assert len(terminals) == 1
@@ -2817,6 +3077,24 @@ async def test_dlq_cte_uses_schema_qualified_table_names() -> None:
         await engine.dispose()
 
 
+async def test_delete_with_lease_raises_when_dlq_payload_but_no_dlq_table() -> None:
+    """P10: a dlq_payload with no dlq_table must raise, not silently degrade to a plain DELETE."""
+    engine = create_async_engine("postgresql+asyncpg://u:p@localhost/db")
+    try:
+        metadata = MetaData()
+        table = make_outbox_table(metadata)
+        client = OutboxClient(engine, table)  # no dlq_table configured
+        with pytest.raises(RuntimeError, match="dlq_table"):
+            await client.delete_with_lease(
+                MagicMock(),  # non-None conn; the guard raises before it's used
+                1,
+                uuid.uuid4(),
+                dlq_payload={"failure_reason": "rejected", "last_exception": None},
+            )
+    finally:
+        await engine.dispose()
+
+
 async def test_fetch_cte_carries_partial_index_predicates_as_conjuncts() -> None:
     """
     T6: each OR arm of the fetch WHERE must carry its partial-index predicate as a conjunct.
@@ -2852,6 +3130,9 @@ async def execute(self, stmt: object, _params: object) -> object:
         sql = str(captured["stmt"].compile(dialect=postgresql.dialect())).lower()
         assert "acquired_token is null" in sql  # Branch A predicate
         assert "acquired_token is not null" in sql  # Branch B conjunct (the naive form drops this)
+        # P11: queues bound as a single ANY(:queues) array (stable SQL across queue counts),
+        # not an OR-of-N-equalities.
+        assert "= any (" in sql  # ANY(:queues) array, not OR-of-equalities
     finally:
         await engine.dispose()
 
@@ -2870,6 +3151,7 @@ def test_make_dlq_table_columns_present() -> None:
         "failed_at",
         "failure_reason",
         "last_exception",
+        "timer_id",
     }
     assert {c.name for c in t.columns} == expected
     assert t.name == "my_dlq"