fix: improve codex continuation logic and add regression tests

- Refactor droid detection to use regex token matching instead of
  substring search, preventing false positives on Android user agents
- Add early return guard in history divergence detection
- Add test for mid-conversation history divergence replay
- Add test verifying Android user agents aren't misclassified as droid

Author: Mateusz

src/connectors/openai_codex/continuation.py

@@ -4,6 +4,7 @@
 
 import hashlib
 import json
+import re
 import time
 from collections import OrderedDict
 from dataclasses import dataclass
@@ -32,7 +33,6 @@
     "factory-cli",
     "factory_cli",
     "factorydroid",
-    "droid",
 )
 
 
@@ -253,10 +253,17 @@ def _normalize_client_family(cls, candidate: str) -> str:
             "clinelike",
         }:
             return "cline_like"
-        if any(pattern in lowered for pattern in _DROID_USER_AGENT_PATTERNS):
+        if cls._is_droid_candidate(lowered):
             return "droid"
         return "generic"
 
+    @staticmethod
+    def _is_droid_candidate(lowered: str) -> bool:
+        if any(pattern in lowered for pattern in _DROID_USER_AGENT_PATTERNS):
+            return True
+        tokens = {token for token in re.split(r"[^a-z0-9]+", lowered) if token}
+        return "droid" in tokens
+
     @classmethod
     def _fingerprint_component(cls, value: Any) -> str | None:
         if value is None:

tests/unit/connectors/openai_codex/test_continuation.py

@@ -134,6 +134,23 @@ def test_continuation_key_uses_user_agent_header_family_signal() -> None:
     assert coordinator.resolve_previous_response_id(generic_context) is None
 
 
+def test_continuation_key_does_not_treat_android_user_agent_as_droid() -> None:
+    coordinator = InMemoryCodexContinuationCoordinator(ttl_seconds=60, max_entries=4)
+    android_context = _context("session-1")
+    generic_context = _context("session-1")
+    metadata = cast(dict[str, object], android_context.metadata)
+    metadata["headers"] = {
+        "User-Agent": (
+            "Mozilla/5.0 (Linux; Android 14; Pixel 8) AppleWebKit/537.36 "
+            "(KHTML, like Gecko) Chrome/124.0 Mobile Safari/537.36"
+        )
+    }
+
+    coordinator.record_response_id(android_context, "resp-generic")
+
+    assert coordinator.resolve_previous_response_id(generic_context) == "resp-generic"
+
+
 def test_continuation_key_separates_different_client_families_in_same_session() -> None:
     coordinator = InMemoryCodexContinuationCoordinator(ttl_seconds=60, max_entries=4)
     opencode_context = _context("session-1")

tests/unit/connectors/openai_codex/test_executor_streaming.py

@@ -1296,6 +1296,126 @@ async def handle_streaming_side_effect(
         assert changed_tools[0]["name"] == "write_file"
         assert captured_payloads[1]["instructions"] == "Full Codex bootstrap"
 
+    @pytest.mark.asyncio
+    async def test_execute_streaming_replays_when_history_diverges_mid_conversation(
+        self,
+        mock_base_connector,
+        mock_credential_manager,
+        sample_context,
+    ) -> None:
+        continuation = InMemoryCodexContinuationCoordinator()
+        executor = ResponseExecutor(
+            mock_base_connector,
+            mock_credential_manager,
+            continuation_coordinator=continuation,
+        )
+        from src.connectors.openai_codex.contracts import CodexInputItem, CodexPayload
+
+        first_payload = CodexPayload(
+            model="gpt-5.1-codex",
+            input=[
+                CodexInputItem.model_validate(
+                    {
+                        "type": "message",
+                        "role": "user",
+                        "content": [{"type": "input_text", "text": "A"}],
+                    }
+                ),
+                CodexInputItem.model_validate(
+                    {
+                        "type": "message",
+                        "role": "assistant",
+                        "content": [{"type": "output_text", "text": "B"}],
+                    }
+                ),
+                CodexInputItem.model_validate(
+                    {
+                        "type": "message",
+                        "role": "user",
+                        "content": [{"type": "input_text", "text": "C"}],
+                    }
+                ),
+            ],
+            tools=[],
+            tool_choice="auto",
+            parallel_tool_calls=False,
+            store=False,
+            stream=True,
+            include=[],
+            prompt_cache_key="test-key",
+            instructions="Full Codex bootstrap",
+        )
+        diverged_payload = first_payload.model_copy(
+            update={
+                "input": [
+                    first_payload.input[0],
+                    first_payload.input[1],
+                    CodexInputItem.model_validate(
+                        {
+                            "type": "message",
+                            "role": "user",
+                            "content": [{"type": "input_text", "text": "X"}],
+                        }
+                    ),
+                    CodexInputItem.model_validate(
+                        {
+                            "type": "message",
+                            "role": "assistant",
+                            "content": [{"type": "output_text", "text": "D"}],
+                        }
+                    ),
+                    CodexInputItem.model_validate(
+                        {
+                            "type": "message",
+                            "role": "user",
+                            "content": [{"type": "input_text", "text": "E"}],
+                        }
+                    ),
+                ]
+            }
+        )
+
+        async def done_iterator(response_id: str):
+            yield ProcessedResponse(
+                content={"id": response_id, "output": []},
+                metadata={"event_type": "response.done", "done": True},
+            )
+
+        captured_payloads: list[dict[str, object]] = []
+
+        async def handle_streaming_side_effect(
+            url, payload_dict, headers, session_id, *args, **kwargs
+        ):
+            captured_payloads.append(dict(payload_dict))
+            stream_handle = MagicMock()
+            stream_handle.headers = {}
+            stream_handle.cancel_callback = AsyncMock()
+            stream_handle.iterator = done_iterator(
+                "resp_first" if len(captured_payloads) == 1 else "resp_second"
+            )
+            return stream_handle
+
+        mock_base_connector._handle_streaming_response = AsyncMock(
+            side_effect=handle_streaming_side_effect
+        )
+
+        first_result = await executor.execute(first_payload, sample_context)
+        assert first_result.content is not None
+        async for _ in first_result.content:
+            pass
+
+        second_result = await executor.execute(diverged_payload, sample_context)
+        assert second_result.content is not None
+        async for _ in second_result.content:
+            pass
+
+        assert len(captured_payloads) == 2
+        assert "previous_response_id" not in captured_payloads[1]
+        assert captured_payloads[1]["instructions"] == "Full Codex bootstrap"
+        assert captured_payloads[1]["input"] == [
+            item.model_dump(exclude_none=True) for item in diverged_payload.input
+        ]
+
     @pytest.mark.asyncio
     async def test_execute_streaming_records_terminal_response_id_in_continuation(
         self,

tests/unit/connectors/openai_codex/test_executor_websocket.py

@@ -1,13 +1,15 @@
 """Unit tests for ResponseExecutor WebSocket support."""
 
-from __future__ import annotations
-
-from unittest.mock import AsyncMock, MagicMock, patch
-
-import pytest
-from src.connectors.openai_codex.executor import _CodexTransportAdapter
-from src.core.common.exceptions import AuthenticationError
-from src.core.domain.responses import ProcessedResponse, StreamingResponseHandle
+from __future__ import annotations
+
+from typing import Any, cast
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from src.connectors.contracts import ConnectorRequestContext
+from src.connectors.openai_codex.executor import _CodexTransportAdapter
+from src.core.common.exceptions import AuthenticationError
+from src.core.domain.responses import ProcessedResponse, StreamingResponseHandle
 
 
 @pytest.mark.asyncio
@@ -60,11 +62,13 @@ async def mock_send_response_create(*args, **kwargs):
         async for chunk in handle.iterator:
             chunks.append(chunk)
 
-        # Verify chunks
-        assert len(chunks) == 2
-        # Websocket transport adapter yields ProcessedResponse objects directly
-        assert chunks[0].content["message"]["content"] == "Hello"
-        assert chunks[1].content["message"]["content"] == "World"
+        # Verify chunks
+        assert len(chunks) == 2
+        # Websocket transport adapter yields ProcessedResponse objects directly
+        first_content = cast(dict[str, Any], chunks[0].content)
+        second_content = cast(dict[str, Any], chunks[1].content)
+        assert cast(dict[str, Any], first_content["message"])["content"] == "Hello"
+        assert cast(dict[str, Any], second_content["message"])["content"] == "World"
 
     async def test_recreates_websocket_client_when_auth_token_changes(self) -> None:
         """Auth refresh retries must not reuse stale WebSocket credentials."""
@@ -113,17 +117,17 @@ async def test_initiate_websocket_streaming_no_auth(self) -> None:
 
         url = "https://chatgpt.com/backend-api/codex/responses"
         payload = {"model": "gpt-4", "input": []}
-        headers = {}  # No authorization header
+        headers: dict[str, str] = {}  # No authorization header
         session_id = "test_session"
 
         with pytest.raises(AuthenticationError, match="No API key"):
             await adapter.initiate_streaming_request(url, payload, headers, session_id)
 
-    async def test_http_fallback_when_websocket_disabled(self) -> None:
-        """Test fallback to HTTP/SSE when WebSocket is disabled."""
-        mock_connector = MagicMock()
-        mock_connector._handle_streaming_response = AsyncMock(
-            return_value=StreamingResponseHandle(
+    async def test_http_fallback_when_websocket_disabled(self) -> None:
+        """Test fallback to HTTP/SSE when WebSocket is disabled."""
+        mock_connector = MagicMock()
+        mock_connector._handle_streaming_response = AsyncMock(
+            return_value=StreamingResponseHandle(
                 iterator=AsyncMock(), headers={}, cancel_callback=AsyncMock()
             )
         )
@@ -140,15 +144,53 @@ async def test_http_fallback_when_websocket_disabled(self) -> None:
         )
 
         # Verify HTTP/SSE method was called
-        mock_connector._handle_streaming_response.assert_called_once_with(
-            url, payload, headers, session_id, "responses"
-        )
-        assert isinstance(handle, StreamingResponseHandle)
-
-    async def test_cleanup_closes_websocket_client(self) -> None:
-        """Test cleanup properly disconnects WebSocket client."""
-        mock_connector = MagicMock()
-        adapter = _CodexTransportAdapter(mock_connector, use_websocket=True)
+        mock_connector._handle_streaming_response.assert_called_once_with(
+            url, payload, headers, session_id, "responses"
+        )
+        assert isinstance(handle, StreamingResponseHandle)
+
+    async def test_http_fallback_accepts_transport_metadata_kwargs(self) -> None:
+        """Transport adapter should accept the executor's keyword metadata contract."""
+        mock_connector = MagicMock()
+        mock_connector._handle_streaming_response = AsyncMock(
+            return_value=StreamingResponseHandle(
+                iterator=AsyncMock(), headers={}, cancel_callback=AsyncMock()
+            )
+        )
+
+        adapter = _CodexTransportAdapter(mock_connector, use_websocket=False)
+
+        url = "https://chatgpt.com/backend-api/codex/responses"
+        payload = {"model": "gpt-4", "input": []}
+        headers = {"Authorization": "Bearer test_key"}
+        session_id = "test_session"
+        request_context = ConnectorRequestContext(
+            request_id="req-1",
+            session_id="sess-1",
+            client_host="127.0.0.1",
+            extensions={},
+        )
+
+        handle = await adapter.initiate_streaming_request(
+            url,
+            payload,
+            headers,
+            session_id,
+            context=request_context,
+            backend="openai-codex",
+            model="gpt-4",
+            key_name="openai-codex",
+        )
+
+        mock_connector._handle_streaming_response.assert_called_once_with(
+            url, payload, headers, session_id, "responses"
+        )
+        assert isinstance(handle, StreamingResponseHandle)
+
+    async def test_cleanup_closes_websocket_client(self) -> None:
+        """Test cleanup properly disconnects WebSocket client."""
+        mock_connector = MagicMock()
+        adapter = _CodexTransportAdapter(mock_connector, use_websocket=True)
 
         # Create mock WebSocket client
         mock_ws_client = AsyncMock()