feat: accept an AsyncConnection wherever publishing takes a session

[lesnik512]

Every publish-side entry point now accepts a SQLAlchemy Core AsyncConnection as well as an AsyncSession: broker.publish, publish_batch, cancel_timer, fetch_unprocessed, OutboxPublisher.publish, OutboxResponse and OutboxPublishCommand. Code that uses Core without the ORM can publish on the connection its domain writes run on:

async with engine.begin() as conn:
    await conn.execute(insert(orders).values(id=1))
    await broker.publish({"order_id": 1}, queue="orders", session=conn)

Design

• The existing session= keyword is widened rather than adding connection=. A second keyword (the faststream-sqlbroker shape) would make both optional, so forgetting both becomes a runtime error instead of a type error unless every entry point gets @overload pairs. It would also need an exactly-one check and a second field on the command and response. Widening keeps the parameter required, breaks no caller, and the docs explain that the name covers both types.
• The parameter stays required. faststream-sqlbroker's optional connection falls back to its own transaction when omitted, which silently drops the outbox guarantee. This PR doesn't copy that.
• NOTIFY dedup keys on the sync transaction object (sync_connection.get_nested_transaction() or get_transaction()), the same way it already does for sessions. The async proxies are regenerated per call and only weakly held, so keying on them would defeat the memo under autobegin. RootTransaction and NestedTransaction can be weakly referenced, so the memo keeps working and a rolled-back savepoint can't suppress a later NOTIFY.
• SessionOrConnection and the shared type guard live in a new _session.py, so the accepted types are defined in one place.

Known edges, not addressed

• Mixing session and await session.connection() in one transaction gives two memo keys, so one extra pg_notify can go out. Postgres coalesces it at delivery, so it's harmless.
• An AsyncConnection that was constructed but never started passes the type check and then fails inside SQLAlchemy, not with our TypeError. engine.connect() and engine.begin() always return a started connection.
• Raw asyncpg connections are out of scope.

Tests

Integration tests on a real connection cover commit, rollback, autobegin (rows land only on commit(), one NOTIFY), savepoint rollback re-emitting NOTIFY, publish_batch sharing the dedup, cancel_timer, fetch_unprocessed, OutboxResponse and OutboxPublisher.publish. A fake-broker test publishes with a connection, and the rejection test now pins the new error message. just test: 754 passed. ruff and ty are clean, and mkdocs build --strict passes.

[github-actions]

Benchmark gate

✅ gate passed

scenario	msg/s	delete/msg	WALrec/msg	WALB/msg	fpi	upd	del	dead_tup
consumer/w1/b10	1300	1.000	6.72	921	242	5000	5000	10000
consumer/w1/b100	1396	1.000	7.11	1154	243	5000	5000	10000
consumer/w2/b10	1570	1.000	6.72	923	242	5000	5000	10000
consumer/w2/b100	1888	1.000	7.11	1154	243	5000	5000	10000
consumer/w4/b10	1641	1.000	6.76	953	242	5000	5000	10000
consumer/w4/b100	1864	1.000	7.11	1153	243	5000	5000	10000
consumer/w1/b100/tfbs100	5664	0.010	6.17	1171	265	5000	5000	10000
consumer/w1/b100/interval1-10	1433	1.000	7.11	1156	244	5000	5000	10000
producer/w1/b100	2934	0.000	3.04	597	1	0	0	0
producer/w1/b100/ppt1	968	0.000	4.04	635	0	0	0	0

Gated (fails the build): delete_calls + tuple counters (upd/del/ins) + the producer's insert_calls, exact; select_calls and other_calls within +2; wal_records within a 10% band; fetch_calls within 2x. msg/s, WAL bytes and total calls are informational (timing/FPI noise).