Summary

Fixes a deadlock in the concurrent source where the main thread (sole queue consumer) blocks on queue.put() when the shared queue is full. Since the main thread is the only consumer, it can never unblock itself — classic self-deadlock.

Root cause: Three code paths in _handle_item cause the main thread to produce into its own queue via ConcurrentMessageRepository.emit_message() → queue.put():

1. PartitionCompleteSentinel → _on_stream_is_done → ensure_at_least_one_state_emitted → queue.put(state)
2. PartitionGenerationCompletedSentinel → same path
3. Partition → on_partition → emit_message(slice_log) → queue.put(log)

Fix: The main thread now uses put(block=False). If the queue is Full, the message is buffered in a deque (_pending) and drained via consume_queue(), which the main thread already calls after processing every queue item. Worker threads are unchanged (blocking put() for backpressure). Thread detection uses threading.get_ident() captured at construction time.

Review & Testing Checklist for Human

• Verify _pending is main-thread-only: _put_on_queue only appends to _pending when get_ident() == _consumer_thread_id, and consume_queue() is only called from the main thread's processing loop. If this invariant holds, no lock is needed. If a worker thread could ever call consume_queue(), this would be a data race.
• Verify consume_queue() is called frequently enough: Previously a no-op (yield from []), it now yields buffered messages. Confirm callers in concurrent_read_processor.py (on_record, on_partition_complete_sentinel, _on_stream_is_done) all do yield from self._message_repository.consume_queue() so buffered messages are drained promptly.
• No new tests for the deadlock scenario: Existing tests pass but none exercise a full-queue + main-thread-emit path. Consider adding a test that fills the queue, calls emit_message from the "main" thread, and verifies the message lands in _pending and is yielded by consume_queue().
• Validate with a real sync: Run a HubSpot sync (or similar high-volume concurrent source) that previously deadlocked, and confirm it completes. This was the original reproducer (job 77457394).

Notes

• threading.get_ident() is captured once in __init__, which is assumed to always run on the main thread. This is true today because ConcurrentSource.__init__ and the declarative source constructor both run on the main thread.
• The deque thread-safety comment references CPython's GIL, but in practice _pending is only accessed from a single thread (main), so it's safe regardless of GIL guarantees.
• consume_queue() was previously a dead no-op. It now has real behavior — any code that calls it will start receiving buffered messages, which is the intended design.

Link to Devin session: https://app.devin.ai/sessions/ad184113df474f0ba37ede09cdac7eaf