Fix UNIQUE constraint violation on messages table when resuming conversations

[olsonjeffery]

Summary

Fixes a UNIQUE constraint failed: messages.project_key, messages.session_id, messages.seq error that occurs when resuming a conversation (sending a follow-up message). The conversation becomes inert — model response events fail to persist and the stream stalls.

Root Causes

RC1: Reader task ignores cancellation signal — In subscribe_and_spawn(), the event reader task is spawned without receiving the cancellation handle. When resume_turn() cancels the old subscription and creates a new one, the old reader task keeps running. Both reader tasks then call persist_event() for the same session, producing duplicate seq values.

RC2: next_seq() is not atomic — next_seq() performs a separate SELECT MAX(seq) before INSERT, making concurrent calls from overlapping reader tasks produce the same seq value.

Changes

Phase 1: Pass cancellation handle to reader task

• Added ReaderCancellation using Arc<tokio::sync::Notify> for per-task cancellation independent of the SDK subscription handle
• Added reader_cancellation.notified() branch to the reader task's tokio::select! loop
• Extracted cancel_inflight() helper to deduplicate cancellation logic across resume_turn(), abort_turn(), and shutdown()

Phase 2: Make seq generation atomic

• Removed next_seq() function entirely
• Inlined atomic seq computation as a subquery in the INSERT INTO messages ... VALUES statement

Refinements

• Code simplification and clippy fixes (deduplicated patterns, early-return refactors, etc.)

Testing

• New test_persist_event_concurrent unit test validates 20 concurrent calls produce unique monotonic seq values
• All 432 unit tests + 9 integration test suites pass