fix: channel bidirectional routing — emit TurnCompleteEvent on all id… (#43)

* fix: channel bidirectional routing — emit TurnCompleteEvent on all idle transitions

Bridge previously only emitted TurnCompleteEvent for MQ-initiated turns
(those with a correlation_id in _active_sends). Server-UI-initiated turns
went idle without emitting TurnCompleteEvent, so the Discord bot's
StreamingMessage never finalized — content accumulated in the buffer and
collided with the next Discord-triggered response.

Now TurnCompleteEvent is emitted unconditionally on every idle transition.
correlation_id is empty for non-MQ turns; SDK client filters by
correlation_id so existing consumers are unaffected.

* fix: remove unused variable flagged by ruff

tests/test_bridge_events.py

@@ -0,0 +1,69 @@
+"""Tests for bridge event publishing — TurnCompleteEvent on idle transitions."""
+
+from unittest.mock import MagicMock, patch
+
+from turnstone.mq.bridge import Bridge
+from turnstone.mq.protocol import StateChangeEvent, TurnCompleteEvent
+
+
+def _make_bridge():
+    """Create a Bridge with a mock broker (no Redis or HTTP needed)."""
+    broker = MagicMock()
+    bridge = Bridge(server_url="http://localhost:8080", broker=broker, node_id="test-node")
+    return bridge
+
+
+class TestIdleTurnComplete:
+    """TurnCompleteEvent should be emitted on every idle transition."""
+
+    def test_idle_emits_turn_complete_with_correlation_id(self):
+        """Bridge-initiated turn: TurnCompleteEvent has the correlation_id."""
+        bridge = _make_bridge()
+        bridge._active_sends["ws-1"] = "cid-abc"
+
+        published = []
+        with patch.object(
+            bridge, "_publish_ws", side_effect=lambda ws, ev: published.append((ws, ev))
+        ):
+            bridge._handle_global_event({"type": "ws_state", "ws_id": "ws-1", "state": "idle"})
+
+        turn_completes = [(ws, ev) for ws, ev in published if isinstance(ev, TurnCompleteEvent)]
+        assert len(turn_completes) == 1
+        ws, ev = turn_completes[0]
+        assert ws == "ws-1"
+        assert ev.correlation_id == "cid-abc"
+        # correlation_id should be removed from _active_sends
+        assert "ws-1" not in bridge._active_sends
+
+    def test_idle_emits_turn_complete_without_correlation_id(self):
+        """Server-UI-initiated turn: TurnCompleteEvent has empty correlation_id."""
+        bridge = _make_bridge()
+        # No entry in _active_sends for this workstream
+
+        published = []
+        with patch.object(
+            bridge, "_publish_ws", side_effect=lambda ws, ev: published.append((ws, ev))
+        ):
+            bridge._handle_global_event({"type": "ws_state", "ws_id": "ws-2", "state": "idle"})
+
+        turn_completes = [(ws, ev) for ws, ev in published if isinstance(ev, TurnCompleteEvent)]
+        assert len(turn_completes) == 1
+        ws, ev = turn_completes[0]
+        assert ws == "ws-2"
+        assert ev.correlation_id == ""
+
+    def test_non_idle_state_does_not_emit_turn_complete(self):
+        """Non-idle state transitions should emit StateChangeEvent but not TurnCompleteEvent."""
+        bridge = _make_bridge()
+
+        published = []
+        with patch.object(
+            bridge, "_publish_ws", side_effect=lambda ws, ev: published.append((ws, ev))
+        ):
+            bridge._handle_global_event({"type": "ws_state", "ws_id": "ws-3", "state": "thinking"})
+
+        state_changes = [ev for _, ev in published if isinstance(ev, StateChangeEvent)]
+        turn_completes = [ev for _, ev in published if isinstance(ev, TurnCompleteEvent)]
+        assert len(state_changes) == 1
+        assert state_changes[0].state == "thinking"
+        assert len(turn_completes) == 0

tests/test_channel_discord.py

@@ -312,6 +312,59 @@ class TestParseFooter:
 # ---------------------------------------------------------------------------
 
 
+class TestWsEventFinalization:
+    """TurnCompleteEvent should finalize streaming messages in the Discord bot."""
+
+    def test_turn_complete_finalizes_streaming(self):
+        """ContentEvent + TurnCompleteEvent(correlation_id='') finalizes the message."""
+        from turnstone.channels.discord.bot import TurnstoneBot
+        from turnstone.mq.protocol import ContentEvent, TurnCompleteEvent
+
+        bot = MagicMock(spec=TurnstoneBot)
+        bot.config = MagicMock()
+        bot.config.max_message_length = 2000
+        bot.config.streaming_edit_interval = 1.5
+        bot.config.auto_approve = False
+        bot.config.auto_approve_tools = []
+        bot._streaming = {}
+
+        # Use the real _on_ws_event method
+        bot._on_ws_event = TurnstoneBot._on_ws_event.__get__(bot, TurnstoneBot)
+
+        thread = AsyncMock()
+
+        # Feed content event
+        content_raw = ContentEvent(ws_id="ws-1", text="Hello world").to_json()
+        _run(bot._on_ws_event("ws-1", thread, content_raw))
+
+        # StreamingMessage should exist
+        assert "ws-1" in bot._streaming
+
+        # Feed turn complete with empty correlation_id (server-UI-initiated)
+        complete_raw = TurnCompleteEvent(ws_id="ws-1", correlation_id="").to_json()
+        _run(bot._on_ws_event("ws-1", thread, complete_raw))
+
+        # StreamingMessage should be removed and finalized
+        assert "ws-1" not in bot._streaming
+
+    def test_turn_complete_no_streaming_is_noop(self):
+        """TurnCompleteEvent without prior content should not error."""
+        from turnstone.channels.discord.bot import TurnstoneBot
+        from turnstone.mq.protocol import TurnCompleteEvent
+
+        bot = MagicMock(spec=TurnstoneBot)
+        bot._streaming = {}
+        bot._on_ws_event = TurnstoneBot._on_ws_event.__get__(bot, TurnstoneBot)
+
+        thread = AsyncMock()
+
+        complete_raw = TurnCompleteEvent(ws_id="ws-1", correlation_id="").to_json()
+        _run(bot._on_ws_event("ws-1", thread, complete_raw))
+
+        # No error, no streaming message
+        assert "ws-1" not in bot._streaming
+
+
 class TestChannelCLI:
     """Tests for the channel CLI entry point."""
 

turnstone/mq/bridge.py

@@ -789,12 +789,13 @@ class Bridge:
                 )
             )
 
-            # Completion detection
+            # Completion detection — emit for all idle transitions so
+            # channel adapters can finalize streaming messages even when
+            # the turn was initiated from the server UI (no correlation_id).
             if state == "idle":
                 with self._lock:
                     cid = self._active_sends.pop(ws_id, None)
-                if cid:
-                    self._publish_ws(ws_id, TurnCompleteEvent(ws_id=ws_id, correlation_id=cid))
+                self._publish_ws(ws_id, TurnCompleteEvent(ws_id=ws_id, correlation_id=cid or ""))
 
         elif etype == "ws_rename":
             self._publish_global(WorkstreamRenameEvent(ws_id=ws_id, name=data.get("name", "")))

turnstone/mq/protocol.py

@@ -264,7 +264,8 @@ class TurnCompleteEvent(OutboundEvent):
     """Emitted when a workstream finishes processing (returns to IDLE).
 
     This is a synthetic event produced by the bridge when it detects
-    the ws_state transition to 'idle' after a send.
+    the ws_state transition to 'idle'.  ``correlation_id`` is set for
+    MQ-initiated turns and empty for turns initiated from the server UI.
     """
 
     type: str = "turn_complete"