feat: Implement turn sequence management and stale context discard

- Added `turn_seq` attribute to `_QueuedItem` for tracking user turns.
- Introduced `discard_stale` method in `ContextInjectionQueue` to remove queued items from older turns, preventing cross-turn carryover.
- Updated `SessionState` to include `user_turn_seq` and `turn_output_seen` for managing user activity and turn state.
- Enhanced `WebSocketHandler` to register user activity, manage turn sequences, and handle silent turns with a watchdog mechanism.
- Modified context injection methods to include `turn_seq` for accurate tracking.
- Added tests for stale context discard functionality and silent turn handling.
- Created a script to generate an animated GIF from system architecture SVG for documentation purposes.

Author: SunflowersLwtech

artifacts/e2e_multiturn/report.json

@@ -0,0 +1,64 @@
+import cairosvg
+import imageio.v2 as imageio
+import os
+import re
+
+# Configuration
+INPUT_SVG = 'assets/system-architecture.svg'
+OUTPUT_GIF = 'assets/system-architecture.gif'
+FRAMES = 12
+DURATION = 0.1  # seconds per frame
+
+def create_gif():
+    print(f"Reading {INPUT_SVG}...")
+    with open(INPUT_SVG, 'r') as f:
+        svg_content = f.read()
+
+    # FIX: Remove missing marker references that cause cairosvg to crash
+    # The SVG references #arrowGreen but doesn't define it in <defs>
+    svg_content = re.sub(r'marker-end="url\(#[^\)]+\)"', '', svg_content)
+
+    images = []
+    
+    # We will animate the 'stroke-dashoffset' to create a flowing effect on dashed lines
+    # The patterns are roughly length 6-8. A 24-unit shift covers multiples of 3, 4, 6, 8 nicely.
+    # Let's use 24 frames shifting by -1 each time, or 12 frames shifting by -2.
+    
+    print("Generating frames...")
+    for i in range(FRAMES):
+        # Calculate offset. We shift backwards to make it look like it's flowing forward usually.
+        offset = i * -2 
+        
+        # Inject stroke-dashoffset into elements that have stroke-dasharray
+        # We use a regex substitution callback to append the offset
+        def add_offset(match):
+            full_match = match.group(0)
+            # If dashoffset already exists, replace it; otherwise append it
+            if 'stroke-dashoffset' in full_match:
+                return re.sub(r'stroke-dashoffset="[^"]*"', f'stroke-dashoffset="{offset}"', full_match)
+            else:
+                return f'{full_match} stroke-dashoffset="{offset}"'
+
+        # Target lines with dasharray
+        frame_svg = re.sub(r'stroke-dasharray="[^"]*"', add_offset, svg_content)
+        
+        try:
+            # Convert to PNG in memory
+            png_data = cairosvg.svg2png(bytestring=frame_svg.encode('utf-8'))
+            
+            # Append to image list
+            images.append(imageio.imread(png_data))
+            print(f"  - Frame {i+1}/{FRAMES} rendered")
+        except Exception as e:
+            print(f"Error rendering frame {i}: {e}")
+            break
+
+    if images:
+        print(f"Saving GIF to {OUTPUT_GIF}...")
+        imageio.mimsave(OUTPUT_GIF, images, duration=DURATION, loop=0)
+        print("Done!")
+    else:
+        print("No frames generated.")
+
+if __name__ == "__main__":
+    create_gif()

scripts/generate_architecture_gif.py

@@ -298,6 +298,7 @@ class _QueuedItem:
     text: str
     priority: int  # lower = more important
     speak: bool
+    turn_seq: int = 0
     enqueued_at: float = dc_field(default_factory=time.monotonic)
 
 
@@ -441,17 +442,54 @@ def enqueue(
         text: str,
         priority: int = 5,
         speak: bool = True,
+        turn_seq: int = 0,
     ) -> None:
         """Queue a context injection.  Always queues; never sends immediately."""
         self._queue[category] = _QueuedItem(
-            category=category, text=text, priority=priority, speak=speak,
+            category=category, text=text, priority=priority, speak=speak, turn_seq=turn_seq,
         )
         logger.info("Queued [%s] (priority=%d, speak=%s, queue_size=%d, state=%s)",
                      category, priority, speak, len(self._queue), self._state.value)
         # Only schedule flush if IDLE and no timer pending
         if self._state == ModelState.IDLE and self._deferred_flush_handle is None:
             self._schedule_deferred_flush()
 
+    def discard_stale(
+        self,
+        *,
+        min_turn_seq: int,
+        categories: set[str] | None = None,
+    ) -> list[str]:
+        """Drop queued items that belong to older user turns.
+
+        This prevents silent control messages or delayed sub-agent results from
+        leaking into a newer user request and causing cross-turn carryover.
+        """
+        dropped: list[str] = []
+        kept: dict[str, _QueuedItem] = {}
+        for category, item in self._queue.items():
+            should_drop = item.turn_seq < min_turn_seq
+            if categories is not None:
+                should_drop = should_drop and category in categories
+            if should_drop:
+                dropped.append(category)
+                continue
+            kept[category] = item
+
+        if not dropped:
+            return []
+
+        self._queue = kept
+        if not self._queue:
+            self._cancel_deferred_flush()
+        logger.info(
+            "Discarded %d stale queued item(s) for turn %d: %s",
+            len(dropped),
+            min_turn_seq,
+            ", ".join(sorted(dropped)),
+        )
+        return dropped
+
     # -- Deferred flush (batching window) ------------------------------------
 
     def _schedule_deferred_flush(self, delay: float = BATCH_WINDOW_SEC):

server.py

@@ -83,6 +83,8 @@ class SessionState:
 
     # -- User activity -------------------------------------------------------
     last_user_activity_at: float = field(default_factory=time.monotonic)
+    user_turn_seq: int = 0
+    turn_output_seen: bool = False
 
     # -- Memory (Phase 4) ----------------------------------------------------
     memory_top3: list[str] = field(default_factory=list)
@@ -99,6 +101,7 @@ class SessionState:
     TRANSCRIPT_FLUSH_TIMEOUT_SEC: float = field(default=1.5, repr=False)
     CAMERA_GRACE_PERIOD_SEC: float = field(default=12.0, repr=False)
     FRAME_TO_GEMINI_INTERVAL: float = field(default=2.0, repr=False)
+    USER_TURN_GAP_SEC: float = field(default=1.0, repr=False)
     SENTENCE_BOUNDARY_RE: re.Pattern = field(
         default_factory=lambda: re.compile(r"[。！？.!?\n]"), repr=False,
     )

session_state.py

@@ -330,6 +330,42 @@ def test_different_categories_both_kept(self, queue, lrq):
         assert "John detected" in text
 
 
+# ---------------------------------------------------------------------------
+# Tests: stale turn discard
+# ---------------------------------------------------------------------------
+
+
+class TestDiscardStale:
+    def test_discards_only_matching_older_turns(self, queue, lrq):
+        queue.set_model_speaking(True)
+        queue.enqueue("turn_boundary", "old boundary", priority=1, speak=False, turn_seq=1)
+        queue.enqueue("vision", "fresh vision", priority=5, speak=True, turn_seq=2)
+        queue.enqueue("camera_toggle", "camera on", priority=4, speak=True, turn_seq=1)
+
+        dropped = queue.discard_stale(
+            min_turn_seq=2,
+            categories={"turn_boundary", "vision"},
+        )
+
+        assert dropped == ["turn_boundary"]
+        assert set(queue._queue) == {"vision", "camera_toggle"}
+        assert len(lrq.sent) == 0
+
+    def test_cancels_deferred_flush_when_queue_becomes_empty(self, queue):
+        queue.set_model_speaking(True)
+        queue.enqueue("turn_boundary", "old boundary", priority=1, speak=False, turn_seq=1)
+        queue._deferred_flush_handle = MagicMock()
+
+        dropped = queue.discard_stale(
+            min_turn_seq=2,
+            categories={"turn_boundary"},
+        )
+
+        assert dropped == ["turn_boundary"]
+        assert queue._queue == {}
+        assert queue._deferred_flush_handle is None
+
+
 # ---------------------------------------------------------------------------
 # Tests: priority sorting in merged output
 # ---------------------------------------------------------------------------

tests/test_context_injection_queue.py

@@ -10,9 +10,11 @@
 
 import pytest
 
+import tools.navigation as navigation_mod
 from tools.navigation import (
     NAVIGATION_FUNCTIONS,
     NAVIGATION_TOOL_DECLARATIONS,
+    _get_client,
     _haversine_distance,
     _maneuver_to_description,
     _strip_html,
@@ -124,6 +126,21 @@ def test_rounding(self):
         assert result == "at 9 o'clock, 48 meters"
 
 
+class TestGoogleMapsClient:
+    def test_get_client_uses_supported_timeout_arguments(self, monkeypatch):
+        monkeypatch.setenv("GOOGLE_MAPS_API_KEY", "test-key")
+        monkeypatch.setattr(navigation_mod, "_client", None)
+
+        with patch("tools.navigation.googlemaps.Client") as mock_client:
+            _get_client()
+
+        kwargs = mock_client.call_args.kwargs
+        assert kwargs["key"] == "test-key"
+        assert kwargs["connect_timeout"] == 5
+        assert kwargs["read_timeout"] == 5
+        assert "timeout" not in kwargs
+
+
 # ---------------------------------------------------------------------------
 # Helper unit tests
 # ---------------------------------------------------------------------------

tests/test_navigation.py

@@ -655,6 +655,18 @@ async def test_activity_start_while_model_speaking_triggers_interrupt(handler, f
     assert session_state.is_interrupted
 
 
+def test_register_user_activity_starts_new_turn_and_discards_stale_context(handler, mock_ctx_queue, session_state):
+    mock_ctx_queue.discard_stale = Mock()
+    session_state.last_user_activity_at = time.monotonic() - 5
+
+    started_new_turn = handler._register_user_activity(explicit_turn_start=True)
+
+    assert started_new_turn is True
+    assert session_state.user_turn_seq == 1
+    mock_ctx_queue.discard_stale.assert_called_once()
+    assert mock_ctx_queue.discard_stale.call_args.kwargs["min_turn_seq"] == 1
+
+
 @pytest.mark.websocket
 @pytest.mark.asyncio
 async def test_interrupt_debounce_rapid_interrupts(handler, fake_ws, session_state, mock_ctx_queue):
@@ -1183,6 +1195,16 @@ async def test_gemini_connection_drop_closes_queue():
     assert fake_queue.closed
 
 
+def test_silent_turn_reconnect_predicate(handler, session_state):
+    session_state.user_turn_seq = 1
+    session_state.turn_output_seen = False
+
+    assert handler._should_reconnect_silent_turn() is True
+
+    session_state.turn_output_seen = True
+    assert handler._should_reconnect_silent_turn() is False
+
+
 @pytest.mark.websocket
 @pytest.mark.error_scenario
 @pytest.mark.asyncio

tests/test_websocket_handler.py

@@ -48,7 +48,6 @@ def _get_client() -> googlemaps.Client:
             raise RuntimeError("GOOGLE_MAPS_API_KEY environment variable not set")
         _client = googlemaps.Client(
             key=api_key,
-            timeout=5,
             connect_timeout=5,
             read_timeout=5,
         )