Hugging Face's logo

Spaces:
WINTER4000
/
syntheogenesis
Running

App Files Community
syntheogenesis / dee /server.py
github-actions[bot]
Deploy d4ed15c
1a5743f
Raw History Blame Contribute Delete
366 kB
"""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 <script>/<style> so the
SERVED page carries no internal engineering notes (source stays commented).
Protects script/style regions before removing comments; keeps conditional
comments; minifies inline JS/CSS with rjsmin/rcssmin."""
import re as _re
blocks: list = []
def _stash(m):
blocks.append(m.group(0))
return f"\x00{len(blocks) - 1}\x00"
tmp = _re.sub(r'<script\b[^>]*>.*?</script>|<style\b[^>]*>.*?</style>',
_stash, html, flags=_re.S | _re.I)
tmp = _re.sub(r'<!--(?!\[if).*?-->', '', tmp, flags=_re.S) # drop HTML comments (keep conditional)
def _restore(m):
blk = blocks[int(m.group(1))]
sm = _re.match(r'(<script\b[^>]*>)(.*)(</script>)', 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'(<style\b[^>]*>)(.*)(</style>)', 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|<accession>|<entry_name>
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
<script> blocks under a strict script-src while blocking any injected inline
script / event-handler (the app uses external JS + addEventListener only)."""
import secrets
from flask import g
n = getattr(g, "csp_nonce", None)
if not n:
n = secrets.token_urlsafe(16)
g.csp_nonce = n
return n
_CORS_ORIGIN_RE = re.compile(r"^https://([a-z0-9-]+\.)?turingdna\.com$")
def _public_cors(resp: Response) -> Response:
"""Opt a response into cross-origin reads from the marketing site, and
ONLY the marketing site — turingdna.com wants a live Field Atlas widget
on the landing page, which means fetching from a different origin
(winter4000-syntheogenesis.hf.space) than the page itself. Every other
endpoint stays same-origin-only (the default), including everything
behind auth — this is applied by hand to exactly the two routes that are
already public, unauthenticated, de-identified reads (/api/atlas,
/api/benchmarks), never as a blanket policy. Origin is echoed rather than
'*' so the allowlist actually restricts who can read the response, and
Vary: Origin keeps any intermediate cache from serving one origin's
CORS grant to another."""
origin = request.headers.get("Origin", "")
if origin and _CORS_ORIGIN_RE.match(origin):
resp.headers["Access-Control-Allow-Origin"] = origin
resp.headers["Vary"] = "Origin"
return resp
def _authed_cors(resp: Response) -> Response:
"""Like _public_cors, but for the account endpoints turingdna.com calls
WITH a Bearer token — the personal dashboard's billing panel and its
delete-account button. An authenticated cross-origin request is never a
"simple" one (the Authorization header alone disqualifies it, GET included),
so the browser gates every one of them behind a preflight; this adds the
method and header allowances that preflight checks.
GET is listed as well as POST because reading the billing panel is a GET
and would otherwise fail at the preflight — with a CORS console error
rather than an HTTP status, which looks like the endpoint not existing.
Still origin-locked to turingdna.com (same regex), and deliberately NOT in
credentials mode: authorisation rides the Authorization header, never a
cookie, so there is no ambient session for a hostile origin to borrow — the
grant only lets turingdna.com's own script reuse the token it already holds.
"""
origin = request.headers.get("Origin", "")
if origin and _CORS_ORIGIN_RE.match(origin):
resp.headers["Access-Control-Allow-Origin"] = origin
resp.headers["Access-Control-Allow-Methods"] = "GET, POST, OPTIONS"
resp.headers["Access-Control-Allow-Headers"] = "authorization, content-type"
resp.headers["Access-Control-Max-Age"] = "600"
resp.headers.setdefault("Vary", "Origin")
return resp
def _rate_limit_rule(path: str):
"""Most-specific matching rule as (prefix, max_requests, window_s), or
None if the path isn't rate-limited."""
for prefix, (limit, window) in _RL_RULES:
if path.startswith(prefix):
return (prefix, limit, window)
return None
def _check_rate_limit(path: str, key: str):
"""Returns (limited: bool, retry_after_s: int) over a sliding window.
The window is per (client, rule) so heavy and cheap endpoints have
independent budgets — bursting the plasmid editor never locks a user out
of running a pipeline. Bounded memory; fail-open on any internal error."""
if _RL_DISABLED:
return (False, 0)
matched = _rate_limit_rule(path)
if matched is None:
return (False, 0)
prefix, limit, window = matched
bucket = f"{key}|{prefix}"
now = time.monotonic()
try:
with _RL_LOCK:
# Protect against an IP-rotating flood bloating the table — but evict
# the STALEST buckets (oldest last-hit), not the whole table. A blunt
# clear() would reset every honest user's window, and a unique-IP
# flood could exploit that to wipe limits on demand.
if len(_RL_HITS) > _RL_MAX_KEYS:
try:
victims = sorted(_RL_HITS.items(),
key=lambda kv: (kv[1][-1] if kv[1] else 0.0))
for vk, _dq in victims[: max(1, len(_RL_HITS) // 5)]:
_RL_HITS.pop(vk, None)
except Exception: # noqa: BLE001
_RL_HITS.clear() # last-ditch fallback
dq = _RL_HITS.get(bucket)
if dq is None:
dq = deque()
_RL_HITS[bucket] = dq
cutoff = now - window
while dq and dq[0] < cutoff:
dq.popleft()
if len(dq) >= limit:
retry = int(dq[0] + window - now) + 1
return (True, max(1, retry))
dq.append(now)
return (False, 0)
except Exception: # noqa: BLE001 — never block a request on limiter failure
return (False, 0)
# How long a library sits with no logged results before Mission Control stops
# saying "ready to build" and starts asking for the numbers. Three weeks is
# past a synthesis-and-assay turnaround for most benches, so by then "you have
# not started" is a less likely explanation than "you have results and nowhere
# obvious to put them". A Space variable, because the right number is an
# empirical question nobody has data for yet.
MISSION_AWAITING_RESULTS_DAYS = int(
os.environ.get("TURINGDNA_MISSION_AWAITING_RESULTS_DAYS", "21"))
def _days_since(iso: Optional[str]) -> int:
"""Whole days since an ISO timestamp. 0 when it cannot be read — an
unparseable date must never promote a card to 'waiting on you'."""
if not iso:
return 0
import datetime as _dt
try:
txt = str(iso).replace("Z", "+00:00")
then = _dt.datetime.fromisoformat(txt)
if then.tzinfo is None:
then = then.replace(tzinfo=_dt.timezone.utc)
return max(0, (_dt.datetime.now(_dt.timezone.utc) - then).days)
except Exception: # noqa: BLE001
return 0
def create_app() -> Flask:
app = Flask(__name__, static_folder=None)
# Global request-body cap (defense-in-depth against memory-exhaustion POSTs).
# 25 MB comfortably covers the largest legitimate payload (CRISPR's 1 Mbp
# paste ≈ 1 MB, sequence-file uploads) while blocking 100 MB+ floods;
# oversized requests get a 413 before any handler/parser runs.
app.config["MAX_CONTENT_LENGTH"] = 25 * 1024 * 1024
# Wire up auth middleware: before_request hook that parses the
# Authorization header + signed anon cookie into flask.g.auth.
# See dee/auth.py for the full design + env-var requirements.
_auth.init_app(app)
@app.get("/healthz")
def _healthz() -> Response:
"""Liveness probe for uptime monitors / the platform supervisor. No auth,
no rate limit (not under /api/), fast + side-effect-free."""
return jsonify({"status": "ok"}), 200
@app.post("/api/ping")
def api_ping() -> Response:
"""Dwell-time heartbeat. The client posts {session_id, visible_ms} every
~30 s while the tab is VISIBLE, and once more on hide via sendBeacon.
Upserts one public.app_sessions row per session (atomic record_ping RPC).
Anonymous visitors are tracked too. Auth optional, best-effort — returns
204 fast and never does work on the request thread."""
try:
data = request.get_json(silent=True) or {}
sid = str(data.get("session_id") or "")[:64]
try:
vis = int(data.get("visible_ms", 0))
except (TypeError, ValueError):
vis = 0
if sid:
_auth.record_ping_async(session_id=sid, visible_ms=vis)
except Exception: # noqa: BLE001 — telemetry never affects the response
pass
# 204: no body; sendBeacon ignores the response anyway.
return ("", 204)
@app.post("/api/collect")
def api_collect() -> Response:
"""First-party UX-behaviour ingest (sibling of /api/ping). The
collector.js batch POSTs {session_id, events:[…]} via sendBeacon —
dead/rage clicks, scroll depth, funnel steps, feature clicks, client
errors. STRUCTURE ONLY: the client sends element ids / data-analytics
labels / tag paths, never sequence content, and never identity — the
server stamps user_id / anon_fingerprint from g.auth in
log_client_events_async, which also whitelists `kind`, caps string
lengths, and drops unexpected keys (never trusts client shape).
Auth optional, best-effort — returns 204 fast and never does work on
the request thread (the writer fires off a daemon thread)."""
try:
data = request.get_json(silent=True) or {}
sid = str(data.get("session_id") or "")[:64]
events = data.get("events")
# Cap events per request so a crafted payload can't fan out into a
# huge batch insert; the writer re-caps + sanitizes each row.
if sid and isinstance(events, list) and events:
_auth.log_client_events_async(sid, events[:100])
except Exception: # noqa: BLE001 — telemetry never affects the response
pass
return ("", 204)
@app.before_request
def _enforce_rate_limit() -> Optional[Response]:
# Throttle only API calls — the SPA shell + static assets are cheap and
# heavily cached. Returning a response here short-circuits the handler.
path = request.path or ""
if not path.startswith("/api/"):
return None
# Start the latency clock BEFORE the rate-limit check, so a 429 is timed
# like any other response. Short-circuiting here still runs
# after_request, which would otherwise find no start time and record 0 ms
# for exactly the requests worth investigating.
g._req_started_at = time.perf_counter()
limited, retry = _check_rate_limit(path, _client_key())
if not limited:
return None
resp = jsonify({
"error": "Too many requests — please slow down and try again in a moment.",
"kind": "rate_limited",
})
resp.status_code = 429
resp.headers["Retry-After"] = str(retry)
return resp
# ── Compression ──────────────────────────────────────────────────────
# Registered BEFORE _security_headers so it runs AFTER it: Flask calls
# after_request hooks in reverse registration order, and nothing may touch
# the body once it has been compressed.
#
# There is no nginx in front of this. gunicorn serves these bytes straight
# out, so if we do not compress, nothing does. Measured on the live Space
# before this: the document, app.js and app.css left the server at 858 KB
# against 204 KB gzipped, and first contentful paint was 4.0s. 654 KB of
# every cold load was pure waste.
_COMPRESSIBLE = frozenset((
"text/html", "text/css", "text/plain", "text/csv", "text/markdown",
"application/javascript", "application/json", "application/xml",
"image/svg+xml",
))
# Below this, gzip's own framing costs more than it saves.
_GZIP_MIN_BYTES = 1024
@app.after_request
def _compress(resp: Response) -> Response:
try:
if (resp.direct_passthrough # send_file streams — leave it alone
or resp.status_code in (204, 304)
or "Content-Encoding" in resp.headers
or "Content-Range" in resp.headers # a range is not the whole body
or (resp.mimetype or "") not in _COMPRESSIBLE):
return resp
# Announce the variance whether or not we compress THIS response.
# Without it a shared cache can hand gzipped bytes to a client that
# never asked for them. .vary is a HeaderSet, so this appends to the
# CORS Vary already present rather than replacing it.
resp.vary.add("Accept-Encoding")
if "gzip" not in request.headers.get("Accept-Encoding", ""):
return resp
data = resp.get_data()
if len(data) < _GZIP_MIN_BYTES:
return resp
# Immutable assets (?v=-busted, max-age=31536000) are worth level 9
# once and caching; the index cannot be cached at all because it
# carries a per-request CSP nonce, so it takes the cheap level.
if "max-age=31536000" in resp.headers.get("Cache-Control", ""):
key = hashlib.blake2b(data, digest_size=16).digest()
out = _GZIP_CACHE.get(key)
if out is None:
out = gzip.compress(data, 9)
if len(_GZIP_CACHE) > 64: # bounded — the asset set is small
_GZIP_CACHE.clear()
_GZIP_CACHE[key] = out
else:
out = gzip.compress(data, 6)
if len(out) >= len(data): # already-dense payload; don't pay for nothing
return resp
resp.set_data(out)
resp.headers["Content-Encoding"] = "gzip"
resp.headers["Content-Length"] = str(len(out))
return resp
except Exception: # noqa: BLE001 — compression must never break a response
logger.exception("compression failed; serving uncompressed")
return resp
@app.after_request
def _security_headers(resp: Response) -> Response:
# Safe, iframe-compatible hardening. setdefault so we never clobber a
# header a handler set deliberately (e.g. Cache-Control on assets).
resp.headers.setdefault("X-Content-Type-Options", "nosniff")
resp.headers.setdefault("Referrer-Policy", "strict-origin-when-cross-origin")
resp.headers.setdefault(
"Permissions-Policy",
"geolocation=(), microphone=(), camera=(), payment=(), usb=()",
)
# HSTS — the engine is always HTTPS on hf.space. No includeSubDomains /
# preload: we don't want to make claims about sibling *.hf.space hosts.
resp.headers.setdefault("Strict-Transport-Security", "max-age=31536000")
# Clickjacking defense that RESPECTS the embed: only turingdna.com and
# huggingface.co may frame the engine (the /app wrapper iframes it, and
# so does the Space's own "App" preview tab on huggingface.co — without
# this the preview shows as a blank/broken frame even though the engine
# is running fine). An allowlist, not a blanket deny — direct
# (un-framed) access to the engine is unaffected.
# Strict-ish CSP. script-src is nonce-based (no 'unsafe-inline'), so an
# injected inline <script> or on*= handler — the realistic XSS vector,
# and the one that would steal the Supabase token from localStorage — is
# refused; our own inline scripts carry the nonce. 'unsafe-eval' + blob:
# are allowed because the bundled 3-D / spreadsheet libraries need them.
# We deliberately DON'T set default-src/connect-src/style-src/img-src, so
# API calls, external structure fetches, fonts and inline styles are
# unaffected — script execution is the only thing locked down here.
resp.headers.setdefault(
"Content-Security-Policy",
"script-src 'self' 'nonce-" + _get_nonce() + "' 'unsafe-eval' blob: "
"https://cdn.jsdelivr.net https://unpkg.com https://cdn.paddle.com https://us-assets.i.posthog.com; "
# No connect-src/frame-src are set (see the note below), so Paddle.js'
# API calls and its checkout overlay iframe (buy.paddle.com /
# sandbox-buy.paddle.com) are already allowed; only its CDN script
# needs naming here.
"object-src 'none'; base-uri 'self'; "
"frame-ancestors 'self' https://turingdna.com https://*.turingdna.com "
"https://huggingface.co",
)
# Engagement telemetry: one event per successful "meaningful action".
# Centralized here so no individual handler needs to know about analytics.
try:
if resp.status_code < 400:
_kind = _EVENT_KINDS.get(request.path)
if _kind:
_auth.log_event(_kind)
except Exception: # noqa: BLE001 — never break a response over analytics
pass
# Request-level telemetry for the analytics console's endpoint panel.
# Separate from the block above on purpose: _EVENT_KINDS means "a user
# did something meaningful", this means "an HTTP request happened". They
# answer different questions and merging them would ruin both.
#
# /api/collect and /api/ping are excluded — logging the telemetry
# endpoints as telemetry is a feedback loop that would make them the two
# busiest routes on the dashboard while telling us nothing.
try:
_p = request.path
if (_p.startswith("/api/")
and _p not in ("/api/collect", "/api/ping")):
_started = g.get("_req_started_at")
_elapsed = (time.perf_counter() - _started) * 1000.0 if _started else 0.0
_auth.log_api_request_async(_p, resp.status_code, _elapsed)
except Exception: # noqa: BLE001 — never break a response over analytics
pass
return resp
@app.get("/")
def index() -> Response:
body = _cleaned_html("index.html") # strip internal comments; source stays commented
if body is not None:
# Stamp the per-request nonce onto our two first-party inline scripts
# so they run under the strict script-src CSP set in _security_headers.
# Only bare "<script>" matches — external "<script src=…>" tags are
# untouched (allowed by origin) — so this hits exactly the inline pair.
body = body.replace("<script>", f'<script nonce="{_get_nonce()}">')
return _no_cache(Response(body, mimetype="text/html"))
return _no_cache(send_from_directory(STATIC_DIR, "index.html"))
@app.get("/static/<path:filename>")
def static_file(filename: str) -> Response:
# Serve minified JS/CSS (cached) for a clean, compact production surface.
if "/" not in filename and filename.endswith((".js", ".css")):
body = _minified_asset(filename)
if body is not None:
mime = "application/javascript" if filename.endswith(".js") else "text/css"
resp = Response(body, mimetype=mime)
# safe to cache hard: every reference is query-string cache-busted (?v=)
resp.headers["Cache-Control"] = "public, max-age=31536000"
return resp
# Strip internal comments from static HTML pages (e.g. assistant.html) so
# view-source / inspect-element shows no engineering notes — same clean
# treatment as the "/" index route. These pages reference external scripts
# only (no inline <script>), so no CSP-nonce stamping is required.
if "/" not in filename and filename.endswith(".html"):
body = _cleaned_html(filename)
if body is not None:
return _no_cache(Response(body, mimetype="text/html"))
return send_from_directory(STATIC_DIR, filename)
@app.post("/api/preview")
def preview() -> Response:
"""Accept a sequence and report what we found (no scoring yet).
Accepts EITHER:
* multipart/form-data with ``file`` field, or
* application/json with ``{"text": "..."}``.
If the input parses cleanly as a single CDS or protein, returns its
metadata. If it's a structured plasmid file or a raw DNA blob with
multiple ORFs, returns a list of choices in ``cds_options`` so the
UI can show a picker before kicking off a run.
"""
try:
tmp_path, original_name = _materialize_input()
payload, session_extras = _summarize(tmp_path, original_name)
sid = _new_session(
{"path": str(tmp_path), "name": original_name, **session_extras}
)
payload["session_id"] = sid
return jsonify(payload)
except SequenceValidationError as exc:
return jsonify(
{
"error": str(exc),
"kind": "validation",
"nt_position": exc.nt_position,
"code": exc.code,
}
), 400
except Exception: # noqa: BLE001 — surface anything else as 500.
logger.exception("Preview failed.")
# Detail is logged; don't leak the raw exception text to the client
# (confusing for users, and avoids exposing internals).
return jsonify({"error": "Couldn't read that input — check it's a valid "
"sequence or file and try again.",
"kind": "internal"}), 500
@app.post("/api/run")
def start_run() -> Response:
# Auth-or-quota gate. Anonymous visitors get ANON_FREE_RUNS of
# them — runs, not minutes; see dee/auth.py for why the clock was
# the wrong unit. After that this returns a 402 with
# kind="auth_required" and the frontend pops the sign-up gate.
# Signed-in users (valid Supabase JWT in Authorization header)
# bypass entirely.
gate = _auth.require_auth_or_quota()
if gate is not None:
return gate
body = request.get_json(force=True, silent=True) or {}
sid = body.get("session_id")
if not sid:
return jsonify({"error": "missing session_id"}), 400
session = _get_session(sid)
if not session:
return jsonify({"error": "session expired; re-upload the sequence"}), 410
try:
settings = _validate_run_settings(body.get("settings") or {})
except ValueError as exc:
return jsonify({"error": str(exc), "kind": "validation"}), 400
cds_feature = body.get("cds_feature")
# Audit M7: require explicit CDS selection for multi-CDS uploads.
# Previously, when the user uploaded an annotated file with >1 CDS
# feature, the pipeline defaulted to the LONGEST CDS if the user
# hadn't picked one — sometimes silently swapping in a 5'-flanking
# ORF instead of the intended target gene. Now: if the session
# offers a chooser (cds_options) and the user didn't pick one,
# 400 instead of guessing.
cds_options = session.get("cds_options") or []
if cds_options and len(cds_options) > 1 and not cds_feature:
return jsonify({
"error": "This input has multiple CDS features. Pick which one "
"to design against — auto-selection is disabled to "
"avoid silently designing the wrong gene.",
"kind": "needs_cds_choice",
"cds_options": cds_options,
}), 400
# Capture the signed-in user_id at request time (while flask.g.auth
# is still populated) so the background pipeline thread can
# attribute the library to them on completion. None for anon users.
_auth_ctx = _auth.get_auth()
job = JobState(
job_id=uuid.uuid4().hex[:12],
user_id=_auth_ctx.user_id if not _auth_ctx.anonymous else None,
)
_cid = str(body.get("construct_id") or "").strip()
if job.user_id and re.match(r"^[0-9a-fA-F-]{36}$", _cid):
job.construct_id = _cid
_put_job(job)
# If the session has an ORF map and the user picked an ORF, materialize
# that ORF's DNA into a fresh temp file and use it as the pipeline input.
run_path = Path(session["path"])
run_name = session["name"]
orf_map = session.get("orfs") or {}
if orf_map and cds_feature and cds_feature in orf_map:
tmp = tempfile.NamedTemporaryFile(
delete=False, suffix=".fasta", mode="w", encoding="utf-8",
dir=str(STATE_DIR),
)
tmp.write(f">{cds_feature}\n{orf_map[cds_feature]}\n")
tmp.close()
run_path = Path(tmp.name)
run_name = f"{Path(run_name).stem}_{cds_feature}"
# Already extracted a clean ORF — don't re-pass the label downstream.
cds_feature = None
thread = threading.Thread(
target=_run_pipeline,
args=(job, run_path, run_name, cds_feature, settings),
daemon=True,
)
thread.start()
# Best-effort: log the run to Supabase public.runs for analytics.
# Never blocks the response. Stores only hash + length of the
# sequence (per the privacy policy — no sequence content leaves
# the engine memory for analytics purposes).
try:
with open(run_path, "rb") as _f:
_sequence_hash = hashlib.sha256(_f.read()).hexdigest()[:32]
_sequence_length = run_path.stat().st_size
except OSError:
_sequence_hash = "unknown"
_sequence_length = 0
_auth.log_run_async(
job_id=job.job_id,
sequence_hash=_sequence_hash,
sequence_length=_sequence_length,
model=settings.get("model"),
host_organism=settings.get("host"),
)
# Lazy cleanup of expired libraries (free-tier TTL, 90d by default).
# Piggybacks on the user's own /api/run so we don't need pg_cron
# or an Edge Function — active users keep their storage tidy.
# Anonymous users skipped (no saved libraries to clean).
_auth.cleanup_expired_libraries_async(_auth.get_auth().user_id)
# If the requester is anonymous, bump their signed-cookie counter
# so subsequent /api/run calls hit the 402 above. Signed-in users
# are no-ops here.
resp = jsonify({"job_id": job.job_id})
return _auth.increment_anon_runs_on_response(resp)
# ─── Ownership guard (audit C1) ────────────────────────────────────
# Engine endpoints that take a ``<job_id>`` historically had no
# ownership check — anyone with the id could read or modify any
# other user's job. Closes that gap: signed-in users can only touch
# their own jobs; anonymous jobs (created during a trial) remain
# accessible to anyone with the id (acceptable for the 5-min trial
# scope and required so anon polling still works without a JWT).
def _require_owner_or_403(job: "JobState") -> Optional[Response]:
if job.user_id is None:
return None # anonymous-created job; id-based access only
auth = _auth.get_auth()
if auth.anonymous or auth.user_id != job.user_id:
return jsonify({"error": "not authorized", "kind": "forbidden"}), 403
return None
_AGENT_MESSAGE_MAX_CHARS = 4000
# ─── The Orchestrator (2026-07-26) ──────────────────────────────────
# A long-running agent RUN. The client starts one, then polls for
# events while the model
# works — so a workflow that takes six tool calls shows six tool calls
# landing live instead of resolving into one silent HTTP response.
#
# Every one of these spawns real billed OpenRouter calls, so each takes
# the require_auth_or_quota() gate and its own tight _RL_RULES entry.
# The poll route is deliberately NOT gated the same way — it spends
# nothing, and rate-limiting it would break the very loop it serves.
def _orch_owner_or_403(run: "_orch.Run") -> Optional[Response]:
"""Same rule as _require_owner_or_403 for jobs: a signed-in user's run
is theirs alone; an anonymous run stays reachable by id so trial
polling works without a JWT."""
if not run.owner:
return None
auth = _auth.get_auth()
if auth.anonymous or auth.user_id != run.owner:
return jsonify({"error": "not authorized", "kind": "forbidden"}), 403
return None
@app.post("/api/orchestrator/start")
def orchestrator_start() -> Response:
gate = _auth.require_auth_or_quota()
if gate is not None:
return gate
# Two different questions, in order: has the ANONYMOUS trial run out
# (above), and has this SIGNED-IN account spent its month's credits
# (here). /reply asks the second question too — a run is charged for
# what it actually spends, every resumed turn included — while
# stop/poll spend nothing and are never gated.
gate = _auth.require_monthly_quota()
if gate is not None:
return gate
body = request.get_json(force=True, silent=True) or {}
message = str(body.get("message") or "").strip()
if not message:
return jsonify({"error": "missing message"}), 400
if len(message) > _AGENT_MESSAGE_MAX_CHARS:
return jsonify({
"error": (
f"Message too long — trim to under {_AGENT_MESSAGE_MAX_CHARS:,} "
"characters. For a full plasmid or long sequence, use the "
"relevant tool directly instead of pasting it into chat."
),
"kind": "message_too_long",
}), 400
auth = _auth.get_auth()
run = _orch.create_run(owner=(auth.user_id or ""), anonymous=auth.anonymous)
# Continuing a run that ran out of budget mid-task. The old ceiling
# said "start a new one to continue" and the new one started cold —
# no target, no plan, no record of what had already happened — so the
# advice was unfollowable and the work was simply lost. Passing the
# source run's ID (never a client-supplied brief: that text lands in
# the model's context) rebuilds the handover server-side, and
# _orch.start enforces that the caller owns it.
carry_id = str(body.get("carry_from") or "").strip()
source = _orch.get_run(carry_id) if carry_id else None
try:
# The model picked under the composer. Validated here rather than
# trusted: effective_model() raises for anything this deployment
# cannot actually run, and a bad value must not silently become a
# different model's numbers.
run.model = _orch.sanitize_model(body.get("model"))
_orch.start(run, message, body.get("workspace"), carry_from=source)
except ValueError:
return jsonify({"error": "That run isn't yours to continue.",
"kind": "carry_forbidden"}), 403
resp = jsonify(_orch.public_state(run))
return _auth.increment_anon_runs_on_response(resp)
# Poll lives under its own /poll/ prefix purely so _RL_RULES can give it a
# generous bucket (it fires ~1/s and spends nothing) while start/reply/stop
# keep the tight billed-LLM budget. Prefix matching can't otherwise tell a
# free poll apart from a paid turn.
def _orch_find(run_id: str):
"""A run from memory, or rebuilt from the database if it was evicted.
Runs live in process memory with a 6-hour sweep, so a deploy or an
overnight gap used to lose them outright. Signed-in runs are now
persisted, so before reporting one gone we try to bring it back —
that turns "your session vanished" into a page refresh.
"""
run = _orch.get_run(run_id)
if run is not None:
return run
auth = _auth.get_auth()
if auth.anonymous or not auth.user_id:
return None
return _orch.rehydrate(run_id, auth.user_id)
@app.get("/api/orchestrator/poll/<run_id>")
def orchestrator_poll(run_id: str) -> Response:
run = _orch_find(run_id)
if not run:
# Nothing in memory and nothing persisted (or the caller is
# anonymous, whose runs are never stored). A distinct kind lets
# the client start a fresh one transparently instead of showing
# the user an error for something they didn't do.
return jsonify({"error": "run not found", "kind": "run_expired"}), 410
denied = _orch_owner_or_403(run)
if denied is not None:
return denied
try:
after = int(request.args.get("after") or 0)
except (TypeError, ValueError):
after = 0
state = _orch.public_state(run)
state["events"] = _orch.events_since(run, after)
return jsonify(state)
@app.post("/api/orchestrator/<run_id>/decision")
def orchestrator_decision(run_id: str) -> Response:
# Viewing a counterfactual or recording a scientist's observation uses
# no model/GPU budget. Ownership is the same as replay and polling.
run = _orch_find(run_id)
if not run:
return jsonify({"error": "run not found", "kind": "run_expired"}), 410
denied = _orch_owner_or_403(run)
if denied is not None:
return denied
from dee.core.scientific_decisions import Conflict, DecisionError
try:
decision = _orch.scientific_decision_action(run, request.get_json(silent=True))
except Conflict as exc:
return jsonify({"error": str(exc), "kind": "decision_conflict"}), 409
except (DecisionError, TypeError) as exc:
return jsonify({"error": str(exc), "kind": "invalid_decision"}), 400
return jsonify({"ok": True, "decision": decision})
@app.post("/api/orchestrator/<run_id>/reply")
def orchestrator_reply(run_id: str) -> Response:
gate = _auth.require_auth_or_quota()
if gate is not None:
return gate
run = _orch_find(run_id)
if not run:
return jsonify({"error": "run not found", "kind": "run_expired"}), 410
denied = _orch_owner_or_403(run)
if denied is not None:
return denied
body = request.get_json(force=True, silent=True) or {}
message = str(body.get("message") or "").strip()
if not message:
return jsonify({"error": "missing message"}), 400
if len(message) > _AGENT_MESSAGE_MAX_CHARS:
return jsonify({"error": "Message too long.", "kind": "message_too_long"}), 400
# Working right now? Then this is a CORRECTION, and it is the most
# valuable message the user can send — they have watched the agent
# start down the wrong path and are stopping it. This used to 409 and
# sit in a client-side queue until the run finished, by which point
# the wrong work was already done and paid for. steer() hands it to
# the run loop, which picks it up at its next step boundary.
# A run whose compute allowance is gone cannot answer this. Spawning
# it again would spend one more model call to rediscover that and show
# the user a second dead end. 410 is the shape the cockpit already
# recovers from (it clears the run and re-sends as a fresh start), and
# the server-side handover is what makes that fresh start land on the
# work instead of a blank page. `carry_from` is returned so a client
# that wants to be explicit about it can be.
if run.budget_spent:
return jsonify({
"error": ("That run used its full compute allowance. Your next "
"message starts a fresh run carrying its target, "
"plan and progress over."),
"kind": "run_budget_spent",
"carry_from": run.run_id,
}), 410
if run.status == "running":
if _orch.steer(run, message):
_orch.set_workspace(run, body.get("workspace"), message)
resp = jsonify(dict(_orch.public_state(run), steered=True))
return _auth.increment_anon_runs_on_response(resp)
return jsonify({"error": "run is still working", "kind": "run_busy"}), 409
# A reply spends model time exactly as a start does — metering in
# credits rather than runs exists so that a long conversation costs
# what it costs — so resuming a finished or parked run passes the same
# wall. Steering a run that is already working (above) does not: that
# run was let through when it began, and a correction adds no step of
# its own.
gate = _auth.require_monthly_quota()
if gate is not None:
return gate
run.model = _orch.sanitize_model(body.get("model"))
if not _orch.reply(run, message, body.get("workspace")):
return jsonify({"error": "run is still working", "kind": "run_busy"}), 409
resp = jsonify(_orch.public_state(run))
return _auth.increment_anon_runs_on_response(resp)
@app.get("/api/orchestrator/runs")
def orchestrator_runs() -> Response:
"""This user's past runs, newest first — the history menu's source.
Metadata only (no transcripts): rendering a dropdown shouldn't ship
every event log the user has ever produced. Anonymous callers get an
empty list rather than a 403; their runs genuinely don't exist
server-side, and the cockpit falls back to its local record.
"""
auth = _auth.get_auth()
if auth.anonymous or not auth.user_id:
return jsonify({"ok": True, "runs": [], "anonymous": True})
return jsonify({"ok": True, "runs": _auth.list_agent_runs(auth.user_id)})
# ── catalog verbs ────────────────────────────────────────────────────
# The history dropdown could only ever OPEN a run. Nothing could be
# removed and nothing could be renamed, so a workspace silted up with
# "Untitled run" and abandoned attempts, and the only cure was to wait
# ninety days for the TTL. Both verbs are owner-enforced twice over: the
# caller's own user_id is what reaches the query (never a client-supplied
# one), and the run_id shape is checked before it is interpolated.
def _run_guard(run_id: str):
"""(auth, None) when the caller may act on this run, else (None, resp)."""
auth = _auth.get_auth()
if auth.anonymous or not auth.user_id:
return None, (jsonify({"ok": False, "kind": "signin_required", "error":
"Sign in to manage your runs."}), 403)
if not re.match(r"^[0-9a-f]{32}$", run_id or ""):
return None, (jsonify({"ok": False, "kind": "bad_run_id", "error":
"That isn't a run id."}), 422)
return auth, None
@app.delete("/api/orchestrator/runs/<run_id>")
def orchestrator_run_delete(run_id: str) -> Response:
"""Delete one run — the row, and the process-memory copy behind it."""
auth, denied = _run_guard(run_id)
if denied:
return denied
result = _auth.delete_agent_run(auth.user_id, run_id)
# Evict AFTER the row is gone. A cached Run keeps answering the
# client's poll and re-persists itself at its next terminal
# transition, which would resurrect the row we just deleted.
if result.get("ok"):
from .core import orchestrator as _orch
_orch.forget_run(run_id, auth.user_id)
return jsonify(result)
@app.patch("/api/orchestrator/runs/<run_id>")
def orchestrator_run_rename(run_id: str) -> Response:
"""Retitle one run."""
auth, denied = _run_guard(run_id)
if denied:
return denied
body = request.get_json(force=True, silent=True) or {}
title = str(body.get("title", "")).strip()[:120]
if not title:
return jsonify({"ok": False, "kind": "bad_title", "error":
"Give the run a name."}), 422
result = _auth.rename_agent_run(auth.user_id, run_id, title)
# Keep the live Run in step. It holds its own title and writes it on
# every persist, so without this the next terminal transition would
# quietly put the old name back.
if result.get("ok"):
from .core import orchestrator as _orch
live = _orch.get_run(run_id)
if live is not None and live.owner == auth.user_id:
live.title = title
return jsonify(result)
# ── Engine ↔ Network: source-backed Research Objects ───────────────
# A browser never sends a publishable card. It names one tool_result in a
# run it owns; the server reconstructs the public-safe projection from the
# recorded result and stores an opaque 30-minute draft. Network is where
# the scientist reviews/removes metadata and explicitly publishes it.
_NETWORK_OBJECT_TOOLS = {
"design_crispr_guides": ("crispr_design", "CRISPR guide design", "crispr"),
"design_variant_library": ("protein_engineering", "Protein engineering run", "design"),
"design_primers": ("primer_design", "Primer design", "primers"),
"fold_structure": ("workflow", "Structure prediction", "structure"),
}
def _network_share_parts(run, call_id: str):
# Result-card buttons name their exact event. Conversational sharing
# ("let's share it") intentionally sends no event id, so resolve the
# most recent compatible successful result from this owned run. The
# browser still cannot supply or alter publishable metadata.
if call_id:
event = next((e for e in run.events if e.get("kind") == "tool_result"
and e.get("id") == call_id and e.get("ok")), None)
else:
event = next((e for e in reversed(run.events)
if e.get("kind") == "tool_result" and e.get("ok")
and e.get("name") in _NETWORK_OBJECT_TOOLS), None)
call_id = str((event or {}).get("id") or "")
call = next((e for e in run.events if e.get("kind") == "tool_call"
and e.get("id") == call_id), None)
if not call or not event or event.get("name") not in _NETWORK_OBJECT_TOOLS:
return None
name = event["name"]
object_type, noun, route = _NETWORK_OBJECT_TOOLS[name]
result = ((event.get("ui") or {}).get("payload") or {})
panel = ((event.get("ui") or {}).get("panel") or {})
args = call.get("args") if isinstance(call.get("args"), dict) else {}
# Only fields emitted by the tool are candidates. Workspace/construct
# names are intentionally excluded: a private project can create a
# public summary without its name becoming public by accident.
target = (result.get("gene_symbol") or result.get("wt_identifier")
or result.get("uniprot") or panel.get("wt_identifier")
or args.get("label") or args.get("gene_symbol") or None)
# UI defaults are navigation labels, not scientific targets. Keeping
# one would make an auto-filled share technically non-empty but useless.
if str(target or "").strip().lower() in {"turing run", "this protein", "result"}:
target = None
organism = result.get("organism") or None
item_count = shortlisted = None
if name == "design_crispr_guides":
item_count = int(result.get("n_guides") or len(result.get("guides") or []))
elif name == "design_variant_library":
item_count = int(result.get("n_pool") or 0) or None
shortlisted = len(result.get("variants") or []) or None
elif name == "design_primers":
item_count = len(result.get("pairs") or []) or None
elif name == "fold_structure":
item_count = 1
title = noun + ((" for " + str(target)) if target else "")
summary_bits = [title + " generated with TuringDNA."]
if name == "design_crispr_guides" and item_count is not None:
summary_bits.append(f"{item_count} guides designed.")
elif name == "design_variant_library":
if item_count is not None:
summary_bits.append(f"{item_count} candidate mutations evaluated.")
if shortlisted is not None:
summary_bits.append(f"{shortlisted} variants shortlisted.")
if result.get("region") and len(result.get("region") or []) >= 2:
summary_bits.append(
f"Residues {result['region'][0]}–{result['region'][-1]} targeted.")
elif name == "design_primers" and item_count is not None:
summary_bits.append(f"{item_count} primer pairs designed.")
projection = {
"object_type": object_type, "origin_tool": name,
"title": title[:160], "target": str(target)[:160] if target else None,
"organism": str(organism)[:120] if organism else None,
"summary": " ".join(summary_bits)[:1200],
"item_count": item_count, "shortlisted_count": shortlisted,
"model_label": str(result.get("model") or "")[:160] or None,
"tool_version": str(result.get("version") or "")[:120] or None,
"status": "designed", "deep_link": "/app/#" + route,
}
# Explicit allowlists. These are methodology controls, never molecular
# inputs or outputs; sequence was already stripped from tool_call.
if name == "design_crispr_guides":
methodology = {k: v for k, v in {
"enzyme": result.get("enzyme") or args.get("enzyme"),
"mode": result.get("mode") or args.get("mode"),
"base_editor": args.get("base_editor"),
"max_results": args.get("max_results"),
}.items() if v is not None}
elif name == "design_variant_library":
methodology = {k: v for k, v in {
"host": result.get("host") or args.get("host"),
"percentile": args.get("percentile"), "k": args.get("k"),
"max_mutations": args.get("max_mutations"),
"region_start": (result.get("region") or [None, None])[0],
"region_end": (result.get("region") or [None, None])[-1],
"model": result.get("model"),
}.items() if v is not None}
elif name == "design_primers":
methodology = {k: args.get(k) for k in ("target_start", "target_end")
if args.get(k) is not None}
else:
methodology = {}
return projection, methodology, call_id
@app.post("/api/network/research-drafts")
def network_research_draft_create() -> Response:
auth = _auth.get_auth()
if auth.anonymous or not auth.user_id:
return jsonify({"ok": False, "kind": "signin_required",
"error": "Sign in to share scientific work."}), 403
body = request.get_json(force=True, silent=True) or {}
run_id = str(body.get("run_id") or "")
call_id = str(body.get("call_id") or "")[:128]
run = _orch_find(run_id)
if not run:
return jsonify({"ok": False, "error": "That Turing run is no longer available."}), 404
denied = _orch_owner_or_403(run)
if denied is not None:
return denied
parts = _network_share_parts(run, call_id)
if not parts:
return jsonify({"ok": False, "error": "That result cannot be shared as a Research Object."}), 422
projection, methodology, source_call_id = parts
made = _auth.create_research_object_draft(
auth.user_id, source_kind="orchestrator_event", source_id=run_id,
source_event_id=source_call_id, projection=projection, methodology=methodology)
if not made.get("ok"):
return jsonify(made), 503
_auth.log_event("engine_to_network_share", meta={"tool": projection["origin_tool"]})
draft = made.get("draft") or {}
return jsonify({"ok": True, "draft_id": draft.get("id"),
"expires_at": draft.get("expires_at")})
@app.get("/api/network/forks/<object_id>")
def network_research_fork_get(object_id: str) -> Response:
auth = _auth.get_auth()
if auth.anonymous or not auth.user_id:
return jsonify({"ok": False, "kind": "signin_required",
"error": "Sign in to open this fork."}), 403
fork = _auth.get_research_object_fork(auth.user_id, object_id)
if not fork:
return jsonify({"ok": False, "error": "That fork is unavailable."}), 404
_auth.log_event("network_to_engine_open", meta={"object_type": fork.get("object_type"),
"fork": True})
return jsonify({"ok": True, "fork": fork})
@app.get("/api/network/objects/<object_id>/open")
def network_research_object_open(object_id: str) -> Response:
"""Open a visible Network object without crossing its privacy boundary.
The safe public projection and methodology are useful to every signed-in
viewer. Only the object's author can receive the private Engine run id,
because that lookup is separately owner-filtered in ``auth.py``.
"""
auth = _auth.get_auth()
if auth.anonymous or not auth.user_id:
return jsonify({"ok": False, "kind": "signin_required",
"error": "Sign in to open this scientific work."}), 403
obj = _auth.get_research_object_open(auth.user_id, object_id)
if not obj:
return jsonify({"ok": False, "error": "That Research Object is unavailable."}), 404
run_id = obj.pop("source_run_id", None)
_auth.log_event("network_to_engine_open", meta={
"object_type": obj.get("object_type"), "original": bool(run_id),
})
payload = {"ok": True, "object": obj}
if run_id:
payload["run_id"] = run_id
return jsonify(payload)
@app.route("/api/notifications", methods=["GET", "POST", "OPTIONS"])
def notifications() -> Response:
from dee.core.notifications import Inbox, Unavailable
def respond(data, status=200):
resp = jsonify(data)
resp.status_code = status
resp.headers["Cache-Control"] = "no-store"
resp.vary.add("Origin")
origin = request.headers.get("Origin", "")
if origin and _CORS_ORIGIN_RE.fullmatch(origin):
resp.headers["Access-Control-Allow-Origin"] = origin
resp.headers["Access-Control-Allow-Methods"] = "GET, POST, OPTIONS"
resp.headers["Access-Control-Allow-Headers"] = "authorization, content-type"
resp.headers["Access-Control-Max-Age"] = "600"
return resp
if request.method == "OPTIONS":
return respond({})
auth = _auth.get_auth()
if auth.anonymous or not auth.user_id or not auth.token:
return respond({"error": "Sign in to see your notifications."}, 401)
try:
inbox = Inbox(_auth.SUPABASE_URL,
_auth.SUPABASE_ANON_KEY or _auth.SUPABASE_SERVICE_KEY,
auth.token, auth.user_id)
if request.method == "POST":
body = request.get_json(silent=True)
if not isinstance(body, dict):
return respond({"error": "Expected a notification timestamp."}, 400)
return respond(inbox.mark_read(body.get("through")))
return respond(inbox.read())
except ValueError as exc:
return respond({"error": str(exc)}, 400)
except Unavailable as exc:
return respond({"error": str(exc)}, 503)
@app.get("/api/usage")
def usage() -> Response:
"""Plan + this month's usage, for the account panel.
Only reports numbers it actually has. A count that couldn't be
fetched comes back as null, never 0 — a confident zero next to work
the user knows they did is worse than an honest blank.
"""
auth = _auth.get_auth()
if auth.anonymous or not auth.user_id:
return jsonify({
"ok": True, "anonymous": True, "plan": "anonymous",
"trial_seconds": _auth.ANON_TRIAL_SECONDS,
})
# One source of truth with the enforcement gate — see
# auth.monthly_credit_allowance. Computing the month boundary or the
# total separately here is how the panel and the wall drift apart.
allowance = _auth.monthly_credit_allowance(auth.user_id)
return jsonify({
"ok": True,
"anonymous": False,
"plan": allowance["plan"],
# Credits, not runs, since 2026-09-11 — see auth.CREDIT_USD.
"credits_used": allowance["used"],
# The ceiling, and whether it will actually be applied. `enforced`
# False means "we could not count, so nothing is being blocked" —
# the panel must not draw a hard limit it isn't backing up.
"credits_limit": allowance["limit"],
"credits_remaining": allowance["remaining"],
"quota_enforced": allowance["enforced"],
# Credits earned by referring (0060). Spent only after the plan's
# allowance, and they do not reset, so they are reported apart
# from it. null when unknown, like every other number here.
"bonus_credits": allowance["bonus_remaining"],
"bonus_granted": allowance["bonus_granted"],
# The end of the billing period on a paid plan, the 1st otherwise
# — the same instant the gate resets on (auth.allowance_window).
"resets_at": allowance["resets_at"],
# What "free" actually means, so the panel can say it rather than
# leaving the user to guess where the ceiling is. The SAME rule the
# lazy cleanup enforces (auth.keeps_work_forever), asked of the plan
# we already hold: this line once said `plan == "pro"` and showed a
# Starter subscriber "kept for 45 days" next to their 75 credits.
"retention_days": (None if _auth.plan_keeps_work_forever(allowance["plan"])
else _auth.FREE_LIBRARY_TTL_DAYS),
})
@app.route("/api/billing/config", methods=["GET", "OPTIONS"])
def billing_config() -> Response:
"""What the browser needs to render the Pro upsell and open checkout.
Public values (client token, price id, environment) for everyone;
the customer id, email and current plan only for the signed-in
requester, about themselves. See dee/core/paddle.public_config.
"""
# The approved checkout origin is the website, not the HF iframe.
# Explicitly allow only the site's two canonical origins, with Bearer
# auth (no cookies). No other API route inherits this permission.
def checkout_response(resp):
resp.headers["Cache-Control"] = "no-store"
resp.vary.add("Origin")
if request.headers.get("Origin") in (
"https://turingdna.com", "https://www.turingdna.com"):
resp.headers["Access-Control-Allow-Origin"] = request.headers["Origin"]
resp.headers["Access-Control-Allow-Methods"] = "GET, OPTIONS"
resp.headers["Access-Control-Allow-Headers"] = "authorization"
resp.headers["Access-Control-Max-Age"] = "600"
return resp
if request.method == "OPTIONS":
return checkout_response(app.make_response(("", 204)))
from .core import paddle as _paddle
auth = _auth.get_auth()
uid = None if auth.anonymous else auth.user_id
return checkout_response(jsonify(_paddle.public_config(uid, auth.email)))
@app.route("/api/account/delete", methods=["POST", "OPTIONS"])
def account_delete() -> Response:
"""Self-service account deletion — the user erases their own account and
every piece of personal data, with no email to anyone (GDPR/UK-GDPR
right to erasure). Called cross-origin from the personal dashboard on
turingdna.com, so it answers the CORS preflight and echoes the origin
(turingdna.com only, via _authed_cors).
Guarded two ways: a valid Bearer JWT — only the account's owner holds
one, and it rides the Authorization header, so there is no cookie for a
cross-site page to ride — and the user_id taken from that verified token
(never from the body). An explicit `confirm:true` from the dashboard's
final step guards against a stray programmatic call. The optional
`reason`/`reason_text` feed an anonymous churn record kept BEFORE the
account is gone (see auth.record_deletion_feedback).
"""
if request.method == "OPTIONS":
return _authed_cors(app.make_response(("", 204)))
auth = _auth.get_auth()
if auth.anonymous or not auth.user_id:
return _authed_cors(jsonify({"ok": False, "kind": "signin_required",
"error": "Sign in to manage your account."})), 403
body = request.get_json(force=True, silent=True) or {}
if body.get("confirm") is not True:
return _authed_cors(jsonify({"ok": False, "kind": "confirm_required",
"error": "Deletion was not confirmed."})), 422
result = _auth.delete_account(
auth.user_id,
reason_code=str(body.get("reason") or "")[:64] or None,
reason_text=str(body.get("reason_text") or "")[:1000] or None,
)
status = 200 if result.get("ok") else 500
return _authed_cors(jsonify(result)), status
@app.route("/api/account/subscription", methods=["GET", "OPTIONS"])
def account_subscription() -> Response:
"""What the dashboard's billing panel renders: plan, next charge, any
scheduled cancellation, and a fresh Paddle customer-portal link.
Owner-only and owner-derived. The user_id comes from the verified
Bearer JWT, never from a query string — the portal URL this returns
authenticates whoever holds it straight into that person's billing, so
there must be no parameter that could name somebody else's account.
Same cross-origin shape as the delete endpoint below it: the personal
dashboard lives on turingdna.com and the engine does not, and
_authed_cors echoes that one origin.
"""
if request.method == "OPTIONS":
return _authed_cors(app.make_response(("", 204)))
auth = _auth.get_auth()
if auth.anonymous or not auth.user_id:
return _authed_cors(jsonify({"ok": False, "kind": "signin_required",
"error": "Sign in to manage your account."})), 403
return _authed_cors(jsonify(_auth.subscription_status(auth.user_id))), 200
@app.route("/api/account/subscription/cancel", methods=["POST", "OPTIONS"])
def account_subscription_cancel() -> Response:
"""End the subscription at the end of the period already paid for,
keeping the account and everything in it.
Guarded exactly like deletion — a valid Bearer JWT (no cookie, so no
cross-site page can ride it), the user_id taken from that token, and an
explicit `confirm:true` so a stray programmatic call cannot cancel
somebody's plan. The optional reason feeds the same anonymous churn
table, tagged 'cancel' so it is never averaged with deletions.
Returns 502, not 500, when Paddle refuses: the engine is fine and the
upstream is not, and the difference decides whether the user should
retry here or go to Paddle directly.
"""
if request.method == "OPTIONS":
return _authed_cors(app.make_response(("", 204)))
auth = _auth.get_auth()
if auth.anonymous or not auth.user_id:
return _authed_cors(jsonify({"ok": False, "kind": "signin_required",
"error": "Sign in to manage your account."})), 403
body = request.get_json(force=True, silent=True) or {}
if body.get("confirm") is not True:
return _authed_cors(jsonify({"ok": False, "kind": "confirm_required",
"error": "Cancellation was not confirmed."})), 422
result = _auth.cancel_subscription(
auth.user_id,
reason_code=str(body.get("reason") or "")[:64] or None,
reason_text=str(body.get("reason_text") or "")[:1000] or None,
)
if result.get("ok"):
status = 200
elif result.get("kind") == "no_subscription":
status = 409
elif result.get("kind") == "not_configured":
# 503, not 502: nothing upstream refused us — we never asked,
# because this deployment has no key to ask with. Retrying will
# not help, and the message says so.
status = 503
else:
status = 502
return _authed_cors(jsonify(result)), status
@app.post("/api/paddle/webhook")
def paddle_webhook() -> Response:
"""Paddle's server calls this when a subscription changes. It is the
ONLY thing that writes billing state — no browser path flips a plan.
Authorised by the signature alone: no JWT (Paddle has none), no CSRF
token (Paddle can't hold one). The raw body must be read UNMODIFIED —
the signature is over these exact bytes, so it is verified before the
JSON is parsed.
"""
from .core import paddle as _paddle
raw = request.get_data() or b""
if not _paddle.verify_webhook(raw, request.headers.get("Paddle-Signature")):
# 403, and nothing logged from the body: an unverified caller does
# not get to write to our log either.
return jsonify({"error": "invalid signature", "kind": "forbidden"}), 403
try:
event = json.loads(raw.decode("utf-8"))
except Exception: # noqa: BLE001
return jsonify({"error": "bad payload"}), 400
result = _paddle.apply_event(event)
# A verified event we understood but could not persist is the one case
# worth a retry (500). Everything else — applied, or deliberately
# ignored — is a 200 so Paddle stops redelivering it.
if result.get("handled") and not result.get("ok"):
return jsonify({"error": "could not apply"}), 500
return jsonify({"ok": True, "handled": bool(result.get("handled"))}), 200
@app.post("/api/orchestrator/<run_id>/stop")
def orchestrator_stop(run_id: str) -> Response:
run = _orch.get_run(run_id)
if not run:
# No rehydrate here on purpose: stopping a run that is not in
# memory is a no-op by definition — there is no thread to stop.
return jsonify({"error": "run not found", "kind": "run_expired"}), 410
denied = _orch_owner_or_403(run)
if denied is not None:
return denied
_orch.stop(run)
return jsonify(_orch.public_state(run))
@app.get("/api/status/<job_id>")
def status(job_id: str) -> Response:
job = _get_job(job_id)
if not job:
return jsonify({"error": "unknown job"}), 404
denied = _require_owner_or_403(job)
if denied is not None:
return denied
return jsonify(job.public())
@app.get("/api/result/<job_id>")
def result(job_id: str) -> Response:
job = _get_job(job_id)
if not job:
return jsonify({"error": "unknown job"}), 404
denied = _require_owner_or_403(job)
if denied is not None:
return denied
if job.status != "done":
return jsonify({"error": f"job not done (status={job.status})"}), 409
# csv_path / desktop_path intentionally omitted (audit H2): they
# leak the server's filesystem layout. Downloads use the
# /api/download/<job_id> endpoint, which constructs the path
# server-side.
return jsonify(
{
"wt_identifier": job.wt_identifier,
"wt_protein": job.wt_protein,
"variants": job.variants or [],
"library_id": job.library_id,
"settings_used": job.settings_used,
"started_at": job.started_at,
"elapsed_seconds": job.elapsed(),
"global_prior": job.global_prior_info or {"applied": False, "substitution_types": 0},
}
)
@app.post("/api/variants/<job_id>/<variant_id>/dna")
def edit_variant_dna(job_id: str, variant_id: str) -> Response:
"""Persist a user-edited DNA sequence for one variant.
The frontend POSTs ``{"dna": "ACGT..."}``. We re-derive every
column from those bytes (translation, mutations diff, PCR metrics,
restriction-site count), replace ``job.variants[i]`` in place, and
rewrite the on-disk CSV atomically so subsequent /api/download
calls return the edited sequence. Returns the full updated row so
the frontend can replace its local copy without a second fetch.
Rejects with 400 if the DNA isn't ACGT or isn't a multiple of 3.
A premature stop is allowed (nonsense/truncation mutation) and simply
truncates the encoded product.
"""
job = _get_job(job_id)
if not job:
return jsonify({"error": "unknown job"}), 404
denied = _require_owner_or_403(job)
if denied is not None:
return denied
if job.status != "done":
return jsonify({"error": f"job not done (status={job.status})"}), 409
if not job.variants:
return jsonify({"error": "job has no variants"}), 404
body = request.get_json(force=True, silent=True) or {}
edited_dna = (body.get("dna") or "").strip()
with _edit_lock_for(job_id):
idx = next(
(i for i, v in enumerate(job.variants)
if v.get("Variant_ID") == variant_id),
None,
)
if idx is None:
return jsonify({"error": f"unknown variant {variant_id!r}"}), 404
try:
new_row = _rederive_variant_row(
variant_id=variant_id,
edited_dna=edited_dna,
wt_protein=job.wt_protein,
original_row=job.variants[idx],
)
except ValueError as exc:
return jsonify({"error": str(exc), "kind": "validation"}), 400
except Exception as exc: # noqa: BLE001
logger.exception("Re-derive failed for %s/%s", job_id, variant_id)
return jsonify({"error": f"{type(exc).__name__}: {exc}"}), 500
job.variants[idx] = new_row
try:
_atomic_rewrite_csv(job)
except Exception as exc: # noqa: BLE001
# The in-memory state was updated but the on-disk CSV
# rewrite failed. Surface this so the user knows downloads
# may temporarily diverge from screen state.
logger.exception("CSV rewrite failed for %s", job_id)
return jsonify(
{"error": f"saved in memory but CSV rewrite failed: {exc}",
"row": new_row, "kind": "csv_warning"},
), 207
return jsonify({"row": new_row})
@app.get("/api/download/<job_id>")
def download(job_id: str) -> Response:
job = _get_job(job_id)
if not job or not job.csv_path:
return jsonify({"error": "no result available"}), 404
denied = _require_owner_or_403(job)
if denied is not None:
return denied
fmt = request.args.get("format", "csv").lower()
stem = Path(job.csv_path).stem
try:
data, mimetype, ext = _render_download(job, fmt)
except ValueError as exc:
return jsonify({"error": str(exc)}), 400
return send_file(
io.BytesIO(data) if isinstance(data, bytes) else data,
mimetype=mimetype,
as_attachment=True,
download_name=f"{stem}.{ext}",
)
@app.post("/api/identify")
def identify_start() -> Response:
"""Kick off an NCBI-BLAST identification for the session's WT protein.
Returns ``{job_id, cached, hits?}``. If we already ran BLAST on this
exact sequence we return the prior result inline with cached=true.
"""
body = request.get_json(force=True, silent=True) or {}
sid = body.get("session_id")
# A raw protein is an ALTERNATIVE to a session, not a shortcut past it.
# Libraries designed inside a Turing conversation never create a
# server-side job, so they have no session at all — and requiring one
# made the wild-type structure lookup permanently dead for every
# agent-designed library while looking like an ordinary button.
raw_protein = str(body.get("protein") or "").strip()
if not sid and not raw_protein:
return jsonify({"error": "missing session_id or protein"}), 400
session = _get_session(sid) if sid else None
if sid and not session:
return jsonify({"error": "session expired"}), 410
# Audit H1: BLAST consent must be enforced server-side, not just
# by the consent modal in the UI. The frontend posts
# `blast_consent: true` only after the user has clicked Continue
# in the modal; we require it here so anyone calling /api/identify
# directly (curl, script, etc.) still has to opt in to leak the
# sequence to NCBI.
if not body.get("blast_consent"):
return jsonify({
"error": (
"BLAST consent is required. The wild-type protein "
"sequence is sent to ncbi.nlm.nih.gov; please confirm "
"in the UI before retrying."
),
"kind": "consent_required",
}), 403
cds_feature = body.get("cds_feature")
if session is None:
# Sequence-only path. Validated the same way every other pasted
# protein is, so a junk body cannot become a BLAST job.
# classify_sequence, not an alphabet test. A, T, G and C are all
# valid amino-acid letters, so a pasted DNA sequence passes any
# naive "are these residues?" check and then gets blastp'd as if it
# were protein — the realistic mistake, and a silently wrong answer
# rather than an error. Anything the classifier will not call
# protein outright is refused.
from dee.core.sequence import classify_sequence
protein = "".join(c for c in raw_protein.upper() if c.isalpha())
if len(protein) < 12 or classify_sequence(protein) != "protein":
return jsonify({
"error": ("'protein' must be 12+ residues and readable as "
"protein. A sequence of only A/C/G/T/U/N is DNA "
"or is indistinguishable from it, and blastp "
"would return a confidently wrong answer."),
}), 400
return _identify_from_protein(protein)
# Re-derive the protein to BLAST. For sessions with discovered ORFs
# the user may have picked a specific one — use that if provided.
orf_map = session.get("orfs") or {}
if orf_map and cds_feature and cds_feature in orf_map:
tmp_path = Path(STATE_DIR / f"blast_input_{uuid.uuid4().hex[:8]}.fa")
tmp_path.write_text(f">{cds_feature}\n{orf_map[cds_feature]}\n")
record = parse_input(tmp_path, require_start=False, require_stop=False)
tmp_path.unlink(missing_ok=True)
else:
record = parse_input(
Path(session["path"]),
require_start=False,
require_stop=False,
cds_feature=cds_feature,
)
protein = record.protein
if not protein:
return jsonify({"error": "no protein available to identify"}), 400
return _identify_from_protein(protein)
def _identify_from_protein(protein: str) -> Response:
"""Submit (or reuse) a BLAST job for one protein.
Shared by both entry paths — session-derived and sequence-supplied — so
the hash-keyed cache, the job registry and the response shape cannot
drift between them.
"""
seq_hash = _hash_sequence(protein)
with _BLAST_LOCK:
cached = _BLAST_CACHE.get(seq_hash)
if cached and cached.status in {"done", "running", "submitting", "parsing", "pending"}:
return jsonify(
{
"job_id": cached.job_id,
"cached": cached.status == "done",
**cached.public(),
}
)
job = BlastJob(job_id=uuid.uuid4().hex[:12], seq_hash=seq_hash)
_BLAST_CACHE[seq_hash] = job
_BLAST_BY_ID[job.job_id] = job
threading.Thread(
target=_run_blast, args=(job, protein), daemon=True
).start()
return jsonify({"job_id": job.job_id, "cached": False, **job.public()})
@app.get("/api/identify/<job_id>")
def identify_status(job_id: str) -> Response:
with _BLAST_LOCK:
job = _BLAST_BY_ID.get(job_id)
if not job:
return jsonify({"error": "unknown job"}), 404
return jsonify(job.public())
# ============================================================ PRIMER PICKER
# Batch-analyse candidate primers: score each against the user's template
# (on-server, private) and against NCBI nt for off-target specificity
# (opt-in), then rank and recommend the best. Sign-in gated like CRISPR.
@app.post("/api/primers/analyze")
def primers_analyze() -> Response:
from .core import primers as _P
auth = _auth.get_auth()
if auth.anonymous:
return jsonify({
"error": "The Primer Picker requires a free account. Sign in to continue.",
"kind": "signin_required",
"signup_url": "https://turingdna.com/signin/?from=primers",
}), 403
body = request.get_json(force=True, silent=True) or {}
primers_text = (body.get("primers") or "").strip()
template = re.sub(r"\s+", "", (body.get("template") or "")).upper()
scan_ncbi = bool(body.get("scan_ncbi"))
strict_3p = bool(body.get("strict_3p"))
organism = (body.get("organism") or "human").lower()
if organism not in _PRIMER_ORGANISMS:
organism = "human"
if not primers_text:
return jsonify({"error": "Paste at least one primer."}), 400
if template and not re.fullmatch(r"[ACGTN]+", template):
return jsonify({"error": "Template must be DNA (A/C/G/T/N) only."}), 400
if len(template) > _P.MAX_TEMPLATE:
return jsonify({"error": (
f"Template is longer than {_P.MAX_TEMPLATE:,} bp — paste just the "
"region you're amplifying.")}), 400
parsed, warnings = _P.parse_primers(primers_text)
if not parsed:
return jsonify({"error": "No valid primers found.", "warnings": warnings}), 400
scored = [_P.score_primer(p, template) for p in parsed]
if not scan_ncbi:
return jsonify({
"status": "done",
"result": _P.rank(scored, template),
"warnings": warnings,
})
# NCBI scan sends the primers off-server → require explicit consent
# (same posture as the BLAST identify gate).
if not body.get("blast_consent"):
return jsonify({
"error": ("The specificity scan sends your primers to "
"blast.ncbi.nlm.nih.gov; confirm in the UI before retrying."),
"kind": "consent_required",
}), 403
seq_hash = _hash_sequence("|".join(p["seq"] for p in parsed)
+ "::" + template + "::" + organism
+ ("::strict" if strict_3p else ""))
with _PRIMER_LOCK:
cached = _PRIMER_CACHE.get(seq_hash)
if cached and cached.status in {"done", "scanning", "pending"}:
return jsonify({
"job_id": cached.job_id,
"status": cached.status,
"result": cached.result(),
"warnings": cached.warnings,
})
job = PrimerJob(job_id=uuid.uuid4().hex[:12], seq_hash=seq_hash,
user_id=auth.user_id,
primers=scored, template=template, organism=organism,
three_prime_strict=strict_3p, warnings=warnings)
_PRIMER_CACHE[seq_hash] = job
_PRIMER_BY_ID[job.job_id] = job
threading.Thread(target=_run_primer_blast, args=(job,), daemon=True).start()
return jsonify({
"job_id": job.job_id,
"status": "scanning",
"result": job.result(), # template-only ranking while NCBI runs
"warnings": warnings,
})
def _primers_signin_gate():
auth = _auth.get_auth()
if auth.anonymous:
return jsonify({"error": "This requires a free account. Sign in to continue.",
"kind": "signin_required",
"signup_url": "https://turingdna.com/signin/?from=primers"}), 403
return None
@app.post("/api/primers/design")
def primers_design() -> Response:
"""De novo primer design from a template (offline — no NCBI)."""
from .core import primers as _P
gate = _primers_signin_gate()
if gate is not None:
return gate
body = request.get_json(force=True, silent=True) or {}
template = re.sub(r"\s+", "", (body.get("template") or "")).upper()
if not template:
return jsonify({"error": "Paste a template to design primers against."}), 400
if not re.fullmatch(r"[ACGTN]+", template):
return jsonify({"error": "Template must be DNA (A/C/G/T/N) only."}), 400
if len(template) > _P.MAX_TEMPLATE:
return jsonify({"error": f"Template longer than {_P.MAX_TEMPLATE:,} bp — trim to the region of interest."}), 400
def _oi(v):
try:
return int(v)
except (TypeError, ValueError):
return None
def _f(v, d):
try:
return float(v)
except (TypeError, ValueError):
return d
s = body.get("settings") if isinstance(body.get("settings"), dict) else {}
kw = dict(
product_min=max(40, _oi(s.get("product_min")) or 100),
product_max=min(_P.MAX_PRODUCT, _oi(s.get("product_max")) or 1000),
len_min=max(_P.MIN_LEN, _oi(s.get("len_min")) or 18),
len_max=min(_P.MAX_LEN, _oi(s.get("len_max")) or 27),
tm_min=_f(s.get("tm_min"), 57.0), tm_max=_f(s.get("tm_max"), 63.0),
tm_opt=_f(s.get("tm_opt"), 60.0),
gc_min=_f(s.get("gc_min"), 40.0), gc_max=_f(s.get("gc_max"), 60.0),
max_pairs=max(1, min(20, _oi(s.get("max_pairs")) or 5)),
)
try:
result = _P.design_primers(template, _oi(body.get("target_start")),
_oi(body.get("target_end")), **kw)
except Exception: # noqa: BLE001
logger.exception("primer design failed")
return jsonify({"error": "Primer design failed — check the inputs."}), 500
return jsonify(result), (200 if result.get("ok") else 400)
@app.post("/api/primers/multiplex")
def primers_multiplex() -> Response:
"""Multiplex compatibility: pairwise cross-dimers + amplicon size clashes."""
from .core import primers as _P
gate = _primers_signin_gate()
if gate is not None:
return gate
body = request.get_json(force=True, silent=True) or {}
prims = body.get("primers")
if isinstance(prims, str):
parsed, _w = _P.parse_primers(prims)
prims = [{"name": p["name"], "seq": p["seq"]} for p in parsed]
if not isinstance(prims, list) or not prims:
return jsonify({"error": "Provide at least two primers."}), 400
products = body.get("products") if isinstance(body.get("products"), list) else None
try:
result = _P.multiplex_check(prims[:_P.MAX_PRIMERS], products=products)
except Exception: # noqa: BLE001
logger.exception("multiplex check failed")
return jsonify({"error": "Multiplex check failed — check the inputs."}), 500
return jsonify({"ok": True, **result})
@app.post("/api/primers/local-specificity")
def primers_local_spec() -> Response:
"""In-silico PCR against a user-uploaded genome/FASTA — runs entirely on
our server, so primers never leave (unlike the NCBI scan)."""
from .core import primers as _P
gate = _primers_signin_gate()
if gate is not None:
return gate
body = request.get_json(force=True, silent=True) or {}
prims = body.get("primers")
if isinstance(prims, str):
parsed, _w = _P.parse_primers(prims)
prims = [{"id": p["id"], "name": p["name"], "seq": p["seq"]} for p in parsed]
if not isinstance(prims, list) or not prims:
return jsonify({"error": "Provide the primers to check."}), 400
fasta = body.get("fasta") or body.get("genome") or ""
if not fasta.strip():
return jsonify({"error": "Upload a genome or paste a FASTA to scan against."}), 400
if len(fasta) > 12_000_000:
return jsonify({"error": "Genome larger than ~12 Mb — for human-scale targets use the NCBI scan instead."}), 413
try:
result = _P.local_specificity(prims[:_P.MAX_PRIMERS], fasta)
except Exception: # noqa: BLE001
logger.exception("local specificity failed")
return jsonify({"error": "Local specificity scan failed — check the inputs."}), 500
return jsonify(result), (200 if result.get("ok") else 400)
@app.get("/api/primers/<job_id>")
def primers_status(job_id: str) -> Response:
with _PRIMER_LOCK:
job = _PRIMER_BY_ID.get(job_id)
if not job:
return jsonify({"error": "unknown job"}), 404
# Same rule as _require_owner_or_403 (audit C1): every PrimerJob has
# an owner (the creating route is sign-in gated, unlike /api/identify
# or /api/run), so unlike those, there is no anonymous-job exception
# here — a bare job_id must never be enough to read someone else's
# primer sequences and template DNA.
auth = _auth.get_auth()
if auth.anonymous or auth.user_id != job.user_id:
return jsonify({"error": "not authorized", "kind": "forbidden"}), 403
return jsonify(job.public())
# ============================================================ PLASMID STUDIO
# Import GenBank/FASTA/raw → annotated plasmid model; restriction analysis +
# digest (1088 REBASE enzymes via Biopython); GenBank/FASTA export; per-user
# saved-plasmid library. Sign-in gated like CRISPR / the Primer Picker.
def _plasmid_signin_guard():
auth = _auth.get_auth()
if auth.anonymous:
return None, (jsonify({
"error": "Plasmid Studio requires a free account. Sign in to continue.",
"kind": "signin_required",
"signup_url": "https://turingdna.com/signin/?from=plasmid",
}), 403)
return auth, None
def _validate_plasmid_body(body):
"""Pull (sequence, topology, name, features) from a request body for the
analysis/export endpoints, with bounds. Returns (data, error_response)."""
from .core import plasmid as _PL
seq = re.sub(r"\s+", "", (body.get("sequence") or "")).upper()
if not seq:
return None, (jsonify({"error": "missing 'sequence'"}), 400)
if len(seq) > _PL.MAX_LEN:
return None, (jsonify({"error": f"Sequence longer than {_PL.MAX_LEN:,} bp."}), 400)
if not re.fullmatch(r"[ACGTNRYSWKMBDHV]+", seq):
return None, (jsonify({"error": "Sequence must be DNA."}), 400)
topology = "linear" if (body.get("topology") == "linear") else "circular"
# Sanitize features: clamp coords, force colors to safe hex, cap count.
# Defends export/save + the SVG/style rendering from malformed or
# injection-laden feature dicts (audit hardening).
return {"sequence": seq, "topology": topology,
"name": str(body.get("name", "") or "plasmid")[:120],
"features": _PL.sanitize_features(body.get("features"), len(seq))}, None
@app.post("/api/plasmid/import")
def plasmid_import() -> Response:
from .core import plasmid as _PL
_auth_ctx, denied = _plasmid_signin_guard()
if denied:
return denied
body = request.get_json(force=True, silent=True) or {}
topology = body.get("topology") if body.get("topology") in ("circular", "linear") else None
b64 = body.get("b64")
try:
if b64:
# Binary upload (e.g. SnapGene .dna) base64-encoded in the JSON
# body. Bound the decoded size BEFORE handing to the parser.
import base64
import binascii
if not isinstance(b64, str) or len(b64) > 4 * (_PL.MAX_INPUT_TEXT // 3 + 4):
return jsonify({"error": "File too large."}), 400
try:
raw = base64.b64decode(b64, validate=True)
except (binascii.Error, ValueError):
return jsonify({"error": "Could not decode the uploaded file."}), 400
filename = str(body.get("filename") or "")[:200]
plasmid = _PL.parse_binary(raw, filename=filename, topology=topology)
else:
text = body.get("text") or ""
plasmid = _PL.parse_plasmid(text, topology=topology)
except ValueError as exc:
return jsonify({"error": str(exc)}), 400
if body.get("annotate", True):
_PL.auto_annotate(plasmid)
return jsonify({"ok": True, "plasmid": plasmid})
@app.post("/api/plasmid/restriction")
def plasmid_restriction() -> Response:
from .core import plasmid as _PL
_auth_ctx, denied = _plasmid_signin_guard()
if denied:
return denied
body = request.get_json(force=True, silent=True) or {}
data, err = _validate_plasmid_body(body)
if err:
return err
enzymes = body.get("enzymes") if isinstance(body.get("enzymes"), list) else None
enzymes = [str(e)[:24] for e in enzymes][:120] if enzymes else None
return jsonify({"ok": True,
"analysis": _PL.restriction_analysis(data["sequence"], data["topology"], enzymes)})
@app.post("/api/plasmid/digest")
def plasmid_digest() -> Response:
from .core import plasmid as _PL
_auth_ctx, denied = _plasmid_signin_guard()
if denied:
return denied
body = request.get_json(force=True, silent=True) or {}
data, err = _validate_plasmid_body(body)
if err:
return err
enzymes = body.get("enzymes") if isinstance(body.get("enzymes"), list) else []
enzymes = [str(e)[:24] for e in enzymes][:20]
if not enzymes:
return jsonify({"error": "Pick at least one enzyme to digest with."}), 400
return jsonify({"ok": True,
"digest": _PL.digest(data["sequence"], data["topology"], enzymes)})
@app.post("/api/plasmid/export")
def plasmid_export() -> Response:
from .core import plasmid as _PL
_auth_ctx, denied = _plasmid_signin_guard()
if denied:
return denied
body = request.get_json(force=True, silent=True) or {}
data, err = _validate_plasmid_body(body)
if err:
return err
fmt = (body.get("format") or "genbank").lower()
plasmid = {"name": data["name"], "sequence": data["sequence"],
"topology": data["topology"], "length": len(data["sequence"]),
"features": data["features"]}
stem = re.sub(r"[^A-Za-z0-9_.-]", "_", data["name"]) or "plasmid"
try:
if fmt in ("fasta", "fa"):
payload, mime, ext = _PL.build_fasta(plasmid), "text/x-fasta", "fasta"
else:
payload, mime, ext = _PL.build_genbank(plasmid), "chemical/x-genbank", "gb"
except Exception as exc: # noqa: BLE001
return jsonify({"error": f"Export failed: {exc}"}), 500
return send_file(io.BytesIO(payload.encode("utf-8")), mimetype=mime,
as_attachment=True, download_name=f"{stem}.{ext}")
@app.post("/api/align")
def align_sequences() -> Response:
"""Pairwise alignment (global/local) of two sequences. Stateless utility,
no sign-in needed; bounded by align()'s size cap + the global rate limit."""
from .core import align as _AL
body = request.get_json(force=True, silent=True) or {}
a, b = body.get("a") or "", body.get("b") or ""
mode = "local" if body.get("mode") == "local" else "global"
try:
result = _AL.align(a, b, mode=mode)
except ValueError as exc:
return jsonify({"error": str(exc)}), 400
except Exception: # noqa: BLE001
logger.exception("alignment failed")
return jsonify({"error": "Alignment failed — check the two sequences."}), 500
return jsonify({"ok": True, **result})
# ── Attribution ───────────────────────────────────────────────────────
# The whole point of the construct being the selected object: when a tool
# saves something and the client sent the id it was working under, the
# link is RECORDED at the moment it happened rather than reconstructed
# later by matching name strings.
#
# Deliberately best-effort and silent. Attribution failing must never fail
# the save the user actually asked for — the artifact still exists in its
# own table, it is only the cross-reference that is missing.
def _attribute(auth, body, kind, result):
try:
cid = str((body or {}).get("construct_id") or "").strip()
new_id = (result or {}).get("id")
if not cid or not new_id or auth.anonymous:
return
_auth.attach_artifact(auth.user_id, cid, kind, str(new_id))
# ...and the EDGE. attach_artifact says the construct has this
# artifact; this says where the artifact came from. The client
# sends `derived_from` when it knows the specific parent (a
# plasmid region handed to CRISPR); when it does not, the honest
# parent is the construct's own sequence, which is the root of
# every graph and is always true.
src = (body or {}).get("derived_from") or {}
from_kind = str(src.get("kind") or "").strip()
from_id = str(src.get("id") or "").strip()
if from_kind and from_id:
origin = "handoff"
else:
from_kind, from_id, origin = "sequence", cid, "handoff"
_auth.record_edge(
auth.user_id,
from_kind=from_kind, from_id=from_id,
to_kind=kind, to_id=str(new_id),
relation=str(src.get("relation") or "derived_from"),
origin=origin, construct_id=cid,
detail={k: v for k, v in (src.get("detail") or {}).items()
if isinstance(k, str)} or {})
except Exception: # noqa: BLE001
app.logger.warning("attribution failed for %s", kind, exc_info=True)
@app.post("/api/plasmid/save")
def plasmid_save() -> Response:
auth, denied = _plasmid_signin_guard()
if denied:
return denied
body = request.get_json(force=True, silent=True) or {}
data, err = _validate_plasmid_body(body)
if err:
return err
try:
gc = float(body.get("gc")) if body.get("gc") is not None else None
except (TypeError, ValueError):
gc = None
result = _auth.save_plasmid(
auth.user_id,
name=data["name"], topology=data["topology"],
length=len(data["sequence"]), gc=gc,
source=str(body.get("source", "") or "")[:16] or None,
sequence=data["sequence"], features=data["features"],
)
_attribute(auth, body, "plasmid", result)
_auth.cleanup_expired_plasmids_async(auth.user_id)
return jsonify(result), (200 if result.get("ok") else 502)
# ── Turing's memory ───────────────────────────────────────────────────
# Facts the agent carries between conversations. The whole point of this
# surface is that the user can SEE and change what is remembered about
# them — an agent whose beliefs are only visible in its behaviour is not a
# glass box. Nothing here is optional polish; without it `remember` would
# be a hidden accumulator.
def _memory_signin_guard():
auth = _auth.get_auth()
if auth.anonymous:
return None, (jsonify({
"error": "Memory needs a free account. Sign in to continue.",
"kind": "signin_required",
"signup_url": "https://turingdna.com/signin/?from=memory",
}), 403)
return auth, None
@app.get("/api/memory")
def memory_list() -> Response:
auth, denied = _memory_signin_guard()
if denied:
return denied
rows = _auth.list_memory(auth.user_id)
return jsonify({
"ok": True,
"memories": rows,
# Split so the client can show proposals as something to decide on
# rather than as facts already in force.
"kept": sum(1 for r in rows if r.get("status") == "kept"),
"proposed": sum(1 for r in rows if r.get("status") == "proposed"),
})
@app.post("/api/memory")
def memory_add() -> Response:
"""The user adding a fact themselves. Lands 'kept' — they are the one
who decides, so there is nothing to confirm."""
auth, denied = _memory_signin_guard()
if denied:
return denied
body = request.get_json(silent=True) or {}
fact = str(body.get("fact") or "")
why = _auth.memory_rejection(fact)
if why:
return jsonify({"ok": False, "error": why}), 400
out = _auth.remember_fact(auth.user_id, fact,
kind=str(body.get("kind") or "context"))
if not out.get("ok"):
return jsonify(out), 400
if out.get("id"):
_auth.keep_memory(auth.user_id, out["id"])
return jsonify({"ok": True, "id": out.get("id"), "status": "kept"})
@app.post("/api/memory/<memory_id>/keep")
def memory_keep(memory_id: str) -> Response:
auth, denied = _memory_signin_guard()
if denied:
return denied
out = _auth.keep_memory(auth.user_id, memory_id)
return jsonify(out), (200 if out.get("ok") else 400)
@app.patch("/api/memory/<memory_id>")
def memory_edit(memory_id: str) -> Response:
auth, denied = _memory_signin_guard()
if denied:
return denied
body = request.get_json(silent=True) or {}
out = _auth.edit_memory(auth.user_id, memory_id, str(body.get("fact") or ""))
return jsonify(out), (200 if out.get("ok") else 400)
@app.delete("/api/memory/<memory_id>")
def memory_forget(memory_id: str) -> Response:
auth, denied = _memory_signin_guard()
if denied:
return denied
out = _auth.forget_memory(auth.user_id, memory_id)
return jsonify(out), (200 if out.get("ok") else 400)
# ── The scientist's own standing instructions (agent_instructions, 0052) ──
# Their CLAUDE.md. Distinct from /api/memory above: memory holds facts the
# AGENT proposed and they kept; this is prose they wrote about how they
# want to be worked with. See orchestrator._instructions_note for why the
# two are injected under different headings.
@app.get("/api/fold")
def fold_capability() -> Response:
"""Can THIS caller fold privately? Asked before the sequence exists.
The client needs this to decide whether to show the "Send sequence to
Meta?" consent panel at all — a paid user must never see it, and
discovering entitlement by POSTing the sequence first would mean
building the whole flow around a request that is usually refused.
Reports `private` only when the plan allows it AND the backend is
actually configured. "Availability must mean reachable": a plan flag
alone would promise a paid feature that 503s.
"""
from .core import folding as _fold
auth = _auth.get_auth()
plan = "free" if auth.anonymous else _auth.plan_for(auth.user_id)
entitled = plan != "free"
return jsonify({
"private": bool(entitled and _fold.is_configured()),
"entitled": entitled,
"plan": plan,
"max_residues": _fold.max_residues(),
})
@app.post("/api/fold")
def fold_private() -> Response:
"""Fold a sequence on OUR GPU. Paid plans only.
The free tier keeps the existing client-side path to
api.esmatlas.com, behind its consent dialog — nothing is taken away.
What a paid plan buys is that the sequence never reaches a third
party at all: browser -> Space -> our Modal GPU.
FAILS CLOSED, TWICE OVER. plan_for() already resolves to 'free' on
any uncertainty (anonymous, Supabase down, unknown plan name), so an
error denies the perk rather than granting it. And if the backend is
unreachable this returns an error rather than falling back to the
public endpoint: quietly shipping a paying customer's sequence to
Meta because our own GPU was down is the one outcome this feature
exists to prevent, and the one they would never find out about.
"""
from .core import folding as _fold
auth = _auth.get_auth()
if auth.anonymous:
return jsonify({
"error": "Private folding needs a signed-in account.",
"kind": "signin_required",
"signup_url": "https://turingdna.com/signin/?from=fold",
}), 403
plan = _auth.plan_for(auth.user_id)
if plan == "free":
return jsonify({
"error": "Private folding is on the paid plans. On Free you can "
"still fold through the public ESM Atlas service.",
"kind": "upgrade_required",
"plan": plan,
"pricing_url": "https://turingdna.com/pricing/",
}), 402
body = request.get_json(silent=True) or {}
sequence = str(body.get("sequence") or "")
try:
out = _fold.fold(sequence)
except _fold.NotConfigured as exc:
# Configured != deployed. Say which it is, rather than implying
# the user did something wrong.
logger.warning("private fold not configured: %s", exc)
return jsonify({"error": "Private folding isn't switched on for this "
"deployment yet.", "kind": "not_configured"}), 503
except _fold.FoldError as exc:
return jsonify({"error": str(exc), "kind": "fold_failed"}), 502
return jsonify({"ok": True, "pdb": out.get("pdb", ""),
"length": out.get("length"),
"seconds": out.get("seconds"),
"source": "TuringDNA GPU"})
@app.get("/api/instructions")
def instructions_get() -> Response:
auth, denied = _memory_signin_guard()
if denied:
return denied
row = _auth.get_instructions(auth.user_id)
return jsonify({
"ok": True,
"body": row.get("body") or "",
"enabled": bool(row.get("enabled", True)),
"updated_at": row.get("updated_at"),
# The client renders the counter against the real server limit
# rather than a hardcoded copy that can drift from the schema.
"limit": _auth._INSTRUCTIONS_MAX,
})
@app.put("/api/instructions")
def instructions_put() -> Response:
auth, denied = _memory_signin_guard()
if denied:
return denied
body = request.get_json(silent=True) or {}
text = str(body.get("body") or "")
why = _auth.instructions_rejection(text)
if why:
return jsonify({"ok": False, "error": why}), 400
enabled = body.get("enabled")
out = _auth.set_instructions(
auth.user_id, text,
enabled=(None if enabled is None else bool(enabled)))
return jsonify(out), (200 if out.get("ok") else 400)
@app.get("/api/plasmid/library")
def plasmid_library() -> Response:
auth, denied = _plasmid_signin_guard()
if denied:
return denied
return jsonify({"ok": True, "plasmids": _auth.list_plasmids(auth.user_id)})
@app.get("/api/plasmid/library/<plasmid_id>")
def plasmid_library_get(plasmid_id: str) -> Response:
auth, denied = _plasmid_signin_guard()
if denied:
return denied
p = _auth.get_plasmid(auth.user_id, plasmid_id)
if p is None:
return jsonify({"ok": False, "error": "Plasmid not found."}), 404
return jsonify({"ok": True, "plasmid": p})
@app.delete("/api/plasmid/library/<plasmid_id>")
def plasmid_library_delete(plasmid_id: str) -> Response:
auth, denied = _plasmid_signin_guard()
if denied:
return denied
return jsonify(_auth.delete_plasmid(auth.user_id, plasmid_id))
@app.post("/api/plasmid/clone")
def plasmid_clone() -> Response:
"""Cloning simulation: Gibson / Golden Gate / restriction. Returns the
assembled construct in the standard plasmid-ish shape."""
from .core import cloning as _CL
_auth_ctx, denied = _plasmid_signin_guard()
if denied:
return denied
body = request.get_json(force=True, silent=True) or {}
method = (body.get("method") or "gibson").lower()
try:
if method == "gibson":
frags = body.get("fragments")
if not isinstance(frags, list):
return jsonify({"error": "Provide a list of fragments."}), 400
res = _CL.gibson(frags, circular=bool(body.get("circular", True)))
elif method == "golden_gate":
frags = body.get("fragments")
if not isinstance(frags, list):
return jsonify({"error": "Provide a list of parts."}), 400
res = _CL.golden_gate(frags, enzyme=str(body.get("enzyme", "BsaI"))[:12])
elif method == "restriction":
res = _CL.restriction_clone(
str(body.get("vector", "")), str(body.get("insert", "")),
str(body.get("enzyme", ""))[:24],
vector_topology="linear" if body.get("vector_topology") == "linear" else "circular")
elif method in ("yeast_hr", "yeast"):
frags = body.get("fragments")
if not isinstance(frags, list):
return jsonify({"error": "Provide a list of fragments (the linearised vector is one of them)."}), 400
res = _CL.yeast_assembly(
frags, circular=bool(body.get("circular", True)),
marker=(str(body.get("marker")).strip()[:40] if body.get("marker") else None))
elif method in ("yeast_integration", "integration"):
res = _CL.yeast_integration(
str(body.get("upstream", "")), str(body.get("downstream", "")),
str(body.get("insert", "") or body.get("cassette", "")),
arm=int(body.get("arm") or 40),
marker=(str(body.get("marker")).strip()[:40] if body.get("marker") else None),
site_name=str(body.get("site") or "locus")[:40])
else:
return jsonify({"error": f"Unknown method: {method}."}), 400
except Exception as exc: # noqa: BLE001
logger.exception("clone failed")
return jsonify({"error": f"Assembly failed: {exc}"}), 500
if not res.get("ok"):
return jsonify(res), 200 # business-level failure (bad overlaps etc.)
return jsonify(res)
# ================================================================= CRISPR
# Cas9 gRNA design for knockout applications. MVP scope: pasted DNA
# → ranked SpCas9 guides with Doench-style heuristic on-target
# scoring. Sign-in gated (free for any account); anonymous users
# get a 403 with a sign-in CTA so the frontend can route them to
# /signin instead of the engine's trial gate. Whole-genome off-target,
# Cas12a, base editors, and HDR donor design are deferred — see
# dee/core/crispr.py module docstring for the full scope ladder.
@app.post("/api/crispr/design")
def crispr_design() -> Response:
# Sign-in gate. Distinct from the trial timer (which is per-anon-
# session): CRISPR requires an account, no anonymous use. The
# 403 carries kind="signin_required" so static auth.js can route
# the parent window to /signin instead of showing the trial modal.
auth = _auth.get_auth()
if auth.anonymous:
return jsonify({
"error": (
"CRISPR design requires a free account. "
"Sign in or create one to keep going."
),
"kind": "signin_required",
"signup_url": "https://turingdna.com/signin/?from=crispr",
}), 403
body = request.get_json(force=True, silent=True) or {}
seq = (body.get("sequence") or "").strip()
if not seq:
return jsonify({"error": "missing 'sequence'"}), 400
# Bound the input so a 10 MB paste doesn't tie up the worker.
# 1 Mbp covers any reasonable cloning-scale region (largest
# human genes are ~2.4 Mbp, but at that scale you'd use a
# whole-genome tool, not a paste box).
if len(seq) > 1_000_000:
return jsonify({
"error": (
"Sequence too long. Cap is 1 Mbp — paste just the "
"gene / region you're editing, not a whole "
"chromosome."
),
"kind": "input_too_large",
}), 400
try:
max_results = int(body.get("max_results", 50))
except (TypeError, ValueError):
max_results = 50
max_results = max(1, min(500, max_results)) # bound
try:
min_score = float(body.get("min_score", 0.0))
except (TypeError, ValueError):
min_score = 0.0
min_score = max(0.0, min(1.0, min_score))
# Enzyme picker — Phase 1 supports SpCas9 and Cas12a. Default cas9
# for backward compat with the existing frontend (which sends no
# `enzyme` field on the current build).
enzyme = str(body.get("enzyme", "cas9")).lower()
if enzyme not in ("cas9", "cas12a"):
enzyme = "cas9"
# Phase 2B-1: optional genome off-target + Ensembl exon context.
# Both are opt-in; default empty means "skip" so existing
# frontends keep working unchanged.
target_organism = str(body.get("target_organism", "")).lower().strip()
gene_symbol = str(body.get("gene_symbol", "")).strip()
# Light input validation — anything else gets ignored.
# Validate against the actual genome registry rather than a hardcoded
# list, so adding an organism to offtarget.GENOME_SOURCES enables it
# end to end. The old literal tuple silently rejected any new organism
# and fell back to "no genome search".
from dee.core.offtarget import GENOME_SOURCES as _GENOMES
if target_organism and target_organism not in _GENOMES:
target_organism = ""
if len(gene_symbol) > 32 or not re.match(r"^[A-Za-z0-9._-]*$", gene_symbol):
gene_symbol = ""
# Phase 2C-1: optional cloning vector for per-guide oligo gen.
from dee.core import crispr_cloning as _cc
vector_id = str(body.get("vector_id", "")).lower().strip()
if vector_id and vector_id not in _cc.CRISPR_VECTORS:
vector_id = ""
# Phase 3 (M1): base-editing mode. Defaults to knockout so the
# existing frontend (which sends no `mode`) is unaffected.
from dee.core import base_editor as _be
mode = str(body.get("mode", "knockout")).lower()
if mode not in ("knockout", "base_edit"):
mode = "knockout"
base_editor = str(body.get("base_editor", "")).lower().strip()
if base_editor and _be.get_base_editor(base_editor) is None:
base_editor = ""
try:
guides = _find_crispr_guides(
seq, enzyme=enzyme, max_results=max_results, min_score=min_score,
target_organism=target_organism, gene_symbol=gene_symbol,
vector_id=vector_id, mode=mode, base_editor=base_editor,
)
except ValueError as exc:
return jsonify({"error": str(exc), "kind": "validation"}), 400
except Exception as exc: # noqa: BLE001
logger.exception("CRISPR design failed.")
return jsonify({
"error": f"{type(exc).__name__}: {exc}",
"kind": "internal",
}), 500
# Flatten dataclasses into plain dicts for JSON. New Phase 1
# fields (composite_score, cfd_max_offtarget, offtarget_count,
# ko_efficacy, ko_reasoning, enzyme) get exposed so the frontend
# table can render them.
#
# Phase 2D: genome_index_status surfaces whether the off-target
# search actually ran. "ready" = index hit, hits are real.
# "building" = first-time index build is in flight, guides
# returned without genome data; the frontend shows a banner
# telling the user to refresh in 2 min. "n/a" = no organism
# requested.
from dee.core import offtarget as _ot_status
genome_index_status = _ot_status.index_status(target_organism)
# Only the top-ranked guides get the (~3.5 s each) genome query, so
# say exactly how many were screened rather than letting the UI imply
# every guide was checked.
from dee.core.crispr import GENOME_OFFTARGET_TOP_N as _GENOME_TOP_N
genome_searched_top_n = (
min(_GENOME_TOP_N, len(guides))
if (genome_index_status == "ready" and enzyme == "cas9") else 0
)
from dee.core import outcomes as _O
# One serializer, shared with the agent's design_crispr_guides tool
# (dee/core/crispr.guide_to_dict) — both feed the same results table.
from dee.core.crispr import guide_to_dict as _guide_to_dict
guide_dicts = [_guide_to_dict(g) for g in guides]
# Cross-user CRISPR on-target prior → fold into scores + re-rank. No-op
# until public.outcome_priors has a 'crispr' aggregate (post effective date).
guide_dicts = _O.calibrate_crispr_guides(guide_dicts, _load_tool_prior("crispr"))
return jsonify({
"input_length": len(seq.strip()),
"n_guides": len(guide_dicts),
"enzyme": enzyme,
"mode": mode,
"base_editor": base_editor or (_be.DEFAULT_BASE_EDITOR if mode == "base_edit" else ""),
"genome_organism": target_organism,
"genome_index_status": genome_index_status,
"genome_searched_top_n": genome_searched_top_n,
"calibrated": any("calibration" in g for g in guide_dicts),
"guides": guide_dicts,
})
@app.get("/api/crispr/methods")
def crispr_methods() -> Response:
"""How every scored column in the CRISPR table was actually computed —
formula, published basis, and where it stops being trustworthy. PUBLIC:
the whole point is that a reviewer can audit the numbers without asking
us. Mirrors dee/core/crispr.py exactly (see crispr_methods.py)."""
from dee.core import crispr_methods as _cm
payload = _cm.methods_payload()
payload["ok"] = True
return jsonify(payload)
@app.get("/api/crispr/vectors")
def crispr_vectors() -> Response:
"""List the curated CRISPR expression vectors the engine supports.
Used by the frontend to populate the vector dropdown above the
results table. Filterable by ?enzyme=cas9|cas12a."""
from dee.core import crispr_cloning as _cc
enzyme = request.args.get("enzyme", "").strip().lower()
if enzyme not in ("", "cas9", "cas12a"):
enzyme = ""
vectors = _cc.list_vectors(enzyme)
return jsonify({
"default": _cc.default_vector_for(enzyme or "cas9"),
"vendors": _cc.VENDORS,
"vectors": [
{
"id": v.id,
"name": v.name,
"addgene_id": v.addgene_id,
"enzyme": v.enzyme,
"nuclease": v.nuclease,
"selection": v.selection,
"description": v.description,
}
for v in vectors
],
})
@app.get("/api/crispr/base-editors")
def crispr_base_editors() -> Response:
"""List the base editors the engine supports, for the base-edit
mode dropdown. Filterable by ?kind=CBE|ABE."""
from dee.core import base_editor as _be
kind = request.args.get("kind", "").strip().upper()
if kind not in ("", "CBE", "ABE"):
kind = ""
editors = _be.list_base_editors(kind)
return jsonify({
"default": _be.DEFAULT_BASE_EDITOR,
"editors": [
{
"id": e.id,
"name": e.name,
"kind": e.kind,
"target_base": e.target_base,
"result_base": e.result_base,
"window": list(e.window),
"citation": e.citation,
"note": e.note,
}
for e in editors
],
})
@app.post("/api/crispr/structure")
def crispr_structure() -> Response:
"""Resolve a gene → UniProt + AlphaFold model URL for the structure
viewer (Phase 3, M5). Body: {organism, gene_symbol}. Sign-in gated.
Only (organism, gene_symbol) leaves the Space."""
auth = _auth.get_auth()
if auth.anonymous:
return jsonify({"ok": False, "error": "signin_required",
"kind": "signin_required"}), 403
body = request.get_json(force=True, silent=True) or {}
organism = str(body.get("organism", "")).lower().strip()
gene_symbol = str(body.get("gene_symbol", "")).strip()
# Any organism: AlphaFold DB covers all of UniProt. The old
# human/mouse gate is why SpCas9 — which has had a model the whole
# time — came back as "no structure".
if not organism or len(organism) > 64 or not re.match(r"^[A-Za-z0-9 .'-]+$", organism):
return jsonify({"ok": False,
"error": "Provide an organism, e.g. human or S. pyogenes."}), 422
if not gene_symbol or len(gene_symbol) > 32 or not re.match(r"^[A-Za-z0-9._-]+$", gene_symbol):
return jsonify({"ok": False, "error": "Provide a valid gene symbol."}), 422
from dee.core import resolve as _resolve
try:
result = _resolve.resolve_uniprot(organism, gene_symbol)
except Exception as exc: # noqa: BLE001
logger.exception("CRISPR structure resolve failed.")
return jsonify({"ok": False, "error": f"{type(exc).__name__}: {exc}"}), 500
return jsonify(result), (200 if result.get("ok") else 422)
@app.post("/api/structure")
def public_structure() -> Response:
"""PUBLIC gene → AlphaFold model resolve. Body: {organism, gene_symbol}.
Deliberately NOT sign-in gated, unlike /api/crispr/structure. Both call
the same resolver, but this one exists because gating it was producing a
FALSEHOOD: the Director folds a protein as part of a workflow, and for an
anonymous visitor every gene outside a small hardcoded list came back 403
→ the UI narrated "no public structure resolved for PTEN", which is
simply untrue (PTEN is P60484 in AlphaFold DB). Telling a scientist a
structure doesn't exist when it does is a credibility failure, and it
made the product un-evaluatable without an account.
Safe to open: the ONLY thing that leaves the Space is (organism,
gene_symbol) — both public identifiers, never a user sequence — and the
answer is a URL into the public AlphaFold DB. Rate-limited like every
other outbound-lookup route.
"""
body = request.get_json(force=True, silent=True) or {}
organism = str(body.get("organism", "")).lower().strip()
gene_symbol = str(body.get("gene_symbol", "")).strip()
if not organism or len(organism) > 64 or not re.match(r"^[A-Za-z0-9 .'-]+$", organism):
return jsonify({"ok": False, "kind": "bad_organism",
"error": "Provide an organism, e.g. human or S. pyogenes."}), 422
if not gene_symbol or len(gene_symbol) > 32 or not re.match(r"^[A-Za-z0-9._-]+$", gene_symbol):
return jsonify({"ok": False, "kind": "bad_symbol",
"error": "Provide a valid gene symbol."}), 422
from dee.core import resolve as _resolve
try:
result = _resolve.resolve_uniprot(organism, gene_symbol)
except Exception: # noqa: BLE001 — never leak internals
logger.exception("public structure resolve failed")
return jsonify({"ok": False, "kind": "lookup_failed",
"error": "Structure lookup is unavailable right now."}), 503
# `not ok` here means the LOOKUP genuinely found nothing — the caller may
# honestly say so. An auth/transport failure never reaches this branch.
if not result.get("ok"):
result.setdefault("kind", "not_found")
return jsonify(result), (200 if result.get("ok") else 404)
@app.post("/api/crispr/resolve")
def crispr_resolve() -> Response:
"""Paste-anything resolver (Phase 3, M2). Body: {text, organism}.
Resolves a gene symbol / accession / raw sequence to an editable
DNA sequence + a human label. Same sign-in gate as /design.
Privacy: only (organism, identifier) leaves the Space for lookups;
a raw sequence makes no outbound call."""
auth = _auth.get_auth()
if auth.anonymous:
return jsonify({
"ok": False,
"error": "Sign in or create a free account to keep going.",
"kind": "signin_required",
"signup_url": "https://turingdna.com/signin/?from=crispr",
}), 403
body = request.get_json(force=True, silent=True) or {}
text = (body.get("text") or "").strip()
if not text:
return jsonify({"ok": False, "error": "missing 'text'"}), 400
# Bound input: a sequence paste can be up to 1 Mbp; an identifier is
# tiny. Cap defensively so a giant blob can't tie up the resolver.
if len(text) > 1_000_000:
return jsonify({"ok": False, "error": "Input too long (1 Mbp cap)."}), 400
# Bound but do NOT allow-list: blanking an unrecognized organism made
# the resolver ask "which species?" right after the user had said.
organism = str(body.get("organism", "")).lower().strip()
if len(organism) > 64 or not re.match(r"^[A-Za-z0-9 .'-]*$", organism):
organism = ""
from dee.core import resolve as _resolve
try:
result = _resolve.resolve_target(text, organism=organism)
except Exception as exc: # noqa: BLE001
logger.exception("CRISPR resolve failed.")
return jsonify({"ok": False, "error": f"{type(exc).__name__}: {exc}"}), 500
return jsonify(result), (200 if result.get("ok") else 422)
@app.post("/api/crispr/ip_radar")
def crispr_ip_radar() -> Response:
"""Live Collision IP Radar (v1, literature via Europe PMC). Opt-in,
metadata-first: only the gene symbol, validated point-mutation set, and
the 20-mer guide spacer are accepted — never a full novel sequence.
Sign-in gated. Non-critical: any failure returns ok:False with 200 so
the UI degrades quietly instead of erroring."""
auth = _auth.get_auth()
if auth.anonymous:
return jsonify({"ok": False, "error": "signin_required",
"kind": "signin_required"}), 403
body = request.get_json(force=True, silent=True) or {}
gene_symbol = str(body.get("gene_symbol", "")).strip()
if len(gene_symbol) > 32 or not re.match(r"^[A-Za-z0-9._-]*$", gene_symbol):
gene_symbol = ""
spacer = str(body.get("spacer", "")).strip().upper()[:25]
if not re.match(r"^[ACGT]*$", spacer):
spacer = ""
raw_muts = body.get("mutations")
mutations = []
if isinstance(raw_muts, list):
for m in raw_muts[:8]:
m = str(m).strip()
# Only accept HGVS-ish point mutations (e.g. T47A) — refuse
# arbitrary strings so the field can't smuggle a raw sequence.
if re.match(r"^[A-Za-z]\d{1,5}[A-Za-z*]$", m):
mutations.append(m)
mode = str(body.get("mode", "knockout")).lower()
from dee.core import ip_radar as _ipr
try:
result = _ipr.check_literature(
gene_symbol=gene_symbol, mutations=mutations,
spacer=spacer, mode=mode,
)
# Patent layer (free PatentsView, gene-level "related patents").
# Off unless PATENTSVIEW_API_KEY is set; never blocks literature.
try:
pat = _ipr.check_patents(gene_symbol=gene_symbol)
result["patents"] = pat.get("patents", [])
result["patents_available"] = pat.get("available", False)
except Exception: # noqa: BLE001
result["patents"] = []
result["patents_available"] = False
except Exception: # noqa: BLE001
logger.exception("IP radar failed.")
return jsonify({"ok": False, "error": "radar_unavailable"}), 200
return jsonify(result)
# ─── Phase 3 (M4): save & revisit designs ─────────────────────────
@app.post("/api/crispr/save")
def crispr_save() -> Response:
"""Save the current guide set to the signed-in user's library."""
auth = _auth.get_auth()
if auth.anonymous:
return jsonify({"ok": False, "error": "signin_required",
"kind": "signin_required",
"signup_url": "https://turingdna.com/signin/?from=crispr"}), 403
body = request.get_json(force=True, silent=True) or {}
guides = body.get("guides")
if not isinstance(guides, list) or not guides:
return jsonify({"ok": False, "error": "Nothing to save — run a design first."}), 400
mode = str(body.get("mode", "knockout")).lower()
if mode not in ("knockout", "base_edit"):
mode = "knockout"
try:
input_length = int(body.get("input_length") or 0)
except (TypeError, ValueError):
input_length = 0
result = _auth.save_crispr_design(
auth.user_id,
label=str(body.get("label", ""))[:120],
mode=mode,
enzyme=str(body.get("enzyme", ""))[:16] or None,
base_editor=str(body.get("base_editor", ""))[:32] or None,
organism=str(body.get("organism", ""))[:16] or None,
gene_symbol=str(body.get("gene_symbol", ""))[:32] or None,
input_length=input_length,
guides=guides,
)
_attribute(auth, body, "crispr", result)
_auth.cleanup_expired_crispr_designs_async(auth.user_id) # lazy sweep
return jsonify(result), (200 if result.get("ok") else 502)
@app.get("/api/crispr/designs")
def crispr_designs_list() -> Response:
"""The signed-in user's saved designs (metadata only)."""
auth = _auth.get_auth()
if auth.anonymous:
return jsonify({"ok": False, "error": "signin_required",
"kind": "signin_required"}), 403
return jsonify({"ok": True, "designs": _auth.list_crispr_designs(auth.user_id)})
@app.get("/api/crispr/designs/<design_id>")
def crispr_design_get(design_id: str) -> Response:
"""One saved design incl. guides (owner-only, server-enforced)."""
auth = _auth.get_auth()
if auth.anonymous:
return jsonify({"ok": False, "error": "signin_required",
"kind": "signin_required"}), 403
design = _auth.get_crispr_design(auth.user_id, design_id)
if design is None:
return jsonify({"ok": False, "error": "Design not found."}), 404
return jsonify({"ok": True, "design": design})
# ── Primer Analysis: saved sets (unified "My designs" hub) ──
# Static routes win over /api/primers/<job_id> in Werkzeug, so no clash.
@app.post("/api/primers/save")
def primers_save() -> Response:
"""Save the current primer analysis to the signed-in user's library."""
auth = _auth.get_auth()
if auth.anonymous:
return jsonify({"ok": False, "error": "signin_required",
"kind": "signin_required",
"signup_url": "https://turingdna.com/signin/?from=primers"}), 403
body = request.get_json(force=True, silent=True) or {}
result_obj = body.get("result")
if not isinstance(result_obj, dict) or not result_obj.get("primers"):
return jsonify({"ok": False, "error": "Nothing to save — analyse some primers first."}), 400
try:
template_len = int(body.get("template_len") or 0)
except (TypeError, ValueError):
template_len = 0
try:
n_primers = int(body.get("n_primers") or len(result_obj.get("primers") or []))
except (TypeError, ValueError):
n_primers = len(result_obj.get("primers") or [])
result = _auth.save_primer_analysis(
auth.user_id,
label=str(body.get("label", ""))[:120],
organism=str(body.get("organism", ""))[:32] or None,
template_len=template_len,
n_primers=n_primers,
result=result_obj,
)
_attribute(auth, body, "primer", result)
_auth.cleanup_expired_primer_analyses_async(auth.user_id) # lazy sweep
return jsonify(result), (200 if result.get("ok") else 502)
@app.get("/api/primers/analyses")
def primers_analyses_list() -> Response:
"""The signed-in user's saved primer sets (metadata only)."""
auth = _auth.get_auth()
if auth.anonymous:
return jsonify({"ok": False, "error": "signin_required",
"kind": "signin_required"}), 403
return jsonify({"ok": True, "analyses": _auth.list_primer_analyses(auth.user_id)})
@app.get("/api/primers/analyses/<analysis_id>")
def primers_analysis_get(analysis_id: str) -> Response:
"""One saved primer set incl. result (owner-only, server-enforced)."""
auth = _auth.get_auth()
if auth.anonymous:
return jsonify({"ok": False, "error": "signin_required",
"kind": "signin_required"}), 403
analysis = _auth.get_primer_analysis(auth.user_id, analysis_id)
if analysis is None:
return jsonify({"ok": False, "error": "Primer set not found."}), 404
return jsonify({"ok": True, "analysis": analysis})
@app.delete("/api/primers/analyses/<analysis_id>")
def primers_analysis_delete(analysis_id: str) -> Response:
"""Delete one saved primer set (owner-only)."""
auth = _auth.get_auth()
if auth.anonymous:
return jsonify({"ok": False, "error": "signin_required",
"kind": "signin_required"}), 403
return jsonify(_auth.delete_primer_analysis(auth.user_id, analysis_id))
@app.post("/api/primers/export")
def primers_export() -> Response:
"""CSV / XLSX of a primer analysis the client already holds — so
researchers can take the ranked table to the bench. The client posts
the analyze result back (no re-run); we flatten + stream it. Sign-in
gated like the rest of Primer Analysis."""
auth = _auth.get_auth()
if auth.anonymous:
return jsonify({"error": "signin_required", "kind": "signin_required"}), 403
body = request.get_json(force=True, silent=True) or {}
result = body.get("result")
primers = result.get("primers") if isinstance(result, dict) else None
if not isinstance(primers, list) or not primers:
return jsonify({"error": "Nothing to export — analyse some primers first."}), 400
primers = primers[:1000] # bound the export size
fmt = str(body.get("format", "xlsx")).lower().strip()
if fmt not in ("csv", "xlsx"):
fmt = "xlsx"
header = ["Rank", "Name", "Sequence", "Length (nt)", "GC %", "Tm (C)",
"3' G/C clamp", "Sites on template", "Off-target products", "Fitness (0-1)"]
def _row(i, p):
if not isinstance(p, dict):
p = {}
ot = p.get("on_template") if isinstance(p.get("on_template"), dict) else {}
sp = p.get("specificity") if isinstance(p.get("specificity"), dict) else {}
offt = sp.get("offtarget_products")
if offt is None:
offt = sp.get("strong_offtargets")
fit = p.get("fitness")
return [
i + 1,
str(p.get("name", "") or ""),
str(p.get("seq", "") or ""),
p.get("length", ""),
p.get("gc", ""),
p.get("tm", ""),
"yes" if p.get("gc_clamp") else "no",
ot.get("n_strong_sites", "") if ot else "",
offt if offt is not None else "",
round(fit, 3) if isinstance(fit, (int, float)) else "",
]
rows = [header] + [_row(i, p) for i, p in enumerate(primers)]
if fmt == "csv":
import csv as _csv
from io import StringIO as _SIO
from flask import make_response as _mr
sio = _SIO()
w = _csv.writer(sio, quoting=_csv.QUOTE_MINIMAL, lineterminator="\r\n")
for r in rows:
w.writerow(r)
data = ("" + sio.getvalue()).encode("utf-8") # BOM → Excel detects UTF-8
resp = _mr(data)
resp.headers["Content-Type"] = "text/csv; charset=utf-8"
resp.headers["Content-Disposition"] = 'attachment; filename="turingdna_primers.csv"'
return resp
from io import BytesIO as _BIO
from openpyxl import Workbook as _WB
from openpyxl.styles import Font as _Font
wb = _WB()
ws = wb.active
ws.title = "Primers"
for r in rows:
ws.append(r)
for cell in ws[1]:
cell.font = _Font(bold=True)
buf = _BIO()
wb.save(buf)
buf.seek(0)
return send_file(
buf,
mimetype="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
as_attachment=True,
download_name="turingdna_primers.xlsx",
)
# ── Directed-Evolution libraries: per-user list + signed CSV download ──
# Per-user, RLS-safe replacement for the shared-filesystem /api/library
# removed in audit C2 — reads public.libraries filtered by the JWT user.
@app.get("/api/library")
def library_list() -> Response:
"""The signed-in user's saved DE libraries, each with its loop state.
n_outcomes is what makes an OPEN LOOP visible. A library with none is
an experiment still running — the numbers have not come back from the
bench — but it rendered identically to one that was finished and done
with, which is a large part of why nobody ever returned to close one.
One extra query for the whole list, not one per library: see
de_outcome_counts_by_library.
"""
auth = _auth.get_auth()
if auth.anonymous:
return jsonify({"ok": False, "error": "signin_required",
"kind": "signin_required"}), 403
libs = _auth.list_libraries(auth.user_id)
tested = _auth.de_outcome_counts_by_library(auth.user_id)
for lib in libs:
lib["n_outcomes"] = int(tested.get(lib.get("id"), 0))
return jsonify({"ok": True, "libraries": libs})
@app.get("/api/library/<library_id>")
def library_get(library_id: str) -> Response:
"""One saved library, in the shape a fresh run result has.
This is the door back into round 2. Everything the learning loop needs
already existed on the server — the library_id, the wild type it was
designed against, /api/de/outcomes/<id> to read back what was already
logged — but a library could only be DOWNLOADED, so the round-2 panel
was reachable only in the session that created the library. That is
precisely the session in which nobody can have wet-lab results yet.
Returns the same keys renderResults() consumes so a reopened library
renders as itself rather than as a summary of itself.
"""
auth = _auth.get_auth()
if auth.anonymous:
return jsonify({"ok": False, "error": "signin_required",
"kind": "signin_required"}), 403
lib = _auth.load_library(auth.user_id, library_id)
if not lib:
return jsonify({"ok": False, "error": "Library not found."}), 404
return jsonify({"ok": True, "reopened": True, **lib})
@app.get("/api/library/<library_id>/download")
def library_download(library_id: str) -> Response:
"""Return a short-lived signed URL for the library CSV (owner-only).
The browser fetches the CSV straight from Storage; the service key
never leaves the server."""
auth = _auth.get_auth()
if auth.anonymous:
return jsonify({"ok": False, "error": "signin_required",
"kind": "signin_required"}), 403
url = _auth.library_csv_url(auth.user_id, library_id)
if not url:
return jsonify({"ok": False, "error": "Library not found."}), 404
return jsonify({"ok": True, "url": url})
# ── Assays: what a measured number means ──────────────────────────────
# An assay is defined ONCE and reused by every outcome it produced.
# Retyping the readout and the wild-type per measurement is how a lab ends
# up with six near-identical assays and no way to pool any of them.
@app.get("/api/assays")
def assays_list() -> Response:
auth = _auth.get_auth()
if auth.anonymous:
return jsonify({"ok": True, "gated": True, "assays": []})
# Each row carries its own poolability verdict, computed by the same
# `check_assay` the create path uses. The client must never re-derive
# this: a second implementation in JavaScript is a second set of rules
# about what counts as comparable evidence, and the two would drift.
rows = _auth.list_assays(auth.user_id)
for row in rows:
probe = _assay.check_assay(_assay.Assay.from_row(row))
row["poolable"] = probe.comparable
row["poolable_why"] = probe.why
return jsonify({"ok": True, "gated": False, "assays": rows})
@app.post("/api/assays")
def assays_create() -> Response:
auth = _auth.get_auth()
if auth.anonymous:
return jsonify({"error": "Sign in to record assays.",
"kind": "signin_required"}), 401
body = request.get_json(force=True, silent=True) or {}
name = str(body.get("name") or "").strip()
if not name:
return jsonify({"error": "An assay needs a name."}), 400
res = _auth.save_assay(
auth.user_id, name=name,
readout=str(body.get("readout") or ""),
unit=str(body.get("unit") or ""),
scale=str(body.get("scale") or "ratio"),
direction=str(body.get("direction") or "higher_is_better"),
wt_value=body.get("wt_value"), wt_stdev=body.get("wt_stdev"),
wt_n=body.get("wt_n"),
conditions=body.get("conditions") if isinstance(
body.get("conditions"), dict) else {},
notes=str(body.get("notes") or ""))
if not res.get("ok"):
return jsonify({"error": res.get("error") or "Couldn't save."}), 400
# Tell the caller straight away whether results against this assay will
# be poolable, rather than letting them discover it after 40
# measurements. The missing field is almost always the wild-type, and
# it is far cheaper to add before the plate is read than after.
probe = _assay.check_assay(_assay.Assay.from_row(res.get("assay")))
return jsonify({"ok": True, "assay": res.get("assay"),
"poolable": probe.comparable,
"poolable_why": probe.why})
# ── DE learning flywheel: log outcomes + active-learning round 2 ──
@app.post("/api/de/outcomes")
def de_outcomes_save() -> Response:
"""Save this user's measured variants for a saved DE library."""
auth = _auth.get_auth()
if auth.anonymous:
return jsonify({"ok": False, "error": "signin_required",
"kind": "signin_required"}), 403
body = request.get_json(force=True, silent=True) or {}
library_id = str(body.get("library_id", "")).strip()
outcomes = body.get("outcomes")
if not isinstance(outcomes, list):
return jsonify({"ok": False, "error": "No outcomes provided."}), 400
# An assay id is only written after it is confirmed to belong to this
# user. Forwarding an unverified one would attach another lab's
# readout, unit and wild-type to these numbers — worse than no assay
# at all, because the result would then look poolable.
assay_id = str(body.get("assay_id") or "").strip()
assay_row = _auth.get_assay(auth.user_id, assay_id) if assay_id else None
if assay_id and not assay_row:
return jsonify({"ok": False, "error": "That assay wasn't found."}), 404
result = _auth.save_de_outcomes(auth.user_id, library_id, outcomes,
assay_id=(assay_row or {}).get("id"))
# THE EDGE THAT MATTERS MOST. Everything else in the graph is a
# prediction pointing at another prediction; this is the one arrow that
# touches reality — a library and what the bench actually measured. It
# is also the edge the whole outcome loop is built to accumulate, so it
# is recorded the moment results arrive rather than reconstructed.
#
# The results node id is derived from the library, not generated: one
# results node per library, so logging a second batch adds observations
# to the same node instead of littering the graph with orphans.
if result.get("ok") and library_id:
try:
_auth.record_edge(
auth.user_id,
from_kind="library", from_id=library_id,
to_kind="results", to_id="results:" + library_id,
relation="measured", origin="handoff",
construct_id=str(body.get("construct_id") or "") or None,
detail={"n_outcomes": len(outcomes)})
except Exception: # noqa: BLE001
app.logger.warning("lineage: outcome edge failed", exc_info=True)
# Say what was actually banked. "Saved" alone lets a lab log forty
# measurements and only discover months later — when the commons is
# built — that none of them could be pooled. The moment of saving is
# the last point at which the fix (record your wild-type) is cheap.
if result.get("ok"):
a = _assay.Assay.from_row(assay_row)
result["pooling"] = _assay.summarise_pooling(
[_assay.normalise(r.get("measured_value"), a)
for r in outcomes if isinstance(r, dict)])
_auth.cleanup_expired_de_outcomes_async(auth.user_id) # lazy sweep
return jsonify(result), (200 if result.get("ok") else 502)
@app.get("/api/de/outcomes/<library_id>")
def de_outcomes_list(library_id: str) -> Response:
auth = _auth.get_auth()
if auth.anonymous:
return jsonify({"ok": False, "error": "signin_required",
"kind": "signin_required"}), 403
return jsonify({"ok": True, "outcomes": _auth.list_de_outcomes(auth.user_id, library_id)})
# ── Did round 2 actually beat round 1? ────────────────────────────────
# The product's central claim, measured against this user's own recorded
# loops rather than asserted in copy. See dee/core/uplift.py — it refuses
# far more often than it measures, and refusing is the feature.
@app.get("/api/de/uplift")
def de_uplift() -> Response:
auth = _auth.get_auth()
if auth.anonymous:
# Not an error: a signed-out visitor simply has no loops. Returning
# a zero-uplift shape here would be a claim, and a false one.
return jsonify({"ok": True, "gated": True,
"summary": _uplift.summarise([]), "pairs": []})
pairs = _uplift.pair_rounds(_auth.list_lineage(auth.user_id))
if not pairs:
return jsonify({"ok": True, "gated": False,
"summary": _uplift.summarise([]), "pairs": []})
# One read per library involved, de-duplicated: a library that is
# round 2 of one pair is often round 1 of the next.
wanted = {lid for p in pairs for lid in (p.round1, p.round2)}
outcomes = {lid: _auth.list_de_outcomes(auth.user_id, lid) for lid in wanted}
assays = {str(a.get("id")): a for a in _auth.list_assays(auth.user_id)}
results = [_uplift.measure_pair(p, outcomes.get(p.round1, []),
outcomes.get(p.round2, []), assays)
for p in pairs]
return jsonify({"ok": True, "gated": False,
"summary": _uplift.summarise(results),
"pairs": [vars(r) for r in results]})
# ── "I already have data" ─────────────────────────────────────────────
# The Learn phase used to be gated behind a wet-lab cycle: design a
# library, wait three months, come back and type numbers in. Almost nobody
# completes that, so the flywheel could not start and the engine's central
# claim went untested by anyone but us.
#
# Most protein engineers arrive WITH data. /api/de/round2 never actually
# required a prior library — it takes a wild-type and a list of
# measurements — so the only thing between that spreadsheet and a
# same-session result was a way to read it.
#
# Deliberately open to anonymous visitors, like the other design tools: a
# scientist deciding whether to trust this should be able to check it
# against results they already trust, before making an account.
@app.post("/api/de/match")
def de_match() -> Response:
"""Read a results file against ONE library's variants.
After three weeks at the bench a scientist has a plate-reader export,
not sixty numbers to retype into a form. This takes the file and fills
the round-2 panel from it.
The parsing is dee/core/measurements.py — the same module the "I
already have data" door uses, deliberately. It is careful about the
two things that quietly ruin this: silent loss (every rejected row
comes back with its own reason) and residue numbering (a table written
against the mature protein scores as garbage against a precursor
wild-type, and "your data does not correlate" is the most damaging
wrong answer this product can give). Re-implementing that in the
browser would duplicate the careful half and drift from it.
Three outcomes are reported separately, because they need different
things from the user: rows we could not read, rows whose numbering
does not fit the wild-type, and rows that read fine but name a variant
this library does not contain.
"""
auth = _auth.get_auth()
if auth.anonymous:
return jsonify({"ok": False, "error": "signin_required",
"kind": "signin_required"}), 403
body = request.get_json(force=True, silent=True) or {}
lib = _auth.load_library(auth.user_id, str(body.get("library_id") or ""))
if not lib:
return jsonify({"ok": False, "error": "Library not found."}), 404
wt = re.sub(r"\s+", "", str(lib.get("wt_protein") or "")).upper()
if not wt:
return jsonify({"ok": False, "error":
"This library has no stored wild-type, so there is nothing to "
"check the mutation numbering against."}), 400
# A FILE is not a pasted two-column table. parse_table takes the value
# from the last field and the mutation from everything before it,
# which is right for a paste and wrong for a plate-reader export whose
# first column is a well id. reduce_columns finds the two columns that
# matter and says which it chose; it declines rather than guesses, and
# a two-column file falls straight through to the original path.
text = str(body.get("text") or "")
reduced, columns_note = _meas.reduce_columns(text)
parsed = _meas.parse_table(reduced if reduced is not None else text)
rejects = [vars(j) for j in parsed.rejects]
if not parsed.rows:
return jsonify({"ok": True, "matched": [], "unmatched": [],
"rejects": rejects, "offset_hint": None,
"columns": columns_note, "n_parsed": 0})
rows = parsed.rows
try:
offset = int(body.get("offset") or 0)
except (TypeError, ValueError):
offset = 0
if offset:
rows = _meas.apply_offset(rows, offset)
# Never renumber someone's data unasked. Report the shift and let them
# accept it; off-by-one on residue numbering is the classic way to
# lose a month, and it is free to catch here.
hint = None if offset else _meas.detect_offset(rows, wt)
if hint:
return jsonify({"ok": True, "matched": [], "unmatched": [],
"rejects": rejects, "offset_hint": hint,
"columns": columns_note,
"n_parsed": len(parsed.rows)})
kept, wt_bad = _meas.check_against_wt(rows, wt)
rejects += [vars(j) for j in wt_bad]
# A12V,L7M and L7M,A12V are the same variant. Matching on the joined
# string would drop the row and call it "not in this library", which
# is the silent loss this whole module exists to prevent.
def _key(label):
return ",".join(sorted(p.strip().upper()
for p in str(label or "").split(",") if p.strip()))
by_key = {}
for v in (lib.get("variants") or []):
label = v.get("Mutations_AA")
if label:
by_key.setdefault(_key(label), label)
matched, unmatched = [], []
for r in kept:
hit = by_key.get(_key(r.mutations))
if hit:
# Return the LIBRARY's spelling: the panel's inputs are keyed
# by it, so the user's ordering must not reach the DOM lookup.
matched.append({"mutations": hit, "measured_value": r.measured_value})
else:
unmatched.append({"mutations": r.mutations,
"measured_value": r.measured_value, "line": r.line})
return jsonify({"ok": True, "matched": matched, "unmatched": unmatched,
"rejects": rejects, "offset_hint": None,
"columns": columns_note,
"n_parsed": len(parsed.rows)})
@app.post("/api/de/import")
def de_import() -> Response:
body = request.get_json(force=True, silent=True) or {}
wt = re.sub(r"\s+", "", str(body.get("wt_protein", ""))).upper()
if not wt or not re.match(r"^[ACDEFGHIKLMNPQRSTVWY*]+$", wt):
return jsonify({"ok": False,
"error": "Paste the wild-type protein sequence too — "
"without it there is nothing to check the "
"mutation numbering against."}), 400
if len(wt) > _DE_ROUND2_MAX_AA:
return jsonify({"ok": False, "error":
f"This runs in real time for proteins up to {_DE_ROUND2_MAX_AA} aa "
f"on the free tier — yours is longer."}), 400
parsed = _meas.parse_table(str(body.get("text") or ""))
if not parsed.rows:
return jsonify({"ok": True, "n_parsed": 0, "header": parsed.header,
"rejects": [vars(j) for j in parsed.rejects],
"offset_hint": None, "calibration": None,
"measurements": []})
rows = parsed.rows
# An explicit offset means the user saw the suggestion and accepted it.
# Renumbering somebody's data without being asked would be the worst
# thing this endpoint could do, so it only ever happens on their say-so.
try:
offset = int(body.get("offset") or 0)
except (TypeError, ValueError):
offset = 0
if offset:
rows = _meas.apply_offset(rows, offset)
# Before spending an ESM pass: does the table even describe this
# sequence? A whole table numbered from the mature protein would score
# as garbage, and "your data doesn't correlate" is the single most
# damaging wrong answer this product could give.
hint = None if offset else _meas.detect_offset(rows, wt)
if hint:
return jsonify({"ok": True, "n_parsed": len(parsed.rows),
"header": parsed.header,
"rejects": [vars(j) for j in parsed.rejects],
"offset_hint": hint, "calibration": None,
"measurements": []})
kept, wt_bad = _meas.check_against_wt(rows, wt)
rejects = [vars(j) for j in parsed.rejects] + [vars(j) for j in wt_bad]
measurements = [{"mutations": r.mutations, "measured_value": r.measured_value}
for r in kept]
calibration = None
calibration_error = ""
if kept:
# THE ANSWER A SCIENTIST CAME FOR: does this engine agree with
# results they already trust? Same additive ΔLL the design engine
# ranks by (benchmark.predict_additive), and the same two metrics
# the published benchmarks report — so the number on their own
# protein is computed by the code path as the number in the
# Field Atlas, not a second one invented for this screen.
try:
from dee.core import benchmark as _bm
scorer = _scoring.get_scorer("small")
scores_df = _scoring.score_guarded(scorer, wt)
labels = [r.mutations for r in kept]
pred = _bm.predict_additive(scores_df, labels)
meas_vals = [r.measured_value for r in kept]
pts = [{"mutations": l, "predicted": (None if p != p else round(float(p), 4)),
"measured": m}
for l, p, m in zip(labels, pred, meas_vals)]
scored = [(p, m) for p, m in zip(pred, meas_vals) if p == p]
calibration = {
"n": len(scored),
"n_unscored": len(labels) - len(scored),
"spearman": _bm.spearman([p for p, _ in scored],
[m for _, m in scored]),
# top_decile_precision is shared with the published
# benchmarks, which run on thousands of variants. On a
# pasted table of 13 the "top 10%" is ONE variant, and
# reporting "0%" from a single pick reads as a damning
# finding rather than as too little data. The function is
# left alone (changing its floor would move a published
# number); the floor that matters here is applied here.
"top_decile_precision": (
_bm.top_decile_precision([p for p, _ in scored],
[m for _, m in scored])
if len(scored) >= _IMPORT_TDP_MIN_N else None),
"top_decile_min_n": _IMPORT_TDP_MIN_N,
"points": pts,
"basis": "ESM-2 additive ΔLL, the same ranking the design engine uses",
}
except Exception: # noqa: BLE001 — the import still succeeded
app.logger.exception("import calibration failed")
calibration = None
# A missing number must never read as "your data didn't
# correlate". The table was read and is still usable; what
# failed is the scorer, and saying so is the difference
# between an outage and an indictment of their data.
calibration_error = (
"The scoring model isn't available on this deployment right "
"now, so the rank-agreement check couldn't run. Your table "
"was read and you can still design a round from it.")
# Direction matters and only the assay knows it. Without one the
# correlation above is still real, it just cannot be compared with
# anybody else's — same rule as everywhere else.
pooling = None
aid = str(body.get("assay_id") or "").strip()
auth = _auth.get_auth()
if aid and not auth.anonymous:
arow = _auth.get_assay(auth.user_id, aid)
a = _assay.Assay.from_row(arow)
pooling = _assay.summarise_pooling(
[_assay.normalise(m["measured_value"], a) for m in measurements])
return jsonify({"ok": True, "n_parsed": len(parsed.rows),
"header": parsed.header, "rejects": rejects,
# The wild-type's own measurement, if the pasted table
# carried one. It is the anchor that decides whether
# any of these results can leave the account, and it is
# usually already in the file — see dee/core/assay.py.
"wt_value": parsed.wt_value,
"offset_hint": None, "offset_applied": offset,
"calibration": calibration,
"calibration_error": calibration_error, "pooling": pooling,
"measurements": measurements})
@app.post("/api/de/round2")
def de_round2() -> Response:
"""Propose a round-2 library learned from this user's logged outcomes."""
auth = _auth.get_auth()
if auth.anonymous:
return jsonify({"ok": False, "error": "signin_required",
"kind": "signin_required",
"signup_url": "https://turingdna.com/signin/?from=de"}), 403
body = request.get_json(force=True, silent=True) or {}
library_id = str(body.get("library_id", "")).strip()
wt_protein = re.sub(r"\s+", "", str(body.get("wt_protein", ""))).upper()
if not wt_protein or not re.match(r"^[ACDEFGHIKLMNPQRSTVWY*]+$", wt_protein):
return jsonify({"ok": False, "error": "Missing or non-standard wild-type protein."}), 400
if len(wt_protein) > _DE_ROUND2_MAX_AA:
return jsonify({"ok": False, "error":
f"Round 2 runs in real time for proteins up to {_DE_ROUND2_MAX_AA} aa on the "
"free tier — yours is longer."}), 400
# Outcomes: DB is authoritative; allow inline only as a fallback.
rows = _auth.list_de_outcomes(auth.user_id, library_id) if library_id else []
if not rows and isinstance(body.get("measurements"), list):
rows = body["measurements"]
_OUT_VAL = {"worked": 1.0, "partial": 0.5, "failed": 0.0}
measurements = []
for r in rows:
if not isinstance(r, dict):
continue
mv = r.get("measured_value")
try:
val = (float(mv) if mv is not None and mv != ""
else _OUT_VAL.get(str(r.get("outcome", "")).lower()))
except (TypeError, ValueError):
val = None
if val is None:
continue
measurements.append((r.get("mutations", ""), val))
# Sanitize settings — clamp model + host to known values (never load an
# arbitrary HF model id from the client).
s = body.get("settings") if isinstance(body.get("settings"), dict) else {}
model = s.get("model", "small")
if model not in ("small", "medium", "large"):
model = "small"
host = s.get("host", "e_coli")
if host not in ("e_coli", "yeast", "human"):
host = "e_coli"
try:
pct = float(s.get("percentile", 85.0))
except (TypeError, ValueError):
pct = 85.0
pct = min(99.0, max(1.0, pct))
def _clampi(v, lo, hi, d):
try:
return max(lo, min(hi, int(v)))
except (TypeError, ValueError):
return d
settings = {
"model": model, "host": host, "percentile": pct,
"k": _clampi(s.get("k", 30), 1, 200, 30),
"min_mutations": _clampi(s.get("min_mutations", 2), 1, 10, 2),
"max_mutations": _clampi(s.get("max_mutations", 5), 1, 12, 5),
"restarts": _clampi(s.get("restarts", 8), 1, 24, 8),
"steps": _clampi(s.get("steps", 1200), 100, 5000, 1200),
"seed": s.get("seed"),
}
try:
rows2, surrogate = _de_round2_library(wt_protein, settings, measurements)
except Exception: # noqa: BLE001
logger.exception("DE round-2 failed.")
return jsonify({"ok": False, "error":
"Couldn't compute round 2 — check the sequence and try again."}), 500
_auth.cleanup_expired_de_outcomes_async(auth.user_id)
# What of this could be compared with another lab's work, and what
# could not. Reported even though round 2 uses ALL of the user's own
# measurements — their own results are comparable with each other by
# construction. The split matters for what leaves this account, and
# showing it here is where a scientist is most likely to act on it:
# the fix is usually one number, and it is far cheaper to add before
# the next plate than after.
assay_row = None
aid = str(body.get("assay_id") or "").strip()
if aid:
assay_row = _auth.get_assay(auth.user_id, aid)
a = _assay.Assay.from_row(assay_row)
pooling = _assay.summarise_pooling(
[_assay.normalise(v, a) for _, v in measurements])
return jsonify({
"ok": True,
"variants": rows2,
"wt_protein": wt_protein,
"wt_identifier": str(body.get("wt_identifier", "") or "round 2"),
"settings_used": settings,
"surrogate": surrogate,
"round": 2,
"pooling": pooling,
})
@app.post("/api/de/epistasis")
def de_epistasis() -> Response:
"""Interaction Radar — zero-shot epistasis analysis of a library's
multi-mutants. Scores each variant's full mutant sequence with ESM-2
and reports how the model's preference for each mutation SHIFTS once
the others are present (dee.core.epistasis). Opt-in + bounded: signed-
in only, top-N variants, length-capped, since it's ~1 forward pass per
variant on the free CPU tier."""
auth = _auth.get_auth()
if auth.anonymous:
return jsonify({"ok": False, "error": "signin_required",
"kind": "signin_required",
"signup_url": "https://turingdna.com/signin/?from=de"}), 403
body = request.get_json(force=True, silent=True) or {}
wt_protein = re.sub(r"\s+", "", str(body.get("wt_protein", ""))).upper()
if not wt_protein or not re.match(r"^[ACDEFGHIKLMNPQRSTVWY*]+$", wt_protein):
return jsonify({"ok": False, "error": "Missing or non-standard wild-type protein."}), 400
# The 500-aa ceiling protects the shared CPU from free-tier work. ESM-2
# safely windows longer proteins around each mutated position, so Pro
# can analyze Cas9-sized proteins without exceeding model context.
if (len(wt_protein) > _DE_ROUND2_MAX_AA
and not _auth.has_pro_plan(auth.user_id)):
return jsonify({"ok": False, "error":
f"Interaction analysis for proteins above {_DE_ROUND2_MAX_AA} aa requires Pro."}), 400
variants = body.get("variants")
if not isinstance(variants, list) or not variants:
return jsonify({"ok": False, "error": "No variants to analyze."}), 400
# Only keep the fields we use, and bound the count BEFORE any scoring.
clean_variants = []
for v in variants[:60]:
if isinstance(v, dict) and v.get("Mutations_AA"):
clean_variants.append({
"Variant_ID": str(v.get("Variant_ID", "") or ""),
"Mutations_AA": str(v.get("Mutations_AA", "") or ""),
})
model = body.get("model", "small")
if model not in ("small", "medium", "large"):
model = "small"
try:
from dee.core import epistasis as _epi
scorer = _scoring.get_scorer(model)
# score_guarded enforces the same safety/timeout wrapper the rest of
# the DE path uses; epistasis needs raw per-residue log-probs, so we
# hand analyze_library the scorer's masked_log_probs directly.
# MASKED, not log_probs: comparing an unmasked WT background against
# an unmasked mutant one measures the model seeing the residue it is
# scoring, not the interaction. See dee.core.epistasis.
results = _epi.analyze_library(
wt_protein, clean_variants, scorer.masked_log_probs, max_variants=12,
)
except Exception: # noqa: BLE001
logger.exception("epistasis analysis failed")
return jsonify({"ok": False, "error":
"Couldn't analyze interactions — try again."}), 500
n_clash = sum(1 for r in results if r.verdict == "clash")
return jsonify({
"ok": True,
"analysis": [r.as_dict() for r in results],
"n_analyzed": len(results),
"n_clash": n_clash,
"wt_protein_length": len(wt_protein),
})
def _atlas_growth(source: str) -> Optional[Dict[str, Any]]:
"""Honest week-ish-over-week delta for one source, from the append-
only public.atlas_snapshots ledger (migration 0020) — never derived
from mutation_priors.updated_at, which is rewritten on every rebuild
whether or not anything changed. Returns None (not zero) until at
least two real snapshots exist, so a cold-start or pre-migration
deployment reads as "no history yet", not "no growth"."""
import datetime as _dt4
hist = _auth.get_atlas_history(source, limit=60) # newest-first
if len(hist) < 2:
return None
def _parse(ts):
try:
return _dt4.datetime.fromisoformat(str(ts).replace("Z", "+00:00"))
except (TypeError, ValueError):
return None
latest = hist[0]
latest_at = _parse(latest.get("taken_at"))
if latest_at is None:
return None
target = latest_at - _dt4.timedelta(days=7)
best, best_at, best_diff = None, None, None
for snap in hist[1:]:
at = _parse(snap.get("taken_at"))
if at is None:
continue
diff = abs((at - target).total_seconds())
if best_diff is None or diff < best_diff:
best, best_at, best_diff = snap, at, diff
if best is None:
return None
return {
"substitutions_delta": int(latest.get("substitutions", 0) or 0) - int(best.get("substitutions", 0) or 0),
"observations_delta": int(latest.get("observations", 0) or 0) - int(best.get("observations", 0) or 0),
"since": best.get("taken_at"),
"days": round((latest_at - best_at).total_seconds() / 86400.0, 1),
}
_COMMONS_CACHE: Dict[str, Any] = {}
_COMMONS_TTL = 300.0 # seconds
def _commons_summary() -> Dict[str, Any]:
"""The one place that reads public.mutation_priors and splits it by
provenance. /api/atlas and /api/mission both call this so they can
never drift apart on what "contributors" means — see migration 0018:
a DMS row's n_users counts independent PUBLISHED STUDIES, a user row's
n_users counts real platform labs, and summing or conflating them
overstates lab adoption this product has not earned."""
# CACHED, because this is a FULL-TABLE read of mutation_priors over the
# network and it ran on every page load — including for signed-out
# visitors, since /api/mission builds the public trust block before it
# checks auth. Measured at 680ms of the mission call's 680ms TTFB.
#
# Safe to cache and honest to: these are deployment-wide public
# aggregates that move when the weekly literature harvest lands or a
# lab logs a result, not per-request state. 300s matches the TTL this
# repo already uses for availability probes. It is NOT a durable cache
# — a restart re-reads, and five minutes is well inside the cadence
# anything here actually changes at.
_now = time.monotonic()
_hit = _COMMONS_CACHE.get("v")
if _hit is not None and (_now - _hit[0]) < _COMMONS_TTL:
return _hit[1]
rows = _auth.get_mutation_priors() # [{substitution, n_users, n_obs, mean_effect, source}]
total_obs = 0
by_source = {"user": 0, "dms": 0}
obs_by_source = {"user": 0, "dms": 0}
labs_by_source = {"user": 0, "dms": 0}
for r in rows:
src = r.get("source") or "user"
try:
n = int(r.get("n_obs", 0) or 0)
except (TypeError, ValueError):
n = 0
try:
nu = int(r.get("n_users", 0) or 0)
except (TypeError, ValueError):
nu = 0
total_obs += n
if src in by_source:
by_source[src] += 1
obs_by_source[src] += n
labs_by_source[src] = max(labs_by_source[src], nu)
out = {
"rows": rows,
"count": len(rows),
"total_observations": total_obs,
"provenance": {
"from_published_studies": {
"substitutions": by_source["dms"],
"observations": obs_by_source["dms"],
"studies": labs_by_source["dms"],
"unit": "independent published DMS assays",
},
"from_platform_labs": {
"substitutions": by_source["user"],
"observations": obs_by_source["user"],
"labs": labs_by_source["user"],
"unit": "distinct labs logging wet-lab outcomes",
},
},
"growth": {"dms": _atlas_growth("dms"), "user": _atlas_growth("user")},
}
# Only a REAL read is worth caching. get_mutation_priors() returns []
# on a misconfigured or unreachable Supabase, and freezing that for
# five minutes would turn a blip into a commons that reads empty long
# after it recovered — the same "availability must mean reachable"
# rule this repo applies to model tiers.
if rows:
_COMMONS_CACHE["v"] = (_now, out)
return out
@app.get("/api/atlas")
def atlas() -> Response:
"""The Field Atlas — a PUBLIC, de-identified read of the cross-user
substitution-effect aggregate (Pillar 2: the living commons). Every
row is a substitution TYPE (e.g. 'W>L'), pooled across labs, k-
anonymized (kept only when ≥ MIN_USERS distinct users measured it) and
z-scored per library before pooling — so no raw value, position, or
user identity survives. Safe to expose openly: the commons being
visible IS the product. Empty until the first post-effective-date
rebuild accrues enough data (that's the honest cold-start, shown as
such in the UI)."""
from dee.core import aggregate as _agg
summary = _commons_summary()
return _public_cors(jsonify({
"ok": True,
"substitutions": summary["rows"],
"count": summary["count"],
"total_observations": summary["total_observations"],
"min_users": _agg.MIN_USERS,
"effective_date": _agg.EFFECTIVE_DATE.isoformat(),
# The honest breakdown. `from_published_studies` is prior
# knowledge the field already had; `from_platform_labs` is the
# only number that measures THIS commons growing.
"provenance": summary["provenance"],
# Real week-ish deltas from atlas_snapshots, split by source; null
# (never 0) until enough history exists to diff against.
"growth": summary["growth"],
}))
@app.get("/api/models")
def models() -> Response:
"""Which scoring models this deployment can actually run.
The picker is built from THIS, not from a hardcoded list, so the UI can
never offer a model the server would refuse. `available` is the whole
point: 650M needs a GPU, the free CPU tier has none, and the honest
answer is "not here" rather than quietly serving 35M under its name
(see dee/core/scoring.effective_model).
"""
gpu = _scoring.gpu_available()
out = []
# 'large' (3B) is deliberately NOT offered. It needs an A10G-class GPU
# for a ~12 GB fp32 load where 650M fits a T4 at ~2.6 GB, and cold start
# — not inference — dominates the bill. ESM-2's zero-shot variant-effect
# performance is also not monotonic in size; 650M is the usual sweet
# spot. Offering a model that costs several times more for no MEASURED
# gain is exactly the claim this codebase refuses to make. It stays in
# MODEL_PRESETS so dee/core/benchmark.py can settle the question; put it
# back here when a benchmark says it earns its cost.
for key in ("small", "medium"):
needs_gpu = key in ("medium", "large")
# "available" must mean the run will WORK, not that a backend
# exists somewhere. Modal being configured is necessary but not
# sufficient: nothing routes a scoring call to it yet, so offering
# 650M here would hand the user a picker entry that errors. That is
# the same dishonesty as faking the number, inverted — promising a
# capability rather than a result.
available = _scoring.runnable(key)
note = _scoring.MODEL_NOTES[key]
# Key the note off `available` — the SAME predicate the flag uses.
# It was keyed off gpu_available() instead, and the two diverged the
# moment remote GPU scoring started routing: runnable() is
# `gpu_available() OR modal_client.is_configured()`, so on the Space
# (no local CUDA, MODAL_SUBMIT_URL set) Prometheus went out as
# available=true carrying the note "Not available on this deployment
# yet". A picker that contradicts itself in the same row is worse
# than either answer alone — and it is the same two-sources-of-truth
# bug as the DNA tool gate, which is why both now read one function.
if needs_gpu and not available:
note = ("Not available on this workbench yet. "
f"{_scoring.MODEL_LABELS['small']} is what runs here.")
out.append({
"id": key,
"label": _scoring.MODEL_LABELS[key],
"note": note,
"available": available,
# Must agree with `available`: "not-routed" alongside
# available:true said two opposite things about the same model.
"where": ("local" if not needs_gpu
else "local-gpu" if gpu
else "remote-gpu" if available
else "unavailable"),
})
# DNA tiers (Evo 2, via Modal) — a SEPARATE id-space from the protein
# tiers above ("achilles"/"prometheus", not "small"/"medium"), since
# neither picker key means anything for the other modality. Same
# honesty contract: `available` reflects dna_scoring.runnable(), not
# "a backend exists somewhere". There is no local/free tier here —
# unlike protein's 35M, every DNA call is a metered GPU call; see
# dee/core/dna_scoring.py's module docstring.
dna_out = []
for tier in ("achilles", "prometheus"):
dna_out.append({
"id": tier,
"label": _dna_scoring.TIER_LABELS[tier],
"note": _dna_scoring.TIER_NOTES[tier],
"available": _dna_scoring.runnable(tier),
# Two different numbers, deliberately, because conflating them
# is what let a request 122x larger than anything that can run
# get accepted. `context` is the CHECKPOINT's architectural
# window — a fact about Evo 2. `max_reference_nt` is what this
# deployment will actually accept, which is far smaller and is
# the only one a caller can plan against. Reporting `context`
# alone advertised a capacity no request could ever use.
"context": _dna_scoring.TIER_CONTEXT[tier],
"max_reference_nt": _dna_scoring.MAX_REFERENCE_NT,
"max_total_score_bases": _dna_scoring.MAX_TOTAL_SCORE_TOKENS,
"generation": tier in _dna_scoring.GENERATION_TIERS,
})
# Never cached. This reports what the deployment can do RIGHT NOW, and
# the two pickers (the bench form and the Turing composer) fetch it at
# different moments — one on DOMContentLoaded, one when the rail mounts.
# With no cache directive the browser cached it heuristically, so the
# two disagreed about whether 650M existed: one showed it, one did not,
# from the same deployment. A capability answer that can go stale is
# worse than no answer.
resp = jsonify({"models": out, "default": "small",
"dna_models": dna_out, "dna_default": "achilles"})
resp.headers["Cache-Control"] = "no-store, max-age=0"
return resp
# ── DNA-level scoring / generation (Evo 2) ───────────────────────────
# Until now the ONLY route to Evo 2 was the agent deciding to call
# score_or_generate_dna mid-conversation. The engine computed dna_models
# for /api/models and no frontend read it, so a capability that runs on a
# 7B genome model was, in practice, unreachable by clicking. These are the
# two endpoints the DNA view calls.
#
# Both are deliberately thin: dee/core/dna_scoring.py already owns tier
# rules, bounds and the "not configured is not the same as broken"
# distinction, and duplicating any of that here would give the product two
# answers to the same question.
def _dna_signin_gate(where: str):
"""Same contract as CRISPR/primers: no anonymous use. Distinct from
the trial timer — the 403 carries kind=signin_required so auth.js
routes to /signin rather than showing the trial modal."""
auth = _auth.get_auth()
if auth.anonymous:
return jsonify({
"error": ("DNA scoring requires a free account. Sign in or "
"create one to keep going."),
"kind": "signin_required",
"signup_url": f"https://turingdna.com/signin/?from={where}",
}), 403
return None
@app.post("/api/dna/score")
def dna_score() -> Response:
gate = _dna_signin_gate("dna")
if gate is not None:
return gate
body = request.get_json(force=True, silent=True) or {}
reference = "".join(str(body.get("reference") or "").split()).upper()
raw_variants = body.get("variants") or []
tier = str(body.get("tier") or "achilles").strip().lower()
if not reference:
return jsonify({"error": "missing 'reference'"}), 400
if not isinstance(raw_variants, list) or not raw_variants:
return jsonify({"error": "missing 'variants'"}), 400
variants = [str(v).strip().upper() for v in raw_variants if str(v).strip()]
if not variants:
return jsonify({"error": "missing 'variants'"}), 400
try:
result = _dna_scoring.score_dna_variants(reference, variants, tier=tier)
except _dna_scoring.DnaModelUnavailable as exc:
# Customer-safe by construction — dna_scoring keeps env-var names
# and repo paths in the log, not in this message.
return jsonify({"error": str(exc), "kind": "dna_unavailable"}), 503
except ValueError as exc:
return jsonify({"error": str(exc)}), 400
except Exception: # noqa: BLE001
app.logger.exception("dna_score failed")
return jsonify({
"error": ("The DNA model didn't return a result for that "
"request. Nothing was scored — try again, and if it "
"keeps failing the backend is having a moment."),
"kind": "dna_error",
}), 502
return jsonify(result)
@app.post("/api/dna/generate")
def dna_generate() -> Response:
gate = _dna_signin_gate("dna")
if gate is not None:
return gate
body = request.get_json(force=True, silent=True) or {}
prompt = "".join(str(body.get("prompt") or "").split()).upper()
tier = str(body.get("tier") or "prometheus").strip().lower()
if not prompt:
return jsonify({"error": "missing 'prompt'"}), 400
try:
n_tokens = int(body.get("n_tokens") or 200)
temperature = float(body.get("temperature") or 1.0)
top_k = int(body.get("top_k") or 4)
except (TypeError, ValueError):
return jsonify({"error": "n_tokens/temperature/top_k must be numbers"}), 400
try:
result = _dna_scoring.generate_dna_sequence(
prompt, n_tokens=n_tokens, tier=tier,
temperature=temperature, top_k=top_k)
except _dna_scoring.DnaModelUnavailable as exc:
# Also the path for "generation is Prometheus-only" — enforced in
# dna_scoring AND again in modal/evo2_scoring, so a caller that
# skips this route still gets refused.
return jsonify({"error": str(exc), "kind": "dna_unavailable"}), 503
except ValueError as exc:
return jsonify({"error": str(exc)}), 400
except Exception: # noqa: BLE001
app.logger.exception("dna_generate failed")
return jsonify({
"error": ("The DNA model didn't return a sequence for that "
"prompt. Nothing was generated — try again."),
"kind": "dna_error",
}), 502
return jsonify(result)
# ── Therapeutic compiler ─────────────────────────────────────────────
# Lowers a pathogenic variant to an editing strategy, or refuses with a
# diagnostic. See dee/core/compiler.py for the scope this operates in —
# somatic design and assessment only, never a clinical decision.
def _score_variant_consequence(wt: str, patient: str, window: str, offset: int):
"""Ask Evo 2 what the PATIENT allele does, in its real sequence context.
The variant scored is the pathogenic one against the wild-type window
— `<wt><1-based pos><patient>` — so a negative delta log-likelihood
means the model finds the patient's sequence less likely than
wild-type. That is supporting evidence about the lesion, not proof:
zero-shot, no training on this locus, no validated relationship to
clinical consequence.
Returns None when there is nothing to score (no window, or a lesion
with no single-base correction). Never raises — a scoring failure must
degrade the record's completeness, not the compile.
"""
if not window or offset < 0 or offset >= len(window):
return None
if len(wt) != 1 or len(patient) != 1:
return None # indels are not point substitutions
if window[offset].upper() != wt.upper():
return None # the verify pass will refuse this anyway
label = f"{wt.upper()}{offset + 1}{patient.upper()}"
try:
res = _dna_scoring.score_dna_variants(window, [label], tier="achilles")
except Exception as exc: # noqa: BLE001
app.logger.warning("consequence scoring failed: %s", exc)
return {"ok": False, "label": label,
"error": "the DNA model did not return a score"}
scores = (res or {}).get("scores") or []
hit = next((s for s in scores if s.get("label") == label), None)
if hit is None:
skipped = (res or {}).get("skipped") or []
return {"ok": False, "label": label,
"error": (f"the model returned no score for {label}"
+ (f" (skipped: {skipped})" if skipped else ""))}
return {"ok": True, "label": label, "delta_ll": hit.get("delta_ll"),
"reference_ll": (res or {}).get("reference_ll"),
"checkpoint": (res or {}).get("checkpoint"),
"tier": (res or {}).get("tier")}
def _score_labels(window: str, labels):
"""Score a batch of forward-strand variant labels in ONE model call.
Batched on purpose: a 7B GPU round trip per bystander would make the
blast-radius view unusable. Returns a label->delta_ll map containing
only what actually came back — a missing label stays missing, and the
caller leaves it as None rather than defaulting it to neutral.
"""
if not window or not labels:
return {}
try:
res = _dna_scoring.score_dna_variants(window, list(labels),
tier="achilles")
except Exception as exc: # noqa: BLE001
app.logger.warning("bystander scoring failed: %s", exc)
return {}
out = {}
for row in (res or {}).get("scores") or []:
lab, d = row.get("label"), row.get("delta_ll")
if lab and isinstance(d, (int, float)):
out[lab] = d
return out
@app.post("/api/compiler/compile")
def compiler_compile() -> Response:
gate = _dna_signin_gate("compiler")
if gate is not None:
return gate
body = request.get_json(force=True, silent=True) or {}
# A notation is the normal way to name a variant. Resolving it here
# means the caller never hand-types alleles or coordinates — and the
# forward-strand alleles come back RECONCILED against the reference,
# not taken on trust from the notation's strand.
resolved = None
notation = str(body.get("variant") or "").strip()
if notation:
try:
rv = _variant_resolve.resolve(notation)
except _variant_resolve.ResolveError as exc:
return jsonify({"error": str(exc), "kind": "unresolved"}), 422
except Exception: # noqa: BLE001
app.logger.exception("variant resolution failed")
return jsonify({
"error": ("Could not reach the variant databases just now. "
"Nothing was resolved — try again, or supply the "
"alleles and reference window directly."),
"kind": "resolver_unavailable"}), 503
resolved = _variant_resolve.resolved_to_dict(rv)
body = dict(body)
body["wt_allele"] = rv.wt_base
body["patient_allele"] = rv.patient_base
body["window"] = rv.window
body["offset"] = rv.offset
wt = str(body.get("wt_allele") or "")
patient = str(body.get("patient_allele") or "")
window = "".join(str(body.get("window") or "").split()).upper()
try:
offset = int(body.get("offset", -1))
except (TypeError, ValueError):
return jsonify({"error": "offset must be an integer"}), 400
if not wt and not patient:
return jsonify({"error": "missing 'wt_allele' and 'patient_allele'"}), 400
# Consequence scoring is OPT-IN. The compile itself is pure logic and
# returns in milliseconds; an Evo 2 call is a rented 7B GPU and ~80s
# cold, so running it on every compile would make the fast, honest
# part of the tool feel broken. Off by default means the pass reports
# `unavailable` — which is true — rather than quietly costing money.
consequence = None
if body.get("score_consequence"):
consequence = _score_variant_consequence(wt, patient, window, offset)
# Actually design. Cheap, deterministic, no network.
strategies, pegrnas, enum_diags = None, None, None
call = _compiler.classify_lesion(wt, patient)
if call.correction is not None and window and 0 <= offset < len(window):
try:
strategies, enum_diags = _compiler.plan_base_edit_strategies(
window, offset, call.correction,
editor_id=str(body.get("editor") or ""))
except Exception: # noqa: BLE001
app.logger.exception("guide enumeration failed")
strategies, enum_diags = None, None
elif call.route == "prime_editing" and window and offset >= 0:
# Transversions and small indels land here — the majority of
# pathogenic variation, and everything that used to stop at
# `prime_editing_unavailable`.
try:
pegrnas, enum_diags = _prime_editor.design_pegrnas(
window, offset, wt, patient)
except Exception: # noqa: BLE001
app.logger.exception("pegRNA design failed")
pegrnas, enum_diags = None, None
# The blast radius: what ELSE this guide changes, scored in context.
# Standard practice is to assume a silent or intronic bystander is
# benign. Nothing checks that today; this does.
if strategies and body.get("score_consequence"):
try:
labels = _compiler.label_bystanders(window, strategies)
if labels:
_compiler.attach_bystander_scores(
strategies, _score_labels(window, labels))
except ValueError:
# A label that disagrees with the genome means the mapping is
# broken; leave every bystander unscored rather than score a
# position we cannot vouch for.
app.logger.exception("bystander labelling refused")
try:
report = _compiler.compile_report(
wt, patient, window=window, offset=offset,
germline=bool(body.get("germline")),
strategies=strategies, pegrnas=pegrnas,
enumerate_diags=enum_diags,
consequence=consequence,
can_check_specificity=False,
)
except _compiler.GermlineRefused as exc:
# 422, not 400: the request is well-formed and understood, and is
# being refused on scope. Distinct code so nothing retries it.
return jsonify({"error": str(exc), "kind": "out_of_scope"}), 422
out = _compiler.report_to_dict(report)
out["resolved"] = resolved
# The deliverable: a design and its complete justification, in a form
# that diffs and pastes. Deterministic — same inputs, same bytes.
out["record"] = _compiler.design_record(
report, variant=notation, resolved=resolved,
provenance={"engine": "TuringDNA compiler",
"record": _compiler.RECORD_VERSION})
# The certificate. The record above is the human document; this is its
# identity. Content-addressed, so "certificate 8f2a…" names exactly one
# compile and nobody can rename a result into agreeing with them — and
# `not_established` is hashed with everything else, so the caveats
# cannot be quietly dropped while the id survives.
try:
cert = _certificate.build(
kind="therapeutic_compile",
inputs={"variant": notation, "wt_allele": wt,
"patient_allele": patient,
"window_sha256": _hash_sequence(window) if window else "",
"window_length": len(window), "offset": offset,
"scored_consequence": bool(body.get("score_consequence"))},
outputs={"compiled": bool(report.compiled),
"route": (report.lesion.route if report.lesion else ""),
"n_strategies": len(report.strategies or []),
"n_pegrnas": len(report.pegrnas or [])},
passes=[{"name": p.name, "title": p.title, "status": p.status,
"detail": p.detail} for p in report.passes],
not_established=report.incomplete_because,
versions={"record": _compiler.RECORD_VERSION,
"cert": _certificate.CERT_VERSION},
scope=report.scope)
out["certificate"] = cert
out["certificate_text"] = _certificate.render_text(cert)
except _certificate.CanonicalError as exc:
# Refusing beats minting an id over a value we cannot reproduce.
app.logger.warning("certificate refused: %s", exc)
out["certificate"] = None
out["certificate_error"] = str(exc)
return jsonify(out)
@app.post("/api/certificate/verify")
def certificate_verify() -> Response:
"""Recompute a certificate's id from its content. PUBLIC and unauthenticated
on purpose: a document that only its issuer can verify is not evidence.
Nothing is stored — the certificate is supplied by the caller."""
body = request.get_json(force=True, silent=True) or {}
cert = body.get("certificate")
if not isinstance(cert, dict):
return jsonify({"ok": False,
"error": "Send {certificate: {...}}."}), 400
ok, why = _certificate.verify(cert)
return jsonify({"ok": True, "valid": ok, "explanation": why,
"id": cert.get("id")})
@app.get("/api/compiler/validation")
def compiler_validation() -> Response:
"""Landmark designs the compiler is checked against. Public: the whole
value is that anyone can look at it, including the gaps."""
report = _compiler_validation.run_validation()
resp = jsonify(_compiler_validation.report_to_dict(report))
resp.headers["Cache-Control"] = "no-store, max-age=0"
return resp
@app.get("/api/benchmarks")
def benchmarks() -> Response:
"""The receipts — how well the engine's zero-shot ranking predicts
measured DMS fitness (Spearman ρ + top-decile precision), per published
dataset. PUBLIC read. Served from dee/data/benchmarks.json, which is
produced ONLY by scripts/run_benchmarks.py against real data with the
live model — never hand-entered. Ships empty (honest 'pending' state)
until a real run populates it; no fabricated numbers ever reach here."""
import json as _json
path = Path(__file__).resolve().parent / "data" / "benchmarks.json"
try:
data = _json.loads(path.read_text(encoding="utf-8"))
except Exception: # noqa: BLE001
data = {"generated_at": None, "model": None,
"summary": {"n_datasets": 0, "n_variants": 0,
"median_spearman": None, "median_top_decile_precision": None},
"datasets": []}
data["ok"] = True
return _public_cors(jsonify(data))
# Collaboration is an additive API; private tool routes retain their
# owner-only authorization and do not depend on the workspace migration.
from .core.project_workspaces import register as register_workspaces
register_workspaces(app, _auth)
# Referrals ride the same rail (0058) and the same JWT-only transport.
from .core.referrals import register as register_referrals
register_referrals(app, _auth)
# ── Constructs: the object every tool operates on ────────────────────
# The inversion. Until now a tool was a destination you pasted a sequence
# into, and each of the eight tool views had its own empty input. Nothing
# carried across them, so "which project does this guide set belong to?"
# could only ever be reconstructed afterwards by matching name strings.
#
# A construct is the noun. The tools become verbs applied to it. Once the
# client holds a selected construct and sends its id with every tool call,
# attribution stops being a reconstruction and becomes a record.
#
# SIGNED OUT these endpoints all 401 by design, and the client keeps its
# selection purely in local storage. Public tool access was a deliberate
# decision and must not regress: nothing here may become a precondition
# for running a tool.
def _construct_out(row):
"""Trim a row for the client. The full DNA/protein is only sent by the
single-construct GET — the switcher must not ship megabytes."""
if not row:
return None
return {
"id": row.get("id"), "name": row.get("name"),
"phase": row.get("phase") or "Design",
"sequence_hash": row.get("sequence_hash"),
"wt_identifier": row.get("wt_identifier"),
"created_at": row.get("created_at"),
"updated_at": row.get("updated_at"),
"n_plasmids": len(row.get("plasmid_ids") or []),
"n_crispr": len(row.get("crispr_ids") or []),
"n_primers": len(row.get("primer_ids") or []),
"library_id": row.get("library_id"),
}
@app.get("/api/constructs")
def constructs_list() -> Response:
auth = _auth.get_auth()
if auth.anonymous:
# Not an error. A signed-out visitor has no saved projects and the
# UI shows its local-only selection instead.
return jsonify({"ok": True, "gated": True, "constructs": []})
rows = _auth.list_constructs(auth.user_id)
return jsonify({"ok": True, "gated": False,
"constructs": [_construct_out(r) for r in rows]})
@app.post("/api/constructs")
def constructs_create() -> Response:
"""Create, or return the one already grouping this sequence.
Pasting the same sequence twice means "the thing I am already working
on", not a second project. Silently creating a duplicate is how a
workspace turns into a list the user has to tidy, so the hash decides.
"""
auth = _auth.get_auth()
if auth.anonymous:
return jsonify({"error": "Sign in to keep projects.",
"kind": "signin_required"}), 401
body = request.get_json(force=True, silent=True) or {}
dna = "".join(str(body.get("sequence_dna") or "").split()).upper()
protein = "".join(str(body.get("wt_protein") or "").split()).upper()
basis = dna or protein
if not basis:
return jsonify({"error": "A project needs a sequence."}), 400
seq_hash = _hash_sequence(basis)
existing = _auth.find_construct_by_hash(auth.user_id, seq_hash)
if existing:
return jsonify({"ok": True, "reused": True,
"construct": _construct_out(existing)})
res = _auth.save_construct(
auth.user_id,
name=str(body.get("name") or "").strip() or "Untitled construct",
sequence_hash=seq_hash,
wt_identifier=str(body.get("wt_identifier") or ""),
wt_protein=protein, sequence_dna=dna,
phase=str(body.get("phase") or "Design"))
if not res.get("ok"):
return jsonify({"error": res.get("error") or "Couldn't create."}), 400
return jsonify({"ok": True, "reused": False,
"construct": _construct_out(res.get("construct"))})
@app.get("/api/constructs/<construct_id>")
def constructs_get(construct_id: str) -> Response:
auth = _auth.get_auth()
if auth.anonymous:
return jsonify({"error": "Sign in.", "kind": "signin_required"}), 401
row = _auth.get_construct(auth.user_id, construct_id)
if not row:
return jsonify({"error": "Not found."}), 404
out = _construct_out(row)
# Only here, and only for the one the user actually selected.
out["sequence_dna"] = row.get("sequence_dna") or ""
out["wt_protein"] = row.get("wt_protein") or ""
# What the project last had on the bench (project.js): which
# structure, library and conversation to bring back when it is
# reopened, on this machine or another.
out["workspace"] = _auth.clean_workspace((row.get("meta") or {}).get("workspace"))
return jsonify({"ok": True, "construct": out})
@app.put("/api/constructs/<construct_id>/workspace")
def constructs_workspace(construct_id: str) -> Response:
"""Record what a project has on the bench.
Pointers only — an AlphaFold accession, a library id, a run id — each
validated to its own shape, so this can never become a place to park
a sequence or arbitrary JSON on the construct row."""
auth = _auth.get_auth()
if auth.anonymous:
return jsonify({"error": "Sign in.", "kind": "signin_required"}), 401
body = request.get_json(force=True, silent=True) or {}
res = _auth.save_workspace(auth.user_id, construct_id, body.get("workspace"))
if not res.get("ok"):
return jsonify({"error": res.get("error") or "Couldn't save."}), 400
return jsonify({"ok": True, "workspace": res.get("workspace")})
@app.patch("/api/constructs/<construct_id>")
def constructs_patch(construct_id: str) -> Response:
auth = _auth.get_auth()
if auth.anonymous:
return jsonify({"error": "Sign in.", "kind": "signin_required"}), 401
body = request.get_json(force=True, silent=True) or {}
res = _auth.update_construct(
auth.user_id, construct_id,
**{k: body[k] for k in ("name", "phase") if k in body})
if not res.get("ok"):
return jsonify({"error": res.get("error") or "Couldn't update."}), 400
return jsonify({"ok": True, "construct": _construct_out(res.get("construct"))})
@app.delete("/api/constructs/<construct_id>")
def constructs_delete(construct_id: str) -> Response:
auth = _auth.get_auth()
if auth.anonymous:
return jsonify({"error": "Sign in.", "kind": "signin_required"}), 401
res = _auth.delete_construct(auth.user_id, construct_id)
if not res.get("ok"):
return jsonify({"error": res.get("error") or "Couldn't delete."}), 400
return jsonify({"ok": True})
@app.get("/api/constructs/<construct_id>/lineage")
def constructs_lineage(construct_id: str) -> Response:
"""The recorded graph for one project: what came from what.
Nodes are derived from the edges, never stored separately, so the view
cannot render a relationship no edge records.
"""
auth = _auth.get_auth()
if auth.anonymous:
return jsonify({"error": "Sign in.", "kind": "signin_required"}), 401
edges = _auth.list_lineage(auth.user_id, construct_id)
graph = _auth.lineage_graph(edges)
# State the composition of what is being drawn. A caller that cannot
# tell observed edges from reconstructed ones will eventually present
# a guess as a fact.
by_origin = {}
for e in graph["edges"]:
by_origin[e["origin"]] = by_origin.get(e["origin"], 0) + 1
graph["by_origin"] = by_origin
return jsonify({"ok": True, "construct_id": construct_id, **graph})
@app.get("/api/mission")
def mission() -> Response:
"""Command-center rollup for the Mission Control home — ONE call that
assembles the whole front-door state so the home never fans out into a
dozen requests. Everything here is REAL, cheaply-available data:
• constructs — the user's saved work, grouped best-effort by name,
each enriched with its real loop phase, top-variant fitness, variant
count, how many variants they've measured, and a computed
next-best-action (round 2 ready / build / open).
• loop — Design/Build/Edit counts + the honest Learn signal (results
logged, loops closed). No fabricated improvement %.
• trust — the PUBLIC receipts (benchmark ρ) + commons size, so even a
signed-out visitor (an investor) sees the engine is validated.
The trust block is public; constructs/loop require sign-in (gated flag
so the UI can prompt without erroring). Running jobs are tracked
client-side (per tab), not here."""
import json as _json
# ── trust: public receipts + commons (works signed-out) ──────────
bpath = Path(__file__).resolve().parent / "data" / "benchmarks.json"
try:
bench = _json.loads(bpath.read_text(encoding="utf-8")).get("summary", {})
except Exception: # noqa: BLE001
bench = {}
# Same helper /api/atlas uses — was previously its own inline
# max(n_users) over ALL rows regardless of source, which is the exact
# conflation migration 0018 exists to prevent: with the commons
# currently 100% DMS-seeded, that used to report a DMS row's
# independent-study count as "N labs contributing" on the home page.
summary = _commons_summary()
prov = summary["provenance"]
trust = {
"spearman": bench.get("median_spearman"),
"n_datasets": bench.get("n_datasets", 0),
"n_variants": bench.get("n_variants", 0),
"commons_substitutions": summary["count"],
"commons_observations": summary["total_observations"],
"commons_studies": prov["from_published_studies"]["studies"],
"commons_labs": prov["from_platform_labs"]["labs"],
"commons_growth": summary["growth"],
}
auth = _auth.get_auth()
if auth.anonymous:
return jsonify({"ok": True, "gated": True, "constructs": [],
"loop": {"design": 0, "build": 0, "edit": 0,
"learn_logged": 0, "loops_closed": 0},
"trust": trust})
libs = _auth.list_libraries(auth.user_id)
plas = _auth.list_plasmids(auth.user_id)
cris = _auth.list_crispr_designs(auth.user_id)
prim = _auth.list_primer_analyses(auth.user_id)
projects = _auth.list_constructs(auth.user_id)
tested = _auth.de_outcome_counts_by_library(auth.user_id) # {library_id: n}
def _norm(s):
return re.sub(r"[^a-z0-9]+", "", str(s or "").lower())[:40]
# Flatten every artifact into a common shape, carrying the library-only
# enrichment (fitness / variant count / measured count) where present.
raw = []
project_names = {}
artifact_projects = {}
for c in projects:
cid = c.get("id")
if not cid:
continue
project_names.setdefault(_norm(c.get("name")), []).append(cid)
# Every library filed under the project (attach_library keeps the
# history in meta; library_id is only the newest).
lib_ids = [c.get("library_id")] + list(((c.get("meta") or {}).get("library_ids")) or [])
for kind, ids in (("library", lib_ids),
("plasmid", c.get("plasmid_ids") or []),
("crispr", c.get("crispr_ids") or []),
("primer", c.get("primer_ids") or [])):
for aid in ids:
if aid:
artifact_projects[(kind, aid)] = cid
# Explicit allowlist: the rollup must never serialize sequences or
# arbitrary project metadata, even if a storage helper returns them.
raw.append({"kind": "construct", "id": cid,
"name": c.get("name") or "Construct",
"at": c.get("updated_at") or c.get("created_at"),
"created_at": c.get("created_at"),
"phase": {"Design": 0, "Build": 1, "Edit": 2, "Learn": 3}.get(c.get("phase"), 0)})
for l in libs:
lid = l.get("id")
raw.append({"kind": "library", "id": lid, "name": l.get("name") or "Library",
"at": l.get("created_at"), "phase": 0,
"top_fitness": l.get("top_fitness"), "n_variants": l.get("n_variants"),
"n_tested": int(tested.get(lid, 0)),
# For the card's thumbnail: the same protein the library
# was built on, so a construct can show a real fold
# instead of a glyph — resolved client-side, lazily.
"gene": _gene_token(l.get("wt_identifier")),
"organism": (l.get("host_organism") or "")})
for p in plas:
raw.append({"kind": "plasmid", "id": p.get("id"), "name": p.get("name") or "Plasmid",
"at": p.get("created_at"), "phase": 1})
for d in cris:
raw.append({"kind": "crispr", "id": d.get("id"),
"name": d.get("label") or d.get("gene_symbol") or "CRISPR design",
"at": d.get("created_at"), "phase": 2,
"gene": _gene_token(d.get("gene_symbol")),
"organism": (d.get("organism") or "")})
for a in prim:
raw.append({"kind": "primer", "id": a.get("id"), "name": a.get("label") or "Primer set",
"at": a.get("created_at"), "phase": 1})
# Group by normalized name into constructs; the most-recent member is the
# card's "primary" (drives click + kind); phase = furthest step reached,
# bumped to Learn(3) once any grouped library has measured results.
groups = {}
for r in raw:
matches = project_names.get(_norm(r["name"]), [])
cid = (r["id"] if r["kind"] == "construct" else
artifact_projects.get((r["kind"], r["id"])) or
(matches[0] if len(matches) == 1 else None))
key = ("construct:" + str(cid)) if cid else (_norm(r["name"]) or (r["kind"] + str(r["id"])))
g = groups.get(key)
if not g:
# The key the client files this project's workspace under
# (project.js). It must not change as work lands, so it is
# never the newest artifact's id while a steadier one exists.
client_key = ("construct:" + str(cid) if cid else
("name:" + _norm(r["name"])) if _norm(r["name"]) else
(r["kind"] + ":" + str(r["id"])))
# `created_at` is when this project FIRST appeared (min
# across grouped members); `updated_at` is when the newest
# piece of work landed on it (max). Legacy `at` = updated
# (existing behaviour preserved for older clients).
# The construct id the card belongs to, or None for a group
# that only name-matched. The client needs a STABLE id to join
# collaboration state onto: `primary.id` is whichever artifact
# happens to be newest, so it changes as work lands.
g = {"construct_id": cid, "key": client_key, "members": {},
"name": r["name"], "at": r["at"], "phase": r["phase"], "primary": r,
"created_at": r.get("created_at") or r["at"], "updated_at": r["at"],
"top_fitness": None, "n_variants": None, "n_tested": 0,
"gene": "", "organism": ""}
groups[key] = g
if cid and not g.get("construct_id"):
g["construct_id"] = cid
# Every artifact in the project, by kind, so a reopened project
# can put each tool's own work back — not only the one artifact
# the card happens to lead with.
if r["kind"] != "construct" and r.get("id"):
g["members"].setdefault(r["kind"], []).append((str(r.get("at") or ""), str(r["id"])))
g["phase"] = max(g["phase"], r["phase"])
# Carry a protein identity for the thumbnail. A library's is the
# truest (it IS the protein being engineered), so it wins; otherwise
# the first member that has one.
if r.get("gene") and (not g["gene"] or r["kind"] == "library"):
g["gene"], g["organism"] = r["gene"], r.get("organism") or ""
if r["kind"] == "library":
g["top_fitness"] = r.get("top_fitness")
g["n_variants"] = r.get("n_variants")
g["n_tested"] = max(g["n_tested"], r.get("n_tested", 0))
if r.get("n_tested", 0) > 0:
g["phase"] = 3 # measured → the loop reached Learn
if r["at"]:
if not g["updated_at"] or r["at"] > g["updated_at"]:
g["updated_at"] = r["at"]
g["at"] = r["at"]
born = r.get("created_at") or r["at"]
if not g["created_at"] or born < g["created_at"]:
g["created_at"] = born
# A saved result remains directly reopenable even if the enclosing
# project was renamed/touched more recently than that result.
p = g["primary"]
if r["kind"] != "construct" and (p["kind"] == "construct" or str(r["at"] or "") > str(p["at"] or "")):
g["primary"] = r
def _next_action(g):
p = g["primary"]
if g.get("n_tested", 0) > 0:
n = g["n_tested"]
return {"type": "round2",
"text": f"<b>Round 2 is ready.</b> Turing can re-rank from your "
f"{n} logged result{'s' if n != 1 else ''}.",
"cta": "Open round 2", "route": "design"}
if p["kind"] == "library" and (g.get("n_variants") or 0) > 0:
# A LIBRARY THAT HAS BEEN SITTING A WHILE IS WAITING ON A BENCH.
#
# "Ready to build" is right for a library designed this morning
# and wrong for one designed a month ago: by then the person is
# far more likely to be holding numbers than to have forgotten
# to start. The product cannot see a bench, but it can see age,
# and age is the honest proxy — so past the threshold the card
# asks for results instead of pointing at the plasmid editor.
#
# This is the return path that needs no email. Whenever they
# come back, for any reason, the open loop is what greets them.
if _days_since(g.get("at")) >= MISSION_AWAITING_RESULTS_DAYS:
return {"type": "awaiting",
"text": "<b>Waiting on your results.</b> Designed "
f"{_days_since(g.get('at'))} days ago and nothing "
"logged yet — open it and drop in what you measured.",
"cta": "Log results", "route": "library"}
return {"type": "build",
"text": "<b>Library's ready to build.</b> Map the construct, design "
"primers, then order your top picks — and log what you measure.",
"cta": "Build construct", "route": "plasmid"}
return {"type": "open",
"text": f"Open this {p['kind']} to keep working.",
"cta": "Open", "route": p["kind"]}
# OPEN LOOPS FIRST, then recency. Sorting purely by date buries the
# one card that is waiting on the user under whatever they opened most
# recently — and the whole reason this surface exists is to put the
# unfinished thing in front of them.
def _rank(g):
act = _next_action(g)
waiting = 1 if act.get("type") in ("awaiting", "round2") else 0
return (waiting, str(g["at"] or ""))
constructs = []
for g in sorted(groups.values(), key=_rank, reverse=True):
p = g["primary"]
members = {k: [i for _, i in sorted(v, reverse=True)][:10]
for k, v in (g.get("members") or {}).items()}
constructs.append({
"kind": p["kind"], "id": p["id"], "name": g["name"], "at": g["at"],
"construct_id": g.get("construct_id"),
"key": g.get("key"), "members": members,
# Two dates for the card:
# created_at = earliest member.created_at (project's birth)
# updated_at = latest member.created_at (last activity)
# `at` is preserved (== updated_at) so older clients still work.
"created_at": g.get("created_at"),
"updated_at": g.get("updated_at"),
"phase": g["phase"],
"top_fitness": g.get("top_fitness"), "n_variants": g.get("n_variants"),
"n_tested": g.get("n_tested", 0),
# Protein identity for the card thumbnail (may be ""). The client
# resolves this to a UniProt accession and draws the fold, or —
# if it doesn't resolve — shows an honest "no structure" plate.
"gene": g.get("gene") or "", "organism": g.get("organism") or "",
"next": _next_action(g),
})
total_tested = sum(tested.values())
loops_closed = sum(1 for v in tested.values() if v > 0)
trust["you_contributed"] = total_tested
return jsonify({
"ok": True, "gated": False, "constructs": constructs,
"loop": {"design": len(libs), "build": len(plas) + len(prim),
"edit": len(cris), "learn_logged": total_tested,
"loops_closed": loops_closed},
"trust": trust,
})
@app.get("/api/structure/resolve")
def structure_resolve() -> Response:
"""Resolve a gene symbol (+ organism) to a UniProt accession for a
construct's thumbnail — the same lookup the CRISPR structure viewer uses
(dee.core.resolve), cached so the catalog grid doesn't re-hit UniProt.
Returns {ok:true, uniprot} on a confident hit, else {ok:false}: the card
then draws no structure rather than a fold for the wrong protein — the
same rule the structure card already follows for a run with no accession.
Public data (a gene maps to a public accession), so no auth gate."""
gene = _gene_token(request.args.get("gene"))
organism = (request.args.get("organism") or "").strip()
if not gene:
return jsonify({"ok": False, "error": "bad_gene"}), 422
if not organism or len(organism) > 64 or not re.match(r"^[A-Za-z0-9 .'-]+$", organism):
return jsonify({"ok": False, "error": "bad_organism"}), 422
key = (gene.lower(), organism.lower())
if key in _CONSTRUCT_ACC_CACHE:
acc = _CONSTRUCT_ACC_CACHE[key]
return jsonify({"ok": bool(acc), "uniprot": acc, "cached": True})
from dee.core import resolve as _resolve
try:
result = _resolve.resolve_uniprot(organism, gene)
except Exception: # noqa: BLE001
logger.exception("construct accession resolve failed")
return jsonify({"ok": False, "error": "resolve_failed"}), 200
acc = result.get("uniprot") if result.get("ok") else None
# Cache the answer — including a confirmed miss — but never a transient
# UniProt outage, so a blip isn't remembered forever as "no structure".
if result.get("ok") or not result.get("transient"):
_CONSTRUCT_ACC_CACHE[key] = acc
return jsonify({"ok": bool(acc), "uniprot": acc,
"protein_name": result.get("protein_name")})
@app.get("/api/admin/mutation-priors")
def admin_mutation_priors() -> Response:
"""View the stored cross-user aggregate (de-identified). Admin-token gated."""
from dee.core import aggregate as _agg
if not _admin_ok():
return jsonify({"ok": False, "error": "forbidden"}), 403
rows = _auth.get_mutation_priors()
return jsonify({"ok": True, "count": len(rows), "priors": rows,
"effective_date": _agg.EFFECTIVE_DATE.isoformat(),
"min_users": _agg.MIN_USERS})
@app.get("/api/admin/learn-signal")
def admin_learn_signal() -> Response:
"""What Turing is worst at, ranked by how often users hit it.
The Learn phase was built on `log_outcome` — a scientist synthesises a
design, measures it, logs the numbers. That is the honest loop and it
stays, but nobody has ever closed it (platform labs = 0). This reads
the signal that accrues whether or not the user does anything: every
run already persists its steers, tool failures, asks and abandons, and
in aggregate that has never been read.
Cross-user, so it inherits the mutation-commons privacy regime —
`field_report` is date-gated, k-anonymous, and emits no user text and
no tool arguments. Admin-token gated on top, because the underlying
scan uses service-role credentials to cross users at all.
?days=30 window (default 30, max 365)
?limit= runs to read (default 2000, max 5000)
?format=text for the pasteable version
"""
from dee.core import learn_signal as _ls
from dee.core.aggregate import AggregationGateError as _Gate
import datetime as _dt4
if not _admin_ok():
return jsonify({"ok": False, "error": "forbidden"}), 403
try:
days = max(1, min(int(request.args.get("days", 30)), 365))
except ValueError:
days = 30
try:
limit = max(1, min(int(request.args.get("limit", 2000)), 5000))
except ValueError:
limit = 2000
since = (_dt4.datetime.now(_dt4.timezone.utc)
- _dt4.timedelta(days=days)).isoformat()
runs = _auth.scan_agent_runs(since_iso=since, limit=limit)
try:
rep = _ls.field_report(runs)
except _Gate as exc:
return jsonify({"ok": False, "error": str(exc)}), 403
rep["window_days"] = days
rep["ok"] = True
if request.args.get("format") == "text":
return Response(_ls.as_text(rep), mimetype="text/plain")
return jsonify(rep)
@app.get("/api/admin/worked-examples")
def admin_worked_examples() -> Response:
"""Tool chains real users have completed cleanly.
The other half of /api/admin/learn-signal: that one reads the corpus
for what went wrong, this one for what went right. Same privacy
regime — tool names only, k-anonymous, date-gated. These are fed back
into the agent's system prompt, so this endpoint is also how you see
what it is currently being told.
"""
from dee.core import worked_examples as _we
from dee.core.aggregate import AggregationGateError as _Gate
if not _admin_ok():
return jsonify({"ok": False, "error": "forbidden"}), 403
try:
limit = max(1, min(int(request.args.get("limit", 400)), 5000))
except ValueError:
limit = 400
runs = _auth.scan_agent_runs(limit=limit)
try:
rep = _we.promote(runs)
except _Gate as exc:
return jsonify({"ok": False, "error": str(exc)}), 403
if request.args.get("format") == "prompt":
return Response(_we.as_prompt_section(rep) or
"(nothing has cleared the k-anonymity floor yet)",
mimetype="text/plain")
return jsonify(rep)
@app.get("/api/admin/cache-stats")
def admin_cache_stats() -> Response:
"""Resolution-cache hit rate — the number that says whether repeat
lookups are actually compounding."""
from dee.core import resolution_cache as _rc
if not _admin_ok():
return jsonify({"ok": False, "error": "forbidden"}), 403
return jsonify({"ok": True, "resolution_cache": _rc.stats()})
@app.post("/api/admin/run-benchmarks")
def admin_run_benchmarks() -> Response:
"""Produce REAL validation numbers — runs the live ESM-2 model already
loaded in this Space against the bundled public DMS fixtures
(dee/data/dms_fixtures/manifest.json) and writes dee/data/benchmarks.json.
This is the ONLY code path that may populate that file; the app never
hand-enters a number. Admin-token gated — this is a heavy call (loads
the model, scores thousands of variants across every bundled assay).
Provenance for the numbers this writes lives in the manifest itself
(citation, exact fitness column used, WT-residue cross-validation)."""
import datetime as _dt3
from dee.core import benchmark as _bm
from dee.core.dms_seed import parse_proteingym_csv as _parse_dms
if not _admin_ok():
return jsonify({"ok": False, "error": "forbidden"}), 403
body = request.get_json(force=True, silent=True) or {}
model = body.get("model", "small")
if model not in ("small", "medium", "large"):
model = "small"
fixtures_dir = Path(__file__).resolve().parent / "data" / "dms_fixtures"
manifest_path = fixtures_dir / "manifest.json"
if not manifest_path.exists():
return jsonify({"ok": False, "error": "no fixtures bundled"}), 404
try:
assays = json.loads(manifest_path.read_text(encoding="utf-8"))
except Exception: # noqa: BLE001
return jsonify({"ok": False, "error": "manifest unreadable"}), 500
scorer = _scoring.get_scorer(model)
results = []
failed = []
for a in assays:
name = a.get("name", a.get("protein", "?"))
try:
csv_path = fixtures_dir / a["csv"]
recs = _parse_dms(csv_path.read_text(encoding="utf-8"))
if not recs:
failed.append({"name": name, "error": "no parseable records in csv"})
continue
labels = [lab for (lab, _v) in recs]
measured = [v for (_lab, v) in recs]
scores_df = _scoring.score_guarded(scorer, a["sequence"])
predicted = _bm.predict_additive(scores_df, labels)
results.append(_bm.evaluate_dataset(
name, a.get("protein", ""),
predicted, measured, source=a.get("source", ""),
))
except Exception as exc: # noqa: BLE001 — one bad assay must not kill the run
logger.exception("benchmark assay failed: %s", name)
failed.append({"name": name, "error": f"{type(exc).__name__}: {exc}"})
out = {
"generated_at": _dt3.datetime.now(_dt3.timezone.utc).isoformat(),
"model": model,
"note": "Computed live on this Space against dee/data/dms_fixtures/ "
"(see manifest.json for provenance/citations). Never hand-entered.",
"summary": _bm.summarize(results),
"datasets": [r.as_dict() for r in results],
}
if failed:
out["failed"] = failed
out_path = Path(__file__).resolve().parent / "data" / "benchmarks.json"
try:
out_path.write_text(json.dumps(out, indent=2), encoding="utf-8")
except Exception: # noqa: BLE001 — the response below still carries the real result
logger.exception("could not persist benchmarks.json to disk")
out["ok"] = True
return jsonify(out)
@app.post("/api/admin/seed-commons")
def admin_seed_commons() -> Response:
"""Seed the Field Atlas from the SAME bundled public DMS fixtures —
ESM-free (pure aggregation of measured values, dee.core.dms_seed),
pushed to public.mutation_priors via this Space's own Supabase
credentials. Admin-token gated AND date-gated (the same Privacy/Terms
v2.0 effective-date + k-anonymity floor as user-data aggregation —
each bundled assay counts as one contributor)."""
from dee.core import aggregate as _agg
from dee.core.dms_seed import parse_proteingym_csv as _parse_dms
from dee.core.dms_seed import seed_rows as _seed_rows
if not _admin_ok():
return jsonify({"ok": False, "error": "forbidden"}), 403
fixtures_dir = Path(__file__).resolve().parent / "data" / "dms_fixtures"
manifest_path = fixtures_dir / "manifest.json"
if not manifest_path.exists():
return jsonify({"ok": False, "error": "no fixtures bundled"}), 404
try:
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
except Exception: # noqa: BLE001
return jsonify({"ok": False, "error": "manifest unreadable"}), 500
assays = []
failed = []
for a in manifest:
name = a.get("name", a.get("protein", "?"))
try:
csv_path = fixtures_dir / a["csv"]
recs = _parse_dms(csv_path.read_text(encoding="utf-8"))
if recs:
assays.append((name, recs))
else:
failed.append({"name": name, "error": "no parseable records in csv"})
except Exception as exc: # noqa: BLE001 — one bad assay must not kill the seed
logger.exception("seed-commons assay failed: %s", name)
failed.append({"name": name, "error": f"{type(exc).__name__}: {exc}"})
try:
rows = _seed_rows(assays) # enforces the effective-date gate
except _agg.AggregationGateError as exc:
return jsonify({"ok": False, "error": "gated", "detail": str(exc),
"effective_date": _agg.EFFECTIVE_DATE.isoformat()}), 423
# source="dms": replaces only the published-study rows, leaving any
# user-contributed commons untouched (migration 0018). _seed_rows
# already stamps each row source="dms".
result = _auth.replace_mutation_priors(rows, source="dms")
_GLOBAL_PRIOR_CACHE["data"] = None # bust so the next design sees it immediately
out = {"ok": bool(result.get("ok")), "substitutions": len(rows),
"contributing_assays": len(assays)}
if failed:
out["failed"] = failed
return jsonify(out)
@app.post("/api/admin/rebuild-priors")
def admin_rebuild_priors() -> Response:
"""Recompute the cross-user aggregate from all logged outcomes and store
it. Admin-token gated AND date-gated (refuses before the v2.0 effective
date). De-identified by construction; k-anonymity floor MIN_USERS."""
from dee.core import aggregate as _agg
if not _admin_ok():
return jsonify({"ok": False, "error": "forbidden"}), 403
# Opt-in richer aggregate: split each substitution by its ESM ΔLL bin.
# Heavy (one ESM pass per library) → off by default; the plain path is
# the tested default. Any ESM failure degrades to unbinned keys.
esm_bins = bool((request.get_json(force=True, silent=True) or {}).get("esm_bins"))
import datetime as _dt2
gate_open = _dt2.date.today() >= _agg.EFFECTIVE_DATE
ll_maps = None
if esm_bins and gate_open: # don't load ESM pre-gate
grouped_lib = _auth.list_all_de_outcomes_grouped_with_lib()
grouped = [(uid, meas) for uid, _lib, meas in grouped_lib]
try:
ll_maps = _compute_de_ll_maps(grouped_lib)
except Exception: # noqa: BLE001
logger.exception("esm-bin ll_maps failed; falling back to plain aggregate")
ll_maps = None
else:
grouped = _auth.list_all_de_outcomes_grouped()
try:
prior = _agg.build_priors(grouped, ll_maps=ll_maps) # enforces the 2026-07-08 gate
except _agg.AggregationGateError as exc:
return jsonify({"ok": False, "error": "gated", "detail": str(exc),
"effective_date": _agg.EFFECTIVE_DATE.isoformat()}), 423
rows = prior.to_rows("user")
result = _auth.replace_mutation_priors(rows, source="user")
_GLOBAL_PRIOR_CACHE["data"] = None # bust the scorer cache
return jsonify({"ok": bool(result.get("ok")),
"substitutions": len(rows),
"contributing_libraries": len(grouped),
"esm_binned": bool(esm_bins and ll_maps and any(ll_maps)),
"written": result.get("written", 0),
"priors": rows})
# ── Unified outcomes framework — generic capture + per-tool aggregate ─────
@app.post("/api/outcomes")
def outcomes_save() -> Response:
"""Save this user's wet-lab outcomes for any tool's saved design."""
auth = _auth.get_auth()
if auth.anonymous:
return jsonify({"ok": False, "error": "signin_required", "kind": "signin_required"}), 403
body = request.get_json(force=True, silent=True) or {}
tool = str(body.get("tool", "")).strip().lower()
design_id = str(body.get("design_id", "")).strip()
rows = body.get("outcomes")
if not isinstance(rows, list):
return jsonify({"ok": False, "error": "No outcomes provided."}), 400
result = _auth.save_outcomes(auth.user_id, tool, design_id, rows)
return jsonify(result), (200 if result.get("ok") else 400)
@app.get("/api/outcomes/<tool>/<design_id>")
def outcomes_list(tool: str, design_id: str) -> Response:
auth = _auth.get_auth()
if auth.anonymous:
return jsonify({"ok": False, "error": "signin_required", "kind": "signin_required"}), 403
return jsonify({"ok": True,
"outcomes": _auth.list_outcomes(auth.user_id, tool.lower(), design_id)})
@app.get("/api/admin/priors/<tool>")
def admin_priors(tool: str) -> Response:
from dee.core import outcomes as _O
if not _admin_ok():
return jsonify({"ok": False, "error": "forbidden"}), 403
rows = _auth.get_outcome_priors(tool.lower())
return jsonify({"ok": True, "tool": tool.lower(), "count": len(rows), "priors": rows,
"effective_date": _O.EFFECTIVE_DATE.isoformat(),
"registered_tools": _O.registered_tools()})
@app.post("/api/admin/rebuild-priors/<tool>")
def admin_rebuild_priors_tool(tool: str) -> Response:
"""Recompute one tool's cross-user aggregate from public.outcomes.
Admin-token + date gated. De-identified; k-anonymity per the ToolSpec."""
from dee.core import outcomes as _O
if not _admin_ok():
return jsonify({"ok": False, "error": "forbidden"}), 403
tool = tool.lower()
if _O.get_spec(tool) is None:
return jsonify({"ok": False, "error": "unknown_tool",
"registered_tools": _O.registered_tools()}), 404
grouped = _auth.list_all_outcomes_grouped(tool)
try:
prior = _O.build_priors(tool, grouped) # enforces the effective-date gate
except _O.AggregationGateError as exc:
return jsonify({"ok": False, "error": "gated", "detail": str(exc),
"effective_date": _O.EFFECTIVE_DATE.isoformat()}), 423
rows = prior.to_rows()
result = _auth.replace_outcome_priors(tool, rows)
return jsonify({"ok": bool(result.get("ok")), "tool": tool,
"keys": len(rows), "contributing_designs": len(grouped),
"written": result.get("written", 0), "priors": rows})
@app.post("/api/crispr/download")
def crispr_download() -> Response:
"""Return the CSV body for a fresh design. We re-design from the
sequence on download rather than caching per-job to avoid a
whole job-state layer for what's essentially a pure function
of the input. Same sign-in gate as /api/crispr/design."""
auth = _auth.get_auth()
if auth.anonymous:
return jsonify({"error": "signin_required"}), 403
body = request.get_json(force=True, silent=True) or {}
seq = (body.get("sequence") or "").strip()
if not seq:
return jsonify({"error": "missing 'sequence'"}), 400
if len(seq) > 1_000_000:
return jsonify({"error": "input_too_large"}), 400
try:
max_results = int(body.get("max_results", 50))
except (TypeError, ValueError):
max_results = 50
max_results = max(1, min(500, max_results))
enzyme = str(body.get("enzyme", "cas9")).lower()
if enzyme not in ("cas9", "cas12a"):
enzyme = "cas9"
# Phase 2B-1 — same optional context as /api/crispr/design so the
# downloaded XLSX matches what the user saw in the table.
target_organism = str(body.get("target_organism", "")).lower().strip()
gene_symbol = str(body.get("gene_symbol", "")).strip()
# Validate against the actual genome registry rather than a hardcoded
# list, so adding an organism to offtarget.GENOME_SOURCES enables it
# end to end. The old literal tuple silently rejected any new organism
# and fell back to "no genome search".
from dee.core.offtarget import GENOME_SOURCES as _GENOMES
if target_organism and target_organism not in _GENOMES:
target_organism = ""
if len(gene_symbol) > 32 or not re.match(r"^[A-Za-z0-9._-]*$", gene_symbol):
gene_symbol = ""
# Phase 2C-1 passthrough
from dee.core import crispr_cloning as _cc_dl
vector_id = str(body.get("vector_id", "")).lower().strip()
if vector_id and vector_id not in _cc_dl.CRISPR_VECTORS:
vector_id = ""
# Phase 3 (M1) passthrough so the XLSX matches the on-screen mode.
from dee.core import base_editor as _be_dl
mode = str(body.get("mode", "knockout")).lower()
if mode not in ("knockout", "base_edit"):
mode = "knockout"
base_editor = str(body.get("base_editor", "")).lower().strip()
if base_editor and _be_dl.get_base_editor(base_editor) is None:
base_editor = ""
# Export format. XLSX (default) is locale-safe for Excel users in
# ;-delimiter locales; CSV is the universal format for pipelines /
# pandas / R that some users explicitly need.
fmt = str(body.get("format", "xlsx")).lower().strip()
if fmt not in ("xlsx", "csv"):
fmt = "xlsx"
try:
guides = _find_crispr_guides(
seq, enzyme=enzyme, max_results=max_results,
target_organism=target_organism, gene_symbol=gene_symbol,
vector_id=vector_id, mode=mode, base_editor=base_editor,
)
except ValueError as exc:
return jsonify({"error": str(exc)}), 400
rows = _crispr_csv_rows(guides)
# ── CSV export ──────────────────────────────────────────────────
# Comma-delimited, RFC-4180 quoting (csv module quotes any field
# containing a comma/quote/newline — e.g. KO reasoning — so it stays
# in one column even when opened in a ;-delimiter locale). A UTF-8
# BOM makes Excel (any locale) detect the encoding correctly.
if fmt == "csv":
import csv as _csv
from io import StringIO as _SIO
from flask import make_response as _mr_csv
sio = _SIO()
writer = _csv.writer(sio, quoting=_csv.QUOTE_MINIMAL, lineterminator="\r\n")
for r in rows:
writer.writerow(r)
# Lead with a UTF-8 BOM so Excel auto-detects the encoding.
body_bytes = ("" + sio.getvalue()).encode("utf-8")
resp = _mr_csv(body_bytes)
resp.headers["Content-Type"] = "text/csv; charset=utf-8"
resp.headers["Content-Disposition"] = (
'attachment; filename="turingdna_crispr_guides.csv"'
)
return resp
# ── XLSX export (default) ───────────────────────────────────────
# openpyxl produces a true Excel file with explicit column types;
# opens cleanly in Excel/Numbers/Google Sheets regardless of locale.
from io import BytesIO as _BIO
from openpyxl import Workbook as _WB
from openpyxl.styles import Font as _Font, Alignment as _Align, PatternFill as _Fill
wb = _WB()
ws = wb.active
ws.title = "Guides"
header_row, data_rows = rows[0], rows[1:]
ws.append(header_row)
# Numeric-typed cells where appropriate so Excel sorts + formats them
# correctly (sort by Composite is the most useful thing the user
# will do in Excel; force scientific notation off, keep 3 decimals).
# Header → indices for the cells we want numeric.
header_index = {name: i for i, name in enumerate(header_row)}
num_cols = {
header_index.get("Rank"),
header_index.get("Position"),
header_index.get("Composite score"),
header_index.get("On-target"),
header_index.get("Max self-CFD"),
header_index.get("Off-target count"),
header_index.get("KO efficacy"),
header_index.get("GC %"),
# Phase 2A numeric columns
header_index.get("Frameshift %"),
header_index.get("Top outcome dominance %"),
# Phase 2B-1 numeric columns
header_index.get("Genome off-target count"),
header_index.get("Genome off-target max CFD"),
# Phase 3 (M1) numeric column
header_index.get("Editability"),
}
num_cols.discard(None)
for r in data_rows:
converted = []
for col_idx, cell in enumerate(r):
if col_idx in num_cols:
try:
converted.append(float(cell) if "." in str(cell) else int(cell))
except (TypeError, ValueError):
converted.append(cell)
else:
converted.append(cell)
ws.append(converted)
# Header styling — bold + light paper-tone fill so the header
# row reads as the table caption, not data.
bold = _Font(bold=True, color="0A0A0A")
paper_fill = _Fill(start_color="EFECE5", end_color="EFECE5", fill_type="solid")
for cell in ws[1]:
cell.font = bold
cell.fill = paper_fill
cell.alignment = _Align(vertical="center")
# Freeze the header row so it sticks when the user scrolls down a
# large guide list. Also widen columns to something sensible —
# Excel's autosize logic is browser-dependent.
ws.freeze_panes = "A2"
widths = {
"Rank": 6, "Enzyme": 8, "Strand": 8, "Position": 10,
"Spacer": 28, "PAM": 8, "Target context (5'→3')": 36,
"Composite score": 16, "On-target": 12, "Max self-CFD": 14,
"Off-target count": 16, "KO efficacy": 12, "KO reasoning": 56,
"GC %": 8, "Flags": 18,
# Phase 2A
"Top predicted indel": 22, "Frameshift %": 14, "Top outcome dominance %": 22,
"Indel distribution": 64, "Base-editor compatibility": 30,
# Phase 2B-1
"Genome searched": 14, "Genome off-target count": 18,
"Genome off-target max CFD": 22, "Top genome off-target": 44,
"Exon context": 32,
# Phase 2C-1
"Cloning vector": 36, "Addgene #": 12, "Cloning enzyme": 14,
"Sense oligo (5'→3')": 32, "Antisense oligo (5'→3')": 32,
}
from openpyxl.utils import get_column_letter as _col_letter
for i, name in enumerate(header_row, start=1):
ws.column_dimensions[_col_letter(i)].width = widths.get(name, 14)
bio = _BIO()
wb.save(bio)
body_bytes = bio.getvalue()
from flask import make_response as _mr
resp = _mr(body_bytes)
resp.headers["Content-Type"] = (
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
)
resp.headers["Content-Disposition"] = (
'attachment; filename="turingdna_crispr_guides.xlsx"'
)
return resp
# NOTE: /api/library, /api/library/<filename> (GET + DELETE), and
# /api/shutdown were removed on 2026-05-26 per SECURITY_AUDIT findings
# C2 + C3.
#
# The library endpoints listed every CSV under the Space-shared
# ~/.dee/output/ directory and let anyone download or delete files
# by guessing filenames — they pre-dated the per-user storage we now
# ship in Supabase. The dashboard (planned task #98) will be the
# replacement, querying public.libraries via Supabase RLS-gated reads.
#
# /api/shutdown was an unauthenticated DoS vector — POSTing to it
# killed the Flask process. Reasonable for the original macOS .app
# use case, dangerous in the hosted multi-tenant deployment. The
# HF Space's own restart mechanism replaces this for ops needs.
return app
# ----------------------------------------------------------------- helpers
# NOTE: the ASSISTANT block (gradio_client connection to
# winter4000/turingdna-assistant, _AssistantUnavailable, _get_assistant_client,
# _format_history_into_message, _call_assistant) was removed on 2026-05-25.
# Full implementation lives in _llm_backup_2026-05-25/server/server.py.pre-strip
# and can be restored when the assistant is ready to ship again.
_VALID_MODELS = {"small", "medium", "large"}
_VALID_HOSTS = {"e_coli", "ecoli", "yeast", "s_cerevisiae", "human", "h_sapiens"}
_VALID_DEVICES = {None, "", "cpu", "cuda", "mps", "auto"}
_VALID_QUANT = {None, "", "none", "int8", "int4", "fp16", "bf16"}
# ─── ESM-2 scorer cache + CPU model gate ──────────────────────────────────
# Moved to dee/core/scoring.py (2026-07-10) so the Turing agent's
# design_variant_library tool can share the SAME cached model instance and
# scoring concurrency guard, without dee/core/ needing to import this file
# (which would be circular — this file already imports dee/core/agent.py).
from dee.core import scoring as _scoring
# DNA-level tiers (Evo 2, via Modal) — same "/api/models reports what's
# actually runnable" contract, separate id-space (see the models() route).
from dee.core import compiler as _compiler
from dee.core import compiler_validation as _compiler_validation
from dee.core import prime_editor as _prime_editor
from dee.core import certificate as _certificate
from dee.core import assay as _assay
from dee.core import measurements as _meas
from dee.core import uplift as _uplift
from dee.core import variant_resolve as _variant_resolve
from dee.core import dna_scoring as _dna_scoring
def _validate_run_settings(raw: Dict[str, Any]) -> Dict[str, Any]:
"""Normalize and bounds-check the user-supplied run settings.
Returns a clean dict ready to hand to ``_run_pipeline``. Raises
``ValueError`` (the route turns this into HTTP 400) on any invalid
value — better to fail loudly at submission than have the worker
thread crash silently or produce a degenerate library.
The original code did ``int(settings.get("k", 30))`` with no bounds,
so a malformed request (k=-1, max_mutations=99999, percentile=200,
host="postgres") would either crash the SearchConfig with an
InternalError or return a 200 with a useless result. The bounds here
are intentionally generous — the goal is to catch garbage, not
second-guess the user's intent.
"""
def _num(field: str, default, lo, hi, kind):
value = raw.get(field, default)
try:
value = kind(value)
except (TypeError, ValueError):
raise ValueError(
f"{field!r} must be a {kind.__name__}; got {raw.get(field)!r}.",
)
if value < lo or value > hi:
raise ValueError(
f"{field!r}={value} is outside the allowed range [{lo}, {hi}].",
)
return value
model = (raw.get("model") or "small")
if model not in _VALID_MODELS:
raise ValueError(f"model must be one of {sorted(_VALID_MODELS)}; got {model!r}.")
host = (raw.get("host") or "e_coli")
if host not in _VALID_HOSTS:
raise ValueError(f"host must be one of {sorted(_VALID_HOSTS)}; got {host!r}.")
device = raw.get("device")
if device not in _VALID_DEVICES:
raise ValueError(f"device must be one of {sorted(d or '<auto>' for d in _VALID_DEVICES)}; got {device!r}.")
quant = raw.get("quantization")
if quant not in _VALID_QUANT:
raise ValueError(f"quantization must be one of {sorted(q or '<none>' for q in _VALID_QUANT)}; got {quant!r}.")
# EVERY RUN GETS A SEED, whether or not one was asked for.
#
# The search is simulated annealing over restarts, so an unseeded run is
# not repeatable — the same sequence and the same settings give a slightly
# different library each time, and an outside audit reported exactly that:
# ranks moving by one or two between identical runs. The plumbing to fix
# it already existed end to end (main.py --seed, SearchConfig(seed=...)),
# and nothing ever supplied a value.
#
# Minting one here rather than adding a field to the form is the whole
# point: an optional seed only helps the person who thought to set it
# BEFORE the run they now want to repeat. A seed that is always recorded
# makes every run repeatable after the fact, including the one that
# produced a result worth chasing.
seed = raw.get("seed")
if seed in (None, ""):
import secrets as _secrets
seed = _secrets.randbelow(2 ** 31)
if seed not in (None, ""):
try:
seed = int(seed)
except (TypeError, ValueError):
raise ValueError(f"seed must be an integer or null; got {seed!r}.")
percentile = _num("percentile", 85.0, 0.0, 100.0, float)
k = _num("k", 30, 1, 500, int)
min_mutations = _num("min_mutations", 2, 1, 50, int)
max_mutations = _num("max_mutations", 5, 1, 50, int)
restarts = _num("restarts", 8, 1, 100, int)
steps = _num("steps", 1200, 50, 50000, int)
if min_mutations > max_mutations:
raise ValueError(
f"min_mutations ({min_mutations}) must be ≤ max_mutations ({max_mutations}).",
)
# CPU-tier safety: downgrade GPU-class models + drop a 'cuda'/quant request
# so the run can't OOM the shared instance. Reflected in the returned dict
# so analytics + the methods note describe what actually ran.
model = _scoring.effective_model(model)
if not _scoring.gpu_available():
device = None
quant = None
return {
"model": model,
"percentile": percentile,
"k": k,
"min_mutations": min_mutations,
"max_mutations": max_mutations,
"restarts": restarts,
"steps": steps,
"host": host,
"seed": seed,
"device": device,
"quantization": quant,
}
def _safe_decode_text(path: Path) -> str:
"""Decode an uploaded sequence file as text without silently losing bytes.
The previous code used ``path.read_text(errors="ignore")`` in the
ORF-fallback branch, which silently drops any byte that isn't valid
UTF-8 — a corrupted upload could yield a different ORF than the user
intended. This helper tries common encodings in order of likelihood for
a sequence file:
1. UTF-8 with BOM (UTF-8-SIG) — Windows Notepad's default for files
containing any non-ASCII char.
2. Plain UTF-8 — the modern default everywhere else.
3. UTF-16 LE/BE — older Windows tooling, especially when the user
exported from Excel as "Unicode Text".
4. Latin-1 — never raises, accepts all 256 byte values. Last-resort
fallback for genuinely odd inputs (will pass through whatever
bytes are there; downstream alphabet validation rejects them).
Sequences are pure ASCII anyway (ACGTU and amino-acid letters), so for
every legitimate input one of the early encodings produces a usable
string. The point of the chain is to never SILENTLY drop a byte.
"""
raw = path.read_bytes()
for enc in ("utf-8-sig", "utf-8", "utf-16", "utf-16-le", "utf-16-be"):
try:
return raw.decode(enc)
except UnicodeDecodeError:
continue
# latin-1 cannot fail; this branch is the universal fallback.
return raw.decode("latin-1")
def _materialize_input() -> tuple[Path, str]:
"""Pull the request's sequence into a temp file we can hand to parse_input.
For uploads we also impose a generous-but-finite size cap (8 MiB) so a
runaway client or accidental binary-file drop can't blow up the server.
Real protein CDS uploads top out under ~50 kB even for the largest
genes; 8 MiB covers full plasmid GenBank files with comfortable margin.
"""
MAX_UPLOAD_BYTES = 8 * 1024 * 1024 # 8 MiB
if "file" in request.files:
file = request.files["file"]
suffix = Path(file.filename or "").suffix or ".txt"
tmp = tempfile.NamedTemporaryFile(
delete=False, suffix=suffix, dir=str(STATE_DIR)
)
file.save(tmp.name)
tmp.close()
if Path(tmp.name).stat().st_size > MAX_UPLOAD_BYTES:
Path(tmp.name).unlink(missing_ok=True)
raise SequenceValidationError(
f"Uploaded file exceeds the {MAX_UPLOAD_BYTES // (1024 * 1024)} MiB limit. "
"Trim to just the gene of interest before uploading.",
)
return Path(tmp.name), file.filename or "uploaded"
body = request.get_json(force=True, silent=True) or {}
text = body.get("text", "").strip()
if not text:
raise SequenceValidationError("No sequence provided.")
if len(text.encode("utf-8")) > MAX_UPLOAD_BYTES:
raise SequenceValidationError(
f"Pasted text exceeds the {MAX_UPLOAD_BYTES // (1024 * 1024)} MiB limit.",
)
tmp = tempfile.NamedTemporaryFile(
delete=False, suffix=".txt", mode="w", encoding="utf-8", dir=str(STATE_DIR)
)
tmp.write(text)
tmp.close()
return Path(tmp.name), body.get("name") or "pasted"
def _humanize_paste_name(original_name: str) -> str:
"""A clean display label for a raw paste/upload that carries no real name."""
stem = Path(original_name or "").stem.strip()
if stem and stem.lower() not in ("pasted", "uploaded"):
return stem # a real uploaded filename, e.g. "TP53_exon2"
return "Pasted sequence"
def _clean_record_identifier(record, path: Path, original_name: str) -> None:
"""When a raw (non-FASTA) sequence has no header, parse_input falls back to
the file stem as the identifier — and for our temp files that stem is a
random 'tmpXXXXXXXX'. That leaked through as the displayed identifier
("random letters"). Replace it in place with a human label so it never
surfaces. FASTA headers / GenBank LOCUS names are untouched (their
identifier never equals the temp stem)."""
try:
if record.identifier and record.identifier == path.stem:
record.identifier = _humanize_paste_name(original_name)
except Exception: # noqa: BLE001 — display nicety must never break parsing
pass
def _summarize(path: Path, original_name: str) -> tuple[Dict[str, Any], Dict[str, Any]]:
"""Return (JSON-friendly summary, extra session payload to cache).
Three flow modes:
1. Structured file (.dna/.gb/...) → list annotated CDS features.
2. Raw input parses cleanly as a single CDS or protein → return preview.
3. Raw input fails CDS validation (e.g. user pasted a whole plasmid) →
fall back to a 6-frame ORF scan and return ORF choices.
"""
suffix = path.suffix.lower()
# Mode 1 — annotated structured file.
if suffix in {".dna", ".gb", ".gbk", ".genbank", ".embl"}:
features = list_cds_features(path)
cds_options = [
{"label": label, "length_nt": length} for label, length in features
]
record = parse_input(path, require_start=False, require_stop=False)
_clean_record_identifier(record, path, original_name)
return (
{
"original_name": original_name,
"detected_kind": "plasmid",
"identifier": record.identifier,
"protein_length": len(record.protein),
"protein_preview": record.protein[:80]
+ ("…" if len(record.protein) > 80 else ""),
"cds_options": cds_options,
},
# cds_options carried into the session so /api/run can enforce
# the "pick one explicitly" check (audit M7).
{"mode": "structured_file", "cds_options": cds_options},
)
# Mode 2 — try clean single-CDS / protein parse.
try:
record = parse_input(path, require_start=False, require_stop=False)
_clean_record_identifier(record, path, original_name)
return (
{
"original_name": original_name,
"detected_kind": "protein" if not record.dna else "dna",
"identifier": record.identifier,
"protein_length": len(record.protein),
"protein_preview": record.protein[:80]
+ ("…" if len(record.protein) > 80 else ""),
"cds_options": None,
},
{"mode": "single"},
)
except SequenceValidationError as clean_exc:
# Mode 3 — fall back to 6-frame ORF discovery for raw DNA. Use the
# safe-decode chain instead of read_text(errors="ignore") so a UTF-16
# or BOM-prefixed file doesn't silently lose every other byte and
# surface a different ORF than the user pasted.
text = _safe_decode_text(path)
orfs = find_orfs_in_dna(text)
if not orfs:
# Not recoverable; bubble the original parse error up.
raise clean_exc
cds_options = [
{
"label": o.label,
"length_nt": len(o.dna),
"frame": o.frame,
"protein_length": len(o.protein),
}
for o in orfs
]
return (
{
"original_name": original_name,
"detected_kind": "multi_orf",
"identifier": f"{_humanize_paste_name(original_name)} (no annotation)",
"protein_length": 0,
"protein_preview": (
f"{len(orfs)} ORFs detected across 6 reading frames "
"(no annotated CDS features). Pick one below."
),
"cds_options": cds_options,
},
# Same here: carry cds_options into the session so the
# multi-ORF chooser is enforced server-side too (audit M7).
{
"mode": "orf",
"orfs": {o.label: o.dna for o in orfs},
"cds_options": cds_options,
},
)
_DE_ROUND2_MAX_AA = 500 # sync round-2 length cap (bounds ESM score time on free CPU)
# ── Cross-user aggregate prior (the data moat) ──────────────────────────
# Admin access is gated on a shared secret (env TURINGDNA_ADMIN_TOKEN). If the
# env var is unset, the admin endpoints are CLOSED (403) — no admin by default.
_GLOBAL_PRIOR_CACHE: Dict[str, Any] = {"data": None, "ts": 0.0}
_GLOBAL_PRIOR_TTL_S = 600.0
def _admin_ok() -> bool:
token = os.environ.get("TURINGDNA_ADMIN_TOKEN", "")
if not token:
return False
import hmac
supplied = request.headers.get("X-Admin-Token", "")
return bool(supplied) and hmac.compare_digest(supplied, token)
def _load_global_prior():
"""Load + cache the stored aggregate prior. Returns a GlobalPrior (possibly
empty). Empty → apply_global_prior is a no-op, so this is inert until the
first post-effective-date rebuild populates public.mutation_priors."""
import time as _t
from dee.core import aggregate as _agg
now = _t.time()
if _GLOBAL_PRIOR_CACHE["data"] is not None and (now - _GLOBAL_PRIOR_CACHE["ts"]) < _GLOBAL_PRIOR_TTL_S:
return _GLOBAL_PRIOR_CACHE["data"]
try:
prior = _agg.GlobalPrior.from_rows(_auth.get_mutation_priors())
except Exception: # noqa: BLE001
prior = _agg.GlobalPrior(effects={}, n_users={}, n_obs={})
_GLOBAL_PRIOR_CACHE["data"] = prior
_GLOBAL_PRIOR_CACHE["ts"] = now
return prior
# Per-tool unified-outcomes prior cache (CRISPR, primer, …). Empty until the
# tool's post-effective-date rebuild populates public.outcome_priors → inert.
_TOOL_PRIOR_CACHE: Dict[str, Any] = {}
def _load_tool_prior(tool: str):
"""Cached outcomes.Prior for a tool from public.outcome_priors."""
import time as _t
from dee.core import outcomes as _O
now = _t.time()
ent = _TOOL_PRIOR_CACHE.get(tool)
if ent is not None and (now - ent["ts"]) < _GLOBAL_PRIOR_TTL_S:
return ent["data"]
spec = _O.get_spec(tool)
weight = spec.weight if spec else 0.2
try:
prior = _O.Prior.from_rows(tool, _auth.get_outcome_priors(tool), weight=weight)
except Exception: # noqa: BLE001
prior = _O.Prior(tool=tool, effects={}, n_users={}, n_obs={}, weight=weight)
_TOOL_PRIOR_CACHE[tool] = {"data": prior, "ts": now}
return prior
# Rebuilding the aggregate priors used to require a human to POST to
# /api/admin/rebuild-priors by hand with TURINGDNA_ADMIN_TOKEN — meaning
# "the model compounds on aggregated results" (landing page, §3) was true of
# the mechanism's design but not actually happening unless someone remembered
# to trigger it. This background loop makes it self-sustaining instead.
# Once a day is plenty (outcomes accrue slowly, and a rebuild scans every
# logged outcome — not something to run on every request). Safe to run from
# a single thread: Dockerfile pins gunicorn to --workers 1, so there is
# exactly one of these loops per running instance, never a rebuild race.
_PRIOR_REBUILD_INTERVAL_S = 24 * 3600.0
def _rebuild_all_priors_once() -> None:
"""One rebuild pass: the global DE prior, then every registered per-tool
prior. Each is independent and best-effort — one failing (or still being
before the Privacy/Terms v2.0 effective date, harmless) never blocks the
others."""
from dee.core import aggregate as _agg
from dee.core import outcomes as _O
try:
grouped = _auth.list_all_de_outcomes_grouped()
rows = _agg.build_priors(grouped).to_rows("user")
_auth.replace_mutation_priors(rows, source="user")
_GLOBAL_PRIOR_CACHE["data"] = None # this route's own cache
_agg.bust_cached_global_prior() # the shared one design_variant_library (chat) reads
logger.info("prior rebuild: de -> %d substitutions from %d libraries", len(rows), len(grouped))
except _agg.AggregationGateError:
pass # before the effective date — expected, not an error
except Exception: # noqa: BLE001 — one bad cycle must not kill the loop
logger.exception("prior rebuild failed for de")
for tool in _O.registered_tools():
try:
grouped = _auth.list_all_outcomes_grouped(tool)
rows = _O.build_priors(tool, grouped).to_rows()
_auth.replace_outcome_priors(tool, rows)
_TOOL_PRIOR_CACHE.pop(tool, None) # this route's own cache
_O.bust_cached_tool_prior(tool) # the shared one chat tools read
logger.info("prior rebuild: %s -> %d keys from %d designs", tool, len(rows), len(grouped))
except _O.AggregationGateError:
pass
except Exception: # noqa: BLE001
logger.exception("prior rebuild failed for %s", tool)
def _start_prior_rebuild_scheduler() -> None:
"""Daemon thread: first rebuild shortly after startup (so a fresh deploy
doesn't wait a full day for the first pass), then once every
_PRIOR_REBUILD_INTERVAL_S. Mirrors the off-target prewarm pattern in
build_app() below — fire-and-forget, never blocks request handling."""
import threading
import time as _t
def _loop() -> None:
_t.sleep(30.0) # let the rest of startup (prewarm, etc.) settle first
while True:
_rebuild_all_priors_once()
_t.sleep(_PRIOR_REBUILD_INTERVAL_S)
threading.Thread(target=_loop, name="prior-rebuild", daemon=True).start()
def _compute_de_ll_maps(grouped_with_lib):
"""Best-effort {single-site label: ESM ΔLL} per library, for the richer
(substitution × ESM-bin) DE aggregate. HEAVY (one ESM pass per distinct
library) → only the opt-in rebuild calls it. Any failure for a library
yields None for that unit (its keys stay unbinned). Returns a list parallel
to grouped_with_lib."""
cache: Dict[str, Any] = {}
scorer = None
maps = []
for _uid, lib, _rows in grouped_with_lib:
if lib in cache:
maps.append(cache[lib]); continue
m = None
try:
wt = _auth.get_library_wt(lib)
if (wt and re.match(r"^[ACDEFGHIKLMNPQRSTVWY*]+$", wt)
and len(wt) <= _DE_ROUND2_MAX_AA):
if scorer is None:
scorer = _scoring.get_scorer("small")
df = _scoring.score_guarded(scorer, wt)
m = {f"{r.wt_aa}{int(r.position) + 1}{r.mut_aa}": float(r.delta_ll)
for r in df.itertuples(index=False)}
except Exception: # noqa: BLE001
logger.exception("ll-map compute failed for library %s", lib)
m = None
cache[lib] = m
maps.append(m)
return maps
# Below this the "top 10%" is one or two variants and its precision is a
# coin flip printed as a percentage. 30 makes the top set at least 3.
_IMPORT_TDP_MIN_N = 30
_POOL_DELTAS_MAX = 40 # cap the "what changed" payload — biggest movers only
def _de_round2_library(wt_protein: str, settings: Dict[str, Any],
measurements: list) -> tuple[list, Dict[str, Any]]:
"""Active-learning round 2 (Design→Build→Test→Learn). Score the WT, blend
the cross-user field-wide prior (parity with round 1), fit the personal
surrogate on the user's measured variants, re-rank the single-site pool by
the learned acquisition score, evolve, and assemble the same variant table
shape as a normal run. Returns (variant_rows, surrogate_info) — the latter
now also carries `pool_deltas` (per-mutation before/after + plain-language
reason, sorted by round-2 score) and `global_prior` (whether the field-wide
aggregate contributed), so the frontend can show a genuine before/after
re-rank instead of a silent number change."""
from dee.core import active_learning as _al
from dee.core import aggregate as _agg
from dee.optimizer.search import Mutation as _Mut
scorer = _scoring.get_scorer(
settings.get("model", "small"),
settings.get("device"),
settings.get("quantization"),
)
scores_df = _scoring.score_guarded(scorer, wt_protein)
pool_df = top_percentile_pool(scores_df, percentile=float(settings.get("percentile", 85.0)))
muts = [_Mut(int(r.position), str(r.wt_aa), str(r.mut_aa), float(r.delta_ll))
for r in pool_df.itertuples(index=False)]
# Same field-wide blend round 1 gets (soft nudge by substitution TYPE only,
# no-op until a post-effective-date rebuild has populated the aggregate) —
# applied BEFORE the personal surrogate so the two signals compose the same
# way in both rounds: ESM-2 prior → +field-wide nudge → +your own data.
global_prior_info = {"applied": False, "substitution_types": 0}
try:
_gp = _load_global_prior()
if _gp and _gp.effects:
muts = _agg.apply_global_prior(muts, _gp)
global_prior_info = {"applied": True, "substitution_types": len(_gp.effects)}
except Exception: # noqa: BLE001 — global-prior blend is a nice-to-have, never fatal
logger.exception("global-prior blend skipped (round 2)")
# Fit the surrogate on the (possibly field-nudged) single-site pool + the
# user's results, then overwrite the pool's per-mutation score with the
# learned acquisition (additive ⇒ evolve() consumes it unchanged).
surrogate = _al.fit_surrogate(muts, measurements)
baseline = {(m.position, m.mut_aa): float(m.delta_ll) for m in muts}
adj_df = pool_df.copy()
adj_df["delta_ll"] = [
surrogate.adjusted.get((int(r.position), str(r.mut_aa)), float(r.delta_ll))
for r in pool_df.itertuples(index=False)
]
# "What Turing learned" — every pool mutation's prior vs adjusted score
# plus a plain-language reason, sorted by the new score so the biggest
# movers (and the ones round 2 now favors most) lead. Cheap: template
# text only, no LLM call, and it's a straight decomposition of numbers
# fit_surrogate already computed — nothing here can fabricate a reason
# the data doesn't support.
pool_deltas = []
for m in muts:
key = (m.position, m.mut_aa)
prior_score = baseline.get(key, float(m.delta_ll))
adjusted_score = surrogate.adjusted.get(key, prior_score)
comp = surrogate.components.get(key, {"prior": prior_score, "learned": 0.0,
"explore": 0.0, "n_measured": 0})
pool_deltas.append({
"label": f"{m.wt_aa}{m.position + 1}{m.mut_aa}",
"prior_score": round(prior_score, 4),
"adjusted_score": round(adjusted_score, 4),
"delta": round(adjusted_score - prior_score, 4),
"n_measured": int(comp.get("n_measured", 0)),
"reason": _al.narrate(comp.get("prior", prior_score), comp.get("learned", 0.0),
comp.get("explore", 0.0), int(comp.get("n_measured", 0))),
})
pool_deltas.sort(key=lambda d: d["adjusted_score"], reverse=True)
pool_deltas = pool_deltas[:_POOL_DELTAS_MAX]
variants = evolve(adj_df, SearchConfig(
k=int(settings.get("k", 30)),
max_mutations=int(settings.get("max_mutations", 5)),
min_mutations=int(settings.get("min_mutations", 2)),
n_restarts=int(settings.get("restarts", 8)),
steps_per_restart=int(settings.get("steps", 1200)),
seed=settings.get("seed"),
))
df = variants_to_dataframe(
wt_protein, variants,
host=settings.get("host", "e_coli"),
forbidden_sites=DEFAULT_FORBIDDEN_SITES,
)
rows2 = df.to_dict(orient="records")
# Glass-box certainty on the round-2 rows too (free, reuses scores_df).
try:
from dee.core import confidence as _conf
_conf.attach_confidence(rows2, _conf.position_confidence_from_scores(scores_df))
except Exception: # noqa: BLE001
logger.exception("confidence annotation skipped (round 2)")
# ── Epistasis in the loop ────────────────────────────────────────────
# Round 2 doesn't just re-rank on your data — it re-ranks on predicted
# INTERACTION too. We score the top variants' full sequences with ESM-2
# and demote combinations it thinks will antagonize (clashers sink,
# cooperative combos rise), so the delivered library already avoids the
# duds. Honest scope: this re-ranks the SA's output by ESM-2's contextual
# opinion; the additive search itself doesn't explore epistatically.
# Bounded (top 8, a forward pass each) to stay responsive on the free CPU.
epistasis_info = {"applied": False, "n_clash": 0, "n_analyzed": 0}
try:
from dee.core import epistasis as _epi
analyzed = _epi.analyze_library(wt_protein, rows2, scorer.masked_log_probs,
max_variants=8)
by_mut = {a.mutations: a for a in analyzed}
for r in rows2:
base_fit = float(r.get("Predicted_Fitness_Score", 0.0) or 0.0)
a = by_mut.get(str(r.get("Mutations_AA", "") or ""))
if a is not None:
r["Epistasis_Verdict"] = a.verdict
r["Epistasis_Signed"] = round(a.epistasis_signed, 4)
# Adjust by the EXCESS over a random same-size recombination,
# never by the raw Σshift. Σshift grows with the number of
# mutations whether or not anything interacts, so adding it
# here handed every large variant a bonus and then re-sorted
# the library by it — changing which variants a user is told
# to build, on the strength of their size. excess_signed is
# 0.0 for an uncalibrated variant, leaving its rank untouched.
r["Epistasis_Excess"] = round(a.excess_signed, 4)
r["Interaction_Adjusted_Fitness"] = round(base_fit + a.excess_signed, 4)
else:
r["Epistasis_Verdict"] = ""
r["Epistasis_Signed"] = 0.0
r["Epistasis_Excess"] = 0.0
r["Interaction_Adjusted_Fitness"] = round(base_fit, 4)
if analyzed:
# Re-rank the delivered library by the interaction-adjusted fitness,
# then renumber so the visible ordering reflects it.
rows2.sort(key=lambda r: r.get("Interaction_Adjusted_Fitness", 0.0), reverse=True)
for i, r in enumerate(rows2, start=1):
r["Variant_ID"] = f"V{i:04d}"
n_clash = sum(1 for a in analyzed if a.verdict == "clash")
epistasis_info = {"applied": True, "n_clash": n_clash, "n_analyzed": len(analyzed)}
except Exception: # noqa: BLE001 — interaction re-rank is additive, never fatal
logger.exception("epistasis re-rank skipped (round 2)")
return rows2, {
"learned": surrogate.learned,
"n_train": surrogate.n_train,
"w_prior": round(surrogate.w_prior, 3),
"n_effects": surrogate.n_effects,
"note": surrogate.note,
"pool_deltas": pool_deltas,
"global_prior": global_prior_info,
"epistasis": epistasis_info,
}
def _run_pipeline(
job: JobState,
input_path: Path,
original_name: str,
cds_feature: Optional[str],
settings: Dict[str, Any],
) -> None:
"""Background worker. Updates ``job`` in place as it makes progress."""
try:
job.status = "parsing"
job.message = "Reading sequence…"
job.progress = 0.05
record = parse_input(
input_path,
require_start=False,
require_stop=False,
cds_feature=cds_feature,
)
_clean_record_identifier(record, input_path, original_name)
job.wt_identifier = record.identifier
job.wt_protein = record.protein
job.settings_used = {
"model": settings.get("model", "small"),
"percentile": float(settings.get("percentile", 85.0)),
"k": int(settings.get("k", 30)),
"min_mutations": int(settings.get("min_mutations", 2)),
"max_mutations": int(settings.get("max_mutations", 5)),
"restarts": int(settings.get("restarts", 8)),
"steps_per_restart": int(settings.get("steps", 1200)),
"host": settings.get("host", "e_coli"),
"seed": settings.get("seed"),
"device": settings.get("device") or "cpu",
"cds_feature": cds_feature,
}
job.status = "scoring"
job.message = (
f"Scoring {len(record.protein)} residues × 19 substitutions with ESM-2…"
)
job.progress = 0.15
scorer = _scoring.get_scorer(
settings.get("model", "small"),
settings.get("device"),
settings.get("quantization"),
)
scores_df = _scoring.score_guarded(scorer, record.protein)
job.progress = 0.55
pool = top_percentile_pool(scores_df, percentile=float(settings.get("percentile", 85.0)))
job.message = f"Filtered to {len(pool)} high-impact single-site mutations."
job.progress = 0.60
# Soft-blend the cross-user AGGREGATE prior into round-1 scores: every
# design benefits from the field's pooled wet-lab results. No-op until a
# post-effective-date rebuild populates public.mutation_priors (so this
# is inert before 2026-07-08). Individual designs stay ESM-driven; this
# is a small nudge by amino-acid substitution type only.
try:
_gp = _load_global_prior()
if _gp and _gp.effects:
_w = 0.3
pool = pool.copy()
pool["delta_ll"] = [
float(r.delta_ll) + _w * _gp.effects.get(
(str(r.wt_aa).upper(), str(r.mut_aa).upper()), 0.0)
for r in pool.itertuples(index=False)
]
job.message = (f"Filtered to {len(pool)} mutations; blended field-wide "
f"priors ({len(_gp.effects)} substitution types).")
# job.message gets overwritten by later status updates (searching,
# done, ...), so by the time the job finishes this fact would
# otherwise be lost — record it structurally so /api/result can
# still tell the user "this was informed by other labs" after
# the run is done, not just mid-flight in a progress string.
job.global_prior_info = {"applied": True, "substitution_types": len(_gp.effects)}
except Exception: # noqa: BLE001
logger.exception("global-prior blend skipped")
job.status = "searching"
job.message = (
f"Optimizing multi-mutants (K={int(settings.get('k', 30))}, "
f"max {int(settings.get('max_mutations', 5))} subs)…"
)
variants = evolve(
pool,
SearchConfig(
k=int(settings.get("k", 30)),
max_mutations=int(settings.get("max_mutations", 5)),
min_mutations=int(settings.get("min_mutations", 2)),
n_restarts=int(settings.get("restarts", 8)),
steps_per_restart=int(settings.get("steps", 1200)),
seed=settings.get("seed"),
),
)
job.progress = 0.85
job.status = "encoding"
job.message = f"Reverse-translating and scrubbing restriction sites for {len(variants)} variants…"
df = variants_to_dataframe(
record.protein,
variants,
host=settings.get("host", "e_coli"),
forbidden_sites=DEFAULT_FORBIDDEN_SITES,
)
# Per-variant model-certainty (glass-box): how sharp ESM-2's posterior
# is at the changed residues, weakest-link. Free — reconstructed from
# the ΔLL table we already have (softmax is shift-invariant), no extra
# forward pass. Flows into the table, the CSV, and the result payload.
try:
from dee.core import confidence as _conf
_pos_conf = _conf.position_confidence_from_scores(scores_df)
_cv = [_conf.variant_confidence(str(mm or ""), _pos_conf)
for mm in df["Mutations_AA"]]
df["Confidence"] = [c for (c, _w) in _cv]
df["Confidence_Weakest"] = [w for (_c, w) in _cv]
except Exception: # noqa: BLE001 — confidence is additive, never fatal
logger.exception("confidence annotation skipped (round 1)")
# ---------------------------------------------------------------
# Build a WT pseudo-row with the SAME codon-optimization and
# restriction-site scrub applied to variants, so the user can
# order the unmutated wild-type alongside the evolved variants.
# Variant_ID 'WT' is special-cased on the client.
# ---------------------------------------------------------------
try:
wt_host = settings.get("host", "e_coli")
wt_raw_dna = reverse_translate(record.protein, host=wt_host, append_stop=True)
wt_clean_dna, wt_report = scrub_restriction_sites(
wt_raw_dna,
record.protein + "*",
host=wt_host,
forbidden_sites=DEFAULT_FORBIDDEN_SITES,
)
wt_pcr = pcr_metrics(wt_clean_dna)
wt_row = {
"Variant_ID": "WT",
"Mutations_AA": "",
"Mutant_AA_Seq": record.protein,
"Optimized_DNA_Seq": wt_clean_dna,
"Predicted_Fitness_Score": 0.0,
"Length_bp": wt_pcr.get("length_bp"),
"GC_Percent": wt_pcr.get("gc_percent"),
"Primer_Fwd": wt_pcr.get("primer_fwd"),
"Primer_Fwd_Tm_C": wt_pcr.get("primer_fwd_tm_c"),
"Primer_Fwd_GC_Percent": wt_pcr.get("primer_fwd_gc"),
"Primer_Rev": wt_pcr.get("primer_rev"),
"Primer_Rev_Tm_C": wt_pcr.get("primer_rev_tm_c"),
"Primer_Rev_GC_Percent": wt_pcr.get("primer_rev_gc"),
"Annealing_Temp_C": wt_pcr.get("annealing_temp_c"),
"Restriction_Sites_Found": sum(wt_report.sites_found.values()),
"Restriction_Sites_Unresolved": len(wt_report.unresolved),
"Confidence": None, # WT isn't a prediction — no certainty score
"Confidence_Weakest": "",
}
except Exception as wt_exc: # noqa: BLE001 — never fail the whole run for the WT row
# Audit H5: previously this branch set wt_row = None and the WT
# row silently vanished from the table. The user's reference
# sample (the unmutated sequence they paid us to design around)
# was gone with no signal. Now we emit a sentinel row with the
# error message in the Mutations_AA column so the UI can render
# an explicit "WT row couldn't be built — <reason>" indicator
# instead of silently omitting it.
logger.warning("Couldn't build WT pseudo-row: %s", wt_exc)
wt_row = {
"Variant_ID": "WT",
"Mutations_AA": f"ERROR: {wt_exc}",
"Mutant_AA_Seq": record.protein,
"Optimized_DNA_Seq": "",
"Predicted_Fitness_Score": 0.0,
"Length_bp": 0,
"GC_Percent": 0.0,
"Primer_Fwd": "",
"Primer_Fwd_Tm_C": None,
"Primer_Fwd_GC_Percent": 0.0,
"Primer_Rev": "",
"Primer_Rev_Tm_C": None,
"Primer_Rev_GC_Percent": 0.0,
"Annealing_Temp_C": None,
"Restriction_Sites_Found": 0,
"Restriction_Sites_Unresolved": 0,
# Sentinel flag the frontend can check to render the error
# banner inline on the WT row instead of trying to display
# an empty DNA sequence as if it were valid.
"WT_Error": str(wt_exc),
}
job.progress = 0.95
stem = Path(original_name).stem or "library"
safe_id = "".join(c if c.isalnum() else "_" for c in record.identifier) or "wt"
ts = time.strftime("%Y%m%d_%H%M%S")
filename = f"{stem}__{safe_id}__{ts}_dee_library.csv"
primary = OUTPUT_DIR / filename
write_library_csv(df, str(primary))
job.csv_path = str(primary)
# Best-effort copy to Desktop. Requires the user to have granted the
# .app Files & Folders → Desktop access (NSDesktopFolderUsageDescription
# in Info.plist triggers the system prompt).
desktop_dest = DESKTOP_DIR / filename
try:
desktop_dest.write_bytes(primary.read_bytes())
job.desktop_path = str(desktop_dest)
except (PermissionError, OSError) as exc:
logger.warning("Could not copy to Desktop (%s); CSV stays in ~/.dee/output/.", exc)
job.desktop_path = None
variant_rows = df.to_dict(orient="records")
# WT row first — UI sorts by Variant_ID so 'WT' sorts naturally last by
# default; the client explicitly pins it to row 0.
job.variants = ([wt_row] + variant_rows) if wt_row else variant_rows
job.status = "done"
job.progress = 1.0
job.message = f"Done. {len(variants)} variants ready."
job.finished_at = time.time()
# Save the library to Supabase Storage so the user can access it
# from their dashboard. Signed-in users only — anonymous runs no-op
# inside save_library_async. Fire-and-forget; failures are logged
# but don't propagate to the user.
try:
top_fitness = None
if variant_rows:
# Variants are sorted descending by fitness; the first one's
# Predicted_Fitness_Score (if present) is the top score.
top_fitness = variant_rows[0].get("Predicted_Fitness_Score")
if isinstance(top_fitness, (int, float)):
top_fitness = float(top_fitness)
else:
top_fitness = None
# Library name: prefer the original input filename's stem; fall
# back to the WT identifier. Truncated to keep dashboard rows tidy.
library_name = (
Path(original_name).stem
or record.identifier
or "library"
)[:120]
# Pre-assign the library id so the client gets it in the result and
# can log wet-lab outcomes against it (the learning flywheel).
job.library_id = uuid.uuid4().hex
_auth.save_library_async(
user_id=job.user_id,
job_id=job.job_id,
csv_path=job.csv_path,
name=library_name,
wt_identifier=record.identifier,
wt_protein=record.protein,
host_organism=settings.get("host", "e_coli"),
n_variants=len(variants),
top_fitness=top_fitness,
library_id=job.library_id,
)
# File it under the project it was designed in. Until this, a
# library reached its project only by matching names in
# /api/mission, so renaming either lost it — and reopening the
# project showed whichever library happened to be on screen.
if job.user_id and job.construct_id:
_auth.attach_library_async(job.user_id, job.construct_id, job.library_id)
except Exception as exc: # noqa: BLE001
# save_library_async is itself fire-and-forget, but the kwargs
# build above could in principle raise. Never block the user.
logger.warning("save_library kickoff failed: %s", exc)
except SequenceValidationError as exc:
job.status = "error"
job.error = f"Couldn't parse input: {exc}"
job.message = job.error
job.finished_at = time.time()
except Exception as exc: # noqa: BLE001
logger.exception("Pipeline crashed.")
job.status = "error"
job.error = f"{type(exc).__name__}: {exc}"
job.message = job.error
job.finished_at = time.time()
finally:
# Persist the run's terminal state to public.runs. The row was inserted
# with status='started' when the job kicked off; this is the only place
# status/finished_at/elapsed_seconds/n_variants get written. Best-effort.
try:
_n = len(job.variants) if getattr(job, "variants", None) else None
_auth.mark_run_finished_async(
job_id=job.job_id,
status=job.status,
elapsed_seconds=job.elapsed(),
n_variants=_n,
)
except Exception: # noqa: BLE001 — telemetry must never affect a run
pass
# ============================================================== EDIT PIPELINE
# Re-derive every column of a variant row from a user-edited DNA string. The
# input is treated as authoritative — we do NOT scrub restriction sites or
# silently mutate the user's bases. We translate it, diff against WT to
# rebuild the mutation label, recompute PCR metrics, and report restriction
# sites as found (cleared count is always 0 — they wrote it, they keep it).
# Returns the new row dict in the same shape as variants_to_dataframe rows
# so it can drop into job.variants[i] directly. Raises ValueError on
# user-correctable issues (callers should turn that into HTTP 400).
def _diff_mutations_against_wt(wt_protein: str, mutant: str) -> str:
"""Build a 'A23V,K45R' label from the WT-vs-edited protein diff.
Same convention used elsewhere in the codebase: 1-indexed AA positions,
single-letter codes, '*' for stop. If lengths differ we still emit a
label for the overlapping prefix and prefix the whole string with a
length-change marker so the UI can render the warning prominently.
"""
overlap = min(len(wt_protein), len(mutant))
muts = []
for i in range(overlap):
if wt_protein[i] != mutant[i]:
muts.append(f"{wt_protein[i]}{i + 1}{mutant[i]}")
label = ",".join(muts)
if len(wt_protein) != len(mutant):
marker = f"length:WT={len(wt_protein)}aa,edited={len(mutant)}aa"
return f"{marker}|{label}" if label else marker
return label
def _scan_restriction_sites(
dna: str,
forbidden_sites: Optional[Dict[str, str]] = None,
) -> Dict[str, int]:
"""Scan both strands for forbidden recognition sites. Pure observation —
no scrubbing. Returns {enzyme: count} for sites present in the user's
edited DNA, summed across forward + reverse strands so a Type IIS site
on either strand is counted once per occurrence."""
forbidden = forbidden_sites or DEFAULT_FORBIDDEN_SITES
from Bio.Seq import Seq
rc = str(Seq(dna).reverse_complement())
counts: Dict[str, int] = {}
for enzyme, motif in forbidden.items():
fwd = sum(1 for _ in re.finditer(f"(?={re.escape(motif)})", dna))
rev = sum(1 for _ in re.finditer(f"(?={re.escape(motif)})", rc))
counts[enzyme] = fwd + rev
return counts
def _rederive_variant_row(
*,
variant_id: str,
edited_dna: str,
wt_protein: str,
original_row: Dict[str, Any],
) -> Dict[str, Any]:
"""Recompute every derived column of a variant row from a new DNA string.
The user's edit is treated as authoritative. We validate (ACGT only,
length multiple of 3), translate (a premature stop is allowed — it
truncates the product, e.g. an intentional nonsense/knockout mutation),
diff against WT for the Mutations_AA label, recompute PCR metrics, and
report restriction-site presence honestly.
Raises ValueError with a user-readable message on validation failure —
callers should turn that into HTTP 400 so the frontend can display it.
"""
from Bio.Seq import Seq
if not edited_dna:
raise ValueError("Edited DNA is empty.")
cleaned = edited_dna.upper()
bad = set(cleaned) - set("ACGT")
if bad:
raise ValueError(
f"Edited DNA contains non-ACGT character(s): {sorted(bad)!r}. "
"Only A, C, G, T are allowed. RNA U should be converted to T "
"and IUPAC ambiguity codes must be resolved before saving."
)
if len(cleaned) % 3 != 0:
raise ValueError(
f"Edited DNA length ({len(cleaned)} bp) is not a multiple of 3 — "
"frame would be broken. Add or remove bases until the length "
"is divisible by 3 before saving."
)
# Translate. Stop codons appear as '*'. A premature stop is a LEGITIMATE
# edit — a nonsense / truncation mutation (e.g. an intentional knockout or
# a single-base SNP that happens to create a stop). We don't block it:
# synthesis vendors synthesize the DNA as-is, they don't reject internal
# stops. The encoded protein simply ends at the first stop; the user's
# client shows a non-blocking heads-up. (Previously this hard-rejected the
# save with a false "vendors will reject this" claim — an audit finding.)
raw_protein = str(Seq(cleaned).translate(table=1, to_stop=False))
protein = raw_protein.split("*")[0]
# Mutations label — diff against WT. For length-changing edits this
# prefixes a 'length:...' marker so the frontend can warn loudly.
is_wt_row = variant_id.upper() == "WT"
mutations_label = "" if is_wt_row and protein == wt_protein else (
_diff_mutations_against_wt(wt_protein, protein)
)
# PCR metrics + restriction-site count from the actual edited bytes.
pcr = pcr_metrics(cleaned)
re_counts = _scan_restriction_sites(cleaned)
# Compose the row. Preserve fields we don't recompute (Variant_ID,
# Predicted_Fitness_Score) from the original row so the variant keeps
# its identity in the table.
return {
"Variant_ID": original_row.get("Variant_ID", variant_id),
"Mutations_AA": mutations_label,
"Mutant_AA_Seq": protein,
"Optimized_DNA_Seq": cleaned,
"Predicted_Fitness_Score": original_row.get(
"Predicted_Fitness_Score", 0.0
),
"Length_bp": pcr.get("length_bp"),
"GC_Percent": pcr.get("gc_percent"),
"Primer_Fwd": pcr.get("primer_fwd"),
"Primer_Fwd_Tm_C": pcr.get("primer_fwd_tm_c"),
"Primer_Fwd_GC_Percent": pcr.get("primer_fwd_gc"),
"Primer_Rev": pcr.get("primer_rev"),
"Primer_Rev_Tm_C": pcr.get("primer_rev_tm_c"),
"Primer_Rev_GC_Percent": pcr.get("primer_rev_gc"),
"Annealing_Temp_C": pcr.get("annealing_temp_c"),
"Restriction_Sites_Found": sum(re_counts.values()),
# We don't scrub edits, so 'Unresolved' equals 'Found' here. The
# frontend treats anything > 0 as a hard warning before synthesis.
"Restriction_Sites_Unresolved": sum(re_counts.values()),
# Marker so exports + UI can show "edited" indicators without
# second-guessing equality with original_row.
"Edited": True,
}
def _atomic_rewrite_csv(job: JobState) -> None:
"""Rewrite the job's library CSV from current job.variants atomically.
Writes to a sibling .tmp file and renames into place so a concurrent
/api/download never sees a half-written file. WT-row 'Edited' flag and
other UI-only fields are dropped from the CSV — the on-disk shape stays
identical to what variants_to_dataframe originally wrote.
"""
import pandas as pd
if not job.csv_path or not job.variants:
return
target = Path(job.csv_path)
tmp = target.with_suffix(target.suffix + ".tmp")
df_rows = []
for v in job.variants:
row = {k: val for k, val in v.items() if k != "Edited"}
df_rows.append(row)
pd.DataFrame(df_rows).to_csv(str(tmp), index=False)
tmp.replace(target)
EXPORT_DISCLAIMER = (
"RESEARCH USE ONLY — NOT FOR DIAGNOSTIC OR THERAPEUTIC USE. "
"Sequences in this file are computational predictions generated by "
"TuringDNA. Visually verify every sequence before ordering DNA from a "
"synthesis vendor. The TuringDNA authors disclaim all warranties; "
"you are responsible for validating outputs against your experimental "
"requirements."
)
def _disclaimer_lines(prefix: str = "") -> List[str]:
"""Split the disclaimer into wrapped lines with an optional comment prefix.
Used by formats whose native comment syntax is line-based (FASTA's ``;``,
GenBank ``COMMENT`` blocks, shell-style ``#``)."""
import textwrap
wrapped = textwrap.wrap(EXPORT_DISCLAIMER, width=72)
return [f"{prefix}{line}" for line in wrapped]
def _fasta_from_job(job: JobState, *, of: str = "protein") -> str:
"""Serialize a job's variants as FASTA. ``of`` selects which sequence to
emit per record — protein or DNA. Protein is the more common ask (mutants
aligned against WT); DNA is included for completeness.
Prepends the export disclaimer as ``;``-comment lines (the standard
FASTA comment syntax — every conforming parser skips lines starting
with ``;``). The disclaimer is also embedded in every record's
description so a downstream tool that extracts a single record still
carries the "predicted output; verify before use" annotation.
"""
if not job.variants:
return ""
lines: List[str] = []
lines.extend(_disclaimer_lines("; "))
lines.append(f"; Generated by TuringDNA · {time.strftime('%Y-%m-%d %H:%M:%S')}")
lines.append(";")
seq_key = "Mutant_AA_Seq" if of == "protein" else "Optimized_DNA_Seq"
for v in job.variants:
header = (
f">{v.get('Variant_ID')} {v.get('Mutations_AA')} "
f"fitness={v.get('Predicted_Fitness_Score')} "
f"[PREDICTED · verify before synthesis]"
)
seq = v.get(seq_key, "")
lines.append(header)
for i in range(0, len(seq), 60):
lines.append(seq[i : i + 60])
return "\n".join(lines) + "\n"
def _genbank_from_job(job: JobState) -> str:
"""Build a multi-record GenBank file with one record per variant.
Each record gets a CDS feature whose ``/translation`` is the mutant
protein, plus a misc_feature per substitution so the mutation positions
are visible when opened in SnapGene / Benchling.
"""
from Bio.Seq import Seq
from Bio.SeqFeature import FeatureLocation, SeqFeature
from Bio.SeqRecord import SeqRecord
from Bio import SeqIO
records = []
for v in job.variants or []:
dna = v.get("Optimized_DNA_Seq", "")
protein = v.get("Mutant_AA_Seq", "")
muts = v.get("Mutations_AA", "")
record = SeqRecord(
Seq(dna),
id=v.get("Variant_ID", "variant"),
name=v.get("Variant_ID", "variant"),
description=(
f"TuringDNA predicted variant — RESEARCH USE ONLY · "
f"verify before synthesis. mutations={muts}; "
f"fitness={v.get('Predicted_Fitness_Score')}"
),
annotations={
"molecule_type": "DNA",
"organism": "synthetic construct",
# GenBank COMMENT block — SnapGene / Benchling render this
# at the top of the entry when the file is opened.
"comment": EXPORT_DISCLAIMER,
},
)
# Whole-CDS feature with the translation in /translation.
cds_qualifiers = {
"gene": [job.wt_identifier or "designed_variant"],
"product": ["designed variant"],
"note": [f"mutations={muts}"],
}
if protein:
cds_qualifiers["translation"] = [protein]
record.features.append(
SeqFeature(
FeatureLocation(0, len(dna)),
type="CDS",
qualifiers=cds_qualifiers,
)
)
# One misc_feature per substitution at the nucleotide it touches.
# Audit H6: the previous parser did
# pos_aa = int("".join(c for c in mut[1:-1] if c.isdigit()))
# which would happily turn a malformed label like "A1V2K3R" into the
# position string "123" and place a feature at nt 367 — completely
# wrong. It would also silently truncate at the first non-digit
# character. Use the same strict regex the frontend's detail panel
# uses (^[A-Z](\d+)[A-Z*]$) so labels that don't match the format
# are skipped rather than parsed into garbage positions. The
# length-marker prefix from manual edits ("length:WT=8aa,...|G3V")
# is handled by taking only the post-"|" portion if present.
_MUT_RE = re.compile(r"^([A-Z])(\d+)([A-Z*])$")
mut_payload = muts.split("|", 1)[-1] if "|" in muts else muts
for mut in mut_payload.split(","):
mut = mut.strip()
m = _MUT_RE.match(mut)
if not m:
continue
pos_aa = int(m.group(2))
if pos_aa < 1:
continue
nt_start = (pos_aa - 1) * 3
nt_end = nt_start + 3
if nt_end > len(dna):
continue
record.features.append(
SeqFeature(
FeatureLocation(nt_start, nt_end),
type="misc_feature",
qualifiers={"label": [mut], "note": [f"Substitution {mut}"]},
)
)
records.append(record)
buf = io.StringIO()
SeqIO.write(records, buf, "genbank")
return buf.getvalue()
def _xlsx_from_job(job: JobState) -> bytes:
"""Write the variant table to a production-quality .xlsx workbook.
Two sheets:
* "Variants" — the table itself, with bold-colored header row, frozen
top row, auto-filter, column widths sized per content type, number
formatting per column (3 decimals for fitness, 1 for GC/Tm, integer
for bp), and horizontally-clamped sequence columns so the long
Mutant_AA_Seq / Optimized_DNA_Seq cells don't visually drown out
everything else.
* "Run info" — key-value pairs of every parameter the engine used,
plus the auto-generated methods paragraph for copy-pasting into a
Materials & Methods section.
Requires ``openpyxl``. Returns raw bytes for the Flask handler to stream.
"""
import openpyxl
from openpyxl.styles import Alignment, Border, Font, PatternFill, Side
from openpyxl.utils import get_column_letter
wb = openpyxl.Workbook()
ws = wb.active
ws.title = "Variants"
variants = job.variants or []
if not variants:
wb.remove(ws)
ws = wb.create_sheet("Variants")
ws["A1"] = "No variants in this run."
buf = io.BytesIO()
wb.save(buf)
buf.seek(0)
return buf.getvalue()
headers = list(variants[0].keys())
# Header row — bold white on deep brand background, all caps, slightly
# taller so it reads as a real header band.
header_fill = PatternFill("solid", fgColor="0A5F77")
header_font = Font(name="Inter", size=11, bold=True, color="FFFFFF")
header_align = Alignment(horizontal="left", vertical="center", wrap_text=False)
body_font = Font(name="Inter", size=10)
body_align = Alignment(horizontal="left", vertical="top", wrap_text=False)
num_align = Alignment(horizontal="right", vertical="top")
mono_font = Font(name="JetBrains Mono", size=9)
band_fill = PatternFill("solid", fgColor="F4F6F9") # alternating row color
thin = Side(style="thin", color="E0E5EC")
border = Border(left=thin, right=thin, top=thin, bottom=thin)
# Column geometry — width sized to typical content per column type.
# Number formats keyed to specific column names (rest = text).
COL_WIDTH = {
"Variant_ID": 11,
"Mutations_AA": 28,
"Mutant_AA_Seq": 52,
"Optimized_DNA_Seq": 70,
"Predicted_Fitness_Score": 14,
"Length_bp": 9,
"GC_Percent": 10,
"Primer_Fwd": 32,
"Primer_Fwd_Tm_C": 12,
"Primer_Fwd_GC_Percent": 14,
"Primer_Rev": 32,
"Primer_Rev_Tm_C": 12,
"Primer_Rev_GC_Percent": 14,
"Annealing_Temp_C": 14,
"Restriction_Sites_Found": 14,
"Restriction_Sites_Unresolved": 16,
}
NUM_FORMAT = {
"Predicted_Fitness_Score": "0.000",
"GC_Percent": "0.0",
"Primer_Fwd_Tm_C": "0.0",
"Primer_Fwd_GC_Percent": "0.0",
"Primer_Rev_Tm_C": "0.0",
"Primer_Rev_GC_Percent": "0.0",
"Annealing_Temp_C": "0.0",
"Length_bp": "0",
"Restriction_Sites_Found": "0",
"Restriction_Sites_Unresolved": "0",
}
MONO_COLS = {"Mutations_AA", "Mutant_AA_Seq", "Optimized_DNA_Seq",
"Primer_Fwd", "Primer_Rev", "Variant_ID"}
NUM_COLS = set(NUM_FORMAT.keys())
# Disclaimer banner row at the very top. Merged across all data
# columns, amber background, bold red text — meant to be the first
# thing the user sees when they open the workbook. Drops the actual
# header row to row 2 and data to row 3 onwards.
disclaimer_fill = PatternFill("solid", fgColor="FEF3C7")
disclaimer_font = Font(name="Inter", size=10, bold=True, color="7F1D1D")
disclaimer_align = Alignment(horizontal="left", vertical="center", wrap_text=True)
ws.cell(row=1, column=1, value=EXPORT_DISCLAIMER)
ws.merge_cells(start_row=1, start_column=1, end_row=1, end_column=max(1, len(headers)))
banner = ws.cell(row=1, column=1)
banner.fill = disclaimer_fill
banner.font = disclaimer_font
banner.alignment = disclaimer_align
ws.row_dimensions[1].height = 48
# Write header row.
for col_idx, col in enumerate(headers, start=1):
cell = ws.cell(row=2, column=col_idx, value=col)
cell.fill = header_fill
cell.font = header_font
cell.alignment = header_align
cell.border = border
ws.row_dimensions[2].height = 22
# Write data rows with alternating band fill + column-specific formatting.
# Data now starts at row 3 because of the banner + header rows above.
for row_idx, variant in enumerate(variants, start=3):
is_band = (row_idx % 2 == 0)
for col_idx, col in enumerate(headers, start=1):
val = variant.get(col)
cell = ws.cell(row=row_idx, column=col_idx, value=val)
cell.border = border
if is_band:
cell.fill = band_fill
if col in NUM_COLS:
cell.alignment = num_align
cell.number_format = NUM_FORMAT[col]
cell.font = body_font
elif col in MONO_COLS:
cell.alignment = body_align
cell.font = mono_font
else:
cell.alignment = body_align
cell.font = body_font
# Column widths.
for col_idx, col in enumerate(headers, start=1):
width = COL_WIDTH.get(col, 14)
ws.column_dimensions[get_column_letter(col_idx)].width = width
# Freeze the header row + first two columns so Variant_ID + Mutations stay
# in view while horizontally scrolling through primer + sequence columns.
# Now C3 because the banner + header rows live above.
ws.freeze_panes = "C3"
# Auto-filter the data range only (skips the banner row).
from openpyxl.utils import get_column_letter as _gcl
last_col_letter = _gcl(len(headers))
last_row = 2 + len(variants)
ws.auto_filter.ref = f"A2:{last_col_letter}{last_row}"
# ---- Sheet 2: Run info ------------------------------------------------
info = wb.create_sheet("Run info")
label_font = Font(name="Inter", size=11, bold=True, color="0A5F77")
value_font = Font(name="Inter", size=11)
methods_font = Font(name="Inter", size=10)
# Disclaimer banner at the top of Run info too. Mirrors the Variants
# sheet so a user who jumps straight to Run info still sees the
# research-use-only framing.
info.cell(row=1, column=1, value=EXPORT_DISCLAIMER)
info.merge_cells(start_row=1, start_column=1, end_row=1, end_column=4)
info_banner = info.cell(row=1, column=1)
info_banner.fill = disclaimer_fill
info_banner.font = disclaimer_font
info_banner.alignment = disclaimer_align
info.row_dimensions[1].height = 48
INFO_ROW_OFFSET = 2 # data starts at row 3 (banner row 1, blank row 2)
s = job.settings_used or {}
model_label = {
"small": "ESM-2 35M (facebook/esm2_t12_35M_UR50D)",
"medium": "ESM-2 650M (facebook/esm2_t33_650M_UR50D)",
"large": "ESM-2 3B (facebook/esm2_t36_3B_UR50D)",
}.get(s.get("model"), s.get("model", ""))
host_label = {
"e_coli": "Escherichia coli",
"yeast": "Saccharomyces cerevisiae",
"human": "Homo sapiens",
}.get(s.get("host"), s.get("host", ""))
rows = [
("Wild-type identifier", job.wt_identifier),
("Wild-type length", f"{len(job.wt_protein)} aa"),
("Model", model_label),
("Percentile cutoff",
f"≥ p{s.get('percentile')} (top {100 - float(s.get('percentile', 85)):.0f}%)"),
("K variants returned", s.get("k")),
("Min mutations / variant", s.get("min_mutations")),
("Max mutations / variant", s.get("max_mutations")),
("SA restarts", s.get("restarts")),
("SA steps / restart", s.get("steps_per_restart")),
("Expression host", host_label),
("Compute device", s.get("device")),
("Random seed", s.get("seed") if s.get("seed") is not None else "(random)"),
("CDS feature picked", s.get("cds_feature") or "(none / single CDS)"),
("Run started (epoch s)", job.started_at),
("Wall time (s)", job.elapsed()),
]
for i, (k, v) in enumerate(rows, start=1 + INFO_ROW_OFFSET):
info.cell(row=i, column=1, value=k).font = label_font
info.cell(row=i, column=2, value=v).font = value_font
# Methods paragraph. Offset accounts for the disclaimer banner block.
info.cell(row=len(rows) + 2 + INFO_ROW_OFFSET, column=1, value="Materials & methods (copy-paste)").font = label_font
methods = (
f"Variants of {job.wt_identifier} ({len(job.wt_protein)} aa) were designed "
f"in silico with {model_label} (Lin et al., Science 2022) using the "
f"wild-type marginal log-likelihood scoring scheme of Meier et al. "
f"(Adv. Neural Inf. Process. Syst. 2021). Single-point substitutions "
f"in the top {100 - float(s.get('percentile', 85)):.0f}% by ΔLL were "
f"retained as the combinatorial search space. {s.get('k')} multi-mutant "
f"variants with {s.get('min_mutations')}–{s.get('max_mutations')} "
f"simultaneous substitutions were generated by simulated annealing "
f"({s.get('restarts')} restarts × {s.get('steps_per_restart')} steps, "
f"geometric cooling) maximizing cumulative ΔLL with stop-codon and "
f"duplicate-position penalties. Optimized DNA was reverse-translated "
f"using {host_label} codon-usage frequencies and synonymously cleaned "
f"of BsaI, BsmBI, and NotI recognition sites for Golden Gate "
f"compatibility."
)
methods_row = len(rows) + 3 + INFO_ROW_OFFSET
mcell = info.cell(row=methods_row, column=1, value=methods)
mcell.font = methods_font
mcell.alignment = Alignment(wrap_text=True, vertical="top")
info.merge_cells(start_row=methods_row, start_column=1, end_row=methods_row, end_column=4)
info.row_dimensions[methods_row].height = 140
info.column_dimensions["A"].width = 32
info.column_dimensions["B"].width = 48
info.column_dimensions["C"].width = 16
info.column_dimensions["D"].width = 16
buf = io.BytesIO()
wb.save(buf)
buf.seek(0)
return buf.getvalue()
def _json_from_job(job: JobState) -> bytes:
"""Compact, programmer-friendly JSON dump of the run.
Includes ``_disclaimer`` and ``_generated_at`` fields at the top of
the payload so any downstream consumer sees the "predicted; verify
before use" annotation alongside the data.
"""
payload = {
"_disclaimer": EXPORT_DISCLAIMER,
"_generated_at": time.strftime("%Y-%m-%dT%H:%M:%S"),
"_generator": "TuringDNA",
"wt_identifier": job.wt_identifier,
"wt_protein": job.wt_protein,
"wt_protein_length": len(job.wt_protein),
"variants": job.variants or [],
}
return json.dumps(payload, indent=2).encode("utf-8")
def _tsv_from_job(job: JobState) -> bytes:
"""Tab-separated equivalent of the CSV — friendlier for some analysis
pipelines (R, awk, Perl) that hate quoted commas. The disclaimer is
prepended as ``#``-prefixed comment lines; pandas / R / awk users
pass ``comment='#'`` (or equivalent) to skip them, while a vendor
eyeballing the file in a text editor still sees the disclaimer first.
"""
import pandas as pd
df = pd.DataFrame(job.variants or [])
body = df.to_csv(sep="\t", index=False)
header = "\n".join(_disclaimer_lines("# ")) + "\n"
header += f"# Generated by TuringDNA · {time.strftime('%Y-%m-%d %H:%M:%S')}\n"
return (header + body).encode("utf-8")
# Format registry — keyed by the value of ``?format=``. Each entry produces
# (payload, mimetype, file-extension). New formats are a one-line addition.
def _render_download(job: JobState, fmt: str):
if fmt == "csv":
# Excel in many European locales (incl. the user's Georgian Excel)
# treats CSV with ',' delimiter as a single column unless told
# otherwise. We prepend Excel's "sep=," hint line and a UTF-8 BOM so
# double-clicking the file opens it with proper columns everywhere.
#
# Disclaimer placement: appended as a final row after all variant
# data. Putting it BEFORE the sep=, hint breaks Excel's auto-
# detection; putting it ABOVE the header row breaks pandas /
# csv.DictReader. A final row with the disclaimer in column A and
# blanks elsewhere is the format-neutral compromise — every tool
# reads it as one extra row, and a vendor opening the file in
# Excel sees the disclaimer plainly at the bottom of the data.
#
# Audit M6: the previous implementation did
# raw = read_bytes(); decoded = raw.decode('utf-8','replace'); …; decoded.encode('utf-8')
# which is a needless round-trip — and silently corrupted any
# byte that wasn't valid UTF-8 (the 'replace' arg). Now we read
# bytes and work in bytes throughout; the only decoding is the
# first-line column count, done on a bounded prefix.
with open(job.csv_path, "rb") as f:
raw = f.read()
# Read the first line in bytes to count columns without decoding
# the whole file. Sequence columns are pure ASCII so split('\n')
# on bytes is safe here.
first_line_bytes = raw.split(b"\n", 1)[0] if raw else b""
n_cols = max(1, first_line_bytes.count(b",") + 1) if first_line_bytes else 1
# The disclaimer row is pure ASCII (EXPORT_DISCLAIMER above is
# constructed from ASCII chars only), so encoding once is safe.
disclaimer_row = (
'"' + EXPORT_DISCLAIMER.replace('"', '""') + '"'
+ ("," * (n_cols - 1))
+ "\r\n"
).encode("utf-8")
bom = b"\xef\xbb\xbf"
return bom + b"sep=,\r\n" + raw + disclaimer_row, "text/csv", "csv"
if fmt == "tsv":
return _tsv_from_job(job), "text/tab-separated-values", "tsv"
if fmt == "fasta":
return _fasta_from_job(job, of="protein").encode("utf-8"), "text/x-fasta", "fasta"
if fmt == "fasta-dna":
return _fasta_from_job(job, of="dna").encode("utf-8"), "text/x-fasta", "fna"
if fmt in ("gb", "genbank"):
return _genbank_from_job(job).encode("utf-8"), "text/x-genbank", "gb"
if fmt in ("xlsx", "excel"):
return (
_xlsx_from_job(job),
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
"xlsx",
)
if fmt == "json":
return _json_from_job(job), "application/json", "json"
raise ValueError(
"format must be one of: csv, tsv, fasta, fasta-dna, gb, xlsx, json"
)
# ----------------------------------------------------------------- CLI entry
def serve(host: Optional[str] = None, port: Optional[int] = None) -> None:
"""Run the Flask server.
Honors ``HOST`` and ``PORT`` env vars first (so the same image runs under
Hugging Face Spaces' fixed ``PORT=7860`` contract, Fly.io's ``8080``,
Render's ``10000``, etc.) before falling back to local-dev defaults
(``127.0.0.1:4789``).
"""
env_host = os.environ.get("HOST")
env_port = os.environ.get("PORT")
host = host or env_host or "127.0.0.1"
port = port or (int(env_port) if env_port else 4789)
app = build_app()
logger.info("Digital Evolution Engine listening on http://%s:%s", host, port)
print(f"DEE_READY http://{host}:{port}", flush=True)
app.run(host=host, port=port, debug=False, use_reloader=False, threaded=True)
def build_app() -> Flask:
"""Construct the app + run one-time startup side-effects (logging config +
off-target prewarm). This is the WSGI entry point used in production by
gunicorn (``dee.server:build_app()``); ``serve()`` reuses it for local dev so
both paths get identical startup behaviour.
Prewarms the CRISPR off-target kmer indexes in daemon threads so user
requests don't pay the 60-300 s cold-build cost (ecoli ~3-5 s; human/mouse
60-180 s). A Design click before the build finishes returns empty + a
"building" flag rather than blocking the request past the ~120 s cap.
"""
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)s %(name)s: %(message)s",
)
app = create_app()
try:
from dee.core import offtarget as _ot
for _org in ("ecoli", "human", "mouse"):
if _ot.is_organism_ready(_org):
logger.info("Prewarming kmer index for %s in background", _org)
_ot.kick_off_build(_org)
except Exception as exc: # noqa: BLE001
logger.warning("Off-target prewarm failed (will still build on demand): %s", exc)
try:
_start_prior_rebuild_scheduler()
except Exception as exc: # noqa: BLE001
logger.warning("Prior-rebuild scheduler failed to start: %s", exc)
return app
if __name__ == "__main__":
serve()