ghpsolutions/quadro

Quadro: team boards with roles, history and an MCP endpoint for Claude and ChatGPT

• dodil-challenge
• dodil-challenge-boards

A remix template · 0 remixes · Remix ghpsolutions/quadro

app/boards/infrastructure/dodil.py

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

The remix creates the service account and stores its id and secret as CI secrets;
.dodil/deploy.yaml hands them to the app as environment variables.
"""
from __future__ import annotations

import os
import threading
import time

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")
MODEL_TIMEOUT = float(os.environ.get("MODEL_TIMEOUT", "50"))

_lock, _tok = threading.Lock(), {"v": None, "exp": 0.0}
_http = httpx.Client(timeout=MODEL_TIMEOUT)


def token() -> str:
    """A short-lived service-account token (client_credentials), cached until near expiry.
    It is both the DataK3 password and the Models bearer."""
    with _lock:
        if _tok["v"] and time.time() < _tok["exp"] - 60:
            return _tok["v"]
        r = _http.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"]


def connect() -> psycopg.Connection:
    """DataK3 speaks Postgres: the database is the bucket, the user is 'token', TLS is required.
    DATABASE_URL overrides this for running the SQL repository against a local Postgres."""
    url = os.environ.get("DATABASE_URL")
    if url:
        return psycopg.connect(url, autocommit=True)
    return psycopg.connect(host=PG_HOST, port=5432, dbname=os.environ["BUCKET"], user="token",
                           password=token(), sslmode="require", autocommit=True,
                           connect_timeout=15)


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


def chat(prompt: str, max_tokens: int = 4096) -> str:
    """One chat call to Dodil Models. The model thinks before it answers, so it gets a generous
    token cap, and an empty answer is retried once."""
    for attempt in range(2):
        r = _http.post(f"{MODELS_BASE}/chat/completions",
                       headers={"Authorization": f"Bearer {token()}"},
                       json={"model": CHAT_MODEL, "max_tokens": max_tokens,
                             "messages": [{"role": "user", "content": prompt}]})
        r.raise_for_status()
        text = (_unwrap(r.json())["choices"][0]["message"].get("content") or "").strip()
        if text:
            return text
    return ""


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


class ModelsAssistant:
    def checklist(self, title: str, description: str) -> list:
        text = chat("Break this task into a short checklist of 3 to 7 concrete, checkable steps. "
                    "Answer with one step per line, no numbering, no extra text.\n\n"
                    f"Task: {title}\nDetails: {description or '(none)'}", max_tokens=3000)
        steps = [line.strip().lstrip("-*•0123456789.) ").strip() for line in text.splitlines()]
        return [s for s in steps if s and not s.endswith(":")][:8]

    def summary(self, facts: str) -> str:
        return chat("Write a short daily stand-up summary of this task board for the team. "
                    "Plain text, at most 6 short lines: what moved, what is overdue or blocked, "
                    "and what needs attention today. Use card references like OPS-12. "
                    "Only use these facts:\n\n" + facts, max_tokens=3000)

    def embed(self, texts: list) -> list:
        return embed(texts)