Add model conformance check CLI command for live provider testing

- Introduce ModelConformanceResult and ModelConformanceReport data
  types for per-provider and aggregate status (passed/failed/skipped)
- Add run_model_conformance() that attempts a simple prompt across
  selected or all configured providers and reports tokens and cost
- Add `model conformance` CLI subcommand with --provider, --model,
  --prompt, and --max-tokens flags

Author: johnteee

teaagent/__init__.py

@@ -46,6 +46,11 @@
     check_llm_configuration,
     create_llm_adapter,
 )
+from teaagent.llm_conformance import (
+    ModelConformanceReport,
+    ModelConformanceResult,
+    run_model_conformance,
+)
 from teaagent.mcp_http import build_mcp_http_server, serve_mcp_http
 from teaagent.mcp_server import handle_mcp_request, serve_mcp_stdio
 from teaagent.memory import MemoryCatalog, MemoryEntry
@@ -142,6 +147,8 @@
     'LLMResponse',
     'MemoryCatalog',
     'MemoryEntry',
+    'ModelConformanceReport',
+    'ModelConformanceResult',
     'ModelRoute',
     'ModelDecisionEngine',
     'PermissionMode',
@@ -218,4 +225,5 @@
     'serve_mcp_stdio',
     'run_chat_agent',
     'run_eval',
+    'run_model_conformance',
 ]

teaagent/cli.py

@@ -20,6 +20,7 @@
     check_llm_configuration,
     create_llm_adapter,
 )
+from teaagent.llm_conformance import run_model_conformance
 from teaagent.mcp_http import DEFAULT_PORT as MCP_HTTP_DEFAULT_PORT
 from teaagent.mcp_http import serve_mcp_http
 from teaagent.mcp_server import serve_mcp_stdio
@@ -386,6 +387,29 @@ def build_parser() -> argparse.ArgumentParser:
     )
     smoke_model.set_defaults(func=model_smoke)
 
+    conformance_model = model_subparsers.add_parser(
+        'conformance', help='Run live conformance checks across providers.'
+    )
+    conformance_model.add_argument(
+        '--provider',
+        action='append',
+        choices=available_providers(),
+        default=None,
+        help='Provider to check. Can be repeated. Defaults to all providers.',
+    )
+    conformance_model.add_argument(
+        '--model',
+        default=None,
+        help='Override model name for all selected providers.',
+    )
+    conformance_model.add_argument(
+        '--prompt', default='Reply with exactly: ok', help='Prompt to send.'
+    )
+    conformance_model.add_argument(
+        '--max-tokens', type=int, default=32, help='Maximum output tokens.'
+    )
+    conformance_model.set_defaults(func=model_conformance)
+
     route = model_subparsers.add_parser(
         'route', help='Classify a task and choose a provider-specific model.'
     )
@@ -822,6 +846,17 @@ def model_smoke(args: argparse.Namespace) -> int:
     return 0
 
 
+def model_conformance(args: argparse.Namespace) -> int:
+    report = run_model_conformance(
+        args.provider,
+        prompt=args.prompt,
+        max_tokens=args.max_tokens,
+        model=args.model,
+    )
+    print_json(report.as_dict())
+    return 0 if report.ok else 1
+
+
 def model_route(args: argparse.Namespace) -> int:
     print_json(
         route_model(args.task, provider=args.provider, model=args.model).to_dict()

teaagent/llm_conformance.py

@@ -0,0 +1,148 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Callable, Iterable
+
+from teaagent.llm import (
+    LLMMessage,
+    LLMRequest,
+    available_providers,
+    check_llm_configuration,
+    create_llm_adapter,
+)
+
+
+@dataclass(frozen=True)
+class ModelConformanceResult:
+    provider: str
+    status: str
+    model: str | None = None
+    content: str = ''
+    error: str | None = None
+    input_tokens: int = 0
+    output_tokens: int = 0
+    estimated_cost_cents: float = 0.0
+
+    def as_dict(self) -> dict[str, object]:
+        payload: dict[str, object] = {
+            'provider': self.provider,
+            'status': self.status,
+        }
+        if self.model is not None:
+            payload['model'] = self.model
+        if self.content:
+            payload['content'] = self.content
+        if self.error:
+            payload['error'] = self.error
+        if self.status == 'passed':
+            payload['input_tokens'] = self.input_tokens
+            payload['output_tokens'] = self.output_tokens
+            payload['estimated_cost_cents'] = self.estimated_cost_cents
+        return payload
+
+
+@dataclass(frozen=True)
+class ModelConformanceReport:
+    results: list[ModelConformanceResult]
+
+    @property
+    def passed(self) -> int:
+        return sum(1 for result in self.results if result.status == 'passed')
+
+    @property
+    def failed(self) -> int:
+        return sum(1 for result in self.results if result.status == 'failed')
+
+    @property
+    def skipped(self) -> int:
+        return sum(1 for result in self.results if result.status == 'skipped')
+
+    @property
+    def ok(self) -> bool:
+        return self.failed == 0
+
+    def as_dict(self) -> dict[str, object]:
+        return {
+            'ok': self.ok,
+            'passed': self.passed,
+            'failed': self.failed,
+            'skipped': self.skipped,
+            'results': [result.as_dict() for result in self.results],
+        }
+
+
+AdapterFactory = Callable[..., object]
+ConfigurationChecker = Callable[[str], tuple[bool, str]]
+
+
+def run_model_conformance(
+    providers: Iterable[str] | None = None,
+    *,
+    prompt: str = 'Reply with exactly: ok',
+    max_tokens: int = 32,
+    model: str | None = None,
+    adapter_factory: AdapterFactory = create_llm_adapter,
+    configuration_checker: ConfigurationChecker = check_llm_configuration,
+) -> ModelConformanceReport:
+    selected_providers = list(providers or available_providers())
+    results = [
+        _run_provider_conformance(
+            provider,
+            prompt=prompt,
+            max_tokens=max_tokens,
+            model=model,
+            adapter_factory=adapter_factory,
+            configuration_checker=configuration_checker,
+        )
+        for provider in selected_providers
+    ]
+    return ModelConformanceReport(results=results)
+
+
+def _run_provider_conformance(
+    provider: str,
+    *,
+    prompt: str,
+    max_tokens: int,
+    model: str | None,
+    adapter_factory: AdapterFactory,
+    configuration_checker: ConfigurationChecker,
+) -> ModelConformanceResult:
+    configured, message = configuration_checker(provider)
+    if not configured:
+        return ModelConformanceResult(
+            provider=provider, status='skipped', error=message
+        )
+
+    try:
+        adapter = adapter_factory(provider, model=model)
+        response = adapter.complete(  # type: ignore[attr-defined]
+            LLMRequest(
+                messages=[LLMMessage(role='user', content=prompt)],
+                max_tokens=max_tokens,
+            )
+        )
+    except Exception as exc:
+        return ModelConformanceResult(
+            provider=provider,
+            status='failed',
+            error=f'{type(exc).__name__}: {exc}',
+        )
+
+    content = response.content.strip()
+    if not content:
+        return ModelConformanceResult(
+            provider=provider,
+            status='failed',
+            model=response.model,
+            error='provider returned empty content',
+        )
+    return ModelConformanceResult(
+        provider=provider,
+        status='passed',
+        model=response.model,
+        content=content,
+        input_tokens=response.input_tokens,
+        output_tokens=response.output_tokens,
+        estimated_cost_cents=response.estimated_cost_cents,
+    )

tests/test_llm_conformance.py

@@ -0,0 +1,73 @@
+from __future__ import annotations
+
+import io
+import json
+import unittest
+from contextlib import redirect_stdout
+from unittest.mock import patch
+
+from conftest import FakeAdapter
+
+from teaagent.cli import main
+from teaagent.llm_conformance import run_model_conformance
+
+
+class FailingAdapter:
+    def complete(self, request: object) -> object:
+        raise RuntimeError('boom')
+
+
+class LLMConformanceTests(unittest.TestCase):
+    def test_run_model_conformance_reports_pass_skip_and_fail(self) -> None:
+        adapters = {'gpt': FakeAdapter(['ok']), 'claude': FailingAdapter()}
+
+        def check(provider: str) -> tuple[bool, str]:
+            if provider == 'gemini':
+                return False, 'GEMINI_API_KEY is not set'
+            return True, 'configured'
+
+        def factory(provider: str, *, model: str | None = None) -> object:
+            return adapters[provider]
+
+        report = run_model_conformance(
+            ['gpt', 'gemini', 'claude'],
+            adapter_factory=factory,
+            configuration_checker=check,
+        )
+
+        self.assertFalse(report.ok)
+        self.assertEqual(report.passed, 1)
+        self.assertEqual(report.skipped, 1)
+        self.assertEqual(report.failed, 1)
+        payload = report.as_dict()
+        self.assertEqual(payload['results'][0]['status'], 'passed')
+        self.assertEqual(payload['results'][1]['status'], 'skipped')
+        self.assertEqual(payload['results'][2]['status'], 'failed')
+
+    def test_cli_model_conformance_outputs_report(self) -> None:
+        output = io.StringIO()
+
+        with (
+            patch('teaagent.cli.run_model_conformance') as run_model_conformance,
+            redirect_stdout(output),
+        ):
+            run_model_conformance.return_value.ok = True
+            run_model_conformance.return_value.as_dict.return_value = {
+                'ok': True,
+                'passed': 1,
+                'failed': 0,
+                'skipped': 0,
+                'results': [{'provider': 'gpt', 'status': 'passed'}],
+            }
+            exit_code = main(['model', 'conformance', '--provider', 'gpt'])
+
+        payload = json.loads(output.getvalue())
+        self.assertEqual(exit_code, 0)
+        self.assertTrue(payload['ok'])
+        run_model_conformance.assert_called_once_with(
+            ['gpt'], prompt='Reply with exactly: ok', max_tokens=32, model=None
+        )
+
+
+if __name__ == '__main__':
+    unittest.main()