refactor: move the outbox insert and its NOTIFY into the client

[lesnik512]

First of two PRs for #236. This one moves the outbox INSERT and its NOTIFY behind the client interface, with no user-visible change. The second PR switches TestOutboxBroker to the real producer.

Why

TestOutboxBroker can only run the production producer if inserts go through something it can swap. The test broker already swaps the client, so the insert moves there, next to fetch and the terminal writes.

The change

• AbstractOutboxClient.insert / insert_batch take the caller's session. insert returns the row id, or None when a live row holds the timer_id. Their docstrings point callers to broker.publish, because calling them directly skips publish middlewares and the published metric.
• OutboxClient takes over the producer's eight prebuilt INSERT statements, the batch SELECT pg_notify and the per-transaction NOTIFY memo. The SQL is moved verbatim.
• OutboxProducer keeps encoding and the published event, and inserts through broker_config.client, read on each publish.
• FakeOutboxClient gets the same operations: the existing Postgres-style validation, ids, and timer_id dedup. A batch with one rejected row stores nothing.

Brokers without an engine

#236 says to always build OutboxClient with an optional engine. I didn't. broker_config.client is None is what the broker and subscriber use for "not connected" (five places), and the test broker also runs without an engine, so "no engine" can't mean "not connected".

Instead the SQL lives in a small _OutboxInserter that needs no engine. OutboxClient uses it, and the producer falls back to its own instance when the broker has no client. A broker without an engine keeps publishing on the caller's session, as it does today.

Tests

• Contract suite: new insert scenarios run against both clients, real Postgres and the fake.

  • the returned id and the stored row;
  • None on a live timer_id conflict, and the same timer_id on another queue;
  • timer_id reuse after delivery;
  • activate_in gating, activate_at stored as given, and a past activate_at eligible at once;
  • batch order, batch gating, and nothing stored when one row is rejected;
  • headers JSONB can't hold. Each adapter declares its own exception, because the fake raises ValueError and Postgres raises a SQLAlchemy StatementError.

  Over-long queue and timer_id aren't in the contract. SQLAlchemy 2.0.0 binds them as $1::VARCHAR(255), and Postgres truncates an explicit cast instead of raising, so the real client doesn't reject them on every supported version. queue is validated before the INSERT anyway; timer_id isn't, which I'll fix in a separate PR.

  They failed before the change, since neither client had insert.

• Integration: test_broker_without_engine_publishes_on_the_callers_session covers a single publish and a batch publish on a real session, with one NOTIFY. It also passes on main; it's there to guard the fallback.

• Unchanged: the existing NOTIFY dedup tests (savepoints, autobegin, batch after single, timer conflicts) pass unchanged.

Verification

• 734 tests pass in docker at 100% coverage.
• ruff and ty are clean.
• just bench-check passes with unchanged counters.

[github-actions]

Benchmark gate

✅ gate passed

scenario	msg/s	delete/msg	WALrec/msg	WALB/msg	fpi	upd	del	dead_tup
consumer/w1/b10	1375	1.000	6.70	914	242	5000	5000	10000
consumer/w1/b100	1457	1.000	7.09	1143	243	5000	5000	10000
consumer/w2/b10	1649	1.000	6.70	915	242	5000	5000	10000
consumer/w2/b100	1964	1.000	7.09	1143	243	5000	5000	10000
consumer/w4/b10	1693	1.000	6.74	945	242	5000	5000	10000
consumer/w4/b100	1982	1.000	7.10	1143	243	5000	5000	10000
consumer/w1/b100/tfbs100	6463	0.010	6.06	1117	243	5000	5000	10000
consumer/w1/b100/interval1-10	1468	1.000	7.15	1186	265	5000	5000	10000
producer/w1/b100	3035	0.000	3.04	597	1	0	0	0
producer/w1/b100/ppt1	990	0.000	4.04	637	1	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).