opensandbox-group--opensandbox/server/opensandbox_server/api/devops.py

# Copyright 2026 Alibaba Group Holding Ltd.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""
API routes for OpenSandbox DevOps diagnostics.

Requests that include `scope` target the stable Diagnostics API. The open-source
server has not implemented that API yet, so these requests return a uniform
not-implemented error. Requests without `scope` preserve the deprecated DevOps
plain-text behavior for legacy humans, agents, and CLI clients.
"""

import logging
from typing import Optional

from fastapi import APIRouter, HTTPException, Query, status
from fastapi.responses import JSONResponse, PlainTextResponse

from opensandbox_server.api.lifecycle import sandbox_service

logger = logging.getLogger(__name__)
router = APIRouter(tags=["DevOps"])


def _diagnostics_not_implemented_response() -> JSONResponse:
    return JSONResponse(
        status_code=status.HTTP_501_NOT_IMPLEMENTED,
        content={
            "code": "DIAGNOSTICS_NOT_IMPLEMENTED",
            "message": "The stable Diagnostics API is not implemented by this OpenSandbox server.",
        },
    )


def _deprecated_plain_text_response(content: str) -> PlainTextResponse:
    return PlainTextResponse(
        content=content,
        headers={"Deprecation": "true"},
    )


@router.get(
    "/sandboxes/{sandbox_id}/diagnostics/logs",
    response_model=None,
    status_code=status.HTTP_200_OK,
    responses={
        200: {
            "description": "Deprecated plain-text logs when scope is omitted",
            "content": {"text/plain": {}},
        },
        501: {
            "description": "Stable Diagnostics API is not implemented by this server",
            "content": {"application/json": {}},
        },
        404: {"description": "Sandbox not found"},
    },
)
def get_sandbox_logs(
    sandbox_id: str,
    scope: Optional[str] = Query(
        None,
        description="Required for stable Diagnostics JSON responses. Omit only for deprecated plain-text logs.",
    ),
    tail: int = Query(
        100,
        ge=1,
        le=10000,
        deprecated=True,
        description="Deprecated plain-text logs only. Number of trailing log lines.",
    ),
    since: Optional[str] = Query(
        None,
        deprecated=True,
        description="Deprecated plain-text logs only. Only return logs newer than this duration (e.g. 10m, 1h).",
    ),
    container: Optional[str] = Query(
        None,
        deprecated=True,
        description=(
            "Deprecated plain-text logs only. Container name to read logs from. "
            "Defaults to the canonical user container (typically 'sandbox') when "
            "the runtime supports multi-container pods."
        ),
    ),
) -> JSONResponse | PlainTextResponse:
    """Retrieve diagnostic logs for a sandbox."""
    if scope is not None:
        return _diagnostics_not_implemented_response()
    text = sandbox_service.get_sandbox_logs(
        sandbox_id, tail=tail, since=since, container=container
    )
    return _deprecated_plain_text_response(text)


@router.get(
    "/sandboxes/{sandbox_id}/diagnostics/inspect",
    response_class=PlainTextResponse,
    status_code=status.HTTP_200_OK,
    deprecated=True,
    responses={
        200: {"description": "Container inspection as plain text", "content": {"text/plain": {}}},
        404: {"description": "Sandbox not found"},
    },
)
def get_sandbox_inspect(sandbox_id: str) -> PlainTextResponse:
    """Retrieve detailed inspection info for a sandbox container."""
    text = sandbox_service.get_sandbox_inspect(sandbox_id)
    return PlainTextResponse(content=text)


@router.get(
    "/sandboxes/{sandbox_id}/diagnostics/events",
    response_model=None,
    status_code=status.HTTP_200_OK,
    responses={
        200: {
            "description": "Deprecated plain-text events when scope is omitted",
            "content": {"text/plain": {}},
        },
        501: {
            "description": "Stable Diagnostics API is not implemented by this server",
            "content": {"application/json": {}},
        },
        404: {"description": "Sandbox not found"},
    },
)
def get_sandbox_events(
    sandbox_id: str,
    scope: Optional[str] = Query(
        None,
        description="Required for stable Diagnostics JSON responses. Omit only for deprecated plain-text events.",
    ),
    limit: int = Query(
        50,
        ge=1,
        le=500,
        deprecated=True,
        description="Deprecated plain-text events only. Maximum number of events to return.",
    ),
) -> JSONResponse | PlainTextResponse:
    """Retrieve diagnostic events for a sandbox."""
    if scope is not None:
        return _diagnostics_not_implemented_response()
    text = sandbox_service.get_sandbox_events(sandbox_id, limit=limit)
    return _deprecated_plain_text_response(text)


@router.get(
    "/sandboxes/{sandbox_id}/diagnostics/summary",
    response_class=PlainTextResponse,
    status_code=status.HTTP_200_OK,
    deprecated=True,
    responses={
        200: {"description": "Combined diagnostics summary as plain text", "content": {"text/plain": {}}},
        404: {"description": "Sandbox not found"},
    },
)
def get_sandbox_diagnostics_summary(
    sandbox_id: str,
    tail: int = Query(50, ge=1, le=10000, description="Number of trailing log lines"),
    event_limit: int = Query(20, ge=1, le=500, description="Maximum number of events"),
) -> PlainTextResponse:
    """One-shot diagnostics summary: inspect + events + logs."""
    sections: list[str] = []

    sections.append("=" * 72)
    sections.append("SANDBOX DIAGNOSTICS SUMMARY")
    sections.append(f"Sandbox ID: {sandbox_id}")
    sections.append("=" * 72)

    # Inspect — let HTTPException (e.g. 404) propagate so callers get a proper error
    sections.append("")
    sections.append("-" * 40)
    sections.append("INSPECT")
    sections.append("-" * 40)
    try:
        sections.append(sandbox_service.get_sandbox_inspect(sandbox_id))
    except HTTPException:
        raise
    except Exception:
        logger.exception("Failed to collect sandbox inspect diagnostics for %s", sandbox_id)
        sections.append("[error] Failed to collect inspect diagnostics.")

    # Events
    sections.append("")
    sections.append("-" * 40)
    sections.append("EVENTS")
    sections.append("-" * 40)
    try:
        sections.append(sandbox_service.get_sandbox_events(sandbox_id, limit=event_limit))
    except HTTPException:
        raise
    except Exception:
        logger.exception("Failed to collect sandbox event diagnostics for %s", sandbox_id)
        sections.append("[error] Failed to collect event diagnostics.")

    # Logs
    sections.append("")
    sections.append("-" * 40)
    sections.append("LOGS (last {} lines)".format(tail))
    sections.append("-" * 40)
    try:
        sections.append(sandbox_service.get_sandbox_logs(sandbox_id, tail=tail))
    except HTTPException:
        raise
    except Exception:
        logger.exception("Failed to collect sandbox log diagnostics for %s", sandbox_id)
        sections.append("[error] Failed to collect log diagnostics.")

    return PlainTextResponse(content="\n".join(sections) + "\n")