EPIC R03 (Team Duy) — Model Providers & Routing. All six tasks done.
R03-T02 — Provider catalogue
domain/models/provider_descriptor.py ProviderDescriptor (frozen), WireProtocol, AuthKind
infrastructure/providers/provider_registry.py
thread-safe registry: id/alias lookup, dynamic
lookup by model id, adapter selection by protocol
providers/factory.py drops its own _REGISTRY table and delegates to the
registry, still raising ProviderError for callers
R03-T03 — RoutingApplicationService (pure Python, 4 modes)
application/model_routing/routing_models.py
RoutingMode (off/auto/manual/fallback),
RoutingRequest (immutable snapshot), RouteEvaluation,
RoutingOutcome
application/model_routing/routing_application_service.py
the single decision flow, reached through two narrow
ports plus a caller-supplied confirm callback, so no
Qt import is needed
application/model_routing/core_routing_adapter.py
binds the ports to core/routing and AppContext
Fallback is a new resilience mode: keep the selected model while it can serve the turn,
re-route only when it cannot. Wired end to end through config.py, state.py,
ui/routing_toggle.py and i18n.py (EN/JA/VI).
R03-T04 / T05 — Remove the duplicated routing flow
ui/chat_panel.py (#L638), ui/co4e_tab.py, ui/folder_tab.py each drop ~35 lines of copied
logic and call the shared service; the widgets now only build a RoutingRequest, host the
Manual-mode modal and render the outcome.
R03-T06 — Token usage as an event
infrastructure/telemetry/usage_sink.py UsageEvent + UsageEventSink protocol, with tracker,
in-memory and composite sinks
providers/openai_compat.py, providers/anthropic.py
publish a UsageEvent instead of writing to the
usage tracker themselves
core/usage_tracker.py adds current_context() so a sink can borrow and
restore a thread's attribution
R03-T01 — Contract tests
tests/contracts/test_providers.py parametrises over every provider in the registry: chat()
signature, canonical assistant message, normalised tool calls, response closed, tool schema
translation, ProviderError, list_models/test_connection, one UsageEvent per turn.
Test infrastructure fix (required to verify any of the above): tests/conftest.py used to put
the repository's PARENT directory on sys.path, so `import cowork_local.*` resolved against
whichever sibling folder happened to carry that name — on a dev machine, an unrelated older
checkout. The suite reported green while exercising different code. The conftest now binds
this checkout to the cowork_local name in sys.modules.
Verification
pytest tests/ 236 passed in ~1.8s (102 before this change)
scripts/check_imports.py PASS, 0 forbidden imports in domain/ and application/
new production files largest is 288 lines, all under the 400 LOC ceiling
new tests 134 (50 contract, 70 unit, 14 integration), all offline
scripts/run_quality_gate.py does not exist yet (R10-T02), so DoD item 7 was covered by
check_imports.py plus the full suite.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
280 lines
12 KiB
Python
280 lines
12 KiB
Python
"""R03-T01 — the contract every LLM provider adapter must satisfy.
|
|
|
|
Parametrised over EVERY provider in the central registry
|
|
(``infrastructure/providers/provider_registry.py``), so registering a new
|
|
provider automatically subjects it to the same specification and a provider that
|
|
drifts from the canonical shapes fails here.
|
|
|
|
The contract, in one list:
|
|
|
|
* construction — the registry builds a real ``Provider`` for every id;
|
|
* ``chat()`` — canonical signature, canonical assistant message, streamed text
|
|
delivered through ``on_text``, tool calls normalised to
|
|
``{"id", "name", "arguments": dict}``, response always closed;
|
|
* tool schema translation matches the adapter's wire protocol;
|
|
* failures raise ``ProviderError`` — never a bare transport exception;
|
|
* ``list_models()`` / ``test_connection()`` report a reason instead of a silent
|
|
empty list;
|
|
* telemetry — exactly one ``UsageEvent`` per turn (R03-T06), with the real
|
|
counts when the stream reports them.
|
|
|
|
Everything runs offline: ``Provider._request`` is replaced by a scripted
|
|
transport, so the suite needs no network, no API key and no Qt event loop.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
import requests
|
|
from cowork_local.infrastructure.providers.provider_registry import (
|
|
BUILTIN_DESCRIPTORS,
|
|
ProviderRegistry,
|
|
)
|
|
from cowork_local.infrastructure.telemetry import usage_sink
|
|
from cowork_local.providers.base import Provider, ProviderError, ToolSpec
|
|
from cowork_local.tests.contracts.provider_stubs import (
|
|
EXPECTED_CACHED_TOKENS,
|
|
EXPECTED_INPUT_TOKENS,
|
|
EXPECTED_OUTPUT_TOKENS,
|
|
EXPECTED_TEXT,
|
|
EXPECTED_TOOL_CALL,
|
|
PROTOCOL_FIXTURES,
|
|
FakeStreamResponse,
|
|
ScriptedTransport,
|
|
)
|
|
|
|
# Every provider id in the catalogue — the parametrisation that makes this a
|
|
# contract suite rather than a per-adapter unit test.
|
|
PROVIDER_IDS = [d.provider_id for d in BUILTIN_DESCRIPTORS]
|
|
|
|
# Minimal config: enough for any adapter to build a URL and headers offline.
|
|
BASE_CONF = {"base_url": "https://gateway.test/v1", "api_key": "test-key"}
|
|
|
|
SAMPLE_MESSAGES = [
|
|
{"role": "system", "content": "You are a helpful assistant."},
|
|
{"role": "user", "content": "Say hello"},
|
|
]
|
|
|
|
SAMPLE_TOOL = ToolSpec(
|
|
name="read_file",
|
|
description="Read a file from disk",
|
|
parameters={"type": "object", "properties": {"path": {"type": "string"}}},
|
|
)
|
|
|
|
|
|
@pytest.fixture()
|
|
def registry() -> ProviderRegistry:
|
|
"""A private registry per test so registrations never leak between tests."""
|
|
return ProviderRegistry(BUILTIN_DESCRIPTORS)
|
|
|
|
|
|
@pytest.fixture()
|
|
def collected_usage(monkeypatch) -> usage_sink.InMemoryUsageSink:
|
|
"""Swap the process-wide telemetry sink for an in-memory one.
|
|
|
|
Restored by monkeypatch after each test, so a contract run never appends to
|
|
the developer's real ``~/.cowork_local/usage/`` files.
|
|
"""
|
|
sink = usage_sink.InMemoryUsageSink()
|
|
monkeypatch.setattr(usage_sink, "_sink", usage_sink.CompositeUsageSink([sink]))
|
|
return sink
|
|
|
|
|
|
def _fixtures_for(registry: ProviderRegistry, provider_id: str) -> dict:
|
|
"""The stream/model-list script matching this provider's wire protocol."""
|
|
protocol = registry.get(provider_id).wire_protocol.value
|
|
return PROTOCOL_FIXTURES[protocol]
|
|
|
|
|
|
def _build(registry: ProviderRegistry, provider_id: str, transport=None) -> Provider:
|
|
"""Build a provider and (optionally) replace its transport with a script."""
|
|
provider = registry.build(provider_id, dict(BASE_CONF))
|
|
if transport is not None:
|
|
# Patch the INSTANCE, not the class: parallel parametrised cases must
|
|
# not see each other's scripted transport.
|
|
provider._request = transport
|
|
return provider
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Construction & interface shape
|
|
# --------------------------------------------------------------------------- #
|
|
@pytest.mark.parametrize("provider_id", PROVIDER_IDS)
|
|
def test_registry_builds_a_provider_for_every_registered_id(registry, provider_id) -> None:
|
|
"""Every catalogued provider must be constructible — a descriptor with no
|
|
working adapter is a broken entry, not a feature flag."""
|
|
provider = _build(registry, provider_id)
|
|
|
|
assert isinstance(provider, Provider)
|
|
# The registry fills in the descriptor's default model when config omits it,
|
|
# so a half-configured provider still names a concrete model.
|
|
assert provider.model, f"{provider_id} built without a model id"
|
|
assert provider.describe() == f"{provider.name}:{provider.model}"
|
|
|
|
|
|
@pytest.mark.parametrize("provider_id", PROVIDER_IDS)
|
|
def test_chat_signature_is_uniform(registry, provider_id) -> None:
|
|
"""All adapters accept the same call, so the agent runtime can swap
|
|
providers without knowing which one it holds."""
|
|
import inspect
|
|
|
|
provider = _build(registry, provider_id)
|
|
params = list(inspect.signature(provider.chat).parameters)
|
|
|
|
assert params == ["messages", "tools", "on_text", "cancel", "on_reasoning"]
|
|
|
|
|
|
@pytest.mark.parametrize("provider_id", PROVIDER_IDS)
|
|
def test_tool_schema_matches_the_wire_protocol(registry, provider_id) -> None:
|
|
"""A ToolSpec must translate into the exact shape the endpoint expects."""
|
|
descriptor = registry.get(provider_id)
|
|
|
|
if descriptor.wire_protocol.value == "anthropic":
|
|
translated = SAMPLE_TOOL.to_anthropic()
|
|
assert translated["input_schema"] == SAMPLE_TOOL.parameters
|
|
assert translated["name"] == "read_file"
|
|
else:
|
|
translated = SAMPLE_TOOL.to_openai()
|
|
assert translated["type"] == "function"
|
|
assert translated["function"]["parameters"] == SAMPLE_TOOL.parameters
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# The turn itself
|
|
# --------------------------------------------------------------------------- #
|
|
@pytest.mark.parametrize("provider_id", PROVIDER_IDS)
|
|
def test_chat_returns_the_canonical_assistant_message(registry, provider_id, collected_usage) -> None:
|
|
"""Whatever the wire format, one turn yields the same canonical result."""
|
|
fixtures = _fixtures_for(registry, provider_id)
|
|
response = FakeStreamResponse(lines=fixtures["stream_lines"]())
|
|
transport = ScriptedTransport([response])
|
|
provider = _build(registry, provider_id, transport)
|
|
|
|
streamed: list = []
|
|
result = provider.chat(
|
|
SAMPLE_MESSAGES, tools=[SAMPLE_TOOL], on_text=streamed.append,
|
|
)
|
|
|
|
assert result["role"] == "assistant"
|
|
assert result["content"] == EXPECTED_TEXT
|
|
# Text must arrive incrementally, not only in the final message — the chat
|
|
# UI streams from these callbacks.
|
|
assert "".join(streamed) == EXPECTED_TEXT
|
|
assert len(streamed) >= 2
|
|
# Tool calls are normalised: parsed arguments, never the raw JSON fragments.
|
|
assert result["tool_calls"] == [EXPECTED_TOOL_CALL]
|
|
assert response.closed, "provider left the streaming response open"
|
|
|
|
|
|
@pytest.mark.parametrize("provider_id", PROVIDER_IDS)
|
|
def test_chat_publishes_exactly_one_usage_event(registry, provider_id, collected_usage) -> None:
|
|
"""R03-T06: a turn reports its token usage through the telemetry sink, with
|
|
the server's real counts when the stream carried them."""
|
|
fixtures = _fixtures_for(registry, provider_id)
|
|
transport = ScriptedTransport([FakeStreamResponse(lines=fixtures["stream_lines"]())])
|
|
provider = _build(registry, provider_id, transport)
|
|
|
|
provider.chat(SAMPLE_MESSAGES, tools=[SAMPLE_TOOL])
|
|
|
|
events = collected_usage.snapshot()
|
|
assert len(events) == 1, "a turn must publish exactly one usage event"
|
|
event = events[0]
|
|
assert event.provider == provider.name
|
|
assert event.model == provider.model
|
|
assert event.input_tokens == EXPECTED_INPUT_TOKENS
|
|
assert event.output_tokens == EXPECTED_OUTPUT_TOKENS
|
|
assert event.cached_tokens == EXPECTED_CACHED_TOKENS
|
|
# Real counts were available, so the event must NOT be flagged as a guess.
|
|
assert event.estimated is False
|
|
|
|
|
|
@pytest.mark.parametrize("provider_id", PROVIDER_IDS)
|
|
def test_usage_is_estimated_when_the_stream_reports_none(registry, provider_id, collected_usage) -> None:
|
|
"""Gateways that never send usage still produce a dashboard row — clearly
|
|
flagged as an estimate rather than silently recorded as zero."""
|
|
# Only text; no usage block anywhere in the stream.
|
|
silent_stream = ['data: ' + '{"choices": [{"delta": {"content": "hi"}}]}', "data: [DONE]"]
|
|
if registry.get(provider_id).wire_protocol.value == "anthropic":
|
|
silent_stream = [
|
|
'data: {"type": "content_block_start", "index": 0, "content_block": {"type": "text"}}',
|
|
'data: {"type": "content_block_delta", "index": 0,'
|
|
' "delta": {"type": "text_delta", "text": "hi"}}',
|
|
]
|
|
transport = ScriptedTransport([FakeStreamResponse(lines=silent_stream)])
|
|
provider = _build(registry, provider_id, transport)
|
|
|
|
provider.chat(SAMPLE_MESSAGES)
|
|
|
|
events = collected_usage.snapshot()
|
|
assert len(events) == 1
|
|
assert events[0].estimated is True
|
|
# An estimate still has to be a positive number to be worth showing.
|
|
assert events[0].total_tokens > 0
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Failure behaviour
|
|
# --------------------------------------------------------------------------- #
|
|
@pytest.mark.parametrize("provider_id", PROVIDER_IDS)
|
|
def test_http_error_becomes_provider_error(registry, provider_id, collected_usage) -> None:
|
|
"""Callers handle exactly one exception type; adapters must not leak
|
|
transport- or JSON-level errors past their boundary."""
|
|
failing = FakeStreamResponse(status_code=401, body='{"error": {"message": "bad key"}}')
|
|
transport = ScriptedTransport([failing])
|
|
provider = _build(registry, provider_id, transport)
|
|
|
|
with pytest.raises(ProviderError):
|
|
provider.chat(SAMPLE_MESSAGES)
|
|
|
|
assert failing.closed, "provider left a failed response open"
|
|
|
|
|
|
@pytest.mark.parametrize("provider_id", PROVIDER_IDS)
|
|
def test_list_models_and_test_connection_report_a_reason(registry, provider_id) -> None:
|
|
"""A failed model load must explain itself: ``last_error`` is what Settings
|
|
shows instead of an unexplained empty dropdown."""
|
|
def _boom(*_args, **_kwargs):
|
|
# A transport failure, i.e. what actually happens when the gateway is
|
|
# unreachable — adapters translate this class of error, not arbitrary
|
|
# programming errors, which must still surface as bugs.
|
|
raise requests.ConnectionError("network down")
|
|
|
|
provider = _build(registry, provider_id, _boom)
|
|
|
|
models = provider.list_models()
|
|
|
|
assert provider.last_error, f"{provider_id} swallowed a model-load failure"
|
|
ok, message = provider.test_connection()
|
|
assert ok is False
|
|
assert message
|
|
# Anthropic answers with a built-in fallback catalogue; a gateway answers
|
|
# with nothing. Both are acceptable — the contract is only that a failure is
|
|
# never reported as success.
|
|
assert isinstance(models, list)
|
|
|
|
|
|
@pytest.mark.parametrize("provider_id", PROVIDER_IDS)
|
|
def test_list_models_returns_ids_on_success(registry, provider_id) -> None:
|
|
"""The happy path returns plain model-id strings, not raw API objects."""
|
|
fixtures = _fixtures_for(registry, provider_id)
|
|
transport = ScriptedTransport([
|
|
FakeStreamResponse(status_code=200, payload=fixtures["models_payload"]),
|
|
])
|
|
provider = _build(registry, provider_id, transport)
|
|
|
|
models = provider.list_models()
|
|
|
|
assert models == fixtures["expected_models"]
|
|
assert provider.last_error == ""
|
|
assert all(isinstance(m, str) for m in models)
|
|
|
|
|
|
@pytest.mark.parametrize("provider_id", PROVIDER_IDS)
|
|
def test_strip_think_removes_inline_reasoning(registry, provider_id) -> None:
|
|
"""Reasoning must never leak into a final answer, whichever adapter ran."""
|
|
provider = _build(registry, provider_id)
|
|
|
|
cleaned = provider.strip_think("<think>secret plan</think>Visible answer")
|
|
|
|
assert cleaned == "Visible answer"
|