diff --git a/docs/introduction/how-it-works.md b/docs/introduction/how-it-works.md
index bcaceab..4038d01 100644
--- a/docs/introduction/how-it-works.md
+++ b/docs/introduction/how-it-works.md
@@ -34,8 +34,8 @@ transactions are the better fit.*
 ## Producer side
 
 `broker.publish(body, *, queue, session, ...)` inserts an outbox row through
-the caller's `AsyncSession`. It does not flush, commit, or open its own
-transaction; the row must commit with the caller's domain writes:
+the caller's `AsyncSession` (or `AsyncConnection`, for Core code). It does not
+flush, commit, or open its own transaction; the row must commit with the caller's domain writes:
 
 ```python
 async with session_factory() as session, session.begin():
diff --git a/docs/usage/basic.md b/docs/usage/basic.md
index b553647..9fb8beb 100644
--- a/docs/usage/basic.md
+++ b/docs/usage/basic.md
@@ -64,7 +64,9 @@ async with session_factory() as session, session.begin():
     # session.begin() commits both atomically on exit
 ```
 
-Passing anything that is not an `AsyncSession` raises `TypeError`. The whole
+Code that uses SQLAlchemy Core without the ORM can pass its `AsyncConnection`
+instead; see [Publishing on a Core connection](./publisher.md#publishing-on-a-core-connection).
+Passing anything else raises `TypeError`. The whole
 point of the outbox pattern is that the row commits atomically with your
 domain writes; opening a separate session would defeat it.
 
diff --git a/docs/usage/publisher.md b/docs/usage/publisher.md
index 7313c73..e7d6c6b 100644
--- a/docs/usage/publisher.md
+++ b/docs/usage/publisher.md
@@ -8,8 +8,8 @@ There are three ways to write an outbox row:
    per-queue config and AsyncAPI spec coverage.
 
 All three share the same transactional contract: the caller supplies an
-`AsyncSession`, and the row commits with the caller's domain writes. The
-broker does not flush, commit, or open its own transaction.
+`AsyncSession` or an `AsyncConnection`, and the row commits with the caller's
+domain writes. The broker does not flush, commit, or open its own transaction.
 
 For "consume from A → enqueue to B" relay flows, a fourth path is
 available: returning `OutboxResponse(...)` from a handler. See [Chained
@@ -37,7 +37,7 @@ await broker.publish(
     body,
     *,
     queue: str,
-    session: AsyncSession,
+    session: AsyncSession | AsyncConnection,
     headers: dict[str, str] | None = None,
     correlation_id: str | None = None,
     activate_in: timedelta | None = None,
@@ -50,7 +50,26 @@ Returns the inserted row's `id`, or `None` if a `timer_id` conflict made
 the insert a no-op. See [Timers](./timers.md) for `activate_in` /
 `activate_at` / `timer_id`.
 
-Passing anything that is not an `AsyncSession` raises `TypeError`.
+Passing anything other than an `AsyncSession` or `AsyncConnection` raises
+`TypeError`.
+
+### Publishing on a Core connection
+
+If your code uses SQLAlchemy Core without the ORM, pass the `AsyncConnection`
+your domain writes run on. The parameter is still called `session`:
+
+```python
+async with engine.begin() as conn:
+    await conn.execute(insert(orders).values(id=1))  # domain write
+    await broker.publish({"order_id": 1}, queue="orders", session=conn)
+```
+
+Everything else works the same: the row commits or rolls back with `conn`'s
+transaction, a savepoint from `conn.begin_nested()` scopes it like a session
+savepoint, and a connection opened with `engine.connect()` without `begin()`
+starts its transaction on the first publish, so you still have to call
+`await conn.commit()`. `publish_batch`, `cancel_timer`, `fetch_unprocessed`,
+`OutboxPublisher.publish` and `OutboxResponse` accept a connection too.
 
 ## `broker.publish_batch`
 
@@ -73,7 +92,7 @@ Full signature:
 await broker.publish_batch(
     *bodies,
     queue: str,
-    session: AsyncSession,
+    session: AsyncSession | AsyncConnection,
     headers: dict[str, str] | None = None,
     activate_in: timedelta | None = None,
     activate_at: datetime | None = None,
diff --git a/faststream_outbox/_session.py b/faststream_outbox/_session.py
new file mode 100644
index 0000000..1faf694
--- /dev/null
+++ b/faststream_outbox/_session.py
@@ -0,0 +1,12 @@
+"""The caller-owned handle an outbox row is written on; a runtime union, so it also works with ``isinstance``."""
+
+from sqlalchemy.ext.asyncio import AsyncConnection, AsyncSession
+
+
+SessionOrConnection = AsyncSession | AsyncConnection
+
+
+def require_session_or_connection(session: object, context: str) -> None:
+    if not isinstance(session, SessionOrConnection):
+        msg = f"{context} requires an sqlalchemy.ext.asyncio.AsyncSession or AsyncConnection"
+        raise TypeError(msg)
diff --git a/faststream_outbox/broker.py b/faststream_outbox/broker.py
index f460c8f..2a7d748 100644
--- a/faststream_outbox/broker.py
+++ b/faststream_outbox/broker.py
@@ -26,10 +26,10 @@
 from faststream.exceptions import IncorrectState
 from faststream.specification.schema import BrokerSpec
 from faststream.specification.schema.extra import Tag, TagDict
-from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.pool import QueuePool
 from typing_extensions import override
 
+from faststream_outbox._session import SessionOrConnection, require_session_or_connection
 from faststream_outbox.client import AbstractOutboxClient, OutboxClient
 from faststream_outbox.configs import LastExceptionRenderer, OutboxBrokerConfig
 from faststream_outbox.message import OutboxInnerMessage
@@ -413,7 +413,7 @@ async def publish(  # ty: ignore[invalid-method-override]
         body: typing.Any,
         *,
         queue: str,
-        session: AsyncSession,
+        session: SessionOrConnection,
         headers: dict[str, str] | None = None,
         correlation_id: str | None = None,
         activate_in: dt.timedelta | None = None,
@@ -422,6 +422,7 @@ async def publish(  # ty: ignore[invalid-method-override]
     ) -> int | None:
         """Insert one outbox row using *session*'s open transaction.
 
+        *session* is an ``AsyncSession`` or, for Core code, an ``AsyncConnection``.
         Must be called inside a transaction the caller owns (typically inside an
         ``async with session.begin():`` block). ``publish`` does not flush, commit,
         or open its own transaction — that is the whole point of the transactional
@@ -453,7 +454,7 @@ async def publish_batch(  # ty: ignore[invalid-method-override]
         self,
         *bodies: typing.Any,
         queue: str,
-        session: AsyncSession,
+        session: SessionOrConnection,
         headers: dict[str, str] | None = None,
         activate_in: dt.timedelta | None = None,
         activate_at: dt.datetime | None = None,
@@ -467,9 +468,7 @@ async def publish_batch(  # ty: ignore[invalid-method-override]
         """
         # 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)
+        require_session_or_connection(session, "broker.publish_batch")
         if not bodies:
             # Validate queue + activate args even when there's no work so an empty
             # batch rejects the same misconfigurations a non-empty one does (F4-06).
@@ -499,7 +498,7 @@ async def cancel_timer(
         *,
         queue: str,
         timer_id: str,
-        session: AsyncSession,
+        session: SessionOrConnection,
     ) -> bool:
         """Delete a not-yet-leased timer row. Idempotent at the ``(queue, timer_id)`` level.
 
@@ -534,7 +533,7 @@ async def cancel_timer(
     async def fetch_unprocessed(
         self,
         *,
-        session: AsyncSession,
+        session: SessionOrConnection,
         queue: str | None = None,
         limit: int = 1000,
     ) -> list[OutboxInnerMessage]:
diff --git a/faststream_outbox/client.py b/faststream_outbox/client.py
index 636b828..6b87420 100644
--- a/faststream_outbox/client.py
+++ b/faststream_outbox/client.py
@@ -44,6 +44,7 @@
 
 from faststream_outbox import schema_validation
 from faststream_outbox._scheduling import is_future_dated, resolve_next_attempt_client_side
+from faststream_outbox._session import SessionOrConnection, require_session_or_connection
 from faststream_outbox._time import utcnow
 from faststream_outbox.message import OutboxInnerMessage
 from faststream_outbox.schema import (
@@ -90,7 +91,7 @@ def table(self) -> "Table": ...
     @abc.abstractmethod
     async def insert(
         self,
-        session: "AsyncSession",
+        session: SessionOrConnection,
         *,
         queue: str,
         payload: bytes,
@@ -109,7 +110,7 @@ async def insert(
     @abc.abstractmethod
     async def insert_batch(
         self,
-        session: "AsyncSession",
+        session: SessionOrConnection,
         *,
         queue: str,
         rows: "Sequence[tuple[bytes, dict[str, str] | None]]",
@@ -163,13 +164,13 @@ async def mark_pending_with_lease(
     ) -> bool: ...
 
     @abc.abstractmethod
-    async def cancel_timer(self, *, queue: str, timer_id: str, session: "AsyncSession") -> bool: ...
+    async def cancel_timer(self, *, queue: str, timer_id: str, session: SessionOrConnection) -> bool: ...
 
     @abc.abstractmethod
     async def fetch_unprocessed(
         self,
         *,
-        session: "AsyncSession",
+        session: SessionOrConnection,
         queue: str | None = None,
         limit: int = 1000,
     ) -> list[OutboxInnerMessage]: ...
@@ -282,7 +283,7 @@ def __init__(self, table: "Table") -> None:
 
     async def insert(
         self,
-        session: "AsyncSession",
+        session: SessionOrConnection,
         *,
         queue: str,
         payload: bytes,
@@ -316,7 +317,7 @@ async def insert(
 
     async def insert_batch(
         self,
-        session: "AsyncSession",
+        session: SessionOrConnection,
         *,
         queue: str,
         rows: "Sequence[tuple[bytes, dict[str, str] | None]]",
@@ -339,15 +340,15 @@ async def insert_batch(
         if not is_future_dated(activate_in, activate_at, now):
             await self._notify(session, queue)
 
-    def _notified_queues(self, session: "AsyncSession") -> set[str] | None:
+    def _notified_queues(self, session: SessionOrConnection) -> set[str] | None:
         """Return the queues already notified in *session*'s innermost transaction, or None outside one."""
-        sync_session = session.sync_session
-        txn = sync_session.get_nested_transaction() or sync_session.get_transaction()
+        sync_owner = session.sync_session if isinstance(session, AsyncSession) else session.sync_connection
+        txn = None if sync_owner is None else sync_owner.get_nested_transaction() or sync_owner.get_transaction()
         if txn is None:
             return None
         return self._notified.setdefault(txn, set())
 
-    async def _notify(self, session: "AsyncSession", queue: str) -> None:
+    async def _notify(self, session: SessionOrConnection, queue: str) -> None:
         # ``pg_notify(:channel, :payload)`` -- parameterized so channel and payload
         # bind cleanly (raw NOTIFY accepts only literals -- injection-prone). Runs on
         # the caller's session so NOTIFY commits with the row insert; rollback silently
@@ -359,12 +360,13 @@ async def _notify(self, session: "AsyncSession", queue: str) -> None:
         # active transaction (get_nested_transaction() during a savepoint, else the outer
         # transaction) -- each a distinct, weak-referenceable object that GCs when the
         # transaction ends, so a rolled-back savepoint cannot suppress a later real NOTIFY.
-        # Key on the *sync* SessionTransaction, not the async AsyncSessionTransaction proxy:
-        # SQLAlchemy regenerates that proxy on every call and only weak-references it, so
-        # under autobegin (no explicit `session.begin()`) it would GC between publishes and
-        # defeat the memo. The Session holds the sync transaction strongly for its lifetime,
-        # so the key survives across publishes whether the caller uses `session.begin()` or
-        # autobegin. No active transaction -> emit unconditionally.
+        # Key on the *sync* transaction (the Session's SessionTransaction, or the Connection's
+        # RootTransaction / NestedTransaction), not the async proxy around it: SQLAlchemy
+        # regenerates that proxy on every call and only weak-references it, so under autobegin
+        # (no explicit `begin()`) it would GC between publishes and defeat the memo. The sync
+        # Session or Connection holds its transaction strongly while it is open, so the key
+        # survives across publishes whether the caller uses `begin()` or autobegin. No active
+        # transaction -> emit unconditionally.
         emitted = self._notified_queues(session)
         if emitted is not None:
             if queue in emitted:
@@ -410,7 +412,7 @@ def engine(self) -> "AsyncEngine":
 
     async def insert(
         self,
-        session: "AsyncSession",
+        session: SessionOrConnection,
         *,
         queue: str,
         payload: bytes,
@@ -431,7 +433,7 @@ async def insert(
 
     async def insert_batch(
         self,
-        session: "AsyncSession",
+        session: SessionOrConnection,
         *,
         queue: str,
         rows: "Sequence[tuple[bytes, dict[str, str] | None]]",
@@ -618,11 +620,9 @@ async def mark_pending_with_lease(
         result = await conn.execute(stmt, {"delay": max(0.0, delay_seconds)})
         return (result.rowcount or 0) > 0
 
-    async def cancel_timer(self, *, queue: str, timer_id: str, session: "AsyncSession") -> bool:
+    async def cancel_timer(self, *, queue: str, timer_id: str, session: SessionOrConnection) -> bool:
         """Delete a not-yet-leased ``(queue, timer_id)`` row on the caller's session."""
-        if not isinstance(session, AsyncSession):
-            msg = "OutboxClient.cancel_timer requires an sqlalchemy.ext.asyncio.AsyncSession"
-            raise TypeError(msg)
+        require_session_or_connection(session, "OutboxClient.cancel_timer")
         t = self._table
         stmt = delete(t).where(
             t.c.queue == queue,
@@ -635,14 +635,12 @@ async def cancel_timer(self, *, queue: str, timer_id: str, session: "AsyncSessio
     async def fetch_unprocessed(
         self,
         *,
-        session: "AsyncSession",
+        session: SessionOrConnection,
         queue: str | None = None,
         limit: int = 1000,
     ) -> list[OutboxInnerMessage]:
         """Read up to *limit* rows (optionally filtered by *queue*) on the caller's session."""
-        if not isinstance(session, AsyncSession):
-            msg = "OutboxClient.fetch_unprocessed requires an sqlalchemy.ext.asyncio.AsyncSession"
-            raise TypeError(msg)
+        require_session_or_connection(session, "OutboxClient.fetch_unprocessed")
         if limit < 1:
             # F4-04: a non-positive limit otherwise hits SQL (LIMIT -1 → DB error) or
             # silently returns nothing (LIMIT 0); reject it up front, consistently with
diff --git a/faststream_outbox/publisher/producer.py b/faststream_outbox/publisher/producer.py
index 382f02c..6bd31eb 100644
--- a/faststream_outbox/publisher/producer.py
+++ b/faststream_outbox/publisher/producer.py
@@ -4,7 +4,7 @@
 route through this producer via FastStream's ``_basic_publish(cmd, producer=...)``
 middleware stack, so encode → insert → ``published`` semantics live in one place. The
 insert itself (and its NOTIFY) belongs to the broker's client. The producer never opens
-its own session: every command carries the caller's ``AsyncSession`` so the row commits
+its own session: every command carries the caller's ``AsyncSession`` or ``AsyncConnection`` so the row commits
 atomically with the caller's domain writes.
 """
 
diff --git a/faststream_outbox/publisher/usecase.py b/faststream_outbox/publisher/usecase.py
index 9e2932c..4e52ba2 100644
--- a/faststream_outbox/publisher/usecase.py
+++ b/faststream_outbox/publisher/usecase.py
@@ -12,9 +12,9 @@
 import typing
 
 from faststream._internal.endpoint.publisher import PublisherUsecase
-from sqlalchemy.ext.asyncio import AsyncSession
 from typing_extensions import override
 
+from faststream_outbox._session import SessionOrConnection
 from faststream_outbox.publisher.config import OutboxPublisherConfig
 from faststream_outbox.publisher.specification import OutboxPublisherSpecification
 from faststream_outbox.response import _REQUEST_UNSUPPORTED_MSG, OutboxPublishCommand
@@ -68,7 +68,7 @@ async def publish(  # ty: ignore[invalid-method-override]
         self,
         body: typing.Any,  # parity with broker.publish (both build OutboxPublishCommand(body: Any))
         *,
-        session: AsyncSession,
+        session: SessionOrConnection,
         headers: dict[str, str] | None = None,
         correlation_id: str | None = None,
         activate_in: dt.timedelta | None = None,
diff --git a/faststream_outbox/response.py b/faststream_outbox/response.py
index 1af3844..dcf6ed5 100644
--- a/faststream_outbox/response.py
+++ b/faststream_outbox/response.py
@@ -1,8 +1,8 @@
 """Publish-side DTO for the outbox transport.
 
 ``OutboxPublishCommand`` carries the domain fields that travel from the
-publisher / broker into ``OutboxProducer.publish``: the user's ``AsyncSession``
-(required — outbox rows commit with the caller's transaction), the queue name,
+publisher / broker into ``OutboxProducer.publish``: the user's ``AsyncSession`` or
+``AsyncConnection`` (required — outbox rows commit with the caller's transaction), the queue name,
 scheduling args, the timer dedup key. Centralizing the activate-args validation
 in the constructor ensures ``broker.publish``, ``broker.publish_batch`` and
 ``OutboxPublisher.publish`` all reject the same misconfigurations.
@@ -17,8 +17,8 @@
 
 from faststream.response.publish_type import PublishType
 from faststream.response.response import BatchPublishCommand, PublishCommand, Response
-from sqlalchemy.ext.asyncio import AsyncSession
 
+from faststream_outbox._session import SessionOrConnection, require_session_or_connection
 from faststream_outbox.message import CORRELATION_ID_HEADER
 
 
@@ -55,9 +55,7 @@ def _validate_publish_args(
     if activate_at is not None and activate_at.tzinfo is None:
         msg = f"{context} requires activate_at to be timezone-aware"
         raise ValueError(msg)
-    if not isinstance(session, AsyncSession):
-        msg = f"{context} requires an sqlalchemy.ext.asyncio.AsyncSession"
-        raise TypeError(msg)
+    require_session_or_connection(session, context)
     # ``queue`` reaches SQL otherwise unvalidated — empty / non-str / over the
     # ``String(255)`` column would surface as an opaque DB error or silent truncation.
     if not isinstance(queue, str):
@@ -91,7 +89,7 @@ def __init__(
         /,
         *bodies: typing.Any,
         queue: str,
-        session: AsyncSession,
+        session: SessionOrConnection,
         headers: dict[str, str] | None = None,
         correlation_id: str | None = None,
         activate_in: dt.timedelta | None = None,
@@ -195,7 +193,7 @@ def __init__(
         body: typing.Any,
         *,
         queue: str,
-        session: AsyncSession,
+        session: SessionOrConnection,
         headers: dict[str, str] | None = None,
         correlation_id: str | None = None,
         activate_in: dt.timedelta | None = None,
diff --git a/faststream_outbox/testing.py b/faststream_outbox/testing.py
index 339f392..fe4d3c8 100644
--- a/faststream_outbox/testing.py
+++ b/faststream_outbox/testing.py
@@ -25,6 +25,7 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from faststream_outbox._scheduling import resolve_next_attempt_client_side
+from faststream_outbox._session import SessionOrConnection
 from faststream_outbox._time import utcnow
 from faststream_outbox.broker import OutboxBroker
 from faststream_outbox.client import _NO_DLQ_TABLE_MSG, AbstractOutboxClient
@@ -139,7 +140,7 @@ def feed(
 
     async def insert(
         self,
-        session: "AsyncSession",
+        session: SessionOrConnection,
         *,
         queue: str,
         payload: bytes,
@@ -164,7 +165,7 @@ async def insert(
 
     async def insert_batch(
         self,
-        session: "AsyncSession",
+        session: SessionOrConnection,
         *,
         queue: str,
         rows: "Sequence[tuple[bytes, dict[str, str] | None]]",
@@ -341,7 +342,7 @@ async def mark_pending_with_lease(
                 return True
         return False
 
-    async def cancel_timer(self, *, queue: str, timer_id: str, session: "AsyncSession") -> bool:
+    async def cancel_timer(self, *, queue: str, timer_id: str, session: SessionOrConnection) -> bool:
         """Mirror :meth:`OutboxBroker.cancel_timer` — drop a not-yet-leased timer row."""
         del session  # ignored (no real DB), matching the fake publish contract
         for i, row in enumerate(self._rows):
@@ -353,7 +354,7 @@ async def cancel_timer(self, *, queue: str, timer_id: str, session: "AsyncSessio
     async def fetch_unprocessed(
         self,
         *,
-        session: "AsyncSession",
+        session: SessionOrConnection,
         queue: str | None = None,
         limit: int = 1000,
     ) -> list[OutboxInnerMessage]:
diff --git a/tests/test_fake.py b/tests/test_fake.py
index d7abde2..df72ba6 100644
--- a/tests/test_fake.py
+++ b/tests/test_fake.py
@@ -14,7 +14,7 @@
 from faststream.exceptions import NackMessage, StopConsume
 from faststream.middlewares import AckPolicy
 from sqlalchemy import MetaData
-from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.ext.asyncio import AsyncConnection, AsyncSession
 from typing_extensions import get_protocol_members
 
 from faststream_outbox import (
@@ -110,6 +110,20 @@ async def handle(body: dict) -> None:
     assert test_broker.fake_client.rows == []  # row deleted after ack
 
 
+async def test_publish_on_a_core_connection_dispatches_under_the_test_broker() -> None:
+    broker = _make_broker()
+    received: list[dict] = []
+
+    @broker.subscriber("orders")
+    async def handle(body: dict) -> None:
+        received.append(body)
+
+    async with TestOutboxBroker(broker):
+        await broker.publish({"order_id": 1}, queue="orders", session=AsyncMock(spec=AsyncConnection))
+
+    assert received == [{"order_id": 1}]
+
+
 async def test_sync_batch_inserts_whole_batch_before_dispatch_and_published_first() -> None:
     """S5: sync-mode batch inserts the whole batch before any handler runs, and ``published`` precedes delivery."""
     events: list[str] = []
diff --git a/tests/test_integration.py b/tests/test_integration.py
index e69be4e..1240d4e 100644
--- a/tests/test_integration.py
+++ b/tests/test_integration.py
@@ -2706,3 +2706,116 @@ async def test_notify_deduped_in_autobegin_transaction(pg_engine, outbox_table)
                 await broker.publish({"n": 1}, queue="orders", session=session)
             await session.commit()
     assert count[0] == 1  # 5 publishes, one queue, one autobegin transaction -> one pg_notify
+
+
+async def test_publish_on_a_core_connection_commits_with_its_transaction(
+    pg_engine: AsyncEngine, outbox_table: Table
+) -> None:
+    broker = OutboxBroker(pg_engine, outbox_table=outbox_table)
+    async with pg_engine.begin() as conn:
+        rid = await broker.publish({"n": 1}, queue="orders", session=conn)
+        assert await _row_count(pg_engine, outbox_table) == 0
+    assert rid is not None
+    assert await _row_count(pg_engine, outbox_table) == 1
+
+
+async def test_publish_on_a_core_connection_rolls_back_with_its_transaction(
+    pg_engine: AsyncEngine, outbox_table: Table
+) -> None:
+    broker = OutboxBroker(pg_engine, outbox_table=outbox_table)
+    async with pg_engine.connect() as conn:
+        trans = await conn.begin()
+        await broker.publish({"n": 1}, queue="orders", session=conn)
+        await trans.rollback()
+    assert await _row_count(pg_engine, outbox_table) == 0
+
+
+async def test_publish_on_an_autobegun_core_connection_lands_on_commit(
+    pg_engine: AsyncEngine, outbox_table: Table
+) -> None:
+    broker = OutboxBroker(pg_engine, outbox_table=outbox_table)
+    with _count_pg_notify(pg_engine) as count:
+        async with pg_engine.connect() as conn:
+            for _ in range(5):
+                await broker.publish({"n": 1}, queue="orders", session=conn)
+            assert await _row_count(pg_engine, outbox_table) == 0
+            await conn.commit()
+    assert await _row_count(pg_engine, outbox_table) == 5
+    assert count[0] == 1
+
+
+async def test_notify_on_a_core_connection_re_emits_after_savepoint_rollback(
+    pg_engine: AsyncEngine, outbox_table: Table
+) -> None:
+    broker = OutboxBroker(pg_engine, outbox_table=outbox_table)
+
+    class _RollbackError(Exception):
+        pass
+
+    with _count_pg_notify(pg_engine) as count:
+        async with pg_engine.begin() as conn:
+            with contextlib.suppress(_RollbackError):
+                async with conn.begin_nested():
+                    await broker.publish({"n": 1}, queue="orders", session=conn)
+                    raise _RollbackError
+            count[0] = 0
+            await broker.publish({"n": 2}, queue="orders", session=conn)
+            await broker.publish({"n": 3}, queue="orders", session=conn)
+    assert count[0] == 1
+
+
+async def test_publish_batch_on_a_core_connection_shares_the_notify_dedup(
+    pg_engine: AsyncEngine, outbox_table: Table
+) -> None:
+    broker = OutboxBroker(pg_engine, outbox_table=outbox_table)
+    with _count_pg_notify(pg_engine) as count:
+        async with pg_engine.begin() as conn:
+            await broker.publish_batch({"n": 1}, {"n": 2}, queue="orders", session=conn)
+            await broker.publish({"n": 3}, queue="orders", session=conn)
+    assert await _row_count(pg_engine, outbox_table) == 3
+    assert count[0] == 1
+
+
+async def test_cancel_timer_on_a_core_connection_deletes_the_pending_timer(
+    pg_engine: AsyncEngine, outbox_table: Table
+) -> None:
+    broker = OutboxBroker(pg_engine, outbox_table=outbox_table)
+    async with pg_engine.begin() as conn:
+        await broker.publish(
+            {"n": 1}, queue="orders", session=conn, activate_in=dt.timedelta(minutes=5), timer_id="t-1"
+        )
+    async with pg_engine.begin() as conn:
+        assert await broker.cancel_timer(queue="orders", timer_id="t-1", session=conn) is True
+    assert await _row_count(pg_engine, outbox_table) == 0
+
+
+async def test_fetch_unprocessed_on_a_core_connection_sees_its_uncommitted_rows(
+    pg_engine: AsyncEngine, outbox_table: Table
+) -> None:
+    broker = OutboxBroker(pg_engine, outbox_table=outbox_table)
+    async with pg_engine.begin() as conn:
+        await broker.publish("pre-commit", queue="orders", session=conn)
+        rows = await broker.fetch_unprocessed(session=conn, queue="orders")
+    assert [r.queue for r in rows] == ["orders"]
+
+
+async def test_outbox_response_on_a_core_connection_commits_with_its_transaction(
+    pg_engine: AsyncEngine, outbox_table: Table
+) -> None:
+    broker = OutboxBroker(pg_engine, outbox_table=outbox_table)
+    response_publisher = OutboxFakePublisher(broker.config.producer)
+    async with pg_engine.begin() as conn:
+        response = OutboxResponse(body={"chained": True}, queue="downstream", session=conn)
+        await response_publisher._publish(response.as_publish_command(), _extra_middlewares=[])  # noqa: SLF001
+        assert await _row_count(pg_engine, outbox_table) == 0
+    assert await _row_count(pg_engine, outbox_table) == 1
+
+
+async def test_publisher_publish_on_a_core_connection_commits_with_its_transaction(
+    pg_engine: AsyncEngine, outbox_table: Table
+) -> None:
+    broker = OutboxBroker(pg_engine, outbox_table=outbox_table)
+    pub = broker.publisher("orders")
+    async with pg_engine.begin() as conn:
+        assert await pub.publish({"n": 1}, session=conn) is not None
+    assert await _row_count(pg_engine, outbox_table) == 1
diff --git a/tests/test_unit.py b/tests/test_unit.py
index 5f8ff8d..7f0fb25 100644
--- a/tests/test_unit.py
+++ b/tests/test_unit.py
@@ -598,15 +598,15 @@ def test_broker_with_engine_has_client() -> None:
 
 
 async def test_broker_publish_rejects_non_async_session() -> None:
-    """INVARIANT: a non-``AsyncSession`` session argument is refused at the publish boundary.
+    """INVARIANT: a session that is neither ``AsyncSession`` nor ``AsyncConnection`` is refused at the boundary.
 
-    The transactional contract is only enforceable through a real session; anything else silently
-    degrades it. Accepting a duck-typed object (a sync ``Session``, a connection, a mock in
-    production code) would let the insert run outside the caller's transaction and fail as lost
-    messages rather than as a ``TypeError`` at the call site.
+    The transactional contract is only enforceable through a handle that owns the caller's transaction;
+    anything else silently degrades it. Accepting a duck-typed object (a sync ``Session``, a raw driver
+    connection, a mock in production code) would let the insert run outside the caller's transaction and
+    fail as lost messages rather than as a ``TypeError`` at the call site.
     """
     broker = _make_broker()
-    with pytest.raises(TypeError, match="AsyncSession"):
+    with pytest.raises(TypeError, match="AsyncSession or AsyncConnection"):
         await broker.publish(b"x", queue="orders", session=object())  # ty: ignore[invalid-argument-type]