sirenamc/stackline

Stackline, a team task board you run from the web or from Claude/ChatGPT. FastAPI + DataK3 + appid + Models + MCP.

A remix template · 0 remixes · Remix sirenamc/stackline

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 json
import re

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.7-code")
EMBED_MODEL = os.environ.get("EMBED_MODEL", "jina-embeddings-v4")

_lock, _tok = threading.Lock(), {"v": None, "exp": 0.0}
_local = threading.local()
# DataK3 serves 16 concurrent reads per bucket; stay well under it, whatever the thread pool does.
_slots = threading.BoundedSemaphore(int(os.environ.get("DB_CONCURRENCY", "8")))


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


def connect() -> psycopg.Connection:
    """DataK3 speaks Postgres: the database is the bucket, TLS is required."""
    return psycopg.connect(host=PG_HOST, port=5432, dbname=os.environ["BUCKET"], user="token",
                           password=token(), sslmode="require", autocommit=True)


@contextmanager
def db():
    """One connection per worker thread, reused while its token is fresh. A TLS handshake per
    query made the board page slow; a broken connection is dropped and the next call reconnects."""
    tok = token()
    with _slots:
        conn = _connection(tok)
        try:
            yield conn
        except (psycopg.OperationalError, psycopg.InterfaceError):
            conn.close()
            _local.conn = None
            raise


def _connection(tok: str) -> psycopg.Connection:
    conn = getattr(_local, "conn", None)
    if conn is None or conn.closed or getattr(_local, "tok", None) != tok:
        if conn is not None and not conn.closed:
            conn.close()
        conn, _local.tok = connect(), tok
        _local.conn = conn
    return conn


def _models(path: str, body: dict) -> dict:
    """Dodil Models is OpenAI-compatible. The gateway wraps the OpenAI body as
    {"status": ..., "data": {...}}, so unwrap it.

    Ignite gives a request 60 s. Measured answer times for chat models range from 2 to 40 s with
    the odd 500, so: 40 s per call, and one retry only when the first attempt failed fast."""
    started = time.time()
    for attempt in (1, 2):
        try:
            r = httpx.post(f"{MODELS_BASE}{path}", timeout=40, json=body,
                           headers={"Authorization": f"Bearer {token()}"})
            if r.status_code < 500:
                break
        except httpx.TimeoutException:
            if attempt == 2 or time.time() - started > 15:
                raise
        if attempt == 2 or time.time() - started > 15:
            break
    r.raise_for_status()
    out = r.json()
    inner = out.get("data")
    return inner if isinstance(inner, dict) else out


def chat(prompt: str, system: str = "", max_tokens: int = 0) -> str:
    messages = ([{"role": "system", "content": system}] if system else []) + [
        {"role": "user", "content": prompt}]
    request = {"model": CHAT_MODEL, "messages": messages}
    if max_tokens:
        request["max_tokens"] = max_tokens  # a cap on answer length is a cap on answer time
    body = _models("/chat/completions", request)
    return body["choices"][0]["message"]["content"] or ""


class ModelsAssistant:
    def checklist(self, title: str, description: str) -> list:
        task = title + (f"\n\nDetails: {description}" if description else "")
        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: " + task)
        return [line.strip("-*• ").strip() for line in text.splitlines() if line.strip()][:8]

    def summary(self, facts: str) -> str:
        return chat(facts, system=(
            "You write a team's daily stand-up for a task board. Use only the facts given. "
            "At most 6 short lines: what moved, what is late and who owns it, what is blocked. "
            "Refer to cards as #number. No greeting, no advice, no markdown headings."))

    def plan(self, goal: str, deadline: str, today: str) -> list:
        """A first plan for set_up_board. The model only drafts: the use case checks every step
        before anything is written."""
        text = chat(
            f"Goal: {goal}\nToday: {today}. Deadline: {deadline}.",
            system=(
                "You plan work for a small team's task board. Reply with a JSON array only, "
                "6 to 10 steps in the order they happen. Each step: "
                '{"title": short imperative under 80 characters, '
                '"days_before": whole days before the deadline it is due (0 = on the deadline), '
                '"after": [numbers of earlier steps it waits on, 1 = the first step]}. '
                "The last step is the goal itself. No commentary, no markdown."),
            max_tokens=900)
        match = re.search(r"\[.*\]", text, re.S)
        steps = json.loads(match.group(0)) if match else []
        return [s for s in steps if isinstance(s, dict) and s.get("title")][:15]

    def embed(self, texts: list) -> list:
        body = _models("/embeddings", {"model": EMBED_MODEL, "input": texts})
        rows = sorted(body["data"], key=lambda d: d.get("index", 0))
        return [d["embedding"] for d in rows]