smart-org/apps/api/app/workers/sandbox_gateway.py

"""Sandbox gateway worker — housekeeping for the only sanctioned execution path.

Two jobs, both about not leaking resources:

1. **Reap idle sessions.** A sandbox session that nobody has touched for hours
   is still consuming CPU and disk on the provider. The reaper stops it and
   writes an audit event, because "the platform killed my sandbox" must be
   answerable from the audit log.

2. **Report the policy in force.** The gateway is the one place where untrusted
   code runs, so the worker logs the effective network policy and the absolute
   caps at startup. An operator should be able to see the safety envelope
   without reading the source.

The worker never executes a command. It only stops sessions — the command path
is ``SandboxService.run_command``, reachable from the API under a tenant check.
"""

from __future__ import annotations

from datetime import timedelta

from app.application.container import build_sandbox_service
from app.application.ports.realtime import emit_sandbox_reaped
from app.core.config import settings
from app.domain.audit.models import (
    AuditAction,
    AuditEvent,
    ErrorSeverity,
    ErrorSource,
)
from app.domain.base import utcnow
from app.infrastructure.error_log import record_error
from app.infrastructure.repositories.audit import AuditRepository
from app.infrastructure.repositories.sandbox import (
    SandboxSessionRepository,
    WorkspaceRepository,
)
from app.workers.base import WorkerOutcome, worker_session

#: Sessions are considered abandoned after this long with no activity.
DEFAULT_IDLE_MINUTES = 120


class SandboxGatewayWorker:
    """Stops idle sandbox sessions and records why."""

    name = "sandbox-gateway"

    async def run_pass(
        self, *, idle_minutes: int = DEFAULT_IDLE_MINUTES, limit: int = 200
    ) -> WorkerOutcome:
        """One housekeeping pass: idle sessions first, then orphan containers.

        Combined into one pass so a cron entry needs a single invocation and an
        operator sees one summary line per run.
        """
        outcome = await self.reap(idle_minutes=idle_minutes, limit=limit)
        orphans = await self.reap_orphans(limit=limit)
        outcome.processed += orphans.processed
        outcome.failed += orphans.failed
        for note in orphans.notes:
            outcome.note(note)
        return outcome

    async def reap(
        self, *, idle_minutes: int = DEFAULT_IDLE_MINUTES, limit: int = 200
    ) -> WorkerOutcome:
        outcome = WorkerOutcome(name=self.name)
        idle_before = utcnow() - timedelta(minutes=max(1, idle_minutes))

        async with worker_session() as session:
            sandbox = build_sandbox_service(session)
            workspaces = WorkspaceRepository(session)
            audit = AuditRepository(session)

            stopped = await sandbox.reap_idle_sessions(
                idle_before=idle_before, limit=limit
            )

            for sandbox_session in stopped:
                # Resolve the tenant for the audit trail. A session whose
                # workspace has vanished is still stopped, just recorded
                # without an organization.
                workspace = await workspaces.get(sandbox_session.workspace_id)
                await audit.record(
                    event=AuditEvent(
                        action=AuditAction.SANDBOX_REAP,
                        organization_id=workspace.organization_id if workspace else None,
                        target_type="sandbox_session",
                        target_id=sandbox_session.id,
                        metadata_json={
                            "workspace_id": sandbox_session.workspace_id,
                            "idle_minutes": idle_minutes,
                            "reason": "idle_timeout",
                        },
                    )
                )
                outcome.processed += 1
                outcome.note(f"stopped session {sandbox_session.id}")
                # A reaped sandbox is a tenant-visible fact: push it so the
                # project page's terminal stops showing a live session.
                await emit_sandbox_reaped(
                    organization_id=workspace.organization_id if workspace else None,
                    session_id=sandbox_session.id,
                    workspace_id=sandbox_session.workspace_id,
                )

            if not stopped:
                outcome.skipped += 1

        return outcome

    def policy_report(self) -> WorkerOutcome:
        """Describe the safety envelope this gateway is running under."""
        outcome = WorkerOutcome(name=self.name)
        outcome.note(f"sandbox provider = {settings.sandbox_provider}")
        outcome.note(
            f"command timeout = {settings.sandbox_command_timeout_seconds}s "
            f"(absolute cap enforced in the service layer)"
        )
        outcome.note(f"max output = {settings.sandbox_max_output_bytes} bytes")
        outcome.note("egress policy default = deny_all (set per resource limit)")
        return outcome

    async def reap_orphans(self, *, limit: int = 200) -> WorkerOutcome:
        """Remove labelled containers that no live session row points at.

        ``reap`` above stops *sessions*; this stops *containers*. The two are
        not the same: a session row can be stopped (or the database restored
        from a backup) while the container keeps running, and a container can be
        started by a process that then died before writing its row. Without this
        pass those containers hold CPU, memory and disk forever, because nothing
        in the application knows they exist.

        Only the container runtime is inspected. A provider without
        ``list_managed_containers`` (the mock, the remote adapters) has nothing
        to reap and is skipped silently — this is not an error condition.
        """
        outcome = WorkerOutcome(name=self.name)

        from app.infrastructure.adapters.registry import get_sandbox_provider

        provider = get_sandbox_provider()
        lister = getattr(provider, "list_managed_containers", None)
        if lister is None:
            outcome.skipped += 1
            outcome.note("provider does not expose managed containers; nothing to reap")
            return outcome

        try:
            containers = await lister()
        except Exception as exc:  # noqa: BLE001 - a CLI outage must not kill the worker
            await record_error(
                kind="sandbox_orphan_list_failed",
                message=str(exc),
                exc=exc,
                source=ErrorSource.WORKER,
                severity=ErrorSeverity.WARNING,
                metadata={"worker": self.name},
            )
            outcome.failed += 1
            return outcome

        async with worker_session() as session:
            live = await SandboxSessionRepository(session).list_active_external_ids()
            audit = AuditRepository(session)

            reaped = 0
            for container in containers:
                if reaped >= limit:
                    break
                if container.container_id in live:
                    continue
                # A demo/build container is owned by the delivery pipeline, which
                # tracks it in its own table. Until that table exists it is left
                # alone rather than guessed at.
                if container.role and container.role != "workspace":
                    outcome.skipped += 1
                    continue
                removed = await provider.remove_container(container.container_id)
                if not removed:
                    outcome.failed += 1
                    continue
                reaped += 1
                outcome.processed += 1
                outcome.note(f"removed orphan container {container.container_id}")
                await audit.record(
                    event=AuditEvent(
                        action=AuditAction.SANDBOX_REAP,
                        organization_id=container.organization_id or None,
                        target_type="sandbox_container",
                        target_id=container.container_id,
                        metadata_json={
                            "reason": "orphan_container",
                            "project_id": container.project_id,
                            "role": container.role,
                        },
                    )
                )

            if not reaped:
                outcome.skipped += 1

        return outcome


__all__ = ["DEFAULT_IDLE_MINUTES", "SandboxGatewayWorker"]