diff --git a/docs/operations/troubleshooting.md b/docs/operations/troubleshooting.md
index 28adf78..bd3405a 100644
--- a/docs/operations/troubleshooting.md
+++ b/docs/operations/troubleshooting.md
@@ -17,6 +17,7 @@ owns the underlying design.
 | [`AckPolicy.ACK_FIRST` raises `ValueError` at registration](#ackpolicyack_first-raises-valueerror-at-registration) | By design (would defeat outbox reliability) |
 | [`OutboxResponse(...)` + foreign-publisher decorator logs a configuration error](#outboxresponse-foreign-publisher-decorator-config-error) | By design (dual-fire footgun) |
 | [Chained `OutboxResponse` retries after the handler "succeeded"](#outboxresponse-relay-publish-failure) | Follow-on publish fails post-handler; nacks the inbound row |
+| [`OutboxResponse` raises "no open transaction"](#outboxresponse-no-open-transaction) | The response's session closed before the publish |
 | [`validate_schema()` raises `ImportError`](#validate_schema-raises-importerror) | `[validate]` extra not installed |
 
 ## `event=lease_lost` recurring in logs { #event-lease_lost-recurring-in-logs }
@@ -245,14 +246,14 @@ See [Subscriber § Ack policy](../usage/subscriber.md#ack-policy).
 
 A handler with both `@kafka_pub` and an `OutboxResponse(...)` return
 value logs an ERROR on every dispatch:
-`Outbox configuration error (fix required; row left to lease-expiry retry)`.
+`Outbox configuration error (fix required; row retried per its strategy)`.
 
 This is by design. The combination would both insert a row into the
 outbox *and* publish to Kafka, a dual-fire that doubles delivery. The
 subscriber refuses the chain composition after the handler returns. The
-worker logs the error and moves on without nacking, so the retry
-strategy is not consulted. The row's lease expires, a later fetch
-reclaims it, and the cycle repeats until the configuration is fixed.
+worker logs the error and writes the nack the ack middleware recorded, so
+the row follows its retry strategy until the configuration is fixed (or,
+with `max_deliveries` set, until it goes terminal).
 
 To confirm, inspect the handler's decorator stack and return type.
 
@@ -291,6 +292,24 @@ a redelivery's insert is a no-op.
 
 See [Publisher § Chained publishing](../usage/publisher.md#chained-publishing).
 
+## `OutboxResponse` raises "no open transaction" { #outboxresponse-no-open-transaction }
+
+A handler that returns `OutboxResponse(...)` logs an ERROR on every
+dispatch saying the response's session has no open transaction, and the
+inbound row keeps retrying.
+
+The response is published after the handler returns. A session from
+`Depends(...)`, or from an `async with` block inside the handler, is
+already closed by then, so the follow-on row would sit in a transaction
+nobody commits. The subscriber refuses to publish it and the inbound row
+follows its retry strategy.
+
+To fix it, open the session in a broker middleware so it outlives the
+handler, or call `broker.publish(..., session=session)` inside the
+handler instead of returning `OutboxResponse`.
+
+See [Publisher § Chained publishing](../usage/publisher.md#chained-publishing).
+
 ## `validate_schema()` raises `ImportError` { #validate_schema-raises-importerror }
 
 Calling `await broker.validate_schema()` raises:
diff --git a/docs/usage/fastapi.md b/docs/usage/fastapi.md
index d7d8179..6689a0b 100644
--- a/docs/usage/fastapi.md
+++ b/docs/usage/fastapi.md
@@ -96,9 +96,12 @@ delivery, opened in a `session.begin()` block, committed on handler return,
 rolled back on exception.
 
 A handler's `AsyncSession` is therefore resolved exactly as in an HTTP
-route (a fresh session per delivery, not a shared instance), and
-`OutboxResponse(session=...)` commits the follow-on row with the handler's
-domain writes. See [Chained
+route (a fresh session per delivery, not a shared instance). Publish
+follow-on rows with `router.broker.publish(..., session=session)` inside
+the handler so they commit with its domain writes. Returning
+`OutboxResponse(session=session)` with this session does not work: the
+dependency is torn down before the response is published, so the dispatch
+raises and the inbound row is retried. See [Chained
 publishing](./publisher.md#chained-publishing).
 
 ## Annotated context shortcuts
diff --git a/docs/usage/publisher.md b/docs/usage/publisher.md
index 3060e16..7313c73 100644
--- a/docs/usage/publisher.md
+++ b/docs/usage/publisher.md
@@ -149,35 +149,46 @@ with your domain writes):
 
 !!! note "The `session` must outlive the handler return"
     The returned `OutboxResponse` is published after the handler
-    returns, so its `session` must still be open at that point. The
-    requirement is about session *lifetime*, not any particular framework:
-    provide the session through a dependency that the framework tears down
-    *after* the response flow. FastAPI's `Depends(get_session)` or
-    FastStream's own `Depends` / `Context` session both do this. Opening
-    your own `async with session_factory() as session:` inside the handler
-    does not work here: that session closes on `return`, before the row
-    is inserted. In that case, call `broker.publish(..., session=session)`
-    directly inside the `async with` instead (see
-    [§ Not a relay decorator](#not-a-relay-decorator)).
+    returns, so its `session` must still hold an open transaction at that
+    point. A session from `Depends(...)` does not: FastStream and FastAPI
+    both tear dependencies down inside the handler call, before the
+    response is published. The same goes for an
+    `async with session_factory() as session:` block inside the handler.
+    In those cases the dispatch raises `RuntimeError` naming the missing
+    transaction, logs it at ERROR, and retries the inbound row. Either
+    open the session in a broker middleware, as below, or call
+    `broker.publish(..., session=session)` inside the handler instead
+    (see [§ Not a relay decorator](#not-a-relay-decorator)).
+
+A broker middleware's `on_receive` / `after_processed` hooks wrap the
+whole dispatch, response publish included, so a session opened there
+covers both the handler's writes and the follow-on row. The commit runs
+before the ack decision, so a failed commit retries the inbound row:
 
 ```python
-from fastapi import Depends
+from faststream import BaseMiddleware, Context
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from faststream_outbox import OutboxResponse
-from faststream_outbox.annotations import OutboxMessage
+from faststream_outbox import OutboxBroker, OutboxResponse
 
-# `get_session` is your app's session dependency — the same one your HTTP
-# routes use. `Depends(...)` resolves inside a handler only under the FastAPI
-# integration; see the engine / sessionmaker setup in the FastAPI guide.
+
+class SessionPerMessage(BaseMiddleware):
+    async def on_receive(self) -> None:
+        self._begin = session_factory.begin()
+        self.context.set_local("session", await self._begin.__aenter__())
+        await super().on_receive()
+
+    async def after_processed(self, exc_type=None, exc_val=None, exc_tb=None) -> bool | None:
+        await self._begin.__aexit__(exc_type, exc_val, exc_tb)
+        return await super().after_processed(exc_type, exc_val, exc_tb)
+
+
+broker = OutboxBroker(engine, outbox_table=outbox_table, middlewares=[SessionPerMessage])
 
 
 @broker.subscriber("orders")
-async def handle(
-    msg: OutboxMessage,
-    session: AsyncSession = Depends(get_session),
-) -> OutboxResponse:
-    ...  # process inbound
+async def handle(body: dict, session: AsyncSession = Context()) -> OutboxResponse:
+    ...  # domain writes on `session`
     return OutboxResponse(
         body={"chained": True},
         queue="downstream",
@@ -185,8 +196,9 @@ async def handle(
     )
 ```
 
-`correlation_id` propagates from the inbound message if you don't set one
-explicitly, which is useful for trace stitching. Plain returns (`None`, `dict`,
+`correlation_id` comes from `headers["correlation_id"]` when you set the
+header but not the argument, and otherwise from the inbound message, which
+is useful for trace stitching. Plain returns (`None`, `dict`,
 etc.) are silently skipped, so handlers that don't want to chain just
 return normally.
 
diff --git a/faststream_outbox/publisher/fake.py b/faststream_outbox/publisher/fake.py
index 18a4175..ae0bb7d 100644
--- a/faststream_outbox/publisher/fake.py
+++ b/faststream_outbox/publisher/fake.py
@@ -19,7 +19,7 @@
 
 from faststream.response.publish_type import PublishType
 
-from faststream_outbox.response import OutboxPublishCommand
+from faststream_outbox.response import OutboxPublishCommand, _OutboxConfigError
 
 
 if typing.TYPE_CHECKING:
@@ -43,6 +43,14 @@ async def _publish(
     ) -> typing.Any:
         if not isinstance(cmd, OutboxPublishCommand):
             return None
+        if not cmd.session.in_transaction():
+            msg = (
+                "OutboxResponse is published after the handler returns, but its session has no open "
+                "transaction by then, so the follow-on row would never commit. Open the session and its "
+                "transaction in a broker middleware so they outlive the handler, or call "
+                "broker.publish(..., session=session) inside the handler instead."
+            )
+            raise _OutboxConfigError(msg)
         cmd.publish_type = PublishType.REPLY
         # ty diagnostic: ``producer.publish`` narrows to ``OutboxPublishCommand``
         # while ``PublisherMiddleware`` is generic over the base ``PublishCommand``.
diff --git a/faststream_outbox/response.py b/faststream_outbox/response.py
index b880b0f..a9e612f 100644
--- a/faststream_outbox/response.py
+++ b/faststream_outbox/response.py
@@ -19,6 +19,8 @@
 from faststream.response.response import BatchPublishCommand, PublishCommand, Response
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from faststream_outbox.message import CORRELATION_ID_HEADER
+
 
 # Matches the ``queue`` column width in ``make_outbox_table`` (``String(255)``).
 _MAX_QUEUE_LENGTH = 255
@@ -68,6 +70,14 @@ def _validate_publish_args(
         raise ValueError(msg)
 
 
+# Marker exception raised by programming guards on the dispatch path (e.g. the
+# OutboxResponse + foreign-publisher dual-fire guard). Inherits from
+# ``RuntimeError`` so ``pytest.raises(RuntimeError, ...)`` in tests catches it, but
+# is a distinct subclass so ``consume()`` and ``dispatch_one`` can re-raise it while
+# still swallowing plain handler-raised ``RuntimeError``s.
+class _OutboxConfigError(RuntimeError): ...
+
+
 class OutboxPublishCommand(BatchPublishCommand):
     """Outbox-specific publish command: carries session + scheduling fields end-to-end."""
 
@@ -166,9 +176,13 @@ class OutboxResponse(Response):
     re-runs the same validator on ``as_publish_command()``, so it stays the authoritative
     single source of truth and the eager checks can never drift from it.
 
-    ``correlation_id`` defaults to the inbound message's correlation_id when not set
-    — FastStream's ``SubscriberUsecase.process_message`` does the inheritance before
-    publishing the response.
+    ``correlation_id`` defaults to ``headers["correlation_id"]``, then to the inbound
+    message's correlation_id, which FastStream's ``SubscriberUsecase.process_message``
+    fills in before publishing the response.
+
+    The response is published after the handler returns, so ``session`` must still
+    hold an open transaction then; otherwise the dispatch raises ``RuntimeError`` and
+    the inbound row is retried.
     """
 
     def __init__(
@@ -190,6 +204,8 @@ def __init__(
             activate_in=activate_in,
             activate_at=activate_at,
         )
+        if correlation_id is None and headers:
+            correlation_id = headers.get(CORRELATION_ID_HEADER)
         super().__init__(body=body, headers=headers, correlation_id=correlation_id)
         self.queue = queue
         self.session = session
diff --git a/faststream_outbox/subscriber/usecase.py b/faststream_outbox/subscriber/usecase.py
index 5b269f2..5feedae 100644
--- a/faststream_outbox/subscriber/usecase.py
+++ b/faststream_outbox/subscriber/usecase.py
@@ -43,7 +43,7 @@
 from faststream_outbox.message import ENVELOPE_MANAGED_HEADERS, OutboxInnerMessage, Retry, Terminal
 from faststream_outbox.parser.parser import OutboxParser
 from faststream_outbox.publisher.fake import OutboxFakePublisher
-from faststream_outbox.response import OutboxResponse
+from faststream_outbox.response import OutboxResponse, _OutboxConfigError
 from faststream_outbox.subscriber.config import OutboxSubscriberConfig, OutboxSubscriberSpecificationConfig
 
 
@@ -79,14 +79,6 @@ class _PendingFlush:
     tags: dict[str, typing.Any]
 
 
-# Marker exception raised by programming guards inside ``process_message`` (e.g.
-# the OutboxResponse + foreign-publisher dual-fire guard). Inherits from
-# ``RuntimeError`` so ``pytest.raises(RuntimeError, ...)`` in tests catches it, but
-# is a distinct subclass so ``consume()`` and ``dispatch_one`` can re-raise it while
-# still swallowing plain handler-raised ``RuntimeError``s.
-class _OutboxConfigError(RuntimeError): ...
-
-
 # Cap for the ``last_exception`` string written to the DLQ. Some exceptions carry
 # huge payloads (validation errors with the full request body, asyncpg ``DataError``
 # with the rejected row, etc.). An unbounded ``repr`` would extend the writer
@@ -643,11 +635,11 @@ async def _worker_inner(self, *, writer_conn: "AsyncConnection | None") -> None:
                     # 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.
+                    # rows. dispatch_one already wrote the recorded outcome, so the retry
+                    # strategy applies; log it and continue until the configuration is fixed.
                     self._log(
                         log_level=logging.ERROR,
-                        message=f"Outbox configuration error (fix required; row left to lease-expiry retry): {e!r}",
+                        message=f"Outbox configuration error (fix required; row retried per its strategy): {e!r}",
                         exc_info=e,
                     )
                 finally:
@@ -722,14 +714,12 @@ async def dispatch_one(  # linear pipeline: guard, consume, branch on outcome, f
         try:
             await self.consume(row)
         except _OutboxConfigError:
-            # _OutboxConfigError escaping consume() is a programming guard (e.g.
-            # the OutboxResponse + foreign-publisher dual-fire guard). Re-raise so
-            # callers see the error immediately — sync test dispatch surfaces it
-            # from outbox.publish; the worker loop's outer except re-raises it via
-            # _run_with_reconnect so it's logged at ERROR level with a reconnect
-            # backoff rather than silently ignored.
+            # A programming guard tripped after the handler ran. Write the outcome the ack
+            # middleware recorded, then re-raise: sync test dispatch surfaces it from
+            # outbox.publish, and _worker_inner logs it at ERROR. Inline (buffer=None),
+            # because a buffered row would be task_done'd twice once the error unwinds.
             row.handler_duration_seconds = time.perf_counter() - start_perf
-            self._emit_dispatch_abandoned(row)
+            await self._write_outcome(row, base, buffer=None, writer_conn=writer_conn)
             raise
         except Exception as e:  # noqa: BLE001
             # No ack/nack metric here: the row was never marked terminal/retry, so
@@ -745,6 +735,18 @@ async def dispatch_one(  # linear pipeline: guard, consume, branch on outcome, f
             self._emit_dispatch_abandoned(row)
             return False
         row.handler_duration_seconds = time.perf_counter() - start_perf
+        return await self._write_outcome(row, base, buffer=buffer, writer_conn=writer_conn)
+
+    async def _write_outcome(
+        self,
+        row: OutboxInnerMessage,
+        base: dict[str, typing.Any],
+        *,
+        buffer: "list[_PendingFlush] | None",
+        writer_conn: "AsyncConnection | None",
+    ) -> bool:
+        """Write the outcome recorded during consume and emit its metric. Returns True iff buffered."""
+        logger = self._outer_config.logger.logger.logger if self._outer_config.logger else None
         # Shutdown race: SubscriberUsecase.consume() returns None without invoking
         # process_message when self.running has been flipped to False by stop().
         # Detecting that here lets us preserve the row instead of falling through
@@ -1006,10 +1008,9 @@ async def consume(self, msg: OutboxInnerMessage) -> typing.Any:
         ``OutboxResponse + foreign-publisher`` dual-fire check in
         ``process_message``) raise ``_OutboxConfigError`` (a ``RuntimeError``
         subclass) so the guard is distinguishable from handler-raised
-        ``RuntimeError``s. Re-raising here lets ``dispatch_one`` surface it to
-        the caller (sync test dispatch exposes it from ``outbox.publish``; the
-        worker loop re-raises so the error is logged and the lease expires rather
-        than silently swallowing a configuration mistake).
+        ``RuntimeError``s. Re-raising here lets ``dispatch_one`` write the recorded
+        outcome and surface the error: sync test dispatch exposes it from
+        ``outbox.publish``, and the worker loop logs it at ERROR.
 
         # Upstream equivalent (extended):
         #   SubscriberUsecase.consume
diff --git a/tests/test_fake.py b/tests/test_fake.py
index 6d69825..c377e50 100644
--- a/tests/test_fake.py
+++ b/tests/test_fake.py
@@ -6,7 +6,7 @@
 import warnings as _warnings
 from collections.abc import Awaitable, Callable, Mapping
 from types import TracebackType
-from unittest.mock import AsyncMock, patch
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 from faststream import BaseMiddleware, FastStream, StreamMessage
@@ -2375,3 +2375,63 @@ async def handle(body: str) -> None:
         await _wait_until(lambda: not test_broker.fake_client.rows, timeout=5.0)
 
     assert sorted(handled) == sorted(f"m{i}" for i in range(30))
+
+
+async def test_handler_returning_outbox_response_takes_correlation_id_from_headers() -> None:
+    broker = _make_broker()
+    downstream_cors: list[str] = []
+
+    @broker.subscriber("orders")
+    async def handle(body: dict) -> OutboxResponse:
+        del body
+        return OutboxResponse(
+            body={"v": 1}, queue="downstream", session=_fake_session(), headers={"correlation_id": "from-header"}
+        )
+
+    @broker.subscriber("downstream")
+    async def handle_next(body: dict, correlation_id: str = _Context("message.correlation_id")) -> None:
+        del body
+        downstream_cors.append(correlation_id)
+
+    async with TestOutboxBroker(broker):
+        await broker.publish({"x": 1}, queue="orders", correlation_id="inbound")  # ty: ignore[missing-argument]
+
+    assert downstream_cors == ["from-header"]
+
+
+async def test_outbox_response_on_a_session_without_a_transaction_retries_the_inbound_row() -> None:
+    broker = _make_broker()
+    closed_session = _fake_session()
+    closed_session.in_transaction = MagicMock(return_value=False)
+
+    @broker.subscriber("orders")
+    async def handle(body: dict) -> OutboxResponse:
+        return OutboxResponse(body=body, queue="downstream", session=closed_session)
+
+    test_broker = TestOutboxBroker(broker)
+    async with test_broker:
+        with pytest.raises(RuntimeError, match="transaction"):
+            await broker.publish({"x": 1}, queue="orders")  # ty: ignore[missing-argument]
+
+    assert [(r.queue, r.attempts_count, r.acquired_token) for r in test_broker.fake_client.rows] == [
+        ("orders", 1, None)
+    ]
+
+
+async def test_config_error_under_manual_ack_retries_the_row() -> None:
+    broker = _make_broker()
+    closed_session = _fake_session()
+    closed_session.in_transaction = MagicMock(return_value=False)
+
+    @broker.subscriber("orders", ack_policy=AckPolicy.MANUAL)
+    async def handle(body: dict) -> OutboxResponse:
+        return OutboxResponse(body=body, queue="downstream", session=closed_session)
+
+    test_broker = TestOutboxBroker(broker)
+    async with test_broker:
+        with pytest.raises(RuntimeError, match="transaction"):
+            await broker.publish({"x": 1}, queue="orders")  # ty: ignore[missing-argument]
+
+    assert [(r.queue, r.attempts_count, r.acquired_token) for r in test_broker.fake_client.rows] == [
+        ("orders", 1, None)
+    ]
diff --git a/tests/test_integration.py b/tests/test_integration.py
index 82ba2a6..3b5f981 100644
--- a/tests/test_integration.py
+++ b/tests/test_integration.py
@@ -10,6 +10,7 @@
 from unittest import mock
 
 import pytest
+from faststream import BaseMiddleware, Context, Depends
 from faststream.kafka import KafkaBroker, TestKafkaBroker
 from sqlalchemy import MetaData, Table, event, insert, select, text
 from sqlalchemy.exc import IntegrityError
@@ -470,6 +471,94 @@ async def test_outbox_response_followon_row_commits_with_handler_transaction(pg_
     assert [r.queue for r in rows] == ["downstream"]
 
 
+async def test_outbox_response_on_a_session_closed_before_the_publish_is_retried(
+    pg_engine: AsyncEngine,
+    outbox_table: Table,
+) -> None:
+    """INVARIANT: an ``OutboxResponse`` whose session has no open transaction is refused and the inbound row retried.
+
+    ``Depends`` tears its session down inside the handler call, before the response is published, so
+    the follow-on INSERT would land in a transaction nobody commits. Publishing anyway acks the inbound
+    row and loses the follow-on row without an error.
+    """
+    session_factory = async_sessionmaker(pg_engine, expire_on_commit=False)
+
+    async def get_session() -> Any:
+        async with session_factory() as session, session.begin():
+            yield session
+
+    session_dep = Depends(get_session)
+    broker = OutboxBroker(pg_engine, outbox_table=outbox_table)
+
+    @broker.subscriber(
+        "orders", min_fetch_interval=0.05, max_fetch_interval=0.2, retry_strategy=ConstantRetry(delay_seconds=60.0)
+    )
+    async def handle(body: dict[str, Any], session: AsyncSession = session_dep) -> OutboxResponse:
+        return OutboxResponse(body=body, queue="downstream", session=session)
+
+    errors: list[str] = []
+    async with broker:
+        sub = next(iter(broker._subscribers))  # noqa: SLF001
+        original_log = sub._log  # noqa: SLF001
+
+        def spy_log(*args: Any, **kwargs: Any) -> Any:
+            if kwargs.get("log_level") == logging.ERROR:
+                errors.append(str(kwargs.get("message", "")))
+            return original_log(*args, **kwargs)
+
+        with mock.patch.object(sub, "_log", side_effect=spy_log):
+            async with session_factory() as session, session.begin():
+                await broker.publish({"order_id": 1}, queue="orders", session=session)
+            await _wait_until(lambda: bool(errors), timeout=10.0)
+            await asyncio.sleep(0.2)
+
+    async with pg_engine.connect() as conn:
+        rows = (await conn.execute(select(outbox_table.c.queue, outbox_table.c.attempts_count))).all()
+    assert [(r.queue, r.attempts_count) for r in rows] == [("orders", 1)]
+    assert any("transaction" in e for e in errors), errors
+
+
+class _SessionPerMessage(BaseMiddleware):
+    """Open the handler's session and transaction around the whole dispatch, response publish included."""
+
+    session_factory: async_sessionmaker[AsyncSession]
+
+    async def on_receive(self) -> None:
+        self._begin = self.session_factory.begin()
+        self.context.set_local("session", await self._begin.__aenter__())
+        await super().on_receive()
+
+    async def after_processed(self, exc_type: Any = None, exc_val: Any = None, exc_tb: Any = None) -> bool | None:
+        await self._begin.__aexit__(exc_type, exc_val, exc_tb)
+        return await super().after_processed(exc_type, exc_val, exc_tb)
+
+
+async def test_outbox_response_on_a_middleware_scoped_session_commits_the_followon_row(
+    pg_engine: AsyncEngine,
+    outbox_table: Table,
+) -> None:
+    """A session opened in a broker middleware outlives the response publish, so the follow-on row commits."""
+    _SessionPerMessage.session_factory = async_sessionmaker(pg_engine, expire_on_commit=False)
+    broker = OutboxBroker(pg_engine, outbox_table=outbox_table, middlewares=[_SessionPerMessage])
+    handled: list[dict[str, Any]] = []
+    session_ctx = Context()
+
+    @broker.subscriber("orders", min_fetch_interval=0.05, max_fetch_interval=0.2)
+    async def handle(body: dict[str, Any], session: AsyncSession = session_ctx) -> OutboxResponse:
+        handled.append(body)
+        return OutboxResponse(body=body, queue="downstream", session=session)
+
+    async with broker:
+        async with _SessionPerMessage.session_factory() as session, session.begin():
+            await broker.publish({"order_id": 1}, queue="orders", session=session)
+        await _wait_until(lambda: bool(handled), timeout=10.0)
+        await asyncio.sleep(0.3)
+
+    async with pg_engine.connect() as conn:
+        queues = (await conn.execute(select(outbox_table.c.queue))).scalars().all()
+    assert queues == ["downstream"]
+
+
 async def test_publish_payload_is_decodable_by_subscriber(pg_engine, outbox_table) -> None:
     received: list[dict] = []
     broker = OutboxBroker(pg_engine, outbox_table=outbox_table)
@@ -1907,8 +1996,8 @@ async def test_relay_dual_fire_guard_through_worker_loop_leaves_row_and_logs(
     Every existing relay-chain test runs in ``TestOutboxBroker(run_loops=False)``,
     where the handler runs synchronously inside ``publish`` and the guard raises out
     of ``dispatch_one``. The *worker-loop* contract is different: ``_worker_inner``
-    catches the ``_OutboxConfigError``, logs it at ERROR, and leaves the row for
-    lease-expiry retry (it must NOT delete it or fire the foreign publish). This is
+    catches the ``_OutboxConfigError`` and logs it at ERROR, and the row is retried
+    per its strategy (never deleted, and the foreign publish never fires). This is
     the first test that exercises that catch path end-to-end against real Postgres.
     """
     broker_outbox = OutboxBroker(pg_engine, outbox_table=outbox_table)
@@ -1948,8 +2037,8 @@ def spy_log(*args: Any, **kwargs: Any) -> Any:
         # Asserted inside the TestKafkaBroker context: outside it, `.mock` raises SetupError.
         publisher_kafka.mock.assert_not_called()
 
-    # Row left in place (lease held, not deleted) for lease-expiry retry.
-    assert await _row_count(pg_engine, outbox_table) == 1, "config-error row must be left for lease-expiry, not deleted"
+    # Row kept for retry, not deleted.
+    assert await _row_count(pg_engine, outbox_table) == 1, "config-error row must be retried, not deleted"
     # And the _OutboxConfigError was logged at ERROR by the worker loop.
     assert errors, "worker loop must log the _OutboxConfigError at ERROR"
 
diff --git a/tests/test_relay.py b/tests/test_relay.py
index c4bb0f0..b21c417 100644
--- a/tests/test_relay.py
+++ b/tests/test_relay.py
@@ -209,7 +209,7 @@ async def test_outbox_response_with_foreign_publisher_raises() -> None:
     The two compose into a silent dual-fire: the row lands in the outbox *and* the same body is
     published to the foreign broker, so downstream sees it twice with no error anywhere. The guard
     turns an undetectable delivery bug into a loud ``_OutboxConfigError`` at dispatch, which the
-    worker loop then swallows without nacking so the row simply waits for the wiring to be fixed.
+    worker loop logs at ERROR while the row retries until the wiring is fixed.
     """
     metadata = MetaData()
     outbox_table = make_outbox_table(metadata)
@@ -255,6 +255,32 @@ async def relay(body: dict[str, Any]) -> OutboxResponse:
         assert not any(r.queue == "next_queue" for r in test_outbox.fake_client.rows)
 
 
+async def test_outbox_response_with_foreign_publisher_writes_the_recorded_retry() -> None:
+    """INVARIANT: a configuration error after the handler writes the outcome the ack middleware recorded.
+
+    The error unwinds through the ack middleware, which nacks, so the retry strategy has already
+    decided. Dropping that outcome leaves the row leased until expiry and, with ``max_deliveries=None``,
+    re-runs the handler every ``lease_ttl_seconds`` forever.
+    """
+    metadata = MetaData()
+    broker_outbox = OutboxBroker(outbox_table=make_outbox_table(metadata))
+    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]) -> OutboxResponse:
+        return OutboxResponse(body=body, queue="next_queue", session=mock.AsyncMock(spec=AsyncSession))
+
+    test_outbox = TestOutboxBroker(broker_outbox, run_loops=False)
+    async with TestKafkaBroker(broker_kafka), test_outbox as outbox:
+        with pytest.raises(RuntimeError, match="OutboxResponse"):
+            await outbox.publish({"x": 1}, queue="relay_queue", session=None)  # ty: ignore[invalid-argument-type]
+        rows = [(r.queue, r.attempts_count, r.acquired_token) for r in test_outbox.fake_client.rows]
+
+    assert rows == [("relay_queue", 1, None)]
+
+
 async def test_unstarted_foreign_broker_warns_on_start(caplog: pytest.LogCaptureFixture) -> None:
     """Log one WARNING per unstarted foreign broker at start() time.