diff --git a/api/.importlinter b/api/.importlinter
index d74bd46d533474..510b83529f9363 100644
--- a/api/.importlinter
+++ b/api/.importlinter
@@ -164,6 +164,34 @@ forbidden_modules =
     sqlalchemy
     werkzeug
 
+[importlinter:contract:consumer-service-boundary]
+name = Account consumer use cases are framework and persistence neutral
+type = forbidden
+source_modules =
+    services.consumer_service
+forbidden_modules =
+    controllers
+    extensions
+    repositories
+    models
+    configs
+    tasks
+    flask
+    sqlalchemy
+    werkzeug
+allow_indirect_imports = True
+
+[importlinter:contract:consumer-transport-boundary]
+name = Account consumer transport uses application services
+type = forbidden
+source_modules =
+    controllers.web.me
+forbidden_modules =
+    repositories
+    models
+    sqlalchemy
+allow_indirect_imports = True
+
 [importlinter:contract:machinery-framework-boundary]
 name = API machinery is framework neutral
 type = forbidden
diff --git a/api/controllers/web/__init__.py b/api/controllers/web/__init__.py
index 1af0f9a3d1aae2..f41f7b7cc23ab1 100644
--- a/api/controllers/web/__init__.py
+++ b/api/controllers/web/__init__.py
@@ -60,6 +60,7 @@ def _register_controllers() -> tuple[ModuleType, ...]:
         human_input_file_upload,
         human_input_form,
         login,
+        me,
         message,
         migration,
         passport,
@@ -84,6 +85,7 @@ def _register_controllers() -> tuple[ModuleType, ...]:
         human_input_file_upload,
         human_input_form,
         login,
+        me,
         message,
         migration,
         passport,
@@ -109,6 +111,7 @@ def _register_controllers() -> tuple[ModuleType, ...]:
     human_input_file_upload,
     human_input_form,
     login,
+    me,
     message,
     migration,
     passport,
@@ -137,6 +140,7 @@ def _register_controllers() -> tuple[ModuleType, ...]:
     "human_input_file_upload",
     "human_input_form",
     "login",
+    "me",
     "message",
     "passport",
     "remote_files",
diff --git a/api/controllers/web/me.py b/api/controllers/web/me.py
new file mode 100644
index 00000000000000..af0fc9c6dd12d9
--- /dev/null
+++ b/api/controllers/web/me.py
@@ -0,0 +1,216 @@
+"""Account consumer APIs authenticated by the WebApp login cookie."""
+
+from collections.abc import Generator
+from contextlib import contextmanager
+from datetime import datetime
+from urllib.parse import urlsplit
+from uuid import UUID
+
+from flask import request
+from flask_restx import Resource
+from pydantic import BaseModel, ConfigDict, Field, ValidationError, field_serializer
+from werkzeug.exceptions import Forbidden, NotFound, ServiceUnavailable, Unauthorized
+
+from configs import dify_config
+from controllers.common.end_user_connections import (
+    ConnectionConflictHTTPError,
+    ConnectionNotFoundHTTPError,
+    raise_connection_error,
+)
+from controllers.common.schema import query_params_from_model, register_response_schema_models
+from controllers.web import web_ns
+from extensions.ext_application_services import application_services
+from fields.base import ResponseModel
+from libs.datetime_utils import to_utc_timestamp
+from libs.external_api import error_responses
+from libs.helper import dump_response
+from libs.token import extract_webapp_cookie_token
+from services.consumer_service import (
+    ConsumerApp,
+    ConsumerAppNotFoundError,
+    ConsumerAppsQuery,
+    ConsumerConnection,
+    ConsumerIdentity,
+)
+from services.end_user_connections.contracts import ListConnectionsInput
+from services.end_user_connections.errors import ConnectionError
+from services.web_identity_service import WebIdentityError
+from services.web_passport_service import WebPassportUnauthorizedError
+from services.webapp_access_query_service import WebAppAccessUnavailableError
+
+
+def _origin(url: str) -> tuple[str, str, int]:
+    parsed = urlsplit(url)
+    if (
+        parsed.scheme not in {"http", "https"}
+        or parsed.hostname is None
+        or parsed.username is not None
+        or parsed.password is not None
+        or parsed.query
+        or parsed.fragment
+    ):
+        raise ValueError("Invalid origin.")
+    port = parsed.port
+    return parsed.scheme, parsed.hostname, port if port is not None else (443 if parsed.scheme == "https" else 80)
+
+
+def _identity() -> ConsumerIdentity:
+    """Cookie authentication requires a trusted browser origin for account-wide mutations."""
+    origin = request.headers.get("Origin")
+    if origin is not None or request.method not in {"GET", "HEAD", "OPTIONS"}:
+        try:
+            trusted = next(
+                url
+                for url in (dify_config.APP_WEB_URL, dify_config.CONSOLE_WEB_URL, request.host_url)
+                if urlsplit(url).netloc
+            )
+            if origin is None or _origin(origin) != _origin(trusted):
+                raise ValueError("Untrusted origin.")
+        except ValueError:
+            raise Forbidden("WebApp request origin is not allowed.") from None
+    try:
+        return application_services().consumer.authenticate(extract_webapp_cookie_token(request))
+    except (WebIdentityError, WebPassportUnauthorizedError):
+        raise Unauthorized("WebApp Account login required.") from None
+
+
+@contextmanager
+def _errors() -> Generator[None]:
+    try:
+        yield
+    except ValidationError:
+        raise ValueError("Invalid consumer request.") from None
+    except ConsumerAppNotFoundError:
+        raise NotFound("App not found.") from None
+    except WebAppAccessUnavailableError:
+        raise ServiceUnavailable("App access is temporarily unavailable.") from None
+    except ConnectionError as error:
+        raise_connection_error(error)
+
+
+class ConsumerAppResponse(ConsumerApp, ResponseModel):
+    @field_serializer("last_used_at")
+    def serialize_usage(self, value: datetime | None) -> int | None:
+        return to_utc_timestamp(value) if value is not None else None
+
+
+class ConsumerAppsResponse(ResponseModel):
+    data: list[ConsumerAppResponse]
+    has_more: bool
+    next_cursor: str | None
+
+
+class ConsumerConnectionResponse(ConsumerConnection, ResponseModel):
+    pass
+
+
+class ConsumerConnectionsResponse(ResponseModel):
+    data: list[ConsumerConnectionResponse]
+    has_more: bool
+    next_cursor: str | None
+
+
+class ConsumerConnectionsQuery(ListConnectionsInput):
+    limit: int = Field(default=20, ge=1, le=100, strict=False)
+
+
+class ConsumerConnectionAppsQuery(BaseModel):
+    model_config = ConfigDict(extra="forbid", frozen=True)
+
+    cursor: UUID | None = None
+    limit: int = Field(default=20, ge=1, le=100)
+
+
+register_response_schema_models(
+    web_ns,
+    ConsumerAppResponse,
+    ConsumerAppsResponse,
+    ConsumerConnectionResponse,
+    ConsumerConnectionsResponse,
+)
+
+
+@web_ns.route("/me/apps")
+class ConsumerAppsApi(Resource):
+    @web_ns.doc(
+        description="List accessible Chatflow and Workflow Apps using the WebApp Account login cookie. "
+        "Recent use is derived from retained published execution logs. "
+        "No App Passport or workspace membership is required."
+    )
+    @web_ns.doc(params=query_params_from_model(ConsumerAppsQuery))
+    @web_ns.response(200, "Accessible personal Apps", web_ns.models[ConsumerAppsResponse.__name__])
+    def get(self):
+        identity = _identity()
+        with _errors():
+            query = ConsumerAppsQuery.model_validate(request.args.to_dict(flat=True))
+            result = application_services().consumer.list_apps(identity, query)
+        return dump_response(ConsumerAppsResponse, result)
+
+
+@web_ns.route("/me/apps/<uuid:app_id>/favorite")
+class ConsumerFavoriteApi(Resource):
+    @web_ns.response(204, "App favorited")
+    def put(self, app_id: UUID):
+        identity = _identity()
+        with _errors():
+            application_services().consumer.set_favorite(identity, str(app_id), enabled=True)
+        return "", 204
+
+    @web_ns.response(204, "App unfavorited")
+    def delete(self, app_id: UUID):
+        identity = _identity()
+        with _errors():
+            application_services().consumer.set_favorite(identity, str(app_id), enabled=False)
+        return "", 204
+
+
+@web_ns.route("/me/connections")
+class ConsumerConnectionsApi(Resource):
+    @web_ns.doc(
+        description="List personal Connections grouped by Provider, account label and Connection type. "
+        "Connections without an account label form individual groups. WebApp counts include only accessible Apps."
+    )
+    @web_ns.doc(params=query_params_from_model(ConsumerConnectionsQuery))
+    @web_ns.response(200, "Account Connections", web_ns.models[ConsumerConnectionsResponse.__name__])
+    def get(self):
+        identity = _identity()
+        with _errors():
+            query = ConsumerConnectionsQuery.model_validate(request.args.to_dict(flat=True))
+            result = application_services().consumer.list_connections(
+                identity,
+                provider=query.provider,
+                cursor=query.cursor,
+                limit=query.limit,
+            )
+        return dump_response(ConsumerConnectionsResponse, result)
+
+
+@web_ns.route("/me/connections/<uuid:connection_key>/apps")
+class ConsumerConnectionAppsApi(Resource):
+    @web_ns.doc(description="List accessible WebApps associated with a Connection group returned by /me/connections.")
+    @web_ns.doc(params=query_params_from_model(ConsumerConnectionAppsQuery))
+    @web_ns.response(200, "Associated WebApps", web_ns.models[ConsumerAppsResponse.__name__])
+    @error_responses(ConnectionNotFoundHTTPError)
+    def get(self, connection_key: UUID):
+        identity = _identity()
+        with _errors():
+            query = ConsumerConnectionAppsQuery.model_validate(request.args.to_dict(flat=True))
+            result = application_services().consumer.list_connection_apps(
+                identity, connection_key, cursor=str(query.cursor) if query.cursor else None, limit=query.limit
+            )
+        return dump_response(ConsumerAppsResponse, result)
+
+
+@web_ns.route("/me/connections/<uuid:connection_key>")
+class ConsumerConnectionApi(Resource):
+    @web_ns.response(204, "Connection group disconnected")
+    @web_ns.doc(
+        description="Disconnect all owned Connections matching a group key returned by /me/connections. "
+        "Cookie-authenticated browser mutations require the configured WebApp Origin."
+    )
+    @error_responses(ConnectionNotFoundHTTPError, ConnectionConflictHTTPError)
+    def delete(self, connection_key: UUID):
+        identity = _identity()
+        with _errors():
+            application_services().consumer.delete_connection(identity, connection_key)
+        return "", 204
diff --git a/api/extensions/application_services/consumer.py b/api/extensions/application_services/consumer.py
new file mode 100644
index 00000000000000..bf8c6af2a4e807
--- /dev/null
+++ b/api/extensions/application_services/consumer.py
@@ -0,0 +1,31 @@
+"""Compose Account consumer operations using existing App and Connection owners."""
+
+from sqlalchemy.orm import Session, sessionmaker
+
+from repositories.app.console_repository import ConsoleAppRepository
+from repositories.end_user_connection_repository import EndUserConnectionRepository
+from services.consumer_adapters import SharedConsumerAppDiscovery
+from services.consumer_service import ConsumerAppAccess, ConsumerService
+from services.end_user_connections.management import ConnectionService
+from services.web_identity_service import WebIdentityService
+from services.web_passport_service import WebPassportTokenGateway
+
+
+def build_consumer_service(
+    *,
+    database_client: sessionmaker[Session],
+    access: ConsumerAppAccess,
+    shared_apps_enabled: bool,
+    management: ConnectionService,
+    identities: WebIdentityService,
+    tokens: WebPassportTokenGateway,
+) -> ConsumerService:
+    return ConsumerService(
+        apps=ConsoleAppRepository(session_factory=database_client),
+        connections=EndUserConnectionRepository(session_factory=database_client),
+        access=access,
+        shared_apps=SharedConsumerAppDiscovery(enabled=shared_apps_enabled),
+        management=management,
+        identities=identities,
+        tokens=tokens,
+    )
diff --git a/api/extensions/ext_application_services.py b/api/extensions/ext_application_services.py
index ed90b02115b6ca..27d7b97cbe5b45 100644
--- a/api/extensions/ext_application_services.py
+++ b/api/extensions/ext_application_services.py
@@ -27,6 +27,7 @@
 from extensions.application_services.account import AccountServices, build_account_services
 from extensions.application_services.agent import AgentAppServices, build_agent_app_services
 from extensions.application_services.app import AppServices, build_app_api_key_service, build_app_services
+from extensions.application_services.consumer import build_consumer_service
 from extensions.application_services.data_sources import (
     DataSourceServices,
     build_data_source_credentials,
@@ -136,6 +137,7 @@
 from services.billing_portal_service import BillingPortalService
 from services.billing_service import BillingService
 from services.compliance_download_service import ComplianceDownloadService
+from services.consumer_service import ConsumerService
 from services.conversation_variable_query import ConversationVariableQuery
 from services.credentials.query import CredentialQuery
 from services.end_user_connection_adapters import PluginApiKeyCredentialGateway
@@ -292,6 +294,7 @@ class AppScopedEndUserServices:
 
 @dataclass(frozen=True, slots=True)
 class ApplicationServices:
+    consumer: ConsumerService
     rbac: RBACServices
     agent_apps: AgentAppServices
     advanced_prompt_templates: AdvancedPromptTemplateService
@@ -596,6 +599,7 @@ def build_application_services(
         invitation_tokens=invitation_tokens,
     )
     end_user_connections = build_end_user_connections(database_client=database_client)
+    end_user_connection_oauth = build_end_user_connection_oauth(database_client=database_client)
     end_user_connection_preflight = build_end_user_connection_preflight(
         database_client=database_client, connections=end_user_connections
     )
@@ -612,7 +616,21 @@ def build_application_services(
         now=lambda: datetime.now(UTC),
         access_token_expire_minutes=dify_config.ACCESS_TOKEN_EXPIRE_MINUTES,
     )
+    installed_apps = build_installed_app_services(
+        database_client=database_client,
+        webapp_access=webapp_access,
+        get_workspace_role=workspace_repository.get_account_role,
+        webapp_auth_enabled=webapp_auth_enabled,
+    )
     return ApplicationServices(
+        consumer=build_consumer_service(
+            database_client=database_client,
+            access=installed_apps.access,
+            shared_apps_enabled=webapp_auth_enabled,
+            management=end_user_connections,
+            identities=web_identity,
+            tokens=web_passport_tokens,
+        ),
         rbac=rbac,
         accounts=account_services,
         apps=apps,
@@ -702,12 +720,7 @@ def build_application_services(
             audit=LoggingWebAuthenticationAuditGateway(logger=logging.getLogger("controllers.web.login")),
             private_app_access_enabled=deployment_edition == DeploymentEdition.ENTERPRISE,
         ),
-        installed_apps=build_installed_app_services(
-            database_client=database_client,
-            webapp_access=webapp_access,
-            get_workspace_role=workspace_repository.get_account_role,
-            webapp_auth_enabled=webapp_auth_enabled,
-        ),
+        installed_apps=installed_apps,
         web_app_runtime=WebAppRuntimeQueryService(
             runtime=app_definition_repository,
             file_service=file_service,
@@ -844,7 +857,7 @@ def build_application_services(
         ),
         end_user_connections=end_user_connections,
         end_user_connection_preflight=end_user_connection_preflight,
-        end_user_connection_oauth=build_end_user_connection_oauth(database_client=database_client),
+        end_user_connection_oauth=end_user_connection_oauth,
         web_passport=web_passport,
         api_based_extensions=APIBasedExtensionApplicationService(
             extensions=APIBasedExtensionRepository(session_factory=database_client),
diff --git a/api/migrations/versions/2026_10_30_1600-b73e9c40a215_add_consumer_execution_indexes.py b/api/migrations/versions/2026_10_30_1600-b73e9c40a215_add_consumer_execution_indexes.py
new file mode 100644
index 00000000000000..a8937eb3eb289e
--- /dev/null
+++ b/api/migrations/versions/2026_10_30_1600-b73e9c40a215_add_consumer_execution_indexes.py
@@ -0,0 +1,27 @@
+"""Index Account App mappings and their latest published executions.
+
+Revision ID: b73e9c40a215
+Revises: a91d5e7c3b20
+Create Date: 2026-10-30 16:00:00.000000
+"""
+
+from alembic import op
+
+revision = "b73e9c40a215"
+down_revision = "a91d5e7c3b20"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_index("end_user_account_apps_idx", "end_users", ["account_id", "type", "tenant_id", "app_id", "id"])
+    op.create_index(
+        "workflow_run_end_user_recent_idx",
+        "workflow_runs",
+        ["tenant_id", "app_id", "created_by_role", "created_by", "triggered_from", "created_at"],
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("workflow_run_end_user_recent_idx", table_name="workflow_runs")
+    op.drop_index("end_user_account_apps_idx", table_name="end_users")
diff --git a/api/migrations/versions/2026_11_03_1200-4f2a90b8c317_index_consumer_connection_groups.py b/api/migrations/versions/2026_11_03_1200-4f2a90b8c317_index_consumer_connection_groups.py
new file mode 100644
index 00000000000000..2bedd655c139d5
--- /dev/null
+++ b/api/migrations/versions/2026_11_03_1200-4f2a90b8c317_index_consumer_connection_groups.py
@@ -0,0 +1,23 @@
+"""Index Account-wide Connection group lookups.
+
+Revision ID: 4f2a90b8c317
+Revises: b73e9c40a215
+Create Date: 2026-11-03 12:00:00.000000
+"""
+
+from alembic import op
+
+revision = "4f2a90b8c317"
+down_revision = "b73e9c40a215"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_index(
+        "end_user_connection_consumer_idx", "end_user_connections", ["account_id", "provider", "credential_type", "id"]
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("end_user_connection_consumer_idx", table_name="end_user_connections")
diff --git a/api/models/end_user_connection.py b/api/models/end_user_connection.py
index 358348eff996bc..acd15f2e97cf7b 100644
--- a/api/models/end_user_connection.py
+++ b/api/models/end_user_connection.py
@@ -140,6 +140,7 @@ class EndUserConnection(DefaultFieldsDCMixin, TypeBase):
         # Selection FKs prevent a valid ID from being paired with another tenant or Provider.
         sa.UniqueConstraint("id", "tenant_id", "provider", name="end_user_connection_scope_key"),
         sa.Index("end_user_connection_account_idx", "tenant_id", "account_id", "id"),
+        sa.Index("end_user_connection_consumer_idx", "account_id", "provider", "credential_type", "id"),
         sa.Index("end_user_connection_end_user_idx", "tenant_id", "end_user_id", "id"),
         sa.Index("end_user_connection_activity_idx", "last_active_at", "id"),
     )
diff --git a/api/models/model.py b/api/models/model.py
index b8e6c9bce68b02..34bcdbcc013b5a 100644
--- a/api/models/model.py
+++ b/api/models/model.py
@@ -2140,6 +2140,7 @@ class EndUser(UserMixin, TypeBase):
         sa.Index("end_user_session_id_idx", "session_id", "type"),
         sa.Index("end_user_tenant_session_id_idx", "tenant_id", "session_id", "type"),
         sa.UniqueConstraint("tenant_id", "app_id", "type", "account_id", name="end_user_app_account_key"),
+        sa.Index("end_user_account_apps_idx", "account_id", "type", "tenant_id", "app_id", "id"),
     )
 
     tenant_id: Mapped[str] = mapped_column(StringUUID, nullable=False, default=None)
diff --git a/api/models/workflow.py b/api/models/workflow.py
index a00e811999f1e1..dda9ef2d919f85 100644
--- a/api/models/workflow.py
+++ b/api/models/workflow.py
@@ -794,6 +794,15 @@ class WorkflowRun(Base):
         sa.PrimaryKeyConstraint("id", name="workflow_run_pkey"),
         sa.Index("workflow_run_triggerd_from_idx", "tenant_id", "app_id", "triggered_from"),
         sa.Index("workflow_run_created_at_id_idx", "created_at", "id"),
+        sa.Index(
+            "workflow_run_end_user_recent_idx",
+            "tenant_id",
+            "app_id",
+            "created_by_role",
+            "created_by",
+            "triggered_from",
+            "created_at",
+        ),
         sa.Index("workflow_run_task_id_idx", "task_id", postgresql_where=sa.text("task_id IS NOT NULL")),
     )
 
diff --git a/api/openapi/markdown/web-openapi.md b/api/openapi/markdown/web-openapi.md
index c389c111dc49bb..9b7e362f4eb1c3 100644
--- a/api/openapi/markdown/web-openapi.md
+++ b/api/openapi/markdown/web-openapi.md
@@ -812,6 +812,108 @@ Logout user from web application
 | 200 | Logout successful | **application/json**: [SimpleResultResponse](#simpleresultresponse)<br> |
 | default | Error | **application/json**: [ErrorResponse](#errorresponse)<br> |
 
+### [GET] /me/apps
+List accessible Chatflow and Workflow Apps using the WebApp Account login cookie. Recent use is derived from retained published execution logs. No App Passport or workspace membership is required.
+
+#### Parameters
+
+| Name | Located in | Description | Required | Schema |
+| ---- | ---------- | ----------- | -------- | ------ |
+| cursor | query |  | No | string |
+| limit | query |  | No | integer, <br>**Default:** 20 |
+| name | query |  | No | string |
+| view | query |  | No | string, <br>**Available values:** "favorites", "recent", "shared", <br>**Default:** recent |
+
+#### Responses
+
+| Code | Description | Schema |
+| ---- | ----------- | ------ |
+| 200 | Accessible personal Apps | **application/json**: [ConsumerAppsResponse](#consumerappsresponse)<br> |
+| default | Error | **application/json**: [ErrorResponse](#errorresponse)<br> |
+
+### [DELETE] /me/apps/{app_id}/favorite
+#### Parameters
+
+| Name | Located in | Description | Required | Schema |
+| ---- | ---------- | ----------- | -------- | ------ |
+| app_id | path |  | Yes | string (uuid) |
+
+#### Responses
+
+| Code | Description | Schema |
+| ---- | ----------- | ------ |
+| 204 | App unfavorited |  |
+| default | Error | **application/json**: [ErrorResponse](#errorresponse)<br> |
+
+### [PUT] /me/apps/{app_id}/favorite
+#### Parameters
+
+| Name | Located in | Description | Required | Schema |
+| ---- | ---------- | ----------- | -------- | ------ |
+| app_id | path |  | Yes | string (uuid) |
+
+#### Responses
+
+| Code | Description | Schema |
+| ---- | ----------- | ------ |
+| 204 | App favorited |  |
+| default | Error | **application/json**: [ErrorResponse](#errorresponse)<br> |
+
+### [GET] /me/connections
+List personal Connections grouped by Provider, account label and Connection type. Connections without an account label form individual groups. WebApp counts include only accessible Apps.
+
+#### Parameters
+
+| Name | Located in | Description | Required | Schema |
+| ---- | ---------- | ----------- | -------- | ------ |
+| cursor | query |  | No | string |
+| limit | query |  | No | integer, <br>**Default:** 20 |
+| provider | query |  | No | string |
+
+#### Responses
+
+| Code | Description | Schema |
+| ---- | ----------- | ------ |
+| 200 | Account Connections | **application/json**: [ConsumerConnectionsResponse](#consumerconnectionsresponse)<br> |
+| default | Error | **application/json**: [ErrorResponse](#errorresponse)<br> |
+
+### [DELETE] /me/connections/{connection_key}
+Disconnect all owned Connections matching a group key returned by /me/connections. Cookie-authenticated browser mutations require the configured WebApp Origin.
+
+#### Parameters
+
+| Name | Located in | Description | Required | Schema |
+| ---- | ---------- | ----------- | -------- | ------ |
+| connection_key | path |  | Yes | string (uuid) |
+
+#### Responses
+
+| Code | Description | Schema |
+| ---- | ----------- | ------ |
+| 204 | Connection group disconnected |  |
+| 404 | Not Found | **application/json**: [ConnectionNotFoundError](#connectionnotfounderror)<br>[ErrorResponse](#errorresponse)<br> |
+| 409 | Conflict | **application/json**: [ConnectionConflictError](#connectionconflicterror)<br>[ErrorResponse](#errorresponse)<br> |
+| default | Error | **application/json**: [ErrorResponse](#errorresponse)<br> |
+
+### [GET] /me/connections/{connection_key}/apps
+List accessible WebApps associated with a Connection group returned by /me/connections.
+
+#### Parameters
+
+| Name | Located in | Description | Required | Schema |
+| ---- | ---------- | ----------- | -------- | ------ |
+| cursor | query |  | No | string (uuid) |
+| limit | query |  | No | integer, <br>**Default:** 20 |
+| connection_key | path |  | Yes | string (uuid) |
+
+#### Responses
+
+| Code | Description | Schema |
+| ---- | ----------- | ------ |
+| 200 | Associated WebApps | **application/json**: [ConsumerAppsResponse](#consumerappsresponse)<br> |
+| 404 | Not Found | **application/json**: [ConnectionNotFoundError](#connectionnotfounderror)<br>[ErrorResponse](#errorresponse)<br> |
+| default | Error | **application/json**: [ErrorResponse](#errorresponse)<br> |
+
 ### [GET] /messages
 Retrieve paginated list of messages from a conversation in a chat application.
 
@@ -1746,6 +1848,48 @@ The connection requirements changed. Refresh and confirm again.
 | ---- | ---- | ----------- | -------- |
 | ConnectionRevisionChangedError |  | The connection requirements changed. Refresh and confirm again. |  |
 
+#### ConsumerAppResponse
+
+| Name | Type | Description | Required |
+| ---- | ---- | ----------- | -------- |
+| app_code | string |  | Yes |
+| description | string |  | Yes |
+| icon | string |  | Yes |
+| icon_background | string |  | Yes |
+| icon_type | string |  | Yes |
+| id | string |  | Yes |
+| is_favorite | boolean |  | Yes |
+| last_used_at | integer | Creation time of the Account's latest retained published execution. | Yes |
+| mode | string |  | Yes |
+| name | string |  | Yes |
+
+#### ConsumerAppsResponse
+
+| Name | Type | Description | Required |
+| ---- | ---- | ----------- | -------- |
+| data | [ [ConsumerAppResponse](#consumerappresponse) ] |  | Yes |
+| has_more | boolean |  | Yes |
+| next_cursor | string |  | Yes |
+
+#### ConsumerConnectionResponse
+
+| Name | Type | Description | Required |
+| ---- | ---- | ----------- | -------- |
+| account_label | string |  | Yes |
+| connection_type | [ConnectionCredentialType](#connectioncredentialtype) |  | Yes |
+| key | string (uuid) | Representative Connection ID for associated WebApp queries and disconnect. | Yes |
+| name | string | Name of the group member with the smallest Connection ID. | Yes |
+| provider | string |  | Yes |
+| webapps_count | integer | Number of distinct accessible WebApps associated with this group. | Yes |
+
+#### ConsumerConnectionsResponse
+
+| Name | Type | Description | Required |
+| ---- | ---- | ----------- | -------- |
+| data | [ [ConsumerConnectionResponse](#consumerconnectionresponse) ] |  | Yes |
+| has_more | boolean |  | Yes |
+| next_cursor | string |  | Yes |
+
 #### ConversationInfiniteScrollPagination
 
 | Name | Type | Description | Required |
diff --git a/api/repositories/app/console_repository.py b/api/repositories/app/console_repository.py
index 08787fde1393b1..0bcbc32dc04f1e 100644
--- a/api/repositories/app/console_repository.py
+++ b/api/repositories/app/console_repository.py
@@ -7,24 +7,29 @@
 
 from __future__ import annotations
 
+import base64
+import json
 import logging
 from collections.abc import Sequence
 from dataclasses import replace
 from datetime import datetime
 from typing import Any, cast, override
+from uuid import UUID
 
 import sqlalchemy as sa
 from pydantic import TypeAdapter
 from sqlalchemy import ColumnElement, delete, select
+from sqlalchemy.engine import Row
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session, sessionmaker
 
 from core.agent.publish_visibility import agent_has_workflow_callable_active_snapshot
+from core.app.public_runtime import published_app_filter
 from core.trigger.constants import TRIGGER_NODE_TYPES
 from libs.datetime_utils import naive_utc_now
 from libs.pagination import PaginatedResult, paginate_query
 from machinery.context import AppRequestContext, RequestContext
-from models.account import Account, Tenant
+from models.account import Account, AccountStatus, Tenant
 from models.agent import (
     APP_BACKED_AGENT_SOURCES,
     Agent,
@@ -35,10 +40,10 @@
     WorkflowAgentBindingType,
     WorkflowAgentNodeBinding,
 )
-from models.enums import AppStatus
-from models.model import App, AppMode, AppModelConfig, AppStar, IconType
+from models.enums import AppStatus, CreatorUserRole, EndUserType, WorkflowRunTriggeredFrom
+from models.model import App, AppMode, AppModelConfig, AppStar, EndUser, IconType, Site
 from models.skill import AgentSkillBinding
-from models.workflow import Workflow
+from models.workflow import Workflow, WorkflowRun
 from repositories.app.response import app_record, app_summary
 from repositories.app_definition_query_repository import load_app_parameter_config
 from repositories.tag_repository import TagRepository
@@ -50,6 +55,7 @@
 from services.app.query_service import AppDescription, DiscoveryApps, RelatedApps
 from services.app_creation_records import create_installed_app_record, create_site_record
 from services.app_tracing_config_service import AppTraceSettings
+from services.consumer_service import ConsumerApp, ConsumerAppNotFoundError, ConsumerAppsQuery, ConsumerIdentity
 from services.entities.app_entities import (
     RECENT_APP_MODES,
     AppAgentBinding,
@@ -78,6 +84,8 @@
 
 logger = logging.getLogger(__name__)
 
+type ConsumerAppRow = tuple[App, str, str | None, datetime | None, datetime | None]
+
 _app_trace_settings_adapter = TypeAdapter(AppTraceSettings)
 
 
@@ -121,6 +129,193 @@ class ConsoleAppRepository(ConsoleApps, DiscoveryApps, RelatedApps):
     def __init__(self, *, session_factory: sessionmaker[Session]) -> None:
         self._session_factory = session_factory
 
+    @staticmethod
+    def _consumer_query(identity: ConsumerIdentity, *, recent_only: bool) -> sa.Select[ConsumerAppRow]:
+        # Resolve recent use through the owner index without aggregating Run history.
+        last_run = (
+            select(WorkflowRun.created_at)
+            .where(
+                WorkflowRun.tenant_id == EndUser.tenant_id,
+                WorkflowRun.app_id == EndUser.app_id,
+                WorkflowRun.created_by_role == CreatorUserRole.END_USER,
+                WorkflowRun.created_by == EndUser.id,
+                WorkflowRun.triggered_from == WorkflowRunTriggeredFrom.APP_RUN,
+            )
+            .order_by(WorkflowRun.created_at.desc())
+            .limit(1)
+            .correlate(EndUser)
+            .scalar_subquery()
+        )
+        usage = (
+            select(EndUser.tenant_id, EndUser.app_id, last_run.label("last_used_at"))
+            .where(EndUser.account_id == identity.account_id, EndUser.type == EndUserType.BROWSER)
+            .subquery()
+        )
+        stmt = (
+            select(App, Site.code, AppStar.id, usage.c.last_used_at, AppStar.created_at)
+            .join(Site, Site.app_id == App.id)
+            .outerjoin(
+                AppStar,
+                sa.and_(
+                    AppStar.tenant_id == App.tenant_id,
+                    AppStar.app_id == App.id,
+                    AppStar.account_id == identity.account_id,
+                ),
+            )
+            .join(
+                usage,
+                sa.and_(
+                    usage.c.tenant_id == App.tenant_id,
+                    usage.c.app_id == App.id,
+                ),
+                isouter=not recent_only,
+            )
+            .where(
+                App.status == AppStatus.NORMAL,
+                App.mode.in_((AppMode.ADVANCED_CHAT, AppMode.WORKFLOW)),
+                App.enable_site.is_(True),
+                Site.status == AppStatus.NORMAL,
+                Site.code.is_not(None),
+                Site.code != "",
+                published_app_filter(),
+            )
+        )
+        # SQLAlchemy does not infer outer-join nullability or the Site code predicate.
+        return cast(sa.Select[ConsumerAppRow], stmt)
+
+    @staticmethod
+    def _consumer_app(identity: ConsumerIdentity, row: Row[ConsumerAppRow], view: str) -> ConsumerApp:
+        app, code, star_id, last_used, starred_at = row
+        timestamp = last_used if view == "recent" else starred_at if view == "favorites" else None
+        continuation = base64.urlsafe_b64encode(
+            json.dumps(
+                [
+                    identity.account_id,
+                    view,
+                    timestamp.isoformat() if timestamp is not None else None,
+                    app.id,
+                ]
+            ).encode()
+        ).decode()
+        return ConsumerApp(
+            id=app.id,
+            name=app.name,
+            description=app.description,
+            mode=app.mode,
+            icon_type=app.icon_type,
+            icon=app.icon,
+            icon_background=app.icon_background,
+            app_code=code,
+            is_favorite=star_id is not None,
+            last_used_at=last_used,
+            continuation=continuation,
+        )
+
+    def candidates(
+        self,
+        identity: ConsumerIdentity,
+        query: ConsumerAppsQuery,
+        *,
+        after: str | None,
+        limit: int,
+        shared_ids: Sequence[str] | None,
+    ) -> tuple[ConsumerApp, ...]:
+        stmt = self._consumer_query(identity, recent_only=query.view == "recent")
+        timestamp = None
+        if query.view == "recent":
+            timestamp = stmt.selected_columns.last_used_at
+            stmt = stmt.where(timestamp.is_not(None))
+        elif query.view == "favorites":
+            stmt = stmt.where(AppStar.id.is_not(None))
+            timestamp = AppStar.created_at
+        else:
+            stmt = stmt.where(App.id.in_(shared_ids or ()))
+        if query.name:
+            stmt = stmt.where(App.name.icontains(query.name, autoescape=True))
+        if after is not None:
+            try:
+                owner, view, time_value, app_id = json.loads(base64.b64decode(after, altchars=b"-_", validate=True))
+                if owner != identity.account_id or view != query.view:
+                    raise ValueError
+                app_id = str(UUID(app_id))
+                if timestamp is None:
+                    stmt = stmt.where(App.id > app_id)
+                else:
+                    moment = datetime.fromisoformat(time_value)
+                    stmt = stmt.where(sa.or_(timestamp < moment, sa.and_(timestamp == moment, App.id > app_id)))
+            except (ValueError, TypeError, UnicodeError):
+                raise ValueError("Invalid consumer App cursor.") from None
+        if timestamp is not None:
+            stmt = stmt.order_by(timestamp.desc())
+        stmt = stmt.order_by(App.id).limit(limit)
+        with self._session_factory() as session:
+            return tuple(self._consumer_app(identity, row, query.view) for row in session.execute(stmt))
+
+    def by_ids(self, identity: ConsumerIdentity, app_ids: Sequence[str]) -> tuple[ConsumerApp, ...]:
+        if not app_ids:
+            return ()
+        with self._session_factory() as session:
+            return tuple(
+                self._consumer_app(identity, row, "shared")
+                for row in session.execute(
+                    self._consumer_query(identity, recent_only=False).where(App.id.in_(app_ids)).order_by(App.id)
+                )
+            )
+
+    def favorite(self, identity: ConsumerIdentity, app_id: str, *, enabled: bool) -> None:
+        with self._session_factory.begin() as session:
+            account = session.scalar(
+                select(Account.id)
+                .where(
+                    Account.id == identity.account_id,
+                    Account.status.in_((AccountStatus.ACTIVE, AccountStatus.UNINITIALIZED)),
+                )
+                .with_for_update()
+            )
+            if account is None:
+                raise ConsumerAppNotFoundError()
+            if not enabled:
+                star = session.scalar(
+                    select(AppStar)
+                    .where(
+                        AppStar.account_id == identity.account_id,
+                        AppStar.app_id == app_id,
+                    )
+                    .limit(1)
+                )
+                if star is not None:
+                    session.execute(
+                        delete(AppStar).where(
+                            AppStar.id == star.id,
+                            AppStar.account_id == identity.account_id,
+                            AppStar.tenant_id == star.tenant_id,
+                            AppStar.app_id == app_id,
+                        )
+                    )
+                return
+            row = session.execute(self._consumer_query(identity, recent_only=False).where(App.id == app_id)).first()
+            if row is None:
+                raise ConsumerAppNotFoundError()
+            app, _, star_id, _, _ = row
+            if star_id is None:
+                try:
+                    with session.begin_nested():
+                        session.add(AppStar(tenant_id=app.tenant_id, account_id=identity.account_id, app_id=app.id))
+                        session.flush()
+                except IntegrityError:
+                    # Console writers share the unique key without taking the consumer lock.
+                    if (
+                        session.scalar(
+                            select(AppStar.id).where(
+                                AppStar.tenant_id == app.tenant_id,
+                                AppStar.account_id == identity.account_id,
+                                AppStar.app_id == app_id,
+                            )
+                        )
+                        is None
+                    ):
+                        raise
+
     def get_agent_binding(self, workspace_id: str, app_id: str) -> AppAgentBinding | None:
         with self._session_factory() as session:
             app = session.scalar(select(App).where(App.id == app_id, App.tenant_id == workspace_id))
diff --git a/api/repositories/end_user_connection_repository.py b/api/repositories/end_user_connection_repository.py
index f8dc315e20a553..1e22042bd8aad3 100644
--- a/api/repositories/end_user_connection_repository.py
+++ b/api/repositories/end_user_connection_repository.py
@@ -1,9 +1,10 @@
 """Owner-scoped Connection storage; external I/O never runs in these transactions."""
 
-from collections.abc import Mapping
+from collections.abc import Mapping, Sequence
 from datetime import datetime
 from hashlib import sha256
 from typing import Any, cast
+from uuid import UUID
 
 import sqlalchemy as sa
 from sqlalchemy.engine import CursorResult
@@ -32,11 +33,15 @@
 from models.model import App, AppMode, Conversation, EndUser, Site
 from models.provider_ids import ToolProviderID
 from models.tools import ToolOAuthSystemClient, ToolOAuthTenantClient
+from models.types import StringUUID
 from models.workflow import Workflow
+from services.consumer_service import ConsumerConnectionGroup, ConsumerConnectionKey, ConsumerIdentity
 from services.end_user_connections.claim import ConnectionClaimSource
 from services.end_user_connections.contracts import ConnectionPage, ConnectionSummary, ListConnectionsInput
 from services.end_user_connections.entities import (
+    AccountConnectionContext,
     ConnectionAccessRecord,
+    ConnectionManagementContext,
     ConnectionMutationGuard,
     ConnectionOAuthClient,
     ConnectionOwner,
@@ -46,6 +51,7 @@
     ConnectionConfigurationError,
     ConnectionConflictError,
     ConnectionIdentityError,
+    ConnectionNotFoundError,
     ConnectionReauthorizationRequiredError,
     ConnectionRevisionChangedError,
 )
@@ -69,6 +75,186 @@ class EndUserConnectionRepository:
     def __init__(self, *, session_factory: sessionmaker[Session]) -> None:
         self._sessions = session_factory
 
+    def account_connection_context(self, identity: ConsumerIdentity, connection_id: str) -> AccountConnectionContext:
+        with self._sessions() as session:
+            tenant_id = session.scalar(
+                sa.select(EndUserConnection.tenant_id).where(
+                    EndUserConnection.id == connection_id,
+                    EndUserConnection.account_id == identity.account_id,
+                    EndUserConnection.end_user_id.is_(None),
+                )
+            )
+            if tenant_id is None:
+                raise ConnectionNotFoundError()
+            context = AccountConnectionContext(tenant_id, identity.account_id, connection_id)
+            self._require_management_context(session, context, lock=False)
+            return context
+
+    def account_connection_groups(
+        self, identity: ConsumerIdentity, *, provider: str | None, after: ConsumerConnectionKey | None, limit: int
+    ) -> tuple[ConsumerConnectionGroup, ...]:
+        """Page complete Account-owned groups before loading their App associations."""
+        label = sa.func.coalesce(sa.func.nullif(EndUserConnection.account_label, ""), "").label("account_label")
+        id_text = sa.cast(EndUserConnection.id, sa.String(36))
+        connection_id = sa.case((label == "", id_text), else_="").label("connection_id")
+        columns = (EndUserConnection.provider, EndUserConnection.credential_type, label, connection_id)
+        # PostgreSQL UUIDs need text conversion for MIN and the empty-label discriminator.
+        grouped = sa.select(*columns, sa.func.min(id_text).label("representative_id")).where(
+            EndUserConnection.account_id == identity.account_id,
+            EndUserConnection.end_user_id.is_(None),
+        )
+        if provider is not None:
+            grouped = grouped.where(EndUserConnection.provider == provider)
+        if after is not None:
+            grouped = grouped.where(
+                sa.tuple_(*columns)
+                > (
+                    after.provider,
+                    after.connection_type,
+                    after.account_label or "",
+                    str(after.connection_id) if after.connection_id else "",
+                )
+            )
+        groups = grouped.group_by(*columns).order_by(*columns).limit(limit).subquery()
+        stmt = (
+            sa.select(
+                groups.c.provider,
+                groups.c.credential_type,
+                groups.c.account_label,
+                groups.c.connection_id,
+                groups.c.representative_id,
+                EndUserConnection.name,
+            )
+            .join(EndUserConnection, EndUserConnection.id == sa.cast(groups.c.representative_id, StringUUID()))
+            .order_by(groups.c.provider, groups.c.credential_type, groups.c.account_label, groups.c.connection_id)
+        )
+        with self._sessions() as session:
+            return tuple(
+                ConsumerConnectionGroup(
+                    key=ConsumerConnectionKey(
+                        provider=provider_,
+                        connection_type=type_,
+                        account_label=label_ or None,
+                        connection_id=UUID(id_) if id_ else None,
+                    ),
+                    representative_id=UUID(representative_id),
+                    name=name,
+                )
+                for provider_, type_, label_, id_, representative_id, name in session.execute(stmt)
+            )
+
+    def account_connection_group(self, identity: ConsumerIdentity, connection_id: str) -> ConsumerConnectionKey:
+        stmt = sa.select(
+            EndUserConnection.provider,
+            EndUserConnection.credential_type,
+            EndUserConnection.account_label,
+        ).where(
+            EndUserConnection.id == connection_id,
+            EndUserConnection.account_id == identity.account_id,
+            EndUserConnection.end_user_id.is_(None),
+        )
+        with self._sessions() as session:
+            row = session.execute(stmt).one_or_none()
+            if row is None:
+                raise ConnectionNotFoundError()
+            provider, type_, account_label = row
+            return ConsumerConnectionKey(
+                provider=provider,
+                connection_type=type_,
+                account_label=account_label or None,
+                connection_id=None if account_label else UUID(connection_id),
+            )
+
+    def account_group_connections(
+        self, identity: ConsumerIdentity, keys: Sequence[ConsumerConnectionKey]
+    ) -> Mapping[ConsumerConnectionKey, tuple[str, ...]]:
+        if not keys:
+            return {}
+        label = sa.func.nullif(EndUserConnection.account_label, "")
+        filters = [
+            sa.and_(
+                EndUserConnection.provider == key.provider,
+                EndUserConnection.credential_type == key.connection_type,
+                label == key.account_label
+                if key.account_label is not None
+                else sa.and_(label.is_(None), EndUserConnection.id == str(key.connection_id)),
+            )
+            for key in keys
+        ]
+        stmt = (
+            sa.select(
+                EndUserConnection.id,
+                sa.case(*[(predicate, index) for index, predicate in enumerate(filters)], else_=-1),
+            )
+            .where(
+                EndUserConnection.account_id == identity.account_id,
+                EndUserConnection.end_user_id.is_(None),
+                sa.or_(*filters),
+            )
+            .order_by(EndUserConnection.id)
+        )
+        with self._sessions() as session:
+            result: dict[ConsumerConnectionKey, list[str]] = {}
+            for id_, index in session.execute(stmt):
+                # Use the same database equality as GROUP BY, including configured collations.
+                key = keys[index]
+                result.setdefault(key, []).append(id_)
+            return {key: tuple(ids) for key, ids in result.items()}
+
+    def account_connection_apps(
+        self, identity: ConsumerIdentity, connection_ids: Sequence[str]
+    ) -> Mapping[str, tuple[str, ...]]:
+        if not connection_ids:
+            return {}
+        queries = [
+            sa.select(reference.connection_id, reference.app_id)
+            .join(
+                EndUserConnection,
+                sa.and_(
+                    EndUserConnection.id == reference.connection_id,
+                    EndUserConnection.tenant_id == reference.tenant_id,
+                ),
+            )
+            .join(App, sa.and_(App.id == reference.app_id, App.tenant_id == reference.tenant_id))
+            .where(
+                EndUserConnection.account_id == identity.account_id,
+                EndUserConnection.end_user_id.is_(None),
+                EndUserConnection.id.in_(connection_ids),
+            )
+            for reference in (AppConnection, SessionConnection)
+        ]
+        with self._sessions() as session:
+            rows = session.execute(sa.union(*queries).order_by("connection_id", "app_id"))
+            associations: dict[str, list[str]] = {}
+            for connection_id, app_id in rows:
+                associations.setdefault(connection_id, []).append(app_id)
+            return {key: tuple(ids) for key, ids in associations.items()}
+
+    @classmethod
+    def _require_management_context(
+        cls, session: Session, context: ConnectionManagementContext, *, lock: bool
+    ) -> App | None:
+        if isinstance(context, WebConnectionContext):
+            return cls._require_context(session, context, lock=lock)
+        account = sa.select(Account.id).where(
+            Account.id == context.account_id,
+            Account.status.in_((AccountStatus.ACTIVE, AccountStatus.UNINITIALIZED)),
+        )
+        if lock:
+            account = account.with_for_update()
+        if session.scalar(account) is None:
+            raise ConnectionIdentityError()
+        connection = sa.select(EndUserConnection.id).where(
+            EndUserConnection.id == context.connection_id,
+            EndUserConnection.tenant_id == context.tenant_id,
+            *cls._owner(context.owner),
+        )
+        if lock:
+            connection = connection.with_for_update()
+        if session.scalar(connection) is None:
+            raise ConnectionNotFoundError()
+        return None
+
     def save_revision(self, snapshot: ConnectionContractSnapshot) -> str:
         canonical = snapshot.canonical_json()
         fingerprint = sha256(canonical.encode()).hexdigest()
@@ -698,7 +884,7 @@ def _insert_conversation_selections(
         )
 
     @classmethod
-    def _summary_query(cls, context: WebConnectionContext) -> sa.Select[tuple[EndUserConnection]]:
+    def _summary_query(cls, context: ConnectionManagementContext) -> sa.Select[tuple[EndUserConnection]]:
         return (
             sa.select(EndUserConnection)
             .options(
@@ -743,9 +929,11 @@ def list(self, context: WebConnectionContext, query: ListConnectionsInput, after
             data = tuple(self._summary(row) for row in rows[: query.limit])
             return ConnectionPage(data=data, next_cursor=data[-1].id if len(rows) > query.limit else None)
 
-    def find(self, context: WebConnectionContext, connection_id: str) -> StoredConnection | None:
+    def find(self, context: ConnectionManagementContext, connection_id: str) -> StoredConnection | None:
+        if isinstance(context, AccountConnectionContext) and connection_id != context.connection_id:
+            return None
         with self._sessions() as session:
-            self._require_context(session, context, lock=False)
+            self._require_management_context(session, context, lock=False)
             row = session.scalar(
                 sa.select(EndUserConnection).where(
                     EndUserConnection.id == connection_id,
@@ -827,9 +1015,9 @@ def update(
                 self._record_revision_acceptance(session, context, changes.revision_acceptance)
             return self._stored(row)
 
-    def delete(self, context: WebConnectionContext, guard: ConnectionMutationGuard) -> None:
+    def delete(self, context: ConnectionManagementContext, guard: ConnectionMutationGuard) -> None:
         with self._sessions.begin() as session:
-            self._require_context(session, context, lock=True)
+            self._require_management_context(session, context, lock=True)
             result = cast(
                 CursorResult, session.execute(sa.delete(EndUserConnection).where(*self._guard(context, guard)))
             )
@@ -1064,9 +1252,14 @@ def save_oauth(
             return self._summary(row)
 
     def _guard(
-        self, context: WebConnectionContext, guard: ConnectionMutationGuard
+        self, context: ConnectionManagementContext, guard: ConnectionMutationGuard
     ) -> tuple[sa.ColumnElement[bool], ...]:
-        if guard.tenant_id != context.tenant_id or guard.owner != context.owner:
+        if (
+            guard.tenant_id != context.tenant_id
+            or guard.owner != context.owner
+            or isinstance(context, AccountConnectionContext)
+            and guard.connection_id != context.connection_id
+        ):
             raise ConnectionConflictError()
         return (
             EndUserConnection.id == guard.connection_id,
diff --git a/api/services/consumer_adapters.py b/api/services/consumer_adapters.py
new file mode 100644
index 00000000000000..c9a7c3a85a5910
--- /dev/null
+++ b/api/services/consumer_adapters.py
@@ -0,0 +1,36 @@
+"""Deployment discovery adapters for Account consumer queries."""
+
+import httpx
+
+from services.enterprise.enterprise_service import EnterpriseService
+from services.errors.enterprise import EnterpriseServiceError
+from services.webapp_access_query_service import WebAppAccessUnavailableError
+
+
+class SharedConsumerAppDiscovery:
+    def __init__(self, *, enabled: bool) -> None:
+        self._enabled = enabled
+
+    def __call__(self) -> tuple[str, ...]:
+        if not self._enabled:
+            return ()
+        app_ids: list[str] = []
+        seen: set[str] = set()
+        page = 1
+        while True:
+            try:
+                result = EnterpriseService.WebAppAuth.list_externally_accessible_apps(page=page, limit=100)
+                rows = result["data"]
+                ids = [row["appId"] for row in rows]
+                has_more = result["hasMore"]
+                if not isinstance(has_more, bool) or any(not isinstance(id_, str) for id_ in ids):
+                    raise ValueError("Malformed App discovery response")
+                if has_more and not set(ids).difference(seen):
+                    raise ValueError("App discovery did not advance")
+            except (EnterpriseServiceError, httpx.RequestError, KeyError, TypeError, ValueError) as error:
+                raise WebAppAccessUnavailableError from error
+            app_ids.extend(ids)
+            seen.update(ids)
+            if not has_more:
+                return tuple(dict.fromkeys(app_ids))
+            page += 1
diff --git a/api/services/consumer_service.py b/api/services/consumer_service.py
new file mode 100644
index 00000000000000..d5eb8e9e791d84
--- /dev/null
+++ b/api/services/consumer_service.py
@@ -0,0 +1,329 @@
+"""Account consumer queries and personal App preferences without Studio admission.
+
+Candidate discovery is not permission. Every App projection is filtered through
+the existing deployment access policy, including Apps referenced by Connections.
+Ports return detached, non-secret values and own short transactions.
+"""
+
+import base64
+import json
+from collections.abc import Mapping, Sequence
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Literal, Protocol
+from uuid import UUID
+
+from pydantic import BaseModel, ConfigDict, Field, model_validator
+
+from constants.end_user_connections import ConnectionCredentialType
+from services.end_user_connections.entities import AccountConnectionContext
+from services.end_user_connections.errors import ConnectionNotFoundError
+from services.end_user_connections.management import ConnectionService
+from services.web_identity_service import WebIdentityError, WebIdentityService
+from services.web_passport_service import WebPassportTokenGateway
+
+
+@dataclass(frozen=True, slots=True)
+class ConsumerIdentity:
+    account_id: str
+
+
+class ConsumerApp(BaseModel):
+    model_config = ConfigDict(frozen=True)
+
+    id: str
+    name: str
+    description: str
+    mode: str
+    icon_type: str | None
+    icon: str | None
+    icon_background: str | None
+    app_code: str
+    is_favorite: bool
+    last_used_at: datetime | None = Field(
+        description="Creation time of the Account's latest retained published execution."
+    )
+    continuation: str = Field(exclude=True)
+
+
+class ConsumerAppsQuery(BaseModel):
+    model_config = ConfigDict(extra="forbid", frozen=True, hide_input_in_errors=True)
+
+    view: Literal["recent", "favorites", "shared"] = "recent"
+    name: str | None = Field(default=None, max_length=255)
+    cursor: str | None = Field(default=None, max_length=2048)
+    limit: int = Field(default=20, ge=1, le=100)
+
+
+class ConsumerAppsPage(BaseModel):
+    data: tuple[ConsumerApp, ...]
+    has_more: bool
+    next_cursor: str | None
+
+
+class ConsumerConnectionKey(BaseModel):
+    """Internal group selector and pagination position, independent of the public member ID."""
+
+    model_config = ConfigDict(extra="forbid", frozen=True)
+
+    provider: str = Field(min_length=1, max_length=255)
+    connection_type: ConnectionCredentialType
+    account_label: str | None = Field(min_length=1)
+    connection_id: UUID | None
+
+    @model_validator(mode="after")
+    def require_one_group_identity(self) -> "ConsumerConnectionKey":
+        if (self.account_label is None) == (self.connection_id is None):
+            raise ValueError("A group requires either an account label or a Connection ID.")
+        return self
+
+    def encode(self) -> str:
+        payload = [
+            self.provider,
+            self.connection_type,
+            self.account_label,
+            str(self.connection_id) if self.connection_id else None,
+        ]
+        return base64.urlsafe_b64encode(json.dumps(payload, separators=(",", ":")).encode()).decode().rstrip("=")
+
+    @classmethod
+    def decode(cls, value: str) -> "ConsumerConnectionKey":
+        try:
+            provider, connection_type, account_label, connection_id = json.loads(
+                base64.b64decode(value + "=" * (-len(value) % 4), altchars=b"-_", validate=True)
+            )
+            key = cls(
+                provider=provider,
+                connection_type=connection_type,
+                account_label=account_label,
+                connection_id=connection_id,
+            )
+            if key.encode() != value:
+                raise ValueError
+            return key
+        except (ValueError, TypeError):
+            raise ValueError("Invalid Connection group key.") from None
+
+
+@dataclass(frozen=True, slots=True)
+class ConsumerConnectionGroup:
+    key: ConsumerConnectionKey
+    representative_id: UUID
+    name: str
+
+
+class ConsumerConnection(BaseModel):
+    key: UUID = Field(description="Representative Connection ID for associated WebApp queries and disconnect.")
+    provider: str
+    connection_type: ConnectionCredentialType
+    name: str = Field(description="Name of the group member with the smallest Connection ID.")
+    account_label: str | None
+    webapps_count: int = Field(ge=0, description="Number of distinct accessible WebApps associated with this group.")
+
+
+class ConsumerConnectionsPage(BaseModel):
+    data: tuple[ConsumerConnection, ...]
+    has_more: bool
+    next_cursor: str | None
+
+
+class ConsumerAppNotFoundError(LookupError):
+    """An unavailable and a foreign App have the same public result."""
+
+
+class ConsumerAppsStore(Protocol):
+    def candidates(
+        self,
+        identity: ConsumerIdentity,
+        query: ConsumerAppsQuery,
+        *,
+        after: str | None,
+        limit: int,
+        shared_ids: Sequence[str] | None,
+    ) -> tuple[ConsumerApp, ...]: ...
+
+    def by_ids(self, identity: ConsumerIdentity, app_ids: Sequence[str]) -> tuple[ConsumerApp, ...]: ...
+
+    def favorite(self, identity: ConsumerIdentity, app_id: str, *, enabled: bool) -> None:
+        """Serialize on the Account; revalidate App availability in the write transaction.
+
+        The existing tenant/Account/App unique key also prevents duplicate stars.
+        No workspace membership is required or created.
+        """
+        ...
+
+
+class ConsumerConnectionsStore(Protocol):
+    def account_connection_groups(
+        self, identity: ConsumerIdentity, *, provider: str | None, after: ConsumerConnectionKey | None, limit: int
+    ) -> tuple[ConsumerConnectionGroup, ...]: ...
+
+    def account_group_connections(
+        self, identity: ConsumerIdentity, keys: Sequence[ConsumerConnectionKey]
+    ) -> Mapping[ConsumerConnectionKey, tuple[str, ...]]: ...
+
+    def account_connection_group(self, identity: ConsumerIdentity, connection_id: str) -> ConsumerConnectionKey:
+        """Resolve an owned member ID into its current group without loading credentials."""
+        ...
+
+    def account_connection_context(
+        self, identity: ConsumerIdentity, connection_id: str
+    ) -> AccountConnectionContext: ...
+
+    def account_connection_apps(
+        self, identity: ConsumerIdentity, connection_ids: Sequence[str]
+    ) -> Mapping[str, tuple[str, ...]]:
+        """Current App defaults and Session references, deduplicated by App.
+
+        References do not grant ownership. Filter Connections by Account and join
+        each reference on tenant as well as ID. Never read historical Run snapshots.
+        """
+        ...
+
+
+class ConsumerAppAccess(Protocol):
+    def get_visible_app_ids(self, *, user_id: str, app_ids: Sequence[str]) -> frozenset[str]: ...
+
+
+class SharedConsumerApps(Protocol):
+    def __call__(self) -> tuple[str, ...]: ...
+
+
+class ConsumerService:
+    def __init__(
+        self,
+        *,
+        apps: ConsumerAppsStore,
+        connections: ConsumerConnectionsStore,
+        access: ConsumerAppAccess,
+        shared_apps: SharedConsumerApps,
+        management: ConnectionService,
+        identities: WebIdentityService,
+        tokens: WebPassportTokenGateway,
+    ) -> None:
+        self._apps = apps
+        self._connections = connections
+        self._access = access
+        self._shared_apps = shared_apps
+        self._management = management
+        self._identities = identities
+        self._tokens = tokens
+
+    def authenticate(self, access_token: str | None) -> ConsumerIdentity:
+        if not access_token:
+            raise WebIdentityError("Account login required.")
+        claims = self._identities.resolve_webapp_account_login(self._tokens.verify(access_token))
+        assert claims.account_id is not None
+        return ConsumerIdentity(claims.account_id)
+
+    def _visible(self, identity: ConsumerIdentity, apps: Sequence[ConsumerApp]) -> tuple[ConsumerApp, ...]:
+        visible = self._access.get_visible_app_ids(user_id=identity.account_id, app_ids=tuple(app.id for app in apps))
+        return tuple(app for app in apps if app.id in visible)
+
+    def list_apps(self, identity: ConsumerIdentity, query: ConsumerAppsQuery) -> ConsumerAppsPage:
+        # The continuation is the last consumed candidate, including denied Apps.
+        # Fill pages after permission filtering rather than leaking candidate counts.
+        data: list[ConsumerApp] = []
+        after = query.cursor
+        last = after
+        shared_ids = self._shared_apps() if query.view == "shared" else None
+        while True:
+            candidates = self._apps.candidates(
+                identity, query, after=after, limit=query.limit + 1, shared_ids=shared_ids
+            )
+            if not candidates:
+                return ConsumerAppsPage(data=tuple(data), has_more=False, next_cursor=None)
+            visible = {app.id for app in self._visible(identity, candidates)}
+            for app in candidates:
+                if app.id in visible:
+                    if len(data) == query.limit:
+                        return ConsumerAppsPage(data=tuple(data), has_more=True, next_cursor=last)
+                    data.append(app)
+                last = app.continuation
+            if len(candidates) < query.limit + 1:
+                return ConsumerAppsPage(data=tuple(data), has_more=False, next_cursor=None)
+            after = last
+
+    def set_favorite(self, identity: ConsumerIdentity, app_id: str, *, enabled: bool) -> None:
+        if enabled:
+            apps = self._apps.by_ids(identity, (app_id,))
+            if not self._visible(identity, apps):
+                raise ConsumerAppNotFoundError()
+        self._apps.favorite(identity, app_id, enabled=enabled)
+
+    def list_connections(
+        self, identity: ConsumerIdentity, *, provider: str | None, cursor: str | None, limit: int
+    ) -> ConsumerConnectionsPage:
+        after = ConsumerConnectionKey.decode(cursor) if cursor is not None else None
+        if after is not None and provider is not None and after.provider != provider:
+            raise ValueError("Invalid Connection group cursor.")
+        rows = self._connections.account_connection_groups(identity, provider=provider, after=after, limit=limit + 1)
+        members = self._connections.account_group_connections(identity, tuple(row.key for row in rows[:limit]))
+        apps = self._group_apps(identity, members)
+        data = tuple(
+            ConsumerConnection(
+                key=row.representative_id,
+                provider=row.key.provider,
+                connection_type=row.key.connection_type,
+                name=row.name,
+                account_label=row.key.account_label,
+                webapps_count=len(apps.get(row.key, ())),
+            )
+            for row in rows[:limit]
+        )
+        more = len(rows) > limit
+        return ConsumerConnectionsPage(
+            data=data, has_more=more, next_cursor=rows[limit - 1].key.encode() if more else None
+        )
+
+    def _group_apps(
+        self, identity: ConsumerIdentity, members: Mapping[ConsumerConnectionKey, tuple[str, ...]]
+    ) -> Mapping[ConsumerConnectionKey, tuple[ConsumerApp, ...]]:
+        connection_ids = tuple(id_ for ids in members.values() for id_ in ids)
+        associations = self._connections.account_connection_apps(identity, connection_ids)
+        app_ids = tuple(dict.fromkeys(app_id for ids in associations.values() for app_id in ids))
+        apps = {app.id: app for app in self._visible(identity, self._apps.by_ids(identity, app_ids))}
+        return {
+            key: tuple(
+                apps[id_]
+                for id_ in sorted(
+                    {
+                        app_id
+                        for connection_id in members.get(key, ())
+                        for app_id in associations.get(connection_id, ())
+                        if app_id in apps
+                    }
+                )
+            )
+            for key in members
+        }
+
+    def list_connection_apps(
+        self, identity: ConsumerIdentity, connection_key: UUID, *, cursor: str | None, limit: int
+    ) -> ConsumerAppsPage:
+        key = self._connections.account_connection_group(identity, str(connection_key))
+        members = self._connections.account_group_connections(identity, (key,))
+        if not members.get(key):
+            raise ConnectionNotFoundError()
+        apps = self._group_apps(identity, members)[key]
+        if cursor is not None:
+            after = str(UUID(cursor))
+            apps = tuple(app for app in apps if app.id > after)
+        more = len(apps) > limit
+        data = apps[:limit]
+        return ConsumerAppsPage(data=data, has_more=more, next_cursor=data[-1].id if more else None)
+
+    def connection_context(self, identity: ConsumerIdentity, connection_id: str) -> AccountConnectionContext:
+        return self._connections.account_connection_context(identity, connection_id)
+
+    def delete_connection(self, identity: ConsumerIdentity, connection_key: UUID) -> None:
+        key = self._connections.account_connection_group(identity, str(connection_key))
+        members = self._connections.account_group_connections(identity, (key,)).get(key, ())
+        if not members:
+            raise ConnectionNotFoundError()
+        for connection_id in members:
+            try:
+                self._management.delete_connection(self.connection_context(identity, connection_id), connection_id)
+            except ConnectionNotFoundError:
+                # Another disconnect may already have removed a member of the snapshot.
+                continue
diff --git a/api/services/end_user_connection_oauth_adapters.py b/api/services/end_user_connection_oauth_adapters.py
index b54953a1025309..3011e2466ced05 100644
--- a/api/services/end_user_connection_oauth_adapters.py
+++ b/api/services/end_user_connection_oauth_adapters.py
@@ -37,7 +37,11 @@
 from models.provider_ids import ToolProviderID
 from services.end_user_connection_adapters import PluginApiKeyCredentialGateway
 from services.end_user_connection_labels import mask_account_label
-from services.end_user_connections.entities import ConnectionMutationGuard, ConnectionOAuthClient, WebConnectionContext
+from services.end_user_connections.entities import (
+    ConnectionManagementContext,
+    ConnectionMutationGuard,
+    ConnectionOAuthClient,
+)
 from services.end_user_connections.errors import (
     ConnectionCheckUnavailableError,
     ConnectionConfigurationError,
@@ -79,7 +83,7 @@ class RedisConnectionOAuthAttempts:
     def save(self, attempt: ConnectionOAuthAttempt) -> str:
         provider = ToolProviderID(attempt.provider)
         context_id = OAuthProxyService.create_proxy_context(
-            user_id=attempt.context.end_user_id,
+            user_id=attempt.context.actor_id,
             tenant_id=attempt.context.tenant_id,
             plugin_id=provider.plugin_id,
             provider=provider.provider_name,
@@ -111,7 +115,7 @@ def consume(self, context_id: str) -> ConnectionOAuthAttempt:
             attempt = self._adapter.validate_python(stored["attempt"])
             provider = ToolProviderID(attempt.provider)
             if (
-                stored["user_id"] != attempt.context.end_user_id
+                stored["user_id"] != attempt.context.actor_id
                 or stored["tenant_id"] != attempt.context.tenant_id
                 or stored["plugin_id"] != provider.plugin_id
                 or stored["provider"] != provider.provider_name
@@ -225,12 +229,14 @@ def redirect_uri(provider: str) -> str:
         # Keep the callback registered by existing workspace/platform OAuth Clients.
         return f"{dify_config.CONSOLE_API_URL}/console/api/oauth/plugin/{provider}/tool/callback"
 
-    def authorization_url(self, context: WebConnectionContext, provider: str, client: ResolvedOAuthClient) -> str:
+    def authorization_url(
+        self, context: ConnectionManagementContext, provider: str, client: ResolvedOAuthClient
+    ) -> str:
         identity = ToolProviderID(provider)
         try:
             return self._handler.get_authorization_url(
                 tenant_id=context.tenant_id,
-                user_id=context.end_user_id,
+                user_id=context.actor_id,
                 plugin_id=identity.plugin_id,
                 provider=identity.provider_name,
                 redirect_uri=self.redirect_uri(provider),
@@ -241,7 +247,7 @@ def authorization_url(self, context: WebConnectionContext, provider: str, client
             raise ConnectionOAuthError() from None
 
     def exchange(
-        self, context: WebConnectionContext, provider: str, client: ResolvedOAuthClient, query_string: bytes
+        self, context: ConnectionManagementContext, provider: str, client: ResolvedOAuthClient, query_string: bytes
     ) -> ConnectionOAuthGrant:
         identity = ToolProviderID(provider)
         try:
@@ -253,7 +259,7 @@ def exchange(
                 callback = Request(builder.get_environ())
                 result = self._handler.get_credentials(
                     tenant_id=context.tenant_id,
-                    user_id=context.end_user_id,
+                    user_id=context.actor_id,
                     plugin_id=identity.plugin_id,
                     provider=identity.provider_name,
                     redirect_uri=self.redirect_uri(provider),
@@ -287,14 +293,18 @@ def _grant(result: PluginOAuthCredentialsResponse) -> ConnectionOAuthGrant:
         )
 
     def refresh(
-        self, context: WebConnectionContext, provider: str, client: ResolvedOAuthClient, credentials: Mapping[str, Any]
+        self,
+        context: ConnectionManagementContext,
+        provider: str,
+        client: ResolvedOAuthClient,
+        credentials: Mapping[str, Any],
     ) -> ConnectionOAuthGrant | CredentialCheckResult:
         identity = ToolProviderID(provider)
         try:
             with use_plugin_daemon_request_timeout(10):
                 result = self._handler.refresh_credentials(
                     tenant_id=context.tenant_id,
-                    user_id=context.end_user_id,
+                    user_id=context.actor_id,
                     plugin_id=identity.plugin_id,
                     provider=identity.provider_name,
                     redirect_uri=self.redirect_uri(provider),
@@ -328,12 +338,16 @@ def refresh(
             )
 
     def check(
-        self, context: WebConnectionContext, provider: str, client: ResolvedOAuthClient, credentials: Mapping[str, Any]
+        self,
+        context: ConnectionManagementContext,
+        provider: str,
+        client: ResolvedOAuthClient,
+        credentials: Mapping[str, Any],
     ) -> CredentialCheckResult:
         try:
             return self._auth.check_provider_credentials(
                 tenant_id=context.tenant_id,
-                user_id=context.end_user_id,
+                user_id=context.actor_id,
                 provider=provider,
                 credentials=credentials,
                 credential_type=CredentialType.OAUTH2,
@@ -352,12 +366,16 @@ def check(
             )
 
     def revoke(
-        self, context: WebConnectionContext, provider: str, client: ResolvedOAuthClient, credentials: Mapping[str, Any]
+        self,
+        context: ConnectionManagementContext,
+        provider: str,
+        client: ResolvedOAuthClient,
+        credentials: Mapping[str, Any],
     ) -> OAuthRevokeResult:
         try:
             return self._auth.revoke_credentials(
                 tenant_id=context.tenant_id,
-                user_id=context.end_user_id,
+                user_id=context.actor_id,
                 provider=provider,
                 system_credentials=client.parameters,
                 plugin_unique_identifier=self._artifact(client),
diff --git a/api/services/end_user_connections/entities.py b/api/services/end_user_connections/entities.py
index 3abe41b149b6c7..593c7a6a6b766c 100644
--- a/api/services/end_user_connections/entities.py
+++ b/api/services/end_user_connections/entities.py
@@ -44,6 +44,38 @@ def owner(self) -> ConnectionOwner:
             return ConnectionOwner(account_id=self.account_id, end_user_id=None)
         return ConnectionOwner(account_id=None, end_user_id=self.end_user_id)
 
+    @property
+    def actor_id(self) -> str:
+        return self.end_user_id
+
+
+@dataclass(frozen=True, slots=True)
+class AccountConnectionContext:
+    """Account-owned Connection authority for queries and disconnect, without an App.
+
+    Repositories revalidate the Account and the exact tenant/Connection/owner chain
+    after asynchronous boundaries. This context never authorizes node execution.
+    """
+
+    tenant_id: str
+    account_id: str
+    connection_id: str
+
+    def __post_init__(self) -> None:
+        if not self.tenant_id or not self.account_id or not self.connection_id:
+            raise ValueError("Incomplete account connection identity.")
+
+    @property
+    def owner(self) -> ConnectionOwner:
+        return ConnectionOwner(account_id=self.account_id, end_user_id=None)
+
+    @property
+    def actor_id(self) -> str:
+        return self.account_id
+
+
+type ConnectionManagementContext = WebConnectionContext | AccountConnectionContext
+
 
 @dataclass(frozen=True, slots=True)
 class ConnectionOAuthClient:
diff --git a/api/services/end_user_connections/management.py b/api/services/end_user_connections/management.py
index 6dd7c8783f2d7d..0ca44b197ea430 100644
--- a/api/services/end_user_connections/management.py
+++ b/api/services/end_user_connections/management.py
@@ -20,7 +20,12 @@
     ListConnectionsInput,
     UpdateConnectionInput,
 )
-from services.end_user_connections.entities import ConnectionAccessRecord, ConnectionMutationGuard, WebConnectionContext
+from services.end_user_connections.entities import (
+    ConnectionAccessRecord,
+    ConnectionManagementContext,
+    ConnectionMutationGuard,
+    WebConnectionContext,
+)
 from services.end_user_connections.errors import (
     ConnectionCheckUnavailableError,
     ConnectionConfigurationError,
@@ -58,7 +63,7 @@ def list(
         self, context: WebConnectionContext, query: ListConnectionsInput, after_id: str | None
     ) -> ConnectionPage: ...
 
-    def find(self, context: WebConnectionContext, connection_id: str) -> StoredConnection | None: ...
+    def find(self, context: ConnectionManagementContext, connection_id: str) -> StoredConnection | None: ...
 
     def create_api_key(
         self,
@@ -79,7 +84,7 @@ def update(
         """Revalidate identity and CAS; a no-change check still verifies the guard."""
         ...
 
-    def delete(self, context: WebConnectionContext, guard: ConnectionMutationGuard) -> None: ...
+    def delete(self, context: ConnectionManagementContext, guard: ConnectionMutationGuard) -> None: ...
 
 
 class ConnectionCipher(Protocol):
@@ -117,7 +122,7 @@ def check(
         self, context: WebConnectionContext, record: StoredConnection, *, plugin_unique_identifier: str
     ) -> ConnectionCheckOutcome: ...
 
-    def revoke_deleted(self, context: WebConnectionContext, record: StoredConnection) -> None:
+    def revoke_deleted(self, context: ConnectionManagementContext, record: StoredConnection) -> None:
         """Best effort after a successful conditional delete; never restore the record."""
         ...
 
@@ -148,7 +153,7 @@ def list_connections(self, context: WebConnectionContext, query: ListConnections
             raise ValueError("Invalid connection cursor.") from None
         return self._store.list(context, query, after_id)
 
-    def get_connection(self, context: WebConnectionContext, connection_id: str) -> ConnectionSummary:
+    def get_connection(self, context: ConnectionManagementContext, connection_id: str) -> ConnectionSummary:
         return self._load(context, connection_id).summary
 
     def get_connection_revision(self, context: WebConnectionContext, connection_id: str) -> ConnectionRevisionView:
@@ -260,7 +265,7 @@ def update_connection(
         )
         return updated.summary
 
-    def delete_connection(self, context: WebConnectionContext, connection_id: str) -> None:
+    def delete_connection(self, context: ConnectionManagementContext, connection_id: str) -> None:
         record = self._load(context, connection_id)
         self._store.delete(context, connection_mutation_guard(context, record.access))
         if record.access.credential_type == ConnectionCredentialType.OAUTH2:
@@ -316,7 +321,7 @@ def check_connection_result(
             revision=current.access.revision,
         )
 
-    def _load(self, context: WebConnectionContext, connection_id: str) -> StoredConnection:
+    def _load(self, context: ConnectionManagementContext, connection_id: str) -> StoredConnection:
         require_personal_connection_access(context)
         record = self._store.find(context, connection_id)
         if record is None or record.access.id != connection_id:
diff --git a/api/services/end_user_connections/oauth_lifecycle.py b/api/services/end_user_connections/oauth_lifecycle.py
index dc58190b691e0f..399416fc96735c 100644
--- a/api/services/end_user_connections/oauth_lifecycle.py
+++ b/api/services/end_user_connections/oauth_lifecycle.py
@@ -9,7 +9,12 @@
 from clients.plugin.auth_entities import CredentialCheckResult, OAuthRevokeResult
 from constants.end_user_connections import ConnectionCredentialStatus, OAuthClientSource
 from services.end_user_connections.contracts import ConnectionCheckOutcome, ConnectionCheckView
-from services.end_user_connections.entities import ConnectionMutationGuard, ConnectionOAuthClient, WebConnectionContext
+from services.end_user_connections.entities import (
+    ConnectionManagementContext,
+    ConnectionMutationGuard,
+    ConnectionOAuthClient,
+    WebConnectionContext,
+)
 from services.end_user_connections.errors import (
     ConnectionConflictError,
     ConnectionNotFoundError,
@@ -31,15 +36,27 @@ def resolve_grant_client(
     ) -> ResolvedOAuthClient: ...
 
     def refresh(
-        self, context: WebConnectionContext, provider: str, client: ResolvedOAuthClient, credentials: Mapping[str, Any]
+        self,
+        context: WebConnectionContext,
+        provider: str,
+        client: ResolvedOAuthClient,
+        credentials: Mapping[str, Any],
     ) -> ConnectionOAuthGrant | CredentialCheckResult: ...
 
     def check(
-        self, context: WebConnectionContext, provider: str, client: ResolvedOAuthClient, credentials: Mapping[str, Any]
+        self,
+        context: WebConnectionContext,
+        provider: str,
+        client: ResolvedOAuthClient,
+        credentials: Mapping[str, Any],
     ) -> CredentialCheckResult: ...
 
     def revoke(
-        self, context: WebConnectionContext, provider: str, client: ResolvedOAuthClient, credentials: Mapping[str, Any]
+        self,
+        context: ConnectionManagementContext,
+        provider: str,
+        client: ResolvedOAuthClient,
+        credentials: Mapping[str, Any],
     ) -> OAuthRevokeResult: ...
 
 
@@ -254,7 +271,7 @@ def _persist(
             client=client.snapshot,
         )
 
-    def revoke_deleted(self, context: WebConnectionContext, record: StoredConnection) -> None:
+    def revoke_deleted(self, context: ConnectionManagementContext, record: StoredConnection) -> None:
         reference = record.access.oauth_client
         if (
             reference is None
diff --git a/api/services/end_user_connections/policy.py b/api/services/end_user_connections/policy.py
index 14b14d6235921f..4a82f469f28a21 100644
--- a/api/services/end_user_connections/policy.py
+++ b/api/services/end_user_connections/policy.py
@@ -4,6 +4,7 @@
 from constants.end_user_connections import ConnectionCredentialType
 from services.end_user_connections.entities import (
     ConnectionAccessRecord,
+    ConnectionManagementContext,
     ConnectionMutationGuard,
     ToolConnectionTarget,
     WebConnectionContext,
@@ -17,13 +18,13 @@
 )
 
 
-def require_personal_connection_access(context: WebConnectionContext) -> None:
+def require_personal_connection_access(context: ConnectionManagementContext) -> None:
     if context.account_id is None and not dify_config.END_USER_AUTH_ALLOW_ANONYMOUS:
         raise ConnectionAccountRequiredError()
 
 
 def require_owned_connection(
-    context: WebConnectionContext, record: ConnectionAccessRecord | None
+    context: ConnectionManagementContext, record: ConnectionAccessRecord | None
 ) -> ConnectionAccessRecord:
     """Ownership allows management even when a node, Provider, or Client changed.
 
@@ -106,7 +107,7 @@ def require_applicable_connection(
 
 
 def connection_mutation_guard(
-    context: WebConnectionContext, record: ConnectionAccessRecord | None
+    context: ConnectionManagementContext, record: ConnectionAccessRecord | None
 ) -> ConnectionMutationGuard:
     owned = require_owned_connection(context, record)
     return ConnectionMutationGuard(
diff --git a/api/tests/helpers/consumer.py b/api/tests/helpers/consumer.py
new file mode 100644
index 00000000000000..9ae4fd07275f26
--- /dev/null
+++ b/api/tests/helpers/consumer.py
@@ -0,0 +1,103 @@
+"""Real Account consumer services with deterministic plugin and access boundaries."""
+
+from collections.abc import Sequence
+from dataclasses import dataclass, field
+from datetime import UTC, datetime, timedelta
+from uuid import UUID
+
+import pytest
+from sqlalchemy.orm import Session, sessionmaker
+
+from graphon.enums import WorkflowExecutionStatus
+from libs.passport import PassportService
+from models.end_user_connection import EndUserConnection
+from models.enums import CreatorUserRole, WorkflowRunTriggeredFrom
+from models.workflow import WorkflowRun, WorkflowType
+from repositories.account.repository import SQLAlchemyAccountRepository
+from repositories.app.console_repository import ConsoleAppRepository
+from services.consumer_service import ConsumerIdentity, ConsumerService
+from services.end_user_connections.contracts import ConnectionSummary
+from services.web_identity_service import WebIdentityService
+from services.web_passport_gateways import PassportTokenGateway
+from tests.helpers.end_user_api_key import ConnectionHarness
+from tests.helpers.end_user_oauth import OAuthHarness, make_oauth_harness
+from tests.helpers.unit.config_override import apply_config_overrides
+
+
+def consumer_connection_key(connection: EndUserConnection | ConnectionSummary) -> UUID:
+    return UUID(connection.id)
+
+
+def new_consumer_run(base: ConnectionHarness, *, created_at: datetime) -> WorkflowRun:
+    return WorkflowRun(
+        tenant_id=base.context.tenant_id,
+        app_id=base.context.app_id,
+        workflow_id=base.workflow_id,
+        type=WorkflowType.WORKFLOW,
+        triggered_from=WorkflowRunTriggeredFrom.APP_RUN,
+        version="published",
+        status=WorkflowExecutionStatus.SUCCEEDED,
+        created_by_role=CreatorUserRole.END_USER,
+        created_by=base.context.end_user_id,
+        created_at=created_at,
+    )
+
+
+@dataclass
+class ConsumerAccess:
+    denied: set[str] = field(default_factory=set)
+
+    def get_visible_app_ids(self, *, user_id: str, app_ids: Sequence[str]) -> frozenset[str]:
+        return frozenset(app_id for app_id in app_ids if app_id not in self.denied)
+
+
+@dataclass
+class ConsumerHarness:
+    oauth: OAuthHarness
+    apps: ConsoleAppRepository
+    access: ConsumerAccess
+    service: ConsumerService
+    identity: ConsumerIdentity
+
+    def token(self, *, source: str | None = "webapp_account_login", subject: str = "Web API Account Login") -> str:
+        return PassportService().issue(
+            {
+                "sub": subject,
+                "user_id": self.identity.account_id,
+                "auth_type": "internal",
+                "token_source": source,
+                "exp": datetime.now(UTC) + timedelta(minutes=10),
+            }
+        )
+
+
+def make_consumer_harness(sessions: sessionmaker[Session], monkeypatch: pytest.MonkeyPatch) -> ConsumerHarness:
+    apply_config_overrides(
+        monkeypatch,
+        SECRET_KEY="consumer-test-key-with-at-least-32-bytes",
+        CONSOLE_API_URL="http://localhost",
+        CONSOLE_WEB_URL="http://localhost",
+    )
+    oauth = make_oauth_harness(sessions, monkeypatch)
+    assert oauth.base.context.account_id is not None
+    apps = ConsoleAppRepository(session_factory=sessions)
+    access = ConsumerAccess()
+    return ConsumerHarness(
+        oauth,
+        apps,
+        access,
+        ConsumerService(
+            apps=apps,
+            connections=oauth.base.repository,
+            access=access,
+            shared_apps=lambda: (oauth.base.context.app_id,),
+            management=oauth.base.service,
+            identities=WebIdentityService(
+                accounts=SQLAlchemyAccountRepository(session_factory=sessions),
+                allow_console_login=True,
+                allow_webapp_login=False,
+            ),
+            tokens=PassportTokenGateway(passport=PassportService()),
+        ),
+        ConsumerIdentity(oauth.base.context.account_id),
+    )
diff --git a/api/tests/helpers/end_user_connection_contract.py b/api/tests/helpers/end_user_connection_contract.py
index 77b2044c710c88..811aba122a938a 100644
--- a/api/tests/helpers/end_user_connection_contract.py
+++ b/api/tests/helpers/end_user_connection_contract.py
@@ -7,7 +7,7 @@
 import pytest
 import sqlalchemy as sa
 from sqlalchemy.exc import DataError, IntegrityError
-from sqlalchemy.orm import Session
+from sqlalchemy.orm import Session, sessionmaker
 
 from libs.uuid_utils import uuidv7
 from models.end_user_connection import (
@@ -18,7 +18,10 @@
     OAuthClientSource,
     SessionConnection,
 )
+from repositories.end_user_connection_repository import EndUserConnectionRepository
+from services.consumer_service import ConsumerIdentity
 from services.end_user_connection_labels import mask_account_label
+from services.end_user_connections.errors import ConnectionNotFoundError
 
 
 def new_connection() -> EndUserConnection:
@@ -63,6 +66,38 @@ def new_session_selection(connection: EndUserConnection) -> SessionConnection:
 
 
 class ConnectionPersistenceTests:
+    def test_consumer_groups_page_and_resolve_all_owned_members(self, connection_session: Session) -> None:
+        api_key = new_connection()
+        assert api_key.account_id is not None
+        oauth = replace(
+            api_key,
+            credential_type=ConnectionCredentialType.OAUTH2,
+            oauth_client_source=OAuthClientSource.WORKSPACE,
+            account_label="al***@example.com",
+        )
+        duplicate = replace(oauth, tenant_id=str(UUID(int=10)))
+        foreign = replace(oauth, account_id=str(UUID(int=11)))
+        connection_session.add_all([api_key, oauth, duplicate, foreign])
+        connection_session.commit()
+        repository = EndUserConnectionRepository(
+            session_factory=sessionmaker(bind=connection_session.get_bind(), expire_on_commit=False)
+        )
+        identity = ConsumerIdentity(api_key.account_id)
+        first = repository.account_connection_groups(identity, provider=None, after=None, limit=1)
+        assert len(first) == 1
+        assert first[0].key.connection_id == UUID(api_key.id)
+        assert first[0].representative_id == UUID(api_key.id)
+        second = repository.account_connection_groups(identity, provider=None, after=first[0].key, limit=1)
+        assert len(second) == 1
+        assert second[0].key.account_label == oauth.account_label
+        assert repository.account_connection_group(identity, duplicate.id) == second[0].key
+        with pytest.raises(ConnectionNotFoundError):
+            repository.account_connection_group(identity, foreign.id)
+        assert repository.account_group_connections(identity, (second[0].key,))[second[0].key] == tuple(
+            sorted((oauth.id, duplicate.id))
+        )
+        assert repository.account_connection_groups(identity, provider=None, after=second[0].key, limit=1) == ()
+
     @pytest.mark.parametrize("account_owned", [True, False])
     def test_multiple_connections_per_owner_and_provider(
         self, connection_session: Session, account_owned: bool
diff --git a/api/tests/unit_tests/controllers/web/test_connections.py b/api/tests/unit_tests/controllers/web/test_connections.py
index 4b15e6763ac05c..deae153df70e19 100644
--- a/api/tests/unit_tests/controllers/web/test_connections.py
+++ b/api/tests/unit_tests/controllers/web/test_connections.py
@@ -232,6 +232,36 @@ def test_standalone_check_route_is_removed(client: FlaskClient, harness: Connect
     assert harness.auth.calls == []
 
 
+def test_connection_write_revision_is_independent_of_authorization_contract(
+    client: FlaskClient, harness: ConnectionHarness
+) -> None:
+    connection = harness.service.create_api_key(harness.context, harness.command())
+    path = f"/api/connections/{connection.id}"
+    app_revision = harness.connection_revision
+    contract_revision = client.get(path + "/revision").get_json()["connection_revision"]
+    before = harness.repository.find(harness.context, connection.id)
+    assert before is not None
+    assert before.access.revision == 1
+
+    assert client.patch(path, json={"name": "Renamed"}).status_code == 200
+    renamed = harness.repository.find(harness.context, connection.id)
+    assert renamed is not None
+    assert renamed.access.revision == 2
+    assert harness.connection_revision == app_revision
+    assert client.get(path + "/revision").get_json()["connection_revision"] == contract_revision
+
+    replaced = client.patch(
+        path,
+        json={"credentials": {"api_key": "replacement"}, "connection_revision": contract_revision},
+    )
+    assert replaced.status_code == 200
+    current = harness.repository.find(harness.context, connection.id)
+    assert current is not None
+    assert current.access.revision == 3
+    assert harness.connection_revision == app_revision
+    assert client.get(path + "/revision").get_json()["connection_revision"] == contract_revision
+
+
 def test_provider_revision_is_available_after_the_authorization_node_is_removed(
     client: FlaskClient, harness: ConnectionHarness
 ) -> None:
diff --git a/api/tests/unit_tests/controllers/web/test_me.py b/api/tests/unit_tests/controllers/web/test_me.py
new file mode 100644
index 00000000000000..6030037f018b75
--- /dev/null
+++ b/api/tests/unit_tests/controllers/web/test_me.py
@@ -0,0 +1,165 @@
+"""Serve consumer APIs through WebApp Account login and same-origin cookies."""
+
+from collections.abc import Iterator
+from datetime import datetime
+from types import SimpleNamespace
+from uuid import uuid4
+
+import pytest
+from flask import Flask
+from flask.testing import FlaskClient
+from sqlalchemy.orm import Session, sessionmaker
+
+from constants import COOKIE_NAME_WEBAPP_ACCESS_TOKEN
+from controllers.web import bp, me
+from models.end_user_connection import AppConnection
+from tests.helpers.consumer import ConsumerHarness, consumer_connection_key, make_consumer_harness, new_consumer_run
+from tests.helpers.end_user_connection_contract import new_connection
+
+
+@pytest.fixture
+def harness(sqlite_session_factory: sessionmaker[Session], monkeypatch: pytest.MonkeyPatch) -> ConsumerHarness:
+    return make_consumer_harness(sqlite_session_factory, monkeypatch)
+
+
+@pytest.fixture
+def client(harness: ConsumerHarness, monkeypatch: pytest.MonkeyPatch) -> Iterator[FlaskClient]:
+    monkeypatch.setattr(me, "application_services", lambda: SimpleNamespace(consumer=harness.service))
+    app = Flask(__name__)
+    app.config.update(TESTING=True, RESTX_ERROR_404_HELP=False)
+    app.register_blueprint(bp)
+    with app.test_client() as client:
+        client.set_cookie(COOKIE_NAME_WEBAPP_ACCESS_TOKEN, harness.token())
+        client.environ_base["HTTP_ORIGIN"] = "http://localhost"
+        yield client
+
+
+def test_http_app_views_favorites_and_timestamps(client: FlaskClient, harness: ConsumerHarness) -> None:
+    base = harness.oauth.base
+    path = f"/api/me/apps/{base.context.app_id}/favorite"
+    assert client.put(path).status_code == 204
+    assert client.put(path).status_code == 204
+    with base.sessions.begin() as session:
+        session.add(new_consumer_run(base, created_at=datetime(2026, 10, 30)))
+    for view in ("recent", "favorites", "shared"):
+        response = client.get(f"/api/me/apps?view={view}&limit=1")
+        assert response.status_code == 200
+        data = response.get_json()
+        assert data["has_more"] is False
+        assert data["data"][0]["id"] == base.context.app_id
+        assert isinstance(data["data"][0]["last_used_at"], int)
+        assert "continuation" not in data["data"][0]
+    assert client.delete(path).status_code == 204
+    assert client.get("/api/me/apps?view=favorites").get_json()["data"] == []
+
+
+def test_connections_support_only_query_and_disconnect(client: FlaskClient, harness: ConsumerHarness) -> None:
+    base = harness.oauth.base
+    created = base.service.create_api_key(base.context, base.command())
+    listed = client.get("/api/me/connections?limit=1")
+    assert listed.status_code == 200
+    listed_connection = listed.get_json()["data"][0]
+    assert listed_connection == {
+        "key": str(consumer_connection_key(created)),
+        "provider": created.provider,
+        "connection_type": "api-key",
+        "name": created.name,
+        "account_label": None,
+        "webapps_count": 0,
+    }
+    path = f"/api/me/connections/{listed_connection['key']}"
+    assert "encrypted_credentials" not in listed.get_data(as_text=True)
+    assert client.get(path).status_code == 405
+    assert client.patch(path, json={"name": "Renamed"}).status_code == 405
+    assert client.get(path + "/credential-schema").status_code == 404
+    assert client.post(path + "/reauthorize").status_code == 404
+    assert client.get(path + "/revision").status_code == 404
+    assert client.get(path + "/apps").get_json()["data"] == []
+    assert client.delete(path).status_code == 204
+    assert client.get(path + "/apps").status_code == 404
+    assert client.get("/api/me/connections").get_json()["data"] == []
+
+
+def test_foreign_connection_matches_unknown_http_id(client: FlaskClient) -> None:
+    response = client.delete(f"/api/me/connections/{consumer_connection_key(new_connection())}")
+    assert response.status_code == 404
+    assert response.get_json()["code"] == "connection_not_found"
+
+
+def test_connection_group_routes_reuse_the_list_key(client: FlaskClient, harness: ConsumerHarness) -> None:
+    base = harness.oauth.base
+    connections = [harness.oauth.finish(harness.oauth.start()) for _ in range(2)]
+    with base.sessions.begin() as session:
+        session.add_all(
+            AppConnection(
+                tenant_id=base.context.tenant_id,
+                app_id=base.context.app_id,
+                workflow_id=base.workflow_id,
+                node_id=f"node-{index}",
+                end_user_id=base.context.end_user_id,
+                provider=connection.provider,
+                connection_id=connection.id,
+            )
+            for index, connection in enumerate(connections)
+        )
+    listed = client.get("/api/me/connections").get_json()
+    assert len(listed["data"]) == 1
+    group = listed["data"][0]
+    assert group["webapps_count"] == 1
+    assert group["connection_type"] == "oauth2"
+    assert "apps" not in group
+    path = f"/api/me/connections/{group['key']}"
+    apps = client.get(path + "/apps?limit=1")
+    assert apps.status_code == 200
+    assert [app["id"] for app in apps.get_json()["data"]] == [base.context.app_id]
+    assert apps.get_json()["has_more"] is False
+    assert client.get(path + "/apps?cursor=invalid").status_code == 400
+    assert client.get(path + "/apps?limit=0").status_code == 400
+    assert client.delete(path).status_code == 204
+    assert len(base.auth.revocations) == 2
+    assert client.get("/api/me/connections").get_json()["data"] == []
+
+
+def test_malformed_connection_group_key_does_not_match_routes(client: FlaskClient) -> None:
+    path = "/api/me/connections/invalid"
+    assert client.get(path + "/apps").status_code == 404
+    assert client.delete(path).status_code == 404
+
+
+@pytest.mark.parametrize("source", [None, "webapp", "webapp_account", "webapp_login_token", "unknown"])
+def test_console_and_app_tokens_cannot_authenticate_consumer_routes(
+    client: FlaskClient, harness: ConsumerHarness, source: str | None
+) -> None:
+    subject = "Console API Passport" if source is None else "Web API Account Login"
+    client.set_cookie(COOKIE_NAME_WEBAPP_ACCESS_TOKEN, harness.token(source=source, subject=subject))
+    assert client.get("/api/me/apps").status_code == 401
+
+
+@pytest.mark.parametrize("origin", [None, "https://evil.example", "null", "http://localhost.evil"])
+def test_account_mutations_require_a_trusted_origin(
+    client: FlaskClient, harness: ConsumerHarness, origin: str | None
+) -> None:
+    client.environ_base.pop("HTTP_ORIGIN")
+    headers: dict[str, str] = {} if origin is None else {"Origin": origin}
+    response = client.put(f"/api/me/apps/{harness.oauth.base.context.app_id}/favorite", headers=headers)
+    assert response.status_code == 403
+    assert client.get("/api/me/apps", headers=headers).status_code == (200 if origin is None else 403)
+
+
+def test_webapp_login_cookie_is_required(client: FlaskClient) -> None:
+    client.delete_cookie(COOKIE_NAME_WEBAPP_ACCESS_TOKEN)
+    assert client.get("/api/me/apps").status_code == 401
+
+
+def test_account_reads_do_not_require_origin(client: FlaskClient) -> None:
+    client.environ_base.pop("HTTP_ORIGIN")
+    assert client.get("/api/me/apps").status_code == 200
+
+
+def test_mutations_require_origin_even_with_a_trusted_referrer(client: FlaskClient) -> None:
+    client.environ_base.pop("HTTP_ORIGIN")
+    response = client.put(
+        f"/api/me/apps/{uuid4()}/favorite",
+        headers={"Referer": "http://localhost/apps/1"},
+    )
+    assert response.status_code == 403
diff --git a/api/tests/unit_tests/migrations/test_2026_10_30_1600-b73e9c40a215_add_consumer_execution_indexes.py b/api/tests/unit_tests/migrations/test_2026_10_30_1600-b73e9c40a215_add_consumer_execution_indexes.py
new file mode 100644
index 00000000000000..7e75b371f93813
--- /dev/null
+++ b/api/tests/unit_tests/migrations/test_2026_10_30_1600-b73e9c40a215_add_consumer_execution_indexes.py
@@ -0,0 +1,65 @@
+"""Verify execution lookup indexes without creating a second usage ledger."""
+
+from importlib import import_module
+
+import sqlalchemy as sa
+from alembic.migration import MigrationContext
+from alembic.operations import Operations
+
+
+def test_execution_indexes_upgrade_and_downgrade_preserve_rows() -> None:
+    migration = import_module("migrations.versions.2026_10_30_1600-b73e9c40a215_add_consumer_execution_indexes")
+    engine = sa.create_engine("sqlite://")
+    metadata = sa.MetaData()
+    users = sa.Table(
+        "end_users",
+        metadata,
+        sa.Column("id", sa.String, primary_key=True),
+        sa.Column("account_id", sa.String),
+        sa.Column("type", sa.String),
+        sa.Column("tenant_id", sa.String),
+        sa.Column("app_id", sa.String),
+    )
+    runs = sa.Table(
+        "workflow_runs",
+        metadata,
+        sa.Column("id", sa.String, primary_key=True),
+        sa.Column("tenant_id", sa.String),
+        sa.Column("app_id", sa.String),
+        sa.Column("created_by_role", sa.String),
+        sa.Column("created_by", sa.String),
+        sa.Column("triggered_from", sa.String),
+        sa.Column("created_at", sa.DateTime),
+    )
+    try:
+        with engine.begin() as connection:
+            metadata.create_all(connection)
+            connection.execute(users.insert().values(id="user", account_id="account"))
+            connection.execute(runs.insert().values(id="run", created_by="user"))
+            with Operations.context(MigrationContext.configure(connection)):
+                migration.upgrade()
+            inspector = sa.inspect(connection)
+            assert set(inspector.get_table_names()) == {"end_users", "workflow_runs"}
+            assert inspector.get_indexes("end_users")[0]["column_names"] == [
+                "account_id",
+                "type",
+                "tenant_id",
+                "app_id",
+                "id",
+            ]
+            assert inspector.get_indexes("workflow_runs")[0]["column_names"] == [
+                "tenant_id",
+                "app_id",
+                "created_by_role",
+                "created_by",
+                "triggered_from",
+                "created_at",
+            ]
+            with Operations.context(MigrationContext.configure(connection)):
+                migration.downgrade()
+            assert sa.inspect(connection).get_indexes("end_users") == []
+            assert sa.inspect(connection).get_indexes("workflow_runs") == []
+            assert connection.execute(sa.select(users.c.id, users.c.account_id)).one() == ("user", "account")
+            assert connection.execute(sa.select(runs.c.id, runs.c.created_by)).one() == ("run", "user")
+    finally:
+        engine.dispose()
diff --git a/api/tests/unit_tests/migrations/test_end_user_connections_migration.py b/api/tests/unit_tests/migrations/test_end_user_connections_migration.py
index 4cebac03fb74bb..14335a1e78c819 100644
--- a/api/tests/unit_tests/migrations/test_end_user_connections_migration.py
+++ b/api/tests/unit_tests/migrations/test_end_user_connections_migration.py
@@ -22,6 +22,7 @@
 
 FEATURE_MIGRATION = "2026_10_26_1300-f3a6b9c2d108_add_connection_revision_contracts.py"
 CONFIGURATION_MIGRATION = "2026_10_29_1200-a91d5e7c3b20_save_connection_authorizations.py"
+CONSUMER_INDEX_MIGRATION = "2026_11_03_1200-4f2a90b8c317_index_consumer_connection_groups.py"
 TABLE_NAMES = {
     "end_user_connections",
     "app_connections",
@@ -61,6 +62,7 @@ def connection_session() -> Iterator[Session]:
         load_migration(connection).upgrade()
         load_migration(connection, FEATURE_MIGRATION).upgrade()
         load_migration(connection, CONFIGURATION_MIGRATION).upgrade()
+        load_migration(connection, CONSUMER_INDEX_MIGRATION).upgrade()
         connection.commit()
         with Session(connection) as session:
             yield session
@@ -71,6 +73,26 @@ class TestMigratedConnections(ConnectionPersistenceTests):
     """Exercise deployed constraints as well as ORM-created tables."""
 
 
+def test_consumer_index_upgrade_and_downgrade_preserve_connections(connection_session: Session) -> None:
+    record = new_connection()
+    connection_session.add(record)
+    connection_session.flush()
+    connection = connection_session.connection()
+    migration = load_migration(connection, CONSUMER_INDEX_MIGRATION)
+    migration.downgrade()
+    assert "end_user_connection_consumer_idx" not in {
+        index["name"] for index in sa.inspect(connection).get_indexes("end_user_connections")
+    }
+    migration.upgrade()
+    index = next(
+        item
+        for item in sa.inspect(connection).get_indexes("end_user_connections")
+        if item["name"] == "end_user_connection_consumer_idx"
+    )
+    assert index["column_names"] == ["account_id", "provider", "credential_type", "id"]
+    assert connection_session.scalar(sa.select(sa.func.count()).select_from(type(record))) == 1
+
+
 def test_migration_schema_matches_models(connection_session: Session) -> None:
     metadata = sa.MetaData()
     for model in (
diff --git a/api/tests/unit_tests/services/test_consumer_adapters.py b/api/tests/unit_tests/services/test_consumer_adapters.py
new file mode 100644
index 00000000000000..56fc1d4edf0c52
--- /dev/null
+++ b/api/tests/unit_tests/services/test_consumer_adapters.py
@@ -0,0 +1,39 @@
+"""Shared discovery uses the existing deployment contract and fails closed."""
+
+import pytest
+
+from services.consumer_adapters import SharedConsumerAppDiscovery
+from services.enterprise.enterprise_service import EnterpriseService
+from services.errors.enterprise import EnterpriseAPIError
+from services.webapp_access_query_service import WebAppAccessUnavailableError
+
+
+def test_shared_discovery_consumes_all_pages_and_deduplicates(monkeypatch: pytest.MonkeyPatch) -> None:
+    def list_apps(*, page: int, limit: int) -> dict[str, object]:
+        assert limit == 100
+        if page == 1:
+            return {"data": [{"appId": "app-1"}], "hasMore": True}
+        assert page == 2
+        return {"data": [{"appId": "app-1"}, {"appId": "app-2"}], "hasMore": False}
+
+    monkeypatch.setattr(EnterpriseService.WebAppAuth, "list_externally_accessible_apps", list_apps)
+    assert SharedConsumerAppDiscovery(enabled=True)() == ("app-1", "app-2")
+    assert SharedConsumerAppDiscovery(enabled=False)() == ()
+
+
+@pytest.mark.parametrize("result", [{}, {"data": [], "hasMore": True}, {"data": [], "hasMore": "false"}])
+def test_invalid_shared_discovery_response_is_unavailable(
+    monkeypatch: pytest.MonkeyPatch, result: dict[str, object]
+) -> None:
+    monkeypatch.setattr(EnterpriseService.WebAppAuth, "list_externally_accessible_apps", lambda **_kwargs: result)
+    with pytest.raises(WebAppAccessUnavailableError):
+        SharedConsumerAppDiscovery(enabled=True)()
+
+
+def test_discovery_outage_is_unavailable(monkeypatch: pytest.MonkeyPatch) -> None:
+    def unavailable(**_kwargs) -> dict[str, object]:
+        raise EnterpriseAPIError("Unavailable", status_code=503)
+
+    monkeypatch.setattr(EnterpriseService.WebAppAuth, "list_externally_accessible_apps", unavailable)
+    with pytest.raises(WebAppAccessUnavailableError):
+        SharedConsumerAppDiscovery(enabled=True)()
diff --git a/api/tests/unit_tests/services/test_consumer_connection_groups.py b/api/tests/unit_tests/services/test_consumer_connection_groups.py
new file mode 100644
index 00000000000000..532b8eb1e9397b
--- /dev/null
+++ b/api/tests/unit_tests/services/test_consumer_connection_groups.py
@@ -0,0 +1,182 @@
+"""Account-scoped Connection grouping, associated Apps and group disconnects."""
+
+from dataclasses import replace
+from uuid import uuid4
+
+import pytest
+import sqlalchemy as sa
+from sqlalchemy.orm import Session, sessionmaker
+
+from constants.end_user_connections import ConnectionCredentialType, OAuthClientSource
+from models.end_user_connection import AppConnection, EndUserConnection, SessionConnection
+from models.enums import CustomizeTokenStrategy
+from models.model import App, Site
+from services.end_user_connections.errors import ConnectionNotFoundError
+from tests.helpers.consumer import ConsumerHarness, consumer_connection_key, make_consumer_harness
+from tests.helpers.end_user_api_key import ConnectionHarness, make_harness
+from tests.helpers.end_user_connection_contract import new_connection
+
+
+@pytest.fixture
+def harness(sqlite_session_factory: sessionmaker[Session], monkeypatch: pytest.MonkeyPatch) -> ConsumerHarness:
+    return make_consumer_harness(sqlite_session_factory, monkeypatch)
+
+
+def _oauth_row(harness: ConsumerHarness, *, tenant_id: str, label: str | None) -> EndUserConnection:
+    return replace(
+        new_connection(),
+        account_id=harness.identity.account_id,
+        tenant_id=tenant_id,
+        credential_type=ConnectionCredentialType.OAUTH2,
+        oauth_client_source=OAuthClientSource.WORKSPACE,
+        account_label=label,
+    )
+
+
+def _selection(connection: EndUserConnection, base: ConnectionHarness, *, node_id: str) -> AppConnection:
+    return AppConnection(
+        tenant_id=base.context.tenant_id,
+        app_id=base.context.app_id,
+        workflow_id=base.workflow_id,
+        end_user_id=base.context.end_user_id,
+        node_id=node_id,
+        provider=connection.provider,
+        connection_id=connection.id,
+    )
+
+
+def test_groups_merge_across_tenants_and_deduplicate_accessible_webapps(harness: ConsumerHarness) -> None:
+    base = harness.oauth.base
+    other = make_harness(base.sessions)
+    first = _oauth_row(harness, tenant_id=base.context.tenant_id, label="al***@example.com")
+    duplicate = replace(first, name="Another name")
+    cross_tenant = replace(first, tenant_id=other.context.tenant_id)
+    foreign = replace(first, account_id=str(uuid4()))
+    guest = replace(first, account_id=None, end_user_id=str(uuid4()))
+    provider = replace(first, provider="vendor/other/calendar")
+    without_label = [replace(first, account_label=label) for label in (None, "")]
+    api_key = replace(new_connection(), account_id=harness.identity.account_id)
+    with base.sessions.begin() as session:
+        session.add_all([first, duplicate, cross_tenant, foreign, guest, provider, *without_label, api_key])
+        session.add(
+            Site(
+                app_id=other.context.app_id,
+                title="Other App",
+                default_language="en-US",
+                customize_token_strategy=CustomizeTokenStrategy.NOT_ALLOW,
+                code=str(uuid4()),
+            )
+        )
+        session.add_all(
+            [
+                _selection(first, base, node_id="one"),
+                _selection(duplicate, base, node_id="two"),
+                _selection(cross_tenant, other, node_id="one"),
+                SessionConnection(
+                    tenant_id=base.context.tenant_id,
+                    app_id=base.context.app_id,
+                    workflow_id=base.workflow_id,
+                    node_id="one",
+                    session_id=str(uuid4()),
+                    provider=first.provider,
+                    connection_id=first.id,
+                ),
+            ]
+        )
+
+    page = harness.service.list_connections(harness.identity, provider=None, cursor=None, limit=20)
+    assert len(page.data) == 5
+    representative = min((first, duplicate, cross_tenant), key=lambda row: row.id)
+    key = consumer_connection_key(representative)
+    group = next(item for item in page.data if item.key == key)
+    assert group.webapps_count == 2
+    assert group.name == representative.name
+    assert group.connection_type == ConnectionCredentialType.OAUTH2
+    assert group.account_label == first.account_label
+    for connection in without_label:
+        assert consumer_connection_key(connection) in {item.key for item in page.data}
+        assert base.repository.account_connection_group(harness.identity, connection.id).connection_id is not None
+    assert all(len(str(item.key)) == 36 for item in page.data)
+
+    app_ids = sorted((base.context.app_id, other.context.app_id))
+    apps = harness.service.list_connection_apps(harness.identity, key, cursor=None, limit=1)
+    assert [app.id for app in apps.data] == app_ids[:1]
+    assert apps.has_more
+    rest = harness.service.list_connection_apps(harness.identity, key, cursor=apps.next_cursor, limit=1)
+    assert [app.id for app in rest.data] == app_ids[1:]
+    assert not rest.has_more
+
+    seen = []
+    cursor = None
+    for _ in range(5):
+        current = harness.service.list_connections(harness.identity, provider=None, cursor=cursor, limit=1)
+        seen.extend(current.data)
+        cursor = current.next_cursor
+        if not current.has_more:
+            break
+    assert len(seen) == len({item.key for item in seen}) == 5
+    assert next(item for item in seen if item.key == key).webapps_count == 2
+    assert cursor is None
+
+    harness.access.denied.add(other.context.app_id)
+    page = harness.service.list_connections(harness.identity, provider=first.provider, cursor=None, limit=20)
+    assert next(item for item in page.data if item.key == key).webapps_count == 1
+    assert all(item.provider == first.provider for item in page.data)
+    assert [
+        app.id for app in harness.service.list_connection_apps(harness.identity, key, cursor=None, limit=20).data
+    ] == [base.context.app_id]
+    with base.sessions.begin() as session:
+        session.execute(sa.update(App).where(App.id == base.context.app_id).values(enable_site=False))
+    assert not harness.service.list_connection_apps(harness.identity, key, cursor=None, limit=20).data
+
+
+def test_group_disconnect_removes_all_members_and_revokes_each_owned_oauth_grant(harness: ConsumerHarness) -> None:
+    base = harness.oauth.base
+    created = harness.oauth.finish(harness.oauth.start())
+    with base.sessions.begin() as session:
+        original = session.get(EndUserConnection, created.id)
+        assert original is not None
+        duplicate = replace(original, name="Second Connection")
+        foreign = replace(original, account_id=str(uuid4()))
+        guest = replace(original, account_id=None, end_user_id=str(uuid4()))
+        session.add_all([duplicate, foreign, guest])
+    key = consumer_connection_key(created)
+    page = harness.service.list_connections(harness.identity, provider=None, cursor=None, limit=20)
+    assert [str(item.key) for item in page.data] == [min(created.id, duplicate.id)]
+    harness.service.delete_connection(harness.identity, consumer_connection_key(duplicate))
+    with base.sessions() as session:
+        assert set(session.scalars(sa.select(EndUserConnection.id))) == {foreign.id, guest.id}
+    assert len(base.auth.revocations) == 2
+    assert not harness.service.list_connections(harness.identity, provider=None, cursor=None, limit=20).data
+    with pytest.raises(ConnectionNotFoundError):
+        harness.service.list_connection_apps(harness.identity, key, cursor=None, limit=20)
+
+
+def test_any_owned_member_resolves_the_group_and_refresh_replaces_a_deleted_key(harness: ConsumerHarness) -> None:
+    base = harness.oauth.base
+    members = [_oauth_row(harness, tenant_id=base.context.tenant_id, label="al***") for _ in range(2)]
+    members.sort(key=lambda row: row.id)
+    with base.sessions.begin() as session:
+        session.add_all(members)
+    key = consumer_connection_key(members[0])
+    assert (
+        harness.service.list_connection_apps(
+            harness.identity, consumer_connection_key(members[1]), cursor=None, limit=20
+        ).data
+        == ()
+    )
+    with base.sessions.begin() as session:
+        session.execute(sa.delete(EndUserConnection).where(EndUserConnection.id == min(row.id for row in members)))
+    page = harness.service.list_connections(harness.identity, provider=None, cursor=None, limit=20)
+    assert [item.key for item in page.data] == [consumer_connection_key(members[1])]
+    with pytest.raises(ConnectionNotFoundError):
+        harness.service.delete_connection(harness.identity, key)
+    cursor = base.repository.account_connection_group(harness.identity, members[1].id).encode()
+    with pytest.raises(ValueError, match="cursor"):
+        harness.service.list_connections(harness.identity, provider="vendor/other/calendar", cursor=cursor, limit=20)
+
+
+@pytest.mark.parametrize("key", ["invalid", "W10", "bnVsbA", "e30", "Li4vLi4v", ""])
+def test_malformed_group_cursors_cannot_query_connections(harness: ConsumerHarness, key: str) -> None:
+    with pytest.raises(ValueError, match="group key"):
+        harness.service.list_connections(harness.identity, provider=None, cursor=key, limit=20)
diff --git a/api/tests/unit_tests/services/test_consumer_service.py b/api/tests/unit_tests/services/test_consumer_service.py
new file mode 100644
index 00000000000000..e1a9d20946eae1
--- /dev/null
+++ b/api/tests/unit_tests/services/test_consumer_service.py
@@ -0,0 +1,312 @@
+"""Consumer identity, ownership, usage and access-policy behavior with real persistence."""
+
+from dataclasses import replace
+from datetime import datetime, timedelta
+from uuid import uuid4
+
+import pytest
+import sqlalchemy as sa
+from sqlalchemy.orm import Session, sessionmaker
+
+from models.account import Account, AccountStatus, TenantAccountJoin
+from models.end_user_connection import AppConnection, EndUserConnection
+from models.enums import CreatorUserRole, CustomizeTokenStrategy, EndUserType, WorkflowRunTriggeredFrom
+from models.model import App, AppMode, AppStar, EndUser, Site
+from models.workflow import WorkflowRun, WorkflowType
+from repositories.web_passport_repository import WebPassportRepository
+from services.consumer_service import (
+    ConsumerAppNotFoundError,
+    ConsumerAppsQuery,
+    ConsumerIdentity,
+)
+from services.end_user_connections.entities import AccountConnectionContext
+from services.end_user_connections.errors import ConnectionNotFoundError
+from services.web_identity_service import WebIdentityError
+from tests.helpers.consumer import ConsumerHarness, consumer_connection_key, make_consumer_harness, new_consumer_run
+from tests.helpers.end_user_api_key import make_harness
+from tests.helpers.end_user_connection_contract import new_connection
+
+
+@pytest.fixture
+def harness(sqlite_session_factory: sessionmaker[Session], monkeypatch: pytest.MonkeyPatch) -> ConsumerHarness:
+    return make_consumer_harness(sqlite_session_factory, monkeypatch)
+
+
+def test_account_authentication_does_not_provision_workspace(harness: ConsumerHarness) -> None:
+    assert harness.service.authenticate(harness.token()) == harness.identity
+    with harness.oauth.base.sessions() as session:
+        assert session.scalar(sa.select(sa.func.count()).select_from(TenantAccountJoin)) == 0
+
+
+@pytest.mark.parametrize("source", [None, "webapp", "webapp_account", "webapp_login_token", "unknown"])
+def test_app_tokens_cannot_become_consumer_identity(harness: ConsumerHarness, source: str) -> None:
+    with pytest.raises(WebIdentityError):
+        harness.service.authenticate(harness.token(source=source))
+
+
+def test_banned_account_cannot_authenticate_or_continue_connection_management(harness: ConsumerHarness) -> None:
+    base = harness.oauth.base
+    connection = base.service.create_api_key(base.context, base.command())
+    context = harness.service.connection_context(harness.identity, connection.id)
+    with base.sessions.begin() as session:
+        session.execute(
+            sa.update(Account).where(Account.id == harness.identity.account_id).values(status=AccountStatus.BANNED)
+        )
+    with pytest.raises(WebIdentityError):
+        harness.service.authenticate(harness.token())
+    from services.end_user_connections.errors import ConnectionIdentityError
+
+    with pytest.raises(ConnectionIdentityError):
+        base.service.delete_connection(context, connection.id)
+
+
+@pytest.mark.parametrize("mode", [AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
+def test_favorites_and_recent_are_account_scoped_and_recheck_access(harness: ConsumerHarness, mode: AppMode) -> None:
+    base = harness.oauth.base
+    app_id = base.context.app_id
+    for _ in range(2):
+        harness.service.set_favorite(harness.identity, app_id, enabled=True)
+    now = datetime(2026, 10, 30, 10)
+    with base.sessions.begin() as session:
+        session.execute(sa.update(App).where(App.id == app_id).values(mode=mode))
+        for moment in (now, now - timedelta(days=1), now + timedelta(hours=1)):
+            run = new_consumer_run(base, created_at=moment)
+            run.type = WorkflowType.CHAT if mode == AppMode.ADVANCED_CHAT else WorkflowType.WORKFLOW
+            session.add(run)
+    for view in ("recent", "favorites", "shared"):
+        result = harness.service.list_apps(harness.identity, ConsumerAppsQuery(view=view))
+        assert [app.id for app in result.data] == [app_id]
+        assert result.data[0].is_favorite
+        assert result.data[0].last_used_at == now + timedelta(hours=1)
+    other = ConsumerIdentity(str(uuid4()))
+    assert not harness.service.list_apps(other, ConsumerAppsQuery()).data
+    assert not harness.service.list_apps(other, ConsumerAppsQuery(view="favorites")).data
+    with base.sessions() as session:
+        assert session.scalar(sa.select(sa.func.count()).select_from(AppStar)) == 1
+    harness.access.denied.add(app_id)
+    assert not harness.service.list_apps(harness.identity, ConsumerAppsQuery(view="favorites")).data
+    with pytest.raises(ConsumerAppNotFoundError):
+        harness.service.set_favorite(harness.identity, app_id, enabled=True)
+    for _ in range(2):
+        harness.service.set_favorite(harness.identity, app_id, enabled=False)
+
+
+def test_unpublished_app_cannot_be_favorited_or_listed(harness: ConsumerHarness) -> None:
+    base = harness.oauth.base
+    with base.sessions.begin() as session:
+        session.execute(sa.update(App).where(App.id == base.context.app_id).values(enable_site=False))
+    with pytest.raises(ConsumerAppNotFoundError):
+        harness.service.set_favorite(harness.identity, base.context.app_id, enabled=True)
+    assert not harness.service.list_apps(harness.identity, ConsumerAppsQuery(view="shared")).data
+
+
+@pytest.mark.parametrize("mode", [AppMode.CHAT, AppMode.COMPLETION, AppMode.AGENT_CHAT, AppMode.AGENT])
+def test_consumer_apps_support_only_chatflow_and_workflow(harness: ConsumerHarness, mode: AppMode) -> None:
+    base = harness.oauth.base
+    harness.service.set_favorite(harness.identity, base.context.app_id, enabled=True)
+    with base.sessions.begin() as session:
+        session.add(new_consumer_run(base, created_at=datetime(2026, 10, 30)))
+        session.execute(sa.update(App).where(App.id == base.context.app_id).values(mode=mode))
+    for view in ("recent", "favorites", "shared"):
+        assert not harness.service.list_apps(harness.identity, ConsumerAppsQuery(view=view)).data
+    with pytest.raises(ConsumerAppNotFoundError):
+        harness.service.set_favorite(harness.identity, base.context.app_id, enabled=True)
+
+
+def test_passport_without_execution_is_not_recent_and_log_deletion_removes_recent_usage(
+    harness: ConsumerHarness,
+) -> None:
+    base = harness.oauth.base
+    harness.service.set_favorite(harness.identity, base.context.app_id, enabled=True)
+    assert not harness.service.list_apps(harness.identity, ConsumerAppsQuery()).data
+    favorites = harness.service.list_apps(harness.identity, ConsumerAppsQuery(view="favorites"))
+    assert favorites.data[0].last_used_at is None
+    with base.sessions.begin() as session:
+        run = new_consumer_run(base, created_at=datetime(2026, 10, 30))
+        session.add(run)
+        session.flush()
+        run_id = run.id
+    assert harness.service.list_apps(harness.identity, ConsumerAppsQuery()).data[0].last_used_at == run.created_at
+    with base.sessions.begin() as session:
+        session.execute(sa.delete(WorkflowRun).where(WorkflowRun.id == run_id))
+    assert not harness.service.list_apps(harness.identity, ConsumerAppsQuery()).data
+    assert harness.service.list_apps(harness.identity, ConsumerAppsQuery(view="favorites")).data[0].last_used_at is None
+
+
+@pytest.mark.parametrize("mismatch", ["tenant", "app", "role", "source", "account", "guest", "type"])
+def test_recent_usage_requires_the_complete_passport_execution_owner_chain(
+    harness: ConsumerHarness, mismatch: str
+) -> None:
+    base = harness.oauth.base
+    now = datetime(2026, 10, 30)
+    with base.sessions.begin() as session:
+        session.add(new_consumer_run(base, created_at=now))
+        unrelated = new_consumer_run(base, created_at=now + timedelta(days=1))
+        if mismatch == "tenant":
+            unrelated.tenant_id = str(uuid4())
+        elif mismatch == "app":
+            unrelated.app_id = str(uuid4())
+        elif mismatch == "role":
+            unrelated.created_by_role = CreatorUserRole.ACCOUNT
+        elif mismatch == "source":
+            unrelated.triggered_from = WorkflowRunTriggeredFrom.DEBUGGING
+        else:
+            account_id = None
+            if mismatch == "type":
+                account_id = harness.identity.account_id
+            elif mismatch == "account":
+                account_id = str(uuid4())
+            user = EndUser(
+                tenant_id=base.context.tenant_id,
+                app_id=base.context.app_id,
+                type=EndUserType.SERVICE_API if mismatch == "type" else EndUserType.BROWSER,
+                account_id=account_id,
+                session_id=str(uuid4()),
+            )
+            session.add(user)
+            unrelated.created_by = user.id
+        session.add(unrelated)
+    result = harness.service.list_apps(harness.identity, ConsumerAppsQuery())
+    assert [app.id for app in result.data] == [base.context.app_id]
+    assert result.data[0].last_used_at == now
+
+
+def test_connection_queries_do_not_decrypt_or_expose_foreign_and_guest_credentials(harness: ConsumerHarness) -> None:
+    base = harness.oauth.base
+    owned = base.service.create_api_key(base.context, base.command())
+    foreign = replace(new_connection(), tenant_id=base.context.tenant_id, account_id=str(uuid4()))
+    guest = replace(new_connection(), tenant_id=base.context.tenant_id, account_id=None, end_user_id=str(uuid4()))
+    with base.sessions.begin() as session:
+        session.add_all([foreign, guest])
+        session.add(
+            AppConnection(
+                tenant_id=base.context.tenant_id,
+                app_id=base.context.app_id,
+                workflow_id=base.workflow_id,
+                node_id="tool-1",
+                end_user_id=base.context.end_user_id,
+                provider=owned.provider,
+                connection_id=owned.id,
+            )
+        )
+    result = harness.service.list_connections(harness.identity, provider=None, cursor=None, limit=20)
+    assert [entry.key for entry in result.data] == [consumer_connection_key(owned)]
+    assert result.data[0].webapps_count == 1
+    apps = harness.service.list_connection_apps(harness.identity, result.data[0].key, cursor=None, limit=20)
+    assert [app.id for app in apps.data] == [base.context.app_id]
+    assert "original-secret" not in result.model_dump_json()
+    for connection in (foreign, guest, new_connection()):
+        with pytest.raises(ConnectionNotFoundError):
+            harness.service.delete_connection(harness.identity, consumer_connection_key(connection))
+        with pytest.raises(ConnectionNotFoundError):
+            harness.service.list_connection_apps(
+                harness.identity, consumer_connection_key(connection), cursor=None, limit=20
+            )
+    harness.access.denied.add(base.context.app_id)
+    result = harness.service.list_connections(harness.identity, provider=None, cursor=None, limit=20)
+    assert result.data[0].webapps_count == 0
+    assert not harness.service.list_connection_apps(harness.identity, result.data[0].key, cursor=None, limit=20).data
+
+
+def test_orphan_connection_can_be_disconnected_without_app(harness: ConsumerHarness) -> None:
+    base = harness.oauth.base
+    owned = base.service.create_api_key(base.context, base.command())
+    context = harness.service.connection_context(harness.identity, owned.id)
+    assert isinstance(context, AccountConnectionContext)
+    with base.sessions.begin() as session:
+        session.execute(sa.delete(EndUser).where(EndUser.id == base.context.end_user_id))
+        session.execute(sa.delete(App).where(App.id == base.context.app_id))
+    result = harness.service.list_connections(harness.identity, provider=None, cursor=None, limit=20)
+    assert [entry.key for entry in result.data] == [consumer_connection_key(owned)]
+    harness.service.delete_connection(harness.identity, consumer_connection_key(owned))
+    with base.sessions() as session:
+        assert session.get(EndUserConnection, owned.id) is None
+
+
+def test_disconnect_rejects_foreign_connections_without_calling_revoke(harness: ConsumerHarness) -> None:
+    base = harness.oauth.base
+    with base.sessions.begin() as session:
+        foreign = replace(new_connection(), tenant_id=base.context.tenant_id, account_id=str(uuid4()))
+        session.add(foreign)
+    with pytest.raises(ConnectionNotFoundError):
+        harness.service.delete_connection(harness.identity, consumer_connection_key(foreign))
+    assert base.auth.revocations == []
+
+
+def test_disconnect_oauth_connection_revokes_original_grant_without_app(harness: ConsumerHarness) -> None:
+    base = harness.oauth.base
+    connection = harness.oauth.finish(harness.oauth.start())
+    with base.sessions.begin() as session:
+        session.execute(sa.delete(App).where(App.id == base.context.app_id))
+        session.execute(sa.delete(EndUser).where(EndUser.id == base.context.end_user_id))
+    harness.service.delete_connection(harness.identity, consumer_connection_key(connection))
+    with base.sessions() as session:
+        assert session.get(EndUserConnection, connection.id) is None
+    assert len(base.auth.revocations) == 1
+    assert base.auth.revocations[0][1] == harness.identity.account_id
+
+
+@pytest.mark.parametrize("same_time", [False, True])
+def test_recent_pagination_fills_pages_after_denials_and_handles_equal_timestamps(
+    harness: ConsumerHarness, same_time: bool
+) -> None:
+    base = harness.oauth.base
+    bases = [base]
+    for _ in range(5):
+        other = make_harness(base.sessions)
+        bases.append(other)
+        with base.sessions.begin() as session:
+            session.add(
+                Site(
+                    app_id=other.context.app_id,
+                    title="Another App",
+                    default_language="en-US",
+                    customize_token_strategy=CustomizeTokenStrategy.NOT_ALLOW,
+                    code=str(uuid4()),
+                )
+            )
+    passports = WebPassportRepository(session_factory=base.sessions, generate_session_id=lambda: str(uuid4()))
+    moment = datetime(2026, 10, 30)
+    used_at_by_app: dict[str, datetime] = {}
+    for index, item in enumerate(bases):
+        with base.sessions() as session:
+            code = session.scalar(sa.select(Site.code).where(Site.app_id == item.context.app_id))
+            assert code is not None
+        app = passports.get_active_web_app(code)
+        assert app is not None
+        resolution = passports.resolve_account_end_user(app, harness.identity.account_id)
+        assert resolution.end_user is not None
+        with base.sessions.begin() as session:
+            used_at = moment if same_time else moment + timedelta(minutes=index)
+            used_at_by_app[item.context.app_id] = used_at
+            run = new_consumer_run(item, created_at=used_at)
+            run.created_by = resolution.end_user.id
+            session.add(run)
+    app_ids = [item.context.app_id for item in bases]
+    app_ids.sort(key=lambda app_id: (-used_at_by_app[app_id].timestamp(), app_id))
+    harness.access.denied.update(app_ids[:2])
+    first = harness.service.list_apps(harness.identity, ConsumerAppsQuery(limit=2))
+    assert [app.id for app in first.data] == app_ids[2:4]
+    assert first.has_more
+    second = harness.service.list_apps(harness.identity, ConsumerAppsQuery(limit=2, cursor=first.next_cursor))
+    assert [app.id for app in second.data] == app_ids[4:]
+    assert not second.has_more
+    assert second.next_cursor is None
+    with pytest.raises(ValueError, match="cursor"):
+        harness.service.list_apps(ConsumerIdentity(str(uuid4())), ConsumerAppsQuery(cursor=first.next_cursor))
+    with pytest.raises(ValueError, match="cursor"):
+        harness.service.list_apps(harness.identity, ConsumerAppsQuery(view="favorites", cursor=first.next_cursor))
+
+
+def test_connections_paginate_across_tenants_without_merging_provider_names(harness: ConsumerHarness) -> None:
+    base = harness.oauth.base
+    rows = [replace(new_connection(), tenant_id=str(uuid4()), account_id=harness.identity.account_id) for _ in range(3)]
+    with base.sessions.begin() as session:
+        session.add_all(rows)
+    first = harness.service.list_connections(harness.identity, provider=None, cursor=None, limit=2)
+    second = harness.service.list_connections(harness.identity, provider=None, cursor=first.next_cursor, limit=2)
+    assert len(first.data) == 2
+    assert first.has_more
+    assert len(second.data) == 1
+    assert not second.has_more
+    assert {row.key for row in (*first.data, *second.data)} == {consumer_connection_key(row) for row in rows}
diff --git a/packages/contracts/generated/api/web/errors.gen.ts b/packages/contracts/generated/api/web/errors.gen.ts
index b46c1e2316074a..4b509f6f3deec4 100644
--- a/packages/contracts/generated/api/web/errors.gen.ts
+++ b/packages/contracts/generated/api/web/errors.gen.ts
@@ -794,7 +794,14 @@ export const post20Errors = {
   authentication_failed: { status: 401, data: zAuthenticationFailedError },
 }
 
-export const get12Errors = {
+export const get13Errors = { connection_not_found: { status: 404, data: zConnectionNotFoundError } }
+
+export const delete4Errors = {
+  connection_conflict: { status: 409, data: zConnectionConflictError },
+  connection_not_found: { status: 404, data: zConnectionNotFoundError },
+}
+
+export const get15Errors = {
   app_more_like_this_disabled: { status: 403, data: zAppMoreLikeThisDisabledError },
   completion_request_error: { status: 400, data: zCompletionRequestError },
   model_currently_not_support: { status: 400, data: zModelCurrentlyNotSupportError },
@@ -803,7 +810,7 @@ export const get12Errors = {
   provider_quota_exceeded: { status: 400, data: zProviderQuotaExceededError },
 }
 
-export const get13Errors = {
+export const get16Errors = {
   app_suggested_questions_after_answer_disabled: {
     status: 403,
     data: zAppSuggestedQuestionsAfterAnswerDisabledError,
@@ -815,9 +822,9 @@ export const get13Errors = {
   provider_quota_exceeded: { status: 400, data: zProviderQuotaExceededError },
 }
 
-export const get14Errors = { not_chat_app: { status: 400, data: zNotChatAppError } }
+export const get17Errors = { not_chat_app: { status: 400, data: zNotChatAppError } }
 
-export const get15Errors = { app_unavailable: { status: 400, data: zAppUnavailableError } }
+export const get18Errors = { app_unavailable: { status: 400, data: zAppUnavailableError } }
 
 export const post23Errors = {
   connection_claim_preflight_failed: { status: 400, data: zConnectionClaimPreflightFailedError },
@@ -825,12 +832,12 @@ export const post23Errors = {
   end_user_login_required: { status: 401, data: zEndUserLoginRequiredError },
 }
 
-export const get16Errors = {
+export const get19Errors = {
   agent_not_published: { status: 400, data: zAgentNotPublishedError },
   app_unavailable: { status: 400, data: zAppUnavailableError },
 }
 
-export const get17Errors = {
+export const get20Errors = {
   end_user_login_required: { status: 401, data: zEndUserLoginRequiredError },
 }
 
@@ -851,7 +858,7 @@ export const post26Errors = {
   unsupported_file_type: { status: 415, data: zUnsupportedFileTypeError },
 }
 
-export const get18Errors = {
+export const get21Errors = {
   remote_file_access_denied: { status: 400, data: zRemoteFileAccessDeniedError },
   remote_file_invalid_response: { status: 502, data: zRemoteFileInvalidResponseError },
   remote_file_invalid_url: { status: 400, data: zRemoteFileInvalidUrlError },
@@ -860,9 +867,9 @@ export const get18Errors = {
   remote_file_url_blocked: { status: 400, data: zRemoteFileUrlBlockedError },
 }
 
-export const delete3Errors = { not_completion_app: { status: 400, data: zNotCompletionAppError } }
+export const delete5Errors = { not_completion_app: { status: 400, data: zNotCompletionAppError } }
 
-export const get19Errors = { not_completion_app: { status: 400, data: zNotCompletionAppError } }
+export const get22Errors = { not_completion_app: { status: 400, data: zNotCompletionAppError } }
 
 export const post27Errors = { not_completion_app: { status: 400, data: zNotCompletionAppError } }
 
@@ -874,9 +881,9 @@ export const post28Errors = {
   provider_quota_exceeded: { status: 400, data: zProviderQuotaExceededError },
 }
 
-export const get22Errors = { invalid_param: { status: 400, data: zInvalidParamError } }
+export const get25Errors = { invalid_param: { status: 400, data: zInvalidParamError } }
 
-export const get24Errors = { invalid_param: { status: 400, data: zInvalidParamError } }
+export const get27Errors = { invalid_param: { status: 400, data: zInvalidParamError } }
 
 export const post29Errors = {
   completion_request_error: { status: 400, data: zCompletionRequestError },
diff --git a/packages/contracts/generated/api/web/orpc.gen.ts b/packages/contracts/generated/api/web/orpc.gen.ts
index 0fc57ca067f66e..e67103a281d563 100644
--- a/packages/contracts/generated/api/web/orpc.gen.ts
+++ b/packages/contracts/generated/api/web/orpc.gen.ts
@@ -70,6 +70,10 @@ import {
   zDeleteConnectionsByConnectionIdResponse,
   zDeleteConversationsByCIdPath,
   zDeleteConversationsByCIdResponse,
+  zDeleteMeAppsByAppIdFavoritePath,
+  zDeleteMeAppsByAppIdFavoriteResponse,
+  zDeleteMeConnectionsByConnectionKeyPath,
+  zDeleteMeConnectionsByConnectionKeyResponse,
   zDeleteSavedMessagesByMessageIdPath,
   zDeleteSavedMessagesByMessageIdResponse,
   zGetAccountProfileResponse,
@@ -93,6 +97,13 @@ import {
   zGetFormHumanInputByFormTokenResponse,
   zGetLoginStatusQuery,
   zGetLoginStatusResponse,
+  zGetMeAppsQuery,
+  zGetMeAppsResponse,
+  zGetMeConnectionsByConnectionKeyAppsPath,
+  zGetMeConnectionsByConnectionKeyAppsQuery,
+  zGetMeConnectionsByConnectionKeyAppsResponse,
+  zGetMeConnectionsQuery,
+  zGetMeConnectionsResponse,
   zGetMessagesByMessageIdMoreLikeThisPath,
   zGetMessagesByMessageIdMoreLikeThisQuery,
   zGetMessagesByMessageIdMoreLikeThisResponse,
@@ -189,6 +200,8 @@ import {
   zPostWorkflowsTasksByTaskIdStopResponse,
   zPutConnectionsConfigurationBody,
   zPutConnectionsConfigurationResponse,
+  zPutMeAppsByAppIdFavoritePath,
+  zPutMeAppsByAppIdFavoriteResponse,
 } from './zod.gen.ts'
 
 /**
@@ -1250,6 +1263,138 @@ export const logout = {
   post: post21,
 }
 
+export const delete3 = oc
+  .route({
+    inputStructure: 'detailed',
+    method: 'DELETE',
+    operationId: 'deleteMeAppsByAppIdFavorite',
+    path: '/me/apps/{app_id}/favorite',
+    successStatus: 204,
+    tags: ['web'],
+  })
+  .input(z.object({ params: zDeleteMeAppsByAppIdFavoritePath }))
+  .output(zDeleteMeAppsByAppIdFavoriteResponse)
+
+export const put2 = oc
+  .route({
+    inputStructure: 'detailed',
+    method: 'PUT',
+    operationId: 'putMeAppsByAppIdFavorite',
+    path: '/me/apps/{app_id}/favorite',
+    successStatus: 204,
+    tags: ['web'],
+  })
+  .input(z.object({ params: zPutMeAppsByAppIdFavoritePath }))
+  .output(zPutMeAppsByAppIdFavoriteResponse)
+
+export const favorite = {
+  delete: delete3,
+  put: put2,
+}
+
+export const byAppId = {
+  favorite,
+}
+
+/**
+ * List accessible Chatflow and Workflow Apps using the WebApp Account login cookie. Recent use is derived from retained published execution logs. No App Passport or workspace membership is required.
+ */
+export const get12 = oc
+  .route({
+    description:
+      'List accessible Chatflow and Workflow Apps using the WebApp Account login cookie. Recent use is derived from retained published execution logs. No App Passport or workspace membership is required.',
+    inputStructure: 'detailed',
+    method: 'GET',
+    operationId: 'getMeApps',
+    path: '/me/apps',
+    tags: ['web'],
+  })
+  .input(z.object({ query: zGetMeAppsQuery.optional() }).optional())
+  .output(zGetMeAppsResponse)
+
+export const apps = {
+  get: get12,
+  byAppId,
+}
+
+/**
+ * List accessible WebApps associated with a Connection group returned by /me/connections.
+ */
+export const get13 = oc
+  .route({
+    description:
+      'List accessible WebApps associated with a Connection group returned by /me/connections.',
+    inputStructure: 'detailed',
+    method: 'GET',
+    operationId: 'getMeConnectionsByConnectionKeyApps',
+    path: '/me/connections/{connection_key}/apps',
+    tags: ['web'],
+  })
+  .input(
+    z.object({
+      params: zGetMeConnectionsByConnectionKeyAppsPath,
+      query: zGetMeConnectionsByConnectionKeyAppsQuery.optional(),
+    }),
+  )
+  .output(zGetMeConnectionsByConnectionKeyAppsResponse)
+  .errors({ connection_not_found: { status: 404, data: zConnectionNotFoundError } })
+
+export const apps2 = {
+  get: get13,
+}
+
+/**
+ * Disconnect all owned Connections matching a group key returned by /me/connections. Cookie-authenticated browser mutations require the configured WebApp Origin.
+ */
+export const delete4 = oc
+  .route({
+    description:
+      'Disconnect all owned Connections matching a group key returned by /me/connections. Cookie-authenticated browser mutations require the configured WebApp Origin.',
+    inputStructure: 'detailed',
+    method: 'DELETE',
+    operationId: 'deleteMeConnectionsByConnectionKey',
+    path: '/me/connections/{connection_key}',
+    successStatus: 204,
+    tags: ['web'],
+  })
+  .input(z.object({ params: zDeleteMeConnectionsByConnectionKeyPath }))
+  .output(zDeleteMeConnectionsByConnectionKeyResponse)
+  .errors({
+    connection_conflict: { status: 409, data: zConnectionConflictError },
+    connection_not_found: { status: 404, data: zConnectionNotFoundError },
+  })
+
+export const byConnectionKey = {
+  delete: delete4,
+  apps: apps2,
+}
+
+/**
+ * List personal Connections grouped by Provider, account label and Connection type. Connections without an account label form individual groups. WebApp counts include only accessible Apps.
+ */
+export const get14 = oc
+  .route({
+    description:
+      'List personal Connections grouped by Provider, account label and Connection type. Connections without an account label form individual groups. WebApp counts include only accessible Apps.',
+    inputStructure: 'detailed',
+    method: 'GET',
+    operationId: 'getMeConnections',
+    path: '/me/connections',
+    tags: ['web'],
+  })
+  .input(z.object({ query: zGetMeConnectionsQuery.optional() }).optional())
+  .output(zGetMeConnectionsResponse)
+
+export const connections2 = {
+  get: get14,
+  byConnectionKey,
+}
+
+export const me = {
+  apps,
+  connections: connections2,
+}
+
 /**
  * Submit feedback (like/dislike) for a specific message.
  */
@@ -1278,7 +1423,7 @@ export const feedbacks = {
 /**
  * Generate a new completion similar to an existing message (completion apps only).
  */
-export const get12 = oc
+export const get15 = oc
   .route({
     description: 'Generate a new completion similar to an existing message (completion apps only).',
     inputStructure: 'detailed',
@@ -1304,13 +1449,13 @@ export const get12 = oc
   })
 
 export const moreLikeThis = {
-  get: get12,
+  get: get15,
 }
 
 /**
  * Get suggested follow-up questions after a message (chat apps only).
  */
-export const get13 = oc
+export const get16 = oc
   .route({
     description: 'Get suggested follow-up questions after a message (chat apps only).',
     inputStructure: 'detailed',
@@ -1334,7 +1479,7 @@ export const get13 = oc
   })
 
 export const suggestedQuestions = {
-  get: get13,
+  get: get16,
 }
 
 export const byMessageId = {
@@ -1346,7 +1491,7 @@ export const byMessageId = {
 /**
  * Retrieve paginated list of messages from a conversation in a chat application.
  */
-export const get14 = oc
+export const get17 = oc
   .route({
     description: 'Retrieve paginated list of messages from a conversation in a chat application.',
     inputStructure: 'detailed',
@@ -1360,7 +1505,7 @@ export const get14 = oc
   .errors({ not_chat_app: { status: 400, data: zNotChatAppError } })
 
 export const messages = {
-  get: get14,
+  get: get17,
   byMessageId,
 }
 
@@ -1369,7 +1514,7 @@ export const messages = {
  *
  * Retrieve the metadata for a specific app.
  */
-export const get15 = oc
+export const get18 = oc
   .route({
     description: 'Retrieve the metadata for a specific app.',
     inputStructure: 'detailed',
@@ -1383,7 +1528,7 @@ export const get15 = oc
   .errors({ app_unavailable: { status: 400, data: zAppUnavailableError } })
 
 export const meta = {
-  get: get15,
+  get: get18,
 }
 
 /**
@@ -1441,7 +1586,7 @@ export const migration = {
  *
  * Retrieve the parameters for a specific app.
  */
-export const get16 = oc
+export const get19 = oc
   .route({
     description: 'Retrieve the parameters for a specific app.',
     inputStructure: 'detailed',
@@ -1458,13 +1603,13 @@ export const get16 = oc
   })
 
 export const parameters = {
-  get: get16,
+  get: get19,
 }
 
 /**
  * Get authentication passport for web application access
  */
-export const get17 = oc
+export const get20 = oc
   .route({
     description: 'Get authentication passport for web application access',
     inputStructure: 'detailed',
@@ -1498,7 +1643,7 @@ export const post25 = oc
   })
 
 export const passport = {
-  get: get17,
+  get: get20,
   post: post25,
 }
 
@@ -1578,7 +1723,7 @@ export const upload2 = {
  * Raises:
  * HTTPException: If the remote file cannot be accessed
  */
-export const get18 = oc
+export const get21 = oc
   .route({
     description:
       'Get information about a remote file\nRetrieves basic information about a file located at a remote URL,\nincluding content type and content length.\n\nArgs:\n    app_model: The associated application model\n    end_user: The end user making the request\n    url: URL-encoded path to the remote file\n\nReturns:\n    dict: Remote file information including type and length\n\nRaises:\n    HTTPException: If the remote file cannot be accessed',
@@ -1601,7 +1746,7 @@ export const get18 = oc
   })
 
 export const byUrl = {
-  get: get18,
+  get: get21,
 }
 
 export const remoteFiles = {
@@ -1612,7 +1757,7 @@ export const remoteFiles = {
 /**
  * Remove a message from saved messages.
  */
-export const delete3 = oc
+export const delete5 = oc
   .route({
     description: 'Remove a message from saved messages.',
     inputStructure: 'detailed',
@@ -1627,13 +1772,13 @@ export const delete3 = oc
   .errors({ not_completion_app: { status: 400, data: zNotCompletionAppError } })
 
 export const byMessageId2 = {
-  delete: delete3,
+  delete: delete5,
 }
 
 /**
  * Retrieve paginated list of saved messages for a completion application.
  */
-export const get19 = oc
+export const get22 = oc
   .route({
     description: 'Retrieve paginated list of saved messages for a completion application.',
     inputStructure: 'detailed',
@@ -1663,7 +1808,7 @@ export const post27 = oc
   .errors({ not_completion_app: { status: 400, data: zNotCompletionAppError } })
 
 export const savedMessages = {
-  get: get19,
+  get: get22,
   post: post27,
   byMessageId: byMessageId2,
 }
@@ -1673,7 +1818,7 @@ export const savedMessages = {
  *
  * Retrieve app site information and configuration.
  */
-export const get20 = oc
+export const get23 = oc
   .route({
     description: 'Retrieve app site information and configuration.',
     inputStructure: 'detailed',
@@ -1686,7 +1831,7 @@ export const get20 = oc
   .output(zGetSiteResponse)
 
 export const site = {
-  get: get20,
+  get: get23,
 }
 
 /**
@@ -1698,7 +1843,7 @@ export const site = {
  *
  * Authentication configuration must be available before the authentication flow can be selected.
  */
-export const get21 = oc
+export const get24 = oc
   .route({
     description:
       'Get the non-sensitive bootstrap snapshot exposed before Console or Web authentication. This is not a general feature registry.\nThis endpoint is akin to the `SystemFeatureApi` endpoint in api/controllers/console/feature.py,\nexcept it is intended for use by the web app, instead of the console dashboard.\n\nAuthentication configuration must be available before the authentication flow can be selected.',
@@ -1712,7 +1857,7 @@ export const get21 = oc
   .output(zGetSystemFeaturesResponse)
 
 export const systemFeatures = {
-  get: get21,
+  get: get24,
 }
 
 /**
@@ -1747,7 +1892,7 @@ export const textToAudio = {
 /**
  * Retrieve the access mode for a web application (public or restricted).
  */
-export const get22 = oc
+export const get25 = oc
   .route({
     description: 'Retrieve the access mode for a web application (public or restricted).',
     inputStructure: 'detailed',
@@ -1761,13 +1906,13 @@ export const get22 = oc
   .errors({ invalid_param: { status: 400, data: zInvalidParamError } })
 
 export const accessMode = {
-  get: get22,
+  get: get25,
 }
 
 /**
  * Check if user has permission to access a web application.
  */
-export const get23 = oc
+export const get26 = oc
   .route({
     description: 'Check if user has permission to access a web application.',
     inputStructure: 'detailed',
@@ -1780,7 +1925,7 @@ export const get23 = oc
   .output(zGetWebappPermissionResponse)
 
 export const permission = {
-  get: get23,
+  get: get26,
 }
 
 export const webapp = {
@@ -1795,7 +1940,7 @@ export const webapp = {
  *
  * Returns Server-Sent Events stream.
  */
-export const get24 = oc
+export const get27 = oc
   .route({
     description: 'GET /api/workflow/<task_id>/events\n\nReturns Server-Sent Events stream.',
     inputStructure: 'detailed',
@@ -1810,7 +1955,7 @@ export const get24 = oc
   .errors({ invalid_param: { status: 400, data: zInvalidParamError } })
 
 export const events = {
-  get: get24,
+  get: get27,
 }
 
 export const byTaskId3 = {
@@ -1921,6 +2066,7 @@ export const contract = {
   humanInputForms,
   login,
   logout,
+  me,
   messages,
   meta,
   migration,
diff --git a/packages/contracts/generated/api/web/types.gen.ts b/packages/contracts/generated/api/web/types.gen.ts
index e0ce6109fd79c4..8965725b970186 100644
--- a/packages/contracts/generated/api/web/types.gen.ts
+++ b/packages/contracts/generated/api/web/types.gen.ts
@@ -297,6 +297,40 @@ export type ConnectionRevisionChangedError = ErrorResponse & {
   status: 409
 }
 
+export type ConsumerAppResponse = {
+  app_code: string
+  description: string
+  icon: string | null
+  icon_background: string | null
+  icon_type: string | null
+  id: string
+  is_favorite: boolean
+  last_used_at: number | null
+  mode: string
+  name: string
+}
+
+export type ConsumerAppsResponse = {
+  data: Array<ConsumerAppResponse>
+  has_more: boolean
+  next_cursor: string | null
+}
+
+export type ConsumerConnectionResponse = {
+  account_label: string | null
+  connection_type: ConnectionCredentialType
+  key: string
+  name: string
+  provider: string
+  webapps_count: number
+}
+
+export type ConsumerConnectionsResponse = {
+  data: Array<ConsumerConnectionResponse>
+  has_more: boolean
+  next_cursor: string | null
+}
+
 export type ConversationInfiniteScrollPagination = {
   data: Array<SimpleConversation>
   has_more: boolean
@@ -2680,6 +2714,151 @@ export type PostLogoutResponses = {
 
 export type PostLogoutResponse = PostLogoutResponses[keyof PostLogoutResponses]
 
+export type GetMeAppsData = {
+  body?: never
+  path?: never
+  query?: {
+    cursor?: string
+    limit?: number
+    name?: string
+    view?: 'favorites' | 'recent' | 'shared'
+  }
+  url: '/me/apps'
+}
+
+export type GetMeAppsErrors = {
+  default: ErrorResponse
+}
+
+export type GetMeAppsError = GetMeAppsErrors[keyof GetMeAppsErrors]
+
+export type GetMeAppsResponses = {
+  200: ConsumerAppsResponse
+}
+
+export type GetMeAppsResponse = GetMeAppsResponses[keyof GetMeAppsResponses]
+
+export type DeleteMeAppsByAppIdFavoriteData = {
+  body?: never
+  path: {
+    app_id: string
+  }
+  query?: never
+  url: '/me/apps/{app_id}/favorite'
+}
+
+export type DeleteMeAppsByAppIdFavoriteErrors = {
+  default: ErrorResponse
+}
+
+export type DeleteMeAppsByAppIdFavoriteError =
+  DeleteMeAppsByAppIdFavoriteErrors[keyof DeleteMeAppsByAppIdFavoriteErrors]
+
+export type DeleteMeAppsByAppIdFavoriteResponses = {
+  204: void
+}
+
+export type DeleteMeAppsByAppIdFavoriteResponse =
+  DeleteMeAppsByAppIdFavoriteResponses[keyof DeleteMeAppsByAppIdFavoriteResponses]
+
+export type PutMeAppsByAppIdFavoriteData = {
+  body?: never
+  path: {
+    app_id: string
+  }
+  query?: never
+  url: '/me/apps/{app_id}/favorite'
+}
+
+export type PutMeAppsByAppIdFavoriteErrors = {
+  default: ErrorResponse
+}
+
+export type PutMeAppsByAppIdFavoriteError =
+  PutMeAppsByAppIdFavoriteErrors[keyof PutMeAppsByAppIdFavoriteErrors]
+
+export type PutMeAppsByAppIdFavoriteResponses = {
+  204: void
+}
+
+export type PutMeAppsByAppIdFavoriteResponse =
+  PutMeAppsByAppIdFavoriteResponses[keyof PutMeAppsByAppIdFavoriteResponses]
+
+export type GetMeConnectionsData = {
+  body?: never
+  path?: never
+  query?: {
+    cursor?: string
+    limit?: number
+    provider?: string
+  }
+  url: '/me/connections'
+}
+
+export type GetMeConnectionsErrors = {
+  default: ErrorResponse
+}
+
+export type GetMeConnectionsError = GetMeConnectionsErrors[keyof GetMeConnectionsErrors]
+
+export type GetMeConnectionsResponses = {
+  200: ConsumerConnectionsResponse
+}
+
+export type GetMeConnectionsResponse = GetMeConnectionsResponses[keyof GetMeConnectionsResponses]
+
+export type DeleteMeConnectionsByConnectionKeyData = {
+  body?: never
+  path: {
+    connection_key: string
+  }
+  query?: never
+  url: '/me/connections/{connection_key}'
+}
+
+export type DeleteMeConnectionsByConnectionKeyErrors = {
+  404: ConnectionNotFoundError | ErrorResponse
+  409: ConnectionConflictError | ErrorResponse
+  default: ErrorResponse
+}
+
+export type DeleteMeConnectionsByConnectionKeyError =
+  DeleteMeConnectionsByConnectionKeyErrors[keyof DeleteMeConnectionsByConnectionKeyErrors]
+
+export type DeleteMeConnectionsByConnectionKeyResponses = {
+  204: void
+}
+
+export type DeleteMeConnectionsByConnectionKeyResponse =
+  DeleteMeConnectionsByConnectionKeyResponses[keyof DeleteMeConnectionsByConnectionKeyResponses]
+
+export type GetMeConnectionsByConnectionKeyAppsData = {
+  body?: never
+  path: {
+    connection_key: string
+  }
+  query?: {
+    cursor?: string
+    limit?: number
+  }
+  url: '/me/connections/{connection_key}/apps'
+}
+
+export type GetMeConnectionsByConnectionKeyAppsErrors = {
+  404: ConnectionNotFoundError | ErrorResponse
+  default: ErrorResponse
+}
+
+export type GetMeConnectionsByConnectionKeyAppsError =
+  GetMeConnectionsByConnectionKeyAppsErrors[keyof GetMeConnectionsByConnectionKeyAppsErrors]
+
+export type GetMeConnectionsByConnectionKeyAppsResponses = {
+  200: ConsumerAppsResponse
+}
+
+export type GetMeConnectionsByConnectionKeyAppsResponse =
+  GetMeConnectionsByConnectionKeyAppsResponses[keyof GetMeConnectionsByConnectionKeyAppsResponses]
+
 export type GetMessagesData = {
   body?: never
   path?: never
diff --git a/packages/contracts/generated/api/web/zod.gen.ts b/packages/contracts/generated/api/web/zod.gen.ts
index f8dd2e15738e4b..22bf4d474eb53a 100644
--- a/packages/contracts/generated/api/web/zod.gen.ts
+++ b/packages/contracts/generated/api/web/zod.gen.ts
@@ -269,6 +269,52 @@ export const zConnectionRevisionChangeDetails = z.object({
   connection_revision: z.uuid().nullable(),
 })
 
+/**
+ * ConsumerAppResponse
+ */
+export const zConsumerAppResponse = z.object({
+  app_code: z.string(),
+  description: z.string(),
+  icon: z.string().nullable(),
+  icon_background: z.string().nullable(),
+  icon_type: z.string().nullable(),
+  id: z.string(),
+  is_favorite: z.boolean(),
+  last_used_at: z.int().nullable(),
+  mode: z.string(),
+  name: z.string(),
+})
+
+/**
+ * ConsumerAppsResponse
+ */
+export const zConsumerAppsResponse = z.object({
+  data: z.array(zConsumerAppResponse),
+  has_more: z.boolean(),
+  next_cursor: z.string().nullable(),
+})
+
+/**
+ * ConsumerConnectionResponse
+ */
+export const zConsumerConnectionResponse = z.object({
+  account_label: z.string().nullable(),
+  connection_type: zConnectionCredentialType,
+  key: z.uuid(),
+  name: z.string(),
+  provider: z.string(),
+  webapps_count: z.int().gte(0),
+})
+
+/**
+ * ConsumerConnectionsResponse
+ */
+export const zConsumerConnectionsResponse = z.object({
+  data: z.array(zConsumerConnectionResponse),
+  has_more: z.boolean(),
+  next_cursor: z.string().nullable(),
+})
+
 /**
  * ConversationListQuery
  */
@@ -2835,6 +2881,70 @@ export const zGetLoginStatusResponse = zLoginStatusResponse
  */
 export const zPostLogoutResponse = zSimpleResultResponse
 
+export const zGetMeAppsQuery = z.object({
+  cursor: z.string().max(2048).optional(),
+  limit: z.int().gte(1).lte(100).optional().default(20),
+  name: z.string().max(255).optional(),
+  view: z.enum(['favorites', 'recent', 'shared']).optional().default('recent'),
+})
+
+/**
+ * Accessible personal Apps
+ */
+export const zGetMeAppsResponse = zConsumerAppsResponse
+
+export const zDeleteMeAppsByAppIdFavoritePath = z.object({
+  app_id: z.uuid(),
+})
+
+/**
+ * App unfavorited
+ */
+export const zDeleteMeAppsByAppIdFavoriteResponse = z.void()
+
+export const zPutMeAppsByAppIdFavoritePath = z.object({
+  app_id: z.uuid(),
+})
+
+/**
+ * App favorited
+ */
+export const zPutMeAppsByAppIdFavoriteResponse = z.void()
+
+export const zGetMeConnectionsQuery = z.object({
+  cursor: z.string().min(1).max(1024).optional(),
+  limit: z.int().gte(1).lte(100).optional().default(20),
+  provider: z.string().min(1).max(255).optional(),
+})
+
+/**
+ * Account Connections
+ */
+export const zGetMeConnectionsResponse = zConsumerConnectionsResponse
+
+export const zDeleteMeConnectionsByConnectionKeyPath = z.object({
+  connection_key: z.uuid(),
+})
+
+/**
+ * Connection group disconnected
+ */
+export const zDeleteMeConnectionsByConnectionKeyResponse = z.void()
+
+export const zGetMeConnectionsByConnectionKeyAppsPath = z.object({
+  connection_key: z.uuid(),
+})
+
+export const zGetMeConnectionsByConnectionKeyAppsQuery = z.object({
+  cursor: z.uuid().optional(),
+  limit: z.int().gte(1).lte(100).optional().default(20),
+})
+
+/**
+ * Associated WebApps
+ */
+export const zGetMeConnectionsByConnectionKeyAppsResponse = zConsumerAppsResponse
+
 export const zGetMessagesQuery = z.object({
   conversation_id: z.string(),
   first_id: z.string().optional(),