feat: add phylo router (NJ/UPGMA/ML) with PhyML binary

Dockerfile

@@ -0,0 +1,27 @@
+FROM python:3.11-slim
+
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    build-essential gcc autoconf automake pkg-config wget && \
+    rm -rf /var/lib/apt/lists/*
+
+# Build PhyML from source (~2 min)
+RUN wget -qO /tmp/phyml.tar.gz \
+      https://github.com/stephaneguindon/phyml/archive/refs/tags/v3.3.20250515.tar.gz && \
+    tar xzf /tmp/phyml.tar.gz -C /tmp && \
+    cd /tmp/phyml-3.3.20250515 && \
+    ./autogen.sh && \
+    ./configure --enable-phyml && \
+    make -j$(nproc) && \
+    make install && \
+    cd / && \
+    rm -rf /tmp/phyml-3.3.20250515 /tmp/phyml.tar.gz
+
+WORKDIR /app
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+# primer3-py is optional (requires C compiler) — skip silently if it fails
+RUN pip install --no-cache-dir primer3-py>=2.0.3 2>/dev/null || echo "primer3-py skipped (optional)"
+COPY . .
+
+EXPOSE 7860
+CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "7860"]

app/main.py

@@ -0,0 +1,114 @@
+import logging
+from dotenv import load_dotenv
+load_dotenv()
+
+from fastapi import FastAPI, HTTPException, Request
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import JSONResponse
+from slowapi import Limiter, _rate_limit_exceeded_handler
+from slowapi.util import get_remote_address
+from slowapi.errors import RateLimitExceeded
+from app.config import settings
+from app.routers import pipelines, pipeline_v2, ai, jobs, share, profile, sequences, uniprot, alignment, structures, pathways, domains, interactions, primers, structure_analysis, phylo
+from app.services.cache import init_redis
+
+logger = logging.getLogger(__name__)
+
+limiter = Limiter(key_func=get_remote_address, default_limits=["30/minute"])
+
+app = FastAPI(title="Bio Nexus API", version="0.2.0")
+app.state.limiter = limiter
+app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
+
+PROD_ORIGIN = settings.CORS_ORIGIN
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=[
+        "http://localhost:3000",
+        "http://localhost:3001",
+        PROD_ORIGIN,
+        "https://bioai-platform.vercel.app",
+    ],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+app.include_router(pipelines.router, prefix="/api/pipelines", tags=["pipelines"])
+app.include_router(pipeline_v2.router, prefix="/api/pipeline/v2", tags=["pipeline_v2"])
+app.include_router(ai.router, prefix="/api/ai", tags=["ai"])
+app.include_router(jobs.router, prefix="/api/jobs", tags=["jobs"])
+app.include_router(share.router, prefix="/api/share", tags=["share"])
+app.include_router(profile.router, prefix="/api/profile", tags=["profile"])
+app.include_router(sequences.router, prefix="/api/sequences", tags=["sequences"])
+app.include_router(uniprot.router, prefix="/api/uniprot", tags=["uniprot"])
+app.include_router(alignment.router, prefix="/api/alignment", tags=["alignment"])
+app.include_router(structures.router, prefix="/api/structures", tags=["structures"])
+app.include_router(pathways.router, prefix="/api/pathways", tags=["pathways"])
+app.include_router(domains.router)
+app.include_router(interactions.router)
+app.include_router(primers.router)
+app.include_router(structure_analysis.router)
+app.include_router(phylo.router)
+
+TERMINAL_STATUSES = {"complete", "failed"}
+NON_TERMINAL_STATUSES = {
+    "submitted_to_ncbi", "polling_ncbi", "parsing",
+    "fetching_uniprot", "fetching_alphafold", "interpreting",
+}
+
+
+async def _fail_stuck_jobs():
+    try:
+        import httpx
+        from app.config import settings
+        headers = {
+            "apikey": settings.SUPABASE_SERVICE_ROLE_KEY,
+            "Authorization": f"Bearer {settings.SUPABASE_SERVICE_ROLE_KEY}",
+            "Content-Type": "application/json",
+            "Prefer": "return=minimal",
+        }
+        url = f"{settings.SUPABASE_URL}/rest/v1/jobs"
+        quoted = ",".join(f'"{s}"' for s in NON_TERMINAL_STATUSES)
+        select_url = f"{url}?select=id&status=in.({quoted})"
+        async with httpx.AsyncClient(timeout=10) as client:
+            resp = await client.get(select_url, headers=headers)
+            if resp.status_code != 200:
+                logger.warning(f"Startup resume: failed to query jobs ({resp.status_code})")
+                return
+            stuck = resp.json()
+            for job in stuck:
+                jid = job["id"]
+                logger.info(f"Startup resume: marking stuck job {jid} as failed")
+                await client.patch(
+                    f"{url}?id=eq.{jid}",
+                    headers=headers,
+                    json={"status": "failed", "error": "Worker lost on restart — please re-run"},
+                )
+            if stuck:
+                logger.info(f"Startup resume: marked {len(stuck)} stuck job(s) as failed")
+    except Exception as e:
+        logger.warning(f"Startup resume: error: {e}")
+
+
+@app.on_event("startup")
+async def startup():
+    init_redis()
+    await _fail_stuck_jobs()
+
+
+@app.get("/health")
+async def health():
+    return {"status": "ok"}
+
+
+@app.exception_handler(Exception)
+async def global_exception_handler(request: Request, exc: Exception):
+    if isinstance(exc, HTTPException):
+        return JSONResponse(status_code=exc.status_code, content={"detail": exc.detail})
+    logger.exception("Unhandled exception")
+    return JSONResponse(
+        status_code=500,
+        content={"detail": "Internal server error"},
+    )

app/routers/phylo.py

@@ -0,0 +1,430 @@
+"""
+Phylogenetic tree router — three methods:
+  nj    : Neighbor-Joining via Clustal Omega guide tree (~60s)
+  upgma : UPGMA computed locally from Clustal alignment (adds ~0s after MSA)
+  ml    : Maximum Likelihood via local PhyML binary (~3-5 min, includes bootstrap)
+
+PhyML binary must be installed at build time (see Dockerfile).
+Download source from: https://github.com/stephaneguindon/phyml
+Compile: ./configure --enable-phyml && make && make install
+
+Job lifecycle (in-memory, thread-safe):
+  queued -> msa_running -> msa_done -> tree_running -> complete | error
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import threading
+import time
+import uuid
+from typing import Literal, Optional
+
+import httpx
+from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException
+from pydantic import BaseModel, Field
+
+logger = logging.getLogger(__name__)
+router = APIRouter(prefix="/phylo", tags=["phylo"])
+
+# ── EBI base URLs ──────────────────────────────────────────────────────────────
+_EBI_CLUSTALO = "https://www.ebi.ac.uk/Tools/services/rest/clustalo"
+_EMAIL        = "bionexus@demo.com"
+
+# ── PhyML protein models (most-used first) ────────────────────────────────────
+PROTEIN_MODELS = ["LG", "WAG", "JTT", "Blosum62", "MtREV", "Dayhoff"]
+DNA_MODELS     = ["GTR", "HKY85", "K80", "F81", "TN93", "SYM"]
+
+# ─── Models ───────────────────────────────────────────────────────────────────
+
+Method      = Literal["nj", "ml", "upgma"]
+SeqType     = Literal["protein", "dna"]
+JobPhase    = Literal["queued", "msa_running", "msa_done", "tree_running", "complete", "error"]
+
+
+class PhyloRequest(BaseModel):
+    sequences: list[dict]
+    method: Method     = "nj"
+    seq_type: SeqType  = "protein"
+    # ML-only options
+    model: str         = "LG"
+    bootstrap: int     = Field(100, ge=0, le=1000)
+
+
+class PhyloJob(BaseModel):
+    job_id: str
+    method: Method
+    seq_type: SeqType
+    model: Optional[str]
+    bootstrap: Optional[int]
+    phase: JobPhase
+    aln_fasta:   Optional[str] = None
+    newick:      Optional[str] = None
+    stats:       Optional[str] = None
+    error:       Optional[str] = None
+    created_at:  float = 0.0
+    msa_done_at: Optional[float] = None
+    done_at:     Optional[float] = None
+
+
+class RunResponse(BaseModel):
+    job_id: str
+    status: str
+
+
+# ─── In-memory store ──────────────────────────────────────────────────────────
+
+_jobs: dict[str, dict] = {}
+_lock = threading.Lock()
+
+
+def _init(job_id: str, req: PhyloRequest) -> None:
+    with _lock:
+        _jobs[job_id] = {
+            "job_id":    job_id,
+            "method":    req.method,
+            "seq_type":  req.seq_type,
+            "model":     req.model,
+            "bootstrap": req.bootstrap,
+            "phase":     "queued",
+            "aln_fasta": None,
+            "newick":    None,
+            "stats":     None,
+            "error":     None,
+            "created_at": time.time(),
+            "msa_done_at": None,
+            "done_at":   None,
+            "_req":      req.model_dump(),
+        }
+
+
+def _patch(job_id: str, **kw) -> None:
+    with _lock:
+        if job_id in _jobs:
+            _jobs[job_id].update(kw)
+
+
+def _read(job_id: str) -> dict | None:
+    with _lock:
+        return dict(_jobs[job_id]) if job_id in _jobs else None
+
+
+# ─── EBI helpers (same pattern as alignment.py) ───────────────────────────────
+
+async def _ebi_submit(url: str, data: dict) -> str:
+    async with httpx.AsyncClient(timeout=30) as client:
+        r = await client.post(url + "/run", data=data)
+        r.raise_for_status()
+        return r.text.strip()
+
+
+async def _ebi_poll(url: str, ebi_job: str, interval: float = 2.0, max_polls: int = 150) -> str:
+    async with httpx.AsyncClient(timeout=10) as client:
+        for _ in range(max_polls):
+            await asyncio.sleep(interval)
+            r = await client.get(f"{url}/status/{ebi_job}")
+            status = r.text.strip()
+            if status in ("FINISHED", "FAILED", "ERROR", "NOT_FOUND"):
+                return status
+    return "TIMEOUT"
+
+
+async def _ebi_result(url: str, ebi_job: str, result_type: str, retries: int = 3) -> str:
+    async with httpx.AsyncClient(timeout=60) as client:
+        for attempt in range(retries):
+            try:
+                r = await client.get(f"{url}/result/{ebi_job}/{result_type}")
+                if r.status_code == 200:
+                    return r.text
+                await asyncio.sleep(2 ** attempt)
+            except httpx.RequestError:
+                await asyncio.sleep(2 ** attempt)
+    raise RuntimeError(f"Failed to fetch EBI result {result_type} after {retries} attempts")
+
+
+# ─── MSA via Clustal Omega ────────────────────────────────────────────────────
+
+def _to_fasta(sequences: list[dict]) -> str:
+    return "\n".join(f">{s['id']}\n{s['sequence']}" for s in sequences)
+
+
+async def _run_clustalo(job_id: str, sequences: list[dict], stype: str) -> tuple[str, str] | None:
+    _patch(job_id, phase="msa_running")
+    fasta = _to_fasta(sequences)
+
+    try:
+        ebi_job = await _ebi_submit(_EBI_CLUSTALO, {
+            "email":    _EMAIL,
+            "sequence": fasta,
+            "outfmt":   "fa",
+            "stype":    "protein" if stype == "protein" else "dna",
+        })
+        logger.info(f"[{job_id}] Clustal Omega job: {ebi_job}")
+    except Exception as e:
+        _patch(job_id, phase="error", error=f"Clustal Omega submission failed: {e}")
+        return None
+
+    status = await _ebi_poll(_EBI_CLUSTALO, ebi_job)
+    if status != "FINISHED":
+        _patch(job_id, phase="error", error=f"Clustal Omega ended with status: {status}")
+        return None
+
+    try:
+        aln_fasta = await _ebi_result(_EBI_CLUSTALO, ebi_job, "fa")
+        nj_newick = await _ebi_result(_EBI_CLUSTALO, ebi_job, "phylotree")
+    except Exception as e:
+        _patch(job_id, phase="error", error=f"Clustal Omega result fetch failed: {e}")
+        return None
+
+    _patch(job_id, phase="msa_done", aln_fasta=aln_fasta, msa_done_at=time.time())
+    return aln_fasta, nj_newick
+
+
+# ─── UPGMA (pure Python) ──────────────────────────────────────────────────────
+
+def _parse_aligned_fasta(fasta: str) -> dict[str, str]:
+    seqs: dict[str, str] = {}
+    cur = None
+    for line in fasta.strip().splitlines():
+        stripped = line.strip()
+        if stripped.startswith(">"):
+            cur = stripped[1:].split()[0]
+            seqs[cur] = ""
+        elif cur:
+            seqs[cur] += stripped
+    return seqs
+
+
+def _p_distance(s1: str, s2: str) -> float:
+    pairs = [(a, b) for a, b in zip(s1, s2) if a != "-" and b != "-"]
+    if not pairs:
+        return 1.0
+    return sum(1 for a, b in pairs if a != b) / len(pairs)
+
+
+def _upgma_newick(aln_fasta: str) -> str:
+    seqs = _parse_aligned_fasta(aln_fasta)
+    names = list(seqs.keys())
+    n = len(names)
+
+    if n < 2:
+        return f"({names[0]}:0.0);" if names else "();"
+
+    dist: dict[tuple[str, str], float] = {}
+    for i in range(n):
+        for j in range(i + 1, n):
+            d = _p_distance(seqs[names[i]], seqs[names[j]])
+            dist[(names[i], names[j])] = d
+            dist[(names[j], names[i])] = d
+
+    clusters: dict[str, dict] = {
+        nm: {"newick": nm, "height": 0.0, "size": 1} for nm in names
+    }
+
+    counter = 0
+    while len(clusters) > 1:
+        ckeys = list(clusters.keys())
+        min_d = float("inf")
+        best = ("", "")
+        for i in range(len(ckeys)):
+            for j in range(i + 1, len(ckeys)):
+                a, b = ckeys[i], ckeys[j]
+                d = dist.get((a, b), float("inf"))
+                if d < min_d:
+                    min_d = d
+                    best = (a, b)
+
+        a, b = best
+        new_h = min_d / 2.0
+        bl_a  = max(0.0, new_h - clusters[a]["height"])
+        bl_b  = max(0.0, new_h - clusters[b]["height"])
+        new_nw = f"({clusters[a]['newick']}:{bl_a:.6f},{clusters[b]['newick']}:{bl_b:.6f})"
+        new_sz = clusters[a]["size"] + clusters[b]["size"]
+
+        counter += 1
+        new_id = f"__c{counter}"
+
+        for c in ckeys:
+            if c in (a, b):
+                continue
+            da = dist.get((a, c), dist.get((c, a), 0.0))
+            db = dist.get((b, c), dist.get((c, b), 0.0))
+            nd = (da * clusters[a]["size"] + db * clusters[b]["size"]) / new_sz
+            dist[(new_id, c)] = nd
+            dist[(c, new_id)] = nd
+
+        del clusters[a]
+        del clusters[b]
+        clusters[new_id] = {"newick": new_nw, "height": new_h, "size": new_sz}
+
+    root = next(iter(clusters.values()))
+    return root["newick"] + ";"
+
+
+# ─── PhyML local (subprocess) ────────────────────────────────────────────────
+
+def fasta_to_phylip(fasta: str) -> str:
+    """Convert aligned FASTA to relaxed PHYLIP (names up to 100 chars)."""
+    seqs: dict[str, str] = {}
+    cur: str | None = None
+    for line in fasta.strip().splitlines():
+        t = line.strip()
+        if t.startswith(">"):
+            cur = t[1:].split()[0][:100]
+            seqs[cur] = ""
+        elif cur:
+            seqs[cur] += t.upper()
+    if not seqs:
+        return ""
+    n = len(seqs)
+    L = len(next(iter(seqs.values())))
+    lines = [f"{n} {L}"]
+    for name, s in seqs.items():
+        lines.append(f"{name:<100}{s}")
+    return "\n".join(lines)
+
+
+async def _run_phyml_local(job_id: str, aln_fasta: str, req: PhyloRequest) -> None:
+    """Run PhyML as a subprocess on a temp PHYLIP file."""
+    _patch(job_id, phase="tree_running")
+
+    import os
+    import tempfile
+
+    fd, phy_path = tempfile.mkstemp(suffix=".phy")
+    os.close(fd)
+    try:
+        with open(phy_path, "w") as f:
+            f.write(fasta_to_phylip(aln_fasta))
+
+        datatype = "aa" if req.seq_type == "protein" else "nt"
+        model = req.model if req.model else ("LG" if req.seq_type == "protein" else "GTR")
+
+        proc = await asyncio.create_subprocess_exec(
+            "phyml",
+            "-i", phy_path,
+            "-d", datatype,
+            "-m", model,
+            "-b", str(req.bootstrap if req.bootstrap else 0),
+            "-o", "tlr",
+            "--no_memory_check",
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+        try:
+            _, stderr = await asyncio.wait_for(proc.communicate(), timeout=900)
+        except asyncio.TimeoutError:
+            proc.kill()
+            await proc.communicate()
+            _patch(job_id, phase="error", error="PhyML timed out after 15 minutes")
+            return
+
+        if proc.returncode != 0:
+            err_text = stderr.decode("utf-8", errors="replace")[:500] if stderr else ""
+            _patch(job_id, phase="error",
+                   error=f"PhyML failed (exit {proc.returncode}): {err_text}")
+            return
+
+        tree_path = phy_path + "_phyml_tree.txt"
+        stats_path = phy_path + "_phyml_stats.txt"
+
+        newick = None
+        stats = None
+        if os.path.exists(tree_path):
+            with open(tree_path) as f:
+                newick = f.read().strip()
+        if os.path.exists(stats_path):
+            with open(stats_path) as f:
+                stats = f.read().strip()
+
+        if not newick:
+            _patch(job_id, phase="error", error="PhyML produced no output tree")
+            return
+
+        _patch(job_id, phase="complete", newick=newick, stats=stats, done_at=time.time())
+    except Exception as e:
+        _patch(job_id, phase="error", error=f"PhyML error: {e}")
+    finally:
+        for suffix in ["", "_phyml_tree.txt", "_phyml_stats.txt", "_phyml_boot_trees.txt"]:
+            p = phy_path + suffix
+            if os.path.exists(p):
+                os.remove(p)
+
+
+# ─── Main pipeline worker ─────────────────────────────────────────────────────
+
+async def _worker(job_id: str) -> None:
+    job = _read(job_id)
+    if not job:
+        return
+
+    req = PhyloRequest(**job["_req"])
+
+    result = await _run_clustalo(job_id, req.sequences, req.seq_type)
+    if result is None:
+        return
+
+    aln_fasta, nj_newick = result
+
+    if req.method == "nj":
+        _patch(job_id, phase="complete",
+               newick=nj_newick.strip(), done_at=time.time())
+
+    elif req.method == "upgma":
+        _patch(job_id, phase="tree_running")
+        try:
+            newick = _upgma_newick(aln_fasta)
+            _patch(job_id, phase="complete",
+                   newick=newick, done_at=time.time())
+        except Exception as e:
+            _patch(job_id, phase="error", error=f"UPGMA computation failed: {e}")
+
+    elif req.method == "ml":
+        await _run_phyml_local(job_id, aln_fasta, req)
+
+    else:
+        _patch(job_id, phase="error", error=f"Unknown method: {req.method}")
+
+
+# ─── API endpoints ─────────────────────────────────────────────────────────────
+
+@router.post("/run", response_model=RunResponse)
+async def run_phylo(
+    req: PhyloRequest,
+    background_tasks: BackgroundTasks,
+):
+    if len(req.sequences) < 2:
+        raise HTTPException(400, detail="At least 2 sequences are required")
+    if len(req.sequences) > 50:
+        raise HTTPException(400, detail="Maximum 50 sequences per run")
+
+    valid_models = PROTEIN_MODELS if req.seq_type == "protein" else DNA_MODELS
+    if req.method == "ml" and req.model not in valid_models:
+        raise HTTPException(
+            400,
+            detail=f"Model '{req.model}' not valid for {req.seq_type}. "
+                   f"Choose from: {', '.join(valid_models)}"
+        )
+
+    job_id = str(uuid.uuid4())
+    _init(job_id, req)
+    background_tasks.add_task(_worker, job_id)
+    return RunResponse(job_id=job_id, status="queued")
+
+
+@router.get("/status/{job_id}")
+async def get_status(job_id: str):
+    job = _read(job_id)
+    if not job:
+        raise HTTPException(404, detail=f"Job {job_id} not found")
+    return {k: v for k, v in job.items() if not k.startswith("_")}
+
+
+@router.get("/models")
+async def list_models(seq_type: SeqType = "protein"):
+    return {
+        "seq_type": seq_type,
+        "models": PROTEIN_MODELS if seq_type == "protein" else DNA_MODELS,
+        "default": "LG" if seq_type == "protein" else "GTR",
+    }