refactor(api): separate console conversation variable queries (#43537)

Co-authored-by: LingYi-Liang <270316896+LingYi-Liang@users.noreply.github.com>

api/.importlinter

@@ -817,6 +817,33 @@ forbidden_modules =
     sqlalchemy
     werkzeug
 
+[importlinter:contract:conversation-variable-query-boundary]
+name = Conversation variable query is framework and persistence neutral
+type = forbidden
+source_modules =
+    services.conversation_variable_query
+forbidden_modules =
+    configs
+    controllers
+    extensions
+    flask
+    models
+    repositories
+    sqlalchemy
+    werkzeug
+
+[importlinter:contract:console-conversation-variable-controller-boundary]
+name = Console conversation variables reach persistence through the composition root
+type = forbidden
+source_modules =
+    controllers.console.app.conversation_variables
+forbidden_modules =
+    extensions.ext_database
+    models
+    repositories
+    sqlalchemy
+allow_indirect_imports = True
+
 [importlinter:contract:tag-application-service-boundary]
 name = Tag application service is framework and persistence neutral
 type = forbidden

api/controllers/console/app/conversation_variables.py

@@ -2,30 +2,26 @@ from __future__ import annotations
 
 from datetime import datetime
 from typing import Any
+from uuid import UUID
 
 from flask_restx import Resource
 from pydantic import BaseModel, Field, field_validator
-from sqlalchemy import select
-from sqlalchemy.orm import sessionmaker
 
 from controllers.common.rbac import PlainApp, RBACCheck
 from controllers.common.schema import query_params_from_model, register_schema_models
 from controllers.console import console_ns
-from controllers.console.app.wraps import get_app_model
+from controllers.console.app.error import AppNotFoundError
+from controllers.console.flask_admission import console_account_admission
 from controllers.console.wraps import (
     RBACPermission,
-    account_initialization_required,
-    model_validate,
-    rbac_permission_required,
-    setup_required,
+    validate_request,
 )
-from extensions.ext_database import db
+from extensions.ext_application_services import application_services
 from fields._value_type_serializer import serialize_value_type
 from fields.base import ResponseModel
 from libs.helper import dump_response, to_timestamp
-from libs.login import login_required
-from models import ConversationVariable
-from models.model import App, AppMode
+from machinery.context import RequestContext
+from services.conversation_variable_query import CONVERSATION_VARIABLE_LIMIT, ConversationVariableAppNotFoundError
 
 
 class ConversationVariablesQuery(BaseModel):
@@ -96,34 +92,24 @@ class ConversationVariablesApi(Resource):
         "Conversation variables retrieved successfully",
         console_ns.models[PaginatedConversationVariableResponse.__name__],
     )
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @rbac_permission_required(RBACCheck(RBACPermission.APP_CREATE_AND_MANAGEMENT, PlainApp()))
-    @get_app_model(mode=AppMode.ADVANCED_CHAT)
-    @model_validate(ConversationVariablesQuery)
-    def get(self, req_data: ConversationVariablesQuery, app_model: App):
-
-        stmt = (
-            select(ConversationVariable)
-            .where(ConversationVariable.app_id == app_model.id)
-            .order_by(ConversationVariable.created_at)
-        )
-        stmt = stmt.where(ConversationVariable.conversation_id == req_data.conversation_id)
-
-        # NOTE: This is a temporary solution to avoid performance issues.
-        page = 1
-        page_size = 100
-        stmt = stmt.limit(page_size).offset((page - 1) * page_size)
-
-        with sessionmaker(db.engine, expire_on_commit=False).begin() as session:
-            rows = session.scalars(stmt).all()
+    @console_account_admission(
+        rbac_checks=[RBACCheck(RBACPermission.APP_CREATE_AND_MANAGEMENT, PlainApp())],
+    )
+    def get(self, request_context: RequestContext, app_id: UUID):
+        queries = application_services().conversation_variables
+        # Preserve the legacy app-not-found/mode check before query validation.
+        try:
+            queries.require_app(request_context, str(app_id))
+        except ConversationVariableAppNotFoundError as error:
+            raise AppNotFoundError(str(error)) from error
+        req_data = validate_request(ConversationVariablesQuery)
+        rows = queries.list_variables(request_context, str(app_id), req_data.conversation_id)
 
         return dump_response(
             PaginatedConversationVariableResponse,
             {
-                "page": page,
-                "limit": page_size,
+                "page": 1,
+                "limit": CONVERSATION_VARIABLE_LIMIT,
                 "total": len(rows),
                 "has_more": False,
                 "data": [
@@ -131,7 +117,7 @@ class ConversationVariablesApi(Resource):
                         {
                             "created_at": row.created_at,
                             "updated_at": row.updated_at,
-                            **row.to_variable().model_dump(),
+                            **row.variable.model_dump(),
                             "id": row.id,
                         }
                     )

api/extensions/ext_application_services.py

@@ -61,6 +61,7 @@ from repositories.app_definition_query_repository import AppDefinitionQueryRepos
 from repositories.app_preview_query_repository import AppPreviewQueryRepository
 from repositories.app_scoped_end_user_repository import AppScopedEndUserRepo
 from repositories.app_statistic_query_repository import AppStatisticQueryRepository
+from repositories.conversation_variable_query_repository import ConversationVariableQueryRepository
 from repositories.credentials.query_repository import CredentialQueryRepository
 from repositories.explore_banner_query_repository import ExploreBannerQueryRepository
 from repositories.factory import DifyAPIRepositoryFactory
@@ -120,6 +121,7 @@ from services.app_tracing_config_service import AppTracingConfigService
 from services.billing_portal_service import BillingPortalService
 from services.billing_service import BillingService
 from services.compliance_download_service import ComplianceDownloadService
+from services.conversation_variable_query import ConversationVariableQuery
 from services.credentials.query import CredentialQuery
 from services.enterprise.enterprise_service import EnterpriseService
 from services.entities.file_grant_entities import FileGrantLimits
@@ -262,6 +264,7 @@ class ApplicationServices:
     app_preview_details: AppPreviewDetails
     app_previews: AppPreviewQueryService
     app_sites: AppSiteService
+    conversation_variables: ConversationVariableQuery
     app_statistics: AppStatisticQuery
     app_tracing_configs: AppTracingConfigService
     billing_portal: BillingPortalService
@@ -530,6 +533,7 @@ def build_application_services(
         app_sites=AppSiteService(
             sites=AppSiteCommandRepository(session_factory=database_client),
         ),
+        conversation_variables=ConversationVariableQueryRepository(session_factory=database_client),
         app_api_keys=build_app_api_key_service(database_client=database_client),
         dataset_api_keys=build_dataset_api_key_service(
             database_client=database_client,

api/repositories/conversation_variable_query_repository.py

@@ -0,0 +1,61 @@
+"""Materialized Console conversation-variable reads with owned sessions."""
+
+from typing import override
+
+from sqlalchemy import select
+from sqlalchemy.orm import Session, sessionmaker
+
+from machinery.context import RequestContext
+from models import App, AppMode, ConversationVariable
+from models.enums import AppStatus
+from repositories.app.console_repository import find_console_app
+from services.conversation_variable_query import (
+    CONVERSATION_VARIABLE_LIMIT,
+    ConversationVariableAppNotFoundError,
+    ConversationVariableQuery,
+    ConversationVariableRecord,
+)
+
+
+class ConversationVariableQueryRepository(ConversationVariableQuery):
+    def __init__(self, *, session_factory: sessionmaker[Session]) -> None:
+        self._session_factory = session_factory
+
+    @override
+    def require_app(self, context: RequestContext, app_id: str) -> None:
+        with self._session_factory() as session:
+            app = find_console_app(session, workspace_id=context.active_workspace_id, app_id=app_id)
+            if app is None:
+                raise ConversationVariableAppNotFoundError("App not found.")
+            if AppMode.value_of(app.mode) != AppMode.ADVANCED_CHAT:
+                modes = {AppMode.ADVANCED_CHAT.value}
+                raise ConversationVariableAppNotFoundError(f"App mode is not in the supported list: {modes}")
+
+    @override
+    def list_variables(
+        self, context: RequestContext, app_id: str, conversation_id: str
+    ) -> list[ConversationVariableRecord]:
+        with self._session_factory() as session:
+            rows = session.scalars(
+                select(ConversationVariable)
+                .join(App, App.id == ConversationVariable.app_id)
+                .where(
+                    App.id == app_id,
+                    App.tenant_id == context.active_workspace_id,
+                    App.status == AppStatus.NORMAL,
+                    App.mode == AppMode.ADVANCED_CHAT,
+                    ConversationVariable.conversation_id == conversation_id,
+                )
+                .order_by(ConversationVariable.created_at)
+                .limit(CONVERSATION_VARIABLE_LIMIT)
+            )
+            # Decode inside the owned session; neither rows nor lazy loads escape.
+            return [
+                ConversationVariableRecord(
+                    id=row.id,
+                    variable=row.to_variable(),
+                    created_at=row.created_at,
+                    updated_at=row.updated_at,
+                )
+                for row in rows
+            ]

api/services/conversation_variable_query.py

@@ -0,0 +1,37 @@
+"""Application query boundary for Console conversation variables."""
+
+from collections.abc import Sequence
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Protocol
+
+from graphon.variables import VariableBase
+from machinery.context import RequestContext
+
+# The Console endpoint exposes only the first page, including its legacy
+# total/has_more fields. Changing pagination is a separate API change.
+CONVERSATION_VARIABLE_LIMIT = 100
+
+
+@dataclass(frozen=True, slots=True)
+class ConversationVariableRecord:
+    id: str
+    variable: VariableBase
+    created_at: datetime
+    updated_at: datetime
+
+
+class ConversationVariableAppNotFoundError(Exception):
+    """The app is unavailable to this workspace or has an unsupported mode."""
+
+
+class ConversationVariableQuery(Protocol):
+    def require_app(self, context: RequestContext, app_id: str) -> None:
+        """Check app access before the controller validates query parameters."""
+        ...
+
+    def list_variables(
+        self, context: RequestContext, app_id: str, conversation_id: str
+    ) -> Sequence[ConversationVariableRecord]:
+        """Read the bounded result under the same workspace and app scope."""
+        ...

api/tests/unit_tests/controllers/console/app/test_conversation_variables_api.py

@@ -1,123 +1,245 @@
-from __future__ import annotations
+"""Console conversation-variable behavior through the decorated resource and real SQL."""
 
-from datetime import UTC, datetime
-from inspect import unwrap
-from unittest.mock import PropertyMock, patch
+from collections.abc import Callable
+from datetime import datetime, timedelta
+from typing import TypedDict, cast
+from unittest.mock import create_autospec
 from uuid import UUID
 
 import pytest
 from flask import Flask
-from pydantic import ValidationError
 from sqlalchemy import Engine
-from sqlalchemy.orm import Session
+from sqlalchemy.orm import Session, sessionmaker
+from werkzeug.exceptions import Forbidden, UnprocessableEntity
 
-from controllers.console.app import conversation_variables as conversation_variables_module
+from controllers.console.app.conversation_variables import ConversationVariablesApi
+from controllers.console.app.error import AppNotFoundError
+from controllers.console.workspace.error import AccountNotInitializedError
+from controllers.console.wraps import RBACPermission
+from enums import DeploymentEdition
+from extensions.ext_application_services import build_application_services
+from extensions.ext_database import db
+from extensions.ext_redis import RedisClientWrapper
 from factories import variable_factory
-from graphon.variables.types import SegmentType
-from models import App, AppMode, ConversationVariable
-from tests.unit_tests.model_factories import make_app
+from libs import login
+from models import AppMode, ConversationVariable
+from models.account import Account, AccountStatus, TenantAccountRole
+from models.agent import Agent, AgentScope, AgentSource, AgentStatus
+from services.enterprise.rbac_service import RBACService
+from tests.unit_tests.model_factories import make_account, make_app, make_tenant
+
+APP_ID = "11111111-1111-1111-1111-111111111111"
+OTHER_APP_ID = "22222222-2222-2222-2222-222222222222"
+CONVERSATION_ID = "33333333-3333-3333-3333-333333333333"
 
 
-def _app() -> App:
-    return make_app(name="Conversation variables app", mode=AppMode.ADVANCED_CHAT)
+class VariablePage(TypedDict):
+    page: int
+    limit: int
+    total: int
+    has_more: bool
+    data: list[dict[str, str | int | None]]
 
 
-def test_get_conversation_variables_returns_paginated_response(
+@pytest.fixture(autouse=True)
+def admitted_account(
     app: Flask,
+    monkeypatch: pytest.MonkeyPatch,
+    config_overrides: Callable[..., None],
     sqlite_engine: Engine,
     sqlite_session: Session,
-) -> None:
-    api = conversation_variables_module.ConversationVariablesApi()
-    method = unwrap(api.get)
-
-    created_at = datetime(2026, 1, 1, tzinfo=UTC)
-    updated_at = datetime(2026, 1, 2, tzinfo=UTC)
-    variable = variable_factory.build_conversation_variable_from_mapping(
-        {
-            "id": "var-1",
-            "name": "my_var",
-            "value_type": SegmentType.STRING,
-            "value": "value",
-            "description": "desc",
-        }
+    sqlite_session_factory: sessionmaker[Session],
+) -> Account:
+    config_overrides(DEPLOYMENT_EDITION=DeploymentEdition.CLOUD, LOGIN_DISABLED=True, RBAC_ENABLED=False)
+    account = make_account(tenant=make_tenant(), role=TenantAccountRole.NORMAL)
+    monkeypatch.setattr(login, "_resolve_current_user", lambda: account)
+    # Legacy admission still reads through the request session. New query
+    # adapters use the separately owned session factory bound to the same DB.
+    monkeypatch.setattr(db, "session", sqlite_session)
+    monkeypatch.setattr(type(db), "engine", property(lambda _self: sqlite_engine))
+    services = build_application_services(
+        database_client=sqlite_session_factory,
+        deployment_edition=DeploymentEdition.CLOUD,
+        initialization_password="",
+        redis=create_autospec(RedisClientWrapper, instance=True),
     )
-    row = ConversationVariable.from_variable(app_id="app-1", conversation_id="conv-1", variable=variable)
-    row.created_at = created_at
-    row.updated_at = updated_at
+    monkeypatch.setitem(app.extensions, "application_services", services)
+    sqlite_session.add(make_app(app_id=APP_ID, mode=AppMode.ADVANCED_CHAT))
+    sqlite_session.commit()
+    return account
+
+
+def _get(app: Flask, *, app_id: str = APP_ID, conversation_id: str | None = CONVERSATION_ID) -> VariablePage:
+    params: dict[str, str] = {} if conversation_id is None else {"conversation_id": conversation_id}
+    with app.test_request_context(f"/console/api/apps/{app_id}/conversation-variables", query_string=params):
+        response = ConversationVariablesApi().get(app_id=UUID(app_id))
+    assert isinstance(response, dict)
+    return cast(VariablePage, response)
+
+
+def _variable(
+    name: str,
+    *,
+    app_id: str = APP_ID,
+    conversation_id: str = CONVERSATION_ID,
+    value_type: str = "string",
+    value: object = "value",
+) -> ConversationVariable:
+    variable = variable_factory.build_conversation_variable_from_mapping(
+        {"id": f"draft-{name}", "name": name, "value_type": value_type, "value": value, "description": "description"}
+    )
+    return ConversationVariable.from_variable(app_id=app_id, conversation_id=conversation_id, variable=variable)
+
+
+@pytest.mark.parametrize(
+    ("value_type", "value", "exposed_type", "exposed_value"),
+    [
+        ("string", "hello", "string", "hello"),
+        ("integer", 42, "number", "42"),
+        ("float", 1.5, "number", "1.5"),
+        ("boolean", True, "boolean", "True"),
+        ("array[string]", ["a", "b"], "array[string]", "['a', 'b']"),
+        ("object", {"key": "value"}, "object", "{'key': 'value'}"),
+        ("array[object]", [{"x": 1}], "array[object]", "[{'x': 1}]"),
+    ],
+)
+def test_variable_wire_values_and_storage_id(
+    app: Flask,
+    sqlite_session: Session,
+    value_type: str,
+    value: object,
+    exposed_type: str,
+    exposed_value: str,
+) -> None:
+    row = _variable("example", value_type=value_type, value=value)
+    row.created_at = datetime(2026, 1, 1)
+    row.updated_at = datetime(2026, 1, 2)
     sqlite_session.add(row)
     sqlite_session.commit()
-    sqlite_session.expire(row)
-    expected_created_at = int(row.created_at.timestamp())
-    expected_updated_at = int(row.updated_at.timestamp())
-    with (
-        app.test_request_context(
-            "/console/api/apps/app-1/conversation-variables",
-            method="GET",
-            query_string={"conversation_id": "conv-1"},
-        ),
-        patch.object(type(conversation_variables_module.db), "engine", new_callable=PropertyMock) as engine,
-    ):
-        engine.return_value = sqlite_engine
-        response = method(
-            api,
-            conversation_variables_module.ConversationVariablesQuery(conversation_id="conv-1"),
-            app_model=_app(),
-        )
+
+    response = _get(app)
+
+    assert response == {
+        "page": 1,
+        "limit": 100,
+        "total": 1,
+        "has_more": False,
+        "data": [
+            {
+                "id": row.id,
+                "name": "example",
+                "value_type": exposed_type,
+                "value": exposed_value,
+                "description": "description",
+                "created_at": int(row.created_at.timestamp()),
+                "updated_at": int(row.updated_at.timestamp()),
+            }
+        ],
+    }
+    assert row.id != row.to_variable().id
+
+
+def test_variables_are_scoped_by_app_and_conversation(app: Flask, sqlite_session: Session) -> None:
+    sqlite_session.add_all(
+        [
+            make_app(app_id=OTHER_APP_ID, tenant_id="tenant-2", mode=AppMode.ADVANCED_CHAT),
+            _variable("visible"),
+            _variable("other-app", app_id=OTHER_APP_ID),
+            _variable("other-conversation", conversation_id="44444444-4444-4444-4444-444444444444"),
+        ]
+    )
+    sqlite_session.commit()
+
+    assert [item["name"] for item in _get(app)["data"]] == ["visible"]
+
+
+@pytest.mark.parametrize("count", [0, 101])
+def test_first_hundred_variables_keep_existing_envelope_and_order(
+    app: Flask, sqlite_session: Session, count: int
+) -> None:
+    for index in reversed(range(count)):
+        row = _variable(f"var-{index}")
+        row.created_at = datetime(2026, 1, 1) + timedelta(seconds=index)
+        sqlite_session.add(row)
+    sqlite_session.commit()
+
+    response = _get(app)
 
     assert response["page"] == 1
     assert response["limit"] == 100
-    assert response["total"] == 1
+    assert response["total"] == min(count, 100)
     assert response["has_more"] is False
-    assert response["data"][0]["id"] == row.id
-    UUID(response["data"][0]["id"])
-    assert row.to_variable().id == "var-1"
-    assert response["data"][0]["created_at"] == expected_created_at
-    assert response["data"][0]["updated_at"] == expected_updated_at
+    assert [item["name"] for item in response["data"]] == [f"var-{index}" for index in range(min(count, 100))]
 
 
-@pytest.mark.parametrize("sqlite_session", [(ConversationVariable,)], indirect=True)
-def test_get_conversation_variables_normalizes_value_type_and_value(
-    app: Flask,
-    sqlite_engine: Engine,
-    sqlite_session: Session,
+@pytest.mark.parametrize("condition", ["missing", "other-workspace", "wrong-mode"])
+@pytest.mark.parametrize("conversation_id", [None, CONVERSATION_ID])
+def test_app_admission_precedes_query_validation(
+    app: Flask, sqlite_session: Session, condition: str, conversation_id: str | None
 ) -> None:
-    api = conversation_variables_module.ConversationVariablesApi()
-    method = unwrap(api.get)
-
-    variable = variable_factory.build_conversation_variable_from_mapping(
-        {
-            "id": "var-2",
-            "name": "my_var_2",
-            "value_type": SegmentType.INTEGER,
-            "value": 42,
-            "description": "",
-        }
-    )
-    sqlite_session.add(ConversationVariable.from_variable(app_id="app-1", conversation_id="conv-1", variable=variable))
-    sqlite_session.commit()
-    with (
-        app.test_request_context(
-            "/console/api/apps/app-1/conversation-variables",
-            method="GET",
-            query_string={"conversation_id": "conv-1"},
-        ),
-        patch.object(type(conversation_variables_module.db), "engine", new_callable=PropertyMock) as engine,
-    ):
-        engine.return_value = sqlite_engine
-        response = method(
-            api,
-            conversation_variables_module.ConversationVariablesQuery(conversation_id="conv-1"),
-            app_model=_app(),
+    if condition != "missing":
+        sqlite_session.add(
+            make_app(
+                app_id=OTHER_APP_ID,
+                tenant_id="tenant-2" if condition == "other-workspace" else "tenant-1",
+                mode=AppMode.CHAT if condition == "wrong-mode" else AppMode.ADVANCED_CHAT,
+            )
         )
+        sqlite_session.commit()
 
-    assert response["data"][0]["value_type"] == "number"
-    assert response["data"][0]["value"] == "42"
+    with pytest.raises(AppNotFoundError) as error:
+        _get(app, app_id=OTHER_APP_ID, conversation_id=conversation_id)
+    expected = (
+        "App mode is not in the supported list: {'advanced-chat'}" if condition == "wrong-mode" else "App not found."
+    )
+    assert error.value.description == expected
 
 
-def test_get_conversation_variables_requires_conversation_id(app) -> None:
-    api = conversation_variables_module.ConversationVariablesApi()
-    method = unwrap(api.get)
+@pytest.mark.parametrize("status", [AgentStatus.ACTIVE, AgentStatus.ARCHIVED])
+def test_hidden_workflow_agent_app_stays_hidden(app: Flask, sqlite_session: Session, status: AgentStatus) -> None:
+    sqlite_session.add(make_app(app_id=OTHER_APP_ID, mode=AppMode.AGENT))
+    sqlite_session.add(
+        Agent(
+            tenant_id="tenant-1",
+            name="Workflow agent",
+            scope=AgentScope.WORKFLOW_ONLY,
+            source=AgentSource.WORKFLOW,
+            status=status,
+            backing_app_id=OTHER_APP_ID,
+        )
+    )
+    sqlite_session.commit()
 
-    with app.test_request_context("/console/api/apps/app-1/conversation-variables", method="GET"):
-        with pytest.raises(ValidationError):
-            conversation_variables_module.ConversationVariablesQuery.model_validate({})
+    with pytest.raises(AppNotFoundError, match="App not found"):
+        _get(app, app_id=OTHER_APP_ID)
+
+
+def test_missing_query_is_422_only_after_app_admission(app: Flask) -> None:
+    with pytest.raises(UnprocessableEntity, match="conversation_id"):
+        _get(app, conversation_id=None)
+
+
+@pytest.mark.parametrize("allowed", [False, True])
+def test_rbac_policy_is_enforced_before_query_validation(
+    app: Flask,
+    admitted_account: Account,
+    monkeypatch: pytest.MonkeyPatch,
+    config_overrides: Callable[..., None],
+    allowed: bool,
+) -> None:
+    config_overrides(RBAC_ENABLED=True)
+    check = create_autospec(RBACService.CheckAccess.check, return_value=allowed)
+    monkeypatch.setattr(RBACService.CheckAccess, "check", check)
+
+    expected_error = UnprocessableEntity if allowed else Forbidden
+    with pytest.raises(expected_error):
+        _get(app, conversation_id=None)
+    assert check.call_count == 1
+    assert check.call_args.args == ("tenant-1", admitted_account.id)
+    assert check.call_args.kwargs["scene"] == RBACPermission.APP_CREATE_AND_MANAGEMENT
+
+
+def test_uninitialized_account_cannot_read(app: Flask, admitted_account: Account) -> None:
+    admitted_account.status = AccountStatus.UNINITIALIZED
+    with pytest.raises(AccountNotInitializedError):
+        _get(app, conversation_id=None)

api/tests/unit_tests/pyrefly.toml

@@ -41,7 +41,6 @@ project-excludes = [
   "controllers/console/app/test_app_response_models.py",
   "controllers/console/app/test_audio.py",
   "controllers/console/app/test_conversation_api.py",
-  "controllers/console/app/test_conversation_variables_api.py",
   "controllers/console/app/test_create_app_payload.py",
   "controllers/console/app/test_description_validation.py",
   "controllers/console/app/test_generator_api.py",

api/tests/unit_tests/repositories/test_conversation_variable_query_repository.py

@@ -0,0 +1,107 @@
+"""Persistence scope and session lifetime of the Console variable query."""
+
+from json import JSONDecodeError
+
+import pytest
+from sqlalchemy import Engine, event, select
+from sqlalchemy.orm import Session, sessionmaker
+
+from factories import variable_factory
+from machinery.context import RequestContext
+from models import App, AppMode, ConversationVariable
+from repositories.conversation_variable_query_repository import ConversationVariableQueryRepository
+from services.conversation_variable_query import ConversationVariableAppNotFoundError
+from tests.unit_tests.model_factories import make_app
+
+CONTEXT = RequestContext(request_id="request", trace_id=None, account_id="account-1", active_workspace_id="tenant-1")
+
+
+@pytest.fixture
+def repository(sqlite_session_factory: sessionmaker[Session]) -> ConversationVariableQueryRepository:
+    variable = variable_factory.build_conversation_variable_from_mapping(
+        {"id": "draft-variable", "name": "example", "value_type": "string", "value": "value"}
+    )
+    with sqlite_session_factory.begin() as session:
+        session.add(make_app(mode=AppMode.ADVANCED_CHAT))
+        session.add(
+            ConversationVariable.from_variable(app_id="app-1", conversation_id="conversation-1", variable=variable)
+        )
+    return ConversationVariableQueryRepository(session_factory=sqlite_session_factory)
+
+
+@pytest.mark.parametrize(
+    ("workspace_id", "app_id", "conversation_id", "expected_count"),
+    [
+        ("tenant-1", "app-1", "conversation-1", 1),
+        ("tenant-2", "app-1", "conversation-1", 0),
+        ("tenant-1", "app-2", "conversation-1", 0),
+        ("tenant-1", "app-1", "conversation-2", 0),
+    ],
+)
+def test_read_applies_owner_scope_without_relying_on_prior_admission(
+    repository: ConversationVariableQueryRepository,
+    workspace_id: str,
+    app_id: str,
+    conversation_id: str,
+    expected_count: int,
+) -> None:
+    rows = repository.list_variables(CONTEXT._replace(active_workspace_id=workspace_id), app_id, conversation_id)
+    assert len(rows) == expected_count
+
+
+def test_app_transfer_after_admission_does_not_expose_variables(
+    repository: ConversationVariableQueryRepository, sqlite_session_factory: sessionmaker[Session]
+) -> None:
+    repository.require_app(CONTEXT, "app-1")
+    with sqlite_session_factory.begin() as session:
+        app = session.get_one(App, "app-1")
+        app.tenant_id = "tenant-2"
+
+    assert repository.list_variables(CONTEXT, "app-1", "conversation-1") == []
+    with pytest.raises(ConversationVariableAppNotFoundError):
+        repository.require_app(CONTEXT, "app-1")
+    new_context = CONTEXT._replace(active_workspace_id="tenant-2")
+    repository.require_app(new_context, "app-1")
+    assert len(repository.list_variables(new_context, "app-1", "conversation-1")) == 1
+
+
+@pytest.mark.parametrize("corrupt", [False, True])
+def test_read_releases_connection_even_if_variable_decoding_fails(
+    repository: ConversationVariableQueryRepository,
+    sqlite_engine: Engine,
+    sqlite_session_factory: sessionmaker[Session],
+    corrupt: bool,
+) -> None:
+    if corrupt:
+        with sqlite_session_factory.begin() as session:
+            row = session.scalars(select(ConversationVariable)).one()
+            row.data = "not json"
+    active_connections = 0
+    checkouts = 0
+
+    def checkout(*_args: object) -> None:
+        nonlocal active_connections, checkouts
+        active_connections += 1
+        checkouts += 1
+
+    def checkin(*_args: object) -> None:
+        nonlocal active_connections
+        active_connections -= 1
+
+    event.listen(sqlite_engine, "checkout", checkout)
+    event.listen(sqlite_engine, "checkin", checkin)
+    try:
+        repository.require_app(CONTEXT, "app-1")
+        assert active_connections == 0
+        if corrupt:
+            with pytest.raises(JSONDecodeError):
+                repository.list_variables(CONTEXT, "app-1", "conversation-1")
+        else:
+            rows = repository.list_variables(CONTEXT, "app-1", "conversation-1")
+            assert rows[0].variable.value == "value"
+            assert rows[0].variable.model_dump()["name"] == "example"
+        assert checkouts == 2
+        assert active_connections == 0
+    finally:
+        event.remove(sqlite_engine, "checkout", checkout)
+        event.remove(sqlite_engine, "checkin", checkin)