flc/crewboard

Crewboard - team task board (boards, lists, cards, roles, history, drag and drop, MCP, AI extras). FastAPI + DataK3 + appid + Models.

• dodil-challenge
• dodil-challenge-boards

A remix template · 1 remixes · Remix flc/crewboard

app/boards/infrastructure/dodil.py

"""The Dodil side: a service-account token, DataK3 on the Postgres wire, and Dodil Models.

The remix generated the service account and put its id and secret into the repo's CI secrets;
.dodil/deploy.yaml hands them to the app as environment variables.
"""
import os
import threading
import time
from contextlib import contextmanager

import httpx
import psycopg

TOKEN_URL = "https://id.dodil.io/realms/dodil/protocol/openid-connect/token"
PG_HOST = os.environ.get("PG_HOST", "pg.uk-lon-1.dodil.io")
MODELS_BASE = os.environ.get("MODELS_BASE", "https://api.dodil.io/v1").rstrip("/")
CHAT_MODEL = os.environ.get("CHAT_MODEL", "kimi-k2.6")
EMBED_MODEL = os.environ.get("EMBED_MODEL", "jina-embeddings-v4")
# DataK3 serves 16 concurrent reads per bucket: stay below it, and retry the transient codes.
_slots = threading.BoundedSemaphore(int(os.environ.get("DB_CONCURRENCY", "8")))
RETRY_CODES = {"40001", "08006", "58030", "XX000"}

_lock, _tok = threading.Lock(), {"v": None, "exp": 0.0}


def token() -> str:
    """A short-lived service-account token (client_credentials), cached until near expiry.
    It is the DataK3 password AND the Models bearer."""
    with _lock:
        if _tok["v"] and time.time() < _tok["exp"] - 60:
            return _tok["v"]
        r = httpx.post(TOKEN_URL, timeout=15, data={
            "grant_type": "client_credentials",
            "client_id": os.environ["DODIL_SERVICE_ACCOUNT_ID"],
            "client_secret": os.environ["DODIL_SERVICE_ACCOUNT_SECRET"]})
        r.raise_for_status()
        d = r.json()
        _tok.update(v=d["access_token"], exp=time.time() + float(d.get("expires_in", 300)))
        return _tok["v"]


_idle = []          # open connections waiting for reuse: (connection, opened at)
MAX_AGE = 240       # seconds; reconnect before the token used as password gets old


def _open() -> psycopg.Connection:
    for attempt in range(3):
        try:
            return psycopg.connect(host=PG_HOST, port=5432, dbname=os.environ["BUCKET"],
                                   user="token", password=token(), sslmode="require",
                                   autocommit=True, connect_timeout=15)
        except psycopg.OperationalError as e:
            if attempt == 2 or (getattr(e, "sqlstate", None) or "08006") not in RETRY_CODES:
                raise
            time.sleep(0.3 * (attempt + 1))


@contextmanager
def connect():
    """DataK3 speaks Postgres: the database is the bucket, TLS is required. Connections are
    reused (a TLS handshake per query is slow); a broken or old one is dropped, never reused."""
    with _slots:
        conn = None
        with _lock:
            while _idle and conn is None:
                c, opened = _idle.pop()
                if c.closed or time.time() - opened > MAX_AGE:
                    c.close()
                else:
                    conn = (c, opened)
        if conn is None:
            conn = (_open(), time.time())
        try:
            yield conn[0]
        except psycopg.OperationalError:
            conn[0].close()
            raise
        finally:
            if not conn[0].closed:
                with _lock:
                    _idle.append(conn)


def _unwrap(body: dict) -> dict:
    """The gateway wraps the OpenAI body as {"status": ..., "data": {...}}."""
    inner = body.get("data")
    return inner if isinstance(inner, dict) else body


MAX_TOKENS = int(os.environ.get("CHAT_MAX_TOKENS", "6000"))


def chat(prompt: str) -> str:
    """One chat call to Dodil Models (OpenAI-compatible). kimi-k2.6 thinks before it answers and
    the default cap (2048 tokens) can end mid-thought with an empty answer, so the cap is raised;
    a 5xx ("temporarily unavailable") is retried once. Returns "" when no answer came back."""
    for attempt in range(2):
        r = httpx.post(f"{MODELS_BASE}/chat/completions", timeout=180,
                       headers={"Authorization": f"Bearer {token()}"},
                       json={"model": CHAT_MODEL, "max_tokens": MAX_TOKENS,
                             "messages": [{"role": "user", "content": prompt}]})
        if r.status_code >= 500 and attempt == 0:
            time.sleep(2)
            continue
        r.raise_for_status()
        return (_unwrap(r.json())["choices"][0]["message"].get("content") or "").strip()
    return ""


def embed(texts: list) -> list:
    r = httpx.post(f"{MODELS_BASE}/embeddings", timeout=120,
                   headers={"Authorization": f"Bearer {token()}"},
                   json={"model": EMBED_MODEL, "input": texts})
    r.raise_for_status()
    rows = sorted(_unwrap(r.json())["data"], key=lambda d: d.get("index", 0))
    return [d["embedding"] for d in rows]


class ModelsAssistant:
    def checklist(self, card_title: str, description: str = "") -> list:
        text = chat("Break this task into a short checklist of 3 to 6 concrete steps. "
                    "One step per line, no numbering, nothing else.\n\nTask: " + card_title +
                    ("\nDetails: " + description[:1500] if description else ""))
        return [line.strip("-*• []").strip() for line in text.splitlines() if line.strip()][:8]

    def summary(self, facts: str) -> str:
        text = chat("Write a short daily stand-up summary of this team board for its members: "
                    "at most 6 sentences, plain text, name the cards by id, say what moved, what "
                    "is late and what is stuck. Use only these facts; do not invent anything."
                    "\n\n" + facts)
        # No answer from the model: the facts themselves are still a true summary.
        return text or "(AI summary unavailable right now; the board's facts:)\n" + facts

    def embed(self, texts: list) -> list:
        out = []
        for i in range(0, len(texts), 32):
            out.extend(embed(texts[i:i + 32]))
        return out