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