"""Local web UI for the Digital Evolution Engine.
Single-user Flask server intended to be launched by the desktop ``.app`` and
opened in the user's browser. Stateful by design (long-running jobs are kept
in memory, results held until a new job replaces them) — this is *not* a
multi-tenant service.
API surface:
POST /api/preview parse a sequence (file upload OR JSON body) and
return its metadata + any CDS features.
POST /api/run enqueue a pipeline job. Returns immediately with
a job_id; the work runs in a background thread.
GET /api/status/:id poll job progress.
GET /api/result/:id fetch the variant table once a job has finished.
GET /api/download/:id?format=csv|fasta download outputs.
POST /api/shutdown cleanly stop the server (used by the desktop app
when the user explicitly quits).
"""
from __future__ import annotations
import gzip
import hashlib
import io
import json
import logging
import os
import re
import tempfile
import threading
import time
import uuid
from collections import deque
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Dict, List, Optional
from flask import Flask, Response, g, jsonify, request, send_file, send_from_directory
from dee import auth as _auth
from dee.core import orchestrator as _orch
from dee.core.crispr import find_guides as _find_crispr_guides, guides_to_csv_rows as _crispr_csv_rows
from dee.core.codon import (
DEFAULT_FORBIDDEN_SITES,
pcr_metrics,
reverse_translate,
scrub_restriction_sites,
variants_to_dataframe,
write_library_csv,
)
from dee.core.sequence import (
SequenceValidationError,
find_orfs_in_dna,
list_cds_features,
parse_input,
)
from dee.models.scorer import top_percentile_pool
from dee.optimizer.search import SearchConfig, evolve
logger = logging.getLogger("dee.server")
STATIC_DIR = Path(__file__).resolve().parent / "static"
# Production assets are minified on the fly (cached by mtime) so the served JS/CSS
# is compact + comment-free — source files stay the single source of truth.
# Conservative minifiers (rjsmin/rcssmin); any failure falls back to the raw file.
_MIN_CACHE: Dict[str, tuple] = {}
# Gzipped bodies for immutable assets, keyed by a hash OF THE BYTES rather
# than by path: content-addressed means it can never go stale against an
# edited file, which a path/?v= key does during local development.
_GZIP_CACHE: Dict[bytes, bytes] = {}
def _minified_asset(filename: str) -> Optional[str]:
path = STATIC_DIR / filename
try:
mtime = path.stat().st_mtime
except OSError:
return None
ent = _MIN_CACHE.get(filename)
if ent and ent[0] == mtime:
return ent[1]
try:
src = path.read_text(encoding="utf-8")
if filename.endswith(".js"):
import rjsmin
out = rjsmin.jsmin(src)
else:
import rcssmin
out = rcssmin.cssmin(src)
_MIN_CACHE[filename] = (mtime, out)
return out
except Exception: # noqa: BLE001 — never let minification break asset serving
logger.exception("minify failed for %s; serving raw", filename)
return None
def _clean_html(html: str) -> str:
"""Strip developer HTML comments + minify inline |',
_stash, html, flags=_re.S | _re.I)
tmp = _re.sub(r'', '', tmp, flags=_re.S) # drop HTML comments (keep conditional)
def _restore(m):
blk = blocks[int(m.group(1))]
sm = _re.match(r'()', blk, flags=_re.S | _re.I)
if sm:
if _re.search(r'\bsrc\s*=', sm.group(1), _re.I):
return blk
try:
import rjsmin
return sm.group(1) + rjsmin.jsmin(sm.group(2)) + sm.group(3)
except Exception: # noqa: BLE001
return blk
cm = _re.match(r'()', blk, flags=_re.S | _re.I)
if cm:
try:
import rcssmin
return cm.group(1) + rcssmin.cssmin(cm.group(2)) + cm.group(3)
except Exception: # noqa: BLE001
return blk
return blk
return _re.sub(r'\x00(\d+)\x00', _restore, tmp)
def _no_cache(resp: Response) -> Response:
"""Make an HTML shell revalidate on every load.
Assets are cache-busted by the `?v=` key in index.html, so a hard
max-age on them is safe — but only if the browser actually re-reads the
HTML that carries the new key. With no Cache-Control at all a browser is
free to heuristically reuse the shell, which pins a visitor to the old
key set and hides a deploy for as long as that guess lasts. "no-cache"
still lets the copy be stored; it just requires revalidation first.
"""
resp.headers["Cache-Control"] = "no-cache, must-revalidate"
return resp
def _cleaned_html(filename: str) -> Optional[str]:
"""Cached comment-stripped HTML (mtime-keyed). None on failure → serve raw."""
path = STATIC_DIR / filename
try:
mtime = path.stat().st_mtime
except OSError:
return None
ent = _MIN_CACHE.get("html::" + filename)
if ent and ent[0] == mtime:
return ent[1]
try:
out = _clean_html(path.read_text(encoding="utf-8"))
_MIN_CACHE["html::" + filename] = (mtime, out)
return out
except Exception: # noqa: BLE001
logger.exception("html clean failed for %s; serving raw", filename)
return None
STATE_DIR = Path.home() / ".dee" / "state"
OUTPUT_DIR = Path.home() / ".dee" / "output"
DESKTOP_DIR = Path.home() / "Desktop"
STATE_DIR.mkdir(parents=True, exist_ok=True)
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
def _patch_ssl_for_macos() -> None:
"""Some macOS Python builds ship without a system CA bundle, so
``urllib.request.urlopen`` (which Biopython's NCBIWWW.qblast uses
internally) blows up with ``CERTIFICATE_VERIFY_FAILED: self-signed
certificate in certificate chain`` when it tries to reach
https://blast.ncbi.nlm.nih.gov.
Fix it by installing a default urllib opener whose HTTPS handler uses
certifi's bundled CA roots (certifi is already a transitive dep of
huggingface-hub). Idempotent and safe to call multiple times.
"""
import ssl
import urllib.request
try:
import certifi
except ImportError: # pragma: no cover — certifi is a dep of HF hub
logger.warning("certifi not installed; BLAST may fail on macOS Pythons.")
return
try:
ctx = ssl.create_default_context(cafile=certifi.where())
https_handler = urllib.request.HTTPSHandler(context=ctx)
opener = urllib.request.build_opener(https_handler)
urllib.request.install_opener(opener)
# Also override the module-level default context so callers that go
# around urllib (e.g. http.client direct) still get the certifi roots.
ssl._create_default_https_context = lambda: ssl.create_default_context(
cafile=certifi.where()
)
logger.info("Patched urllib SSL context with certifi CA bundle.")
except Exception as exc: # noqa: BLE001 — best-effort patch
logger.warning("Could not patch SSL CA bundle: %s", exc)
_patch_ssl_for_macos()
# ----------------------------------------------------------------- job state
@dataclass
class JobState:
job_id: str
status: str = "pending"
progress: float = 0.0
message: str = "Queued."
started_at: float = field(default_factory=time.time)
finished_at: Optional[float] = None
wt_identifier: str = ""
wt_protein: str = ""
csv_path: Optional[str] = None
desktop_path: Optional[str] = None
variants: Optional[List[Dict[str, Any]]] = None
library_id: Optional[str] = None # set on save → lets the client log outcomes
# The project (construct) the run was started in. The finished library is
# filed under it, so reopening the project can bring the library back on
# any machine. Only ever an id the client sent; ownership is checked when
# the library is attached (auth.attach_library), not trusted here.
construct_id: Optional[str] = None
error: Optional[str] = None
# Frozen copy of the user's settings + the parameters the engine actually
# used. Surfaces in the UI's metadata pills and the auto-generated
# methods paragraph; also a paper-trail for reproducibility.
settings_used: Dict[str, Any] = field(default_factory=dict)
# Supabase user_id captured at request time, used to attribute the
# library to the user when the pipeline finishes. None for anonymous
# runs; the save-to-Storage path no-ops in that case.
user_id: Optional[str] = None
# Whether/how much the cross-user field-wide aggregate (dee.core.aggregate)
# contributed to this run's scores — was only ever recorded transiently in
# `message` (overwritten by the next status update, so gone by "done").
# Structured here so /api/result can surface it after the job finishes.
global_prior_info: Optional[Dict[str, Any]] = None
def elapsed(self) -> float:
end = self.finished_at if self.finished_at else time.time()
return round(end - self.started_at, 1)
def public(self) -> Dict[str, Any]:
# csv_path / desktop_path intentionally omitted (audit H2): they
# leak the server's filesystem layout. They remain on the dataclass
# for internal use by /api/download and the library-save worker.
return {
"job_id": self.job_id,
"status": self.status,
"progress": round(self.progress, 3),
"message": self.message,
"started_at": self.started_at,
"elapsed_seconds": self.elapsed(),
"wt_identifier": self.wt_identifier,
"wt_protein_length": len(self.wt_protein),
"error": self.error,
}
_JOBS: Dict[str, JobState] = {}
_JOBS_LOCK = threading.Lock()
# Per-job mutex for editing variant rows. The edit handler reads
# job.variants, mutates one row, and rewrites the CSV on disk; without
# this lock two near-simultaneous edits could interleave and write a
# torn library file. _EDIT_LOCKS is itself guarded by _JOBS_LOCK on
# create. Holds one lock per job for the lifetime of the process —
# small enough not to bother evicting.
_EDIT_LOCKS: Dict[str, threading.Lock] = {}
def _edit_lock_for(job_id: str) -> threading.Lock:
with _JOBS_LOCK:
lock = _EDIT_LOCKS.get(job_id)
if lock is None:
lock = threading.Lock()
_EDIT_LOCKS[job_id] = lock
return lock
def _get_job(job_id: str) -> Optional[JobState]:
with _JOBS_LOCK:
return _JOBS.get(job_id)
_JOBS_MAX = 2000 # cap retained jobs so a long-lived process can't grow forever
def _put_job(job: JobState) -> None:
with _JOBS_LOCK:
_JOBS[job.job_id] = job
# Bound memory: previously every job + its edit lock was kept for the
# process lifetime. Dicts preserve insertion order, so evict the oldest
# (long-finished) jobs once we're over the cap.
while len(_JOBS) > _JOBS_MAX:
old_id = next(iter(_JOBS))
_JOBS.pop(old_id, None)
_EDIT_LOCKS.pop(old_id, None)
# ----------------------------------------------------------------- preview cache
# Uploaded files / pasted strings are cached briefly so /api/preview and the
# subsequent /api/run don't have to re-receive the same payload. Keyed by a
# session_id we hand back to the browser.
_SESSIONS: Dict[str, Dict[str, Any]] = {}
_SESSIONS_LOCK = threading.Lock()
# ----------------------------------------------------------------- BLAST jobs
@dataclass
class BlastJob:
"""One asynchronous NCBI-BLAST submission. Keyed by sequence hash so
repeat lookups on the same WT protein return instantly from cache."""
job_id: str
seq_hash: str
status: str = "pending"
started_at: float = field(default_factory=time.time)
finished_at: Optional[float] = None
hits: List[Dict[str, Any]] = field(default_factory=list)
error: Optional[str] = None
def elapsed(self) -> float:
end = self.finished_at if self.finished_at else time.time()
return round(end - self.started_at, 1)
def public(self) -> Dict[str, Any]:
return {
"job_id": self.job_id,
"status": self.status,
"elapsed_seconds": self.elapsed(),
"hits": self.hits,
"error": self.error,
}
_BLAST_CACHE: Dict[str, BlastJob] = {} # seq_hash -> BlastJob
_BLAST_BY_ID: Dict[str, BlastJob] = {} # job_id -> BlastJob
_BLAST_LOCK = threading.Lock()
def _hash_sequence(protein: str) -> str:
return hashlib.sha256(protein.encode("utf-8")).hexdigest()[:16]
def _extract_organism(hit_def: str) -> str:
"""NCBI hit descriptions look like 'GFP [Aequorea victoria]'. Pull out
the bracketed organism if present; fall back to empty string."""
m = re.search(r"\[([^\]]+)\]", hit_def or "")
return m.group(1) if m else ""
def _trim_description(hit_def: str) -> str:
"""Strip the trailing '[organism]' from the description for cleaner
display — the organism is shown separately."""
return re.sub(r"\s*\[[^\]]+\]\s*$", "", hit_def or "").strip()
def _gene_token(raw) -> str:
"""Best-effort gene symbol from a construct's wt_identifier ('>MC1R', 'MSTN',
or a FASTA header). Returns a clean symbol resolve_uniprot will accept, or ''
— anything that isn't a real symbol becomes 'no structure', never a wrong
one. Same shape the CRISPR structure route validates against."""
s = re.sub(r"^[>\s]+", "", str(raw or "")).strip()
s = s.split()[0] if s else ""
return s if re.match(r"^[A-Za-z0-9._-]{1,32}$", s) else ""
# gene+organism -> UniProt accession (or None), cached so a grid of construct
# thumbnails resolves each protein once, not once per card per catalog open.
_CONSTRUCT_ACC_CACHE: Dict[tuple, Optional[str]] = {}
def _extract_uniprot(alignment) -> Optional[str]:
"""Pull a UniProt accession out of an NCBI BLAST alignment if there is one.
NCBI hit_id strings look like:
``sp|P12345|HUMAN_GENE`` (SwissProt — has UniProt accession)
``ref|NP_001234.1|`` (RefSeq — no direct UniProt mapping)
``pdb|1ABC|A`` (PDB)
``gb|AAA12345.1|`` (GenBank)
AlphaFold-DB only indexes UniProt entries, so we only return something
when we see ``sp|`` (SwissProt) or ``tr|`` (TrEMBL) prefixes.
Audit M4: the previous regex ``^[A-Z0-9]+(?:-\\d+)?$`` matched the
modern accession formats only — the canonical ``P12345`` /
``Q9NPB9`` 6-char form, the 10-char form (``A0A1B2C3D4``), and the
isoform suffix (``-2``). It missed two legitimate formats still
occasionally seen in NCBI hits:
* Lowercase-letter accessions returned by some legacy mirrors
(``p12345``) — UniProt's own API is case-insensitive on lookup.
* Underscore-prefixed entry names ("``HUMAN_INSR``" without the
accession token) where the upstream RefSeq pipe is missing.
Now the regex accepts case-insensitive accessions and validates the
structure rather than just the alphabet, and we additionally try the
third pipe-segment as an entry name in case the second segment
didn't match.
"""
# UniProt accession structure (UniProt knowledgebase, kept in sync with
# https://www.uniprot.org/help/accession_numbers): 6-char or 10-char,
# alternating letter / digit groups. The detail is too fiddly to
# encode strictly in a regex without rejecting legitimate hits, so
# we use the documented "anything matching this character class of
# length 6 or 10" form and an optional isoform suffix.
_ACCESSION_RE = re.compile(r"^([A-Za-z0-9]{6}|[A-Za-z0-9]{10})(?:-\d+)?$")
hit_id = getattr(alignment, "hit_id", "") or ""
for token in hit_id.split(";"):
token = token.strip()
parts = token.split("|")
# Format: sp||
if len(parts) >= 2 and parts[0].lower() in ("sp", "tr"):
candidate = parts[1].upper()
m = _ACCESSION_RE.match(candidate)
if m:
# Normalize to the canonical upper-case form.
return m.group(0).upper()
return None
def _swissprot_uniprot(protein: str) -> Optional[Dict[str, Any]]:
"""Resolve the nearest REVIEWED (Swiss-Prot) protein for ``protein``.
Why this exists: the primary search hits ``nr``, which is excellent for
*identifying* a protein but is crowded with redundant GenBank / TPA records
(e.g. thousands of near-identical Cas9 orthologs deposited as ``tpg|HER…``).
Those have no UniProt accession, so a heavily-sequenced protein can come
back from ``nr`` with **zero** UniProt hits in the top results even though a
structure plainly exists — the one curated UniProt entry is simply buried.
Every UniProtKB/Swiss-Prot entry has an AlphaFold-DB model, so a homology
search against the small, curated ``swissprot`` database reliably surfaces
the closest *known* structure (SpCas9 → Q99ZW2). Returns the accession plus
identity / organism for honest labelling, or ``None`` if nothing close.
"""
try:
from Bio.Blast import NCBIWWW, NCBIXML
handle = NCBIWWW.qblast(
program="blastp",
database="swissprot",
sequence=protein,
hitlist_size=1,
expect=1e-3,
)
for record in NCBIXML.parse(handle):
for alignment in record.alignments[:1]:
if not alignment.hsps:
continue
acc = _extract_uniprot(alignment)
if not acc:
continue
hsp = alignment.hsps[0]
return {
"uniprot": acc,
"identity_pct": round(
100.0 * hsp.identities / max(1, hsp.align_length), 1
),
"organism": _extract_organism(alignment.hit_def),
"description": _trim_description(alignment.hit_def),
}
break # only one query record
except Exception: # noqa: BLE001 — best-effort; absence is handled honestly
logger.exception("Swiss-Prot structure-resolution BLAST failed.")
return None
def _attach_structure_homolog(job: BlastJob, protein: str) -> None:
"""If no parsed hit carries a UniProt accession, fall back to a Swiss-Prot
homology search and pin the resolved accession onto the top hit so the
frontend AlphaFold embed lights up. Records the homolog's organism +
identity separately so the UI can label it honestly as the *closest known*
structure rather than the exact query."""
if not job.hits or any(h.get("uniprot") for h in job.hits):
return
sp = _swissprot_uniprot(protein)
if not sp:
return
acc = sp["uniprot"]
top = job.hits[0]
top["uniprot"] = acc
top["alphafold_url"] = f"https://alphafold.ebi.ac.uk/files/AF-{acc}-F1-model_v6.pdb"
top["alphafold_page"] = f"https://alphafold.ebi.ac.uk/entry/{acc}"
# Provenance: this UniProt entry is the nearest reviewed homolog, NOT the
# nr top hit itself — keep its organism / identity distinct for the label.
top["uniprot_via"] = "swissprot_homolog"
top["uniprot_organism"] = sp.get("organism") or ""
top["uniprot_identity_pct"] = sp.get("identity_pct")
top["uniprot_description"] = sp.get("description") or ""
def _run_blast(job: BlastJob, protein: str) -> None:
"""Submit to NCBI BLAST and parse the top hits.
Uses Biopython's NCBIWWW wrapper (the public web service). Polite delays
and rate limits are imposed by NCBI server-side; we keep hitlist small
and expect strict to minimize bandwidth.
"""
try:
# Biopython is heavy; import lazily so the rest of the server boots
# without a hard Bio dependency at startup.
from Bio.Blast import NCBIWWW, NCBIXML
job.status = "submitting"
result_handle = NCBIWWW.qblast(
program="blastp",
database="nr",
sequence=protein,
hitlist_size=8,
expect=1e-5,
)
job.status = "parsing"
for record in NCBIXML.parse(result_handle):
for alignment in record.alignments[:8]:
if not alignment.hsps:
continue
# Use the best (lowest-evalue) HSP from each alignment.
hsp = alignment.hsps[0]
identity_pct = round(
100.0 * hsp.identities / max(1, hsp.align_length), 1
)
coverage_pct = round(
100.0 * hsp.align_length / max(1, len(protein)), 1
)
uniprot = _extract_uniprot(alignment)
job.hits.append(
{
"accession": getattr(alignment, "accession", "") or "",
"hit_id": getattr(alignment, "hit_id", "") or "",
"description": _trim_description(alignment.hit_def),
"organism": _extract_organism(alignment.hit_def),
"length": alignment.length,
"identity_pct": identity_pct,
"coverage_pct": coverage_pct,
"evalue": float(hsp.expect),
"bit_score": float(hsp.bits),
# UniProt accession when available — gates the
# AlphaFold-DB structure embed on the frontend.
"uniprot": uniprot,
# Fallback URL only — the client re-resolves the current
# model version via the AlphaFold prediction API
# (the DB bumps versions over time: v4 → v6 → …).
"alphafold_url": (
f"https://alphafold.ebi.ac.uk/files/AF-{uniprot}-F1-model_v6.pdb"
if uniprot else None
),
"alphafold_page": (
f"https://alphafold.ebi.ac.uk/entry/{uniprot}"
if uniprot else None
),
}
)
break # only one query record
# nr is great for ID but buries the curated UniProt entry under
# redundant GenBank/TPA records — so if nothing came back with a
# UniProt accession, resolve the closest reviewed homolog (which always
# has an AlphaFold model) before finishing.
_attach_structure_homolog(job, protein)
job.status = "done"
except Exception as exc: # noqa: BLE001 — surface anything as an error
logger.exception("BLAST failed.")
job.status = "error"
job.error = f"{type(exc).__name__}: {exc}"
finally:
job.finished_at = time.time()
# --------------------------------------------------- Primer Picker (multi-BLAST)
# Batch primer analysis: the deterministic scoring lives in core/primers.py;
# here we own the async NCBI specificity scan. All primers go to NCBI in ONE
# blastn-short submission (a multi-FASTA query) — literally "blast them all
# together" — and NCBIXML yields one record per primer, in query order.
# Organism targeting for the specificity scan — mirrors Primer-BLAST's
# "Organism" field. Restricts the BLAST via an Entrez query so off-target =
# "would mis-prime in THIS genome", not "matches something, somewhere, in nt".
_PRIMER_ORGANISMS = {
"human": '"Homo sapiens"[Organism]',
"mouse": '"Mus musculus"[Organism]',
"ecoli": '"Escherichia coli"[Organism]',
"yeast": '"Saccharomyces cerevisiae"[Organism]',
"rat": '"Rattus norvegicus"[Organism]',
"zebrafish": '"Danio rerio"[Organism]',
"all": "", # whole nt — broad, noisier
}
@dataclass
class PrimerJob:
job_id: str
seq_hash: str
user_id: Optional[str] = None # owner — every job has one; /api/primers/analyze
# is sign-in gated, unlike the anonymous-eligible
# /api/identify and /api/run jobs
status: str = "pending" # pending|scanning|done|error
started_at: float = field(default_factory=time.time)
finished_at: Optional[float] = None
primers: List[Dict[str, Any]] = field(default_factory=list) # scored, pre-spec
template: str = ""
organism: str = "human"
three_prime_strict: bool = False
warnings: List[str] = field(default_factory=list)
specificity: Optional[Dict[str, Dict[str, Any]]] = None # id -> {...}
products: List[Dict[str, Any]] = field(default_factory=list) # in-silico PCR
error: Optional[str] = None
def elapsed(self) -> float:
end = self.finished_at if self.finished_at else time.time()
return round(end - self.started_at, 1)
def result(self) -> Dict[str, Any]:
from .core import primers as _P
out = _P.rank(self.primers, self.template, self.specificity, self.products)
out["organism"] = self.organism
return out
def public(self) -> Dict[str, Any]:
return {
"job_id": self.job_id,
"status": self.status,
"elapsed_seconds": self.elapsed(),
"warnings": self.warnings,
"specificity_done": self.specificity is not None,
"result": self.result(),
"error": self.error,
}
_PRIMER_CACHE: Dict[str, PrimerJob] = {}
_PRIMER_BY_ID: Dict[str, PrimerJob] = {}
_PRIMER_LOCK = threading.Lock()
_3P_STRICT_LEN = 5 # strict mode: the 3'-terminal window that must match exactly
def _three_prime_clean(hsp, plen: int, k: int = _3P_STRICT_LEN) -> bool:
"""True if the primer's 3'-terminal ``k`` bases match the subject perfectly
(no mismatch, no gap). The 3' end is what licenses extension, so this is the
high-confidence priming test. Reads the tail of the BLAST alignment strings;
falls back to True if they're unavailable (degrade to lenient)."""
if hsp.query_end < plen: # the very 3' base wasn't even aligned
return False
match = getattr(hsp, "match", None)
if not match or len(match) < k:
return True # can't tell → don't over-reject
if " " in match[-k:]: # a mismatch column in the last k bases
return False
q = getattr(hsp, "query", "") or ""
s = getattr(hsp, "sbjct", "") or ""
if "-" in q[-k:] or "-" in s[-k:]: # a gap in the last k bases
return False
return True
def _hsp_to_event(alignment, hsp, primer: Dict[str, Any],
strict_3p: bool = False) -> Optional[Dict[str, Any]]:
"""Turn one BLAST HSP into a 'priming event' for in-silico PCR — but only if
it's 3'-anchored enough to actually extend. Returns None otherwise.
Direction: the primer's 3' end maps to ``hsp.sbjct_end`` in both strand
cases; the polymerase runs toward higher coords when sbjct_end > sbjct_start
('+') and toward lower coords otherwise ('-')."""
plen = len(primer["seq"])
ident = hsp.identities / max(1, hsp.align_length)
# Must cover the primer's 3' end and be a near-full, high-identity match.
if ident < 0.90 or hsp.align_length < plen - 3 or hsp.query_end < plen - 1:
return None
# Strict mode: also require a perfect 3'-terminal match (cuts borderline
# sites that share a body but mismatch at the 3' end and wouldn't extend).
if strict_3p and not _three_prime_clean(hsp, plen):
return None
ss, se = hsp.sbjct_start, hsp.sbjct_end
direction = "+" if se >= ss else "-"
return {
"primer_id": primer["id"],
"name": primer.get("name", primer["id"]),
"pos5": ss, # 5' end of the primer on the subject
"pos3": se, # 3' end (where extension starts)
"dir": direction,
"title": _trim_description(getattr(alignment, "hit_def", ""))[:80],
"organism": _extract_organism(getattr(alignment, "hit_def", "")),
"identity_pct": round(100.0 * ident, 1),
}
def _run_primer_blast(job: PrimerJob) -> None:
"""One blastn-short submission for ALL primers, then in-silico PCR.
Pairs every forward-oriented priming site with every downstream reverse
site on the same subject (within MAX_PRODUCT bp) to predict the actual PCR
products — the same specificity check Primer-BLAST runs, done here so the
user never leaves the site."""
from .core import primers as _P
try:
from Bio.Blast import NCBIWWW, NCBIXML
job.status = "scanning"
fasta = "\n".join(f">{p['id']}\n{p['seq']}" for p in job.primers)
qblast_kw = dict(
program="blastn",
database="nt",
sequence=fasta,
megablast=False, # short, sensitive search
word_size=7,
expect=1000,
nucl_reward=1,
nucl_penalty=-3,
hitlist_size=50,
)
entrez = _PRIMER_ORGANISMS.get(job.organism, "")
if entrez:
qblast_kw["entrez_query"] = entrez
handle = NCBIWWW.qblast(**qblast_kw)
spec: Dict[str, Dict[str, Any]] = {}
events_by_subject: Dict[str, List[Dict[str, Any]]] = {}
records = list(NCBIXML.parse(handle))
# Records come back in query order; align them to job.primers.
for i, record in enumerate(records):
if i >= len(job.primers):
break
p = job.primers[i]
strong, top = 0, []
for alignment in record.alignments:
acc = getattr(alignment, "accession", "") or getattr(alignment, "hit_id", "")
hit_strong = False
for hsp in alignment.hsps:
ev = _hsp_to_event(alignment, hsp, p, job.three_prime_strict)
if ev:
events_by_subject.setdefault(acc, []).append(ev)
hit_strong = True
if hit_strong:
strong += 1
if len(top) < 5:
top.append({
"title": _trim_description(getattr(alignment, "hit_def", ""))[:80],
"organism": _extract_organism(getattr(alignment, "hit_def", "")),
})
spec[p["id"]] = {"strong_offtargets": strong, "top_hits": top}
# In-silico PCR across ALL primers. rank() collapses these redundant
# hits into DISTINCT amplicons and splits intended vs off-target, so we
# just hand it the raw product list here.
job.products = _P.insilico_pcr(events_by_subject)
for p in job.primers:
spec.setdefault(p["id"], {"strong_offtargets": 0, "top_hits": []})
job.specificity = spec
job.status = "done"
except Exception as exc: # noqa: BLE001
logger.exception("Primer BLAST failed.")
job.status = "error"
job.error = f"{type(exc).__name__}: {exc}"
finally:
job.finished_at = time.time()
_SESSION_TTL_SECONDS = 6 * 60 * 60 # 6 h — long enough to cover lunch, calls, etc.
def _new_session(payload: Dict[str, Any]) -> str:
"""Cache an uploaded sequence under a fresh session id and prune stale entries.
Audit M3: previously the TTL was 30 minutes, which kicked the user out
if they paused to refine settings, take a call, or step away for lunch.
Bumping to 6 hours covers a typical workday's "designed a library
before lunch, came back to download it" pattern without unbounded
memory growth (sessions only hold the uploaded path + ORF map).
"""
sid = uuid.uuid4().hex[:12]
payload["created_at"] = time.time()
with _SESSIONS_LOCK:
_SESSIONS[sid] = payload
cutoff = time.time() - _SESSION_TTL_SECONDS
stale = [k for k, v in _SESSIONS.items() if v.get("created_at", 0) < cutoff]
for k in stale:
_SESSIONS.pop(k, None)
return sid
def _get_session(sid: str) -> Optional[Dict[str, Any]]:
with _SESSIONS_LOCK:
return _SESSIONS.get(sid)
# ----------------------------------------------------------------- Flask app
# ─── Rate limiting (in-process, per-IP sliding window) ────────────────
# A single HF Space container serves everyone, and the heavy endpoints
# (run, identify/BLAST, primer design) are CPU-bound and reachable on the
# anonymous trial — so an unthrottled client can starve every other user or
# run up compute cost. This is a lightweight in-memory limiter: no Redis,
# bounded memory, and fail-open on its own errors (a limiter hiccup must never
# block a legitimate request). Tune via env without a code change.
_RL_LOCK = threading.Lock()
_RL_HITS: "Dict[str, deque]" = {} # client-key -> deque[monotonic timestamps]
_RL_MAX_KEYS = int(os.environ.get("TURINGDNA_RL_MAX_KEYS", "20000")) # evict if exceeded
_RL_DISABLED = os.environ.get("TURINGDNA_RL_DISABLED", "") == "1" # escape hatch
# (max_requests, window_seconds) per path prefix. Longest matching prefix wins;
# the "/api" entry is the catch-all default. Heavy endpoints are tighter.
_RL_RULES = [
("/api/run", (20, 60)), # full pipeline — expensive
("/api/de/epistasis", (8, 60)), # up to 12 ESM forward passes — heavy
("/api/de/round2", (8, 60)), # one ESM pass + an evolutionary search
# (attacker-tunable up to steps=5000,
# restarts=24) + up to 8 more ESM passes
# for epistasis re-ranking — same class as
# /api/de/epistasis above, and it was
# missing here, so it fell through to the
# 120/60 default. That matters more than a
# normal rate-limit gap: score_guarded()
# blocks on ONE process-wide semaphore with
# no timeout on this path, and the deploy is
# 1 gunicorn worker / 8 threads total — a
# burst here can exhaust every thread and
# stall the whole app, not just this route.
# Every reopen pulls the library's CSV out of Storage server-side, so this
# is an outbound fetch per call rather than a table read. Covers the list
# and download routes too, which is fine: opening the drawer is one call.
("/api/library", (30, 60)),
("/api/identify", (20, 60)), # BLAST — expensive
("/api/preview", (40, 60)),
("/api/primers/local-specificity", (20, 60)), # genome scan — expensive
("/api/primers/design", (30, 60)),
("/api/primers/multiplex", (30, 60)),
("/api/primers/analyze", (40, 60)),
("/api/crispr", (40, 60)),
# Evo 2 (7B) on a rented GPU — the most expensive call in the product by a
# wide margin, and unlike ESM-2's Achilles tier there is no free local path
# to fall back to. Tighter than every other tool bucket on purpose: this
# limit is about the GPU bill, not about protecting a worker thread.
("/api/dna/generate", (6, 60)), # autoregressive — priciest
("/api/compiler", (30, 60)), # pure logic; probes DNA reach
("/api/dna", (12, 60)),
("/api/assays", (60, 60)),
("/api/de/uplift", (20, 60)), # a few DB reads, no model
("/api/de/import", (10, 60)), # one ESM pass over the WT, same
# score_guarded semaphore as round 2
("/api/certificate", (60, 60)), # pure logic, but
# unauthenticated
("/api/workspaces", (60, 60)),
("/api/workspace-results", (30, 60)),
("/api/constructs", (90, 60)), # cheap CRUD; the
# switcher polls it
("/api/plasmid", (60, 60)),
("/api/ping", (60, 60)), # dwell heartbeat — its own
# bucket so ~2/min never eats
# the shared default budget
("/api/collect", (120, 60)), # UX-behaviour collector — its
# OWN generous bucket (batched
# sendBeacon flushes, up to
# ~1/5s + on hide) so it never
# eats the shared /api budget
("/api/orchestrator/poll", (240, 60)), # event poll — spends NOTHING (reads an
# in-memory event log). Fires ~1/s while
# the agent works, so it needs its own
# generous bucket ABOVE the billed rule
# below, or a normal run throttles itself.
("/api/orchestrator/runs", (60, 60)), # history list — one cheap DB read, no
# LLM. Must sit ABOVE the billed rule
# below (longest prefix wins) or opening
# the run menu would burn the agent's
# own 12/min budget.
("/api/orchestrator", (12, 60)), # start / reply / stop — each spawns a
# real billed OpenRouter loop. Tight, like
# a billed LLM loop; per-run cost cap in
# orchestrator.py is the second guard.
("/api/structure", (30, 60)), # public gene→AlphaFold resolve. One
# outbound UniProt lookup, no LLM, no
# user data; anonymous by design (see
# the route). 30/min blocks scraping.
("/api/project", (30, 60)), # project CRUD (client-side v1) — cheap
("/api", (120, 60)), # default for any other API call
]
# Engagement events: path → kind, logged once per SUCCESSFUL request from the
# after_request hook (one place instead of touching every handler). Only these
# "meaningful action" endpoints count; /api/run is excluded (public.runs is
# authoritative for it, so logging it here too would double-count engagement).
_EVENT_KINDS = {
"/api/plasmid/import": "plasmid_map",
"/api/plasmid/clone": "plasmid_clone",
"/api/plasmid/save": "plasmid_save",
"/api/crispr/design": "crispr_design",
"/api/crispr/save": "crispr_save",
"/api/primers/analyze": "primer_analyze",
"/api/primers/save": "primer_save",
"/api/align": "sequence_align",
"/api/de/round2": "de_round2",
"/api/de/import": "de_import", # the "I already have data" front door
"/api/dna/score": "dna_score",
"/api/dna/generate": "dna_generate",
"/api/compiler/compile": "therapeutic_compile",
}
# Sort longest-prefix-first so the most specific rule matches.
_RL_RULES.sort(key=lambda r: len(r[0]), reverse=True)
def _is_private_ip(ip: str) -> bool:
"""True for loopback / private / link-local addresses (internal proxy hops)."""
try:
import ipaddress
a = ipaddress.ip_address(ip)
return a.is_private or a.is_loopback or a.is_link_local or a.is_reserved
except ValueError:
return True # not an IP → don't trust it as a client key
def _client_key() -> str:
"""Client identifier for rate limiting.
We sit behind a trusted proxy (HF ingress) that appends the real peer IP to
X-Forwarded-For. The LEFTMOST hop is client-supplied and therefore spoofable
— keying on it lets an attacker mint a fresh bucket per request and bypass
the limiter. So we take the RIGHTMOST *public* hop instead (the one the
proxy added), skipping private/loopback entries so we don't collapse every
user onto a shared internal proxy IP. Falls back to remote_addr."""
fwd = request.headers.get("X-Forwarded-For", "")
if fwd:
parts = [p.strip() for p in fwd.split(",") if p.strip()]
for ip in reversed(parts):
if not _is_private_ip(ip):
return ip
if parts:
return parts[-1] # all private — rightmost is closest to us
return request.remote_addr or "unknown"
def _get_nonce() -> str:
"""Per-request CSP nonce, memoized on flask.g. Used to allow our own inline