"""
ARIA Brain — FastAPI-Einstieg.

Phase B Punkt 1: nur Skeleton.
- /health                        → Liveness
- /memory/list                   → alle Punkte (gefiltert)
- /memory/pinned                 → Hot Memory
- /memory/search?q=...&k=5       → semantische Suche
- /memory/save                   → neuen Punkt anlegen
- /memory/update/{id}            → Punkt aendern (re-embed wenn content geaendert)
- /memory/delete/{id}            → Punkt loeschen
- /memory/stats                  → Anzahl Punkte pro Type

/chat (Conversation-Loop) und /skills/* kommen in spaeteren Phasen.
"""

from __future__ import annotations

import logging
import os
from typing import List, Optional

import asyncio
from contextlib import asynccontextmanager

from fastapi import FastAPI, HTTPException, BackgroundTasks, Request, UploadFile, File
from fastapi.responses import Response
from pydantic import BaseModel, Field

from memory import Embedder, VectorStore, MemoryPoint
from conversation import Conversation
from proxy_client import ProxyClient
from agent import Agent
import skills as skills_mod
import metrics as metrics_mod
import triggers as triggers_mod
import watcher as watcher_mod
import background as background_mod
import oauth as oauth_mod
import seed_rules as seed_rules_mod
import projects as projects_mod

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s")
logger = logging.getLogger("aria-brain")

QDRANT_HOST = os.environ.get("QDRANT_HOST", "aria-qdrant")
QDRANT_PORT = int(os.environ.get("QDRANT_PORT", "6333"))

def _seed_spotify_fast_patterns() -> None:
    """One-shot Migration: schreibt Standard-Steuer-Patterns ins Spotify-Skill
    wenn das Skill existiert + aktiv ist + noch keine fast_patterns hat.

    Nach diesem Run kann ARIA die Patterns frei via skill_update aendern."""
    manifest = skills_mod.read_manifest("spotify")
    if not manifest:
        logger.info("[migrate] spotify skill nicht vorhanden — nichts zu tun")
        return
    if manifest.get("fast_patterns"):
        logger.info("[migrate] spotify hat schon fast_patterns (%d) — skip",
                    len(manifest["fast_patterns"]))
        return
    default_patterns = [
        # NEXT
        {"match": r"^(naechster|nächster|naechste|nächste) (track|song|titel|lied)$",
         "args": {"path": "/v1/me/player/next", "method": "POST"},
         "reply": "Spotify: nächster Track ⏭"},
        {"match": r"^(weiter|skip|ueberspringen|überspringen|ueberspring|überspring)$",
         "args": {"path": "/v1/me/player/next", "method": "POST"},
         "reply": "Spotify: nächster Track ⏭"},
        # PREVIOUS
        {"match": r"^(vorheriger|vorheriges|letzter|letztes) (track|song|titel|lied)$",
         "args": {"path": "/v1/me/player/previous", "method": "POST"},
         "reply": "Spotify: vorheriger Track ⏮"},
        {"match": r"^(zurueck|zurück)$",
         "args": {"path": "/v1/me/player/previous", "method": "POST"},
         "reply": "Spotify: vorheriger Track ⏮"},
        # PAUSE
        {"match": r"^(pause|pausiere|pausieren|stop|stopp|halt)$",
         "args": {"path": "/v1/me/player/pause", "method": "PUT"},
         "reply": "Spotify: pausiert ⏸"},
        {"match": r"^(musik|spotify) (pause|aus|stop|stopp)$",
         "args": {"path": "/v1/me/player/pause", "method": "PUT"},
         "reply": "Spotify: pausiert ⏸"},
        # PLAY
        {"match": r"^(play|weiterspielen|weiter spielen|fortsetzen|abspielen)$",
         "args": {"path": "/v1/me/player/play", "method": "PUT"},
         "reply": "Spotify: spielt ▶"},
        {"match": r"^(musik|spotify) (an|wieder an|weiter|fortsetzen)$",
         "args": {"path": "/v1/me/player/play", "method": "PUT"},
         "reply": "Spotify: spielt ▶"},
    ]
    skills_mod.update_skill("spotify", {"fast_patterns": default_patterns})
    logger.info("[migrate] spotify fast_patterns gesetzt (%d Eintraege)",
                len(default_patterns))


@asynccontextmanager
async def lifespan(app: FastAPI):
    """Beim Brain-Start: System-Seed-Regeln idempotent in DB schreiben,
    Trigger-Background-Loop anwerfen. Beim Shutdown: Loop stoppen."""
    try:
        result = seed_rules_mod.apply(store(), embedder())
        logger.info("Lifespan: seed_rules angewendet (%s)", result)
    except Exception as exc:
        logger.exception("Lifespan: seed_rules fehlgeschlagen — Brain startet trotzdem (%s)", exc)

    # Einmalige Migration: Spotify-Skill ohne fast_patterns kriegt die Standard-
    # Patterns injiziert. Idempotent — wenn schon welche da sind, nichts tun.
    # ARIA kann sie spaeter via skill_update beliebig erweitern/ersetzen.
    try:
        _seed_spotify_fast_patterns()
    except Exception as exc:
        logger.warning("Lifespan: spotify fast_patterns Migration: %s", exc)

    # Einmalige Migration: project_id aus conversation.jsonl nach chat_backup.jsonl
    # zurueckschreiben, damit alt-getaggte Projekt-Nachrichten (getaggt bevor
    # chat_backup project_id fuehrte) in der UI wieder im richtigen Projekt
    # landen. Idempotent (Marker), nicht-destruktiv (.bak), atomar.
    try:
        import migrate_backfill_projectid
        res = migrate_backfill_projectid.run()
        logger.info("Lifespan: chat_backup project_id Backfill: %s", res)
    except Exception as exc:
        logger.warning("Lifespan: project_id Backfill Migration: %s", exc)

    task = asyncio.create_task(background_mod.run_loop(agent))
    logger.info("Lifespan: Trigger-Loop gestartet")
    try:
        yield
    finally:
        task.cancel()
        try:
            await task
        except asyncio.CancelledError:
            pass
        logger.info("Lifespan: Trigger-Loop gestoppt")


app = FastAPI(title="ARIA Brain", version="0.1.0", lifespan=lifespan)

_embedder: Optional[Embedder] = None
_store: Optional[VectorStore] = None
_conversation: Optional[Conversation] = None
_proxy: Optional[ProxyClient] = None
_agent: Optional[Agent] = None


def embedder() -> Embedder:
    global _embedder
    if _embedder is None:
        _embedder = Embedder()
    return _embedder


def store() -> VectorStore:
    global _store
    if _store is None:
        _store = VectorStore(host=QDRANT_HOST, port=QDRANT_PORT)
    return _store


def conversation() -> Conversation:
    global _conversation
    if _conversation is None:
        _conversation = Conversation()
    return _conversation


def proxy_client() -> ProxyClient:
    global _proxy
    if _proxy is None:
        _proxy = ProxyClient()
    return _proxy


def agent() -> Agent:
    global _agent
    if _agent is None:
        _agent = Agent(store(), embedder(), conversation(), proxy_client())
    return _agent


# ─── Pydantic-Schemas ─────────────────────────────────────────────────

class MemoryIn(BaseModel):
    type: str = Field(..., description="identity|rule|preference|tool|skill|fact|conversation|reminder")
    title: str
    content: str
    pinned: bool = False
    category: str = ""
    source: str = "manual"
    tags: List[str] = Field(default_factory=list)
    conversation_id: Optional[str] = None
    # Vorhandene Anhang-Metadaten beim Save mitgeben (i.d.R. werden Anhaenge
    # nach dem Save via /memory/{id}/attachments hinzugefuegt — hier eher fuer
    # Bootstrap-Import/Restore-Faelle relevant).
    attachments: List[dict] = Field(default_factory=list)


class MemoryUpdate(BaseModel):
    title: Optional[str] = None
    content: Optional[str] = None
    pinned: Optional[bool] = None
    category: Optional[str] = None
    tags: Optional[List[str]] = None


class MemoryOut(BaseModel):
    id: str
    type: str
    title: str
    content: str
    pinned: bool
    category: str
    source: str
    tags: List[str]
    created_at: str
    updated_at: str
    conversation_id: Optional[str] = None
    score: Optional[float] = None
    attachments: List[dict] = Field(default_factory=list)

    @classmethod
    def from_point(cls, p: MemoryPoint) -> "MemoryOut":
        return cls(**p.__dict__)


class AttachmentUploadBody(BaseModel):
    """Base64-Upload via JSON — Diagnostic schickt Files so."""
    name: str
    data_base64: str


# ─── Health ───────────────────────────────────────────────────────────

@app.get("/health")
def health():
    try:
        n = store().count()
        return {"status": "ok", "memory_count": n, "qdrant": f"{QDRANT_HOST}:{QDRANT_PORT}"}
    except Exception as exc:
        return {"status": "degraded", "error": str(exc), "qdrant": f"{QDRANT_HOST}:{QDRANT_PORT}"}


# ─── Memory-Endpoints ─────────────────────────────────────────────────

@app.get("/memory/get/{point_id}", response_model=MemoryOut)
def memory_get(point_id: str):
    """Einzelner Memory mit allen Feldern (inkl. Anhaengen).
    Pfad-Prefix /memory/get/ vermeidet Konflikt mit /memory/list, /memory/save etc."""
    m = store().get(point_id)
    if not m:
        raise HTTPException(404, f"Memory {point_id} nicht gefunden")
    return MemoryOut.from_point(m)


@app.get("/memory/stats")
def memory_stats():
    s = store()
    points = s.list_all()
    by_type = {}
    pinned = 0
    for p in points:
        by_type[p.type] = by_type.get(p.type, 0) + 1
        if p.pinned:
            pinned += 1
    return {"total": len(points), "pinned": pinned, "by_type": by_type}


@app.get("/memory/list", response_model=List[MemoryOut])
def memory_list(type: Optional[str] = None, limit: int = 200):
    s = store()
    points = s.list_by_type(type, limit=limit) if type else s.list_all(limit=limit)
    return [MemoryOut.from_point(p) for p in points]


@app.get("/memory/pinned", response_model=List[MemoryOut])
def memory_pinned():
    return [MemoryOut.from_point(p) for p in store().list_pinned()]


@app.get("/memory/search-text", response_model=List[MemoryOut])
def memory_search_text(
    q: str,
    k: int = 50,
    type: Optional[str] = None,
    include_pinned: bool = True,
):
    """Volltext-Substring-Suche (case-insensitive) ueber Title + Content +
    Category + Tags. Findet exakte Begriffe — z.B. 'auto' matched 'Stefans Auto'.
    Im Gegensatz zu /memory/search (semantic) keine 'klingt aehnlich'-Treffer."""
    points = store().search_text(
        q, k=k, type_filter=type,
        exclude_pinned=not include_pinned,
    )
    return [MemoryOut.from_point(p) for p in points]


@app.get("/memory/search", response_model=List[MemoryOut])
def memory_search(
    q: str,
    k: int = 5,
    type: Optional[str] = None,
    include_pinned: bool = False,
    score_threshold: Optional[float] = 0.30,
):
    """Semantische Suche. score_threshold filtert schwache Treffer raus
    (Default 0.30 — MiniLM-multilingual liefert <0.25 fuer Rauschen).
    Mit score_threshold=0 wird komplett Top-k zurueckgegeben."""
    vec = embedder().embed(q)
    points = store().search(
        vec, k=k, type_filter=type, exclude_pinned=not include_pinned,
        score_threshold=score_threshold if score_threshold and score_threshold > 0 else None,
    )
    return [MemoryOut.from_point(p) for p in points]


@app.post("/memory/save", response_model=MemoryOut)
def memory_save(body: MemoryIn):
    s = store()
    vec = embedder().embed(body.content)
    point = MemoryPoint(
        id="",
        type=body.type,
        title=body.title,
        content=body.content,
        pinned=body.pinned,
        category=body.category,
        source=body.source,
        tags=body.tags,
        conversation_id=body.conversation_id,
        attachments=body.attachments or [],
    )
    pid = s.upsert(point, vec)
    saved = s.get(pid)
    return MemoryOut.from_point(saved)


@app.patch("/memory/update/{point_id}", response_model=MemoryOut)
def memory_update(point_id: str, body: MemoryUpdate):
    s = store()
    existing = s.get(point_id)
    if not existing:
        raise HTTPException(404, f"Memory {point_id} nicht gefunden")

    content_changed = body.content is not None and body.content != existing.content
    if body.title is not None:
        existing.title = body.title
    if body.content is not None:
        existing.content = body.content
    if body.pinned is not None:
        existing.pinned = body.pinned
    if body.category is not None:
        existing.category = body.category
    if body.tags is not None:
        existing.tags = body.tags

    vec = embedder().embed(existing.content) if content_changed else None
    if vec is None:
        # Vektor unveraendert lassen — nur Payload neu schreiben
        from qdrant_client.http import models as qm
        from memory.vector_store import COLLECTION
        s.client.set_payload(
            collection_name=COLLECTION,
            payload=existing.to_payload() | {"updated_at": __import__("datetime").datetime.now(__import__("datetime").timezone.utc).isoformat()},
            points=[point_id],
        )
        saved = s.get(point_id)
    else:
        s.upsert(existing, vec)
        saved = s.get(point_id)
    return MemoryOut.from_point(saved)


@app.delete("/memory/delete/{point_id}")
def memory_delete(point_id: str):
    s = store()
    if not s.get(point_id):
        raise HTTPException(404, f"Memory {point_id} nicht gefunden")
    s.delete(point_id)
    # Anhaenge mit-loeschen damit nichts verwaist
    try:
        import memory_attachments as mem_att
        n = mem_att.delete_all(point_id)
        if n:
            logger.info("Memory %s + %d Anhaenge geloescht", point_id, n)
    except Exception as exc:
        logger.warning("Anhang-Cleanup fuer %s fehlgeschlagen: %s", point_id, exc)
    return {"deleted": point_id}


# ─── Memory-Anhaenge ──────────────────────────────────────────────────

@app.get("/memory/{point_id}/attachments")
def memory_attachments_list(point_id: str):
    """Liste der Anhaenge zum Memory. Source-of-Truth ist das Payload
    in der DB, aber wir mergen vorsichtshalber mit dem Filesystem-Stand
    (falls ein Upload-Restart zwischendrin schiefging)."""
    import memory_attachments as mem_att
    s = store()
    m = s.get(point_id)
    if not m:
        raise HTTPException(404, f"Memory {point_id} nicht gefunden")
    return {"memory_id": point_id, "attachments": mem_att.list_attachments(point_id)}


def _commit_attachment_meta(point_id: str, meta: dict) -> MemoryOut:
    """Shared-Helper: nach FS-Write das Payload um den neuen Anhang updaten.
    Duplikat-Name wird ersetzt, sonst hinten dran."""
    s = store()
    m = s.get(point_id)
    if not m:
        raise HTTPException(404, f"Memory {point_id} nicht gefunden")
    atts = [a for a in (m.attachments or []) if a.get("name") != meta["name"]]
    atts.append(meta)
    m.attachments = atts
    from memory.vector_store import COLLECTION
    import datetime as _dt
    m.updated_at = _dt.datetime.now(_dt.timezone.utc).isoformat()
    s.client.set_payload(
        collection_name=COLLECTION,
        payload=m.to_payload() | {"updated_at": m.updated_at},
        points=[point_id],
    )
    return MemoryOut.from_point(s.get(point_id))


@app.post("/memory/{point_id}/attachments", response_model=MemoryOut)
def memory_attachments_add(point_id: str, body: AttachmentUploadBody):
    """Anhang als Base64 hochladen — fuer Diagnostic + interne Tools.
    Fuer grosse Files lieber multipart-Variante (/upload) nutzen,
    Base64 sprengt schnell die Bash-ARG_MAX-Grenze beim curl."""
    import memory_attachments as mem_att
    if not store().get(point_id):
        raise HTTPException(404, f"Memory {point_id} nicht gefunden")
    try:
        meta = mem_att.save_from_base64(point_id, body.name, body.data_base64)
    except ValueError as exc:
        raise HTTPException(400, str(exc))
    return _commit_attachment_meta(point_id, meta)


@app.post("/memory/{point_id}/attachments/upload", response_model=MemoryOut)
async def memory_attachments_upload(point_id: str, file: UploadFile = File(...)):
    """Multipart-Upload — Standard fuer Browser-FormData und curl -F.
    Verwendung:
      curl -F file=@foto.jpg "$ARIA_BRAIN_URL/memory/<id>/attachments/upload"
    """
    import memory_attachments as mem_att
    if not store().get(point_id):
        raise HTTPException(404, f"Memory {point_id} nicht gefunden")
    data = await file.read()
    try:
        meta = mem_att.save_attachment(point_id, file.filename or "datei", data)
    except ValueError as exc:
        raise HTTPException(400, str(exc))
    return _commit_attachment_meta(point_id, meta)


@app.delete("/memory/{point_id}/attachments/{filename}", response_model=MemoryOut)
def memory_attachments_delete(point_id: str, filename: str):
    """Einzelnen Anhang loeschen (FS + Payload-Eintrag)."""
    import memory_attachments as mem_att
    s = store()
    m = s.get(point_id)
    if not m:
        raise HTTPException(404, f"Memory {point_id} nicht gefunden")
    removed_fs = mem_att.delete_attachment(point_id, filename)
    safe = filename  # Cleanup synchron mit FS — Payload-Match per name
    atts = [a for a in (m.attachments or []) if a.get("name") not in (filename, safe)]
    m.attachments = atts
    from qdrant_client.http import models as qm
    from memory.vector_store import COLLECTION
    import datetime as _dt
    m.updated_at = _dt.datetime.now(_dt.timezone.utc).isoformat()
    s.client.set_payload(
        collection_name=COLLECTION,
        payload=m.to_payload() | {"updated_at": m.updated_at},
        points=[point_id],
    )
    if not removed_fs and not atts:
        # weder im FS noch im Payload war was — Anhang existierte nicht
        raise HTTPException(404, f"Anhang {filename} nicht gefunden")
    return MemoryOut.from_point(s.get(point_id))


@app.get("/memory/{point_id}/attachments/{filename}")
def memory_attachments_get(point_id: str, filename: str):
    """Liefert die Bytes eines Anhangs. Diagnostic-Server kann das
    durchproxien zur Vorschau/Download in der UI."""
    import memory_attachments as mem_att
    import mimetypes as _mt
    data = mem_att.read_bytes(point_id, filename)
    if data is None:
        raise HTTPException(404, f"Anhang {filename} nicht gefunden")
    mime = _mt.guess_type(filename)[0] or "application/octet-stream"
    return Response(content=data, media_type=mime)


# ─── Migration aus brain-import/ ──────────────────────────────────────

IMPORT_DIR = os.environ.get("IMPORT_DIR", "/import")


@app.post("/memory/migrate")
def memory_migrate():
    """Liest /import/*.md und schreibt atomare Memory-Punkte in die DB.
    Idempotent: bei Re-Run werden Punkte mit gleicher migration_key ersetzt."""
    from pathlib import Path
    from migration import run_migration
    s = store()
    e = embedder()
    result = run_migration(Path(IMPORT_DIR), s, e)
    return result


@app.get("/memory/import-files")
def memory_import_files():
    """Listet was unter /import/ liegt — fuer die Diagnostic-UI."""
    from pathlib import Path
    d = Path(IMPORT_DIR)
    if not d.exists():
        return {"import_dir": str(d), "exists": False, "files": []}
    out = []
    for p in sorted(d.iterdir()):
        if p.is_file():
            try:
                out.append({"name": p.name, "size": p.stat().st_size})
            except Exception:
                pass
    return {"import_dir": str(d), "exists": True, "files": out}


# ─── Bootstrap-Snapshot ───────────────────────────────────────────────
# "Bootstrap" = alle pinned Memories. Export/Import zum schnellen
# Wiederherstellen einer schlanken ARIA nach Wipe.

@app.get("/memory/export-bootstrap")
def memory_export_bootstrap():
    """Gibt alle pinned Memories als JSON zurueck — fuer Browser-Download."""
    s = store()
    pinned = s.list_pinned()
    return {
        "version": 1,
        "exported_at": __import__("datetime").datetime.now(
            __import__("datetime").timezone.utc
        ).isoformat(),
        "count": len(pinned),
        "memories": [
            {
                "type": p.type,
                "title": p.title,
                "content": p.content,
                "pinned": True,
                "category": p.category,
                "source": p.source,
                "tags": p.tags,
            }
            for p in pinned
        ],
    }


class BootstrapBundle(BaseModel):
    version: int = 1
    memories: List[dict]


@app.post("/memory/import-bootstrap")
def memory_import_bootstrap(body: BootstrapBundle):
    """Loescht alle pinned Memories und importiert die im Bundle.
    Cold Memory (unpinned) bleibt unangetastet.

    Wenn keine Memories im Bundle: nur loeschen ist NICHT erlaubt — der
    Caller soll erst exportieren und dann importieren.
    """
    if not body.memories:
        raise HTTPException(400, "Bundle hat keine memories — Abbruch zur Sicherheit")

    s = store()
    e = embedder()

    # Alle aktuell pinned Punkte loeschen
    from qdrant_client.http import models as qm
    from memory.vector_store import COLLECTION
    s.client.delete(
        collection_name=COLLECTION,
        points_selector=qm.FilterSelector(filter=qm.Filter(must=[
            qm.FieldCondition(key="pinned", match=qm.MatchValue(value=True))
        ])),
    )

    # Neue Punkte einspeisen
    created = 0
    for m in body.memories:
        content = (m.get("content") or "").strip()
        if not content:
            continue
        point = MemoryPoint(
            id="",
            type=m.get("type", "fact"),
            title=m.get("title", "(ohne Titel)"),
            content=content,
            pinned=True,
            category=m.get("category", ""),
            source=m.get("source", "bootstrap-import"),
            tags=list(m.get("tags", [])),
        )
        vec = e.embed(content)
        s.upsert(point, vec)
        created += 1

    return {"created": created, "deleted_previous_pinned": True}


# ─── Conversation-Loop ──────────────────────────────────────────────

class ChatIn(BaseModel):
    message: str
    source: str = ""  # "app" / "diagnostic" / "stt" — optional
    # Multi-Threading: Client bestimmt pro Request welches Projekt (leer = Hauptchat).
    # Kein globaler active_project-State mehr im Brain — parallele Requests fuer
    # verschiedene Projekte laufen echt parallel, nur Requests fuers gleiche
    # Projekt queuen (per-Projekt-Lock).
    project_id: str = ""


class ChatOut(BaseModel):
    reply: str
    turns: int
    distilling: bool
    events: list = Field(default_factory=list)
    # Echo der project_id die dieser Turn hatte. Bridge nutzt sie damit die
    # ausgehende Chat-Bubble sauber getaggt in der richtigen Thread-Bahn der
    # UI landet.
    project_id: str = ""


# Per-Projekt async-Locks fuer Queue-Behavior: Requests fuers gleiche Projekt
# warten aufeinander (queue), Requests fuer verschiedene Projekte laufen echt
# parallel. Hauptchat = Lock unter key "" (leerer String).
_project_locks: dict[str, asyncio.Lock] = {}
_project_locks_meta_lock = asyncio.Lock()
# Pro Projekt eine Liste noch-nicht-verarbeiteter Requests. Wird beim Enqueue
# ergaenzt, beim Fertig-Werden gepoppt. Ermoeglicht Queue-Aware-Prompting:
# waehrend ARIA an Task N arbeitet, sieht sie N+1..N+k als System-Prompt-Hinweis
# und kann entscheiden ob eine spaetere Nachricht die aktuelle korrigiert/
# annuliert → dann Skip-Antwort statt Ausfuehren.
_project_pending: dict[str, list[dict]] = {}


async def _get_project_lock(project_id: str) -> asyncio.Lock:
    """Holt (oder erzeugt) den asyncio.Lock fuer ein bestimmtes Projekt.
    Nutzt _project_locks_meta_lock zur Vermeidung von Race Conditions
    beim ersten-Zugriff pro Projekt."""
    async with _project_locks_meta_lock:
        lock = _project_locks.get(project_id)
        if lock is None:
            lock = asyncio.Lock()
            _project_locks[project_id] = lock
        return lock


def _project_queue_snapshot() -> dict:
    """Snapshot fuer /projects/queue-status: welche Projekte arbeiten gerade,
    wieviele wait-in-queue haben, welche sind idle."""
    out = {}
    # Zeige nur Kontexte mit Aktivitaet — locked oder pending
    seen: set = set()
    for pid, lock in _project_locks.items():
        pending = len(_project_pending.get(pid, []))
        is_busy = lock.locked()
        # busy: gerade in Verarbeitung. queue: N weitere warten dahinter.
        # Der Busy-Request zaehlt NICHT in queue (er ist ja aus pending schon "raus").
        out[pid or "__main__"] = {
            "busy": is_busy,
            "queue_size": max(0, pending - (1 if is_busy else 0)),
        }
        seen.add(pid)
    for pid, pend in _project_pending.items():
        if pid in seen:
            continue
        out[pid or "__main__"] = {"busy": False, "queue_size": len(pend)}
    return out


@app.post("/chat", response_model=ChatOut)
async def chat(body: ChatIn, background: BackgroundTasks):
    """Hauptpfad. Antwort kommt synchron. Memory-Destillat laeuft
    im Hintergrund nachdem die Response rausging.

    Multi-Threading: Requests fuers gleiche Projekt (project_id gleich)
    laufen serialisiert durch den per-Projekt-Lock — Queue-Behavior.
    Verschiedene Projekte laufen parallel."""
    pid = (body.project_id or "").strip()
    lock = await _get_project_lock(pid)
    # Vor dem Lock in die Pending-Liste, damit die verlaufende Task sehen kann
    # was NACH ihr in der Warteschlange steht (Queue-Aware Prompting).
    import uuid as _uuid
    req_id = _uuid.uuid4().hex
    _project_pending.setdefault(pid, []).append({
        "id": req_id, "message": body.message, "source": body.source,
    })
    try:
        async with lock:
            # Snapshot: was liegt NACH mir in der Queue?
            after_me = [
                e["message"] for e in _project_pending.get(pid, [])
                if e["id"] != req_id
            ]
            a = agent()
            try:
                # Sync-Aufruf im Executor damit wir den Event-Loop nicht blocken —
                # chat() macht HTTP-Calls (Proxy) die 30-60s dauern koennen.
                loop = asyncio.get_running_loop()
                reply = await loop.run_in_executor(
                    None,
                    lambda: a.chat(
                        body.message, source=body.source, project_id=pid,
                        pending_queue=after_me,
                    ),
                )
            except ValueError as exc:
                raise HTTPException(400, str(exc))
            except RuntimeError as exc:
                logger.error("chat fehlgeschlagen: %s", exc)
                raise HTTPException(502, str(exc))

            needs_distill = a.conversation.needs_distill()
            if needs_distill:
                background.add_task(a.distill_old_turns)
            return ChatOut(
                reply=reply,
                turns=len(a.conversation.turns),
                distilling=needs_distill,
                events=a.pop_events(),
                project_id=pid,
            )
    finally:
        _project_pending[pid] = [
            e for e in _project_pending.get(pid, []) if e["id"] != req_id
        ]


@app.get("/projects/queue-status")
def projects_queue_status():
    """Snapshot: fuer jeden Projekt-Kontext (inkl. Hauptchat unter __main__)
    - busy: True wenn gerade ein Request in Verarbeitung
    - queue_size: wieviele weitere warten dahinter"""
    return {"contexts": _project_queue_snapshot()}


# ── Projekte ────────────────────────────────────────────────────────

@app.get("/projects/status")
def projects_status():
    """Komplett-Status: aktives Projekt + Liste aller (nicht-archivierten)."""
    return projects_mod.status()


@app.get("/projects/list")
def projects_list(include_archived: bool = False):
    return {"projects": projects_mod.list_projects(include_archived=include_archived)}


class ProjectCreateBody(BaseModel):
    name: str
    description: str = ""


@app.post("/projects/create")
def projects_create(body: ProjectCreateBody):
    try:
        p = projects_mod.create_project(body.name, body.description)
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc))
    return p


class ProjectSwitchBody(BaseModel):
    project_id: str = ""


@app.post("/projects/switch")
def projects_switch(body: ProjectSwitchBody):
    """Aktive Projekt-ID setzen. Leerer String → Hauptthread."""
    if body.project_id:
        p = projects_mod.get_project(body.project_id)
        if not p:
            raise HTTPException(status_code=404, detail=f"Projekt {body.project_id} nicht gefunden")
    projects_mod.set_active(body.project_id)
    return projects_mod.status()


@app.post("/projects/{project_id}/end")
def projects_end(project_id: str):
    if not projects_mod.end_project(project_id):
        raise HTTPException(status_code=404, detail=f"Projekt {project_id} nicht gefunden")
    return projects_mod.get_project(project_id) or {"id": project_id, "status": "ended"}


@app.post("/projects/{project_id}/archive")
def projects_archive(project_id: str):
    if not projects_mod.archive_project(project_id):
        raise HTTPException(status_code=404, detail=f"Projekt {project_id} nicht gefunden")
    return {"id": project_id, "status": "archived"}


class ProjectUpdateBody(BaseModel):
    name: Optional[str] = None
    description: Optional[str] = None


@app.patch("/projects/{project_id}")
def projects_update(project_id: str, body: ProjectUpdateBody):
    patch = body.dict(exclude_unset=True)
    p = projects_mod.update_project(project_id, patch)
    if p is None:
        raise HTTPException(status_code=404, detail=f"Projekt {project_id} nicht gefunden")
    return p


@app.get("/conversation/stats")
def conversation_stats():
    return conversation().stats()


@app.post("/conversation/reset")
def conversation_reset():
    """Hardes Reset — der Rolling-Window-Verlauf wird komplett geleert.
    Destillierte facts bleiben in der DB."""
    conversation().reset()
    return {"ok": True, "turns": 0}


class ConvDeleteBody(BaseModel):
    role: str
    content: str
    ts_iso_hint: Optional[str] = None


@app.post("/conversation/delete-turn")
def conversation_delete_turn(body: ConvDeleteBody):
    """Entfernt einen einzelnen Turn aus dem Rolling-Window + jsonl.
    Match per role + content (erstes Vorkommen wenn ts_iso_hint None,
    sonst nahester zur Zeit). 404 wenn kein Match.

    POST statt DELETE weil FastAPI 0.115 keine Bodys auf DELETE
    erlaubt — semantisch trotzdem eine Loeschung."""
    ok = conversation().remove_by_match(
        role=body.role, content=body.content, ts_iso_hint=body.ts_iso_hint,
    )
    if not ok:
        raise HTTPException(404, "Turn mit diesem role+content nicht gefunden")
    return {"ok": True, "turns": len(conversation().turns)}


@app.post("/conversation/distill")
def conversation_distill_now():
    """Manueller Trigger fuer Destillat — fuer Tests oder vor einem
    bewussten Reset."""
    return agent().distill_old_turns()


# ─── Call-Metrics (Token / Quota-Monitoring) ────────────────────────

@app.get("/metrics/calls")
def metrics_calls():
    """Liefert Aggregate fuer 1h / 5h / 24h / 30d.
    Jedes Window: {window_seconds, calls, tokens_in, tokens_out, by_model}."""
    return metrics_mod.stats()


# ─── Triggers (passive Aufweck-Quellen) ─────────────────────────────

class TriggerTimerBody(BaseModel):
    name: str
    fires_at: str  # ISO timestamp
    message: str
    author: str = "stefan"


class TriggerWatcherBody(BaseModel):
    name: str
    condition: str
    message: str
    check_interval_sec: int = 300
    throttle_sec: int = 3600
    author: str = "stefan"


class TriggerPatch(BaseModel):
    active: bool | None = None
    message: str | None = None
    condition: str | None = None
    throttle_sec: int | None = None
    check_interval_sec: int | None = None
    fires_at: str | None = None


@app.get("/triggers/list")
def triggers_list(active_only: bool = False):
    return {"triggers": triggers_mod.list_triggers(active_only=active_only)}


@app.post("/triggers/check-now")
async def triggers_check_now():
    """Sofortiger Trigger-Check, statt auf den naechsten Background-Tick
    zu warten. Wird von der Bridge nach jedem location_update gerufen
    damit GPS-Watcher (near()) den frischen Wert SOFORT sehen — bei
    Auto-Vorbeifahrt durch einen 300m-Radius hat man sonst nur ~20s
    Drinnen-Zeit, was unter TICK_SEC fallen kann."""
    return await background_mod.tick_now()


@app.get("/triggers/conditions")
def triggers_conditions():
    """Verfuegbare Variablen + Funktionen fuer Watcher-Conditions
    (mit aktuellen Werten)."""
    current = watcher_mod.collect_variables()
    # near() ist ein callable in vars_ — fuer die UI rausfiltern
    serializable = {k: v for k, v in current.items() if not callable(v)}
    return {
        "variables": watcher_mod.describe_variables(),
        "functions": watcher_mod.describe_functions(),
        "current": serializable,
    }


@app.get("/triggers/{name}")
def triggers_get(name: str):
    t = triggers_mod.read(name)
    if t is None:
        raise HTTPException(404, f"Trigger '{name}' nicht gefunden")
    return t


@app.get("/triggers/{name}/logs")
def triggers_get_logs(name: str, limit: int = 50):
    return {"logs": triggers_mod.list_logs(name, limit=limit)}


@app.post("/triggers/timer")
def triggers_create_timer(body: TriggerTimerBody):
    try:
        return triggers_mod.create_timer(
            name=body.name, fires_at_iso=body.fires_at,
            message=body.message, author=body.author,
        )
    except ValueError as exc:
        raise HTTPException(400, str(exc))


@app.post("/triggers/watcher")
def triggers_create_watcher(body: TriggerWatcherBody):
    try:
        return triggers_mod.create_watcher(
            name=body.name, condition=body.condition,
            message=body.message,
            check_interval_sec=body.check_interval_sec,
            throttle_sec=body.throttle_sec,
            author=body.author,
        )
    except ValueError as exc:
        raise HTTPException(400, str(exc))


@app.patch("/triggers/{name}")
def triggers_patch(name: str, body: TriggerPatch):
    patch = {k: v for k, v in body.model_dump().items() if v is not None}
    try:
        return triggers_mod.update(name, patch)
    except ValueError as exc:
        raise HTTPException(404, str(exc))


@app.delete("/triggers/{name}")
def triggers_delete(name: str):
    try:
        triggers_mod.delete(name)
    except ValueError as exc:
        raise HTTPException(404, str(exc))
    return {"deleted": name}


# ─── Skills ─────────────────────────────────────────────────────────

class SkillCreate(BaseModel):
    name: str
    description: str
    execution: str  # local-venv | local-bin | bash
    entry_code: str
    readme: str = ""
    args: list = Field(default_factory=list)
    requires: dict = Field(default_factory=dict)
    pip_packages: list = Field(default_factory=list)
    author: str = "stefan"
    config_schema: list = Field(default_factory=list)


class SkillRun(BaseModel):
    name: str
    args: dict = Field(default_factory=dict)
    timeout_sec: int = 300


class SkillPatch(BaseModel):
    description: str | None = None
    active: bool | None = None
    args: list | None = None
    entry_code: str | None = None
    readme: str | None = None
    pip_packages: list | None = None
    config_schema: list | None = None


class SkillConfigSet(BaseModel):
    values: dict


class SkillRollback(BaseModel):
    version_id: str


@app.get("/skills/list")
def skills_list(active_only: bool = False):
    return {"skills": skills_mod.list_skills(active_only=active_only)}


@app.get("/skills/{name}")
def skills_get(name: str):
    m = skills_mod.read_manifest(name)
    if m is None:
        raise HTTPException(404, f"Skill '{name}' nicht gefunden")
    readme = skills_mod.read_readme(name)
    return {"manifest": m, "readme": readme}


class SkillScaffold(BaseModel):
    name: str
    template: str  # oauth-api | apikey-api | file-process
    params: dict = Field(default_factory=dict)
    author: str = "stefan"


@app.get("/skills/templates")
def skills_templates_list():
    """Liste der verfuegbaren Templates — fuer UI und Dokumentation."""
    import skill_templates as st
    return {"templates": st.list_templates()}


@app.post("/skills/scaffold")
def skills_scaffold(body: SkillScaffold):
    """Baut einen Skill aus einem Template (oauth-api / apikey-api / file-process)."""
    try:
        return skills_mod.scaffold_skill(
            name=body.name, template=body.template,
            params=body.params, author=body.author,
        )
    except ValueError as exc:
        raise HTTPException(400, str(exc))


@app.post("/skills/create")
def skills_create(body: SkillCreate):
    try:
        return skills_mod.create_skill(
            name=body.name,
            description=body.description,
            execution=body.execution,
            entry_code=body.entry_code,
            readme=body.readme,
            args=body.args,
            requires=body.requires,
            pip_packages=body.pip_packages,
            author=body.author,
            config_schema=body.config_schema,
        )
    except ValueError as exc:
        raise HTTPException(400, str(exc))


@app.post("/skills/run")
def skills_run(body: SkillRun):
    try:
        return skills_mod.run_skill(body.name, args=body.args, timeout_sec=body.timeout_sec)
    except ValueError as exc:
        raise HTTPException(400, str(exc))


@app.patch("/skills/{name}")
def skills_patch(name: str, body: SkillPatch):
    patch = {k: v for k, v in body.model_dump().items() if v is not None}
    try:
        return skills_mod.update_skill(name, patch)
    except ValueError as exc:
        raise HTTPException(404, str(exc))


@app.delete("/skills/{name}")
def skills_delete(name: str):
    try:
        skills_mod.delete_skill(name)
    except ValueError as exc:
        raise HTTPException(404, str(exc))
    return {"deleted": name}


@app.get("/skills/{name}/logs")
def skills_logs(name: str, limit: int = 50):
    return {"logs": skills_mod.list_logs(name, limit=limit)}


# ── Skill-Configs (P3): statische Werte (API-Keys etc.) je Skill ───

@app.get("/skills/{name}/config")
def skills_config_get(name: str):
    """Liefert config_schema + aktuelle Werte (secret-Felder gemaskt mit
    '***SET***')."""
    manifest = skills_mod.read_manifest(name)
    if manifest is None:
        raise HTTPException(404, f"Skill '{name}' nicht gefunden")
    return {
        "schema": manifest.get("config_schema") or [],
        "values": skills_mod.get_skill_config_masked(name),
    }


@app.post("/skills/{name}/config")
def skills_config_set(name: str, body: SkillConfigSet):
    """Setzt Config-Werte (komplett ueberschreibend). Werte greifen ab dem
    naechsten skill_run. Secret-Felder werden in der Antwort gemaskt."""
    manifest = skills_mod.read_manifest(name)
    if manifest is None:
        raise HTTPException(404, f"Skill '{name}' nicht gefunden")
    skills_mod.set_skill_config(name, body.values)
    return {"ok": True, "values": skills_mod.get_skill_config_masked(name)}


# ── Skill-Versions (P4): rollback ──────────────────────────────────

@app.get("/skills/{name}/versions")
def skills_versions_list(name: str):
    if skills_mod.read_manifest(name) is None:
        raise HTTPException(404, f"Skill '{name}' nicht gefunden")
    return {"versions": skills_mod.list_skill_versions(name)}


@app.post("/skills/{name}/rollback")
def skills_rollback(name: str, body: SkillRollback):
    try:
        return skills_mod.rollback_skill(name, body.version_id)
    except ValueError as exc:
        raise HTTPException(404, str(exc))


@app.delete("/skills/{name}/versions/{version_id}")
def skills_versions_delete(name: str, version_id: str):
    try:
        return skills_mod.delete_skill_version(name, version_id)
    except ValueError as exc:
        raise HTTPException(404, str(exc))


@app.get("/skills/{name}/export")
def skills_export(name: str):
    try:
        data = skills_mod.export_skill(name)
    except ValueError as exc:
        raise HTTPException(404, str(exc))
    return Response(
        content=data,
        media_type="application/gzip",
        headers={"Content-Disposition": f'attachment; filename="skill-{name}.tar.gz"'},
    )


@app.post("/skills/import")
async def skills_import(request: Request, overwrite: bool = False):
    data = await request.body()
    if not data:
        raise HTTPException(400, "Leerer Body")
    try:
        manifest = skills_mod.import_skill(data, overwrite=overwrite)
    except ValueError as exc:
        raise HTTPException(400, str(exc))
    return {"imported": manifest}


# ── OAuth ─────────────────────────────────────────────────────────


@app.get("/oauth/services")
async def oauth_services_list():
    """Liste aller Services mit Status (configured/authenticated/expires)."""
    return {"services": oauth_mod.list_services()}


@app.get("/oauth/apps")
async def oauth_apps_get():
    """Liefert die persistierte Provider-Config (client_id sichtbar, client_secret
    NICHT — wer den Wert braucht muss ihn neu eintragen). Fuer Diagnostic-UI."""
    apps = oauth_mod._load_json(oauth_mod.APPS_FILE)
    safe = {}
    for service, entry in apps.items():
        if not isinstance(entry, dict):
            continue
        safe[service] = {
            "client_id": entry.get("client_id", ""),
            "has_client_secret": bool(entry.get("client_secret")),
            "scopes": entry.get("scopes"),
            "auth_url": entry.get("auth_url"),
            "token_url": entry.get("token_url"),
        }
    return {"apps": safe, "defaults": list(oauth_mod.DEFAULT_PROVIDERS.keys())}


class OAuthAppIn(BaseModel):
    service: str
    client_id: str = ""
    client_secret: str = ""
    scopes: Optional[List[str]] = None
    auth_url: Optional[str] = None
    token_url: Optional[str] = None


@app.post("/oauth/apps")
async def oauth_apps_set(body: OAuthAppIn):
    """Speichert/aktualisiert eine Provider-Config. Leerer client_secret laesst
    den bestehenden Wert stehen (damit man die Form ohne Re-Eingabe absenden
    kann fuer reine scope-Aenderungen)."""
    service = (body.service or "").strip()
    if not service or not service.isidentifier() and not all(c.isalnum() or c in "_-" for c in service):
        raise HTTPException(400, "Ungueltiger service-Name (a-z0-9_- erlaubt)")
    apps = oauth_mod._load_json(oauth_mod.APPS_FILE)
    entry = apps.get(service) or {}
    if body.client_id:
        entry["client_id"] = body.client_id.strip()
    if body.client_secret:
        entry["client_secret"] = body.client_secret.strip()
    if body.scopes is not None:
        entry["scopes"] = body.scopes
    if body.auth_url:
        entry["auth_url"] = body.auth_url.strip()
    if body.token_url:
        entry["token_url"] = body.token_url.strip()
    apps[service] = entry
    oauth_mod._save_json(oauth_mod.APPS_FILE, apps)
    logger.info("OAuth-App %s gespeichert (client_id=%s, has_secret=%s)",
                service, entry.get("client_id", ""), bool(entry.get("client_secret")))
    return {"ok": True, "service": service}


@app.delete("/oauth/apps/{service}")
async def oauth_apps_delete(service: str):
    apps = oauth_mod._load_json(oauth_mod.APPS_FILE)
    if service in apps:
        apps.pop(service)
        oauth_mod._save_json(oauth_mod.APPS_FILE, apps)
    # Token auch wegwerfen
    oauth_mod.revoke(service)
    return {"ok": True}


@app.post("/oauth/{service}/revoke")
async def oauth_revoke_endpoint(service: str):
    return {"ok": oauth_mod.revoke(service)}


@app.get("/oauth/{service}/token")
async def oauth_token_endpoint(service: str):
    """Liefert das aktuelle access_token fuer einen Service (mit Auto-Refresh
    wenn < 60s Restzeit). Nur fuer interne Skill-Aufrufe gedacht — Skills
    sollen NIEMALS hardcoded client_secrets haben, sondern dieses Endpoint
    pollen. Antwort: {access_token, expires_at, expires_in_sec}.
    Bei nicht-autorisiert: 401 mit klarer Message."""
    try:
        rec = oauth_mod.get_token(service)
    except RuntimeError as exc:
        raise HTTPException(401, str(exc))
    expires_at = int(rec.get("expires_at") or 0)
    import time as _t
    return {
        "access_token": rec.get("access_token"),
        "expires_at": expires_at,
        "expires_in_sec": max(0, expires_at - int(_t.time())),
    }


class OAuthAuthorizeIn(BaseModel):
    service: str
    scopes: Optional[List[str]] = None


@app.post("/oauth/authorize")
async def oauth_authorize_endpoint(body: OAuthAuthorizeIn):
    """Baut eine Authorize-URL fuer einen Service. Diagnostic kann das nutzen
    um den Auth-Flow manuell anzustossen. ARIA selbst nutzt das Tool
    `oauth_authorize` (in agent._dispatch_tool gemapped auf die gleiche Logik)."""
    try:
        return oauth_mod.build_authorize_url(body.service, scopes=body.scopes)
    except RuntimeError as exc:
        raise HTTPException(400, str(exc))


@app.post("/internal/oauth-callback")
async def oauth_callback_internal(request: Request):
    """Wird von aria-bridge gerufen wenn ein RVS oauth_callback ankommt.
    Macht den state-Match + token-exchange und persistiert."""
    try:
        body = await request.json()
    except Exception as exc:
        raise HTTPException(400, f"bad json: {exc}")
    service = (body.get("service") or "").strip()
    code = (body.get("code") or "").strip()
    state = (body.get("state") or "").strip()
    err = body.get("error") or None
    err_desc = body.get("errorDescription") or None
    if not service:
        raise HTTPException(400, "service erforderlich")
    result = oauth_mod.handle_callback(service, code, state, error=err, error_description=err_desc)
    return result
