Fix codegate version for continue with DeepSeek (#982)

`continue version` didn't work with DeepSeek because for some reason
they enclose the user message in a sort of an envelope.

To make the change a little future-proof, we use the envelope to detect
Continue as client and then in the cli command try to extract from the
envelope, but fall back to the generic plain regex.

Fixes: #981

Author: jhrozek

src/codegate/clients/clients.py

@@ -12,3 +12,4 @@ class ClientType(Enum):
     COPILOT = "copilot"  # Copilot client
     OPEN_INTERPRETER = "open_interpreter"  # Open Interpreter client
     AIDER = "aider"  # Aider client
+    CONTINUE = "continue"  # Continue client

src/codegate/clients/detector.py

@@ -160,6 +160,24 @@ def client_name(self) -> ClientType:
         return ClientType.OPEN_INTERPRETER
 
 
+class ContinueDetector(BaseClientDetector):
+    """
+    Detector for Continue client based on message content
+    """
+
+    def __init__(self):
+        super().__init__()
+        # This is a hack that really only detects Continue with DeepSeek
+        # we should get a header or user agent for this (upstream PR pending)
+        self.content_detector = ContentDetector(
+            "You are an AI programming assistant, utilizing the DeepSeek Coder model"
+        )
+
+    @property
+    def client_name(self) -> ClientType:
+        return ClientType.CONTINUE
+
+
 class CopilotDetector(BaseClientDetector):
     """
     Detector for Copilot client based on user agent
@@ -191,6 +209,7 @@ def __init__(self):
             KoduDetector(),
             OpenInterpreter(),
             CopilotDetector(),
+            ContinueDetector(),
         ]
 
     def __call__(self, func):

src/codegate/pipeline/cli/cli.py

@@ -13,6 +13,8 @@
 )
 from codegate.pipeline.cli.commands import CustomInstructions, Version, Workspace
 
+codegate_regex = re.compile(r"^codegate(?:\s+(.*))?", re.IGNORECASE)
+
 HELP_TEXT = """
 ## CodeGate CLI\n
 **Usage**: `codegate [-h] <command> [args]`\n
@@ -77,6 +79,22 @@ def _get_cli_from_open_interpreter(last_user_message_str: str) -> Optional[re.Ma
     return re.match(r"^codegate\s*(.*?)\s*$", last_user_block, re.IGNORECASE)
 
 
+def _get_cli_from_continue(last_user_message_str: str) -> Optional[re.Match[str]]:
+    """
+    Continue sends a differently formatted message to the CLI if DeepSeek is used
+    """
+    deepseek_match = re.search(
+        r"utilizing the DeepSeek Coder model.*?### Instruction:\s*codegate\s+(.*?)\s*### Response:",
+        last_user_message_str,
+        re.DOTALL | re.IGNORECASE,
+    )
+    if deepseek_match:
+        command = deepseek_match.group(1).strip()
+        return re.match(r"^(.*?)$", command)  # This creates a match object with the command
+
+    return codegate_regex.match(last_user_message_str)
+
+
 class CodegateCli(PipelineStep):
     """Pipeline step that handles codegate cli."""
 
@@ -110,12 +128,14 @@ async def process(
         if last_user_message is not None:
             last_user_message_str, _ = last_user_message
             last_user_message_str = last_user_message_str.strip()
-            codegate_regex = re.compile(r"^codegate(?:\s+(.*))?", re.IGNORECASE)
 
+            # Check client-specific matchers first
             if context.client in [ClientType.CLINE, ClientType.KODU]:
                 match = _get_cli_from_cline(codegate_regex, last_user_message_str)
             elif context.client in [ClientType.OPEN_INTERPRETER]:
                 match = _get_cli_from_open_interpreter(last_user_message_str)
+            elif context.client in [ClientType.CONTINUE]:
+                match = _get_cli_from_continue(last_user_message_str)
             else:
                 # Check if "codegate" is the first word in the message
                 match = codegate_regex.match(last_user_message_str)

tests/clients/test_detector.py

@@ -10,6 +10,7 @@
     BaseClientDetector,
     ClineDetector,
     ContentDetector,
+    ContinueDetector,
     CopilotDetector,
     DetectClient,
     HeaderDetector,
@@ -291,6 +292,79 @@ async def test_missing_user_agent(self, mock_request):
         assert await detector.detect(mock_request) is False
 
 
+class TestContinueDetector:
+    @pytest.mark.asyncio
+    async def test_successful_detection_via_system_message(self, mock_request):
+        detector = ContinueDetector()
+
+        async def get_json():
+            return {
+                "system": "You are an AI programming assistant, utilizing the DeepSeek Coder model"
+            }
+
+        mock_request.json = get_json
+        assert await detector.detect(mock_request) is True
+        assert detector.client_name == ClientType.CONTINUE
+
+    @pytest.mark.asyncio
+    async def test_detection_in_message_content(self, mock_request):
+        detector = ContinueDetector()
+
+        async def get_json():
+            return {
+                "messages": [
+                    {
+                        "content": "You are an AI programming assistant, utilizing the DeepSeek Coder model"  # noqa
+                    }
+                ]
+            }
+
+        mock_request.json = get_json
+        assert await detector.detect(mock_request) is True
+
+    @pytest.mark.asyncio
+    async def test_failed_detection_with_partial_match(self, mock_request):
+        detector = ContinueDetector()
+
+        async def get_json():
+            return {"system": "You are an AI assistant"}
+
+        mock_request.json = get_json
+        assert await detector.detect(mock_request) is False
+
+    @pytest.mark.asyncio
+    async def test_case_insensitive_match_handling(self, mock_request):
+        detector = ContinueDetector()
+
+        async def get_json():
+            return {
+                "system": "you ARE an ai programming assistant, UTILIZING the deepseek coder MODEL"
+            }
+
+        mock_request.json = get_json
+        assert await detector.detect(mock_request) is False  # Should be case-sensitive
+
+    @pytest.mark.asyncio
+    async def test_empty_system_message(self, mock_request):
+        detector = ContinueDetector()
+
+        async def get_json():
+            return {"system": ""}
+
+        mock_request.json = get_json
+        assert await detector.detect(mock_request) is False
+
+    @pytest.mark.asyncio
+    async def test_malformed_system_field(self, mock_request):
+        detector = ContinueDetector()
+
+        async def get_json():
+            return {"system": {"nested": "You are an AI programming assistant"}}
+
+        mock_request.json = get_json
+        assert await detector.detect(mock_request) is False
+
+
 class TestDetectClient:
     @pytest.mark.asyncio
     async def test_successful_client_detection(self, mock_request):