feat(application): add license notification

application/backend/src/alembic/versions/7a213a27d666_initial_schema.py

@@ -21,6 +21,15 @@
 
 def upgrade() -> None:
     """Upgrade schema - create all tables matching schema.py."""
+    op.create_table(
+        "license_acceptance",
+        sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
+        sa.Column("accepted_version", sa.String(length=64), nullable=False),
+        sa.Column("created_at", sa.DateTime(), server_default=sa.text("(CURRENT_TIMESTAMP)"), nullable=False),
+        sa.Column("updated_at", sa.DateTime(), server_default=sa.text("(CURRENT_TIMESTAMP)"), nullable=False),
+        sa.PrimaryKeyConstraint("id"),
+    )
+
     # Create projects table first (referenced by other tables)
     op.create_table(
         "projects",
@@ -160,3 +169,4 @@ def downgrade() -> None:
     op.drop_table("sources")
     op.drop_table("jobs")
     op.drop_table("projects")
+    op.drop_table("license_acceptance")

application/backend/src/api/endpoints/system_endpoints.py

@@ -14,7 +14,7 @@
 
 from api.dependencies.dependencies import get_system_service
 from api.endpoints import API_PREFIX
-from pydantic_models import SystemInfo
+from pydantic_models import LicenseAcceptanceResponse, LicenseStatus, SystemInfo
 from pydantic_models.system import CameraInfo, DeviceInfo
 from services.system_service import SystemService
 from settings import get_settings
@@ -39,6 +39,22 @@ async def get_system_info(
     return system_service.get_system_info()
 
 
+@system_router.get("/license")
+async def get_license_status(
+    system_service: Annotated[SystemService, Depends(get_system_service)],
+) -> LicenseStatus:
+    """Get license acceptance status for the current application version."""
+    return await system_service.get_license_status()
+
+
+@system_router.post("/license:accept")
+async def accept_license(
+    system_service: Annotated[SystemService, Depends(get_system_service)],
+) -> LicenseAcceptanceResponse:
+    """Accept licenses required for the current application version."""
+    return await system_service.accept_licenses()
+
+
 @system_router.get("/devices/inference")
 async def get_inference_devices(
     system_service: Annotated[SystemService, Depends(get_system_service)],

application/backend/src/db/schema.py

@@ -117,6 +117,19 @@ class SourceDB(Base):
     updated_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.current_timestamp())
 
 
+class LicenseAcceptanceDB(Base):
+    __tablename__ = "license_acceptance"
+
+    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
+    accepted_version: Mapped[str] = mapped_column(String(64), nullable=False)
+    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.current_timestamp())
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime,
+        server_default=func.current_timestamp(),
+        onupdate=func.current_timestamp(),
+    )
+
+
 class SinkDB(Base):
     __tablename__ = "sinks"
 

application/backend/src/pydantic_models/__init__.py

@@ -10,12 +10,20 @@
 from .project import Project, ProjectList, ProjectUpdate
 from .sink import DisconnectedSinkConfig, OutputFormat, Sink, SinkType
 from .source import DisconnectedSourceConfig, Source, SourceType
-from .system import LibraryVersions, SystemInfo
+from .system import (
+    DeploymentType,
+    LibraryVersions,
+    LicenseAcceptanceResponse,
+    LicenseReference,
+    LicenseStatus,
+    SystemInfo,
+)
 from .trainable_model import ModelFamily, TrainableModel, TrainableModelList
 from .video import Video, VideoExtension, VideoList
 
 __all__ = [
     "DatasetSnapshot",
+    "DeploymentType",
     "DisconnectedSinkConfig",
     "DisconnectedSourceConfig",
     "ImageExtension",
@@ -26,6 +34,9 @@
     "JobType",
     "LatencyMetrics",
     "LibraryVersions",
+    "LicenseAcceptanceResponse",
+    "LicenseReference",
+    "LicenseStatus",
     "Media",
     "MediaList",
     "Model",

application/backend/src/pydantic_models/system.py

@@ -16,6 +16,14 @@ class DeviceType(StrEnum):
     NPU = auto()
 
 
+class DeploymentType(StrEnum):
+    """Enumeration of supported Anomalib Studio deployment types."""
+
+    WIN_APP = auto()
+    DOCKER = auto()
+    DEV = auto()
+
+
 class DeviceInfo(BaseModel):
     """Device information schema"""
 
@@ -55,6 +63,7 @@ class SystemInfo(BaseModel):
     os_version: str
     platform: str
     app_version: str
+    deployment_type: DeploymentType
     libraries: LibraryVersions
     devices: list[DeviceInfo]
 
@@ -65,6 +74,7 @@ class SystemInfo(BaseModel):
                 "os_version": "5.15.0-generic",
                 "platform": "Linux-5.15.0-generic-x86_64-with-glibc2.35",
                 "app_version": "0.1.0",
+                "deployment_type": "docker",
                 "libraries": {
                     "anomalib": "2.0.0",
                     "python": "3.11.0",
@@ -100,3 +110,28 @@ class SystemInfo(BaseModel):
             },
         },
     }
+
+
+class LicenseReference(BaseModel):
+    """License reference displayed in the Studio acceptance dialog."""
+
+    name: str = Field(..., description="Display name of the license entry")
+    url: str = Field(..., description="URL pointing to the full license text")
+    required_for: str = Field(..., description="Deployment or model requiring this license")
+
+
+class LicenseStatus(BaseModel):
+    """Current license acceptance status for the running application version."""
+
+    accepted: bool = Field(..., description="Whether the user accepted licenses for the running app version")
+    accepted_version: str | None = Field(None, description="Last accepted application version")
+    app_version: str = Field(..., description="Current application version")
+    deployment_type: DeploymentType = Field(..., description="Current Studio deployment type")
+    licenses: list[LicenseReference] = Field(..., description="Licenses the user must review and accept")
+
+
+class LicenseAcceptanceResponse(BaseModel):
+    """Response returned after accepting licenses."""
+
+    accepted: bool = Field(..., description="Whether acceptance was stored successfully")
+    accepted_version: str = Field(..., description="Application version recorded for acceptance")

application/backend/src/services/__init__.py

@@ -16,6 +16,7 @@
 from .model_service import ModelService
 from .pipeline_service import PipelineService
 from .project_service import ProjectService
+from .system_service import SystemService
 from .training_service import TrainingService
 from .video_service import VideoService
 from .video_stream_service import VideoStreamService
@@ -34,6 +35,7 @@
     "ResourceAlreadyExistsError",
     "ResourceInUseError",
     "ResourceNotFoundError",
+    "SystemService",
     "TrainingService",
     "VideoService",
     "VideoStreamService",

application/backend/src/services/system_service.py

@@ -2,17 +2,32 @@
 # SPDX-License-Identifier: Apache-2.0
 import platform
 import re
+import sys
 from functools import lru_cache
 from importlib.metadata import PackageNotFoundError, version
+from pathlib import Path
 
 import cv2
 import openvino as ov
 import psutil
 import torch
 from cv2_enumerate_cameras import enumerate_cameras
 from loguru import logger
-
-from pydantic_models.system import CameraInfo, DeviceInfo, DeviceType, LibraryVersions, SystemInfo
+from sqlalchemy import desc, select
+
+from db import get_async_db_session_ctx
+from db.schema import LicenseAcceptanceDB
+from pydantic_models.system import (
+    CameraInfo,
+    DeploymentType,
+    DeviceInfo,
+    DeviceType,
+    LibraryVersions,
+    LicenseAcceptanceResponse,
+    LicenseReference,
+    LicenseStatus,
+    SystemInfo,
+)
 from settings import get_settings
 
 DEVICE_PATTERN = re.compile(r"^(cpu|xpu|cuda)(-(\d+))?$")
@@ -22,6 +37,29 @@
     "Linux": cv2.CAP_V4L2,
     "Darwin": cv2.CAP_AVFOUNDATION,
 }
+APACHE_LICENSE_URL = "https://www.apache.org/licenses/LICENSE-2.0"
+INTEL_SIMPLIFIED_LICENSE_URL = "https://www.intel.com/content/www/us/en/content-details/749362/intel-simplified-software-license-version-october-2022.html"
+ANOMALIB_LICENSE_BLOB_ROOT = "https://github.com/open-edge-platform/anomalib/blob/main/src/anomalib/models"
+MODEL_LICENSES: tuple[tuple[str, str, str], ...] = (
+    ("DRAEM", "MIT License", f"{ANOMALIB_LICENSE_BLOB_ROOT}/image/draem/LICENSE"),
+    ("DSR", "Apache 2.0 License", f"{ANOMALIB_LICENSE_BLOB_ROOT}/image/dsr/LICENSE"),
+    (
+        "Reverse Distillation",
+        "MIT License",
+        f"{ANOMALIB_LICENSE_BLOB_ROOT}/image/reverse_distillation/LICENSE",
+    ),
+    (
+        "AI-VAD (CLIP)",
+        "MIT License",
+        f"{ANOMALIB_LICENSE_BLOB_ROOT}/video/ai_vad/clip/LICENSE",
+    ),
+    (
+        "SuperSimpleNet",
+        "MIT License",
+        f"{ANOMALIB_LICENSE_BLOB_ROOT}/image/supersimplenet/LICENSE",
+    ),
+    ("UniNet", "MIT License", f"{ANOMALIB_LICENSE_BLOB_ROOT}/image/uninet/LICENSE"),
+)
 
 
 class SystemService:
@@ -30,6 +68,77 @@ class SystemService:
     def __init__(self) -> None:
         self.process = psutil.Process()
 
+    @staticmethod
+    def get_deployment_type() -> DeploymentType:
+        """Infer the current deployment type from runtime settings and platform."""
+        settings = get_settings()
+
+        if getattr(sys, "frozen", False) and (os_name := platform.system()) == "Windows":
+            logger.debug(f"Detected frozen Windows application deployment: {os_name}")
+            return DeploymentType.WIN_APP
+        if settings.static_files_dir:
+            static_dir = Path(settings.static_files_dir)
+            if static_dir.is_dir() and (static_dir / "index.html").exists():
+                return DeploymentType.DOCKER
+        return DeploymentType.DEV
+
+    @classmethod
+    def get_required_licenses(cls) -> list[LicenseReference]:
+        """Return licenses that must be accepted for the current deployment."""
+        deployment_type = cls.get_deployment_type()
+        primary_license = LicenseReference(
+            name=(
+                "Intel Simplified Software License"
+                if deployment_type == DeploymentType.WIN_APP
+                else "Apache 2.0 License"
+            ),
+            url=(INTEL_SIMPLIFIED_LICENSE_URL if deployment_type == DeploymentType.WIN_APP else APACHE_LICENSE_URL),
+            required_for=(
+                "Windows application"
+                if deployment_type == DeploymentType.WIN_APP
+                else "Docker and development deployments"
+            ),
+        )
+        model_licenses = [
+            LicenseReference(name=name, url=url, required_for=required_for)
+            for required_for, name, url in MODEL_LICENSES
+        ]
+        return [primary_license, *model_licenses]
+
+    @staticmethod
+    async def _get_latest_license_acceptance() -> LicenseAcceptanceDB | None:
+        async with get_async_db_session_ctx() as session:
+            result = await session.execute(select(LicenseAcceptanceDB).order_by(desc(LicenseAcceptanceDB.id)).limit(1))
+            return result.scalar_one_or_none()
+
+    async def get_license_status(self) -> LicenseStatus:
+        """Return whether licenses were accepted for the running application version."""
+        settings = get_settings()
+        acceptance = await self._get_latest_license_acceptance()
+        accepted_version = acceptance.accepted_version if acceptance is not None else None
+        return LicenseStatus(
+            accepted=accepted_version == settings.version,
+            accepted_version=accepted_version,
+            app_version=settings.version,
+            deployment_type=self.get_deployment_type(),
+            licenses=self.get_required_licenses(),
+        )
+
+    async def accept_licenses(self) -> LicenseAcceptanceResponse:
+        """Persist license acceptance for the running application version."""
+        settings = get_settings()
+        async with get_async_db_session_ctx() as session:
+            result = await session.execute(select(LicenseAcceptanceDB).order_by(desc(LicenseAcceptanceDB.id)).limit(1))
+            acceptance = result.scalar_one_or_none()
+            if acceptance is None:
+                acceptance = LicenseAcceptanceDB(accepted_version=settings.version)
+                session.add(acceptance)
+            else:
+                acceptance.accepted_version = settings.version
+            await session.commit()
+
+        return LicenseAcceptanceResponse(accepted=True, accepted_version=settings.version)
+
     def get_memory_usage(self) -> tuple[float, float]:
         """
         Get the memory usage of the process
@@ -314,6 +423,7 @@ def get_system_info(self) -> SystemInfo:
             os_version=platform.release(),
             platform=platform.platform(),
             app_version=get_settings().version,
+            deployment_type=self.get_deployment_type(),
             libraries=self.get_library_versions(),
             devices=self.get_training_devices(),
         )

application/backend/tests/unit/endpoints/test_system.py

@@ -0,0 +1,54 @@
+# Copyright (C) 2026 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+from fastapi import status
+
+from api.dependencies.dependencies import get_system_service
+from main import app
+from pydantic_models.system import DeploymentType, LicenseAcceptanceResponse, LicenseReference, LicenseStatus
+from services import SystemService
+
+
+@pytest.fixture
+def fxt_system_service() -> MagicMock:
+    system_service = MagicMock(spec=SystemService)
+    system_service.get_license_status = AsyncMock(
+        return_value=LicenseStatus(
+            accepted=False,
+            accepted_version=None,
+            app_version="1.2.3",
+            deployment_type=DeploymentType.DOCKER,
+            licenses=[
+                LicenseReference(
+                    name="Apache 2.0 License",
+                    url="https://www.apache.org/licenses/LICENSE-2.0",
+                    required_for="Docker and development deployments",
+                ),
+            ],
+        ),
+    )
+    system_service.accept_licenses = AsyncMock(
+        return_value=LicenseAcceptanceResponse(accepted=True, accepted_version="1.2.3"),
+    )
+    app.dependency_overrides[get_system_service] = lambda: system_service
+    return system_service
+
+
+def test_get_license_status(fxt_client, fxt_system_service):
+    response = fxt_client.get("/api/system/license")
+
+    assert response.status_code == status.HTTP_200_OK
+    assert response.json()["accepted"] is False
+    assert response.json()["app_version"] == "1.2.3"
+    fxt_system_service.get_license_status.assert_awaited_once()
+
+
+def test_accept_license(fxt_client, fxt_system_service):
+    response = fxt_client.post("/api/system/license:accept")
+
+    assert response.status_code == status.HTTP_200_OK
+    assert response.json() == {"accepted": True, "accepted_version": "1.2.3"}
+    fxt_system_service.accept_licenses.assert_awaited_once()