dachdigitalwerksorg/kanvaflow

KanvaFlow — Remixable team task board with RBAC, drag-and-drop Kanban, dependencies, Dodil Models AI, and MCP tools.

• dodil-challenge
• dodil-challenge-boards

A remix template · 0 remixes · Remix dachdigitalwerksorg/kanvaflow

app/boards/infrastructure/dodil.py

"""Dodil Cloud Adapters: service-account OAuth2 token, DataK3 Postgres wire, and Dodil Models AI.

All credentials come from environment variables populated automatically by `.dodil/remix.yaml`
and `.dodil/deploy.yaml` (`DODIL_SERVICE_ACCOUNT_ID`, `DODIL_SERVICE_ACCOUNT_SECRET`, `BUCKET`).
"""
import os
import threading
import time
from typing import Optional

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")

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


def token() -> str:
    """Short-lived service-account token (client_credentials), cached until 60s before expiry."""
    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"]


def connect() -> psycopg.Connection:
    """Connect to DataK3 over the Postgres wire (`sslmode=require`, database = bucket name)."""
    return psycopg.connect(
        host=PG_HOST,
        port=5432,
        dbname=os.environ["BUCKET"],
        user="token",
        password=token(),
        sslmode="require",
        autocommit=True,
    )


def chat(prompt: str) -> str:
    """Call Dodil Models chat completion (`kimi-k2.6`), unwrapping the gateway envelope."""
    r = httpx.post(
        f"{MODELS_BASE}/chat/completions",
        timeout=90,
        headers={"Authorization": f"Bearer {token()}"},
        json={"model": CHAT_MODEL, "messages": [{"role": "user", "content": prompt}]},
    )
    r.raise_for_status()
    body = r.json()
    inner = body.get("data")
    if isinstance(inner, dict) and "choices" in inner:
        body = inner
    return body["choices"][0]["message"]["content"] or ""


def embed_text(text: str) -> Optional[list[float]]:
    """Call Dodil Models embeddings (`jina-embeddings-v4`), unwrapping the gateway envelope."""
    clean = (text or "").strip()
    if not clean:
        return None
    r = httpx.post(
        f"{MODELS_BASE}/embeddings",
        timeout=25,
        headers={"Authorization": f"Bearer {token()}"},
        json={"model": EMBED_MODEL, "input": clean[:2000]},
    )
    r.raise_for_status()
    body = r.json()
    inner = body.get("data")
    if isinstance(inner, dict) and "data" in inner:
        body = inner
    arr = body.get("data") or []
    if arr and isinstance(arr[0], dict) and isinstance(arr[0].get("embedding"), list):
        return [float(x) for x in arr[0]["embedding"]]
    return None


class ModelsAssistant:
    def checklist(self, card_title: str, description: str = "") -> list[str]:
        ctx = f"Task: {card_title}"
        if description:
            ctx += f"\nContext: {description}"
        text = chat(
            "Break this task into a concise, actionable checklist of 3 to 6 concrete steps. "
            "Output one step per line, no numbering, no markdown intro or outro.\n\n" + ctx
        )
        return [line.strip("-*•0123456789.) ").strip() for line in text.splitlines() if line.strip()][:8]

    def summarize_board(self, board_title: str, snapshot: str) -> str:
        return chat(
            f"Write a crisp 3-5 sentence daily standup summary for the team board {board_title!r}. "
            "Highlight overall momentum, call out any overdue or blocked items with owners, "
            "and suggest the single highest-leverage focus for today.\n\n" + snapshot
        ).strip()

    def embed(self, text: str) -> Optional[list[float]]:
        return embed_text(text)