Add structured response format validation for LLM adapters

- Introduce LLMResponseFormatError for provider response parsing
  failures with descriptive error messages
- Extract dedicated _extract_openai_content, _extract_claude_content,
  and _extract_gemini_content helpers with explicit field validation
  instead of silent empty-string fallback
- Validate streaming delta format via _first_choice_delta

Author: johnteee

teaagent/__init__.py

@@ -41,6 +41,7 @@
     LLMMessage,
     LLMRequest,
     LLMResponse,
+    LLMResponseFormatError,
     ProviderConfig,
     available_providers,
     check_llm_configuration,
@@ -145,6 +146,7 @@
     'LLMMessage',
     'LLMRequest',
     'LLMResponse',
+    'LLMResponseFormatError',
     'MemoryCatalog',
     'MemoryEntry',
     'ModelConformanceReport',

teaagent/llm.py

@@ -21,6 +21,10 @@ class LLMHTTPError(LLMAdapterError):
     pass
 
 
+class LLMResponseFormatError(LLMAdapterError):
+    pass
+
+
 @dataclass(frozen=True)
 class LLMMessage:
     role: str
@@ -163,7 +167,7 @@ def complete(self, request: LLMRequest) -> LLMResponse:
             payload,
             timeout=self.timeout,
         )
-        content = response.get('choices', [{}])[0].get('message', {}).get('content', '')
+        content = _extract_openai_content(self.provider, response)
         usage = response.get('usage', {})
         return LLMResponse(
             provider=self.provider,
@@ -201,7 +205,7 @@ def _complete_streaming(
                     if usage:
                         input_tokens = usage.get('prompt_tokens', 0)
                         output_tokens = usage.get('completion_tokens', 0)
-                    delta = parsed.get('choices', [{}])[0].get('delta', {})
+                    delta = _first_choice_delta(self.provider, parsed)
                     chunk = delta.get('content', '')
                     if chunk and request.on_chunk:
                         request.on_chunk(chunk)
@@ -266,12 +270,7 @@ def complete(self, request: LLMRequest) -> LLMResponse:
             payload,
             timeout=self.timeout,
         )
-        content_blocks = response.get('content', [])
-        content = ''.join(
-            block.get('text', '')
-            for block in content_blocks
-            if block.get('type') == 'text'
-        )
+        content = _extract_claude_content(response)
         usage = response.get('usage', {})
         return LLMResponse(
             provider=self.provider,
@@ -318,8 +317,7 @@ def complete(self, request: LLMRequest) -> LLMResponse:
             payload,
             timeout=self.timeout,
         )
-        parts = response.get('candidates', [{}])[0].get('content', {}).get('parts', [])
-        content = ''.join(part.get('text', '') for part in parts)
+        content = _extract_gemini_content(response)
         metadata = response.get('usageMetadata', {})
         return LLMResponse(
             provider=self.provider,
@@ -331,6 +329,73 @@ def complete(self, request: LLMRequest) -> LLMResponse:
         )
 
 
+def _extract_openai_content(provider: str, response: dict[str, Any]) -> str:
+    choices = response.get('choices')
+    if not isinstance(choices, list) or not choices:
+        raise LLMResponseFormatError(f'{provider} response missing choices')
+    first_choice = choices[0]
+    if not isinstance(first_choice, dict):
+        raise LLMResponseFormatError(f'{provider} response choice is not an object')
+    message = first_choice.get('message')
+    if not isinstance(message, dict):
+        raise LLMResponseFormatError(f'{provider} response missing message')
+    content = message.get('content')
+    if not isinstance(content, str) or not content:
+        raise LLMResponseFormatError(f'{provider} response missing text content')
+    return content
+
+
+def _first_choice_delta(provider: str, response: dict[str, Any]) -> dict[str, Any]:
+    choices = response.get('choices')
+    if not isinstance(choices, list) or not choices:
+        raise LLMResponseFormatError(f'{provider} stream chunk missing choices')
+    first_choice = choices[0]
+    if not isinstance(first_choice, dict):
+        raise LLMResponseFormatError(f'{provider} stream choice is not an object')
+    delta = first_choice.get('delta', {})
+    if not isinstance(delta, dict):
+        raise LLMResponseFormatError(f'{provider} stream delta is not an object')
+    return delta
+
+
+def _extract_claude_content(response: dict[str, Any]) -> str:
+    content_blocks = response.get('content')
+    if not isinstance(content_blocks, list):
+        raise LLMResponseFormatError('claude response missing content blocks')
+    text_parts = [
+        block.get('text', '')
+        for block in content_blocks
+        if isinstance(block, dict) and block.get('type') == 'text'
+    ]
+    content = ''.join(part for part in text_parts if isinstance(part, str))
+    if not content:
+        raise LLMResponseFormatError('claude response missing text content')
+    return content
+
+
+def _extract_gemini_content(response: dict[str, Any]) -> str:
+    candidates = response.get('candidates')
+    if not isinstance(candidates, list) or not candidates:
+        raise LLMResponseFormatError('gemini response missing candidates')
+    first_candidate = candidates[0]
+    if not isinstance(first_candidate, dict):
+        raise LLMResponseFormatError('gemini candidate is not an object')
+    content = first_candidate.get('content')
+    if not isinstance(content, dict):
+        raise LLMResponseFormatError('gemini response missing content')
+    parts = content.get('parts')
+    if not isinstance(parts, list):
+        raise LLMResponseFormatError('gemini response missing parts')
+    text = ''.join(
+        part.get('text', '')
+        for part in parts
+        if isinstance(part, dict) and isinstance(part.get('text'), str)
+    )
+    if not text:
+        raise LLMResponseFormatError('gemini response missing text content')
+    return text
+
+
 PROVIDER_CONFIGS = {
     'claude': ProviderConfig(
         name='claude',

tests/test_llm.py

@@ -10,6 +10,7 @@
 from teaagent import (
     LLMMessage,
     LLMRequest,
+    LLMResponseFormatError,
     available_providers,
     check_llm_configuration,
     create_llm_adapter,
@@ -222,6 +223,30 @@ def test_openai_streaming_reads_sse_lines_incrementally(self) -> None:
         self.assertEqual(result.input_tokens, 2)
         self.assertEqual(result.output_tokens, 3)
 
+    def test_openai_adapter_rejects_malformed_response(self) -> None:
+        transport = FakeTransport({'error': {'message': 'blocked'}})
+        with patch.dict(os.environ, {'OPENAI_API_KEY': 'key'}, clear=True):
+            adapter = create_llm_adapter('gpt', transport=transport, model='gpt-test')
+
+            with self.assertRaises(LLMResponseFormatError):
+                adapter.complete(LLMRequest(messages=[LLMMessage('user', 'hi')]))
+
+    def test_claude_adapter_rejects_malformed_response(self) -> None:
+        transport = FakeTransport({'content': [{'type': 'tool_use'}]})
+        with patch.dict(os.environ, {'ANTHROPIC_API_KEY': 'key'}, clear=True):
+            adapter = create_llm_adapter('claude', transport=transport)
+
+            with self.assertRaises(LLMResponseFormatError):
+                adapter.complete(LLMRequest(messages=[LLMMessage('user', 'hi')]))
+
+    def test_gemini_adapter_rejects_malformed_response(self) -> None:
+        transport = FakeTransport({'promptFeedback': {'blockReason': 'SAFETY'}})
+        with patch.dict(os.environ, {'GEMINI_API_KEY': 'key'}, clear=True):
+            adapter = create_llm_adapter('gemini', transport=transport)
+
+            with self.assertRaises(LLMResponseFormatError):
+                adapter.complete(LLMRequest(messages=[LLMMessage('user', 'hi')]))
+
 
 if __name__ == '__main__':
     unittest.main()