From d3ac9c18dc52c7598d045670c79a720772bc84df Mon Sep 17 00:00:00 2001 From: "helen@cloud" Date: Mon, 13 Jul 2026 11:45:21 +0800 Subject: [PATCH 1/3] chore: harden runtime defaults and clarify deployment scope --- README.md | 2 + docs/compatibility.md | 4 +- docs/guide.md | 12 +- docs/maintainer-architecture.md | 9 +- pyproject.toml | 2 +- src/opencode_a2a/auth.py | 9 +- src/opencode_a2a/config.py | 6 +- src/opencode_a2a/execution/session_manager.py | 116 +++++++++++------- tests/config/test_settings.py | 8 ++ .../execution/test_session_lock_lifecycle.py | 35 ++++++ 10 files changed, 143 insertions(+), 60 deletions(-) diff --git a/README.md b/README.md index 2054fe4d..2352461b 100644 --- a/README.md +++ b/README.md @@ -159,6 +159,8 @@ This repository improves the service boundary around OpenCode, but it does not t - Provider auth and default model configuration remain on the OpenCode side; deployment-time precedence details and HOME/XDG state impact are documented in [docs/guide.md](docs/guide.md#troubleshooting-provider-auth-state). - Use `A2A_CLIENT_BEARER_TOKEN` for server-side outbound peer calls initiated by `a2a_call`. - Deployment supervision is intentionally BYO. Use `systemd`, Docker, Kubernetes, or another supervisor if you need long-running operation. +- The supported persistence profile runs one `opencode-a2a` application process against its own local SQLite database. Do not run multiple Uvicorn workers or replicas against the same database file. +- PostgreSQL and other SQLAlchemy dialects are not part of the supported deployment matrix. - For mutually untrusted tenants, run separate instance pairs with isolated users, containers, workspaces, credentials, and ports. Read before deployment: diff --git a/docs/compatibility.md b/docs/compatibility.md index d71bb354..bf628234 100644 --- a/docs/compatibility.md +++ b/docs/compatibility.md @@ -41,7 +41,7 @@ External TCK runs and local conformance experiments are investigation inputs. Th ## Compatibility-Sensitive Surface -This repository still ships as an alpha project. Within that alpha line, these declared surfaces should not drift silently: +This repository ships as a beta project. Its documented public contracts are intended to remain stable, while implementation details and explicitly provider-private surfaces may continue to evolve. These declared surfaces should not drift silently: - core A2A send / stream / task methods - v1-only request/response payloads and enum values @@ -88,7 +88,7 @@ Task-store behavior that should remain stable for clients: - accepted output-mode negotiation for a task is persisted with the task so later reads keep the same filtered output contract - adapter-managed migrations only own adapter state tables; SDK-managed task schema remains SDK-owned -The default SQLite-first profile is intended for local or controlled single-instance deployments. Wider SQLAlchemy dialect compatibility should be treated as implementation latitude rather than a strong public promise unless explicitly documented later. +The supported persistence profile is one application process using its own local SQLite database. Multiple Uvicorn workers or application replicas must not share that SQLite file. PostgreSQL and other SQLAlchemy dialects are not supported deployment targets; any apparent dialect compatibility is implementation latitude rather than a public promise. ## Extension Stability diff --git a/docs/guide.md b/docs/guide.md index 2177f570..1464d375 100644 --- a/docs/guide.md +++ b/docs/guide.md @@ -47,9 +47,9 @@ Key variables to understand protocol behavior: - `A2A_INTERRUPT_REQUEST_TTL_SECONDS`: active retention window for the interrupt request binding registry used by `a2a.interrupt.*` callback methods. Default: `10800` seconds (`180` minutes). - `A2A_INTERRUPT_REQUEST_TOMBSTONE_TTL_SECONDS`: retention window for expired interrupt tombstones after active TTL has elapsed. During this window, repeated replies keep returning `INTERRUPT_REQUEST_EXPIRED` instead of falling through to `INTERRUPT_REQUEST_NOT_FOUND`. Default: `600` seconds (`10` minutes). - `A2A_CANCEL_ABORT_TIMEOUT_SECONDS`: best-effort timeout for upstream `session.abort` in cancel flow. -- `OPENCODE_TIMEOUT` / `OPENCODE_TIMEOUT_STREAM`: upstream request timeout and optional stream timeout override. -- `OPENCODE_MAX_CONCURRENT_REQUESTS`: optional fast-fail concurrency limit for unary/control upstream calls. `0` disables the limit. -- `OPENCODE_MAX_CONCURRENT_STREAMS`: optional fast-fail concurrency limit for long-lived upstream `/event` streams. `0` disables the limit. +- `OPENCODE_TIMEOUT` / `OPENCODE_TIMEOUT_STREAM`: upstream request timeout and stream timeout. Defaults: `120` and `900` seconds. +- `OPENCODE_MAX_CONCURRENT_REQUESTS`: fast-fail concurrency limit for unary/control upstream calls. Default: `32`; `0` disables the limit explicitly. +- `OPENCODE_MAX_CONCURRENT_STREAMS`: fast-fail concurrency limit for long-lived upstream `/event` streams. Default: `8`; `0` disables the limit explicitly. - `A2A_CLIENT_TIMEOUT_SECONDS`: outbound client timeout. Default: `30` seconds. - `A2A_CLIENT_CARD_FETCH_TIMEOUT_SECONDS`: outbound Agent Card fetch timeout. Default: `5` seconds. - `A2A_CLIENT_USE_CLIENT_PREFERENCE`: whether the outbound client prefers its own transport choices. @@ -174,11 +174,13 @@ With the default `database` backend, the unified lightweight persistence layer p - pending preferred-session claims - interrupt request bindings and tombstones -This project is SQLite-first for local single-instance deployments. The runtime configures local durability-oriented SQLite connection settings (`WAL`, `busy_timeout`, `synchronous=NORMAL`) and creates missing parent directories for file-backed database paths. +The supported persistence profile is one `opencode-a2a` application process using its own local SQLite database. Run Uvicorn with one worker and do not share the SQLite file between processes, containers, or replicas. Horizontal application scaling and PostgreSQL deployments are outside the supported deployment matrix. SQLAlchemy remains an internal implementation detail rather than a promise of dialect portability. + +The runtime configures local durability-oriented SQLite connection settings (`WAL`, `busy_timeout`, `synchronous=NORMAL`) and creates missing parent directories for file-backed database paths. The runtime automatically applies lightweight schema migrations for its custom state tables and records the applied version in `a2a_schema_version`. Schema-version writes are idempotent across concurrent first-start races, pending preferred-session claims now persist absolute `expires_at` timestamps, legacy rows without `expires_at` are pruned during migration instead of being reconstructed from historical TTL assumptions, and the built-in path currently targets the local SQLite deployment profile without requiring Alembic. -Database-backed task persistence also keeps the existing first-terminal-state-wins contract while tightening the SQLite path with an atomic terminal-write guard instead of relying only on process-local read-before-write checks. Any wider SQLAlchemy dialect compatibility should be treated as incidental implementation latitude rather than a documented deployment target. +Database-backed task persistence also keeps the existing first-terminal-state-wins contract while tightening the SQLite path with an atomic terminal-write guard instead of relying only on process-local read-before-write checks. At startup, the runtime logs a concise persistence summary covering the active backend, the redacted database URL when applicable, the shared persistence scope, and whether the SQLite local durability profile is active. diff --git a/docs/maintainer-architecture.md b/docs/maintainer-architecture.md index 835fc119..070583e4 100644 --- a/docs/maintainer-architecture.md +++ b/docs/maintainer-architecture.md @@ -21,12 +21,11 @@ flowchart TD subgraph Upstream["OpenCode Upstream Layer"] UpstreamClient["opencode_upstream_client.py"] - Invocation["invocation.py"] end subgraph Extensions["Extension / Contract Layer"] Jsonrpc["jsonrpc/"] - Contracts["contracts/extensions.py"] + Contracts["contracts/extensions/"] Profile["profile/runtime.py"] end @@ -61,7 +60,7 @@ flowchart TD 1. `jsonrpc/application.py` owns the adapter-specific JSON-RPC application boundary. 2. `jsonrpc/dispatch.py` and handler modules under `jsonrpc/handlers/` route provider-private methods. -3. `contracts/extensions.py` remains the SSOT for extension metadata exposed through Agent Card and OpenAPI. +3. `contracts/extensions/` remains the SSOT for extension metadata exposed through Agent Card and OpenAPI. 4. Tests under `tests/contracts/` and `tests/jsonrpc/` guard contract drift. ### Outbound Peer Call Path @@ -89,7 +88,7 @@ flowchart TD ### Extension and Contract Layer -- `contracts/extensions.py`: SSOT for extension metadata, compatibility profile, and wire-contract payloads +- `contracts/extensions/`: SSOT for extension metadata, compatibility profile, and wire-contract payloads - `jsonrpc/`: provider-private JSON-RPC extension surface - `profile/runtime.py`: runtime profile that feeds Agent Card, OpenAPI, and compatibility metadata - `protocol_versions.py`: protocol normalization and negotiation helpers @@ -128,5 +127,5 @@ For maintainers new to the codebase, this order usually gives the fastest payoff 2. `docs/architecture.md` 3. `src/opencode_a2a/server/application.py` 4. `src/opencode_a2a/execution/executor.py` -5. `src/opencode_a2a/contracts/extensions.py` +5. `src/opencode_a2a/contracts/extensions/` 6. `docs/guide.md` diff --git a/pyproject.toml b/pyproject.toml index 382b3b45..76704859 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -14,7 +14,7 @@ authors = [ ] keywords = ["a2a", "opencode", "fastapi", "json-rpc", "sse"] classifiers = [ - "Development Status :: 3 - Alpha", + "Development Status :: 4 - Beta", "Intended Audience :: Developers", "Programming Language :: Python :: 3", "Programming Language :: Python :: 3.11", diff --git a/src/opencode_a2a/auth.py b/src/opencode_a2a/auth.py index 1107cf43..a869e522 100644 --- a/src/opencode_a2a/auth.py +++ b/src/opencode_a2a/auth.py @@ -2,6 +2,7 @@ import base64 import binascii +import secrets from dataclasses import dataclass from starlette.requests import Request @@ -83,7 +84,7 @@ def authenticate_static_credential( for credential in credentials: if credential.auth_scheme != "bearer" or credential.token is None: continue - if auth_value == credential.token: + if secrets.compare_digest(auth_value, credential.token): return AuthenticatedPrincipal( identity=credential.principal, auth_scheme="bearer", @@ -102,7 +103,11 @@ def authenticate_static_credential( for credential in credentials: if credential.auth_scheme != "basic": continue - if credential.username == username and credential.password == password: + if credential.username is None or credential.password is None: + continue + username_matches = secrets.compare_digest(username, credential.username) + password_matches = secrets.compare_digest(password, credential.password) + if username_matches and password_matches: return AuthenticatedPrincipal( identity=credential.principal, auth_scheme="basic", diff --git a/src/opencode_a2a/config.py b/src/opencode_a2a/config.py index ddcf6640..a867f235 100644 --- a/src/opencode_a2a/config.py +++ b/src/opencode_a2a/config.py @@ -154,14 +154,14 @@ class Settings(BaseSettings): opencode_system: str | None = Field(default=None, alias="OPENCODE_SYSTEM") opencode_variant: str | None = Field(default=None, alias="OPENCODE_VARIANT") opencode_timeout: float = Field(default=120.0, alias="OPENCODE_TIMEOUT") - opencode_timeout_stream: float | None = Field(default=None, alias="OPENCODE_TIMEOUT_STREAM") + opencode_timeout_stream: float | None = Field(default=900.0, alias="OPENCODE_TIMEOUT_STREAM") opencode_max_concurrent_requests: int = Field( - default=0, + default=32, ge=0, alias="OPENCODE_MAX_CONCURRENT_REQUESTS", ) opencode_max_concurrent_streams: int = Field( - default=0, + default=8, ge=0, alias="OPENCODE_MAX_CONCURRENT_STREAMS", ) diff --git a/src/opencode_a2a/execution/session_manager.py b/src/opencode_a2a/execution/session_manager.py index c2c3d58e..25263ab8 100644 --- a/src/opencode_a2a/execution/session_manager.py +++ b/src/opencode_a2a/execution/session_manager.py @@ -2,9 +2,12 @@ import asyncio import weakref +from typing import TypeVar from ..server.state_store import MemorySessionStateRepository, SessionStateRepository +_LockKey = TypeVar("_LockKey") + class SessionManager: def __init__( @@ -22,12 +25,30 @@ def __init__( maxsize=session_cache_maxsize, pending_claim_ttl_seconds=pending_session_claim_ttl_seconds, ) - self._lock = asyncio.Lock() + self._registry_lock = asyncio.Lock() self._inflight_session_creates: dict[tuple[str, str], asyncio.Task[str]] = {} + self._context_locks: weakref.WeakValueDictionary[tuple[str, str], asyncio.Lock] = ( + weakref.WeakValueDictionary() + ) + self._ownership_locks: weakref.WeakValueDictionary[str, asyncio.Lock] = ( + weakref.WeakValueDictionary() + ) self._session_locks: weakref.WeakValueDictionary[str, asyncio.Lock] = ( weakref.WeakValueDictionary() ) + async def _get_registered_lock( + self, + registry: weakref.WeakValueDictionary[_LockKey, asyncio.Lock], + key: _LockKey, + ) -> asyncio.Lock: + async with self._registry_lock: + lock = registry.get(key) + if lock is None: + lock = asyncio.Lock() + registry[key] = lock + return lock + async def get_or_create_session( self, identity: str, @@ -44,7 +65,9 @@ async def get_or_create_session( session_id=preferred_session_id, ) if not pending_claim: - async with self._lock: + cache_key = (identity, context_id) + context_lock = await self._get_registered_lock(self._context_locks, cache_key) + async with context_lock: await self._state_repository.set_session( identity=identity, context_id=context_id, @@ -54,49 +77,55 @@ async def get_or_create_session( task: asyncio.Task[str] | None = None cache_key = (identity, context_id) - async with self._lock: + context_lock = await self._get_registered_lock(self._context_locks, cache_key) + async with context_lock: existing = await self._state_repository.get_session( identity=cache_key[0], context_id=cache_key[1], ) if existing: return existing, False - task = self._inflight_session_creates.get(cache_key) - if task is None: - create_session_kwargs: dict[str, str] = {} - if directory is not None: - create_session_kwargs["directory"] = directory - if workspace_id is not None: - create_session_kwargs["workspace_id"] = workspace_id - task = asyncio.create_task( - self._client.create_session( - title=title, - **create_session_kwargs, + async with self._registry_lock: + task = self._inflight_session_creates.get(cache_key) + if task is None: + create_session_kwargs: dict[str, str] = {} + if directory is not None: + create_session_kwargs["directory"] = directory + if workspace_id is not None: + create_session_kwargs["workspace_id"] = workspace_id + task = asyncio.create_task( + self._client.create_session( + title=title, + **create_session_kwargs, + ) ) - ) - self._inflight_session_creates[cache_key] = task + self._inflight_session_creates[cache_key] = task try: session_id = await task except Exception: - async with self._lock: + async with self._registry_lock: if self._inflight_session_creates.get(cache_key) is task: self._inflight_session_creates.pop(cache_key, None) raise - async with self._lock: - owner = await self._state_repository.get_owner(session_id=session_id) - if owner and owner != identity: - if self._inflight_session_creates.get(cache_key) is task: - self._inflight_session_creates.pop(cache_key, None) - raise PermissionError(f"Session {session_id} is not owned by you") - await self._state_repository.set_session( - identity=cache_key[0], - context_id=cache_key[1], - session_id=session_id, - ) - if not owner: - await self._state_repository.set_owner(session_id=session_id, identity=identity) + async with context_lock: + ownership_lock = await self._get_registered_lock(self._ownership_locks, session_id) + async with ownership_lock: + owner = await self._state_repository.get_owner(session_id=session_id) + if owner and owner != identity: + async with self._registry_lock: + if self._inflight_session_creates.get(cache_key) is task: + self._inflight_session_creates.pop(cache_key, None) + raise PermissionError(f"Session {session_id} is not owned by you") + await self._state_repository.set_session( + identity=cache_key[0], + context_id=cache_key[1], + session_id=session_id, + ) + if not owner: + await self._state_repository.set_owner(session_id=session_id, identity=identity) + async with self._registry_lock: if self._inflight_session_creates.get(cache_key) is task: self._inflight_session_creates.pop(cache_key, None) return session_id, False @@ -109,7 +138,9 @@ async def finalize_preferred_session_binding( session_id: str, ) -> None: await self.finalize_session_claim(identity=identity, session_id=session_id) - async with self._lock: + cache_key = (identity, context_id) + context_lock = await self._get_registered_lock(self._context_locks, cache_key) + async with context_lock: await self._state_repository.set_session( identity=identity, context_id=context_id, @@ -117,7 +148,8 @@ async def finalize_preferred_session_binding( ) async def claim_preferred_session(self, *, identity: str, session_id: str) -> bool: - async with self._lock: + ownership_lock = await self._get_registered_lock(self._ownership_locks, session_id) + async with ownership_lock: owner = await self._state_repository.get_owner(session_id=session_id) pending_owner = await self._state_repository.get_pending_claim(session_id=session_id) if owner and owner != identity: @@ -130,7 +162,8 @@ async def claim_preferred_session(self, *, identity: str, session_id: str) -> bo return True async def finalize_session_claim(self, *, identity: str, session_id: str) -> None: - async with self._lock: + ownership_lock = await self._get_registered_lock(self._ownership_locks, session_id) + async with ownership_lock: owner = await self._state_repository.get_owner(session_id=session_id) pending_owner = await self._state_repository.get_pending_claim(session_id=session_id) if owner and owner != identity: @@ -144,19 +177,15 @@ async def finalize_session_claim(self, *, identity: str, session_id: str) -> Non ) async def release_preferred_session_claim(self, *, identity: str, session_id: str) -> None: - async with self._lock: + ownership_lock = await self._get_registered_lock(self._ownership_locks, session_id) + async with ownership_lock: await self._state_repository.clear_pending_claim( session_id=session_id, identity=identity, ) async def get_session_lock(self, session_id: str) -> asyncio.Lock: - async with self._lock: - lock = self._session_locks.get(session_id) - if lock is None: - lock = asyncio.Lock() - self._session_locks[session_id] = lock - return lock + return await self._get_registered_lock(self._session_locks, session_id) async def pop_cached_session( self, @@ -164,6 +193,9 @@ async def pop_cached_session( identity: str, context_id: str, ) -> asyncio.Task[str] | None: - async with self._lock: + cache_key = (identity, context_id) + context_lock = await self._get_registered_lock(self._context_locks, cache_key) + async with context_lock: await self._state_repository.pop_session(identity=identity, context_id=context_id) - return self._inflight_session_creates.pop((identity, context_id), None) + async with self._registry_lock: + return self._inflight_session_creates.pop(cache_key, None) diff --git a/tests/config/test_settings.py b/tests/config/test_settings.py index c34f2339..397e23ee 100644 --- a/tests/config/test_settings.py +++ b/tests/config/test_settings.py @@ -23,6 +23,14 @@ def test_settings_missing_required(): ) +def test_settings_use_bounded_upstream_defaults() -> None: + settings = make_settings() + + assert settings.opencode_timeout_stream == 900.0 + assert settings.opencode_max_concurrent_requests == 32 + assert settings.opencode_max_concurrent_streams == 8 + + def test_settings_valid(): env = { "A2A_STATIC_AUTH_CREDENTIALS": json.dumps( diff --git a/tests/execution/test_session_lock_lifecycle.py b/tests/execution/test_session_lock_lifecycle.py index bad45ce3..6c77fe93 100644 --- a/tests/execution/test_session_lock_lifecycle.py +++ b/tests/execution/test_session_lock_lifecycle.py @@ -1,3 +1,4 @@ +import asyncio import gc import weakref from unittest.mock import AsyncMock @@ -6,6 +7,7 @@ from opencode_a2a.execution.session_manager import SessionManager from opencode_a2a.opencode_upstream_client import OpencodeUpstreamClient +from opencode_a2a.server.state_store import SessionStateRepository @pytest.mark.asyncio @@ -31,3 +33,36 @@ async def test_session_manager_does_not_strongly_retain_idle_locks() -> None: assert lock_ref() is None assert "session-1" not in manager._session_locks + + +@pytest.mark.asyncio +async def test_session_manager_does_not_serialize_repository_reads_across_contexts() -> None: + repository = AsyncMock(spec=SessionStateRepository) + both_reads_started = asyncio.Event() + release_reads = asyncio.Event() + reads_started = 0 + + async def get_session(*, identity: str, context_id: str) -> None: + nonlocal reads_started + del identity, context_id + reads_started += 1 + if reads_started == 2: + both_reads_started.set() + await release_reads.wait() + return None + + repository.get_session.side_effect = get_session + repository.get_owner.return_value = None + client = AsyncMock(spec=OpencodeUpstreamClient) + client.create_session.side_effect = ["session-1", "session-2"] + manager = SessionManager(client=client, state_repository=repository) + + requests = [ + asyncio.create_task(manager.get_or_create_session("user", "context-1", "first")), + asyncio.create_task(manager.get_or_create_session("user", "context-2", "second")), + ] + await asyncio.wait_for(both_reads_started.wait(), timeout=1.0) + release_reads.set() + await asyncio.gather(*requests) + + assert reads_started == 2 From 3304010ddf0a2eca81cd80e52d42a6cc9f48e454 Mon Sep 17 00:00:00 2001 From: "helen@cloud" Date: Mon, 13 Jul 2026 11:49:29 +0800 Subject: [PATCH 2/3] refactor: restrict persistence to SQLite --- src/opencode_a2a/config.py | 6 +++++ src/opencode_a2a/server/task_store.py | 12 --------- .../server/task_store_sdk_compat.py | 22 +--------------- tests/config/test_settings.py | 7 +++++ tests/conftest.py | 22 +++++----------- tests/server/test_state_store.py | 26 ------------------- tests/server/test_task_store_factory.py | 24 ----------------- 7 files changed, 20 insertions(+), 99 deletions(-) diff --git a/src/opencode_a2a/config.py b/src/opencode_a2a/config.py index a867f235..d677c199 100644 --- a/src/opencode_a2a/config.py +++ b/src/opencode_a2a/config.py @@ -292,6 +292,12 @@ def _validate_sandbox_policy(self) -> Settings: raise ValueError( "A2A_TASK_STORE_DATABASE_URL is required when A2A_TASK_STORE_BACKEND=database" ) + if ( + self.a2a_task_store_backend == "database" + and self.a2a_task_store_database_url + and not self.a2a_task_store_database_url.startswith("sqlite+aiosqlite:") + ): + raise ValueError("A2A_TASK_STORE_DATABASE_URL must use the sqlite+aiosqlite scheme") if self.a2a_static_auth_credentials: if not any(credential.enabled for credential in self.a2a_static_auth_credentials): raise ValueError( diff --git a/src/opencode_a2a/server/task_store.py b/src/opencode_a2a/server/task_store.py index be72d662..e23e71ed 100644 --- a/src/opencode_a2a/server/task_store.py +++ b/src/opencode_a2a/server/task_store.py @@ -173,7 +173,6 @@ def __init__( super().__init__(inner) self._write_policy = write_policy or FirstTerminalStateWinsPolicy() self._save_lock = asyncio.Lock() - self._atomic_guard_fallback_logged = False async def save( self, @@ -211,17 +210,6 @@ async def _save_database_task( context: ServerCallContext, ) -> None: compat = DatabaseTaskStoreCompat(task_store) - if not compat.supports_atomic_terminal_guard(): - if not self._atomic_guard_fallback_logged: - logger.warning( - "Database-backed task store dialect does not support atomic terminal guard; " - "falling back to read-before-write policy dialect=%s", - compat.dialect_name, - ) - self._atomic_guard_fallback_logged = True - await self._save_with_read_before_write(task, context) - return - try: if await compat.atomic_save(task, context): return diff --git a/src/opencode_a2a/server/task_store_sdk_compat.py b/src/opencode_a2a/server/task_store_sdk_compat.py index 6f2ca879..ab64f436 100644 --- a/src/opencode_a2a/server/task_store_sdk_compat.py +++ b/src/opencode_a2a/server/task_store_sdk_compat.py @@ -10,13 +10,11 @@ from a2a.types import Task, TaskState from google.protobuf.json_format import MessageToDict, ParseDict from sqlalchemy import inspect, or_, select -from sqlalchemy.dialects.postgresql import insert as postgresql_insert from sqlalchemy.dialects.sqlite import insert as sqlite_insert from ..task_states import TERMINAL_TASK_STATES from .database import redact_database_url_for_logs -_ATOMIC_TERMINAL_GUARD_DIALECTS = frozenset({"postgresql", "sqlite"}) _TERMINAL_TASK_STATE_VALUES = tuple(TaskState.Name(int(state)) for state in TERMINAL_TASK_STATES) _REQUIRED_TASK_MODEL_COLUMNS = frozenset( { @@ -56,13 +54,6 @@ def __init__(self, task_store: DatabaseTaskStore) -> None: self._task_store = task_store self._shape = _resolve_database_task_store_shape(task_store) - @property - def dialect_name(self) -> str: - return self._task_store.engine.dialect.name - - def supports_atomic_terminal_guard(self) -> bool: - return self.dialect_name in _ATOMIC_TERMINAL_GUARD_DIALECTS - async def initialize(self) -> None: await self._task_store.initialize() @@ -91,7 +82,6 @@ async def atomic_save( task=task, owner=self._shape.owner_resolver(context), task_table=self._shape.task_model.__table__, - dialect_name=self.dialect_name, ) async with self._shape.session_maker.begin() as session: result = await session.execute(statement) @@ -165,9 +155,7 @@ def _build_atomic_task_save_statement( task: Task, owner: str | None, task_table: Any, - dialect_name: str, ): - insert = _resolve_atomic_insert_factory(dialect_name) values = _task_row_values(task, owner=owner) status_state = task_table.c.status["state"].as_string() persist_guard = or_( @@ -176,7 +164,7 @@ def _build_atomic_task_save_statement( status_state.not_in(_TERMINAL_TASK_STATE_VALUES), ) return ( - insert(task_table) + sqlite_insert(task_table) .values(**values) .on_conflict_do_update( index_elements=[task_table.c.id], @@ -187,14 +175,6 @@ def _build_atomic_task_save_statement( ) -def _resolve_atomic_insert_factory(dialect_name: str): - if dialect_name == "sqlite": - return sqlite_insert - if dialect_name == "postgresql": - return postgresql_insert - raise ValueError(f"Unsupported atomic task persistence dialect: {dialect_name}") - - def _task_row_values(task: Task, *, owner: str | None) -> dict[str, Any]: return { "id": task.id, diff --git a/tests/config/test_settings.py b/tests/config/test_settings.py index 397e23ee..eea9a5e6 100644 --- a/tests/config/test_settings.py +++ b/tests/config/test_settings.py @@ -150,6 +150,13 @@ def test_settings_allow_explicit_memory_backend() -> None: assert settings.a2a_task_store_backend == "memory" +def test_settings_reject_non_sqlite_database_url() -> None: + with pytest.raises(ValidationError) as excinfo: + make_settings(a2a_task_store_database_url="postgresql+asyncpg://db.example.com/app") + + assert "A2A_TASK_STORE_DATABASE_URL must use the sqlite+aiosqlite scheme" in str(excinfo.value) + + def test_settings_reject_legacy_runtime_auth_envs() -> None: env = { "A2A_BEARER_TOKEN": "test-token", diff --git a/tests/conftest.py b/tests/conftest.py index 423b3894..eb13b142 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,14 +1,14 @@ from __future__ import annotations -import asyncio -from collections.abc import Generator +from collections.abc import AsyncGenerator import pytest +import pytest_asyncio from sqlalchemy.ext.asyncio import AsyncEngine -@pytest.fixture(autouse=True) -def dispose_app_database_engines(monkeypatch: pytest.MonkeyPatch) -> Generator[None]: +@pytest_asyncio.fixture(autouse=True) +async def dispose_app_database_engines(monkeypatch: pytest.MonkeyPatch) -> AsyncGenerator[None]: import opencode_a2a.server.application as app_module tracked_engines: dict[int, AsyncEngine] = {} @@ -22,15 +22,5 @@ def _build_database_engine(settings): # noqa: ANN001 monkeypatch.setattr(app_module, "build_database_engine", _build_database_engine) yield - if not tracked_engines: - return - - async def _dispose_tracked_engines() -> None: - for engine in tracked_engines.values(): - await engine.dispose() - - loop = asyncio.new_event_loop() - try: - loop.run_until_complete(_dispose_tracked_engines()) - finally: - loop.close() + for engine in tracked_engines.values(): + await engine.dispose() diff --git a/tests/server/test_state_store.py b/tests/server/test_state_store.py index 4071abfc..de8fe27f 100644 --- a/tests/server/test_state_store.py +++ b/tests/server/test_state_store.py @@ -5,7 +5,6 @@ import pytest from sqlalchemy import inspect, text -from sqlalchemy.dialects.postgresql import dialect as postgresql_dialect from sqlalchemy.exc import IntegrityError import opencode_a2a.server.migrations as migrations_module @@ -66,31 +65,6 @@ async def _read_pending_claim_row(engine, session_id: str) -> dict[str, object] return None if row is None else dict(row) -def test_add_missing_nullable_column_supports_non_sqlite_dialects(monkeypatch) -> None: - executed: list[str] = [] - - class _FakeInspector: - def get_columns(self, _table_name: str) -> list[dict[str, str]]: - return [] - - class _FakeConnection: - def __init__(self) -> None: - self.dialect = postgresql_dialect() - - def execute(self, clause) -> None: # noqa: ANN001 - executed.append(str(clause)) - - monkeypatch.setattr(migrations_module, "inspect", lambda _connection: _FakeInspector()) - - migrations_module._add_missing_nullable_column( - _FakeConnection(), - table=_INTERRUPT_REQUESTS, - column_name="credential_id", - ) - - assert executed == ["ALTER TABLE a2a_interrupt_requests ADD COLUMN credential_id VARCHAR"] - - def test_write_schema_version_recovers_from_concurrent_first_insert_race() -> None: executed: list[str] = [] diff --git a/tests/server/test_task_store_factory.py b/tests/server/test_task_store_factory.py index c27f90aa..1922d515 100644 --- a/tests/server/test_task_store_factory.py +++ b/tests/server/test_task_store_factory.py @@ -4,7 +4,6 @@ import warnings from pathlib import Path from unittest.mock import AsyncMock, MagicMock -from urllib.parse import parse_qsl, urlsplit import pytest from a2a.server.context import ServerCallContext @@ -173,29 +172,6 @@ def test_redact_database_url_for_logs_masks_sensitive_query_values() -> None: ) -def test_describe_lightweight_persistence_backend_redacts_sensitive_query_values() -> None: - settings = make_settings( - test_bearer_token="test-token", - a2a_task_store_database_url=( - "postgresql+asyncpg://db.example.com/app" - "?sslmode=require&token=super-secret&api_key=top-secret&pool_size=5" - ), - ) - - summary = describe_lightweight_persistence_backend(settings) - - assert summary["backend"] == "database" - assert summary["scope"] == "sdk_tasks_and_adapter_state" - assert summary["sqlite_tuning"] == "not_applicable" - assert summary["database_url"].startswith("postgresql+asyncpg://db.example.com/app?") - assert dict(parse_qsl(urlsplit(summary["database_url"]).query, keep_blank_values=True)) == { - "sslmode": "require", - "token": "***", - "api_key": "***", - "pool_size": "5", - } - - def test_describe_lightweight_persistence_backend_supports_memory_backend() -> None: settings = make_settings( test_bearer_token="test-token", From b27ef21e736e77b972260bc5280e32ccd1e63bba Mon Sep 17 00:00:00 2001 From: "helen@cloud" Date: Mon, 13 Jul 2026 15:14:16 +0800 Subject: [PATCH 3/3] refactor: remove unreachable dialect branches --- src/opencode_a2a/server/database.py | 23 +++++++++-------------- src/opencode_a2a/server/task_store.py | 4 +--- tests/server/test_task_store_factory.py | 4 ++-- 3 files changed, 12 insertions(+), 19 deletions(-) diff --git a/src/opencode_a2a/server/database.py b/src/opencode_a2a/server/database.py index ad060cdf..f9282a2a 100644 --- a/src/opencode_a2a/server/database.py +++ b/src/opencode_a2a/server/database.py @@ -45,18 +45,13 @@ def redact_database_url_for_logs(database_url: str) -> str: def build_database_engine(settings: Settings) -> AsyncEngine: database_url = cast(str, settings.a2a_task_store_database_url) url = make_url(database_url) - if url.drivername.startswith("sqlite"): - database_path = url.database - if database_path and database_path != ":memory:" and not database_path.startswith("file:"): - path = Path(database_path) - if not path.is_absolute(): - path = (Path.cwd() / path).resolve() - path.parent.mkdir(parents=True, exist_ok=True) - - engine = create_async_engine( - database_url, - pool_pre_ping=not url.drivername.startswith("sqlite"), - ) - if url.drivername.startswith("sqlite"): - event.listen(engine.sync_engine, "connect", _configure_sqlite_connection) + database_path = url.database + if database_path and database_path != ":memory:" and not database_path.startswith("file:"): + path = Path(database_path) + if not path.is_absolute(): + path = (Path.cwd() / path).resolve() + path.parent.mkdir(parents=True, exist_ok=True) + + engine = create_async_engine(database_url) + event.listen(engine.sync_engine, "connect", _configure_sqlite_connection) return engine diff --git a/src/opencode_a2a/server/task_store.py b/src/opencode_a2a/server/task_store.py index e23e71ed..612ffb2d 100644 --- a/src/opencode_a2a/server/task_store.py +++ b/src/opencode_a2a/server/task_store.py @@ -329,9 +329,7 @@ def describe_lightweight_persistence_backend(settings: Settings) -> dict[str, st return summary url = make_url(cast(str, settings.a2a_task_store_database_url)) summary["database_url"] = redact_database_url_for_logs(url.render_as_string(hide_password=True)) - summary["sqlite_tuning"] = ( - "local_durability_defaults" if url.drivername.startswith("sqlite") else "not_applicable" - ) + summary["sqlite_tuning"] = "local_durability_defaults" return summary diff --git a/tests/server/test_task_store_factory.py b/tests/server/test_task_store_factory.py index 1922d515..78e0f70d 100644 --- a/tests/server/test_task_store_factory.py +++ b/tests/server/test_task_store_factory.py @@ -162,12 +162,12 @@ def test_describe_lightweight_persistence_backend_marks_sqlite_first_scope() -> def test_redact_database_url_for_logs_masks_sensitive_query_values() -> None: redacted = redact_database_url_for_logs( - "postgresql+asyncpg://user:***@db.example.com/app" + "sqlite+aiosqlite:////var/lib/opencode-a2a/opencode-a2a.db" "?sslmode=require&token=super-secret&API_KEY=top-secret&pool_size=5" ) assert redacted == ( - "postgresql+asyncpg://user:***@db.example.com/app" + "sqlite+aiosqlite:////var/lib/opencode-a2a/opencode-a2a.db" "?sslmode=require&token=%2A%2A%2A&API_KEY=%2A%2A%2A&pool_size=5" )