plane-gptproto-integration/ai-agent/ai_agent/main.py

import json
import logging
import os
import re
from contextlib import asynccontextmanager
from typing import Any

import asyncpg
import httpx
from fastapi import FastAPI, HTTPException, Query, Request
from pydantic import BaseModel, Field

logger = logging.getLogger(__name__)

_db_pool: asyncpg.Pool | None = None


@asynccontextmanager
async def lifespan(app: FastAPI):
    global _db_pool
    db_url = os.environ.get("DATABASE_URL")
    if db_url:
        try:
            _db_pool = await asyncpg.create_pool(
                db_url, min_size=1, max_size=3, command_timeout=5
            )
        except Exception:
            logger.exception("plane-ai DB pool init failed — /data/role will return is_admin=false")
    yield
    if _db_pool:
        await _db_pool.close()
        _db_pool = None


def _extract_json(text: str) -> dict:
    s = text.strip()
    fence = re.search(r"```(?:json)?\s*(\{.*?\})\s*```", s, re.DOTALL)
    if fence:
        s = fence.group(1)
    else:
        m = re.search(r"\{.*\}", s, re.DOTALL)
        if m:
            s = m.group(0)
    return json.loads(s)


app = FastAPI(title="plane-gptproto ai-agent", lifespan=lifespan)


def get_settings() -> dict:
    return {
        "gptproto_base_url": os.environ.get("GPTPROTO_BASE_URL", "https://gptproto.com/v1"),
        "gptproto_api_key": os.environ.get("GPTPROTO_API_KEY", ""),
        "gptproto_model": os.environ.get("GPTPROTO_MODEL", "claude-sonnet-4-6"),
        "plane_api_base": os.environ.get("PLANE_API_BASE", "http://api:8000"),
        "plane_api_token": os.environ.get("PLANE_API_TOKEN", ""),
    }


async def gptproto_chat(messages: list[dict], response_json: bool = False) -> str:
    cfg = get_settings()
    payload: dict[str, Any] = {"model": cfg["gptproto_model"], "messages": messages}
    if response_json:
        payload["response_format"] = {"type": "json_object"}
    headers = {
        "Authorization": f"Bearer {cfg['gptproto_api_key']}",
        "Content-Type": "application/json",
    }
    async with httpx.AsyncClient(timeout=30) as hc:
        r = await hc.post(
            f"{cfg['gptproto_base_url']}/chat/completions", json=payload, headers=headers
        )
    if r.status_code >= 400:
        raise HTTPException(status_code=502, detail=f"gptproto error: {r.text}")
    data = r.json()
    return data["choices"][0]["message"]["content"]


async def plane_request(
    method: str, path: str, json_body: dict | None = None, params: dict | None = None
) -> httpx.Response:
    cfg = get_settings()
    headers = {"X-Api-Key": cfg["plane_api_token"], "Content-Type": "application/json"}
    url = f"{cfg['plane_api_base']}{path}"
    async with httpx.AsyncClient(timeout=30) as hc:
        return await hc.request(method, url, json=json_body, params=params, headers=headers)


@app.get("/healthz")
async def healthz():
    return {"status": "ok"}


class CreateIssueIn(BaseModel):
    prompt: str = Field(min_length=1)
    project_id: str
    workspace_slug: str


@app.post("/ai/create-issue")
async def ai_create_issue(body: CreateIssueIn):
    system = (
        "Convert the user prompt into a Plane issue. "
        "Output MUST be a single JSON object with exactly these keys: "
        '"name" (short title string), "description_html" (HTML body string). '
        "Do NOT wrap in markdown. Do NOT add prose. Output the JSON object only."
    )
    raw = await gptproto_chat(
        [
            {"role": "system", "content": system},
            {"role": "user", "content": body.prompt},
        ],
    )
    try:
        issue = _extract_json(raw)
    except json.JSONDecodeError as e:
        raise HTTPException(status_code=502, detail=f"gptproto bad json: {e}; raw={raw[:200]}")
    if "name" not in issue:
        raise HTTPException(status_code=502, detail="gptproto missing name")

    path = f"/api/v1/workspaces/{body.workspace_slug}/projects/{body.project_id}/issues/"
    plane_resp = await plane_request(
        "POST",
        path,
        json_body={
            "name": issue["name"],
            "description_html": issue.get("description_html", ""),
        },
    )
    if plane_resp.status_code >= 400:
        raise HTTPException(status_code=502, detail=f"plane error: {plane_resp.text}")
    return plane_resp.json()


class SummarizeIn(BaseModel):
    issues: list[dict] = Field(min_length=1)


@app.post("/ai/summarize")
async def ai_summarize(body: SummarizeIn):
    system = "You summarize a list of Plane issues into a concise paragraph."
    user = json.dumps({"issues": body.issues}, ensure_ascii=False)
    summary = await gptproto_chat(
        [
            {"role": "system", "content": system},
            {"role": "user", "content": user},
        ]
    )
    return {"summary": summary}


class SearchIn(BaseModel):
    query: str = Field(min_length=1)
    workspace_slug: str


@app.post("/ai/search")
async def ai_search(body: SearchIn):
    plane_resp = await plane_request(
        "GET",
        f"/api/v1/workspaces/{body.workspace_slug}/issues/search/",
        params={"search": body.query},
    )
    if plane_resp.status_code >= 400:
        raise HTTPException(status_code=502, detail=f"plane error: {plane_resp.text}")
    results = plane_resp.json()

    system = "You write a one-sentence summary of these Plane search results."
    ai_summary = await gptproto_chat(
        [
            {"role": "system", "content": system},
            {
                "role": "user",
                "content": json.dumps(
                    {"query": body.query, "results": results}, ensure_ascii=False
                ),
            },
        ]
    )
    return {"results": results, "ai_summary": ai_summary}


# ---------------------------------------------------------------------------
# Dashboard data proxy endpoints — forward to Plane API with server-side token
# ---------------------------------------------------------------------------


async def _plane_get_list(path: str, params: dict | None = None) -> list:
    resp = await plane_request("GET", path, params=params)
    if resp.status_code >= 400:
        raise HTTPException(status_code=502, detail=f"plane error: {resp.text}")
    data = resp.json()
    return data.get("results", data) if isinstance(data, dict) else data


@app.get("/data/members")
async def data_members(workspace_slug: str = Query(...)):
    return await _plane_get_list(f"/api/v1/workspaces/{workspace_slug}/members/")


@app.get("/data/projects")
async def data_projects(workspace_slug: str = Query(...)):
    return await _plane_get_list(f"/api/v1/workspaces/{workspace_slug}/projects/")


@app.get("/data/issues")
async def data_issues(
    workspace_slug: str = Query(...),
    project_id: str = Query(...),
    assignee_id: str | None = Query(default=None),
):
    params: dict = {"expand": "state_detail", "per_page": "250"}
    if assignee_id:
        params["assignees"] = assignee_id
    return await _plane_get_list(
        f"/api/v1/workspaces/{workspace_slug}/projects/{project_id}/issues/",
        params=params,
    )


@app.get("/data/states")
async def data_states(workspace_slug: str = Query(...), project_id: str = Query(...)):
    return await _plane_get_list(
        f"/api/v1/workspaces/{workspace_slug}/projects/{project_id}/states/"
    )


@app.get("/data/cycles")
async def data_cycles(workspace_slug: str = Query(...), project_id: str = Query(...)):
    return await _plane_get_list(
        f"/api/v1/workspaces/{workspace_slug}/projects/{project_id}/cycles/"
    )


@app.get("/data/cycle-issues")
async def data_cycle_issues(
    workspace_slug: str = Query(...),
    project_id: str = Query(...),
    cycle_id: str = Query(...),
):
    return await _plane_get_list(
        f"/api/v1/workspaces/{workspace_slug}/projects/{project_id}/cycles/{cycle_id}/issues/",
        params={"expand": "state_detail", "per_page": "250"},
    )


# Plane v1.3.0 workspace role enum: Admin=20, Member=15, Guest=5
_ADMIN_ROLE_MIN = 20

_SESSION_COOKIE_NAME = os.environ.get("SESSION_COOKIE_NAME", "session-id")


@app.get("/data/role")
async def data_role(request: Request, workspace_slug: str = Query(...)):
    """Return is_admin for the current session by querying PostgreSQL directly.

    Flow: session-id cookie → sessions table (user_id) → workspace_members table (role).
    Uses direct SQL to avoid dependency on the workspace-admin-gated Plane API endpoint.
    """
    session_key = request.cookies.get(_SESSION_COOKIE_NAME)
    if not session_key or not _db_pool:
        return {"is_admin": False, "role": 0}

    try:
        # Look up user_id from active (non-expired) session
        session_row = await _db_pool.fetchrow(
            "SELECT user_id::text FROM sessions WHERE session_key = $1 AND expire_date > NOW()",
            session_key,
        )
        if not session_row:
            return {"is_admin": False, "role": 0}

        user_id = session_row["user_id"]

        # Look up workspace member role directly in the DB (avoids admin-gated Plane API)
        member_row = await _db_pool.fetchrow(
            """
            SELECT wm.role FROM workspace_members wm
            JOIN workspaces w ON wm.workspace_id = w.id
            WHERE w.slug = $1
              AND wm.member_id = $2::uuid
              AND wm.is_active = TRUE
              AND wm.deleted_at IS NULL
            """,
            workspace_slug,
            user_id,
        )
    except Exception:
        logger.exception("data_role DB query failed")
        return {"is_admin": False, "role": 0}

    if not member_row:
        return {"is_admin": False, "role": 0}

    role = member_row["role"]
    return {"is_admin": role >= _ADMIN_ROLE_MIN, "role": role}