diff --git a/README.md b/README.md index 9e85fc0fef..66ad43959f 100644 --- a/README.md +++ b/README.md @@ -213,6 +213,7 @@ For another API endpoint, open **Providers → Custom providers → Add provider | [OpenAI API](https://platform.openai.com/api-keys) | `OPENAI_API_KEY` | `openai_api/gpt-5.6-sol` | | [GitHub Copilot](https://docs.github.com/en/copilot/how-tos/copilot-sdk/auth/authenticate) | Connect GitHub Copilot in the Admin UI | `github_copilot/` | | [xAI (Grok)](https://console.x.ai/team/default/api-keys) | `XAI_API_KEY` | `xai/grok-4.5` | +| [Alibaba Cloud](https://www.alibabacloud.com/help/en/model-studio/get-api-key) | `ALIBABA_CLOUD_API_KEY` | `alibaba_cloud/` | | [QwenCloud Token Plan](https://home.qwencloud.com/api-keys) | `QWENCLOUD_API_KEY` | `qwencloud/qwen3.7-plus` | | [QwenCloud Coding Plan](https://home.qwencloud.com/api-keys) | `QWENCLOUD_CODING_API_KEY` | `qwencloud_coding/qwen3.7-plus` | | [Together AI](https://api.together.ai/settings/api-keys) | `TOGETHER_API_KEY` | `together/zai-org/GLM-5.2` | @@ -288,6 +289,10 @@ For another API endpoint, open **Providers → Custom providers → Add provider - Kimi Code subscription keys use `kimi_code/`. Kimi API credit keys use `kimi/`. Kimi Code plans are for personal interactive coding-agent use under [Kimi's community guidelines](https://www.kimi.com/code/docs/en/kimi-code/community-guidelines.html). +- Alibaba Cloud uses a Model Studio pay-as-you-go key and defaults to Singapore. + For another region or workspace, set `ALIBABA_CLOUD_BASE_URL` to its + [OpenAI-compatible endpoint](https://www.alibabacloud.com/help/en/model-studio/base-url) + ending in `/compatible-mode/v1`. The key and endpoint must use the same region. - QwenCloud Coding Plan keys use `qwencloud_coding/`. QwenCloud Token Plan keys use `qwencloud/`. The keys and endpoints are not interchangeable. Coding Plan is for local, personal, interactive coding-agent use under the diff --git a/e2e/test_admin_apply.py b/e2e/test_admin_apply.py index cd8fb0c388..2550d029a4 100644 --- a/e2e/test_admin_apply.py +++ b/e2e/test_admin_apply.py @@ -190,7 +190,8 @@ def test_apply_network_error_unlocks_form_and_keeps_edits( def test_restart_waits_for_a_new_running_server(page: Page, admin_base_url: str): - pending: list[Route] = [] + status = {"status": "running", "instance_id": "old-server"} + status_url = f"{admin_base_url}/admin/api/status" page.route( "**/admin/api/config/apply", lambda route: route.fulfill( @@ -210,21 +211,25 @@ def test_restart_waits_for_a_new_running_server(page: Page, admin_base_url: str) page.goto(f"{admin_base_url}/admin") expect(page.locator("#messageArea")).to_have_text("") page.wait_for_function("!state.startupRequest && !state.startupTimer") - page.route("**/admin/api/status", lambda route: pending.append(route)) + page.route(status_url, lambda route: route.fulfill(json=status)) open_provider(page, "nvidia_nim") page.locator("#field-NVIDIA_NIM_API_KEY").fill("new-key") with page.expect_request("**/admin/api/status"): page.get_by_role("button", name="Save", exact=True).click() expect(page.locator("#applyButton")).to_have_text("Reconnecting…") - for status in ( + for next_status in ( {"status": "running", "instance_id": "old-server"}, {"status": "stopping", "instance_id": "new-server"}, ): - with page.expect_request("**/admin/api/status"): - pending.pop(0).fulfill(json=status) + with page.expect_response( + lambda response, expected=next_status: ( + response.url == status_url and response.json() == expected + ) + ): + status = next_status expect(page.locator("#view-providers")).to_have_attribute("inert", "") expect(page.locator("#dirtyState")).to_have_text("Changes saved") - pending.pop(0).fulfill(json={"status": "running", "instance_id": "new-server"}) + status = {"status": "running", "instance_id": "new-server"} expect(page.locator('[data-provider="nvidia_nim"]')).to_be_enabled() expect(page.locator("#dirtyState")).to_have_text("No changes") expect(page.locator("#messageArea")).to_have_text("Applied") diff --git a/pyproject.toml b/pyproject.toml index 97226cddf2..9b5d430612 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -39,6 +39,7 @@ dependencies = [ "grpcio-tools>=1.81.1", "nvidia-riva-client>=2.26.0", "pyperclip>=1.11.0", + "sqlalchemy>=2.1.1", ] [project.scripts] diff --git a/smoke/lib/config.py b/smoke/lib/config.py index 3b00005d69..d82a14bdba 100644 --- a/smoke/lib/config.py +++ b/smoke/lib/config.py @@ -75,6 +75,7 @@ "groq": "groq/openai/gpt-oss-20b", "cline_pass": "cline_pass/cline-pass/deepseek-v4-flash", "xai": "xai/grok-4.5", + "alibaba_cloud": "alibaba_cloud/qwen3-coder-plus", "qwencloud": "qwencloud/qwen3.7-plus", "qwencloud_coding": "qwencloud_coding/qwen3.7-plus", "together": "together/zai-org/GLM-5.2", diff --git a/smoke/product/test_client_product_live.py b/smoke/product/test_client_product_live.py index 7b730923f7..0340567324 100644 --- a/smoke/product/test_client_product_live.py +++ b/smoke/product/test_client_product_live.py @@ -47,6 +47,21 @@ def _json_object_lines(text: str) -> list[JsonObject]: return objects +def _trace_log_events(text: str) -> list[JsonObject]: + events: list[JsonObject] = [] + for row in _json_object_lines(text): + record = row.get("record") + if not isinstance(record, dict): + continue + extra = record.get("extra") + if not isinstance(extra, dict): + continue + payload = extra.get("trace_payload") + if isinstance(payload, dict): + events.append(payload) + return events + + @pytest.mark.smoke_target("clients") def test_vscode_protocol_e2e(smoke_config: SmokeConfig) -> None: provider_model = ProviderMatrixDriver(smoke_config).first_model() @@ -985,7 +1000,7 @@ def test_claude_cli_web_search_e2e(smoke_config: SmokeConfig, tmp_path: Path) -> assert '"type": "server_tool_use"' in automatic_payload assert '"type": "web_search_tool_result"' in automatic_payload assert "github.com" in automatic_payload - log_rows = _json_object_lines(server_log) + log_rows = _trace_log_events(server_log) assert ( sum( row.get("event") == "free_claude_code.api.web_search.automatic_recognized" @@ -1087,7 +1102,7 @@ def test_claude_auto_mode_openai_connected_e2e( ): assert unexpected not in combined_lower - log_rows = _json_object_lines(server_log) + log_rows = _trace_log_events(server_log) policy_rows = [ row for row in log_rows diff --git a/src/free_claude_code/api/admin_static/providers/alibabacloud-color.svg b/src/free_claude_code/api/admin_static/providers/alibabacloud-color.svg new file mode 100644 index 0000000000..56c20781ec --- /dev/null +++ b/src/free_claude_code/api/admin_static/providers/alibabacloud-color.svg @@ -0,0 +1 @@ +AlibabaCloud \ No newline at end of file diff --git a/src/free_claude_code/cli/commands.py b/src/free_claude_code/cli/commands.py index fbc0aa70cc..5421a461cd 100644 --- a/src/free_claude_code/cli/commands.py +++ b/src/free_claude_code/cli/commands.py @@ -259,7 +259,7 @@ def _run_bound( from free_claude_code.runtime.bootstrap import build_asgi_app - from .uvicorn_server import RuntimeServer + from .uvicorn_server import RuntimeServer, uvicorn_log_config asgi_app = build_asgi_app( settings, @@ -270,9 +270,7 @@ def _run_bound( host=settings.host, port=settings.port, log_level="debug", - log_config=( - uvicorn.config.LOGGING_CONFIG if self._console_logging else None - ), + log_config=uvicorn_log_config(console=self._console_logging), timeout_graceful_shutdown=SERVER_GRACEFUL_SHUTDOWN_SECONDS, ) diff --git a/src/free_claude_code/cli/uvicorn_server.py b/src/free_claude_code/cli/uvicorn_server.py index a0fec01084..d4366ccc8a 100644 --- a/src/free_claude_code/cli/uvicorn_server.py +++ b/src/free_claude_code/cli/uvicorn_server.py @@ -2,9 +2,25 @@ import socket from collections.abc import Awaitable, Callable +from copy import deepcopy import uvicorn +from free_claude_code.config.logging_config import InterceptHandler + + +def uvicorn_log_config(*, console: bool) -> dict[str, object]: + """Send Uvicorn records to FCC's file and optional standard console handlers.""" + config = deepcopy(uvicorn.config.LOGGING_CONFIG) + if not console: + config["formatters"] = {} + config["handlers"] = {} + config["handlers"]["fcc"] = {"()": InterceptHandler} + for name in ("uvicorn", "uvicorn.access"): + handlers = config["loggers"][name]["handlers"] + config["loggers"][name]["handlers"] = [*handlers, "fcc"] if console else ["fcc"] + return config + class RuntimeServer(uvicorn.Server): """Notify the runtime before Uvicorn waits for long-lived HTTP responses.""" diff --git a/src/free_claude_code/config/admin/provider_manifest.py b/src/free_claude_code/config/admin/provider_manifest.py index febca10448..d205f836dc 100644 --- a/src/free_claude_code/config/admin/provider_manifest.py +++ b/src/free_claude_code/config/admin/provider_manifest.py @@ -165,6 +165,19 @@ class ProviderFieldOverride(TypedDict, total=False): "models." ), }, + "ALIBABA_CLOUD_API_KEY": { + "description": ( + "Alibaba Cloud Model Studio pay-as-you-go API key. Use a key from " + "the same region as the base URL." + ), + }, + "ALIBABA_CLOUD_BASE_URL": { + "description": ( + "Optional Model Studio OpenAI-compatible base URL ending in " + "/compatible-mode/v1. Defaults to Singapore. Set your regional " + "or workspace URL when using a key from another region." + ), + }, "QWENCLOUD_API_KEY": { "label": "QwenCloud Token Plan API Key", "description": ( diff --git a/src/free_claude_code/config/logging_config.py b/src/free_claude_code/config/logging_config.py index d07880e6bc..6391fd38b4 100644 --- a/src/free_claude_code/config/logging_config.py +++ b/src/free_claude_code/config/logging_config.py @@ -1,14 +1,12 @@ """Loguru-based structured logging configuration. Structured logs are written as JSON lines to a configurable path (default -``logs/server.log``). Stdlib logging is intercepted and funneled to loguru. -Context vars (request_id, node_id, chat_id) from contextualize() are -included at top level for easy grep/filter. +``~/.fcc/logs/server.log``). Stdlib logging is intercepted and funneled to loguru. +Loguru's JSON stores exception metadata in ``record.exception`` and context in +``record.extra``. Structured traces live in ``record.extra.trace_payload``. """ -import json import logging -import re import threading from pathlib import Path @@ -30,63 +28,6 @@ "telegram.ext", ) -# Loguru ``logger.bind()`` key used by structured TRACE payloads; ``core/trace.py`` -# uses the identical string constant ``TRACE_PAYLOAD_BINDING``. -_TRACE_PAYLOAD_BINDING = "trace_payload" - -# Context keys we promote to top-level JSON for traceability / grep -_CONTEXT_KEYS = ( - "request_id", - "node_id", - "chat_id", - "claude_session_id", - "http_method", - "http_path", -) - -_TELEGRAM_BOT_RE = re.compile( - r"(https?://api\.telegram\.org/)bot([0-9]+:[A-Za-z0-9_-]+)(/?)", - re.IGNORECASE, -) -# Authorization: Bearer (HTTP client / proxy debug lines) -_AUTH_BEARER_RE = re.compile( - r"(\bAuthorization\s*:\s*Bearer\s+)([^\s'\"]+)", - re.IGNORECASE, -) - - -def _redact_sensitive_substrings(message: str) -> str: - """Remove obvious API tokens and secrets before JSON log line emission.""" - text = _TELEGRAM_BOT_RE.sub(r"\1bot\3", message) - return _AUTH_BEARER_RE.sub(r"\1", text) - - -def _serialize_with_context(record) -> str: - """Format record as JSON with context vars at top level. - Returns a format template; we inject _json into record for output. - """ - extra = record.get("extra", {}) - out = { - "time": str(record["time"]), - "level": record["level"].name, - "message": _redact_sensitive_substrings(str(record["message"])), - "module": record["name"], - "function": record["function"], - "line": record["line"], - } - trace_payload = extra.get(_TRACE_PAYLOAD_BINDING) - for key in _CONTEXT_KEYS: - if key in extra and extra[key] is not None: - out[key] = extra[key] - if isinstance(trace_payload, dict): - for tk, tv in trace_payload.items(): - if tk in out: - continue - out[tk] = tv - out["trace"] = True - record["_json"] = json.dumps(out, default=str) - return "{_json}\n" - class InterceptHandler(logging.Handler): """Redirect stdlib logging to loguru.""" @@ -129,7 +70,7 @@ def _add_file_sink(log_file: str | Path, level: str) -> int: return logger.add( log_path, level=level, - format=_serialize_with_context, + serialize=True, encoding="utf-8", mode="a", rotation="50 MB", @@ -150,7 +91,7 @@ def configure_logging( Idempotent: skips if already configured with the same path, level, and verbosity. On path or level change, replaces only the file sink without truncating. On verbosity change alone, updates only the third-party logger levels. - Use force=True to reconfigure from scratch. + Use force=True to rebuild handlers while preserving existing log records. When ``verbose_third_party`` is false, managed noisy third-party loggers are capped at WARNING unless explicitly configured otherwise. @@ -174,8 +115,6 @@ def configure_logging( logger.remove() - log_path.write_text("") - _sink_id = _add_file_sink(log_path, level) intercept = InterceptHandler() diff --git a/src/free_claude_code/config/provider_catalog.py b/src/free_claude_code/config/provider_catalog.py index dd93e6fb81..a3aff95ee1 100644 --- a/src/free_claude_code/config/provider_catalog.py +++ b/src/free_claude_code/config/provider_catalog.py @@ -54,6 +54,7 @@ OPENAI_CODEX_DEFAULT_BASE = "https://chatgpt.com/backend-api/codex" # xAI OpenAI-compatible Chat Completions API. XAI_DEFAULT_BASE = "https://api.x.ai/v1" +ALIBABA_CLOUD_DEFAULT_BASE = "https://dashscope-intl.aliyuncs.com/compatible-mode/v1" # QwenCloud Token Plan OpenAI-compatible Chat Completions API. QWENCLOUD_DEFAULT_BASE = ( "https://token-plan.ap-southeast-1.maas.aliyuncs.com/compatible-mode/v1" @@ -219,6 +220,18 @@ def configuration_attrs(self) -> tuple[str, ...]: default_base_url=XAI_DEFAULT_BASE, proxy_attr="xai_proxy", ), + "alibaba_cloud": ProviderDescriptor( + provider_id="alibaba_cloud", + display_name="Alibaba Cloud", + website_url="https://www.alibabacloud.com/en/product/modelstudio", + logo_filename="alibabacloud-color.svg", + credential_env="ALIBABA_CLOUD_API_KEY", + credential_url="https://www.alibabacloud.com/help/en/model-studio/get-api-key", + credential_attr="alibaba_cloud_api_key", + default_base_url=ALIBABA_CLOUD_DEFAULT_BASE, + base_url_attr="alibaba_cloud_base_url", + proxy_attr="alibaba_cloud_proxy", + ), "qwencloud": ProviderDescriptor( provider_id="qwencloud", display_name="QwenCloud Token Plan", diff --git a/src/free_claude_code/config/settings.py b/src/free_claude_code/config/settings.py index 00a00a5c30..da4f93aa33 100644 --- a/src/free_claude_code/config/settings.py +++ b/src/free_claude_code/config/settings.py @@ -290,6 +290,14 @@ def validate_provider_references(self) -> Settings: default=None, validation_alias="XAI_API_KEY" ) + # ==================== Alibaba Cloud Model Studio ==================== + alibaba_cloud_api_key: OptionalNonEmptyString = Field( + default=None, validation_alias="ALIBABA_CLOUD_API_KEY" + ) + alibaba_cloud_base_url: OptionalNonEmptyString = Field( + default=None, validation_alias="ALIBABA_CLOUD_BASE_URL" + ) + # ==================== QwenCloud Token Plan (OpenAI-compatible) ==================== qwencloud_api_key: OptionalNonEmptyString = Field( default=None, validation_alias="QWENCLOUD_API_KEY" @@ -433,6 +441,9 @@ def validate_provider_references(self) -> Settings: xai_proxy: OptionalNonEmptyString = Field( default=None, validation_alias="XAI_PROXY" ) + alibaba_cloud_proxy: OptionalNonEmptyString = Field( + default=None, validation_alias="ALIBABA_CLOUD_PROXY" + ) qwencloud_proxy: OptionalNonEmptyString = Field( default=None, validation_alias="QWENCLOUD_PROXY" ) diff --git a/src/free_claude_code/core/trace.py b/src/free_claude_code/core/trace.py index 88606b47dc..4d3bafeff7 100644 --- a/src/free_claude_code/core/trace.py +++ b/src/free_claude_code/core/trace.py @@ -1,6 +1,6 @@ """Structured DEBUG traces for end-to-end request / CLI / provider logging. -Emitted lines are merged into JSON log rows by ``config.logging_config``. +Trace payloads are preserved in ``record.extra.trace_payload`` by Loguru. Conversation and Claude Code prompts are logged verbatim unless values live under sanitized credential keys (e.g. ``api_key``, ``authorization``). The default INFO log level excludes these detailed request traces. @@ -50,7 +50,7 @@ def sanitize_trace_value(obj: Any) -> Any: def trace_event(*, stage: str, event: str, source: str, **fields: Any) -> None: - """Emit one structured DEBUG trace row merged into JSON by the log sink.""" + """Emit a DEBUG record with a structured payload bound to its extra fields.""" payload = sanitize_trace_value( { "stage": stage, diff --git a/src/free_claude_code/providers/alibaba_cloud/__init__.py b/src/free_claude_code/providers/alibaba_cloud/__init__.py new file mode 100644 index 0000000000..b89ebfd408 --- /dev/null +++ b/src/free_claude_code/providers/alibaba_cloud/__init__.py @@ -0,0 +1,5 @@ +"""Alibaba Cloud provider exports.""" + +from .client import AlibabaCloudProvider + +__all__ = ["AlibabaCloudProvider"] diff --git a/src/free_claude_code/providers/alibaba_cloud/client.py b/src/free_claude_code/providers/alibaba_cloud/client.py new file mode 100644 index 0000000000..b0934fadda --- /dev/null +++ b/src/free_claude_code/providers/alibaba_cloud/client.py @@ -0,0 +1,139 @@ +"""Model Studio chat transport with native DashScope model discovery.""" + +from typing import Any +from urllib.parse import urljoin + +from openai import AsyncOpenAI + +from free_claude_code.core.anthropic import ReasoningReplayMode +from free_claude_code.providers.admission import ( + ProviderAdmissionController, + ProviderOperationKind, +) +from free_claude_code.providers.base import ProviderConfig +from free_claude_code.providers.model_listing import ModelListResponseError +from free_claude_code.providers.openai_chat import ( + NO_REASONING, + OpenAIChatProfile, + OpenAIChatProvider, + OpenAIChatRequestPolicy, + OpenAIModelListing, +) + +_CHAT_CAPABILITIES = frozenset({"TG", "Reasoning", "VU", "Multimodal-Omni"}) + +_PROFILE = OpenAIChatProfile( + OpenAIChatRequestPolicy( + provider_name="ALIBABA_CLOUD", + reasoning_replay=ReasoningReplayMode.REASONING_CONTENT, + ), + # Model Studio hosts several model families with different thinking controls. + NO_REASONING, + model_listing=OpenAIModelListing( + id_field="model", + required_sequence_items=(("features", "function-calling"),), + thinking_sequence_path=("capabilities",), + thinking_tag="Reasoning", + input_modalities_path=("input_modalities",), + context_window_tokens_path=("model_info", "context_window"), + max_output_tokens_path=("model_info", "max_output_tokens"), + ), +) + + +class AlibabaCloudProvider(OpenAIChatProvider): + """Reuse Chat Completions while adapting the paginated model catalog.""" + + def __init__( + self, + config: ProviderConfig, + *, + admission: ProviderAdmissionController, + client: AsyncOpenAI | None = None, + ) -> None: + super().__init__(config, profile=_PROFILE, admission=admission, client=client) + + async def _list_models_payload(self) -> Any: + # An absolute URL avoids the OpenAI SDK appending to /compatible-mode/v1. + url = urljoin(self._base_url, "/api/v1/models") + models: list[dict[str, Any]] = [] + seen_ids: set[str] = set() + expected_total: int | None = None + for page in range(1, 101): + params = { + "page_no": page, + "page_size": 100, + "features": "function-calling", + "supports": "inference", + } + payload = await self._admission.start_execution().run_call( + lambda params=params: self._client.get( + url, cast_to=object, options={"params": params} + ), + operation_kind=ProviderOperationKind.MODEL_DISCOVERY, + provider_failure_override=self._behavior.failure_override, + ) + if not isinstance(payload, dict) or payload.get("success") is not True: + raise ModelListResponseError("ALIBABA_CLOUD model discovery failed") + output = payload.get("output") + if not isinstance(output, dict): + raise ModelListResponseError( + "ALIBABA_CLOUD model-list output is missing" + ) + total = output.get("total") + items = output.get("models") + if ( + type(total) is not int + or total < 0 + or type(output.get("page_no")) is not int + or output["page_no"] != page + or (expected_total is not None and total != expected_total) + or not isinstance(items, list) + or not items + or any(not isinstance(item, dict) for item in items) + or len(seen_ids) + len(items) > total + ): + raise ModelListResponseError( + "ALIBABA_CLOUD model-list page is malformed" + ) + expected_total = total + for item in items: + model_id = item.get("model") + if not isinstance(model_id, str) or not model_id.strip(): + raise ModelListResponseError( + "ALIBABA_CLOUD model-list item is missing its model id" + ) + if model_id in seen_ids: + raise ModelListResponseError( + "ALIBABA_CLOUD model-list contains duplicate model ids" + ) + seen_ids.add(model_id) + # Count all catalog entries before selecting our HTTP chat subset. + capabilities = item.get("capabilities") + if ( + not isinstance(capabilities, list) + or any(not isinstance(value, str) for value in capabilities) + or not _CHAT_CAPABILITIES.intersection(capabilities) + or any(value.startswith("Realtime-") for value in capabilities) + ): + continue + metadata = item.get("inference_metadata") + modalities = ( + metadata.get("request_modality") + if isinstance(metadata, dict) + else None + ) + models.append( + { + **item, + "input_modalities": [value.lower() for value in modalities] + if isinstance(modalities, list) + and all(isinstance(value, str) for value in modalities) + else None, + } + ) + if len(seen_ids) == total: + return {"data": models} + raise ModelListResponseError( + "ALIBABA_CLOUD model-list pagination exceeded 100 pages" + ) diff --git a/src/free_claude_code/providers/runtime/factory.py b/src/free_claude_code/providers/runtime/factory.py index 9533451201..719697421b 100644 --- a/src/free_claude_code/providers/runtime/factory.py +++ b/src/free_claude_code/providers/runtime/factory.py @@ -107,6 +107,19 @@ def construct( return construct +def _load_alibaba_cloud() -> ProviderFactory: + from free_claude_code.providers.alibaba_cloud import AlibabaCloudProvider + + def construct( + config: ProviderConfig, + settings: Settings, + admission: ProviderAdmissionController, + ) -> BaseProvider: + return AlibabaCloudProvider(config, admission=admission) + + return construct + + def _load_lmstudio() -> ProviderFactory: from free_claude_code.providers.lmstudio import LMStudioProvider @@ -208,6 +221,7 @@ def construct( _SPECIAL_PROVIDER_FACTORIES: dict[str, Callable[[], ProviderFactory]] = { + "alibaba_cloud": _load_alibaba_cloud, "nvidia_nim": _load_nvidia_nim, "open_router": _load_open_router, "openai_api": _load_openai_api, diff --git a/src/free_claude_code/runtime/messaging_import.py b/src/free_claude_code/runtime/messaging_import.py index 82d50b4324..1ac3110633 100644 --- a/src/free_claude_code/runtime/messaging_import.py +++ b/src/free_claude_code/runtime/messaging_import.py @@ -201,6 +201,6 @@ def commit(connection: sqlite3.Connection) -> bool: database.execute( lambda connection: connection.execute( "UPDATE messaging_legacy_import SET cleanup_pending=0" - ) + ).close() ) return warning diff --git a/src/free_claude_code/runtime/sqlite_database.py b/src/free_claude_code/runtime/sqlite_database.py index 647b5d8b8b..4e276a528c 100644 --- a/src/free_claude_code/runtime/sqlite_database.py +++ b/src/free_claude_code/runtime/sqlite_database.py @@ -1,7 +1,7 @@ """Connection, transaction, and lifetime ownership of FCC's shared database. -The application holds the owner lock until both features close. Each connection -is opened, used, and closed within initialization or one synchronous transaction. +The application holds the owner lock until both features close and pooled +connections are disposed. Each transaction exclusively leases one connection. """ import asyncio @@ -10,6 +10,11 @@ from collections.abc import Callable from contextlib import closing from pathlib import Path +from typing import cast + +from sqlalchemy.engine.interfaces import DBAPIConnection +from sqlalchemy.exc import TimeoutError as PoolTimeoutError +from sqlalchemy.pool import QueuePool from free_claude_code.core.async_tasks import run_sync_owned from free_claude_code.core.interprocess_lock import InterprocessFileLock @@ -36,15 +41,22 @@ } -def _connect(path: Path, *, existing: bool = False) -> sqlite3.Connection: +def _connect( + path: Path, *, existing: bool = False, check_same_thread: bool = True +) -> sqlite3.Connection: connection = sqlite3.connect( path.as_uri() + ("?mode=rw" if existing else "?mode=rwc"), uri=True, timeout=10, autocommit=True, + check_same_thread=check_same_thread, ) - connection.row_factory = sqlite3.Row - connection.execute("PRAGMA foreign_keys = ON") + try: + connection.row_factory = sqlite3.Row + connection.execute("PRAGMA foreign_keys = ON").close() + except BaseException: + connection.close() + raise return connection @@ -191,6 +203,10 @@ def __init__( self._closing = False self._startup_error: Exception | None = None self._operations: set[asyncio.Task] = set() + self._writer: QueuePool | None = None + self._readers: QueuePool | None = None + self._write_admission = asyncio.Semaphore(1) + self._read_admission = asyncio.Semaphore(4) async def start(self) -> None: async with self._lifecycle: @@ -204,10 +220,10 @@ async def start(self) -> None: await run_sync_owned(self._initialize) except Exception as exc: self._startup_error = exc - self._owner.release() + await run_sync_owned(self._dispose) raise except BaseException: - self._owner.release() + await run_sync_owned(self._dispose) raise self._started = True @@ -224,37 +240,114 @@ def _initialize(self) -> None: for path in (self.path, Path(f"{self.path}-wal"), Path(f"{self.path}-shm")): if path.exists(): path.chmod(0o600) + self._writer = self._pool(1) + self._readers = self._pool(4) + + def _pool(self, size: int) -> QueuePool: + return QueuePool( + lambda: cast(DBAPIConnection, _connect(self.path, check_same_thread=False)), + pool_size=size, + max_overflow=0, + timeout=10, + reset_on_return=None, + ) + + def _dispose(self) -> None: + try: + if self._writer is not None: + self._writer.dispose() + self._writer = None + finally: + if self._readers is not None: + self._readers.dispose() + self._readers = None + self._owner.release() def execute[T]( self, operation: Callable[[sqlite3.Connection], T], *, write: bool = True ) -> T: - """Run one transaction inside an admitted worker, closing its connection. + """Run a writer-pool transaction inside an already admitted worker. Callbacks consume the connection synchronously and return materialized - results. They must not begin or finish the outer transaction themselves. + results. They must not control the outer transaction, retain or close + the connection, or change connection-wide settings. Legacy import uses this primitive for several transactions in one worker. """ - with closing(_connect(self.path)) as connection: + return self._transaction(self._writer, operation, write=write) + + def _transaction[T]( + self, + pool: QueuePool | None, + operation: Callable[[sqlite3.Connection], T], + *, + write: bool, + ) -> T: + if pool is None: + raise sqlite3.OperationalError("FCC storage is closed") + try: + lease = pool.connect() + except PoolTimeoutError as exc: + raise sqlite3.OperationalError( + "Timed out waiting for FCC storage connection capacity" + ) from exc + try: + connection = lease.driver_connection + if not isinstance(connection, sqlite3.Connection): + raise sqlite3.OperationalError("Invalid FCC storage connection") + if connection.in_transaction: + raise sqlite3.OperationalError("FCC storage connection is not clean") + except BaseException as exc: + lease.invalidate(exc) + lease.close() + raise + try: connection.execute("BEGIN IMMEDIATE" if write else "BEGIN") + result = operation(connection) + connection.execute("COMMIT") + return result + except BaseException as exc: try: - result = operation(connection) - connection.execute("COMMIT") - return result - except BaseException: if connection.in_transaction: connection.execute("ROLLBACK") - raise + if connection.in_transaction: + raise sqlite3.OperationalError( + "FCC storage rollback did not finish" + ) + except BaseException as cleanup_error: + raise exc from cleanup_error + finally: + # A traceback can retain a SELECT cursor after ROLLBACK. Never + # let that cursor's snapshot reach the next borrower. + lease.invalidate(exc) + raise + finally: + lease.close() async def run[T]( self, operation: Callable[[sqlite3.Connection], T], *, write: bool = True ) -> T: """Admit and supervise one transaction through committed result delivery.""" - return await self.work(lambda: self.execute(operation, write=write)) + if write: + return await self.work(lambda: self.execute(operation)) + return await self._work( + lambda: self._transaction(self._readers, operation, write=False), + self._read_admission, + ) async def work[T](self, operation: Callable[[], T]) -> T: + return await self._work(operation, self._write_admission) + + async def _work[T]( + self, operation: Callable[[], T], admission: asyncio.Semaphore + ) -> T: if not self._started or self._closing: raise sqlite3.OperationalError("FCC storage is closed") - task = asyncio.create_task(run_sync_owned(operation)) + + async def admitted() -> T: + async with admission: + return await run_sync_owned(operation) + + task = asyncio.create_task(admitted()) self._operations.add(task) try: # Deliver the result before cancellation can separate SQL commit from @@ -279,4 +372,4 @@ async def close(self) -> None: except asyncio.CancelledError: continue self._started = False - await run_sync_owned(self._owner.release) + await run_sync_owned(self._dispose) diff --git a/tests/cli/test_server_logging.py b/tests/cli/test_server_logging.py new file mode 100644 index 0000000000..91295f311f --- /dev/null +++ b/tests/cli/test_server_logging.py @@ -0,0 +1,127 @@ +"""Server logging contracts run outside pytest's process-wide logging state.""" + +import asyncio +import json +import logging +import logging.config +import subprocess +import sys +from contextlib import nullcontext +from copy import deepcopy +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +import uvicorn +from loguru import logger + +from free_claude_code.cli.commands import ServerSupervisor +from free_claude_code.cli.uvicorn_server import RuntimeServer +from free_claude_code.config.logging_config import configure_logging +from free_claude_code.config.settings import Settings +from free_claude_code.runtime.application import ApplicationRuntime + + +def _exercise_server_logging( + path: Path, console: bool, level: str, without_streams: bool +) -> None: + baseline = deepcopy(uvicorn.config.LOGGING_CONFIG) + # An earlier server may already have installed Uvicorn's console handlers. + logging.config.dictConfig(baseline) + settings = Settings(host="127.0.0.1", port=8082, log_level=level) + + def build_app(settings, restart_callback): + configure_logging(path, level=settings.log_level) + runtime = MagicMock( + spec=ApplicationRuntime, + settings=settings, + _draining=False, + _http_ready=asyncio.Event(), + is_closed=True, + begin_shutdown=lambda: None, + close=AsyncMock(return_value=True), + ) + runtime.http_started.side_effect = lambda: ApplicationRuntime.http_started( + runtime + ) + return SimpleNamespace(runtime=runtime) + + def run(server, sockets): + server._on_started() + logging.getLogger("uvicorn.access").info( + '%s - "%s %s HTTP/%s" %d', + "127.0.0.1:1234", + "POST", + "/test-request", + "1.1", + 200, + ) + try: + raise ValueError("controlled server failure") + except ValueError: + logging.getLogger("uvicorn.error").exception("ASGI request failed") + logging.getLogger("test.dependency").warning("dependency warning") + + with ( + patch("free_claude_code.runtime.bootstrap.build_asgi_app", build_app), + patch.object(RuntimeServer, "run", run), + patch.multiple(sys, stdout=None, stderr=None) + if without_streams + else nullcontext(), + ): + supervisor = ServerSupervisor(console_logging=console) + for _ in range(2): + supervisor._run_bound( + settings, [], open_admin_browser=False, restart_generation=0 + ) + logger.complete() + assert baseline == uvicorn.config.LOGGING_CONFIG + + +@pytest.mark.parametrize( + ("console", "without_streams"), + [(True, False), (False, False), (False, True)], + ids=["cli", "desktop", "desktop-no-console"], +) +@pytest.mark.parametrize("level", ["INFO", "WARNING"]) +def test_server_logs_reach_file_once_per_start( + tmp_path, console, level, without_streams +): + path = tmp_path / "server.log" + result = subprocess.run( + [ + sys.executable, + "-c", + "import sys; from pathlib import Path; " + "from tests.cli.test_server_logging import _exercise_server_logging; " + "_exercise_server_logging(Path(sys.argv[1]), sys.argv[2] == 'True', " + "sys.argv[3], sys.argv[4] == 'True')", + str(path), + str(console), + level, + str(without_streams), + ], + capture_output=True, + text=True, + encoding="utf-8", + timeout=30, + ) + assert result.returncode == 0, result.stdout + result.stderr + rows = [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines()] + messages = [row["record"]["message"] for row in rows] + info_count = 2 if level == "INFO" else 0 + assert sum(message.startswith("Admin UI:") for message in messages) == info_count + assert sum("/test-request" in message for message in messages) == info_count + assert messages.count("ASGI request failed") == 2 + assert messages.count("dependency warning") == 2 + failures = [ + row for row in rows if row["record"]["message"] == "ASGI request failed" + ] + for row in failures: + assert row["record"]["exception"]["type"] == "ValueError" + assert "controlled server failure" in row["text"] + assert "Traceback" in row["text"] + assert result.stdout.count("/test-request") == (2 if console else 0) + assert result.stderr.count("Admin UI:") == (2 if console else 0) + assert result.stderr.count("ASGI request failed") == (2 if console else 0) diff --git a/tests/config/test_logging_config.py b/tests/config/test_logging_config.py index 8ea8729cbb..8ae05efa05 100644 --- a/tests/config/test_logging_config.py +++ b/tests/config/test_logging_config.py @@ -2,9 +2,12 @@ import json import logging +import subprocess +import sys from pathlib import Path from unittest.mock import patch +import pytest from loguru import logger from free_claude_code.config import logging_config @@ -22,12 +25,101 @@ def test_log_capture_does_not_feed_back_into_interception(caplog, capsys, monkey def test_configure_logging_creates_parent_directories(tmp_path) -> None: - """Nested log path: parent directories are created before truncating.""" + """Nested log path: parent directories are created before opening.""" log_file = tmp_path / "nested" / "dir" / "app.log" configure_logging(str(log_file), force=True) assert log_file.is_file() +def test_native_json_preserves_bound_and_request_context(tmp_path): + log_file = tmp_path / "context.log" + configure_logging(log_file, force=True) + with logger.contextualize(request_id="req-test", session_id="session-test"): + logger.bind(provider_id="provider-test", run_id="run-test").info("completed") + logger.complete() + row = json.loads(log_file.read_text(encoding="utf-8")) + assert row["record"]["extra"] == { + "request_id": "req-test", + "session_id": "session-test", + "provider_id": "provider-test", + "run_id": "run-test", + } + assert row["record"]["level"]["name"] == "INFO" + assert row["record"]["message"] == "completed" + + +@pytest.mark.parametrize("stdlib", [False, True]) +def test_native_json_preserves_exception_chain_and_stack(tmp_path, stdlib): + log_file = tmp_path / "exception.log" + configure_logging(log_file, force=True) + try: + try: + raise ValueError("upstream HTTP 429: rate limited") + except ValueError as exc: + raise RuntimeError("provider request failed") from exc + except RuntimeError: + if stdlib: + logging.getLogger("test.exception").exception("request failed") + else: + logger.exception("request failed") + logger.complete() + rows = log_file.read_text(encoding="utf-8").splitlines() + assert len(rows) == 1 + row = json.loads(rows[0]) + assert row["record"]["exception"] == { + "type": "RuntimeError", + "value": "provider request failed", + "traceback": True, + } + assert "ValueError: upstream HTTP 429: rate limited" in row["text"] + assert "RuntimeError: provider request failed" in row["text"] + assert "test_native_json_preserves_exception_chain_and_stack" in row["text"] + + +def test_logging_preserves_records_across_process_restarts(tmp_path): + log_file = tmp_path / "server.log" + script = """ +import sys +from loguru import logger +from free_claude_code.config.logging_config import configure_logging +configure_logging(sys.argv[1]) +logger.info(sys.argv[2]) +logger.complete() +logger.remove() +""" + for message in ("first run", "second run"): + subprocess.run( + [sys.executable, "-c", script, str(log_file), message], + check=True, + capture_output=True, + text=True, + timeout=10, + ) + records = [ + json.loads(line) for line in log_file.read_text(encoding="utf-8").splitlines() + ] + assert [record["record"]["message"] for record in records] == [ + "first run", + "second run", + ] + + +def test_forced_logging_reconfiguration_preserves_records(tmp_path): + log_file = tmp_path / "server.log" + configure_logging(log_file, force=True) + logger.info("before reconfiguration") + configure_logging(log_file, force=True) + logger.info("after reconfiguration") + logger.complete() + records = [ + json.loads(line) for line in log_file.read_text(encoding="utf-8").splitlines() + ] + assert [record["record"]["message"] for record in records] == [ + "before reconfiguration", + "after reconfiguration", + ] + + def test_configure_logging_writes_json_to_file(tmp_path): """configure_logging writes JSON lines to the specified file.""" log_file = str(tmp_path / "test.log") @@ -83,26 +175,24 @@ def test_configure_logging_replaces_sink_when_path_changes(tmp_path): assert "second destination" in second_text -def test_telegram_bot_token_redacted_in_message_field(tmp_path) -> None: - log_file = str(tmp_path / "redact.log") - configure_logging(log_file, force=True, verbose_third_party=False) - token = "123456:ABCDEF-ghij-klm" - logger.info("Calling {}", f"https://api.telegram.org/bot{token}/getMe") - logger.complete() - text = Path(log_file).read_text(encoding="utf-8") - assert token not in text - assert "bot/" in text or "redacted" in text - - -def test_bearer_substring_redacted_in_log_file(tmp_path) -> None: - log_file = str(tmp_path / "bearer.log") - configure_logging(log_file, force=True, verbose_third_party=False) - secret = "ya29.secret-token-abc" - logger.info("Request headers: Authorization: Bearer {}", secret) +@pytest.mark.parametrize( + "message", + [ + "Calling https://api.telegram.org/bot123456:synthetic-token/getMe", + "Request headers: Authorization: Bearer synthetic-token", + "first line\nsecond line \u2603", + ], +) +def test_native_json_preserves_message_text(tmp_path, message): + log_file = tmp_path / "message.log" + configure_logging(log_file, force=True) + logger.info(message) logger.complete() - text = Path(log_file).read_text(encoding="utf-8") - assert secret not in text - assert "Bearer" in text + lines = log_file.read_text(encoding="utf-8").splitlines() + assert len(lines) == 1 + row = json.loads(lines[0]) + assert row["record"]["message"] == message + assert message in row["text"] def test_noisy_third_party_loggers_quieted_when_not_verbose(tmp_path) -> None: diff --git a/tests/contracts/test_feature_manifest.py b/tests/contracts/test_feature_manifest.py index 15b8122814..c7740c4b40 100644 --- a/tests/contracts/test_feature_manifest.py +++ b/tests/contracts/test_feature_manifest.py @@ -65,7 +65,10 @@ def test_product_coverage_is_not_satisfied_by_prereq_probes() -> None: def test_provider_and_platform_registries_include_builtins() -> None: + from free_claude_code.providers.alibaba_cloud import AlibabaCloudProvider + specialized_provider_classes = { + "alibaba_cloud": AlibabaCloudProvider, "openai": OpenAICodexProvider, "openai_api": OpenAIAPIProvider, "github_copilot": GitHubCopilotProvider, diff --git a/tests/contracts/test_provider_catalog_order.py b/tests/contracts/test_provider_catalog_order.py index af321f5ade..511d52d353 100644 --- a/tests/contracts/test_provider_catalog_order.py +++ b/tests/contracts/test_provider_catalog_order.py @@ -14,6 +14,7 @@ "openai_api", "github_copilot", "xai", + "alibaba_cloud", "qwencloud", "qwencloud_coding", "together", diff --git a/tests/contracts/test_smoke_child_process.py b/tests/contracts/test_smoke_child_process.py index 4aeb0efbf3..b9f6a9f03d 100644 --- a/tests/contracts/test_smoke_child_process.py +++ b/tests/contracts/test_smoke_child_process.py @@ -20,9 +20,28 @@ from smoke.lib.e2e import ConversationDriver from smoke.lib.http import collect_message_stream from smoke.lib.server import RunningServer +from smoke.product import test_client_product_live as client_smoke from smoke.product import test_provider_product_live as provider_smoke +def test_smoke_reads_native_trace_events_from_mixed_subprocess_output(): + output = ( + 'INFO: server started\n{"type":"result"}\n' + '{"text":"ordinary log record\\n","record":{"extra":{}}}\n' + '{"text":"trace event\\n","record":{"extra":{"trace_payload":' + '{"stage":"execution","event":"test.completed","source":"test","status":200}}}}\n' + ) + assert client_smoke._trace_log_events(output) == [ + { + "stage": "execution", + "event": "test.completed", + "source": "test", + "status": 200, + } + ] + assert {"type": "result"} in client_smoke._json_object_lines(output) + + def test_fcc_server_command_uses_cli_entrypoint() -> None: assert cmd_fcc_server() == [ child_process.python_exe(), diff --git a/tests/core/test_trace.py b/tests/core/test_trace.py index 528e14d74d..deccfb807b 100644 --- a/tests/core/test_trace.py +++ b/tests/core/test_trace.py @@ -54,13 +54,21 @@ def _json_log_rows(log_file: str) -> list[dict]: return [json.loads(line) for line in text.split("\n")] -def test_trace_payload_merged_into_json_line(tmp_path) -> None: +def _trace_payloads(log_file: str) -> list[dict]: + return [ + row["record"]["extra"][TRACE_PAYLOAD_BINDING] + for row in _json_log_rows(log_file) + if TRACE_PAYLOAD_BINDING in row["record"]["extra"] + ] + + +def test_trace_payload_preserved_in_native_json(tmp_path) -> None: log_file = str(tmp_path / "t.log") configure_logging(log_file, force=True, level="DEBUG") trace_event(stage="s", event="e.v1", source="unit", hello="world", n=42) row = _json_log_rows(log_file)[-1] - assert row["level"] == "DEBUG" - assert row["trace"] is True + assert row["record"]["level"]["name"] == "DEBUG" + row = row["record"]["extra"][TRACE_PAYLOAD_BINDING] assert row["stage"] == "s" assert row["event"] == "e.v1" assert row["source"] == "unit" @@ -77,7 +85,7 @@ def test_trace_payload_excluded_from_default_info_logs(tmp_path) -> None: logger.info("visible lifecycle event") rows = _json_log_rows(log_file) - assert [row["message"] for row in rows] == ["visible lifecycle event"] + assert [row["record"]["message"] for row in rows] == ["visible lifecycle event"] def test_sanitize_masks_nested_api_key_strings() -> None: @@ -112,7 +120,7 @@ async def test_traced_async_stream_logs_completion(tmp_path) -> None: assert chunks == ["hello", " world"] assert source.close_calls == 1 - rows = _json_log_rows(log_file) + rows = _trace_payloads(log_file) completed = [row for row in rows if row.get("event") == "stream.completed"] assert len(completed) == 1 assert completed[0]["request_id"] == "req_complete" @@ -144,7 +152,7 @@ async def test_traced_async_stream_logs_real_exception(tmp_path) -> None: assert source.close_calls == 1 - rows = _json_log_rows(log_file) + rows = _trace_payloads(log_file) interrupted = [row for row in rows if row.get("event") == "stream.interrupted"] assert len(interrupted) == 1 assert interrupted[0]["request_id"] == "req_error" @@ -180,7 +188,7 @@ async def test_traced_async_stream_closes_quietly_on_generator_exit(tmp_path) -> await stream.aclose() assert source.close_calls == 1 - rows = _json_log_rows(log_file) + rows = _trace_payloads(log_file) events = {row.get("event") for row in rows} assert "stream.completed" not in events assert "stream.interrupted" not in events diff --git a/tests/providers/test_alibaba_cloud.py b/tests/providers/test_alibaba_cloud.py new file mode 100644 index 0000000000..3cba12dc96 --- /dev/null +++ b/tests/providers/test_alibaba_cloud.py @@ -0,0 +1,376 @@ +"""Alibaba Cloud discovery and configuration contracts.""" + +import json +from contextlib import asynccontextmanager + +import httpx2 +import pytest +from openai import AsyncOpenAI + +from free_claude_code.config.provider_catalog import PROVIDER_CATALOG +from free_claude_code.config.settings import Settings +from free_claude_code.core.anthropic.models import MessagesRequest +from free_claude_code.core.model_capabilities import ModelInputModality +from free_claude_code.core.openai_responses import OpenAIResponsesRequest +from free_claude_code.providers.model_listing import ModelListResponseError +from free_claude_code.providers.runtime import build_provider_config +from tests.providers.support import immediate_admission + +BASE_URL = "https://dashscope-intl.aliyuncs.com/compatible-mode/v1" + + +def model(model_id="qwen-coder", **metadata): + return { + "model": model_id, + "features": ["function-calling"], + "capabilities": ["TG", "Reasoning"], + "inference_metadata": {"request_modality": ["Text", "Image"]}, + "model_info": {"context_window": 131072, "max_output_tokens": 16384}, + **metadata, + } + + +def page(models, *, number=1, total=None): + return { + "success": True, + "output": { + "page_no": number, + "page_size": 100, + "total": len(models) if total is None else total, + "models": models, + }, + } + + +@asynccontextmanager +async def provider_for(handler, base_url=BASE_URL): + from free_claude_code.providers.alibaba_cloud import AlibabaCloudProvider + + settings = Settings( + ALIBABA_CLOUD_API_KEY="test-key", ALIBABA_CLOUD_BASE_URL=base_url + ) + config = build_provider_config(PROVIDER_CATALOG["alibaba_cloud"], settings) + async with AsyncOpenAI( + api_key=config.api_key, + base_url=config.base_url, + max_retries=0, + http_client=httpx2.AsyncClient(transport=httpx2.MockTransport(handler)), + ) as client: + provider = AlibabaCloudProvider( + config, + admission=immediate_admission(provider_name="alibaba_cloud"), + client=client, + ) + try: + yield provider + finally: + await provider.cleanup() + + +def test_configuration_defaults_and_region_override(): + descriptor = PROVIDER_CATALOG["alibaba_cloud"] + settings = Settings(ALIBABA_CLOUD_API_KEY=" key ") + config = build_provider_config(descriptor, settings) + assert config.api_key == "key" + assert config.base_url == BASE_URL + assert descriptor.configuration_attrs() == ("alibaba_cloud_api_key",) + + settings = Settings( + ALIBABA_CLOUD_API_KEY="regional-key", + ALIBABA_CLOUD_BASE_URL="https://workspace.cn-beijing.maas.aliyuncs.com/compatible-mode/v1", + ALIBABA_CLOUD_PROXY="http://proxy.test:8080", + ) + config = build_provider_config(descriptor, settings) + assert config.api_key == "regional-key" + assert config.base_url == settings.alibaba_cloud_base_url + assert config.proxy == "http://proxy.test:8080" + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "endpoint", + [BASE_URL, "https://workspace.cn-beijing.maas.aliyuncs.com/compatible-mode/v1/"], +) +async def test_discovery_fetches_all_pages_with_same_region_and_auth(endpoint): + requests = [] + + def handler(request): + requests.append(request) + number = int(request.url.params["page_no"]) + return httpx2.Response( + 200, json=page([model(f"model-{number}")], number=number, total=2) + ) + + async with provider_for(handler, endpoint) as provider: + infos = await provider.list_model_infos() + + assert {info.model_id for info in infos} == {"model-1", "model-2"} + assert len(requests) == 2 + for request in requests: + assert request.url.host == httpx2.URL(endpoint).host + assert request.url.path == "/api/v1/models" + assert request.headers["authorization"] == "Bearer test-key" + assert request.url.params["features"] == "function-calling" + assert request.url.params["supports"] == "inference" + for info in infos: + assert info.supports_thinking is True + assert info.context_window_tokens == 131072 + assert info.max_output_tokens == 16384 + assert info.input_modalities == frozenset( + {ModelInputModality.TEXT, ModelInputModality.IMAGE} + ) + + +@pytest.mark.asyncio +async def test_discovery_filters_transport_after_counting_every_page(): + requests = [] + pages = [ + [ + model("realtime", capabilities=["Realtime-Omni"]), + model("voice", capabilities=["TG", "Realtime-Chatting"]), + model("audio", capabilities=["ASR"]), + model("unknown", capabilities=[]), + ], + [ + model("chat", capabilities=["TG"]), + model("reasoner", capabilities=["Reasoning"]), + model("vision", capabilities=["VU"]), + model("omni", capabilities=["Multimodal-Omni"]), + ], + ] + + def handler(request): + number = int(request.url.params["page_no"]) + requests.append(number) + return httpx2.Response( + 200, json=page(pages[number - 1], number=number, total=8) + ) + + async with provider_for(handler) as provider: + infos = await provider.list_model_infos() + + assert requests == [1, 2] + assert {info.model_id for info in infos} == {"chat", "reasoner", "vision", "omni"} + + +@pytest.mark.asyncio +@pytest.mark.parametrize("capabilities", [None, [], ["New-API"], "TG", [1]]) +async def test_discovery_does_not_assume_unknown_capabilities_support_chat( + capabilities, +): + payload = page([model(), model("unknown", capabilities=capabilities)]) + async with provider_for(lambda _: httpx2.Response(200, json=payload)) as provider: + infos = await provider.list_model_infos() + assert {info.model_id for info in infos} == {"qwen-coder"} + + +@pytest.mark.asyncio +@pytest.mark.parametrize("same_page", [False, True]) +@pytest.mark.parametrize("capabilities", [["TG"], ["Realtime-Omni"]]) +async def test_discovery_rejects_duplicate_ids_before_publishing_catalog( + same_page, capabilities +): + requests = [] + repeated = model("repeated", capabilities=capabilities) + pages = ( + [[model("chat"), repeated, repeated]] + if same_page + else [[model("chat"), repeated], [repeated]] + ) + + def handler(request): + number = int(request.url.params["page_no"]) + requests.append(number) + return httpx2.Response( + 200, json=page(pages[number - 1], number=number, total=3) + ) + + async with provider_for(handler) as provider: + with pytest.raises(ModelListResponseError, match="duplicate"): + await provider.list_model_infos() + assert requests == ([1] if same_page else [1, 2]) + + +@pytest.mark.asyncio +async def test_discovery_filters_non_tool_models_without_guessing_missing_limits(): + payload = page( + [model(), model("embedding", features=[]), model("unknown", model_info=None)] + ) + async with provider_for(lambda _: httpx2.Response(200, json=payload)) as provider: + infos = {info.model_id: info for info in await provider.list_model_infos()} + assert set(infos) == {"qwen-coder", "unknown"} + assert infos["unknown"].context_window_tokens is None + assert infos["unknown"].max_output_tokens is None + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "payload", + [ + {"success": False, "code": "InvalidApiKey"}, + {"success": True}, + page([]), + page([], total=2), + page([model()], number=2), + page([model()], total=-1), + page([model()], total=True), + page([model()], total="1"), + page([{"features": ["function-calling"]}]), + ], +) +async def test_discovery_rejects_malformed_or_empty_catalog(payload): + async with provider_for(lambda _: httpx2.Response(200, json=payload)) as provider: + with pytest.raises(ModelListResponseError): + await provider.list_model_infos() + + +@pytest.mark.asyncio +async def test_later_page_failure_does_not_publish_partial_catalog(): + def handler(request): + number = int(request.url.params["page_no"]) + payload = ( + page([model()], total=2) if number == 1 else page([], number=2, total=2) + ) + return httpx2.Response(200, json=payload) + + async with provider_for(handler) as provider: + with pytest.raises(ModelListResponseError): + await provider.list_model_infos() + + +@pytest.mark.asyncio +async def test_discovery_retries_only_the_failed_page(): + pages = [] + + def handler(request): + number = int(request.url.params["page_no"]) + pages.append(number) + if pages == [1, 2]: + return httpx2.Response(503, json={"error": {"message": "Unavailable"}}) + return httpx2.Response( + 200, json=page([model(f"model-{number}")], number=number, total=2) + ) + + async with provider_for(handler) as provider: + infos = await provider.list_model_infos() + assert pages == [1, 2, 2] + assert {info.model_id for info in infos} == {"model-1", "model-2"} + + +@pytest.mark.asyncio +@pytest.mark.parametrize("wire", ["messages", "responses"]) +async def test_chat_stream_preserves_reasoning_tools_and_usage(wire): + requests = [] + chunks = [ + {"choices": [{"index": 0, "delta": {"reasoning_content": "Inspecting."}}]}, + { + "choices": [ + { + "index": 0, + "delta": { + "tool_calls": [ + { + "index": 0, + "id": "call_1", + "type": "function", + "function": { + "name": "inspect", + "arguments": '{"path":', + }, + } + ] + }, + } + ] + }, + { + "choices": [ + { + "index": 0, + "delta": { + "tool_calls": [ + {"index": 0, "function": {"arguments": '"README.md"}'}} + ] + }, + "finish_reason": "tool_calls", + } + ] + }, + { + "choices": [], + "usage": {"prompt_tokens": 25, "completion_tokens": 10, "total_tokens": 35}, + }, + ] + + def handler(request): + requests.append(request) + frames = [ + { + "id": "chat_1", + "object": "chat.completion.chunk", + "created": 0, + "model": "qwen-coder", + **chunk, + } + for chunk in chunks + ] + return httpx2.Response( + 200, + headers={"content-type": "text/event-stream"}, + text="".join(f"data: {json.dumps(frame)}\n\n" for frame in frames) + + "data: [DONE]\n\n", + ) + + schema = {"type": "object", "properties": {"path": {"type": "string"}}} + async with provider_for(handler) as provider: + stream = ( + provider.stream_messages( + MessagesRequest( + model="qwen-coder", + max_tokens=2048, + messages=[{"role": "user", "content": "Inspect the README."}], + tools=[{"name": "inspect", "input_schema": schema}], + ) + ) + if wire == "messages" + else provider.stream_responses( + OpenAIResponsesRequest( + model="qwen-coder", + input="Inspect the README.", + max_output_tokens=2048, + tools=[ + {"type": "function", "name": "inspect", "parameters": schema} + ], + ) + ) + ) + output = "".join([event async for event in stream]) + + assert len(requests) == 1 + assert requests[0].url.path == "/compatible-mode/v1/chat/completions" + assert requests[0].headers["authorization"] == "Bearer test-key" + body = json.loads(requests[0].content) + assert body["model"] == "qwen-coder" + assert body["stream"] is True + assert body["stream_options"] == {"include_usage": True} + assert body["max_tokens"] == 2048 + assert body["tools"][0]["function"]["parameters"] == schema + assert "enable_thinking" not in body + assert "Inspecting." in output + assert "inspect" in output + assert "README.md" in output + events = [ + json.loads(line[6:]) + for line in output.splitlines() + if line.startswith("data: {") + ] + if wire == "messages": + terminal = next(event for event in events if event["type"] == "message_delta") + assert terminal["delta"]["stop_reason"] == "tool_use" + assert terminal["usage"]["output_tokens"] == 10 + else: + terminal = next( + event for event in events if event["type"] == "response.completed" + ) + assert terminal["response"]["usage"]["input_tokens"] == 25 + assert terminal["response"]["usage"]["output_tokens"] == 10 diff --git a/tests/providers/test_provider_runtime.py b/tests/providers/test_provider_runtime.py index da9a216f24..39e9e860c5 100644 --- a/tests/providers/test_provider_runtime.py +++ b/tests/providers/test_provider_runtime.py @@ -46,6 +46,7 @@ from free_claude_code.providers.admission import ProviderAdmissionController from free_claude_code.providers.admission_policy import ProviderAdmissionLimits from free_claude_code.providers.admission_registry import ProviderAdmissionRegistry +from free_claude_code.providers.alibaba_cloud import AlibabaCloudProvider from free_claude_code.providers.cloudflare import CloudflareProvider from free_claude_code.providers.deepseek import DeepSeekProvider from free_claude_code.providers.gemini import GeminiProvider @@ -84,6 +85,9 @@ def _make_settings(**overrides): mock.nvidia_nim_api_key = "test_key" mock.open_router_api_key = "test_openrouter_key" mock.xai_api_key = "test_xai_key" + mock.alibaba_cloud_api_key = "test_alibaba_key" + mock.alibaba_cloud_base_url = None + mock.alibaba_cloud_proxy = None mock.qwencloud_api_key = "test_qwencloud_key" mock.qwencloud_coding_api_key = "test_qwencloud_coding_key" mock.together_api_key = "test_together_key" @@ -1097,6 +1101,7 @@ async def test_create_provider_instantiates_each_builtin(): "github_copilot": GitHubCopilotProvider, "cline_pass": OpenAIChatProvider, "xai": OpenAIChatProvider, + "alibaba_cloud": AlibabaCloudProvider, "qwencloud": OpenAIChatProvider, "qwencloud_coding": OpenAIChatProvider, "together": OpenAIChatProvider, diff --git a/tests/runtime/test_code_sessions_sqlite.py b/tests/runtime/test_code_sessions_sqlite.py index f0f899511f..3132f91c30 100644 --- a/tests/runtime/test_code_sessions_sqlite.py +++ b/tests/runtime/test_code_sessions_sqlite.py @@ -2,6 +2,7 @@ import json import sqlite3 import threading +import traceback import uuid from contextlib import closing from typing import Literal @@ -22,6 +23,52 @@ from free_claude_code.runtime.sqlite_database import SQLiteDatabase +@pytest.mark.asyncio +async def test_failed_history_decode_cannot_leak_snapshot_to_next_reader(store): + session = await store.create( + CodeSession(id="s", cwd="/work", model="provider/model", title="before") + ) + + def seed(connection): + for index in range(2): + code_store_module._insert( + connection, + "code_runs", + CodeRun( + id=f"r{index}", + session_id=session.id, + text="test", + model=session.model, + ordinal=index + 1, + status="completed", + ), + ) + connection.execute( + "UPDATE code_runs SET error_details='invalid json' WHERE id='r0'" + ).close() + + await store.database.run(seed) + with pytest.raises(json.JSONDecodeError) as failure: + await store.runs(session.id) + # Keep the traceback, and therefore its unfinished SELECT cursor, alive. + try: + await store.database.run( + lambda connection: connection.execute( + "UPDATE code_sessions SET title='after' WHERE id=?", (session.id,) + ).close() + ) + with closing(sqlite3.connect(store.database.path)) as external: + assert ( + external.execute( + "SELECT title FROM code_sessions WHERE id=?", (session.id,) + ).fetchone()[0] + == "after" + ) + assert (await store.get_session(session.id)).title == "after" + finally: + traceback.clear_frames(failure.value.__traceback__) + + @pytest.mark.asyncio async def test_cancelled_initialization_drains_thread_before_releasing_owner_lock( database_factory, diff --git a/tests/runtime/test_messaging_sqlite.py b/tests/runtime/test_messaging_sqlite.py index 6fc791801a..835d036987 100644 --- a/tests/runtime/test_messaging_sqlite.py +++ b/tests/runtime/test_messaging_sqlite.py @@ -241,7 +241,7 @@ def fail_second_tree(connection, snapshot): assert source.exists() assert (await storage.load_conversation_snapshot()).is_empty assert ( - storage.database.execute( + await storage.database.run( lambda connection: connection.execute( "SELECT count(*) FROM messaging_legacy_import" ).fetchone()[0], @@ -259,8 +259,8 @@ def incoming(node, *, text="prompt"): ) -def import_receipt(database): - return database.execute( +async def import_receipt(database): + return await database.run( lambda connection: dict( connection.execute("SELECT * FROM messaging_legacy_import").fetchone() ), @@ -313,7 +313,7 @@ async def test_import_rejects_storage_invalid_tree_without_reserving_references( after.identity: after, } assert source.read_text() == original - assert import_receipt(storage.database) == { + assert await import_receipt(storage.database) == { "source": "sessions.json", "outcome": "partial", "trees": 2, @@ -357,7 +357,7 @@ async def test_import_message_storage_validation_and_successful_duplicate_owners rejected = value == "\ud800" or field == "kind" assert bool(await import_legacy(storage.database, source)) is rejected assert source.exists() is rejected - rows = storage.database.execute( + rows = await storage.database.run( lambda connection: [ tuple(row) for row in connection.execute( @@ -378,7 +378,7 @@ async def test_import_message_storage_validation_and_successful_duplicate_owners bad["kind"], ) assert len(rows) == (1 if field == "ts" else 2) - receipt = import_receipt(storage.database) + receipt = await import_receipt(storage.database) assert receipt["messages"] == len(rows) assert receipt["skipped"] == int(rejected) @@ -395,7 +395,7 @@ async def test_import_last_tree_identity_wins_even_when_unrecoverable( assert (await storage.load_conversation_snapshot()).trees == ( {} if last_invalid else {last.identity: last} ) - assert import_receipt(storage.database)["skipped"] == 1 + int(last_invalid) + assert (await import_receipt(storage.database))["skipped"] == 1 + int(last_invalid) await storage.commit_trees((tree(root="fresh"),)) @@ -416,7 +416,7 @@ async def test_partial_import_survives_reopen_and_clear_without_replay( ) ) assert await import_legacy(storage.database, source) - receipt = import_receipt(storage.database) + receipt = await import_receipt(storage.database) await storage.database.close() for restart in range(2): database = SQLiteDatabase(storage.database.path, tmp_path / "code.lock") @@ -424,7 +424,7 @@ async def test_partial_import_survives_reopen_and_clear_without_replay( try: reopened = SQLiteMessagingStore(database) assert await import_legacy(database, source) is None - assert import_receipt(database) == receipt + assert await import_receipt(database) == receipt assert (await reopened.load_conversation_snapshot()).is_empty is bool( restart ) @@ -438,8 +438,6 @@ async def test_partial_import_survives_reopen_and_clear_without_replay( async def test_import_outer_failure_rolls_back_released_units( storage, tmp_path, failure_stage ): - from free_claude_code.runtime import sqlite_database - source = tmp_path / "sessions.json" source.write_text( json.dumps( @@ -453,12 +451,10 @@ async def test_import_outer_failure_rolls_back_released_units( } ) ) - connect = sqlite_database._connect + transaction = storage.database._transaction inserted = [] - def faulting_connect(path, *, existing=False): - connection = connect(path, existing=existing) - + def faulting_transaction(pool, operation, *, write): def authorize(action, arg1, arg2, database_name, trigger): if action == sqlite3.SQLITE_INSERT: inserted.append(arg1) @@ -473,12 +469,26 @@ def authorize(action, arg1, arg2, database_name, trigger): return sqlite3.SQLITE_DENY return sqlite3.SQLITE_OK - connection.set_authorizer(authorize) - return connection + observed = [] + + def faulting(connection): + observed.append(connection) + connection.set_authorizer(authorize) + return operation(connection) + result = transaction(pool, faulting, write=write) + # Failed transactions discard their connection; only successful leases + # need the injected connection state restored before another checkout. + for connection in observed: + connection.set_authorizer(None) + return result + + await storage.database.run( + lambda connection: connection.execute("SELECT 1").close() + ) with ( - patch.object(sqlite_database, "_connect", faulting_connect), - pytest.raises(sqlite3.DatabaseError), + patch.object(storage.database, "_transaction", faulting_transaction), + pytest.raises(sqlite3.DatabaseError, match="not authorized"), ): await import_legacy(storage.database, source) assert "messaging_nodes" in inserted and "messaging_managed_messages" in inserted @@ -486,7 +496,7 @@ def authorize(action, arg1, arg2, database_name, trigger): assert (await storage.load_conversation_snapshot()).is_empty assert not await storage.get_tracked_message_ids_for_chat("telegram", "chat") assert ( - storage.database.execute( + await storage.database.run( lambda connection: connection.execute( "SELECT count(*) FROM messaging_legacy_import" ).fetchone()[0], @@ -495,8 +505,8 @@ def authorize(action, arg1, arg2, database_name, trigger): == 0 ) assert await import_legacy(storage.database, source) is None - assert import_receipt(storage.database)["trees"] == 1 - assert import_receipt(storage.database)["messages"] == 1 + assert (await import_receipt(storage.database))["trees"] == 1 + assert (await import_receipt(storage.database))["messages"] == 1 @pytest.mark.parametrize("platform", ["none", "telegram"]) @@ -576,11 +586,11 @@ async def test_startup_partial_import_preserves_links_scopes_and_code( "2", "3", ] - assert import_receipt(storage.database)["skipped"] == 2 - assert import_receipt(storage.database)["messages"] == 3 + assert (await import_receipt(storage.database))["skipped"] == 2 + assert (await import_receipt(storage.database))["messages"] == 3 assert await code.get_session(session.id) == session assert ( - storage.database.execute( + await storage.database.run( lambda connection: connection.execute( "PRAGMA foreign_key_check" ).fetchall(), diff --git a/tests/runtime/test_sqlite_database.py b/tests/runtime/test_sqlite_database.py index e98bf2afb4..a31aa4279d 100644 --- a/tests/runtime/test_sqlite_database.py +++ b/tests/runtime/test_sqlite_database.py @@ -9,6 +9,8 @@ import httpx import pytest +from anyio import to_thread +from sqlalchemy.exc import TimeoutError as PoolTimeoutError from free_claude_code.application.code_sessions.models import CodeUnavailableError from free_claude_code.runtime import sqlite_database @@ -16,6 +18,365 @@ from free_claude_code.runtime.sqlite_database import initialize_database +@pytest.mark.asyncio +async def test_transactions_reuse_connections_until_database_closes( + database_factory, tmp_path +): + database = database_factory(tmp_path / "fcc.db", tmp_path / "fcc.lock") + await database.start() + observed = [] + + def observe(connection): + observed.append(connection) + assert connection.in_transaction + assert connection.autocommit is True + assert connection.execute("PRAGMA synchronous").fetchone()[0] == 2 + assert connection.execute("PRAGMA foreign_keys").fetchone()[0] == 1 + assert connection.execute("PRAGMA journal_mode").fetchone()[0] == "wal" + return connection.execute("SELECT 1 AS value").fetchone()["value"] + + for write in (True, True, False, False): + assert await database.run(observe, write=write) == 1 + assert observed[0] is observed[1] + assert observed[2] is observed[3] + assert observed[0] is not observed[2] + await database.close() + for connection in observed: + with pytest.raises(sqlite3.ProgrammingError, match="closed"): + connection.execute("SELECT 1") + + +@pytest.mark.asyncio +async def test_queued_writers_leave_workers_available_and_drain_on_close( + database_factory, tmp_path +): + database = database_factory(tmp_path / "fcc.db", tmp_path / "fcc.lock") + await database.start() + entered, release = threading.Event(), threading.Event() + limiter = to_thread.current_default_thread_limiter() + previous_limit = limiter.total_tokens + limiter.total_tokens = 2 + + def write(connection, value): + if value == "0": + entered.set() + assert release.wait(10) + connection.execute("INSERT INTO code_deleted VALUES (?)", (value,)).close() + return value + + writers = [asyncio.create_task(database.run(lambda c: write(c, "0")))] + readers = [] + closing_task = None + try: + assert await asyncio.to_thread(entered.wait, 3) + writers.extend( + asyncio.create_task(database.run(lambda c, i=i: write(c, str(i)))) + for i in range(1, 12) + ) + # Let each accepted call register its admission task before the probes. + await asyncio.sleep(0) + await asyncio.sleep(0) + writers[-1].cancel() + readers = [ + asyncio.create_task( + database.run( + lambda c: c.execute("SELECT count(*) FROM code_deleted").fetchone()[ + 0 + ], + write=False, + ) + ), + asyncio.create_task(to_thread.run_sync(lambda: "available")), + ] + done, pending = await asyncio.wait(readers, timeout=3) + assert not pending + assert {task.result() for task in done} == {0, "available"} + closing_task = asyncio.create_task(database.close()) + await asyncio.sleep(0) + assert not closing_task.done() + with pytest.raises(sqlite3.OperationalError, match="closed"): + await database.run(lambda c: None) + other = database_factory(database.path, tmp_path / "fcc.lock") + with pytest.raises(sqlite3.OperationalError, match="another FCC"): + await other.start() + finally: + release.set() + outcomes = await asyncio.gather(*writers, *readers, return_exceptions=True) + if closing_task is not None: + await closing_task + limiter.total_tokens = previous_limit + assert outcomes[:12] == [str(i) for i in range(12)] + reopened = database_factory(database.path, tmp_path / "fcc.lock") + await reopened.start() + assert ( + await reopened.run( + lambda c: c.execute("SELECT count(*) FROM code_deleted").fetchone()[0], + write=False, + ) + == 12 + ) + + +@pytest.mark.asyncio +async def test_reader_leases_are_bounded_and_can_move_between_threads( + database_factory, tmp_path +): + database = database_factory(tmp_path / "fcc.db", tmp_path / "fcc.lock") + await database.start() + four_readers, release = threading.Event(), threading.Event() + observed = [] + guard = threading.Lock() + + def read(connection): + with guard: + observed.append(connection) + if len(observed) == 4: + four_readers.set() + assert release.wait(10) + return connection.execute("SELECT 1").fetchone()[0] + + tasks = [asyncio.create_task(database.run(read, write=False)) for _ in range(10)] + try: + assert await asyncio.to_thread(four_readers.wait, 3) + assert len(observed) == 4 + finally: + release.set() + results = await asyncio.gather(*tasks) + assert results == [1] * 10 + assert len(set(observed)) == 4 + # Force two different worker threads to reuse the single writer connection. + seen = [] + + def cross_threads(): + errors = [] + finished, release_thread = threading.Event(), threading.Event() + + def execute(wait): + try: + database.execute(lambda c: seen.append(c)) + except BaseException as exc: + errors.append(exc) + finally: + if wait: + finished.set() + assert release_thread.wait(5) + + thread = threading.Thread(target=execute, args=(True,)) + thread.start() + try: + assert finished.wait(3) + execute(False) + finally: + release_thread.set() + thread.join(3) + assert not thread.is_alive() + assert not errors + + await database.work(cross_threads) + assert seen[0] is seen[1] + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "failure", ["begin", "callback", "commit", "rollback", "closed"] +) +async def test_failed_transactions_discard_the_connection( + database_factory, tmp_path, failure +): + database = database_factory(tmp_path / "fcc.db", tmp_path / "fcc.lock") + await database.start() + observed = [] + + def authorize(action, arg1, arg2, name, trigger): + denied = {"begin": "BEGIN", "commit": "COMMIT", "rollback": "ROLLBACK"} + if action == sqlite3.SQLITE_TRANSACTION and arg1 == denied.get(failure): + return sqlite3.SQLITE_DENY + return sqlite3.SQLITE_OK + + def configure(connection): + observed.append(connection) + connection.set_authorizer(authorize) + + # Configure before BEGIN without wrapping/replacing sqlite3's transaction logic. + def configure_lease(): + with closing(database._writer.connect()) as lease: + configure(lease.driver_connection) + + await database.work(configure_lease) + + def fail(connection): + connection.execute("INSERT INTO code_deleted VALUES ('failed')").close() + if failure == "closed": + connection.close() + if failure in ("callback", "rollback", "closed"): + raise ValueError("original callback failure") + + error_type = ( + ValueError + if failure in ("callback", "rollback", "closed") + else sqlite3.DatabaseError + ) + with pytest.raises(error_type) as error: + await database.run(fail) + if failure in ("rollback", "closed"): + assert isinstance(error.value.__cause__, sqlite3.Error) + with pytest.raises(sqlite3.ProgrammingError, match="closed"): + observed[0].execute("SELECT 1") + + def verify(connection): + assert connection is not observed[0] + assert ( + connection.execute("SELECT count(*) FROM code_deleted").fetchone()[0] == 0 + ) + connection.execute("INSERT INTO code_deleted VALUES ('success')").close() + + await database.run(verify) + + +@pytest.mark.asyncio +async def test_checkout_timeout_uses_sqlite_error_contract( + database_factory, tmp_path, monkeypatch +): + database = database_factory(tmp_path / "fcc.db", tmp_path / "fcc.lock") + await database.start() + + def timeout(): + raise PoolTimeoutError("injected capacity timeout") + + monkeypatch.setattr(database._writer, "connect", timeout) + with pytest.raises(sqlite3.OperationalError, match="capacity") as error: + await database.run(lambda c: pytest.fail("callback must not run")) + assert isinstance(error.value.__cause__, PoolTimeoutError) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("cancel", [False, True]) +async def test_startup_failure_disposes_connections_before_releasing_owner( + tmp_path, cancel +): + entered, release = threading.Event(), threading.Event() + observed = [] + + class GatedDatabase(sqlite_database.SQLiteDatabase): + def _initialize(self): + super()._initialize() + self.execute(lambda c: observed.append(c)) + entered.set() + assert release.wait(10) + if not cancel: + raise sqlite3.OperationalError("injected startup failure") + + database = GatedDatabase(tmp_path / "fcc.db", tmp_path / "fcc.lock") + starting = asyncio.create_task(database.start()) + try: + assert await asyncio.to_thread(entered.wait, 3) + if cancel: + starting.cancel() + competing = sqlite_database.SQLiteDatabase(database.path, tmp_path / "fcc.lock") + with pytest.raises(sqlite3.OperationalError, match="another FCC"): + await competing.start() + await competing.close() + finally: + release.set() + try: + with pytest.raises( + asyncio.CancelledError if cancel else sqlite3.OperationalError + ): + await starting + with pytest.raises(sqlite3.ProgrammingError, match="closed"): + observed[0].execute("SELECT 1") + reopened = sqlite_database.SQLiteDatabase(database.path, tmp_path / "fcc.lock") + try: + await reopened.start() + finally: + await reopened.close() + finally: + await database.close() + + +def test_connection_setup_failure_closes_raw_connection(tmp_path, monkeypatch): + connect = sqlite3.connect + observed = [] + + class FailedSetup(sqlite3.Connection): + def execute(self, sql, *args): + if sql.startswith("PRAGMA foreign_keys"): + raise sqlite3.OperationalError("injected configuration failure") + return super().execute(sql, *args) + + def create(*args, **kwargs): + connection = connect(*args, **kwargs, factory=FailedSetup) + observed.append(connection) + return connection + + monkeypatch.setattr(sqlite_database.sqlite3, "connect", create) + with pytest.raises(sqlite3.OperationalError, match="configuration"): + sqlite_database._connect(tmp_path / "fcc.db") + with pytest.raises(sqlite3.ProgrammingError, match="closed"): + observed[0].execute("SELECT 1") + + +@pytest.mark.asyncio +async def test_partial_pool_creation_cleans_up_writer(tmp_path, monkeypatch): + database = sqlite_database.SQLiteDatabase( + tmp_path / "fcc.db", tmp_path / "fcc.lock" + ) + create_pool = database._pool + observed = [] + + def partial(size): + if size == 4: + raise RuntimeError("reader pool creation failed") + pool = create_pool(size) + with closing(pool.connect()) as lease: + observed.append(lease.driver_connection) + return pool + + monkeypatch.setattr(database, "_pool", partial) + try: + with pytest.raises(RuntimeError, match="reader pool"): + await database.start() + with pytest.raises(sqlite3.ProgrammingError, match="closed"): + observed[0].execute("SELECT 1") + other = sqlite_database.SQLiteDatabase(database.path, tmp_path / "fcc.lock") + try: + await other.start() + finally: + await other.close() + finally: + await database.close() + + +@pytest.mark.asyncio +async def test_disposal_failure_retains_owner_and_can_be_retried( + database_factory, tmp_path, monkeypatch +): + database = database_factory(tmp_path / "fcc.db", tmp_path / "fcc.lock") + await database.start() + reader_connections = [] + await database.run(lambda c: reader_connections.append(c), write=False) + writer = database._writer + dispose = writer.dispose + + def fail(): + raise RuntimeError("injected disposal failure") + + monkeypatch.setattr(writer, "dispose", fail) + try: + with pytest.raises(RuntimeError, match="disposal"): + await database.close() + with pytest.raises(sqlite3.ProgrammingError, match="closed"): + reader_connections[0].execute("SELECT 1") + competing = database_factory(database.path, tmp_path / "fcc.lock") + with pytest.raises(sqlite3.OperationalError, match="another FCC"): + await competing.start() + finally: + monkeypatch.setattr(writer, "dispose", dispose) + await database.close() + reopened = database_factory(database.path, tmp_path / "fcc.lock") + await reopened.start() + + def historical_database(path: Path, version: int) -> None: path.parent.mkdir(parents=True, exist_ok=True) schema = (Path(__file__).parent / "fixtures/code_schema_v0.sql").read_text() diff --git a/uv.lock b/uv.lock index 7ccc1a0460..794399eab8 100644 --- a/uv.lock +++ b/uv.lock @@ -626,6 +626,7 @@ dependencies = [ { name = "python-telegram-bot" }, { name = "requests", extra = ["socks"] }, { name = "simplejson" }, + { name = "sqlalchemy" }, { name = "tiktoken" }, { name = "tomlkit" }, { name = "uvicorn" }, @@ -680,6 +681,7 @@ requires-dist = [ { name = "python-telegram-bot", specifier = ">=22.8" }, { name = "requests", extras = ["socks"], specifier = ">=2.34.2" }, { name = "simplejson", specifier = ">=4.1.2" }, + { name = "sqlalchemy", specifier = ">=2.1.1" }, { name = "tiktoken", specifier = ">=0.13.0" }, { name = "tomlkit", specifier = ">=0.13.3" }, { name = "torch", marker = "extra == 'voice-local'", specifier = ">=2.14.0", index = "https://download.pytorch.org/whl/cu130" }, @@ -2480,6 +2482,38 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/3c/1b/f84a2570a74094e921bbad5450b2a22a85d58585916e131d9b98029c3e69/soxr-1.0.0-cp314-cp314t-win_amd64.whl", hash = "sha256:a39b519acca2364aa726b24a6fd55acf29e4c8909102e0b858c23013c38328e5", size = 184850, upload-time = "2025-09-07T13:22:14.068Z" }, ] +[[package]] +name = "sqlalchemy" +version = "2.1.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "typing-extensions" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/a8/cb/7c68da82239ffae9378eca076afe14c310e17a4f01e3d315112f266e26c7/sqlalchemy-2.1.1.tar.gz", hash = "sha256:fcf3cbb33bb23bad75d2150157c21804618745731e9931ab27e61879a9f6123b", size = 10503203, upload-time = "2026-09-25T14:59:00.117Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/19/6e/3739dc61a0bd20db049f1b220cf380d55cc6c6e6e80a9cab382b3fe2d826/sqlalchemy-2.1.1-cp314-cp314-macosx_11_0_arm64.whl", hash = "sha256:597e1edf9cbe6ce63974813000c02c7a44da3d21f1da47f07174732ba188e8ae", size = 2455544, upload-time = "2026-09-25T15:12:50.464Z" }, + { url = "https://files.pythonhosted.org/packages/51/9b/29879ca69033f466da33486d6fb238ce1f4465e15529959e5bd3a47a8f8d/sqlalchemy-2.1.1-cp314-cp314-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:e37b1fe2e080bf47f3b693dfbec9aa39d3c3d52bfac3b4fef290a449570bcf3f", size = 4571544, upload-time = "2026-09-25T15:19:00.881Z" }, + { url = "https://files.pythonhosted.org/packages/41/0f/6db5347662a9ed83509a5236e64afc15285947ed4dd71e618912bde83445/sqlalchemy-2.1.1-cp314-cp314-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:18148691d761ef20a92cfc7dced305834e2c6ab4f4f3648f66f5d4028e234089", size = 4601803, upload-time = "2026-09-25T15:19:55.751Z" }, + { url = "https://files.pythonhosted.org/packages/77/e4/073eecd01e8a1bb5b90638c15f06b86c897518ef96d70118a70ab37bb673/sqlalchemy-2.1.1-cp314-cp314-manylinux_2_31_riscv64.manylinux_2_39_riscv64.whl", hash = "sha256:44dff425e7d88b04497b4317478a6af2a694327a087e6f1670a4266180838269", size = 4295583, upload-time = "2026-09-25T16:17:56.955Z" }, + { url = "https://files.pythonhosted.org/packages/25/71/c4e261d50297457c2496bb39b03f2a2ea0578a298defe4edc5ef514d4423/sqlalchemy-2.1.1-cp314-cp314-musllinux_1_2_aarch64.whl", hash = "sha256:6a261e2244fa88e6eded95a078550c0584da55db22d49b0b2db8c075788282fb", size = 4501686, upload-time = "2026-09-25T15:19:02.804Z" }, + { url = "https://files.pythonhosted.org/packages/7c/85/9ec520ba12027bfdbcaa3e6633e977d1984dca5739ab73bc6c9f16cb4900/sqlalchemy-2.1.1-cp314-cp314-musllinux_1_2_riscv64.whl", hash = "sha256:4789593e8979d7c946da1985a560200b6ef6c3dc6dcc5442985d62c6f2908875", size = 4296847, upload-time = "2026-09-25T16:17:59.584Z" }, + { url = "https://files.pythonhosted.org/packages/dd/65/877c7f19fe976dffbffc579148d842bb99665e520af83d28b287a162f843/sqlalchemy-2.1.1-cp314-cp314-musllinux_1_2_x86_64.whl", hash = "sha256:1477f2f1f531697753a729dd8f3e4e98f7019086f001fdc4d50350c929871c8c", size = 4560888, upload-time = "2026-09-25T15:19:57.887Z" }, + { url = "https://files.pythonhosted.org/packages/0c/f3/feb8b0803a6a2d095f4330cdc598aacef46cfb1ab6e83f4b396c6f5b89cb/sqlalchemy-2.1.1-cp314-cp314-win32.whl", hash = "sha256:1ef7bc38d140b3cf08cc6e8823b8e76da2e188b5594997e2c16586c9778af472", size = 2373288, upload-time = "2026-09-25T15:20:32.159Z" }, + { url = "https://files.pythonhosted.org/packages/be/b6/eef7d11752b5b52ad3fecfe03fabf7324e676bf1db5a22788cbcb3ba69c6/sqlalchemy-2.1.1-cp314-cp314-win_amd64.whl", hash = "sha256:73b0f7cb41a3df79dc0427377a9e2a099eb15995a327120d651f21309faa662a", size = 2423714, upload-time = "2026-09-25T15:20:34.012Z" }, + { url = "https://files.pythonhosted.org/packages/22/a0/6ce78eaac3a3197847a1d184a24633869cc343b0885f9207aa605e916bf5/sqlalchemy-2.1.1-cp314-cp314-win_arm64.whl", hash = "sha256:a0421f9eedad616b04a097ae3f155c5a0a31380364369861fd6e0f46074c3fce", size = 2388489, upload-time = "2026-09-25T15:25:20.78Z" }, + { url = "https://files.pythonhosted.org/packages/b1/56/db8f82c576dbe4f52db82e0e430debca0a6398fd0bf6a45bde11e7feb126/sqlalchemy-2.1.1-cp314-cp314t-macosx_11_0_arm64.whl", hash = "sha256:2f0bb0502de28baed425278d4bd98658d8256419fd9ad6ca964138710f933aa0", size = 2494990, upload-time = "2026-09-25T15:27:27.737Z" }, + { url = "https://files.pythonhosted.org/packages/d8/e0/cbadd9f851fa376c619f97adb4abd4b0c9d727841e988c4283dc45007c1f/sqlalchemy-2.1.1-cp314-cp314t-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:e20233c7d000c006a03f3734d8e7a6e63aa878fcd3482c1f412c864a0992e5a0", size = 4899710, upload-time = "2026-09-25T15:28:30.224Z" }, + { url = "https://files.pythonhosted.org/packages/e2/5f/8cb1a5c751161cdac50355b2c91617e2a872c237922a54b47ce014f11dd3/sqlalchemy-2.1.1-cp314-cp314t-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:8b0660ded5f9c090f9bf49fba78d52843ff3ca8c10f71624811f42f11655762d", size = 4810110, upload-time = "2026-09-25T15:49:02.468Z" }, + { url = "https://files.pythonhosted.org/packages/6b/4f/a4329f5a006b6cd2e8836b160d179c522c0cd7edfd494971a6d5a504f33e/sqlalchemy-2.1.1-cp314-cp314t-manylinux_2_31_riscv64.manylinux_2_39_riscv64.whl", hash = "sha256:63c0291bf0f721543db3719c71f2d5ad6649f12abbcbc8324d42805dac112cbc", size = 4490328, upload-time = "2026-09-25T17:35:02.971Z" }, + { url = "https://files.pythonhosted.org/packages/4c/40/7bfd8136b8038bc1ffad70f283bd51aa93e43808ea522dac5a99be7b26b3/sqlalchemy-2.1.1-cp314-cp314t-musllinux_1_2_aarch64.whl", hash = "sha256:092a743f28f3a660321e9b093144b6a5b16f08dbb9b2f52466b47a2a128475c2", size = 4757854, upload-time = "2026-09-25T15:28:32.778Z" }, + { url = "https://files.pythonhosted.org/packages/ab/4b/a25d5f0e8493fa8ad2334a7100479302cd1928f788c6200faa952b47ee27/sqlalchemy-2.1.1-cp314-cp314t-musllinux_1_2_riscv64.whl", hash = "sha256:fdef70d59afce463bedef54580ba6a832f9f2313c1d0054e15cbb4ff46f37a97", size = 4490016, upload-time = "2026-09-25T17:35:05.664Z" }, + { url = "https://files.pythonhosted.org/packages/20/74/075ae28f89ba586a2229dc746aa8b4e5b0b2595f228c96bacadf66940bf4/sqlalchemy-2.1.1-cp314-cp314t-musllinux_1_2_x86_64.whl", hash = "sha256:0f94929be35def6580cf161f20c71930d9364d62f7040b6aed02b13fd3ca09e6", size = 4748267, upload-time = "2026-09-25T15:49:05.376Z" }, + { url = "https://files.pythonhosted.org/packages/57/fd/744a9e6d2788d1d4a077ed711d06cf503762ee8a614f887e284f0869beba/sqlalchemy-2.1.1-cp314-cp314t-win32.whl", hash = "sha256:dabf6a73f2cc1fa2155b29f2a8a85afd82d8613262e6054cc3c932f5dee5d474", size = 2431659, upload-time = "2026-09-25T15:25:37.802Z" }, + { url = "https://files.pythonhosted.org/packages/c7/11/f4a73cb472cf1401ba3f47b02eed7983a0545f0347582afd469d5da2f324/sqlalchemy-2.1.1-cp314-cp314t-win_amd64.whl", hash = "sha256:3353333f8c25fc5e0661f54630238a17b0d6600305e224400cd4642cce4af34f", size = 2499109, upload-time = "2026-09-25T15:25:39.706Z" }, + { url = "https://files.pythonhosted.org/packages/ee/54/21ceb7c4ad51255be2333aa4fb32c973d35fa28b26f9895c6d66767aff4d/sqlalchemy-2.1.1-cp314-cp314t-win_arm64.whl", hash = "sha256:7f15820c84b2377bcf02cf754232690ac2ad90d425978bf1e5ba30cc04f94d55", size = 2417171, upload-time = "2026-09-25T15:20:58.386Z" }, + { url = "https://files.pythonhosted.org/packages/c5/7a/7d1879d78c086737acee5f3250cbff9d285e5bd7cb6bc2c730f24e8d400e/sqlalchemy-2.1.1-py3-none-any.whl", hash = "sha256:4357c1a222e141662251a59d3702f9b124294941171dfa6bd570513e9f4905ee", size = 2047471, upload-time = "2026-09-25T15:20:37.797Z" }, +] + [[package]] name = "starlette" version = "0.52.1"