dee/server.py

"""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 hashlib
import io
import json
import logging
import os
import re
import tempfile
import threading
import time
import traceback
import uuid
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Dict, List, Optional

from flask import Flask, Response, jsonify, request, send_file, send_from_directory

from dee import auth as _auth
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 ESM2Scorer, ScorerConfig, top_percentile_pool
from dee.optimizer.search import SearchConfig, apply_variant, evolve

logger = logging.getLogger("dee.server")

STATIC_DIR = Path(__file__).resolve().parent / "static"
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
    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

    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)


def _put_job(job: JobState) -> None:
    with _JOBS_LOCK:
        _JOBS[job.job_id] = job


# ----------------------------------------------------------------- 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 _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 _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=5,
            expect=1e-5,
        )
        job.status = "parsing"

        for record in NCBIXML.parse(result_handle):
            for alignment in record.alignments[:5]:
                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,
                        "alphafold_url": (
                            f"https://alphafold.ebi.ac.uk/files/AF-{uniprot}-F1-model_v4.pdb"
                            if uniprot else None
                        ),
                        "alphafold_page": (
                            f"https://alphafold.ebi.ac.uk/entry/{uniprot}"
                            if uniprot else None
                        ),
                    }
                )
            break  # only one query record
        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()


_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


def create_app() -> Flask:
    app = Flask(__name__, static_folder=None)

    # 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("/")
    def index() -> Response:
        return send_from_directory(STATIC_DIR, "index.html")

    @app.get("/static/<path:filename>")
    def static_file(filename: str) -> Response:
        return send_from_directory(STATIC_DIR, filename)

    # ─── /api/whoami — auth diagnostic ──────────────────────────────────
    # Temporary debug endpoint that exposes exactly what the backend
    # sees about the requester's auth state. Returns booleans + token
    # prefixes only (never the full token, never the JWT secret). Lets
    # us pinpoint why a signed-in user is being treated as anonymous
    # without needing HF Space log access.
    #
    # Test from the iframe DevTools console:
    #   fetch('/api/whoami').then(r => r.json()).then(console.log)
    #
    # Safe to leave in production — leaks no secrets. Remove once the
    # auth diagnosis is done if you want a smaller API surface.
    @app.get("/api/whoami")
    def whoami() -> Response:
        # Pull straight from the request so we can report what landed at
        # the server, independent of what _load_auth_into_g made of it.
        raw_header = request.headers.get("Authorization", "")
        header_seen = bool(raw_header)
        header_starts_with_bearer = raw_header.lower().startswith("bearer ")

        raw_token = ""
        if header_starts_with_bearer:
            raw_token = raw_header[7:].strip()
        token_len = len(raw_token)
        # First 8 + last 4 chars of the JWT. JWTs are base64url so these
        # characters aren't sensitive on their own; the secret is what
        # signs them. Useful for "is this the SAME JWT we expect?".
        token_fingerprint = (
            f"{raw_token[:8]}...{raw_token[-4:]}" if token_len > 12 else "(too-short-or-empty)"
        )

        # Decode header + payload WITHOUT verification — purely informational
        # so we can tell if the JWT itself is well-formed and what claims
        # it carries (iss, aud, exp). Verification result reported separately.
        decoded_payload: Dict[str, Any] = {}
        decode_ok = False
        try:
            if raw_token.count(".") == 2:
                _h, _p, _s = raw_token.split(".")
                pad = "=" * (-len(_p) % 4)
                import base64 as _b64
                decoded_payload = json.loads(
                    _b64.urlsafe_b64decode((_p + pad).encode("ascii")).decode("utf-8")
                )
                decode_ok = True
        except Exception:
            decoded_payload = {}
            decode_ok = False

        # Now check verification. _verify_jwt_any dispatches on the
        # JWT's `alg` — HS256 against the Legacy Secret, ES256/RS256
        # against the JWKS public key — so this reports `verified` for
        # whichever signing mode the Supabase project is currently in.
        verified = False
        if header_starts_with_bearer and _auth.AUTH_ENABLED:
            verified = _auth._verify_jwt_any(raw_token) is not None

        # Also surface the algorithm so we can tell at a glance whether
        # this project is still on Legacy HS256 or has migrated to ES256.
        decoded_alg = ""
        try:
            if raw_token.count(".") == 2:
                _h = raw_token.split(".")[0]
                pad = "=" * (-len(_h) % 4)
                import base64 as _b64
                decoded_alg = json.loads(
                    _b64.urlsafe_b64decode((_h + pad).encode("ascii")).decode("utf-8")
                ).get("alg", "")
        except Exception:
            decoded_alg = ""

        # Final auth state as the request handler would see it.
        ctx = _auth.get_auth()

        # Server clock and JWT timestamps — exposes clock-skew issues.
        now = int(time.time())
        exp = decoded_payload.get("exp") if isinstance(decoded_payload, dict) else None
        iat = decoded_payload.get("iat") if isinstance(decoded_payload, dict) else None

        return jsonify({
            "auth_enabled":          _auth.AUTH_ENABLED,
            "supabase_secret_set":   bool(_auth.SUPABASE_JWT_SECRET),
            "supabase_secret_len":   len(_auth.SUPABASE_JWT_SECRET or ""),
            "anon_secret_set":       bool(_auth.ANON_SECRET),
            "header_seen":           header_seen,
            "header_starts_bearer":  header_starts_with_bearer,
            "token_len":             token_len,
            "token_fingerprint":     token_fingerprint,
            "decode_ok":             decode_ok,
            "decoded_alg":           decoded_alg or None,
            "decoded_iss":           decoded_payload.get("iss") if decode_ok else None,
            "decoded_aud":           decoded_payload.get("aud") if decode_ok else None,
            "decoded_sub_present":   bool(decoded_payload.get("sub")) if decode_ok else False,
            "decoded_email_present": bool(decoded_payload.get("email")) if decode_ok else False,
            "verified":              verified,
            "anonymous":             ctx.anonymous,
            "resolved_user_id":      ctx.user_id,
            "server_time":           now,
            "jwt_iat":               iat,
            "jwt_exp":               exp,
            "jwt_seconds_until_exp": (exp - now) if isinstance(exp, (int, float)) else None,
            "jwt_seconds_since_iat": (now - iat) if isinstance(iat, (int, float)) else None,
        })

    @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 as exc:  # noqa: BLE001 — surface anything else as 500.
            logger.exception("Preview failed.")
            return jsonify({"error": str(exc), "kind": "internal"}), 500

    @app.post("/api/run")
    def start_run() -> Response:
        # Auth-or-quota gate. Anonymous users get one free run; 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,
        )
        _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

    @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 [],
                "settings_used": job.settings_used,
                "started_at": job.started_at,
                "elapsed_seconds": job.elapsed(),
            }
        )

    @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, isn't a multiple of 3,
        or contains a premature stop — the user must fix it before save.
        """
        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")
        if not sid:
            return jsonify({"error": "missing session_id"}), 400
        session = _get_session(sid)
        if 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")
        # 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

        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())

    # ================================================================= 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.
        if target_organism not in ("", "ecoli", "human", "mouse"):
            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)
        return jsonify({
            "input_length":        len(seq.strip()),
            "n_guides":            len(guides),
            "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,
            "guides": [
                {
                    "rank":              g.rank,
                    "enzyme":            g.enzyme,
                    "strand":            g.strand,
                    "position":          g.position,
                    "spacer":            g.spacer,
                    "pam":               g.pam,
                    "target_context":    g.target_context,
                    "composite_score":   g.composite_score,
                    "on_target_score":   g.on_target_score,
                    "cfd_max_offtarget": g.cfd_max_offtarget,
                    "offtarget_count":   g.offtarget_count,
                    "ko_efficacy":       g.ko_efficacy,
                    "ko_reasoning":      g.ko_reasoning,
                    "gc_pct":            g.gc_pct,
                    "flag_high_gc":      g.flag_high_gc,
                    "flag_low_gc":       g.flag_low_gc,
                    "flag_polyT":        g.flag_polyT,
                    "notes":             g.notes,
                    # Phase 2A — indel prediction + base editor
                    "top_indel_label":   g.top_indel_label,
                    "frameshift_pct":    g.frameshift_pct,
                    "top_dominance_pct": g.top_dominance_pct,
                    "predicted_indels":  g.predicted_indels,   # list[[lbl, freq]]
                    "be_cbe_positions":  g.be_cbe_positions,
                    "be_abe_positions":  g.be_abe_positions,
                    "be_summary":        g.be_summary,
                    # Phase 2B-1 — genome off-target + exon context
                    "genome_organism":           g.genome_organism,
                    "genome_offtarget_count":    g.genome_offtarget_count,
                    "genome_offtarget_max_cfd":  g.genome_offtarget_max_cfd,
                    "genome_offtarget_top_loc":  g.genome_offtarget_top_loc,
                    "exon_number":               g.exon_number,
                    "distance_to_splice":        g.distance_to_splice,
                    "in_nmd_zone":               g.in_nmd_zone,
                    "exon_context_summary":      g.exon_context_summary,
                    # Phase 2C-1 — cloning oligos
                    "cloning_vector_id":   g.cloning_vector_id,
                    "cloning_vector_name": g.cloning_vector_name,
                    "cloning_enzyme":      g.cloning_enzyme,
                    "cloning_addgene_id":  g.cloning_addgene_id,
                    "sense_oligo":         g.sense_oligo,
                    "antisense_oligo":     g.antisense_oligo,
                    # Phase 3 (M1) — base editing
                    "be_editor":           g.be_editor,
                    "be_editability":      g.be_editability,
                    "be_has_bystander":    g.be_has_bystander,
                    "be_edits":            g.be_edits,         # [[pos, from, to, act], …]
                    "be_outcome_label":    g.be_outcome_label,
                    "be_aa_change":        g.be_aa_change,
                    "be_creates_stop":     g.be_creates_stop,
                    # Phase 3 (M5) — structure viewer
                    "cut_residue":         g.cut_residue,
                }
                for g in guides
            ],
        })

    @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()
        if organism not in ("human", "mouse"):
            return jsonify({"ok": False,
                            "error": "Structure view is available for Human / Mouse genes."}), 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/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
        organism = str(body.get("organism", "")).lower().strip()
        if organism not in ("", "ecoli", "human", "mouse"):
            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)

    # ─── 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,
        )
        _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})

    @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()
        if target_organism not in ("", "ecoli", "human", "mouse"):
            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 = ""

        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

        # XLSX, not CSV. The previous CSV export mangled in locales where
        # Excel uses ";" as the default delimiter (Georgian, German,
        # French, etc.) — KO reasoning strings + any field with a comma
        # would split into the wrong column. 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"

        rows = _crispr_csv_rows(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"] = (
            f'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.

    # NOTE: the /api/chat endpoint (BioMistral-7B proxy via gradio_client to
    # winter4000/turingdna-assistant) was removed on 2026-05-25. The full
    # implementation lives in _llm_backup_2026-05-25/server/server.py.pre-strip
    # alongside the frontend chat panel + WebLLM browser-side LLM. Re-wire
    # when the assistant is ready to ship again.

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


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}.")

    seed = raw.get("seed")
    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}).",
        )

    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 _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)
        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)
        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"{Path(original_name).stem or 'sequence'} (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,
            },
        )


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,
        )
        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 = ESM2Scorer(
            ScorerConfig(
                model_name=settings.get("model", "small"),
                device=settings.get("device"),
                quantization=settings.get("quantization"),
            )
        )
        scores_df = scorer.score_all_substitutions(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

        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,
        )
        # ---------------------------------------------------------------
        # 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),
            }
        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]
            _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,
            )
        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()


# ============================================================== 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, no internal stops in non-terminal codons), then
    translate, 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 the whole thing. We use Biopython's standard table; stop
    # codons appear as '*'. Strip a single terminal '*' so the label diff
    # matches WT (which also has the stop stripped by translate_dna).
    raw_protein = str(Seq(cleaned).translate(table=1, to_stop=False))
    if raw_protein.endswith("*"):
        protein = raw_protein[:-1]
    else:
        protein = raw_protein

    # Premature-stop check — anything other than a single terminal stop is
    # a frame-shift / lab disaster, refuse the save.
    if "*" in protein:
        idx = protein.find("*")
        raise ValueError(
            f"Edited DNA contains a premature stop codon at amino-acid "
            f"position {idx + 1}. Synthesis vendors will reject this; "
            "remove the stop before saving."
        )

    # 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)
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
    )
    app = create_app()

    # Prewarm the CRISPR off-target kmer indexes in daemon threads at
    # startup so user requests don't pay the 60-300 s cold-build cost.
    # ecoli is small (3-5 s) and human/mouse are the slow ones (60-180 s).
    # If a user clicks Design with Human before the build finishes,
    # find_genomic_offtargets returns empty + a "building" status flag
    # rather than blocking the HTTP request (which would time out at
    # the HF Space's ~120 s request cap).
    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)

    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)


if __name__ == "__main__":
    serve()