feat(licensing): send Customer Center a usage report every 30 minutes

docker-compose.yml

@@ -84,6 +84,9 @@ services:
       AUTOMATIONS_SERVICE_KEY: ${AUTOMATION_SERVICE_KEY:-local-compose-service-key}
       AUTOMATION_WEBHOOK_SECRET: ${AUTOMATION_WEBHOOK_SECRET:-local-compose-webhook-secret}
       AUTOMATION_EVENT_FORWARDING_ENABLED: 'true'
+      # With a license key, the worker sends Customer Center a usage report every 30 minutes.
+      OPENHANDS_LICENSE_KEY: ${OPENHANDS_LICENSE_KEY:-}
+      LICENSE_SERVER_URL: ${LICENSE_SERVER_URL:-}
     # The first start pulls the agent server image, which can take many minutes.
     healthcheck:
       test: ['CMD', 'python', '-c', "import urllib.request; urllib.request.urlopen('http://localhost:3000/alive')"]

migrations/versions/207_add_instance_id_to_instance_settings.py

@@ -0,0 +1,27 @@
+"""Add instance_id to instance_settings, to identify the install in usage reports.
+
+Revision ID: 207
+Revises: 206
+Create Date: 2026-11-01
+"""
+
+from typing import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = '207'
+down_revision: str | None = '206'
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        'instance_settings',
+        sa.Column('instance_id', sa.Uuid(), nullable=True),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column('instance_settings', 'instance_id')

openhands/app_server/worker/scheduled_jobs.py

@@ -4,6 +4,9 @@
 enabled with ``OH_WORKER_SCHEDULED_JOBS_<JOB>=true``. Never enable a job on a
 database where its CronJob still runs, or both will run it.
 
+The exception is ``usage_report``, which has no CronJob: it runs whenever
+``OPENHANDS_LICENSE_KEY`` is set, and has no switch of its own.
+
 Schedules and time limits match the OpenHands-Cloud chart's CronJobs. A failed
 run isn't retried; the next scheduled run is the retry.
 """
@@ -35,7 +38,16 @@ class ScheduledJobsConfig(BaseModel):
     enrich_user_interaction: bool = False
 
     def enabled(self) -> list[str]:
-        return [name for name, on in self.model_dump().items() if on]
+        jobs = [name for name, on in self.model_dump().items() if on]
+        if _license_key_configured():
+            jobs.append('usage_report')
+        return jobs
+
+
+def _license_key_configured() -> bool:
+    from server.licensing.config import LicenseConfig
+
+    return LicenseConfig.from_env().license_key is not None
 
 
 # Each job imports its module when it runs: some modules build API clients or
@@ -88,6 +100,12 @@ async def enrich_user_interaction(timestamp: int) -> None:
     await enrich_user_interaction_data.main()
 
 
+async def usage_report(timestamp: int) -> None:
+    from server.licensing import usage_report
+
+    await usage_report.run()
+
+
 SCHEDULES: dict[str, tuple[Callable[..., Any], str]] = {
     'budget_maintenance': (budget_maintenance, '*/15 * * * *'),
     'maintenance_tasks': (maintenance_tasks, '0 6 * * *'),
@@ -99,6 +117,7 @@ async def enrich_user_interaction(timestamp: int) -> None:
     'resend_sync': (resend_sync, '0 1 * * *'),
     'install_gitlab_webhooks': (install_gitlab_webhooks, '* * * * *'),
     'enrich_user_interaction': (enrich_user_interaction, '*/10 * * * *'),
+    'usage_report': (usage_report, '*/30 * * * *'),
 }
 
 

server/licensing/usage_report.py

@@ -0,0 +1,225 @@
+"""The usage report this install sends Customer Center.
+
+The worker's ``usage_report`` job (``openhands/app_server/worker/scheduled_jobs.py``)
+runs ``run()`` every 30 minutes while ``OPENHANDS_LICENSE_KEY`` is set, whether
+or not license enforcement is on. Each report carries lifetime totals, not
+changes since the last report, so a missed or duplicate report loses nothing.
+A failed report is logged and not retried; the next run is the retry.
+"""
+
+from __future__ import annotations
+
+from typing import Callable, Literal
+
+import httpx
+from pydantic import BaseModel, Field
+from sqlalchemy import exists, func, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from openhands.app_server.config import get_global_config
+from openhands.app_server.integrations.service_types import ProviderType
+from openhands.app_server.sandbox.docker_sandbox_service import (
+    DockerSandboxServiceInjector,
+)
+from openhands.app_server.sandbox.e2b_sandbox_service import E2BSandboxServiceInjector
+from openhands.app_server.sandbox.k8s_agent_sandbox_service import (
+    K8sAgentSandboxServiceInjector,
+)
+from openhands.app_server.sandbox.process_sandbox_service import (
+    ProcessSandboxServiceInjector,
+)
+from openhands.app_server.sandbox.remote_sandbox_service import (
+    RemoteSandboxServiceInjector,
+)
+from openhands.app_server.sandbox.sandbox_service import SandboxServiceInjector
+from openhands.app_server.utils.logger import openhands_logger as logger
+from server.auth.constants import (
+    ENABLE_ENTERPRISE_SSO,
+    KEYCLOAK_CLIENT_ID,
+    is_oauth_v2_login_enabled,
+)
+from server.licensing.config import LicenseConfig
+from storage.auth_provider import AuthProvider
+from storage.auth_provider_details import (
+    AuthProviderDetails,
+    OAuthProviderDetails,
+    SamlProviderDetails,
+)
+from storage.database import a_session_maker
+from storage.instance_settings_store import InstanceSettingsStore
+from storage.org import Org
+from storage.stored_conversation_metadata import StoredConversationMetadata
+from storage.user import User
+
+USAGE_REPORTS_PATH = '/api/v1/usage-reports'
+REQUEST_TIMEOUT_SECONDS = 10.0
+
+SandboxStrategy = Literal['docker', 'process', 'remote', 'e2b', 'k8s-agent-sandbox']
+
+# Subclasses report as the strategy they extend.
+_SANDBOX_STRATEGIES: tuple[
+    tuple[type[SandboxServiceInjector], SandboxStrategy], ...
+] = (
+    (DockerSandboxServiceInjector, 'docker'),
+    (ProcessSandboxServiceInjector, 'process'),
+    (RemoteSandboxServiceInjector, 'remote'),
+    (E2BSandboxServiceInjector, 'e2b'),
+    (K8sAgentSandboxServiceInjector, 'k8s-agent-sandbox'),
+)
+
+
+class UsageReport(BaseModel):
+    """The body of ``POST /api/v1/usage-reports``."""
+
+    license_key: str
+    instance_id: str = Field(min_length=1, max_length=200)
+    conversations: int = Field(ge=0)
+    users: int = Field(ge=0)
+    # None until this install has a source for the count.
+    automations: int | None = Field(ge=0)
+    orgs: int = Field(ge=0)
+    sso_enabled: bool
+    personal_workspaces_enabled: bool
+    sandbox_strategy: SandboxStrategy
+
+
+async def run(
+    http_client_factory: Callable[[], httpx.AsyncClient] | None = None,
+) -> None:
+    """Collect and send one report. Does nothing without a license key."""
+    config = LicenseConfig.from_env()
+    if config.license_key is None:
+        return
+    report = await collect_usage_report(config.license_key)
+    if report is None:
+        return
+    await send_usage_report(
+        report,
+        config.license_server_url,
+        http_client_factory
+        or (lambda: httpx.AsyncClient(timeout=REQUEST_TIMEOUT_SECONDS)),
+    )
+
+
+async def collect_usage_report(license_key: str) -> UsageReport | None:
+    """This install's lifetime totals and settings, or None if they can't be reported."""
+    strategy = sandbox_strategy(get_global_config().sandbox)
+    if strategy is None:
+        logger.warning(
+            'usage_report_unknown_sandbox',
+            extra={'sandbox': type(get_global_config().sandbox).__name__},
+        )
+        return None
+    instance_id = await InstanceSettingsStore.get_or_create_instance_id()
+    async with a_session_maker() as session:
+        # Lifetime totals: soft-deleted conversations still count.
+        conversations = await session.scalar(
+            select(func.count()).select_from(StoredConversationMetadata)
+        )
+        users = await session.scalar(select(func.count()).select_from(User))
+        # A personal workspace shares its id with its user, so it is already
+        # counted in users.
+        orgs = await session.scalar(
+            select(func.count())
+            .select_from(Org)
+            .where(~exists().where(User.id == Org.id))
+        )
+        sso = await sso_enabled(session)
+    return UsageReport(
+        license_key=license_key,
+        instance_id=str(instance_id),
+        conversations=conversations or 0,
+        users=users or 0,
+        # TODO: count automations. They live in the automation service's own
+        # database, and that service has no global count, so this sends null
+        # ("not reported").
+        automations=None,
+        orgs=orgs or 0,
+        sso_enabled=sso,
+        personal_workspaces_enabled=await personal_workspaces_enabled(),
+        sandbox_strategy=strategy,
+    )
+
+
+async def send_usage_report(
+    report: UsageReport,
+    license_server_url: str,
+    http_client_factory: Callable[[], httpx.AsyncClient],
+) -> bool:
+    """POST the report. Logs and returns False on any failure; never raises."""
+    url = f'{license_server_url}{USAGE_REPORTS_PATH}'
+    try:
+        async with http_client_factory() as client:
+            response = await client.post(url, json=report.model_dump())
+    except httpx.HTTPError as e:
+        logger.warning(
+            'usage_report_failed',
+            extra={'error': f'{type(e).__name__}: {e}', 'url': url},
+        )
+        return False
+    if response.status_code == 204:
+        logger.info('usage_report_sent', extra={'instance_id': report.instance_id})
+        return True
+    if response.status_code == 401:
+        logger.warning('usage_report_failed: key not valid', extra={'url': url})
+    else:
+        logger.warning(
+            'usage_report_failed',
+            extra={
+                'status': response.status_code,
+                'body': response.text[:1000],
+                'url': url,
+            },
+        )
+    return False
+
+
+def sandbox_strategy(
+    injector: SandboxServiceInjector | None,
+) -> SandboxStrategy | None:
+    """The sandbox service in use, as Customer Center names it.
+
+    ``config_from_env`` picks it from ``OH_SANDBOX_KIND``, else from
+    ``RUNTIME`` (unset is Docker, ``local`` is the process sandbox).
+    """
+    for cls, strategy in _SANDBOX_STRATEGIES:
+        if isinstance(injector, cls):
+            return strategy
+    return None
+
+
+async def sso_enabled(session: AsyncSession) -> bool:
+    """Whether users can sign in through a single sign-on identity provider.
+
+    That is the Keycloak SAML provider (``ENABLE_ENTERPRISE_SSO``), or, with
+    OAuth v2 login, an enabled login provider that is SAML or a direct
+    OIDC/OAuth identity provider (Okta, Google, ...). Keycloak's own rows
+    and git providers don't count.
+    """
+    if ENABLE_ENTERPRISE_SSO:
+        return True
+    if not is_oauth_v2_login_enabled():
+        return False
+    providers = await session.scalars(
+        select(AuthProvider).where(
+            AuthProvider.enabled.is_(True), AuthProvider.is_idp.is_(True)
+        )
+    )
+    return any(_is_sso_provider(provider.details) for provider in providers)
+
+
+def _is_sso_provider(details: AuthProviderDetails) -> bool:
+    if isinstance(details, SamlProviderDetails):
+        return True
+    return (
+        isinstance(details, OAuthProviderDetails)
+        and details.provider_type == ProviderType.ENTERPRISE_SSO
+        and details.legacy_keycloak_type is None
+        and not (KEYCLOAK_CLIENT_ID and details.client_id == KEYCLOAK_CLIENT_ID)
+    )
+
+
+async def personal_workspaces_enabled() -> bool:
+    """Whether the web client shows users their personal workspaces."""
+    web_client = await get_global_config().web_client.get_web_client_config()
+    return not web_client.feature_flags.hide_personal_workspaces

storage/instance_settings.py

@@ -10,7 +10,7 @@
 
 
 class InstanceSettings(Base):
-    """The company name and logo shown across the instance, and first-install state.
+    """Instance-wide state: company name and logo, first-install state, instance ID.
 
     This table holds at most one row (enforced by a database constraint).
     """
@@ -43,6 +43,10 @@ class InstanceSettings(Base):
     signup_link_minted_at: Mapped[datetime | None] = mapped_column(
         DateTime(timezone=True), nullable=True
     )
+    # Identifies this install in the usage reports it sends Customer Center.
+    # Set the first time a report needs it and never changed afterward. See
+    # ``storage.instance_settings_store``.
+    instance_id: Mapped[UUID | None] = mapped_column(nullable=True)
     created_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True),
         default=lambda: datetime.now(UTC),

storage/instance_settings_store.py

@@ -0,0 +1,47 @@
+"""Store for the one-row ``instance_settings`` table."""
+
+from uuid import UUID, uuid4
+
+from sqlalchemy import select
+from sqlalchemy.dialects.postgresql import insert
+
+from storage.database import a_session_maker
+from storage.instance_settings import InstanceSettings
+
+
+class InstanceSettingsStore:
+    @staticmethod
+    async def get_or_create_instance_id() -> UUID:
+        """This install's instance ID, created the first time it is asked for.
+
+        Every replica shares the database, so they all get the same ID, and it
+        never changes once set. When replicas race to create it, the first
+        write wins and the others read it back. The row is created if it does
+        not exist yet; the first-user insert in ``UserStore`` fills in its own
+        columns on an existing row, so creating it here first is safe.
+        """
+        async with a_session_maker() as session:
+            instance_id = await session.scalar(
+                select(InstanceSettings.instance_id).where(InstanceSettings.id == 1)
+            )
+            if instance_id is not None:
+                return instance_id
+
+            stmt = insert(InstanceSettings).values(id=1, instance_id=uuid4())
+            # The WHERE makes this a no-op when another replica got there
+            # first, so RETURNING gives no row and the winner's ID is read below.
+            created = await session.scalar(
+                stmt.on_conflict_do_update(
+                    index_elements=[InstanceSettings.id],
+                    set_={'instance_id': stmt.excluded.instance_id},
+                    where=InstanceSettings.instance_id.is_(None),
+                ).returning(InstanceSettings.instance_id)
+            )
+            if created is None:
+                created = await session.scalar(
+                    select(InstanceSettings.instance_id).where(InstanceSettings.id == 1)
+                )
+            await session.commit()
+        if created is None:
+            raise RuntimeError('instance_settings has no instance_id after upsert')
+        return created