Lablet Controller Architecture¶

class CmlLabsClient:
    """
    CML Lab lifecycle management.

    Base endpoint: /api/v0/labs
    """

    async def list_labs(
        self,
        worker_ip: str,
        auth_token: str
    ) -> list[CmlLabInfo]:
        """List all labs on worker."""

    async def get_lab(
        self,
        worker_ip: str,
        lab_id: str,
        auth_token: str
    ) -> CmlLabDetail:
        """Get detailed lab information."""

    async def import_lab(
        self,
        worker_ip: str,
        topology_yaml: str,
        auth_token: str
    ) -> str:
        """Import YAML topology, return lab_id."""

    async def start_lab(
        self,
        worker_ip: str,
        lab_id: str,
        auth_token: str
    ) -> None:
        """Start all nodes in lab."""

    async def stop_lab(
        self,
        worker_ip: str,
        lab_id: str,
        auth_token: str
    ) -> None:
        """Stop all nodes in lab."""

    async def wipe_lab(
        self,
        worker_ip: str,
        lab_id: str,
        auth_token: str
    ) -> None:
        """Wipe lab state (reset to initial)."""

    async def delete_lab(
        self,
        worker_ip: str,
        lab_id: str,
        auth_token: str
    ) -> None:
        """Delete lab permanently."""

class CmlNodesClient:
    """
    CML Node management.

    Base endpoint: /api/v0/labs/{lab_id}/nodes
    """

    async def list_nodes(
        self,
        worker_ip: str,
        lab_id: str,
        auth_token: str
    ) -> list[CmlNodeInfo]:
        """List all nodes in lab."""

    async def get_node_state(
        self,
        worker_ip: str,
        lab_id: str,
        node_id: str,
        auth_token: str
    ) -> CmlNodeState:
        """Get node state (BOOTED, STOPPED, etc)."""

    async def extract_config(
        self,
        worker_ip: str,
        lab_id: str,
        node_id: str,
        auth_token: str
    ) -> str:
        """Extract running configuration from node."""

class CmlInterfacesClient:
    """
    CML Interface and console port management.

    Base endpoint: /api/v0/labs/{lab_id}/nodes/{node_id}
    """

    async def get_console_ports(
        self,
        worker_ip: str,
        lab_id: str,
        auth_token: str
    ) -> dict[str, int]:
        """Get console port mappings for all nodes."""

    async def get_vnc_url(
        self,
        worker_ip: str,
        lab_id: str,
        node_id: str,
        auth_token: str
    ) -> str:
        """Get VNC access URL for graphical nodes."""

class PortAllocation:
    """Tracks allocated ports for a lablet instance."""
    instance_id: str
    worker_id: str
    allocated_ports: list[int]           # [10011, 10012, 10013]
    port_mappings: dict[str, PortInfo]   # {"workstation": PortInfo(...)}
    allocated_at: datetime
    released_at: datetime | None


class PortInfo:
    """Port mapping for a single device."""
    device_label: str      # "workstation"
    original_port: int     # 5065 (from definition)
    allocated_port: int    # 10011 (dynamically assigned)
    protocol: str          # "serial", "vnc", "pat", "http"
    internal_port: int | None  # For PAT: 22

async def release_ports(self, instance_id: str, worker_id: str) -> None:
    """
    Release allocated ports back to the pool.

    Called during TERMINATED reconciliation after lab deletion.
    """
    allocation = await self._get_allocation(instance_id)
    await self._port_registry.release(worker_id, allocation.allocated_ports)
    await self._control_plane.update_worker_allocations(worker_id)

class LabletDefinitionContent:
    """Content payload stored in LabletDefinition aggregate."""

    cml_yaml: str                    # Full CML topology YAML
    device_definitions: list[DeviceDefinition]  # Parsed device specs
    grade_xml: str | None            # Grading criteria (optional)
    pod_xml: str | None              # Pod configuration (optional, TBD)
    content_version: str             # Version from S3 metadata
    form_qualified_name: str         # S3 path identifier
    refreshed_at: datetime           # Last refresh timestamp
    refreshed_by: str                # Admin user who triggered refresh


class DeviceDefinition:
    """Device extracted from cml.yaml for LDS mapping."""

    label: str                       # "workstation", "R1", "SW1"
    node_definition: str             # "ubuntu-desktop-24-04-v2"
    image_definition: str            # "lablet-desktop-v0.1.1"
    port_tags: list[PortTag]         # Parsed from tags array
    is_user_visible: bool            # Whether LDS should expose this device
    access_credentials: Credentials | None  # Default username/password

def determine_device_visibility(node: CmlNode) -> bool:
    """
    Determine if a CML node should be exposed to LDS.

    Rules:
    1. Nodes with 'hidden' tag are NOT visible
    2. Nodes without port tags are NOT visible (no access method)
    3. Infrastructure nodes (e.g., 'external_connector') are NOT visible
    4. All others ARE visible
    """
    if "hidden" in node.tags:
        return False
    if not any(is_port_tag(t) for t in node.tags):
        return False
    if node.node_definition in INFRASTRUCTURE_NODE_TYPES:
        return False
    return True

@dataclass
class RefreshLabletDefinitionContentCommand(Command[OperationResult[RefreshResultDto]]):
    """Refresh content from S3 for a LabletDefinition."""
    definition_id: str
    force: bool = False  # Refresh even if version unchanged


class RefreshLabletDefinitionContentCommandHandler(CommandHandler):
    async def handle_async(self, request, cancellation_token=None):
        definition = await self._repository.get_by_id_async(request.definition_id)
        if not definition:
            return self.not_found("LabletDefinition", request.definition_id)

        # Download from S3
        content_package = await self._s3_client.download(
            definition.state.form_qualified_name
        )

        # Parse and validate
        parsed = self._content_parser.parse(content_package)

        # Update aggregate
        definition.refresh_content(
            cml_yaml=parsed.cml_yaml,
            device_definitions=parsed.devices,
            grade_xml=parsed.grade_xml,
            content_version=parsed.version,
            refreshed_by=self._current_user.id,
        )

        await self._repository.update_async(definition, cancellation_token)

        return self.ok(RefreshResultDto(
            definition_id=request.definition_id,
            version=parsed.version,
            device_count=len(parsed.devices),
        ))

class LabletInstanceState:
    """LabletInstance aggregate state (partial)."""

    # ... existing fields ...

    # LDS Integration Keys
    lds_session_id: str | None       # LDS session identifier (set on provisioning)
    lds_base_url: str | None         # LDS deployment URL (selected at instantiation)
    lds_login_url: str | None        # User login URL (from LDS after provisioning)

    # Port Allocation
    port_mappings: dict[str, PortInfo]  # {device_label: PortInfo}

class LdsDeployment:
    """LDS deployment configuration."""
    id: str                    # "lds-us-west", "lds-eu-central"
    base_url: str              # "https://lds-us-west.example.com"
    region: str                # "us-west-2"
    capacity: int              # Max concurrent sessions
    priority: int              # Selection priority (lower = preferred)
    enabled: bool              # Whether deployment is active

class LdsDeploymentSelector:
    """
    Select LDS deployment for a LabletInstance.

    Strategy:
    1. If LabletDefinition has lds_affinity, use that deployment
    2. Otherwise, round-robin across enabled deployments (weighted by priority)
    """

    async def select_deployment(
        self,
        definition: LabletDefinition,
        instance: LabletInstance,
    ) -> LdsDeployment:
        # Check for affinity override
        if definition.state.lds_affinity:
            deployment = self._get_by_id(definition.state.lds_affinity)
            if deployment and deployment.enabled:
                return deployment
            logger.warning(f"LDS affinity {definition.state.lds_affinity} unavailable")

        # Round-robin selection
        return await self._round_robin_select()

    async def _round_robin_select(self) -> LdsDeployment:
        """Select next deployment using weighted round-robin."""
        enabled = [d for d in self._deployments if d.enabled]
        enabled.sort(key=lambda d: d.priority)

        # Simple round-robin with priority weighting
        selected = enabled[self._counter % len(enabled)]
        self._counter += 1
        return selected

class LabDeliverySPI(Protocol):
    """
    Abstract interface for Lab Delivery System integration.

    Implementation: LdsApiClient in integration/services/
    """

    async def create_session_with_part(
        self,
        username: str,
        timeslot_start: datetime,
        timeslot_end: datetime,
        form_qualified_name: str,
    ) -> LabSessionInfo:
        """
        Create a LabSession with initial LabSessionPart.

        The form_qualified_name identifies content in S3/MinIO.
        LDS uses this to load tasks and device definitions.
        """
        ...

    async def set_devices(
        self,
        session_id: str,
        devices: list[DeviceAccessInfo],
    ) -> None:
        """
        Provision device access info for the session.

        Each device includes:
        - name: Device label (matches content.xml device_label)
        - protocol: Access protocol (ssh, telnet, vnc, http)
        - host: CML worker IP address
        - port: Allocated external port
        - uri: Full connection URI
        - username/password: Device credentials
        """
        ...

    async def get_session_info(self, session_id: str) -> LabSessionInfo:
        """Get session details including login URL."""
        ...

    async def get_login_url(self, session_id: str) -> str:
        """Get user login URL for the session."""
        ...

    async def archive_session(self, session_id: str) -> None:
        """Archive completed session (on TERMINATED)."""
        ...

    async def refresh_content(self, form_qualified_name: str) -> ContentMetadata:
        """
        Trigger LDS to refresh content from S3/MinIO.

        Called when a LabletDefinition is versioned.
        Synchronous - LDS pulls content package and returns metadata.
        """
        ...

    # Future extensions
    async def collect_responses(self, session_id: str) -> ResponseData:
        """Collect user responses from session."""
        ...

    async def collect_user_feedback_by_session(self, session_id: str) -> FeedbackData:
        """Collect user feedback for specific session."""
        ...

    async def collect_user_feedback_by_form(self, form_qualified_name: str) -> FeedbackData:
        """Collect user feedback for all sessions of a form."""
        ...

async def map_devices_to_ports(
    self,
    definition: LabletDefinition,
    allocated_ports: dict[str, int],
    worker_ip: str,
) -> list[DeviceAccessInfo]:
    """
    Map cached device definitions to LDS access info.

    NOTE: Uses cached content from MongoDB - NO S3 access at instantiation.

    1. Get device_definitions from LabletDefinition.content
    2. Filter to user-visible devices only
    3. Get protocol from port_tags
    4. Lookup allocated port for each device
    5. Build DeviceAccessInfo with worker_ip as host
    """

DeviceDefinition(
    label="workstation",
    node_definition="ubuntu-desktop-24-04-v2",
    port_tags=[
        PortTag(type="serial", port=5065),
        PortTag(type="vnc", port=5066),
    ],
    is_user_visible=True,
)

[
    DeviceAccessInfo(
        name="workstation",
        protocol="vnc",         # Primary access method
        host="10.0.1.50",       # Worker IP
        port=10012,             # Dynamically allocated port
        uri="vnc://10.0.1.50:10012",
        username="cisco",
        password="cisco",
    ),
]

async def on_definition_content_refreshed(
    self,
    definition_id: str,
    form_qualified_name: str,
) -> None:
    """
    Called when LabletDefinition content is refreshed by admin.

    Optionally triggers LDS to refresh its content cache.
    This is separate from the LCM content refresh - LDS maintains its own cache.
    """
    # Notify all configured LDS deployments
    for deployment in self._lds_deployments:
        try:
            await deployment.client.refresh_content(form_qualified_name)
            logger.info(f"LDS {deployment.id} notified of content refresh")
        except Exception as e:
            logger.warning(f"Failed to notify LDS {deployment.id}: {e}")

async def on_instance_terminated(
    self,
    instance: LabletInstance,
) -> None:
    """Archive LDS session on instance termination."""
    if instance.lds_session_id:
        await self.lds_spi.archive_session(instance.lds_session_id)
        logger.info(f"LDS session archived: {instance.lds_session_id}")

async def extract_all_configs(
    self,
    instance: LabletInstance
) -> dict[str, str]:
    """
    Extract configurations from all nodes in lab.

    Returns: {node_label: config_text}

    Use cases:
    - Backup before termination
    - Grading/assessment
    - Configuration comparison
    """