Browser_Agent/api/server.py

# api/server.py
"""
Interne FastAPI für Status-Checks, Task-Enqueuing und ephemeren Medien-Transfer.
Läuft auf Port 8765. Medien verbleiben nur flüchtig im RAM (TTL).
"""
import os
import secrets
from typing import Any
from fastapi import Depends, FastAPI, Header, HTTPException, Response
from fastapi.responses import FileResponse
from pydantic import BaseModel, Field
from config.settings import settings
from job_queue.task_queue import task_queue
from api.models import TaskRequest, TaskResponse
from workflows.ephemeral_media import ephemeral_media_store
from security.crypto import encrypt_bytes
from agent.vision_subagent import VisionCheckError, vision_subagent

app = FastAPI(title="Gravity Browser Agent API", version="0.2.0")


class NewTaskRequest(BaseModel):
    task_type: str
    params: dict
    chat_id: int | None = None


class HandoffCompletionRequest(BaseModel):
    revision: int = Field(ge=0)
    cookies: list[dict[str, Any]] = Field(default_factory=list)
    resume_url: str
    submission_confirmed: bool = False


class DecisionRequest(BaseModel):
    confirmation_id: str
    revision: int = Field(ge=0)
    approved: bool


class VisionCheckRequest(BaseModel):
    goal: str = Field(min_length=1, max_length=4_000)
    images: list[str] = Field(min_length=1, max_length=3)


def require_agent_auth(authorization: str | None = Header(default=None)) -> None:
    expected = settings.gravity_token.strip()
    if not expected:
        if settings.require_api_auth:
            raise HTTPException(status_code=503, detail="Agent API authentication is not configured")
        return
    scheme, _, supplied = (authorization or "").partition(" ")
    if scheme.lower() != "bearer" or not secrets.compare_digest(supplied, expected):
        raise HTTPException(status_code=401, detail="Invalid agent access token")


@app.get("/health")
async def health():
    ephemeral_media_store.purge_expired()
    return {"status": "ok", "queue_size": task_queue.size}


@app.get("/ready")
async def ready():
    browser_ready = bool(getattr(__import__("browser.manager", fromlist=["browser_manager"]).browser_manager, "is_ready", False))
    store = task_queue._store
    dependencies = {
        "worker": {
            "ready": task_queue.worker_heartbeat is not None,
            "heartbeat_at": task_queue.worker_heartbeat.isoformat() if task_queue.worker_heartbeat else None,
        },
        "browser": {"ready": browser_ready},
        "supabase": {
            "configured": store.configured,
            "ready": store.ready,
            "error": store.last_error,
        },
        "auth": {"ready": bool(settings.gravity_token) or not settings.require_api_auth},
        "checkpoint_encryption": {
            "ready": len((settings.aes_secret_key or settings.gravity_token).strip()) >= 16
        },
        "gmail": {"configured": bool(settings.agent_gmail_user and settings.agent_gmail_password)},
    }
    core_ready = (
        dependencies["worker"]["ready"]
        and dependencies["auth"]["ready"]
        and dependencies["checkpoint_encryption"]["ready"]
    )
    return {"status": "ready" if core_ready else "degraded", "dependencies": dependencies}


@app.post("/v1/vision/check", dependencies=[Depends(require_agent_auth)])
async def visual_check(req: VisionCheckRequest):
    """Analyze ephemeral screenshots without persisting or logging payloads."""
    try:
        return await vision_subagent.analyze(req.images, req.goal)
    except VisionCheckError as exc:
        raise HTTPException(
            status_code=503 if exc.retryable else 400,
            detail={"code": "VISUAL_CHECK_FAILED", "message": str(exc), "retryable": exc.retryable},
        ) from exc


@app.on_event("shutdown")
async def close_vision_client():
    await vision_subagent.close()


@app.post("/task", dependencies=[Depends(require_agent_auth)])
async def create_task(
    req: NewTaskRequest,
    idempotency_key: str | None = Header(default=None, alias="Idempotency-Key"),
):
    # One-release legacy adapter: old Flutter builds submit captcha_resume as a
    # new task. Resolve it onto the original waiting task instead.
    if req.task_type == "captcha_resume":
        source_task_id = str(req.params.get("source_task_id") or "")
        source_task = task_queue._tasks.get(source_task_id)
        if source_task and source_task.handoff:
            try:
                await task_queue.resume_handoff(
                    source_task_id,
                    str(source_task.handoff["id"]),
                    revision=source_task.revision,
                    cookies=list(req.params.get("exported_cookies") or []),
                    resume_url=str(req.params.get("resume_url") or source_task.handoff.get("url") or ""),
                    submission_confirmed=req.params.get("submission_confirmed") is True,
                )
                return {"task_id": source_task_id, "legacy_resume": True}
            except (KeyError, ValueError) as exc:
                raise HTTPException(status_code=409, detail=str(exc)) from exc
    try:
        task_id = await task_queue.enqueue(
            req.task_type,
            req.params,
            req.chat_id,
            idempotency_key=idempotency_key,
        )
    except ValueError as exc:
        if str(exc) == "idempotency_conflict":
            raise HTTPException(status_code=409, detail="Idempotency key payload conflict") from exc
        raise
    return {"task_id": task_id}


@app.get("/task/{task_id}", dependencies=[Depends(require_agent_auth)])
async def get_task_status(task_id: str):
    status = await task_queue.get_status(task_id)
    if "id" not in status:
        raise HTTPException(status_code=404, detail=status.get("error", "Not found"))
    return status


@app.post("/task/{task_id}/cancel", dependencies=[Depends(require_agent_auth)])
async def cancel_task(task_id: str):
    cancelled = await task_queue.cancel(task_id)
    if not cancelled:
        raise HTTPException(status_code=404, detail="Task nicht gefunden")
    return {"id": task_id, "status": "cancelled"}


@app.post("/v1/tasks", dependencies=[Depends(require_agent_auth)], status_code=202)
async def create_task_v1(
    req: NewTaskRequest,
    idempotency_key: str = Header(alias="Idempotency-Key"),
):
    if not idempotency_key.strip():
        raise HTTPException(status_code=400, detail="Idempotency-Key is required")
    try:
        task_id = await task_queue.enqueue(
            req.task_type,
            req.params,
            req.chat_id,
            idempotency_key=idempotency_key,
        )
    except ValueError as exc:
        if str(exc) == "idempotency_conflict":
            raise HTTPException(status_code=409, detail="Idempotency key payload conflict") from exc
        raise
    return await task_queue.get_status(task_id, typed=True)


@app.get("/v1/tasks/{task_id}", dependencies=[Depends(require_agent_auth)])
async def get_task_v1(task_id: str):
    status = await task_queue.get_status(task_id, typed=True)
    if "id" not in status:
        raise HTTPException(status_code=404, detail=status.get("error", "Not found"))
    return status


@app.get("/v1/tasks/{task_id}/events", dependencies=[Depends(require_agent_auth)])
async def get_task_events(task_id: str, after: int = 0):
    task = task_queue._tasks.get(task_id)
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    return {
        "task_id": task_id,
        "revision": task.revision,
        "events": [event for event in task.events if int(event.get("sequence", 0)) > after],
    }


@app.get(
    "/v1/tasks/{task_id}/handoffs/{handoff_id}/payload",
    dependencies=[Depends(require_agent_auth)],
)
async def get_handoff_payload(task_id: str, handoff_id: str):
    payload = await task_queue.get_handoff_payload(task_id, handoff_id)
    if not payload:
        raise HTTPException(status_code=410, detail="Handoff expired or unavailable")
    return payload


@app.post(
    "/v1/tasks/{task_id}/handoffs/{handoff_id}/complete",
    dependencies=[Depends(require_agent_auth)],
)
async def complete_handoff(task_id: str, handoff_id: str, req: HandoffCompletionRequest):
    try:
        task = await task_queue.resume_handoff(
            task_id,
            handoff_id,
            revision=req.revision,
            cookies=req.cookies,
            resume_url=req.resume_url,
            submission_confirmed=req.submission_confirmed,
        )
    except KeyError as exc:
        raise HTTPException(status_code=404, detail=str(exc)) from exc
    except ValueError as exc:
        code = 410 if str(exc) == "handoff_expired" else 409
        raise HTTPException(status_code=code, detail=str(exc)) from exc
    return await task_queue.get_status(task.id, typed=True)


@app.post("/v1/tasks/{task_id}/decisions", dependencies=[Depends(require_agent_auth)])
async def decide_task(task_id: str, req: DecisionRequest):
    try:
        task = await task_queue.resolve_confirmation(
            task_id,
            confirmation_id=req.confirmation_id,
            revision=req.revision,
            approved=req.approved,
        )
    except KeyError as exc:
        raise HTTPException(status_code=404, detail=str(exc)) from exc
    except ValueError as exc:
        code = 410 if str(exc) == "confirmation_expired" else 409
        raise HTTPException(status_code=code, detail=str(exc)) from exc
    return await task_queue.get_status(task.id, typed=True)


@app.post("/v1/tasks/{task_id}/cancel", dependencies=[Depends(require_agent_auth)])
async def cancel_task_v1(task_id: str):
    if not await task_queue.cancel(task_id):
        raise HTTPException(status_code=404, detail="Task not found")
    return await task_queue.get_status(task_id, typed=True)


# ── Ephemere Medien Endpunkte (RAM-only, keine Festplattenspeicherung) ──────────

@app.get("/api/media/{media_id}", dependencies=[Depends(require_agent_auth)])
async def get_ephemeral_media(media_id: str):
    """
    Liefert das temporär im RAM gehaltene Bild direkt aus.
    Headers verhindern Caching auf Zwischenservern.
    """
    item = ephemeral_media_store.get(media_id)
    if not item:
        raise HTTPException(status_code=404, detail="Medienobjekt abgelaufen oder gelöscht")

    headers = {
        "Cache-Control": "no-store, no-cache, must-revalidate, max-age=0",
        "Pragma": "no-cache",
        "Content-Disposition": f'inline; filename="{item.filename}"',
        "X-Content-Type-Options": "nosniff",
        "Referrer-Policy": "no-referrer",
        "Cross-Origin-Resource-Policy": "same-origin",
        "Permissions-Policy": "interest-cohort=()",
    }
    return Response(content=item.media_bytes, media_type=item.media_type, headers=headers)


@app.get("/api/media/{media_id}/encrypted", dependencies=[Depends(require_agent_auth)])
async def get_encrypted_media(media_id: str):
    """
    Liefert die Bild-Payload AES-verschlüsselt für den sicheren Mobile-Tresor.
    """
    item = ephemeral_media_store.get(media_id)
    if not item:
        raise HTTPException(status_code=404, detail="Medienobjekt abgelaufen oder gelöscht")

    encrypted_payload = encrypt_bytes(item.media_bytes)
    headers = {
        "Cache-Control": "no-store, no-cache, must-revalidate",
        "Content-Disposition": f'attachment; filename="{item.filename}.enc"',
        "X-Content-Type-Options": "nosniff",
        "Referrer-Policy": "no-referrer",
        "Cross-Origin-Resource-Policy": "same-origin",
        "Permissions-Policy": "interest-cohort=()",
    }
    return Response(content=encrypted_payload, media_type="application/octet-stream", headers=headers)


@app.delete("/api/media/{media_id}", dependencies=[Depends(require_agent_auth)])
async def delete_ephemeral_media(media_id: str):
    """
    Vernichtet das Bild sofort und unwiderruflich aus dem Server-Arbeitsspeicher.
    """
    deleted = ephemeral_media_store.delete(media_id)
    return {"deleted": deleted}


@app.get("/downloads/{filename}", dependencies=[Depends(require_agent_auth)])
async def get_download(filename: str):
    downloads_root = settings.downloads_dir.resolve()
    file_path = (downloads_root / filename).resolve()
    if not file_path.is_relative_to(downloads_root):
        raise HTTPException(status_code=403, detail="Forbidden")
    if not file_path.exists() or not file_path.is_file():
        raise HTTPException(status_code=404, detail="File not found")
    return FileResponse(path=file_path, filename=filename)


@app.post("/run_task", response_model=TaskResponse, dependencies=[Depends(require_agent_auth)])
async def run_task(request: TaskRequest):
    task_desc = request.task
    if request.predefined_task:
        from workflows.tasks import get_predefined_task
        predef = get_predefined_task(request.predefined_task)
        if predef:
            task_desc = predef + "\n" + task_desc
            
    if not task_desc:
        raise HTTPException(status_code=400, detail="Kein Task angegeben.")
        
    task_id = await task_queue.enqueue("agent_task", {"task": task_desc})
    return TaskResponse(status="pending", result=f"Task eingereiht mit ID: {task_id}")