np_app/service/server.py

import json
import hashlib
import math
import mimetypes
import multiprocessing
import os
import queue
import re
import secrets
import threading
import time
import traceback
from decimal import Decimal
from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, wait
from http.cookies import SimpleCookie
from http import HTTPStatus
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
import subprocess
import tempfile
from urllib.parse import parse_qs, urlencode, urlparse
from uuid import uuid4

import numpy
import nupack
from nupack import (
    Complex,
    Complementarity,
    Domain,
    Diversity,
    EnergyMatch,
    DesignOptions,
    Library,
    Match,
    Model,
    Pattern,
    Similarity,
    SSM,
    SetSpec,
    Strand,
    TargetComplex,
    TargetStrand,
    TargetTube,
    Tube,
    Weights,
    Window,
    complex_analysis,
    complex_design,
    tube_analysis,
    tube_design,
)
from nupack import config as nupack_config
from account import AccountStore, OIDCAuth
from split_strand_svg import render_split_strands_svg

try:
    import redis
except ImportError:
    redis = None


ROOT = Path(__file__).resolve().parent
INDEX_PATH = ROOT / "index.html"
HOME_PATH = ROOT / "home.html"
CLOUD_PATH = ROOT / "cloud.html"
ACCOUNT_PATH = ROOT / "account.html"
SETTINGS_PATH = ROOT / "settings.html"
ADMIN_PATH = ROOT / "admin.html"
ADMIN_LOGIN_PATH = ROOT / "admin-login.html"
STATIC_PATHS = {
    "/static/app-shell.css": ROOT / "static" / "app-shell.css",
    "/static/workspace-refresh.css": ROOT / "static" / "workspace-refresh.css",
    "/static/portal.js": ROOT / "static" / "portal.js",
}
FAVICON_PATH = ROOT / "favicon.svg"
GUIDE_PATH = ROOT / "design-guide.html"

HOST = os.environ.get("NP_HOST", "0.0.0.0")
PORT = int(os.environ.get("NP_PORT", "18765"))
RNA_PLOT_CMD = os.environ.get("RNA_PLOT_CMD", "RNAplot")
ENABLE_RNAPLOT = os.environ.get("ENABLE_RNAPLOT", "1") != "0"
STRUCTURE_PLOT_MODE = os.environ.get("NP_STRUCTURE_PLOT_MODE", "auto").strip().lower()
RUN_MODE = os.environ.get("NP_RUN_MODE", "server")
REDIS_URL = os.environ.get("NP_REDIS_URL", "").strip()
JOB_QUEUE_KEY = os.environ.get("NP_JOB_QUEUE_KEY", "np_replica:jobs")
JOB_RUNNING_KEY = os.environ.get("NP_JOB_RUNNING_KEY", "np_replica:jobs:running")
WORKER_RESOURCE_KEY = os.environ.get("NP_WORKER_RESOURCE_KEY", "np_replica:worker:resources")
WORKER_CONCURRENCY = os.environ.get("NP_WORKER_CONCURRENCY", "auto").strip().lower()
PER_JOB_THREAD_LIMIT = max(1, int(os.environ.get("NP_PER_JOB_THREAD_LIMIT", "1")))
NUPACK_CACHE_GB = float(os.environ.get("NP_NUPACK_CACHE_GB", "2.0"))
WORKER_MEMORY_GB = float(os.environ.get("NP_WORKER_MEMORY_GB", "0"))
WORKER_MEMORY_RESERVE_GB = max(0.0, float(os.environ.get("NP_WORKER_MEMORY_RESERVE_GB", "2.0")))
ESTIMATED_JOB_MEMORY_GB = max(0.1, float(os.environ.get("NP_ESTIMATED_JOB_MEMORY_GB", "8.0")))
ACCOUNT_DB_PATH = os.environ.get("NP_ACCOUNT_DB_PATH", "/data/np-replica.sqlite3")
ADMIN_TOKEN = os.environ.get("NP_ADMIN_TOKEN", "").strip()
ADMIN_SESSION_TTL_SECONDS = max(300, int(os.environ.get("NP_ADMIN_SESSION_TTL_SECONDS", "28800")))
ADMIN_COOKIE_NAME = os.environ.get("NP_ADMIN_COOKIE_NAME", "np_admin_session")
ADMIN_COOKIE_SECURE = os.environ.get("NP_ADMIN_COOKIE_SECURE", "1") != "0"
ADMIN_SESSION_PREFIX = os.environ.get("NP_ADMIN_SESSION_PREFIX", "np_replica:admin-session")

UNIT_SCALE = {
    "M": 1.0,
    "mM": 1e-3,
    "uM": 1e-6,
    "nM": 1e-9,
    "pM": 1e-12,
}
IUPAC_CODES = "ACGTUWSMKRYBDHVN"
IUPAC_CONSTRAINT_TOKEN = re.compile(rf"[{IUPAC_CODES}](?:\d+)?")
FIXED_BASES = set("ACGTU")
DNA_COMPLEMENT_BASE = str.maketrans({"A": "T", "C": "G", "G": "C", "T": "A", "U": "A"})
RNA_COMPLEMENT_BASE = str.maketrans({"A": "U", "C": "G", "G": "C", "T": "A", "U": "A"})

NUPACK_VERSION = str(getattr(nupack, "__version__", "4.1.0.1"))
MATERIAL_ALIASES = {"rna": "rna06", "dna": "dna04.3", "dna04": "dna04.3"}
MIXED_MATERIAL_PREFIXES = {"rna-dna06": "rd", "rna-merna06": "rm"}
MATERIAL_RULES = {
    "rna": {"sodium": (0.05, 1.0, 1.0), "magnesium": (0.0, 0.0, 0.0)},
    "rna06": {"sodium": (0.05, 1.0, 1.0), "magnesium": (0.0, 0.0, 0.0)},
    "rna95": {"sodium": (1.0, 1.0, 1.0), "magnesium": (0.0, 0.0, 0.0)},
    "merna06": {"sodium": (0.12, 0.12, 0.12), "magnesium": (0.0, 0.0, 0.0)},
    "rna-dna06": {"sodium": (0.12, 1.0, 1.0), "magnesium": (0.0, 0.0, 0.0)},
    "rna-merna06": {"sodium": (0.12, 0.12, 0.12), "magnesium": (0.0, 0.0, 0.0)},
    "dna": {"sodium": (0.05, 1.1, 1.0), "magnesium": (0.0, 0.2, 0.0)},
    "dna04": {"sodium": (0.05, 1.1, 1.0), "magnesium": (0.0, 0.2, 0.0)},
    "dna04.1": {"sodium": (0.05, 1.1, 1.0), "magnesium": (0.0, 0.2, 0.0)},
    "dna04.2": {"sodium": (0.05, 1.1, 1.0), "magnesium": (0.0, 0.2, 0.0)},
    "dna04.3": {"sodium": (0.05, 1.1, 1.0), "magnesium": (0.0, 0.2, 0.0)},
}

VALID_COMPUTE = {"pfunc", "pairs", "mfe", "sample", "subopt", "ensemble_size"}
VALID_UTILITY_OPERATIONS = {
    "pfunc", "structure_energy", "structure_probability", "sample", "pairs", "mfe",
    "subopt", "ensemble_size", "des", "defect", "seq_distance", "struc_distance",
}
FIXED_TARGET_POLICIES = {"exclude_from_optimization", "include"}
CANCEL_REQUESTED_STATUS = "cancel_requested"
CANCELED_STATUS = "canceled"
TERMINAL_JOB_STATUSES = {"success", "error", CANCELED_STATUS}
JOB_STORE = {}
JOB_LOCK = threading.Lock()
JOB_DEDUP_RESERVATIONS = {}
JOB_TTL_SECONDS = int(os.environ.get("NP_JOB_TTL_SECONDS", "3600"))
JOB_HEARTBEAT_SECONDS = max(1, int(os.environ.get("NP_JOB_HEARTBEAT_SECONDS", "30")))
JOB_MAX_COUNT = int(os.environ.get("NP_JOB_MAX_COUNT", "64"))
JOB_DEDUP_PREFIX = os.environ.get("NP_JOB_DEDUP_PREFIX", "np_replica:job-dedup")
SHARE_STORE = {}
SHARE_LOCK = threading.Lock()
ADMIN_SESSION_STORE = {}
ADMIN_SESSION_LOCK = threading.Lock()
SHARE_MAX_COUNT = int(os.environ.get("NP_SHARE_MAX_COUNT", "100"))
SHARE_MAX_BYTES = int(os.environ.get("NP_SHARE_MAX_BYTES", str(12 * 1024 * 1024)))
SHARE_KEY_PREFIX = os.environ.get("NP_SHARE_KEY_PREFIX", "np_replica:shares")
SHARE_INDEX_KEY = f"{SHARE_KEY_PREFIX}:index"
REDIS_CLIENT = None
ACCOUNT_STORE = AccountStore(ACCOUNT_DB_PATH)
OIDC_AUTH = None


def json_bytes(payload, status=HTTPStatus.OK):
    return (
        status,
        "application/json; charset=utf-8",
        json.dumps(payload, ensure_ascii=False, indent=2).encode("utf-8"),
    )


def log_event(message):
    print(f"[np-replica] {time.strftime('%Y-%m-%d %H:%M:%S')} {message}", flush=True)


def detected_worker_memory_gb():
    if WORKER_MEMORY_GB > 0:
        return WORKER_MEMORY_GB
    try:
        raw = Path("/sys/fs/cgroup/memory.max").read_text().strip()
        if raw != "max":
            return int(raw) / (1024 ** 3)
    except (OSError, ValueError):
        pass
    try:
        return os.sysconf("SC_PAGE_SIZE") * os.sysconf("SC_PHYS_PAGES") / (1024 ** 3)
    except (OSError, ValueError):
        return 0.0


def detected_worker_cpu_count():
    capacities = [max(1, os.cpu_count() or 1)]
    try:
        capacities.append(max(1, len(os.sched_getaffinity(0))))
    except (AttributeError, OSError):
        pass

    quota_paths = (
        (Path("/sys/fs/cgroup/cpu.max"), "v2"),
        (Path("/sys/fs/cgroup/cpu/cpu.cfs_quota_us"), "v1"),
    )
    for path, version in quota_paths:
        try:
            raw = path.read_text().strip()
            if version == "v2":
                quota_text, period_text = raw.split()[:2]
                if quota_text == "max":
                    break
                quota, period = int(quota_text), int(period_text)
            else:
                quota = int(raw)
                if quota < 0:
                    break
                period = int(Path("/sys/fs/cgroup/cpu/cpu.cfs_period_us").read_text().strip())
            if quota > 0 and period > 0:
                capacities.append(max(1, quota // period))
            break
        except (OSError, ValueError):
            continue
    return min(capacities)


def worker_resource_plan():
    cpu_count = detected_worker_cpu_count()
    memory_gb = detected_worker_memory_gb()
    if WORKER_CONCURRENCY != "auto":
        try:
            concurrency = max(1, int(WORKER_CONCURRENCY))
        except ValueError as exc:
            raise ValueError("NP_WORKER_CONCURRENCY must be a positive integer or 'auto'.") from exc
        source = "manual"
    else:
        cpu_capacity = max(1, cpu_count // PER_JOB_THREAD_LIMIT)
        usable_memory_gb = max(0.1, memory_gb - WORKER_MEMORY_RESERVE_GB) if memory_gb else 0
        memory_capacity = max(1, int(usable_memory_gb // ESTIMATED_JOB_MEMORY_GB)) if usable_memory_gb else cpu_capacity
        concurrency = max(1, min(cpu_capacity, memory_capacity))
        source = "auto"
    return {
        "concurrency": concurrency,
        "source": source,
        "cpu_count": cpu_count,
        "per_job_threads": PER_JOB_THREAD_LIMIT,
        "memory_gb": round(memory_gb, 3),
        "memory_reserve_gb": WORKER_MEMORY_RESERVE_GB,
        "estimated_job_memory_gb": ESTIMATED_JOB_MEMORY_GB,
    }


def published_worker_resource_plan():
    if redis_enabled():
        try:
            raw = redis_client().get(WORKER_RESOURCE_KEY)
            if raw:
                return json.loads(raw)
        except Exception:
            pass
    return worker_resource_plan()


def apply_thread_limits():
    # Keep native math libraries and NUPACK's own executor from using all CPU cores inside one job.
    thread_limit_int = max(1, PER_JOB_THREAD_LIMIT)
    thread_limit = str(thread_limit_int)
    for key in (
        "OMP_NUM_THREADS",
        "OPENBLAS_NUM_THREADS",
        "MKL_NUM_THREADS",
        "NUMEXPR_NUM_THREADS",
        "VECLIB_MAXIMUM_THREADS",
        "GOTO_NUM_THREADS",
    ):
        os.environ[key] = thread_limit
    nupack_config.threads = thread_limit_int
    nupack_config.cache = max(0.1, NUPACK_CACHE_GB)


def html_bytes(path):
    body = path.read_bytes()
    mime, _ = mimetypes.guess_type(str(path))
    return HTTPStatus.OK, mime or "text/html; charset=utf-8", body


def file_etag(path):
    stat = path.stat()
    return f'W/"{stat.st_mtime_ns:x}-{stat.st_size:x}"'


def redis_enabled():
    return bool(REDIS_URL and redis is not None)


def redis_client():
    global REDIS_CLIENT
    if not redis_enabled():
        return None
    if REDIS_CLIENT is None:
        REDIS_CLIENT = redis.Redis.from_url(REDIS_URL, decode_responses=True)
    return REDIS_CLIENT


def create_admin_session():
    token = secrets.token_urlsafe(32)
    if redis_enabled():
        redis_client().setex(f"{ADMIN_SESSION_PREFIX}:{token}", ADMIN_SESSION_TTL_SECONDS, "1")
    else:
        with ADMIN_SESSION_LOCK:
            ADMIN_SESSION_STORE[token] = time.time() + ADMIN_SESSION_TTL_SECONDS
    return token


def valid_admin_session(token):
    if not token:
        return False
    if redis_enabled():
        return bool(redis_client().get(f"{ADMIN_SESSION_PREFIX}:{token}"))
    now = time.time()
    with ADMIN_SESSION_LOCK:
        expired = [key for key, expires_at in ADMIN_SESSION_STORE.items() if expires_at <= now]
        for key in expired:
            ADMIN_SESSION_STORE.pop(key, None)
        return ADMIN_SESSION_STORE.get(token, 0) > now


def delete_admin_session(token):
    if not token:
        return
    if redis_enabled():
        redis_client().delete(f"{ADMIN_SESSION_PREFIX}:{token}")
    else:
        with ADMIN_SESSION_LOCK:
            ADMIN_SESSION_STORE.pop(token, None)


def admin_cookie_header(token, *, clear=False):
    secure = "; Secure" if ADMIN_COOKIE_SECURE else ""
    if clear:
        return f"{ADMIN_COOKIE_NAME}=; Path=/; Max-Age=0; HttpOnly; SameSite=Strict{secure}"
    return (
        f"{ADMIN_COOKIE_NAME}={token}; Path=/; Max-Age={ADMIN_SESSION_TTL_SECONDS}; "
        f"HttpOnly; SameSite=Strict{secure}"
    )


OIDC_AUTH = OIDCAuth(redis_client)


def queue_size():
    if not redis_enabled():
        with JOB_LOCK:
            return sum(1 for job in JOB_STORE.values() if job.get("status") in {"queued", "running"})
    try:
        return int(redis_client().llen(JOB_QUEUE_KEY))
    except Exception:
        return None


def prune_redis_running_jobs():
    if not redis_enabled():
        return 0
    client = redis_client()
    removed = 0
    try:
        for job_id in client.smembers(JOB_RUNNING_KEY):
            raw = client.get(job_key(job_id))
            if raw is None:
                client.srem(JOB_RUNNING_KEY, job_id)
                removed += 1
                continue
            try:
                if json.loads(raw).get("status") != "running":
                    client.srem(JOB_RUNNING_KEY, job_id)
                    removed += 1
            except Exception:
                client.srem(JOB_RUNNING_KEY, job_id)
                removed += 1
    except Exception:
        return removed
    return removed


def job_stats():
    stats = {
        "queued": 0,
        "running": 0,
        "success": 0,
        "error": 0,
    }
    if redis_enabled():
        prune_redis_running_jobs()
        client = redis_client()
        stats["queued"] = int(client.llen(JOB_QUEUE_KEY))
        stats["running"] = int(client.scard(JOB_RUNNING_KEY))
        return stats

    with JOB_LOCK:
        prune_jobs()
        for job in JOB_STORE.values():
            status = job.get("status")
            if status in stats:
                stats[status] += 1
    return stats


def job_key(job_id):
    return f"np_replica:job:{job_id}"


def job_dedup_key(fingerprint):
    return f"{JOB_DEDUP_PREFIX}:{fingerprint}"


def refresh_redis_job_claim(job_id, fingerprint):
    return redis_client().eval(
        """
        local current = redis.call('GET', KEYS[1])
        if (not current) or current == ARGV[1] then
            redis.call('SET', KEYS[1], ARGV[1], 'EX', ARGV[2])
            return 1
        end
        return 0
        """,
        1,
        job_dedup_key(fingerprint),
        job_id,
        max(60, JOB_TTL_SECONDS),
    )


def delete_redis_job_claim(job_id, fingerprint):
    return redis_client().eval(
        """
        if redis.call('GET', KEYS[1]) == ARGV[1] then
            return redis.call('DEL', KEYS[1])
        end
        return 0
        """,
        1,
        job_dedup_key(fingerprint),
        job_id,
    )


def job_payload_fingerprint(payload, user_id):
    encoded = json.dumps(
        payload,
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
    )
    return hashlib.sha256(f"{user_id}\0{encoded}".encode("utf-8")).hexdigest()


def active_job_status(status):
    return status in {"queued", "running", CANCEL_REQUESTED_STATUS}


def claim_active_job(payload, owner, job_id):
    """Return an existing active duplicate, or claim this payload for job_id."""
    fingerprint = job_payload_fingerprint(payload, owner["user_id"])
    if redis_enabled():
        client = redis_client()
        key = job_dedup_key(fingerprint)
        while True:
            if client.set(key, job_id, nx=True, ex=max(60, JOB_TTL_SECONDS)):
                return job_id, fingerprint, False
            existing_id = client.get(key)
            if not existing_id:
                continue
            existing = get_job_data(existing_id)
            if existing and existing.get("user_id") == owner["user_id"] and active_job_status(existing.get("status")):
                return existing_id, fingerprint, True
            # A terminal or expired job left a stale dedup key behind.
            delete_redis_job_claim(existing_id, fingerprint)

    with JOB_LOCK:
        prune_jobs()
        for existing in JOB_STORE.values():
            if (
                existing.get("user_id") == owner["user_id"]
                and existing.get("dedup_fingerprint") == fingerprint
                and active_job_status(existing.get("status"))
            ):
                return existing["job_id"], fingerprint, True
        reserved_id = JOB_DEDUP_RESERVATIONS.get(fingerprint)
        if reserved_id:
            if reserved_id in JOB_STORE:
                return reserved_id, fingerprint, True
            JOB_DEDUP_RESERVATIONS.pop(fingerprint, None)
        JOB_DEDUP_RESERVATIONS[fingerprint] = job_id
        return job_id, fingerprint, False


def release_job_claim(job_id, fingerprint):
    if redis_enabled():
        delete_redis_job_claim(job_id, fingerprint)
        return
    with JOB_LOCK:
        if JOB_DEDUP_RESERVATIONS.get(fingerprint) == job_id:
            JOB_DEDUP_RESERVATIONS.pop(fingerprint, None)


def replace_job_claim(job_id, fingerprint):
    if redis_enabled():
        redis_client().setex(
            job_dedup_key(fingerprint),
            max(60, JOB_TTL_SECONDS),
            job_id,
        )
        return
    with JOB_LOCK:
        JOB_DEDUP_RESERVATIONS[fingerprint] = job_id


def share_key(share_id):
    return f"{SHARE_KEY_PREFIX}:item:{share_id}"


def normalize_sequence(sequence, material="rna"):
    normalized = re.sub(r"\s+", "", str(sequence or ""))
    if str(material or "").strip().lower() not in MIXED_MATERIAL_PREFIXES:
        return normalized.upper()
    return normalized


def material_rule(material):
    return MATERIAL_RULES.get(str(material or "rna").strip().lower())


def validate_material_salt(material, ion, value):
    rule = material_rule(material)
    if rule is None:
        return
    lower, upper, _default = rule[ion]
    label = "Sodium" if ion == "sodium" else "Magnesium"
    if math.isclose(lower, upper, rel_tol=0.0, abs_tol=1e-12):
        if not math.isclose(value, lower, rel_tol=0.0, abs_tol=1e-12):
            raise ValueError(f"{label} concentration for {material} must be {lower:g} M.")
    elif value < lower or value > upper:
        raise ValueError(
            f"{label} concentration for {material} must be between {lower:g} and {upper:g} M."
        )


def sequence_length(model, sequence):
    try:
        return int(model.alphabet.sequence_length(sequence))
    except Exception as exc:
        raise ValueError(f"Invalid sequence for material {model}: {sequence}") from exc


def validate_analysis_sequence(sequence, material, model, label="Sequence"):
    sequence = normalize_sequence(sequence, material)
    material_name = str(material or "rna").strip().lower()
    prefixes = MIXED_MATERIAL_PREFIXES.get(material_name)
    if prefixes:
        if not re.fullmatch(rf"(?:[{prefixes}][ACGTU]+)+", sequence):
            raise ValueError(
                f"{label} must use explicit lowercase material prefixes "
                f"({', '.join(prefixes)}) followed by A, C, G, T, or U."
            )
    elif not re.fullmatch(r"[ACGTU]+", sequence):
        raise ValueError(f"{label} must contain only A, C, G, T, or U for analysis.")
    try:
        model.alphabet.sequence(sequence)
    except Exception as exc:
        raise ValueError(f"{label} is invalid for material {material_name}.") from exc
    return sequence


def normalize_design_sequence(sequence, material="rna"):
    normalized = re.sub(r"\s+", "", str(sequence or ""))
    if str(material or "rna").strip().lower() in MIXED_MATERIAL_PREFIXES:
        return normalized
    return normalized.upper()


def parse_design_constraint_tokens(sequence, material="rna"):
    material_name = str(material or "rna").strip().lower()
    seq = normalize_design_sequence(sequence, material_name)
    if not seq:
        raise ValueError("Sequence constraint cannot be empty.")
    mixed_prefixes = MIXED_MATERIAL_PREFIXES.get(material_name)
    allowed_prefixes = set((mixed_prefixes or "") + ("w" if mixed_prefixes else ""))
    index = 0
    current_prefix = None
    tokens = []
    while index < len(seq):
        if mixed_prefixes and seq[index] in allowed_prefixes:
            current_prefix = seq[index]
            index += 1
            if index >= len(seq):
                raise ValueError("A material prefix must be followed by an IUPAC constraint.")
        if mixed_prefixes and current_prefix is None:
            expected = ", ".join(sorted(allowed_prefixes))
            raise ValueError(f"Mixed-material constraints must start with a lowercase prefix: {expected}.")
        match = IUPAC_CONSTRAINT_TOKEN.match(seq, index)
        if not match:
            raise ValueError(f"Invalid sequence constraint near: {seq[index:]}")
        token = match.group(0)
        count = int(token[1:] or "1")
        if count < 1:
            raise ValueError("Sequence constraint repeat counts must be positive.")
        tokens.append((current_prefix, token[0], count))
        index = match.end()
    return seq, tokens


def is_valid_iupac_constraint(sequence, material="rna", model=None):
    try:
        seq, _tokens = parse_design_constraint_tokens(sequence, material)
        if model is not None:
            model.alphabet.domain(seq)
    except Exception:
        return False
    return True


def expand_iupac_constraint(sequence, material="rna"):
    _seq, tokens = parse_design_constraint_tokens(sequence, material)
    output = []
    previous_prefix = None
    for prefix, base, count in tokens:
        if prefix is not None and prefix != previous_prefix:
            output.append(prefix)
        output.append(base * count)
        previous_prefix = prefix
    return "".join(output)


def is_mutable_iupac_constraint(sequence, material="rna"):
    _seq, tokens = parse_design_constraint_tokens(sequence, material)
    return any(prefix == "w" or base not in FIXED_BASES for prefix, base, _count in tokens)


def reverse_complement_fixed(sequence, material="rna", model=None):
    material_name = str(material or "rna").strip().lower()
    if model is not None:
        complement = model.alphabet.to_string(model.alphabet.domain(f"~{sequence}"))
        if material_name in {"rna", "rna06", "rna95", "merna06"}:
            complement = complement.replace("T", "U")
        return complement
    table = DNA_COMPLEMENT_BASE if material_name.startswith("dna") else RNA_COMPLEMENT_BASE
    return sequence.upper().translate(table)[::-1]

def parse_domain_tokens(text):
    tokens = [token.strip() for token in re.split(r"[\s,]+", str(text or "").strip()) if token.strip()]
    output = []
    for token in tokens:
        complement = False
        domain_name = token
        if token.startswith("~"):
            complement = True
            domain_name = token[1:]
        elif token.endswith("*"):
            complement = True
            domain_name = token[:-1]
        output.append((domain_name, complement))
    return output


def get_mapping_value(mapping, key, default=None):
    try:
        return mapping[key]
    except Exception:
        return default


def ensure_unit_interval_limits(lower, upper, label):
    if not (0 <= lower <= upper <= 1):
        raise ValueError(f"{label} limits must satisfy 0 <= lower <= upper <= 1.")


def finite_float(value, label):
    try:
        parsed = float(value)
    except (TypeError, ValueError) as exc:
        raise ValueError(f"{label} must be a number.") from exc
    if not math.isfinite(parsed):
        raise ValueError(f"{label} must be finite.")
    return parsed


def unit_to_molar(value, unit):
    if unit not in UNIT_SCALE:
        raise ValueError(f"Unsupported concentration unit: {unit}")
    return finite_float(value, "Concentration") * UNIT_SCALE[unit]


def parse_model_input(model_input):
    material = str(model_input.get("material", "rna") or "rna").strip().lower()
    rule = material_rule(material)
    sodium_default = rule["sodium"][2] if rule else 1.0
    magnesium_default = rule["magnesium"][2] if rule else 0.0
    parsed = {
        "material": material,
        "ensemble": model_input.get("ensemble", "stacking"),
        "celsius": finite_float(model_input.get("celsius", 37.0), "Temperature"),
        "sodium": finite_float(model_input.get("sodium", sodium_default), "Sodium concentration"),
        "magnesium": finite_float(model_input.get("magnesium", magnesium_default), "Magnesium concentration"),
    }
    if parsed["celsius"] <= -273.15:
        raise ValueError("Temperature must be above absolute zero (-273.15 C).")
    if parsed["sodium"] < 0 or parsed["magnesium"] < 0:
        raise ValueError("Sodium and magnesium concentrations must be non-negative.")
    validate_material_salt(material, "sodium", parsed["sodium"])
    validate_material_salt(material, "magnesium", parsed["magnesium"])
    return parsed


def build_model(model_input):
    return Model(**parse_model_input(model_input))


def build_model_summary(model_input):
    summary = parse_model_input(model_input)
    summary["resolved_material"] = MATERIAL_ALIASES.get(summary["material"], summary["material"])
    summary["nupack_version"] = NUPACK_VERSION
    return summary


def parse_compute(payload):
    compute = payload.get("compute") or ["pfunc", "mfe"]
    compute = [item for item in compute if item in VALID_COMPUTE]
    if not compute:
        raise ValueError("At least one compute option is required.")
    return compute


def parse_options(payload):
    raw = payload.get("options") or {}
    options = {
        "num_sample": int(raw.get("num_sample", 20)),
        "energy_gap": finite_float(raw.get("energy_gap", 1.0), "energy_gap"),
        "sparsity_fraction": finite_float(raw.get("sparsity_fraction", 1.0), "sparsity_fraction"),
        "sparsity_threshold": finite_float(raw.get("sparsity_threshold", 0.0), "sparsity_threshold"),
        "single_mfe": bool(raw.get("single_mfe", False)),
        "indistinguishable_search": bool(raw.get("indistinguishable_search", False)),
        "max_subopt_count": int(raw.get("max_subopt_count", 100000)),
        "result_limit": int(raw.get("result_limit", 25)),
        "pairs_preview_size": int(raw.get("pairs_preview_size", 24)),
    }
    if options["num_sample"] < 0 or options["num_sample"] > 1000:
        raise ValueError("num_sample must be between 0 and 1000.")
    if options["energy_gap"] < 0:
        raise ValueError("energy_gap must be non-negative.")
    if not 0 <= options["sparsity_fraction"] <= 1:
        raise ValueError("sparsity_fraction must be between 0 and 1.")
    if not 0 <= options["sparsity_threshold"] <= 1:
        raise ValueError("sparsity_threshold must be between 0 and 1.")
    if options["max_subopt_count"] < 1 or options["max_subopt_count"] > 1000000:
        raise ValueError("max_subopt_count must be between 1 and 1000000.")
    if options["result_limit"] < 1:
        raise ValueError("result_limit must be at least 1.")
    if options["pairs_preview_size"] < 8 or options["pairs_preview_size"] > 128:
        raise ValueError("pairs_preview_size must be between 8 and 128.")
    return options


def parse_design_options(payload):
    raw = payload.get("design") or {}
    options = {
        "trials": int(raw.get("trials", 1)),
        "result_limit": int(raw.get("result_limit", 25)),
        "off_target_max_size": int(raw.get("off_target_max_size", payload.get("tube", {}).get("max_size", 2))),
        "stop_condition": finite_float(raw.get("stop_condition", 0.02), "design stop_condition"),
        "seed": int(raw.get("seed", 0)),
        "wobble_mutations": bool(raw.get("wobble_mutations", False)),
        "max_time_seconds": int(raw.get("max_time_seconds", 0)),
        "fixed_target_policy": str(raw.get("fixed_target_policy", "include")).strip().lower(),
    }
    if options["trials"] < 1 or options["trials"] > 8:
        raise ValueError("design trials must be between 1 and 8.")
    if options["result_limit"] < 1:
        raise ValueError("design result_limit must be at least 1.")
    if options["off_target_max_size"] < 1 or options["off_target_max_size"] > 8:
        raise ValueError("design off_target_max_size must be between 1 and 8.")
    if not 0 < options["stop_condition"] < 1:
        raise ValueError("design stop_condition must be between 0 and 1.")
    if options["max_time_seconds"] < 0:
        raise ValueError("design max_time_seconds must be non-negative.")
    if options["fixed_target_policy"] not in FIXED_TARGET_POLICIES:
        raise ValueError("design fixed_target_policy must be exclude_from_optimization or include.")
    return options


def euler_phi(value):
    result = value
    factor = 2
    remaining = value
    while factor * factor <= remaining:
        if remaining % factor == 0:
            while remaining % factor == 0:
                remaining //= factor
            result -= result // factor
        factor += 1
    if remaining > 1:
        result -= result // remaining
    return result


def cyclic_complex_count(strand_species, max_size):
    """Count strand-order necklaces enumerated by SetSpec up to max_size."""
    strand_species = max(0, int(strand_species))
    max_size = max(0, int(max_size))
    total = 0
    for size in range(1, max_size + 1):
        rotations = sum(
            euler_phi(divisor) * (strand_species ** (size // divisor))
            for divisor in range(1, size + 1)
            if size % divisor == 0
        )
        total += rotations // size
    return total


def canonical_cyclic_identity(identity):
    """Normalize a strand order using NUPACK's rotation-equivalent complex identity."""
    identity = tuple(identity)
    if not identity:
        return identity
    return min(identity[index:] + identity[:index] for index in range(len(identity)))


def build_design_workload_summary(design_domains, target_rows, tube_rows, target_complex_map, design_options):
    mutable_domain_nucleotides = sum(
        item["length"] if "length" in item else len(expand_iupac_constraint(item["constraint"]))
        for item in design_domains
        if item.get("mutable")
    )
    summary = {
        "mutable_domain_nucleotides": mutable_domain_nucleotides,
        "fixed_target_count": sum(1 for item in target_rows if not item.get("mutable", True)),
        "runtime_limit_seconds": design_options["max_time_seconds"] or "unbounded",
    }
    if not tube_rows:
        return summary

    total_upper_bound = 0
    off_target_upper_bound = 0
    largest_tube_upper_bound = 0
    for tube in tube_rows:
        strand_names = set()
        on_target_identities = set()
        for entry in tube.get("on_targets", []):
            complex_obj = target_complex_map.get(entry["complex"])
            if complex_obj is None:
                continue
            identity = canonical_cyclic_identity(strand.name for strand in complex_obj.strands)
            on_target_identities.add(identity)
            strand_names.update(identity)
        include_identities = set()
        exclude_identities = set()
        for field, identities in (
            ("include_complexes", include_identities),
            ("exclude_complexes", exclude_identities),
        ):
            for line in tube.get(field, []):
                names = tuple(token.strip() for token in line.split("+") if token.strip())
                if names:
                    identity = canonical_cyclic_identity(names)
                    identities.add(identity)
                    strand_names.update(identity)

        max_size = tube["max_size"]
        automatic_count = cyclic_complex_count(len(strand_names), max_size)
        excluded_automatic = {
            identity for identity in exclude_identities
            if len(identity) <= max_size and identity not in on_target_identities
        }
        explicit_outside = {
            identity for identity in on_target_identities.union(include_identities)
            if len(identity) > max_size
        }
        tube_upper_bound = automatic_count - len(excluded_automatic) + len(explicit_outside)
        largest_tube_upper_bound = max(largest_tube_upper_bound, tube_upper_bound)
        total_upper_bound += tube_upper_bound
        off_target_upper_bound += max(0, tube_upper_bound - len(on_target_identities))

    summary.update(
        estimated_complexes_upper_bound=total_upper_bound,
        estimated_off_targets_upper_bound=off_target_upper_bound,
        largest_tube_complexes_upper_bound=largest_tube_upper_bound,
    )
    return summary


def build_strands(strand_payload, model, material):
    if not strand_payload:
        raise ValueError("At least one strand is required.")

    strand_map = {}
    ordered = []

    for row in strand_payload:
        name = (row.get("name") or "").strip()
        sequence = normalize_sequence(row.get("sequence") or "", material)
        if not name:
            raise ValueError("Every strand needs a name.")
        if not sequence:
            raise ValueError(f"Strand {name} is missing a sequence.")
        if name in strand_map:
            raise ValueError(f"Duplicate strand name: {name}")
        sequence = validate_analysis_sequence(sequence, material, model, f"Strand {name}")
        concentration = finite_float(row.get("concentration", 0), f"Strand {name} concentration")
        if concentration < 0:
            raise ValueError(f"Strand {name} concentration must be non-negative.")

        strand = Strand(sequence, name=name)
        strand_map[name] = strand
        ordered.append(
            {
                "name": name,
                "sequence": sequence,
                "concentration": concentration,
                "unit": row.get("unit", "uM"),
                "object": strand,
                "length": sequence_length(model, sequence),
            }
        )

    return strand_map, ordered


def build_design_domains(domain_payload, model=None, material="rna"):
    domain_map = {}
    ordered = []
    for row in domain_payload:
        name = (row.get("name") or "").strip()
        sequence_constraint = normalize_design_sequence(row.get("sequence") or "", material)
        if not name:
            raise ValueError("Every design domain needs a name.")
        if not sequence_constraint:
            raise ValueError(f"Design domain {name} is missing a sequence constraint.")
        if name in domain_map:
            raise ValueError(f"Duplicate domain name: {name}")
        if not is_valid_iupac_constraint(sequence_constraint, material, model):
            raise ValueError(f"Design domain {name} contains unsupported constraint characters.")

        domain = Domain(sequence_constraint, name=name)
        domain_map[name] = domain
        mutable = is_mutable_iupac_constraint(sequence_constraint, material)
        length = int(model.alphabet.domain_length(sequence_constraint)) if model is not None else len(
            expand_iupac_constraint(sequence_constraint, material)
        )
        ordered.append(
            {
                "name": name,
                "constraint": sequence_constraint,
                "object": domain,
                "mutable": mutable,
                "fixed_sequence": None if mutable else expand_iupac_constraint(sequence_constraint, material),
                "length": length,
            }
        )
    return domain_map, ordered


def parse_domain_composition(text, domain_map):
    tokens = [token.strip() for token in re.split(r"[\s,]+", str(text or "").strip()) if token.strip()]
    if not tokens:
        raise ValueError("Domain composition cannot be empty.")

    domains = []
    for token in tokens:
        complement = False
        domain_name = token
        if token.startswith("~"):
            complement = True
            domain_name = token[1:]
        elif token.endswith("*"):
            complement = True
            domain_name = token[:-1]

        if not domain_name:
            raise ValueError(f"Invalid domain token: {token}")
        if domain_name not in domain_map:
            raise ValueError(f"Unknown domain name in strand composition: {domain_name}")

        domain = domain_map[domain_name]
        domains.append(~domain if complement else domain)
    return domains


def build_design_strands(strand_payload, domain_map=None, domain_rows=None, material="rna", model=None):
    if not strand_payload:
        raise ValueError("At least one design strand is required.")

    domain_map = domain_map or {}
    domain_info = {item["name"]: item for item in (domain_rows or [])}
    use_domain_composition = bool(domain_map)
    target_strand_map = {}
    ordered = []

    for row in strand_payload:
        name = (row.get("name") or "").strip()
        raw_definition = str(row.get("sequence") or "").strip()
        sequence_constraint = normalize_design_sequence(raw_definition, material)
        if not name:
            raise ValueError("Every design strand needs a name.")
        if not raw_definition:
            raise ValueError(f"Design strand {name} is missing a strand definition.")
        if name in target_strand_map:
            raise ValueError(f"Duplicate strand name: {name}")

        if use_domain_composition:
            try:
                strand_domains = parse_domain_composition(raw_definition, domain_map)
            except ValueError:
                if (" " in raw_definition) or ("," in raw_definition) or ("~" in raw_definition) or ("*" in raw_definition):
                    raise
                if not is_valid_iupac_constraint(sequence_constraint, material, model):
                    raise
                inline_domain = Domain(sequence_constraint, name=name)
                strand_domains = [inline_domain]
                target_strand = TargetStrand(strand_domains, name=name)
                constraint_kind = "sequence_constraint"
                constraint_value = sequence_constraint
                mutable = is_mutable_iupac_constraint(sequence_constraint, material)
                fixed_sequence = None if mutable else expand_iupac_constraint(sequence_constraint, material)
            else:
                inline_domain = None
                target_strand = TargetStrand(strand_domains, name=name)
                constraint_kind = "domain_composition"
                constraint_value = raw_definition
                fixed_parts = []
                mutable = False
                for domain_name, complement in parse_domain_tokens(raw_definition):
                    info = domain_info.get(domain_name)
                    if info is None:
                        mutable = True
                        fixed_parts = []
                        break
                    if info.get("mutable"):
                        mutable = True
                        fixed_parts = []
                        break
                    part = info.get("fixed_sequence") or ""
                    fixed_parts.append(reverse_complement_fixed(part, material, model) if complement else part)
                fixed_sequence = None if mutable else "".join(fixed_parts)
        else:
            if not is_valid_iupac_constraint(sequence_constraint, material, model):
                raise ValueError(f"Design strand {name} contains unsupported constraint characters.")
            inline_domain = Domain(sequence_constraint, name=name)
            strand_domains = [inline_domain]
            target_strand = TargetStrand(strand_domains, name=name)
            constraint_kind = "sequence_constraint"
            constraint_value = sequence_constraint
            mutable = is_mutable_iupac_constraint(sequence_constraint, material)
            fixed_sequence = None if mutable else expand_iupac_constraint(sequence_constraint, material)

        target_strand_map[name] = target_strand
        ordered.append(
            {
                "name": name,
                "definition": raw_definition,
                "constraint": constraint_value,
                "constraint_kind": constraint_kind,
                "object": target_strand,
                "domains": strand_domains,
                "domain": inline_domain,
                "mutable": mutable,
                "fixed_sequence": fixed_sequence,
            }
        )

    return target_strand_map, ordered


def resolve_scope_reference(expr, domain_map, strand_map, *, allow_global=False):
    text = str(expr or "").strip()
    if not text:
        if allow_global:
            return None
        raise ValueError("Constraint scope cannot be empty.")

    if text in strand_map:
        return strand_map[text]
    return parse_domain_composition(text, domain_map)


def resolve_domain_scope(expr, domain_map):
    text = str(expr or "").strip()
    if not text:
        raise ValueError("Domain scope cannot be empty.")
    return parse_domain_composition(text, domain_map)


def resolve_target_complex_scope(expr, target_complex_map):
    text = str(expr or "").strip()
    if not text:
        raise ValueError("Target complex scope cannot be empty.")
    tokens = [token.strip() for token in re.split(r"[\s,]+", text) if token.strip()]
    if not tokens:
        raise ValueError("Target complex scope cannot be empty.")
    missing = [token for token in tokens if token not in target_complex_map]
    if missing:
        raise ValueError(f"Unknown target complex name(s): {', '.join(missing)}")
    return [target_complex_map[token] for token in tokens]


def parse_sequence_list(text, material="rna", model=None):
    if isinstance(text, list):
        values = text
    else:
        normalized = str(text or "").replace("|", "\n").replace(";", "\n")
        values = normalized.splitlines()
    output = [normalize_design_sequence(item, material) for item in values if item and item.strip()]
    if not output:
        raise ValueError("Constraint sequence list cannot be empty.")
    for seq in output:
        if not is_valid_iupac_constraint(seq, material, model):
            raise ValueError(f"Invalid sequence source: {seq!r}")
    return output


def parse_pattern_list(text, material="rna", model=None):
    if isinstance(text, list):
        values = text
    else:
        values = re.split(r"[\n,]+", str(text or ""))
    output = [normalize_design_sequence(item, material) for item in values if item and item.strip()]
    if not output:
        raise ValueError("Pattern constraint requires at least one pattern.")
    for pattern in output:
        if not is_valid_iupac_constraint(pattern, material, model):
            raise ValueError(f"Invalid pattern: {pattern!r}")
    return output


def parse_catalog_list(text, material="rna", model=None):
    raw_lines = [line.strip() for line in str(text or "").splitlines() if line.strip()]
    if not raw_lines:
        raise ValueError("Library constraint requires at least one catalog row.")
    catalog = []
    for line in raw_lines:
        library = [normalize_design_sequence(item, material) for item in re.split(r"[,|]+", line) if item.strip()]
        if not library:
            raise ValueError("Library constraint contains an empty catalog row.")
        for seq in library:
            if not is_valid_iupac_constraint(seq, material, model):
                raise ValueError(f"Invalid library sequence: {seq!r}")
        catalog.append(library)
    return catalog


def build_hard_constraints(payload, domain_map, strand_map, material="rna", model=None):
    constraints = []
    for index, row in enumerate(payload.get("hard_constraints") or [], start=1):
        constraint_type = (row.get("type") or "").strip().lower()
        if not constraint_type:
            continue

        try:
            def require_text(field, label):
                value = str(row.get(field) or "").strip()
                if not value:
                    raise ValueError(f"{label} is required.")
                return value

            if constraint_type == "match":
                constraints.append(
                    Match(
                        resolve_scope_reference(require_text("left", "Match left scope"), domain_map, strand_map),
                        resolve_scope_reference(require_text("right", "Match right scope"), domain_map, strand_map),
                    )
                )
            elif constraint_type == "complementarity":
                constraints.append(
                    Complementarity(
                        resolve_scope_reference(require_text("left", "Complementarity left scope"), domain_map, strand_map),
                        resolve_scope_reference(require_text("right", "Complementarity right scope"), domain_map, strand_map),
                        wobble_mutations=bool(row.get("wobble_mutations", False)),
                    )
                )
            elif constraint_type == "diversity":
                word = int(row.get("word", 4))
                types = int(row.get("types", 2))
                if word < 1:
                    raise ValueError("Diversity word must be >= 1.")
                if types < 1 or types > 4:
                    raise ValueError("Diversity types must be between 1 and 4.")
                kwargs = {
                    "word": word,
                    "types": types,
                }
                scope = resolve_scope_reference(row.get("scope"), domain_map, strand_map, allow_global=True)
                if scope is not None:
                    kwargs["scope"] = scope
                constraints.append(Diversity(**kwargs))
            elif constraint_type == "similarity":
                lower = float(row.get("min_fraction", 0.0))
                upper = float(row.get("max_fraction", 1.0))
                ensure_unit_interval_limits(lower, upper, "Similarity")
                reference = normalize_design_sequence(row.get("reference") or "", material)
                if not is_valid_iupac_constraint(reference, material, model):
                    raise ValueError("Similarity reference must be a valid IUPAC constraint.")
                constraints.append(
                    Similarity(
                        resolve_scope_reference(require_text("scope", "Similarity scope"), domain_map, strand_map),
                        reference,
                        limits=[lower, upper],
                    )
                )
            elif constraint_type == "window":
                constraints.append(
                    Window(
                        resolve_scope_reference(require_text("scope", "Window scope"), domain_map, strand_map),
                        sources=parse_sequence_list(row.get("sources") or "", material, model),
                    )
                )
            elif constraint_type == "library":
                constraints.append(
                    Library(
                        resolve_scope_reference(require_text("scope", "Library scope"), domain_map, strand_map),
                        catalog=parse_catalog_list(row.get("catalog") or "", material, model),
                    )
                )
            elif constraint_type == "pattern":
                kwargs = {"patterns": parse_pattern_list(row.get("patterns") or "", material, model)}
                scope = resolve_scope_reference(row.get("scope"), domain_map, strand_map, allow_global=True)
                if scope is not None:
                    kwargs["scope"] = scope
                constraints.append(Pattern(**kwargs))
            else:
                raise ValueError(f"Unsupported hard constraint type: {constraint_type}")
        except Exception as exc:
            raise ValueError(f"Invalid hard constraint #{index}: {exc}") from exc

    return constraints


def build_soft_constraints(payload, domain_map, strand_map, target_complex_map, material="rna", model=None):
    constraints = []
    for index, row in enumerate(payload.get("soft_constraints") or [], start=1):
        constraint_type = (row.get("type") or "").strip().lower()
        if not constraint_type:
            continue

        try:
            def require_text(field, label):
                value = str(row.get(field) or "").strip()
                if not value:
                    raise ValueError(f"{label} is required.")
                return value

            if constraint_type == "pattern":
                weight = float(row.get("weight", 1.0))
                if weight < 0:
                    raise ValueError("Pattern weight must be non-negative.")
                kwargs = {
                    "patterns": parse_pattern_list(row.get("patterns") or "", material, model),
                    "weight": weight,
                }
                scope = resolve_scope_reference(row.get("scope"), domain_map, strand_map, allow_global=True)
                if scope is not None:
                    kwargs["scope"] = scope
                constraints.append(Pattern(**kwargs))
            elif constraint_type == "similarity":
                lower = float(row.get("min_fraction", 0.0))
                upper = float(row.get("max_fraction", 1.0))
                ensure_unit_interval_limits(lower, upper, "Similarity")
                reference = normalize_design_sequence(row.get("reference") or "", material)
                if not is_valid_iupac_constraint(reference, material, model):
                    raise ValueError("Similarity reference must be a valid IUPAC constraint.")
                weight = float(row.get("weight", 1.0))
                if weight < 0:
                    raise ValueError("Similarity weight must be non-negative.")
                constraints.append(
                    Similarity(
                        resolve_scope_reference(require_text("scope", "Similarity scope"), domain_map, strand_map),
                        reference,
                        limits=[lower, upper],
                        weight=weight,
                    )
                )
            elif constraint_type == "ssm":
                word = int(row.get("word", 4))
                if word < 1:
                    raise ValueError("SSM word must be >= 1.")
                weight = float(row.get("weight", 1.0))
                if weight < 0:
                    raise ValueError("SSM weight must be non-negative.")
                kwargs = {
                    "word": word,
                    "weight": weight,
                }
                scope = resolve_target_complex_scope(row.get("scope"), target_complex_map) if row.get("scope") else None
                if scope is not None:
                    kwargs["scope"] = scope
                constraints.append(SSM(**kwargs))
            elif constraint_type == "energy_match":
                weight = float(row.get("weight", 1.0))
                if weight < 0:
                    raise ValueError("EnergyMatch weight must be non-negative.")
                kwargs = {
                    "domains": resolve_domain_scope(require_text("scope", "EnergyMatch domain scope"), domain_map),
                    "weight": weight,
                }
                if row.get("energy_ref") not in {None, ""}:
                    kwargs["energy_ref"] = float(row.get("energy_ref"))
                constraints.append(EnergyMatch(**kwargs))
            else:
                raise ValueError(f"Unsupported soft constraint type: {constraint_type}")
        except Exception as exc:
            raise ValueError(f"Invalid soft constraint #{index}: {exc}") from exc

    return constraints


def parse_design_complexes(payload, target_strand_map, strand_rows=None):
    target_rows = payload.get("design_complexes") or payload.get("design_targets") or []
    if not target_rows:
        raise ValueError("At least one design target complex is required.")

    targets = []
    target_complex_map = {}
    strand_info = {item["name"]: item for item in (strand_rows or [])}

    for idx, row in enumerate(target_rows, start=1):
        name = (row.get("name") or f"target_{idx}").strip() or f"target_{idx}"
        if name in target_complex_map:
            raise ValueError(f"Duplicate design target complex name: {name}")
        strand_text = (row.get("strands") or "").strip()
        structure = (row.get("structure") or "").strip()
        if not strand_text:
            raise ValueError(f"Design target complex {name} is missing strand membership.")
        if not structure:
            raise ValueError(f"Design target complex {name} is missing a target structure.")

        tokens = [token.strip() for token in strand_text.split("+") if token.strip()]
        if not tokens:
            raise ValueError(f"Design target complex {name} has an invalid strand list.")
        missing = [token for token in tokens if token not in target_strand_map]
        if missing:
            raise ValueError(
                f"Unknown strand name(s) in design target complex {name}: {', '.join(missing)}"
            )

        bonus = float(row.get("bonus", 0) or 0)
        target_complex = TargetComplex(
            [target_strand_map[token] for token in tokens],
            structure,
            bonus=bonus,
            name=name,
        )
        mutable = any(strand_info.get(token, {}).get("mutable", True) for token in tokens)
        target_payload = {
            "name": name,
            "strands": tokens,
            "structure": structure,
            "bonus": bonus,
            "object": target_complex,
            "mutable": mutable,
            "optimization_status": "included",
        }
        targets.append(target_payload)
        target_complex_map[name] = target_complex

    return targets, target_complex_map


def parse_design_set_members(value, target_complex_map, target_strand_map, label):
    members = []
    identities = set()
    lines = [line.strip() for line in str(value or "").splitlines() if line.strip()]
    for index, line in enumerate(lines, start=1):
        if line in target_complex_map:
            member = target_complex_map[line]
            identity = tuple(strand.name for strand in member.strands)
        else:
            names = [token.strip() for token in line.split("+") if token.strip()]
            missing = [name for name in names if name not in target_strand_map]
            if not names or missing:
                detail = f": {', '.join(missing)}" if missing else ""
                raise ValueError(f"Invalid {label} entry #{index}{detail}")
            member = [target_strand_map[name] for name in names]
            identity = tuple(names)
        members.append(member)
        identities.add(identity)
    return members, identities


def lines_from_set_members(members):
    return ["+".join(strand.name for strand in member.strands) if hasattr(member, "strands")
            else "+".join(strand.name for strand in member) for member in members]


def parse_design_tubes(payload, target_rows, target_complex_map, target_strand_map, default_max_size):
    tube_rows = payload.get("design_tubes") or []
    if not tube_rows:
        legacy_targets = payload.get("design_targets") or []
        if legacy_targets:
            on_targets = []
            for row in legacy_targets:
                if row.get("concentration") in {None, ""}:
                    continue
                on_targets.append(
                    {
                        "complex": row.get("name"),
                        "concentration": row.get("concentration"),
                        "unit": row.get("unit", "uM"),
                    }
                )
            if on_targets:
                tube_rows = [{
                    "name": ((payload.get("tube") or {}).get("name") or "design_tube"),
                    "max_size": default_max_size,
                    "on_targets": on_targets,
                }]

    if not tube_rows:
        raise ValueError("Tube design requires at least one target tube.")

    tubes = []
    ordered_rows = []
    tube_name_set = set()
    for index, row in enumerate(tube_rows, start=1):
        tube_name = (row.get("name") or f"tube_{index}").strip() or f"tube_{index}"
        if tube_name in tube_name_set:
            raise ValueError(f"Duplicate target tube name: {tube_name}")
        tube_name_set.add(tube_name)
        max_size = int(row.get("max_size", default_max_size))
        if max_size < 1 or max_size > 8:
            raise ValueError(f"Target tube {tube_name} max_size must be between 1 and 8.")

        on_targets_input = row.get("on_targets") or []
        if not on_targets_input:
            raise ValueError(f"Target tube {tube_name} must include at least one on-target complex.")

        on_targets = {}
        serialized_on_targets = []
        seen_complex_names = set()
        for entry in on_targets_input:
            complex_name = (entry.get("complex") or "").strip()
            if complex_name not in target_complex_map:
                raise ValueError(f"Unknown target complex {complex_name} in tube {tube_name}.")
            if complex_name in seen_complex_names:
                raise ValueError(f"Duplicate on-target complex {complex_name} in tube {tube_name}.")
            seen_complex_names.add(complex_name)
            concentration = finite_float(
                entry.get("concentration", 0),
                f"Target tube {tube_name} concentration",
            )
            unit = entry.get("unit", "uM")
            if concentration <= 0:
                raise ValueError(f"Target tube {tube_name} requires positive target concentrations.")
            concentration_M = unit_to_molar(concentration, unit)
            complex_obj = target_complex_map[complex_name]
            on_targets[complex_obj] = concentration_M
            serialized_on_targets.append(
                {
                    "complex": complex_name,
                    "concentration": concentration,
                    "unit": unit,
                    "target_concentration_M": concentration_M,
                }
            )

        include, include_ids = parse_design_set_members(
            row.get("include_complexes", ""), target_complex_map, target_strand_map,
            f"target tube {tube_name} include list",
        )
        exclude, exclude_ids = parse_design_set_members(
            row.get("exclude_complexes", ""), target_complex_map, target_strand_map,
            f"target tube {tube_name} exclude list",
        )
        overlap = include_ids.intersection(exclude_ids)
        if overlap:
            display = ", ".join("+".join(names) for names in sorted(overlap))
            raise ValueError(f"Target tube {tube_name} complexes cannot be both included and excluded: {display}")

        tube = TargetTube(
            on_targets=on_targets,
            off_targets=SetSpec(max_size=max_size, include=tuple(include), exclude=tuple(exclude)),
            name=tube_name,
        )
        ordered_rows.append(
            {
                "name": tube_name,
                "max_size": max_size,
                "on_targets": serialized_on_targets,
                "include_complexes": lines_from_set_members(include),
                "exclude_complexes": lines_from_set_members(exclude),
                "object": tube,
            }
        )
        tubes.append(tube)

    return ordered_rows, tubes


def payload_with_allowed_design_targets(payload, allowed_target_names):
    allowed = set(allowed_target_names)
    filtered_tubes = []
    for row in payload.get("design_tubes") or []:
        on_targets = [entry for entry in (row.get("on_targets") or []) if (entry.get("complex") or "").strip() in allowed]
        if on_targets:
            tube_row = dict(row)
            tube_row["on_targets"] = on_targets
            filtered_tubes.append(tube_row)

    output = dict(payload)
    output["design_tubes"] = filtered_tubes
    return output


def validate_design_object_names(design_domains, design_strands, target_rows, tube_rows):
    name_map = {}
    for kind, rows in (
        ("domain", design_domains),
        ("strand", design_strands),
        ("target complex", target_rows),
        ("target tube", tube_rows),
    ):
        for row in rows:
            name = (row.get("name") or "").strip()
            if not name:
                continue
            prev = name_map.get(name)
            if prev and prev != kind:
                raise ValueError(
                    f"Object name {name!r} is reused across {prev} and {kind}. "
                    "Design object names must be unique in a job."
                )
            name_map[name] = kind


def build_defect_weights(payload, design_mode, design_domains, design_strands, target_rows, tube_rows):
    rows = payload.get("defect_weights") or []
    if not rows:
        return None

    domain_index = {item["name"]: item["object"] for item in design_domains}
    strand_index = {item["name"]: item["object"] for item in design_strands}
    complex_index = {item["name"]: item["object"] for item in target_rows}
    tube_index = {item["name"]: item["object"] for item in tube_rows}
    weight_base = [item["object"] for item in tube_rows] if design_mode == "tube" else [item["object"] for item in target_rows]
    weights = Weights(weight_base)

    for index, row in enumerate(rows, start=1):
        kind = (row.get("kind") or "").strip().lower()
        value = float(row.get("weight", 1.0))
        if value < 0:
            raise ValueError(f"Invalid defect weight #{index}: weight must be non-negative")
        name = (row.get("name") or "").strip()
        try:
            if kind == "global":
                weights[:] = value
            elif kind == "domain":
                weights[domain_index[name]] = value
            elif kind == "strand":
                weights[:, strand_index[name]] = value
            elif kind == "complex":
                weights[:, :, complex_index[name]] = value
            elif kind == "tube":
                if design_mode != "tube":
                    raise ValueError("tube-level defect weights are only valid in tube design mode")
                weights[:, :, :, tube_index[name]] = value
            else:
                raise ValueError(f"Unsupported defect weight kind: {kind}")
        except KeyError as exc:
            raise ValueError(f"Unknown entity for defect weight #{index}: {name}") from exc
        except Exception as exc:
            raise ValueError(f"Invalid defect weight #{index}: {exc}") from exc

    return weights


def parse_complex_lines(text, strand_map, *, required=True, label="Complex mode", name_prefix="complex"):
    complexes = []
    raw_lines = [line.strip() for line in str(text or "").splitlines() if line.strip()]
    if not raw_lines:
        if required:
            raise ValueError(f"{label} requires at least one complex definition.")
        return complexes

    for idx, line in enumerate(raw_lines, start=1):
        definition_parts = [part.strip() for part in line.split(";") if part.strip()]
        composition = definition_parts[0]
        tokens = [token.strip() for token in composition.split("+") if token.strip()]
        if not tokens:
            raise ValueError(f"Invalid complex definition on line {idx}: {line}")
        bonus = 0.0
        for option in definition_parts[1:]:
            match = re.fullmatch(r"bonus\s*=\s*(.+)", option, flags=re.IGNORECASE)
            if not match:
                raise ValueError(f"Unsupported complex option on line {idx}: {option}")
            bonus = finite_float(match.group(1), f"Complex line {idx} bonus")
        missing = [token for token in tokens if token not in strand_map]
        if missing:
            raise ValueError(
                f"Unknown strand name(s) in complex line {idx}: {', '.join(missing)}"
            )
        strands = [strand_map[token] for token in tokens]
        complexes.append(Complex(strands, name=f"{name_prefix}_{idx}", bonus=bonus))

    return complexes


def build_tube_set_spec(tube_cfg, strand_map, max_size):
    included = parse_complex_lines(
        tube_cfg.get("include_complexes", ""),
        strand_map,
        required=False,
        label="Tube include list",
        name_prefix="included_complex",
    )
    excluded = parse_complex_lines(
        tube_cfg.get("exclude_complexes", ""),
        strand_map,
        required=False,
        label="Tube exclude list",
        name_prefix="excluded_complex",
    )
    overlap = set(included).intersection(excluded)
    if overlap:
        names = ", ".join(sorted(stringify_complex(item) for item in overlap))
        raise ValueError(f"Tube complexes cannot be both included and excluded: {names}")
    return (
        SetSpec(max_size=max_size, include=tuple(included), exclude=tuple(excluded)),
        included,
        excluded,
    )


def seconds_since(started_at):
    return round(time.perf_counter() - started_at, 6)


def build_performance_summary(timings, workload, notes, pipeline):
    measured = {key: round(float(value), 6) for key, value in timings.items()}
    dominant_stage = max(measured, key=measured.get) if measured else None
    return {
        "pipeline": pipeline,
        "timings_seconds": measured,
        "dominant_stage": dominant_stage,
        "workload": workload,
        "notes": notes,
    }

def stringify_complex(complex_obj):
    return " + ".join(strand.name for strand in complex_obj.strands)


def flatten_sequence(complex_obj, alphabet=None, for_plot=False):
    sequences = []
    for strand in complex_obj.strands:
        raw = str(strand)
        if alphabet is not None and for_plot:
            raw = str(alphabet.sequence(raw)).upper()
        sequences.append(raw)
    return "".join(sequences)


def sanitize_structure_for_rnaplot(structure):
    return str(structure).replace("+", "")


def safe_name(name):
    return re.sub(r"[^A-Za-z0-9_.-]+", "_", name).strip("_") or "structure"


def render_rnaplot_svg(complex_obj, structure, suffix, alphabet=None):
    if not ENABLE_RNAPLOT:
        return None

    seq_name = safe_name(f"{stringify_complex(complex_obj)}_{suffix}")
    sequence = flatten_sequence(complex_obj, alphabet=alphabet, for_plot=True)
    structure_text = sanitize_structure_for_rnaplot(structure)

    with tempfile.TemporaryDirectory(prefix="rnaplot-") as tmpdir:
        tmp_path = Path(tmpdir)
        input_path = tmp_path / f"{seq_name}.seq"
        output_path = tmp_path / f"{seq_name}_ss.svg"

        input_path.write_text(
            f">{seq_name}\n{sequence}\n{structure_text}\n",
            encoding="utf-8",
            newline="\n",
        )

        process = subprocess.run(
            [RNA_PLOT_CMD, "-f", "svg", input_path.name],
            cwd=tmpdir,
            capture_output=True,
            text=True,
            timeout=20,
        )

        if process.returncode != 0:
            raise RuntimeError(process.stderr.strip() or "RNAplot failed.")

        if not output_path.exists():
            raise RuntimeError("RNAplot finished without producing an SVG file.")

        return output_path.read_text(encoding="utf-8")


def render_structure_svg(complex_obj, structure, suffix, alphabet=None):
    structure_text = str(structure)
    multistrand = len(complex_obj.strands) > 1 and "+" in structure_text
    should_try_split = STRUCTURE_PLOT_MODE == "split" or (
        STRUCTURE_PLOT_MODE != "rnaplot" and multistrand
    )
    split_error = None

    if should_try_split:
        try:
            return render_split_strands_svg(
                [str(alphabet.sequence(str(strand))).upper() if alphabet is not None else str(strand)
                 for strand in complex_obj.strands],
                structure_text,
                title=safe_name(f"{stringify_complex(complex_obj)}_{suffix}"),
            )
        except Exception as exc:
            split_error = exc
            if STRUCTURE_PLOT_MODE == "split":
                raise RuntimeError(f"Split-strand layout failed: {exc}") from exc

    if STRUCTURE_PLOT_MODE == "split":
        if split_error is not None:
            raise RuntimeError(f"Split-strand layout failed: {split_error}") from split_error
        return None

    try:
        return render_rnaplot_svg(complex_obj, structure, suffix, alphabet=alphabet)
    except Exception as exc:
        if split_error is not None:
            raise RuntimeError(
                f"Split-strand layout failed: {split_error}; RNAplot fallback failed: {exc}"
            ) from exc
        raise


def serialize_structures(complex_obj, items, plot_kind=None, alphabet=None):
    output = []
    for index, item in enumerate(items or []):
        row = {
            "structure": str(item.structure),
            "energy": round(float(item.energy), 6),
            "stack_energy": round(float(item.stack_energy), 6),
        }
        if plot_kind and index == 0:
            try:
                row["rnaplot_svg"] = render_structure_svg(
                    complex_obj, item.structure, plot_kind, alphabet=alphabet
                )
            except Exception as exc:
                row["rnaplot_error"] = str(exc)
        output.append(row)
    return output


def serialize_samples(items):
    return [str(item) for item in (items or [])]


def serialize_pairs(pair_matrix, preview_limit=24):
    if pair_matrix is None:
        return None

    dense = numpy.asarray(pair_matrix.to_array(), dtype=float)
    preview = dense[:preview_limit, :preview_limit]
    return {
        "shape": list(dense.shape),
        "preview": numpy.round(preview, 6).tolist(),
        "preview_size": int(preview_limit),
        "preview_truncated": bool(dense.shape[0] > preview_limit),
    }


def parse_structure_pairs(structure):
    stack = []
    pairs = []
    compact_index = -1
    index_map = []

    for char in str(structure):
        if char == "+":
            continue
        compact_index += 1
        index_map.append(compact_index)
        if char == "(":
            stack.append(compact_index)
        elif char == ")":
            if not stack:
                raise ValueError("Unbalanced structure: missing opening bracket.")
            left = stack.pop()
            pairs.append((left, compact_index))

    if stack:
        raise ValueError("Unbalanced structure: missing closing bracket.")

    pairs.sort()
    return pairs, compact_index + 1


def build_structure_probabilities(structure, pair_matrix):
    if pair_matrix is None:
        return None

    dense = numpy.asarray(pair_matrix.to_array(), dtype=float)
    mfe_pairs, structure_length = parse_structure_pairs(structure)

    if dense.shape[0] != dense.shape[1]:
        return None
    if structure_length != dense.shape[0]:
        return None

    pair_partner = {}
    for left, right in mfe_pairs:
        pair_partner[left] = right
        pair_partner[right] = left

    residue_probabilities = []
    for index in range(structure_length):
        partner = pair_partner.get(index)
        if partner is None:
            probability = float(dense[index, index])
        else:
            probability = float(dense[index, partner])
        residue_probabilities.append(round(probability, 6))

    pair_probabilities = [
        {
            "i": left + 1,
            "j": right + 1,
            "probability": round(float(dense[left, right]), 6),
        }
        for left, right in mfe_pairs
    ]

    return {
        "length": structure_length,
        "residue_probabilities": residue_probabilities,
        "pair_probabilities": pair_probabilities,
    }


def serialize_complex_result(complex_obj, data, pairs_preview_size=24, alphabet=None):
    payload = {
        "name": getattr(complex_obj, "name", None),
        "display": stringify_complex(complex_obj),
        "strand_names": [strand.name for strand in complex_obj.strands],
        "sequence": flatten_sequence(complex_obj),
        "strand_lengths": [
            int(strand.nt(alphabet)) if alphabet is not None else len(str(strand))
            for strand in complex_obj.strands
        ],
        "bonus_kcal_mol": float(getattr(complex_obj, "bonus", 0.0)),
    }

    if data.pfunc is not None:
        payload["pfunc"] = format(data.pfunc, "g") if isinstance(data.pfunc, Decimal) else str(data.pfunc)
    if data.free_energy is not None:
        payload["free_energy_kcal_mol"] = round(float(data.free_energy), 6)
    if data.ensemble_size is not None:
        payload["ensemble_size"] = int(data.ensemble_size)
    if data.mfe_stack is not None:
        payload["mfe_stack_kcal_mol"] = round(float(data.mfe_stack), 6)
    if data.mfe is not None:
        payload["mfe"] = serialize_structures(
            complex_obj, data.mfe, plot_kind="mfe", alphabet=alphabet
        )
    if data.subopt is not None:
        payload["subopt"] = serialize_structures(complex_obj, data.subopt, alphabet=alphabet)
    if data.sample is not None:
        payload["sample"] = serialize_samples(data.sample)
    if data.pairs is not None:
        payload["pairs"] = serialize_pairs(data.pairs, preview_limit=pairs_preview_size)
        if payload.get("mfe"):
            payload["mfe"][0]["pair_probability_annotations"] = build_structure_probabilities(
                payload["mfe"][0]["structure"], data.pairs
            )

    return payload



def serialize_target_defect_rows(df, limit=None):
    if df is None:
        return []
    rows = []
    for record in df.to_dict("records"):
        row = {}
        for key, value in record.items():
            if key in {"tube", "complex"}:
                continue
            if isinstance(value, numpy.floating):
                row[key] = float(value)
            else:
                row[key] = value
        rows.append(row)
    if limit is not None:
        rows = rows[:limit]
    return rows


def serialize_design_result(
    design_result,
    design_mode,
    model,
    model_summary,
    design_options,
    target_rows,
    tube_rows,
    ordered_domains,
    ordered_strands,
):
    analysis_map = getattr(design_result, "to_analysis", {}) or {}
    designed_domains = []
    designed_domain_map = getattr(design_result, "domains", {}) or {}
    for item in ordered_domains:
        designed_domain = designed_domain_map.get(item["object"])
        domain_sequence = str(designed_domain) if designed_domain is not None else item.get("fixed_sequence")
        designed_domains.append(
            {
                "name": item["name"],
                "constraint": item["constraint"],
                "sequence": domain_sequence,
                "length": int(model.alphabet.domain_length(domain_sequence)) if domain_sequence is not None else None,
                "mutable": bool(item.get("mutable", True)),
            }
        )

    designed_strands = []
    designed_strand_by_name = {}
    for item in ordered_strands:
        target_strand = item["object"]
        analysis_strand = get_mapping_value(analysis_map, target_strand)
        sequence = str(analysis_strand) if analysis_strand is not None else item.get("fixed_sequence")
        designed_strands.append(
            {
                "name": item["name"],
                "constraint": item["constraint"],
                "constraint_kind": item.get("constraint_kind", "sequence_constraint"),
                "definition": item.get("definition", item["constraint"]),
                "sequence": sequence,
                "length": sequence_length(model, sequence) if sequence is not None else None,
                "mutable": bool(item.get("mutable", True)),
            }
        )
        if sequence is not None:
            designed_strand_by_name[item["name"]] = sequence

    target_complexes = []
    for target in target_rows:
        target_complex = target["object"]
        analysis_complex = get_mapping_value(analysis_map, target_complex)
        if analysis_complex is not None:
            display = stringify_complex(analysis_complex)
            sequence = flatten_sequence(analysis_complex)
        else:
            display = " + ".join(target["strands"])
            sequence = "".join(designed_strand_by_name.get(name, "") for name in target["strands"])
        target_complexes.append(
            {
                "name": target["name"],
                "display": display,
                "strand_names": list(target["strands"]),
                "structure": target["structure"],
                "bonus": target.get("bonus", 0),
                "sequence": sequence,
                "target_concentration_M": target.get("target_concentration_M"),
                "optimization_status": target.get("optimization_status", "included"),
                "mutable": bool(target.get("mutable", True)),
            }
        )

    concentration_rows = []
    concentration_table = getattr(getattr(design_result, "concentrations", None), "table", None)
    if design_mode == "tube" and concentration_table is not None:
        for record in concentration_table.to_dict("records"):
            concentration_rows.append(
                {
                    "tube_name": record["tube_name"],
                    "complex_name": record["complex_name"],
                    "concentration_M": float(record["concentration"]),
                    "target_concentration_M": float(record["target_concentration"]),
                    "nucleotides": int(record["nucleotides"]),
                }
            )
        concentration_rows.sort(key=lambda row: (row["tube_name"], -row["concentration_M"]))

    tube_results = []
    if design_mode == "tube":
        for tube_row in tube_rows:
            rows = [row for row in concentration_rows if row["tube_name"] == tube_row["name"]]
            tube_results.append({
                "name": tube_row["name"],
                "max_size": tube_row["max_size"],
                "include_complexes": tube_row.get("include_complexes", []),
                "exclude_complexes": tube_row.get("exclude_complexes", []),
                "complex_concentrations": rows[: design_options["result_limit"]],
                "total_complex_concentrations": len(rows),
            })
    displayed_concentration_rows = [
        concentration
        for tube_result in tube_results
        for concentration in tube_result["complex_concentrations"]
    ]

    return {
        "workflow": "design",
        "mode": design_mode,
        "model": model_summary,
        "compute": ["design"],
        "options": design_options,
        "strands": designed_strands,
        "complexes": target_complexes,
        "total_complex_count": len(target_complexes),
        "displayed_complex_count": len(target_complexes),
        "tube": tube_results[0] if tube_results else None,
        "tubes": tube_results,
        "design": {
            "ensemble_defect": float(design_result.ensemble_defect),
            "weighted_ensemble_defect": float(design_result.defects.weighted_ensemble_defect),
            "objective": float(design_objective(design_result)),
            "domains": designed_domains,
            "stats": {
                key: float(value) if isinstance(value, (int, float, numpy.floating)) else value
                for key, value in design_result.stats.items()
            },
            "job_options": {
                "f_stop": design_options["stop_condition"],
                "seed": design_options["seed"],
                "wobble_mutations": design_options["wobble_mutations"],
                "max_time_seconds": design_options["max_time_seconds"],
                "fixed_target_policy": design_options["fixed_target_policy"],
            },
            "optimization": {
                "included_targets": [row["name"] for row in target_rows if row.get("optimization_status") == "included"],
                "excluded_fixed_targets": [row["name"] for row in target_rows if row.get("optimization_status") == "fixed_excluded"],
            },
            "target_tubes": [
                {
                    "name": row["name"],
                    "max_size": row["max_size"],
                    "on_targets": row["on_targets"],
                    "include_complexes": row.get("include_complexes", []),
                    "exclude_complexes": row.get("exclude_complexes", []),
                }
                for row in tube_rows
            ],
            "targets": target_complexes,
            "defects": {
                "tubes": serialize_target_defect_rows(getattr(design_result.defects, "tubes", None)),
                "complexes": serialize_target_defect_rows(getattr(design_result.defects, "complexes", None)),
                "tube_complexes": serialize_target_defect_rows(getattr(design_result.defects, "tube_complexes", None)),
            },
            "concentrations": displayed_concentration_rows,
        },
    }


def design_objective(result):
    objectives = getattr(getattr(result, "defects", None), "objectives", None)
    if objectives is not None and "weighted" in objectives:
        return float(objectives["weighted"].sum())
    return float(result.ensemble_defect)


def serialize_utility_structure_energies(items):
    return [
        {
            "structure": str(item.structure),
            "energy": float(item.energy),
            "stack_energy": float(item.stack_energy),
        }
        for item in items
    ]


def run_utility_payload(payload, model, model_summary, progress_callback=None):
    utility = payload.get("utility") or {}
    operation = str(utility.get("operation", "pfunc")).strip().lower()
    if operation not in VALID_UTILITY_OPERATIONS:
        raise ValueError(f"Unsupported utility operation: {operation}")
    if progress_callback:
        progress_callback("validating", "Validating utility inputs", {"operation": operation})

    rows = payload.get("strands") or []
    material = model_summary["material"]
    sequences = [normalize_sequence(row.get("sequence") or "", material) for row in rows]
    if operation not in {"seq_distance", "struc_distance"}:
        if operation != "des" and (not sequences or any(not sequence for sequence in sequences)):
            raise ValueError(f"{operation} requires at least one sequence.")
        if operation == "des":
            if material in MIXED_MATERIAL_PREFIXES:
                raise ValueError("Mixed-material des utility is not enabled in this release.")
            if any(not sequence or not re.fullmatch(rf"[{IUPAC_CODES}]+", sequence) for sequence in sequences):
                raise ValueError("des received an invalid IUPAC sequence alphabet.")
        else:
            sequences = [
                validate_analysis_sequence(sequence, material, model, f"Utility strand {index + 1}")
                for index, sequence in enumerate(sequences)
            ]

    structure = str(utility.get("structure") or "").strip()
    options = parse_options(payload)
    started = time.perf_counter()
    if progress_callback:
        progress_callback("computing", f"Running NUPACK utility: {operation}", {
            "operation": operation,
            "strand_count": len(sequences),
            "nucleotides": sum(sequence_length(model, sequence) for sequence in sequences),
        })

    output = {}
    if operation == "pfunc":
        partition_function, free_energy = nupack.pfunc(sequences, model)
        output = {"partition_function": str(partition_function), "free_energy_kcal_per_mol": float(free_energy)}
    elif operation == "structure_energy":
        if not structure:
            raise ValueError("structure_energy requires a structure.")
        output = {"energy_kcal_per_mol": float(nupack.structure_energy(
            sequences, structure, model, distinguishable=bool(utility.get("distinguishable", False))
        ))}
    elif operation == "structure_probability":
        if not structure:
            raise ValueError("structure_probability requires a structure.")
        output = {"probability": float(nupack.structure_probability(sequences, structure, model))}
    elif operation == "sample":
        output = {"structures": [str(item) for item in nupack.sample(sequences, options["num_sample"], model)]}
    elif operation == "pairs":
        matrix = nupack.pairs(
            sequences, model,
            sparsity_fraction=options["sparsity_fraction"],
            sparsity_threshold=options["sparsity_threshold"],
        )
        output = {"pairs": serialize_pairs(matrix, preview_limit=options["pairs_preview_size"])}
    elif operation == "mfe":
        output = {"structures": serialize_utility_structure_energies(nupack.mfe(
            sequences,
            model,
            max_subopt_count=options["max_subopt_count"],
            indistinguishable_search=options["indistinguishable_search"],
        ))}
    elif operation == "subopt":
        output = {"structures": serialize_utility_structure_energies(
            nupack.subopt(
                sequences,
                options["energy_gap"],
                model,
                indistinguishable_search=options["indistinguishable_search"],
                max_subopt_count=options["max_subopt_count"],
            )
        )}
    elif operation == "ensemble_size":
        output = {"ensemble_size": int(nupack.ensemble_size(sequences, model))}
    elif operation == "des":
        if not structure:
            raise ValueError("des requires a target structure.")
        output = {"sequences": [str(item) for item in nupack.des(
            structure, strands=sequences or None, model=model
        )]}
    elif operation == "defect":
        if not structure:
            raise ValueError("defect requires a target structure.")
        output = {"defect": float(nupack.defect(structure, sequences, model=model))}
    elif operation == "seq_distance":
        first = normalize_sequence(utility.get("input_a") or "", material)
        second = normalize_sequence(utility.get("input_b") or "", material)
        if not first or not second:
            raise ValueError("seq_distance requires two sequences.")
        first = validate_analysis_sequence(first, material, model, "seq_distance input A")
        second = validate_analysis_sequence(second, material, model, "seq_distance input B")
        output = {"distance": int(model.alphabet.seq_distance(first, second))}
    elif operation == "struc_distance":
        first = str(utility.get("input_a") or "").strip()
        second = str(utility.get("input_b") or "").strip()
        if not first or not second:
            raise ValueError("struc_distance requires two structures.")
        output = {"distance": int(nupack.struc_distance(first, second))}

    compute_seconds = seconds_since(started)
    if progress_callback:
        progress_callback("serializing", "Serializing utility result", {"operation": operation})
    result = {
        "workflow": "utilities",
        "mode": "utility",
        "model": model_summary,
        "compute": [operation],
        "operation": operation,
        "options": options,
        "strands": [{"name": row.get("name") or f"strand_{i + 1}", "sequence": sequence}
                    for i, (row, sequence) in enumerate(zip(rows, sequences))],
        "utility": output,
        "complexes": [],
        "total_complex_count": 0,
        "displayed_complex_count": 0,
    }
    result["performance"] = build_performance_summary(
        {"nupack_compute": compute_seconds},
        {
            "strand_count": len(sequences),
            "input_nucleotides": sum(sequence_length(model, item) for item in sequences),
        },
        [],
        f"utility_{operation}",
    )
    return result


def sort_concentrations(complex_concentrations):
    rows = []
    for complex_obj, value in complex_concentrations.items():
        rows.append(
            {
                "display": stringify_complex(complex_obj),
                "strand_names": [strand.name for strand in complex_obj.strands],
                "concentration_M": float(value),
            }
        )
    rows.sort(key=lambda row: row["concentration_M"], reverse=True)
    return rows


def run_job_payload(payload, progress_callback=None):
    job_started = time.perf_counter()
    workflow = payload.get("workflow", "analysis")
    mode = payload.get("mode", "tube")
    model_input = payload.get("model") or {}
    model = build_model(model_input)
    model_summary = build_model_summary(model_input)

    if workflow == "utilities":
        return run_utility_payload(payload, model, model_summary, progress_callback)

    if workflow == "design":
        if progress_callback:
            progress_callback("preparing_design", "Building NUPACK design specification", {
                "mode": mode,
                "trials": int((payload.get("design") or {}).get("trials", 1)),
            })
        design_options = parse_design_options(payload)
        design_job_option_kwargs = {
            "f_stop": design_options["stop_condition"],
            "seed": design_options["seed"],
            "wobble_mutations": design_options["wobble_mutations"],
        }
        if design_options["max_time_seconds"] > 0:
            design_job_option_kwargs["max_time"] = design_options["max_time_seconds"]
        design_job_options = DesignOptions(**design_job_option_kwargs)
        design_domain_map, design_domains = build_design_domains(
            payload.get("design_domains") or [], model=model, material=model_summary["material"]
        )
        target_strand_map, design_strands = build_design_strands(
            payload.get("strands") or [],
            domain_map=design_domain_map,
            domain_rows=design_domains,
            material=model_summary["material"],
            model=model,
        )
        target_rows, target_complex_map = parse_design_complexes(payload, target_strand_map, strand_rows=design_strands)
        optimization_target_rows = target_rows
        optimization_target_complex_map = target_complex_map
        if design_options["fixed_target_policy"] == "exclude_from_optimization":
            optimization_target_rows = [row for row in target_rows if row.get("mutable", True)]
            fixed_names = {row["name"] for row in target_rows if not row.get("mutable", True)}
            for row in target_rows:
                if row["name"] in fixed_names:
                    row["optimization_status"] = "fixed_excluded"
            optimization_target_complex_map = {
                row["name"]: row["object"] for row in optimization_target_rows
            }
            if not optimization_target_rows:
                raise ValueError("Design contains no mutable target complexes after fixed-target filtering.")
        hard_constraints = build_hard_constraints(
            payload, design_domain_map, target_strand_map, model_summary["material"], model
        )
        soft_constraints = build_soft_constraints(
            payload,
            design_domain_map,
            target_strand_map,
            optimization_target_complex_map,
            model_summary["material"],
            model,
        )
        tube_rows = []
        design_tubes = []
        if mode == "tube":
            optimization_payload = payload
            if design_options["fixed_target_policy"] == "exclude_from_optimization":
                optimization_payload = payload_with_allowed_design_targets(
                    payload,
                    {row["name"] for row in optimization_target_rows},
                )
            tube_rows, design_tubes = parse_design_tubes(
                optimization_payload,
                optimization_target_rows,
                optimization_target_complex_map,
                target_strand_map,
                design_options["off_target_max_size"],
            )
        validate_design_object_names(design_domains, design_strands, target_rows, tube_rows)
        defect_weights = build_defect_weights(
            payload,
            mode,
            design_domains,
            design_strands,
            target_rows,
            tube_rows,
        )
        design_workload = build_design_workload_summary(
            design_domains,
            target_rows,
            tube_rows,
            optimization_target_complex_map,
            design_options,
        )

        if mode == "tube":
            tube_name = ((payload.get("tube") or {}).get("name") or "design_tube").strip() or "design_tube"
            design_job = tube_design(
                tubes=design_tubes,
                model=model,
                options=design_job_options,
                hard_constraints=hard_constraints,
                soft_constraints=soft_constraints,
                defect_weights=defect_weights,
            )
        elif mode == "complex":
            design_job = complex_design(
                complexes=[row["object"] for row in optimization_target_rows],
                model=model,
                options=design_job_options,
                hard_constraints=hard_constraints,
                soft_constraints=soft_constraints,
                defect_weights=defect_weights,
            )
        else:
            raise ValueError(f"Unsupported design mode: {mode}")

        setup_seconds = seconds_since(job_started)
        design_started = time.perf_counter()
        if progress_callback:
            progress_callback("optimizing", "Running NUPACK design optimization", {
                "mode": mode,
                "trials": design_options["trials"],
                "target_complexes": len(optimization_target_rows),
                "target_tubes": len(tube_rows),
                "f_stop": design_options["stop_condition"],
                **design_workload,
            })
        results = design_job.run(trials=design_options["trials"])
        design_seconds = seconds_since(design_started)
        best_result = min(results, key=design_objective)
        serialization_started = time.perf_counter()
        if progress_callback:
            progress_callback("serializing", "Serializing best design result", {
                "completed_trials": len(results),
                "best_objective": design_objective(best_result),
            })
        serialized_result = serialize_design_result(
            best_result,
            mode,
            model,
            model_summary,
            design_options,
            target_rows,
            tube_rows,
            design_domains,
            design_strands,
        )
        serialization_seconds = seconds_since(serialization_started)
        notes = []
        if design_options["trials"] > 1:
            notes.append("trials repeats the complete stochastic design search; runtime grows approximately with trial count.")
        if design_options["stop_condition"] < 0.02:
            notes.append("A strict f_stop can keep the optimizer searching much longer when the target defect is hard to reach.")
        if mode == "tube" and design_options["off_target_max_size"] > 2:
            notes.append("off_target_max_size expands the off-target ensemble combinatorially.")
        serialized_result["performance"] = build_performance_summary(
            {
                "input_setup": setup_seconds,
                "nupack_design": design_seconds,
                "serialization": serialization_seconds,
            },
            {
                "domain_count": len(design_domains),
                "strand_count": len(design_strands),
                "target_complex_count": len(target_rows),
                "target_tube_count": len(tube_rows),
                "trials": design_options["trials"],
                "off_target_max_size": design_options["off_target_max_size"],
                "hard_constraint_count": len(payload.get("hard_constraints") or []),
                "soft_constraint_count": len(payload.get("soft_constraints") or []),
                **design_workload,
            },
            notes,
            "tube_design" if mode == "tube" else "complex_design",
        )
        return serialized_result

    if workflow != "analysis":
        raise ValueError(f"Unsupported workflow: {workflow}")

    compute = parse_compute(payload)
    options = parse_options(payload)
    nupack_options = {
        key: value
        for key, value in options.items()
        if key not in {"result_limit", "pairs_preview_size"}
    }
    strand_map, strands = build_strands(
        payload.get("strands") or [], model, model_summary["material"]
    )

    if mode == "tube":
        tube_compute = [item for item in compute if item != "pfunc"]
        tube_cfg = payload.get("tube") or {}
        max_size = int(tube_cfg.get("max_size", 2))
        if max_size < 1 or max_size > 8:
            raise ValueError("max_size must be between 1 and 8.")
        strand_concentrations = {
            item["object"]: unit_to_molar(item["concentration"], item["unit"])
            for item in strands
        }
        set_spec, included_complexes, excluded_complexes = build_tube_set_spec(tube_cfg, strand_map, max_size)
        tube = Tube(
            strands=strand_concentrations,
            complexes=set_spec,
            name=(tube_cfg.get("name") or "tube1").strip() or "tube1",
        )
        setup_seconds = seconds_since(job_started)
        nupack_started = time.perf_counter()
        if progress_callback:
            progress_callback("computing", "Running NUPACK tube analysis", {
                "max_size": max_size,
                "strand_count": len(strands),
                "compute": tube_compute,
            })
        result = tube_analysis([tube], model=model, compute=tube_compute, options=nupack_options)
        nupack_seconds = seconds_since(nupack_started)
        tube_result = result[tube]
        serialization_started = time.perf_counter()
        if progress_callback:
            progress_callback("serializing", "Serializing tube analysis result", {})
        concentration_rows = sort_concentrations(tube_result.complex_concentrations)
        total_complex_count = len(concentration_rows)
        result_limit = options["result_limit"]
        displayed_concentration_rows = concentration_rows[:result_limit]
        concentration_by_display = {
            row["display"]: row["concentration_M"] for row in displayed_concentration_rows
        }
        complexes = []
        for complex_obj, data in result.complexes.items():
            display_name = stringify_complex(complex_obj)
            if display_name not in concentration_by_display:
                continue
            row = serialize_complex_result(
                complex_obj,
                data,
                pairs_preview_size=options["pairs_preview_size"],
                alphabet=model.alphabet,
            )
            row["concentration_M"] = concentration_by_display[row["display"]]
            complexes.append(row)
        complexes.sort(key=lambda row: row["concentration_M"], reverse=True)
        serialization_seconds = seconds_since(serialization_started)
        notes = []
        if max_size > 2:
            notes.append("max_size expands the combinatorial complex ensemble and is usually the main Tube cost driver.")
        if "pairs" in compute:
            notes.append("pairs computes a full complex-ensemble pair matrix for every Tube complex; sparsity reduces returned matrix density.")
        if "subopt" in compute:
            notes.append("subopt result count can grow rapidly as energy_gap increases.")
        if "sample" in compute:
            notes.append(f"sample requested {options['num_sample']} Boltzmann structures per complex.")
        performance = build_performance_summary(
            {
                "input_setup": setup_seconds,
                "nupack_compute": nupack_seconds,
                "serialization_and_plots": serialization_seconds,
            },
            {
                "strand_count": len(strands),
                "input_nucleotides": sum(item["length"] for item in strands),
                "enumerated_complex_count": total_complex_count,
                "max_complex_size": max_size,
            },
            notes,
            "tube_analysis",
        )
        return {
            "workflow": "analysis",
            "mode": "tube",
            "model": model_summary,
            "compute": compute,
            "options": options,
            "strands": [
                {
                    "name": item["name"],
                    "sequence": item["sequence"],
                    "concentration": item["concentration"],
                    "unit": item["unit"],
                    "concentration_M": unit_to_molar(item["concentration"], item["unit"]),
                }
                for item in strands
            ],
            "tube": {
                "name": tube.name,
                "max_size": max_size,
                "include_complexes": [stringify_complex(item) for item in included_complexes],
                "exclude_complexes": [stringify_complex(item) for item in excluded_complexes],
                "fraction_bases_unpaired": (
                    float(tube_result.fraction_bases_unpaired)
                    if tube_result.fraction_bases_unpaired is not None else None
                ),
                "ensemble_pair_fractions": serialize_pairs(
                    tube_result.ensemble_pair_fractions,
                    preview_limit=options["pairs_preview_size"],
                ),
                "complex_concentrations": displayed_concentration_rows,
                "total_complex_concentrations": total_complex_count,
            },
            "complexes": complexes,
            "total_complex_count": total_complex_count,
            "displayed_complex_count": len(complexes),
            "performance": performance,
        }

    if mode == "complex":
        complexes = parse_complex_lines(payload.get("complexes_text", ""), strand_map)
        setup_seconds = seconds_since(job_started)
        nupack_started = time.perf_counter()
        if progress_callback:
            progress_callback("computing", "Running NUPACK complex analysis", {
                "complex_count": len(complexes),
                "compute": compute,
            })
        result = complex_analysis(complexes, model=model, compute=compute, options=nupack_options)
        nupack_seconds = seconds_since(nupack_started)
        serialization_started = time.perf_counter()
        if progress_callback:
            progress_callback("serializing", "Serializing complex analysis result", {})
        rows = [
            serialize_complex_result(
                complex_obj,
                data,
                pairs_preview_size=options["pairs_preview_size"],
                alphabet=model.alphabet,
            )
            for complex_obj, data in result.complexes.items()
        ]
        rows.sort(key=lambda row: row["display"])
        total_complex_count = len(rows)
        serialization_seconds = seconds_since(serialization_started)
        notes = []
        if "subopt" in compute:
            notes.append("subopt result count can grow rapidly as energy_gap increases.")
        if "sample" in compute:
            notes.append(f"sample requested {options['num_sample']} Boltzmann structures per complex.")
        performance = build_performance_summary(
            {
                "input_setup": setup_seconds,
                "nupack_compute": nupack_seconds,
                "serialization_and_plots": serialization_seconds,
            },
            {
                "strand_count": len(strands),
                "input_nucleotides": sum(item["length"] for item in strands),
                "complex_count": total_complex_count,
                "largest_complex_nucleotides": max((item.nt(model.alphabet) for item in complexes), default=0),
            },
            notes,
            "complex_analysis",
        )
        rows = rows[: options["result_limit"]]
        return {
            "workflow": "analysis",
            "mode": "complex",
            "model": model_summary,
            "compute": compute,
            "options": options,
            "strands": [
                {
                    "name": item["name"],
                    "sequence": item["sequence"],
                }
                for item in strands
            ],
            "complexes": rows,
            "total_complex_count": total_complex_count,
            "performance": performance,
            "displayed_complex_count": len(rows),
        }

    raise ValueError(f"Unsupported analysis mode: {mode}")


def prune_jobs(now=None):
    now = now or time.time()
    expired_ids = []
    for job_id, job in JOB_STORE.items():
        updated_at = job.get("updated_at", job.get("created_at", now))
        if now - updated_at > JOB_TTL_SECONDS:
            expired_ids.append(job_id)
    for job_id in expired_ids:
        expired = JOB_STORE.pop(job_id, None)
        fingerprint = (expired or {}).get("dedup_fingerprint")
        if fingerprint and JOB_DEDUP_RESERVATIONS.get(fingerprint) == job_id:
            JOB_DEDUP_RESERVATIONS.pop(fingerprint, None)

    if len(JOB_STORE) > JOB_MAX_COUNT:
        keep_ids = sorted(
            JOB_STORE,
            key=lambda job_id: JOB_STORE[job_id].get("created_at", 0),
            reverse=True,
        )[:JOB_MAX_COUNT]
        keep_set = set(keep_ids)
        for job_id in list(JOB_STORE):
            if job_id not in keep_set:
                removed = JOB_STORE.pop(job_id, None)
                fingerprint = (removed or {}).get("dedup_fingerprint")
                if fingerprint and JOB_DEDUP_RESERVATIONS.get(fingerprint) == job_id:
                    JOB_DEDUP_RESERVATIONS.pop(fingerprint, None)


def set_job_data(job):
    if redis_enabled():
        client = redis_client()
        key = job_key(job["job_id"])
        encoded = json.dumps(job, ensure_ascii=False)
        if job.get("status") in TERMINAL_JOB_STATUSES:
            client.setex(key, JOB_TTL_SECONDS, encoded)
            if job.get("dedup_fingerprint"):
                delete_redis_job_claim(job["job_id"], job["dedup_fingerprint"])
        else:
            # Active jobs may legitimately outlive the result-retention TTL.
            client.set(key, encoded)
            if job.get("dedup_fingerprint"):
                refresh_redis_job_claim(job["job_id"], job["dedup_fingerprint"])
        return
    with JOB_LOCK:
        prune_jobs(job.get("updated_at"))
        JOB_STORE[job["job_id"]] = dict(job)


def get_job_data(job_id, include_payload=False):
    if redis_enabled():
        raw = redis_client().get(job_key(job_id))
        if raw is None:
            return None
        job = json.loads(raw)
        job.pop("dedup_fingerprint", None)
        if not include_payload:
            job.pop("payload", None)
        return job
    with JOB_LOCK:
        prune_jobs()
        job = JOB_STORE.get(job_id)
        if job is None:
            return None
        output = dict(job)
        output.pop("dedup_fingerprint", None)
        if not include_payload:
            output.pop("payload", None)
        return output


def should_keep_cancel_state(current, updates):
    current_status = current.get("status") if current else None
    next_status = updates.get("status")
    if current_status in {CANCEL_REQUESTED_STATUS, CANCELED_STATUS}:
        return next_status not in {CANCEL_REQUESTED_STATUS, CANCELED_STATUS}
    return False


def update_job_data(job_id, **updates):
    if redis_enabled():
        client = redis_client()
        key = job_key(job_id)
        while True:
            pipe = client.pipeline()
            try:
                pipe.watch(key)
                raw = pipe.get(key)
                current = json.loads(raw) if raw is not None else {"job_id": job_id, "created_at": time.time()}
                if should_keep_cancel_state(current, updates):
                    pipe.unwatch()
                    return current
                current.update(updates)
                current["updated_at"] = time.time()
                pipe.multi()
                encoded = json.dumps(current, ensure_ascii=False)
                if current.get("status") in TERMINAL_JOB_STATUSES:
                    pipe.setex(key, JOB_TTL_SECONDS, encoded)
                else:
                    pipe.set(key, encoded)
                pipe.execute()
                if current.get("dedup_fingerprint"):
                    if current.get("status") in TERMINAL_JOB_STATUSES:
                        delete_redis_job_claim(job_id, current["dedup_fingerprint"])
                    else:
                        refresh_redis_job_claim(job_id, current["dedup_fingerprint"])
                return current
            except redis.WatchError:
                continue
            finally:
                pipe.reset()

    with JOB_LOCK:
        current = JOB_STORE.get(job_id)
        if current is None:
            current = {"job_id": job_id, "created_at": time.time()}
        if should_keep_cancel_state(current, updates):
            return dict(current)
        current.update(updates)
        current["updated_at"] = time.time()
        prune_jobs(current["updated_at"])
        JOB_STORE[job_id] = current
        if current.get("status") in TERMINAL_JOB_STATUSES and current.get("dedup_fingerprint"):
            if JOB_DEDUP_RESERVATIONS.get(current["dedup_fingerprint"]) == job_id:
                JOB_DEDUP_RESERVATIONS.pop(current["dedup_fingerprint"], None)
        return dict(current)


def create_job(payload, owner, force_duplicate=False):
    requested_job_id = uuid4().hex
    job_id, dedup_fingerprint, reused = claim_active_job(payload, owner, requested_job_id)
    if reused:
        log_event(f"deduplicated job_id={job_id} user_id={owner['user_id']}")
        if not force_duplicate:
            return job_id, True
        job_id = requested_job_id
        replace_job_claim(job_id, dedup_fingerprint)
    now = time.time()
    job = {
        "job_id": job_id,
        "user_id": owner["user_id"],
        "dedup_fingerprint": dedup_fingerprint,
        "status": "queued",
        "error": None,
        "result": None,
        "created_at": now,
        "updated_at": now,
        "payload": payload,
        "progress": {
            "stage": "queued",
            "message": "Waiting for a worker",
            "details": {},
            "updated_at": now,
        },
    }
    try:
        ACCOUNT_STORE.create_job(job_id, owner, payload, status="queued", created_at=now)
        set_job_data(job)
    except Exception:
        release_job_claim(job_id, dedup_fingerprint)
        raise
    log_event(f"accepted job_id={job_id} user_id={owner['user_id']} mode={payload.get('mode', 'tube')}")

    if redis_enabled():
        redis_client().lpush(JOB_QUEUE_KEY, job_id)
    else:
        thread = threading.Thread(target=_run_job, args=(job_id, payload), daemon=True)
        thread.start()
    return job_id, False


def _job_process_entry(payload, result_queue):
    apply_thread_limits()
    def report_progress(stage, message, details=None):
        result_queue.put({
            "kind": "progress",
            "progress": {
                "stage": stage,
                "message": message,
                "details": details or {},
                "updated_at": time.time(),
            },
        })

    try:
        result_queue.put({
            "kind": "terminal",
            "status": "success",
            "result": run_job_payload(payload, progress_callback=report_progress),
        })
    except BaseException as exc:
        result_queue.put(
            {
                "kind": "terminal",
                "status": "error",
                "error": {
                    "message": str(exc),
                    "traceback": traceback.format_exc(),
                },
            }
        )


def terminate_process(process):
    if not process.is_alive():
        return
    if hasattr(process, "kill"):
        process.kill()
    else:
        process.terminate()
    process.join(timeout=2)
    if process.is_alive():
        process.terminate()
        process.join(timeout=2)


def _run_job(job_id, payload):
    started_at = time.time()
    last_heartbeat_at = started_at
    current = get_job_data(job_id)
    if current and current.get("status") in {CANCEL_REQUESTED_STATUS, CANCELED_STATUS}:
        elapsed = round(time.time() - started_at, 3)
        update_job_data(job_id, status=CANCELED_STATUS, payload=None, elapsed_seconds=elapsed)
        ACCOUNT_STORE.update_job(job_id, CANCELED_STATUS, error={"message": "Job canceled by user."}, elapsed_seconds=elapsed)
        return

    log_event(f"running job_id={job_id}")
    update_job_data(job_id, status="running")
    ACCOUNT_STORE.update_job(job_id, "running")
    if redis_enabled():
        redis_client().sadd(JOB_RUNNING_KEY, job_id)

    ctx = multiprocessing.get_context("spawn")
    result_queue = ctx.Queue(maxsize=32)
    process = ctx.Process(target=_job_process_entry, args=(payload, result_queue), daemon=True)
    message = None

    try:
        process.start()
        while process.is_alive():
            current = get_job_data(job_id)
            if current and current.get("status") in {CANCEL_REQUESTED_STATUS, CANCELED_STATUS}:
                terminate_process(process)
                elapsed = round(time.time() - started_at, 3)
                update_job_data(
                    job_id,
                    status=CANCELED_STATUS,
                    error={"message": "Job canceled by user."},
                    result=None,
                    payload=None,
                    elapsed_seconds=elapsed,
                )
                ACCOUNT_STORE.update_job(
                    job_id, CANCELED_STATUS, error={"message": "Job canceled by user."}, elapsed_seconds=elapsed
                )
                log_event(f"canceled job_id={job_id} elapsed={elapsed}s")
                return
            now = time.time()
            if now - last_heartbeat_at >= JOB_HEARTBEAT_SECONDS:
                update_job_data(job_id, heartbeat_at=now)
                last_heartbeat_at = now
            try:
                incoming = result_queue.get_nowait()
                if incoming.get("kind") == "progress":
                    update_job_data(
                        job_id,
                        progress=incoming["progress"],
                        elapsed_seconds=round(time.time() - started_at, 3),
                    )
                    continue
                message = incoming
                break
            except queue.Empty:
                time.sleep(0.25)

        process.join(timeout=2)
        if message is None:
            while True:
                try:
                    incoming = result_queue.get_nowait()
                    if incoming.get("kind") == "progress":
                        update_job_data(
                            job_id,
                            progress=incoming["progress"],
                            elapsed_seconds=round(time.time() - started_at, 3),
                        )
                    else:
                        message = incoming
                except queue.Empty:
                    break

        current = get_job_data(job_id)
        if current and current.get("status") in {CANCEL_REQUESTED_STATUS, CANCELED_STATUS}:
            elapsed = round(time.time() - started_at, 3)
            update_job_data(
                job_id,
                status=CANCELED_STATUS,
                error={"message": "Job canceled by user."},
                result=None,
                payload=None,
                elapsed_seconds=elapsed,
            )
            ACCOUNT_STORE.update_job(
                job_id, CANCELED_STATUS, error={"message": "Job canceled by user."}, elapsed_seconds=elapsed
            )
            log_event(f"canceled job_id={job_id} elapsed={elapsed}s")
            return

        elapsed = round(time.time() - started_at, 3)
        if message and message.get("status") == "success":
            update_job_data(
                job_id,
                status="success",
                result=message["result"],
                payload=None,
                elapsed_seconds=elapsed,
                progress={
                    "stage": "complete",
                    "message": "Calculation complete",
                    "details": {},
                    "updated_at": time.time(),
                },
            )
            ACCOUNT_STORE.update_job(job_id, "success", result=message["result"], elapsed_seconds=elapsed)
            log_event(f"success job_id={job_id} elapsed={elapsed}s")
        elif message and message.get("status") == "error":
            error = message.get("error") or {"message": "Job failed."}
            update_job_data(
                job_id,
                status="error",
                error=error,
                payload=None,
                elapsed_seconds=elapsed,
                progress={
                    "stage": "error",
                    "message": error.get("message", "Job failed."),
                    "details": {},
                    "updated_at": time.time(),
                },
            )
            ACCOUNT_STORE.update_job(job_id, "error", error=error, elapsed_seconds=elapsed)
            log_event(f"error job_id={job_id} elapsed={elapsed}s message={message.get('error', {}).get('message')}")
        else:
            error = {"message": f"Job process exited with code {process.exitcode}."}
            update_job_data(
                job_id,
                status="error",
                error=error,
                payload=None,
                elapsed_seconds=elapsed,
            )
            ACCOUNT_STORE.update_job(job_id, "error", error=error, elapsed_seconds=elapsed)
            log_event(f"error job_id={job_id} elapsed={elapsed}s exitcode={process.exitcode}")
    finally:
        terminate_process(process)
        result_queue.close()
        result_queue.join_thread()
        if redis_enabled():
            redis_client().srem(JOB_RUNNING_KEY, job_id)


def cancel_job(job_id):
    current = get_job_data(job_id, include_payload=True)
    if current is None:
        return None
    status = current.get("status")
    if status in TERMINAL_JOB_STATUSES:
        output = dict(current)
        output.pop("payload", None)
        return output

    if redis_enabled() and status == "queued":
        redis_client().lrem(JOB_QUEUE_KEY, 0, job_id)

    if status == "queued":
        updated = update_job_data(
            job_id,
            status=CANCELED_STATUS,
            error={"message": "Job canceled by user."},
            result=None,
            payload=None,
        )
        ACCOUNT_STORE.update_job(job_id, CANCELED_STATUS, error={"message": "Job canceled by user."})
    else:
        updated = update_job_data(
            job_id,
            status=CANCEL_REQUESTED_STATUS,
            error={"message": "Cancellation requested."},
        )
        ACCOUNT_STORE.update_job(job_id, CANCEL_REQUESTED_STATUS, error={"message": "Cancellation requested."})
    updated.pop("payload", None)
    return updated


def get_job(job_id):
    job = get_job_data(job_id, include_payload=False)
    if job and job.get("status") in {"queued", "running", CANCEL_REQUESTED_STATUS}:
        job["elapsed_seconds"] = round(time.time() - job.get("created_at", time.time()), 3)
    return job


def prune_shares(now=None):
    if SHARE_MAX_COUNT < 1:
        return
    if redis_enabled():
        client = redis_client()
        extra = client.zcard(SHARE_INDEX_KEY) - SHARE_MAX_COUNT
        if extra <= 0:
            return
        stale_ids = client.zrange(SHARE_INDEX_KEY, 0, extra - 1)
        if stale_ids:
            client.delete(*(share_key(share_id) for share_id in stale_ids))
            client.zrem(SHARE_INDEX_KEY, *stale_ids)
        return

    keep_ids = sorted(
        SHARE_STORE,
        key=lambda share_id: SHARE_STORE[share_id].get("created_at", 0),
        reverse=True,
    )[:SHARE_MAX_COUNT]
    keep_set = set(keep_ids)
    for share_id in list(SHARE_STORE):
        if share_id not in keep_set:
            SHARE_STORE.pop(share_id, None)


def create_share(record):
    payload = record.get("payload")
    result = record.get("result")
    if not isinstance(payload, dict):
        raise ValueError("Share payload must contain an input payload object.")
    if not isinstance(result, dict):
        raise ValueError("Share payload must contain a result object.")

    now = time.time()
    share_id = uuid4().hex[:16]
    item = {
        "id": share_id,
        "created_at": now,
        "created_at_iso": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(now)),
        "payload": payload,
        "result": result,
        "result_summary": record.get("result_summary") or {},
    }

    if redis_enabled():
        client = redis_client()
        client.set(share_key(share_id), json.dumps(item, ensure_ascii=False))
        client.zadd(SHARE_INDEX_KEY, {share_id: now})
        prune_shares(now)
        return item

    with SHARE_LOCK:
        SHARE_STORE[share_id] = item
        prune_shares(now)
        return dict(item)


def get_share(share_id):
    share_id = str(share_id or "").strip()
    if not re.fullmatch(r"[0-9a-fA-F]{8,64}", share_id):
        return None
    if redis_enabled():
        raw = redis_client().get(share_key(share_id))
        return json.loads(raw) if raw else None
    with SHARE_LOCK:
        item = SHARE_STORE.get(share_id)
        return dict(item) if item else None


def public_share_payload(share_id, *, record_access=True):
    metadata = ACCOUNT_STORE.share_metadata(share_id)
    if metadata is None:
        return get_share(share_id)

    live = get_job_data(metadata["job_id"], include_payload=True)
    if live is not None:
        if record_access:
            ACCOUNT_STORE.record_share_access(share_id)
        payload = live.get("payload")
        if payload is None:
            account_job = ACCOUNT_STORE.get_job(metadata["user_id"], metadata["job_id"], include_content=True)
            payload = account_job.get("payload") if account_job else None
        result = live.get("result")
        error = live.get("error")
        return {
            "id": share_id,
            "share_id": share_id,
            "job_id": metadata["job_id"],
            "created_at": metadata["created_at"],
            "status": live.get("status"),
            "updated_at": live.get("updated_at"),
            "elapsed_seconds": live.get("elapsed_seconds"),
            "payload": payload,
            "result": result,
            "error": error,
            "result_summary": build_history_summary(payload, live),
        }

    return ACCOUNT_STORE.resolve_share(share_id, record_access=record_access)


def build_history_summary(payload, fallback=None):
    payload = payload or {}
    fallback = fallback or {}
    model = payload.get("model") or {}
    workflow = payload.get("workflow") or fallback.get("workflow") or "analysis"
    mode = payload.get("mode") or fallback.get("mode") or "tube"
    strands = payload.get("strands") or []
    if workflow == "design":
        complexes = payload.get("design_complexes") or payload.get("design_targets") or []
        tube_sizes = [int(row.get("max_size", 0) or 0) for row in (payload.get("design_tubes") or [])]
        max_size = max(tube_sizes, default=int((payload.get("design") or {}).get("off_target_max_size", 0) or 0))
    else:
        complexes = [line for line in str(payload.get("complexes_text") or "").splitlines() if line.strip()]
        max_size = int((payload.get("tube") or {}).get("max_size", 0) or 0)
    return {
        "workflow": workflow,
        "mode": mode,
        "material": model.get("material", fallback.get("material", "rna")),
        "celsius": float(model.get("celsius", fallback.get("celsius", 37)) or 37),
        "sodium": float(model.get("sodium", fallback.get("sodium", 0)) or 0),
        "magnesium": float(model.get("magnesium", fallback.get("magnesium", 0)) or 0),
        "max_size": max_size,
        "strand_count": len(strands),
        "complex_count": len(complexes),
        "compute": list(
            ([((payload.get("utility") or {}).get("operation") or "pfunc")] if workflow == "utilities" else payload.get("compute"))
            or fallback.get("compute")
            or ([] if workflow != "design" else ["design"])
        ),
        "trials": int((payload.get("design") or {}).get("trials", fallback.get("trials", 0)) or 0),
        "stop_condition": float((payload.get("design") or {}).get("stop_condition", fallback.get("stop_condition", 0)) or 0),
        "max_time_seconds": int((payload.get("design") or {}).get("max_time_seconds", fallback.get("max_time_seconds", 0)) or 0),
    }


def recover_interrupted_jobs():
    client = redis_client()
    recovered = 0
    canceled = 0
    for job_id in client.smembers(JOB_RUNNING_KEY):
        raw = client.get(job_key(job_id))
        if raw is None:
            client.srem(JOB_RUNNING_KEY, job_id)
            continue
        job = json.loads(raw)
        status = job.get("status")
        if status == "running":
            client.lrem(JOB_QUEUE_KEY, 0, job_id)
            update_job_data(
                job_id,
                status="queued",
                recovered_at=time.time(),
                recovery_count=int(job.get("recovery_count", 0)) + 1,
            )
            client.lpush(JOB_QUEUE_KEY, job_id)
            ACCOUNT_STORE.update_job(job_id, "queued")
            recovered += 1
        elif status == CANCEL_REQUESTED_STATUS:
            update_job_data(
                job_id,
                status=CANCELED_STATUS,
                error={"message": "Job canceled while the worker was restarting."},
                result=None,
                payload=None,
            )
            ACCOUNT_STORE.update_job(
                job_id,
                CANCELED_STATUS,
                error={"message": "Job canceled while the worker was restarting."},
            )
            canceled += 1
        client.srem(JOB_RUNNING_KEY, job_id)
    if recovered or canceled:
        log_event(f"recovered jobs queued={recovered} canceled={canceled}")


def run_worker_loop():
    if not redis_enabled():
        raise RuntimeError("Worker mode requires NP_REDIS_URL and the redis package.")

    client = redis_client()
    recover_interrupted_jobs()
    resource_plan = worker_resource_plan()
    resource_plan["published_at"] = time.time()
    worker_count = resource_plan["concurrency"]
    client.setex(WORKER_RESOURCE_KEY, 90, json.dumps(resource_plan))
    log_event(
        f"Starting worker loop on Redis queue {JOB_QUEUE_KEY} "
        f"(concurrency={worker_count}, source={resource_plan['source']}, "
        f"per_job_threads={PER_JOB_THREAD_LIMIT}, memory_gb={resource_plan['memory_gb']}, "
        f"estimated_job_memory_gb={ESTIMATED_JOB_MEMORY_GB})"
    )

    with ThreadPoolExecutor(max_workers=worker_count, thread_name_prefix="np-job") as executor:
        inflight = set()
        resource_published_at = time.time()

        while True:
            if time.time() - resource_published_at >= 30:
                resource_plan["published_at"] = time.time()
                client.setex(WORKER_RESOURCE_KEY, 90, json.dumps(resource_plan))
                resource_published_at = time.time()
            finished = {future for future in inflight if future.done()}
            if finished:
                inflight.difference_update(finished)
                for future in finished:
                    future.result()

            if len(inflight) >= worker_count:
                done, _ = wait(inflight, return_when=FIRST_COMPLETED, timeout=1)
                inflight.difference_update(done)
                for future in done:
                    future.result()
                continue

            item = client.brpop(JOB_QUEUE_KEY, timeout=2)
            if not item:
                continue
            _, job_id = item
            raw = client.get(job_key(job_id))
            if raw is None:
                continue
            job = json.loads(raw)
            if job.get("status") in {CANCEL_REQUESTED_STATUS, CANCELED_STATUS}:
                continue
            payload = job.get("payload")
            if payload is None:
                continue
            inflight.add(executor.submit(_run_job, job_id, payload))


EXAMPLE_PAYLOAD = {
    "workflow": "analysis",
    "mode": "tube",
    "model": {
        "material": "rna",
        "ensemble": "stacking",
        "celsius": 37,
        "sodium": 1.0,
        "magnesium": 0.0,
    },
    "compute": ["pfunc", "mfe", "pairs"],
    "options": {
        "num_sample": 20,
        "energy_gap": 1.0,
        "sparsity_fraction": 1.0,
        "sparsity_threshold": 0.0,
        "single_mfe": False,
        "result_limit": 25,
        "pairs_preview_size": 24,
    },
    "strands": [
        {"name": "A", "sequence": "AGUCUAGGAU", "concentration": 1.0, "unit": "uM"},
        {"name": "B", "sequence": "UUAACCCACG", "concentration": 2.0, "unit": "uM"},
    ],
    "tube": {"name": "tube1", "max_size": 2, "include_complexes": "", "exclude_complexes": ""},
    "complexes_text": "A\nB\nA+B",
    "design": {
        "trials": 1,
        "result_limit": 25,
        "off_target_max_size": 2,
        "stop_condition": 0.02,
        "seed": 0,
        "wobble_mutations": False,
        "max_time_seconds": 0,
        "fixed_target_policy": "include",
    },
    "design_domains": [
        {"name": "a", "sequence": "N10"},
        {"name": "b", "sequence": "N10"},
    ],
    "design_complexes": [
        {
            "name": "AB_target",
            "strands": "A+B",
            "structure": "(10+)10",
        }
    ],
    "design_tubes": [
        {
            "name": "Tube 1",
            "max_size": 2,
            "on_targets": [
                {"complex": "AB_target", "concentration": 1.0, "unit": "uM"},
            ],
        }
    ],
    "hard_constraints": [],
    "soft_constraints": [],
    "defect_weights": [],
}

# Use A=a and B=~a so the full-duplex target "(10+)10" is base-pair consistent.
DESIGN_TUBE_EXAMPLE_PAYLOAD = {
    "workflow": "design",
    "mode": "tube",
    "model": {
        "material": "rna",
        "ensemble": "stacking",
        "celsius": 37,
        "sodium": 1.0,
        "magnesium": 0.0,
    },
    "compute": ["pfunc", "mfe", "pairs"],
    "options": {
        "num_sample": 20,
        "energy_gap": 1.0,
        "sparsity_fraction": 1.0,
        "sparsity_threshold": 0.0,
        "single_mfe": False,
        "result_limit": 25,
        "pairs_preview_size": 24,
    },
    "strands": [
        {"name": "A", "sequence": "a", "concentration": 1.0, "unit": "uM"},
        {"name": "B", "sequence": "~a", "concentration": 1.0, "unit": "uM"},
    ],
    "tube": {"name": "Design Tube", "max_size": 2},
    "complexes_text": "A+B",
    "design": {
        "trials": 1,
        "result_limit": 25,
        "off_target_max_size": 2,
        "stop_condition": 0.05,
        "seed": 1,
        "wobble_mutations": False,
        "max_time_seconds": 0,
        "fixed_target_policy": "include",
    },
    "design_domains": [
        {"name": "a", "sequence": "N10"},
    ],
    "design_complexes": [
        {
            "name": "AB_target",
            "strands": "A+B",
            "structure": "(10+)10",
        }
    ],
    "design_tubes": [
        {
            "name": "Design Tube",
            "max_size": 2,
            "on_targets": [
                {"complex": "AB_target", "concentration": 1.0, "unit": "uM"},
            ],
        }
    ],
    "hard_constraints": [],
    "soft_constraints": [],
    "defect_weights": [],
}

DESIGN_COMPLEX_EXAMPLE_PAYLOAD = {
    **DESIGN_TUBE_EXAMPLE_PAYLOAD,
    "mode": "complex",
    "tube": {"name": "Design Tube", "max_size": 1},
    "design": {
        **DESIGN_TUBE_EXAMPLE_PAYLOAD["design"],
        "off_target_max_size": 1,
    },
    "design_tubes": [],
}

UTILITIES_EXAMPLE_PAYLOAD = {
    **EXAMPLE_PAYLOAD,
    "workflow": "utilities",
    "mode": "utility",
    "strands": [
        {"name": "A", "sequence": "GCGCUUCGCG", "concentration": 0, "unit": "uM"},
    ],
    "utility": {
        "operation": "mfe",
        "structure": "..........",
        "input_a": "",
        "input_b": "",
        "distinguishable": False,
    },
}


def get_example_payload(query):
    params = parse_qs(query)
    workflow = (params.get("workflow") or ["analysis"])[0]
    mode = (params.get("mode") or ["tube"])[0]
    if workflow == "utilities":
        return UTILITIES_EXAMPLE_PAYLOAD
    if workflow == "design":
        if mode == "complex":
            return DESIGN_COMPLEX_EXAMPLE_PAYLOAD
        return DESIGN_TUBE_EXAMPLE_PAYLOAD
    return EXAMPLE_PAYLOAD


class AppHandler(BaseHTTPRequestHandler):
    server_version = "NPReplica/0.1"

    def _session(self):
        if not hasattr(self, "_cached_session"):
            self._cached_session = OIDC_AUTH.current_session(self.headers)
            if self._cached_session:
                ACCOUNT_STORE.upsert_user(self._cached_session["user"])
        return self._cached_session

    def _require_user(self):
        session = self._session()
        if session is None:
            self._respond(*json_bytes({"error": "Authentication required", "login_url": "/auth/login"}, status=HTTPStatus.UNAUTHORIZED))
            return None
        return session["user"]

    def _read_json(self, max_bytes=16 * 1024 * 1024):
        length = int(self.headers.get("Content-Length", "0"))
        if length < 0 or length > max_bytes:
            raise ValueError(f"Request payload is too large. Limit is {max_bytes} bytes.")
        return json.loads(self.rfile.read(length).decode("utf-8"))

    def _redirect(self, location, *, cookie=None):
        headers = {"Location": location}
        if cookie:
            headers["Set-Cookie"] = cookie
        self._respond(HTTPStatus.FOUND, "text/plain; charset=utf-8", b"Redirecting", extra_headers=headers)

    def _require_page_user(self, next_path):
        if self._session() is not None:
            return True
        self._redirect(f"/auth/login?{urlencode({'next': next_path})}")
        return False

    def _admin_session_token(self):
        cookie = SimpleCookie()
        try:
            cookie.load(self.headers.get("Cookie", ""))
        except Exception:
            return None
        morsel = cookie.get(ADMIN_COOKIE_NAME)
        return morsel.value if morsel else None

    def _require_admin(self):
        token = self._admin_session_token()
        if valid_admin_session(token):
            return token
        self._respond(
            *json_bytes({"error": "Admin authentication required"}, status=HTTPStatus.UNAUTHORIZED)
        )
        return None

    def _owned_job(self, user, job_id, include_content=True):
        account_metadata = ACCOUNT_STORE.get_job(
            user["user_id"],
            job_id,
            include_content=False,
        )
        if account_metadata is None:
            return None
        live = get_job(job_id)
        if live is not None:
            live.pop("user_id", None)
            return live
        return ACCOUNT_STORE.get_job(user["user_id"], job_id, include_content=include_content)

    def do_GET(self):
        parsed = urlparse(self.path)

        if parsed.path == "/admin":
            if valid_admin_session(self._admin_session_token()):
                self._redirect("/admin/panel")
            else:
                self._respond_file(ADMIN_LOGIN_PATH, cache_control="private, no-store")
            return

        if parsed.path == "/admin/panel":
            if not valid_admin_session(self._admin_session_token()):
                self._redirect("/admin")
                return
            self._respond_file(ADMIN_PATH, cache_control="private, no-store")
            return

        if parsed.path == "/api/admin/jobs":
            if self._require_admin() is None:
                return
            params = {key: values[0] for key, values in parse_qs(parsed.query).items() if values}
            self._respond(*json_bytes({
                "status": "success",
                **ACCOUNT_STORE.list_all_jobs(params),
                "live": job_stats(),
                "worker": published_worker_resource_plan(),
            }))
            return

        if parsed.path == "/api/admin/settings":
            if self._require_admin() is None:
                return
            retention_seconds = ACCOUNT_STORE.trash_retention_seconds()
            self._respond(*json_bytes({
                "status": "success",
                "trash_retention_seconds": retention_seconds,
                "trash_retention_days": retention_seconds / 86400,
            }))
            return

        if parsed.path == "/auth/login":
            next_path = (parse_qs(parsed.query).get("next") or ["/"])[0]
            try:
                self._redirect(OIDC_AUTH.begin_login(next_path))
            except Exception as exc:
                self._respond(*json_bytes({"error": f"Unable to start login: {exc}"}, status=HTTPStatus.BAD_GATEWAY))
            return

        if parsed.path == "/auth/callback":
            try:
                session_id, user, next_path = OIDC_AUTH.complete_login(parse_qs(parsed.query))
                ACCOUNT_STORE.upsert_user(user)
                self._redirect(next_path, cookie=OIDC_AUTH.cookie_header(session_id))
            except Exception as exc:
                log_event(f"OIDC callback failed: {exc}")
                self._respond(*json_bytes({"error": f"Login failed: {exc}"}, status=HTTPStatus.BAD_REQUEST))
            return

        if parsed.path == "/auth/logout":
            session = self._session()
            try:
                location = OIDC_AUTH.logout_url(session)
            except Exception:
                location = "/"
            OIDC_AUTH.delete_session(session)
            self._redirect(location, cookie=OIDC_AUTH.clear_cookie_header())
            return

        share_page = re.fullmatch(r"/share/([A-Za-z0-9_-]{8,128})", parsed.path)
        if share_page:
            self._respond_file(INDEX_PATH, cache_control="public, max-age=60, stale-while-revalidate=86400")
            return

        legacy_share = (parse_qs(parsed.query).get("share") or [""])[0]
        if parsed.path == "/" and re.fullmatch(r"[A-Za-z0-9_-]{8,128}", legacy_share):
            self._respond_file(INDEX_PATH, cache_control="public, max-age=60, stale-while-revalidate=86400")
            return

        if parsed.path == "/":
            if not self._require_page_user(self.path):
                return
            self._respond_file(HOME_PATH, cache_control="private, no-cache")
            return

        page_paths = {
            "/workspace": INDEX_PATH,
            "/cloud": CLOUD_PATH,
            "/account": ACCOUNT_PATH,
            "/settings": SETTINGS_PATH,
        }
        if parsed.path in page_paths:
            if not self._require_page_user(self.path):
                return
            self._respond_file(page_paths[parsed.path], cache_control="private, no-cache")
            return

        if parsed.path in STATIC_PATHS:
            self._respond_file(STATIC_PATHS[parsed.path], cache_control="public, max-age=3600")
            return

        if parsed.path in {"/favicon.svg", "/favicon.ico"}:
            self._respond_file(FAVICON_PATH, cache_control="public, max-age=86400")
            return

        if parsed.path == "/design-guide.html":
            if not self._require_page_user(self.path):
                return
            self._respond_file(GUIDE_PATH, cache_control="public, max-age=3600")
            return

        if parsed.path == "/health":
            stats = job_stats()
            resource_plan = published_worker_resource_plan()
            self._respond(
                *json_bytes(
                    {
                        "status": "ok",
                        "job_backend": "redis" if redis_enabled() else "memory",
                        "run_mode": RUN_MODE,
                        "queue_depth": queue_size(),
                        "jobs_running": stats["running"],
                        "jobs_queued": stats["queued"],
                        "job_ttl_seconds": JOB_TTL_SECONDS,
                        "worker_concurrency": resource_plan["concurrency"],
                        "worker_concurrency_source": resource_plan["source"],
                        "worker_resource_plan": resource_plan,
                        "per_job_thread_limit": PER_JOB_THREAD_LIMIT,
                        "nupack_threads": int(getattr(nupack_config, "threads", PER_JOB_THREAD_LIMIT)),
                        "nupack_cache_gb": float(getattr(nupack_config, "cache", NUPACK_CACHE_GB)),
                    }
                )
            )
            return

        if parsed.path == "/api/me":
            user = self._require_user()
            if user is None:
                return
            self._respond(*json_bytes({"status": "success", "user": user, "usage": ACCOUNT_STORE.usage(user["user_id"])}))
            return

        if parsed.path == "/api/history":
            user = self._require_user()
            if user is None:
                return
            params = {key: values[0] for key, values in parse_qs(parsed.query).items() if values}
            self._respond(*json_bytes({"status": "success", **ACCOUNT_STORE.list_jobs(user["user_id"], params)}))
            return

        if parsed.path.startswith("/api/history/"):
            user = self._require_user()
            if user is None:
                return
            job_id = parsed.path.rsplit("/", 1)[-1]
            job = ACCOUNT_STORE.get_job(user["user_id"], job_id, include_content=True)
            if job is None:
                self._respond(*json_bytes({"error": "History record not found"}, status=HTTPStatus.NOT_FOUND))
                return
            self._respond(*json_bytes({"status": "success", "item": job}))
            return

        if parsed.path == "/api/account/shares":
            user = self._require_user()
            if user is None:
                return
            self._respond(*json_bytes({"status": "success", "items": ACCOUNT_STORE.list_shares(user["user_id"])}))
            return

        if parsed.path == "/api/example":
            user = self._require_user()
            if user is None:
                return
            self._respond(*json_bytes(get_example_payload(parsed.query)))
            return

        if parsed.path.startswith("/api/jobs/"):
            user = self._require_user()
            if user is None:
                return
            job_id = parsed.path.rsplit("/", 1)[-1]
            job = self._owned_job(user, job_id)
            if job is None:
                self._respond(*json_bytes({"error": "Job not found"}, status=HTTPStatus.NOT_FOUND))
                return
            self._respond(*json_bytes(job))
            return

        if parsed.path.startswith("/api/shares/"):
            share_id = parsed.path.rsplit("/", 1)[-1]
            share = public_share_payload(share_id, record_access=(parse_qs(parsed.query).get("poll") or ["0"])[0] != "1")
            if share is None:
                self._respond(*json_bytes({"error": "Share not found"}, status=HTTPStatus.NOT_FOUND))
                return
            self._respond(*json_bytes({"status": "success", "share": share}))
            return

        self._respond(*json_bytes({"error": "Not found"}, status=HTTPStatus.NOT_FOUND))

    def do_POST(self):
        parsed = urlparse(self.path)
        try:
            if parsed.path == "/api/admin/login":
                body = self._read_json(max_bytes=4096)
                password = str(body.get("password") or "")
                if not ADMIN_TOKEN:
                    self._respond(*json_bytes(
                        {"error": "Admin password is not configured"},
                        status=HTTPStatus.SERVICE_UNAVAILABLE,
                    ))
                    return
                if not secrets.compare_digest(password, ADMIN_TOKEN):
                    self._respond(*json_bytes(
                        {"error": "Admin password is incorrect"},
                        status=HTTPStatus.UNAUTHORIZED,
                    ))
                    return
                token = create_admin_session()
                self._respond(
                    *json_bytes({"status": "success", "url": "/admin/panel"}),
                    extra_headers={"Set-Cookie": admin_cookie_header(token)},
                )
                return

            if parsed.path == "/api/admin/logout":
                delete_admin_session(self._admin_session_token())
                self._respond(
                    *json_bytes({"status": "success"}),
                    extra_headers={"Set-Cookie": admin_cookie_header("", clear=True)},
                )
                return

            if parsed.path == "/api/admin/settings":
                if self._require_admin() is None:
                    return
                body = self._read_json(max_bytes=4096)
                seconds = ACCOUNT_STORE.set_trash_retention_days(body.get("trash_retention_days"))
                self._respond(*json_bytes({
                    "status": "success",
                    "trash_retention_seconds": seconds,
                    "trash_retention_days": seconds / 86400,
                }))
                return

            admin_action = re.fullmatch(
                r"/api/admin/jobs/([A-Za-z0-9_-]{8,128})/(cancel|trash|restore|delete)",
                parsed.path,
            )
            if admin_action:
                if self._require_admin() is None:
                    return
                job_id, action = admin_action.groups()
                if action == "cancel":
                    result = cancel_job(job_id)
                    found = result is not None
                elif action == "trash":
                    found = ACCOUNT_STORE.admin_trash_job(job_id)
                    result = {"job_id": job_id, "deleted": found}
                elif action == "restore":
                    found = ACCOUNT_STORE.restore_job(job_id)
                    result = {"job_id": job_id, "restored": found}
                else:
                    found = ACCOUNT_STORE.permanently_delete_job(job_id)
                    result = {"job_id": job_id, "permanently_deleted": found}
                if not found:
                    self._respond(*json_bytes({"error": "Job not found"}, status=HTTPStatus.NOT_FOUND))
                    return
                self._respond(*json_bytes({"status": "success", "job": result}))
                return
        except Exception as exc:
            self._respond(*json_bytes({"status": "error", "error": str(exc)}, status=HTTPStatus.BAD_REQUEST))
            return

        user = self._require_user()
        if user is None:
            return
        try:
            if parsed.path.startswith("/api/jobs/") and parsed.path.endswith("/cancel"):
                job_id = parsed.path.split("/")[-2]
                if ACCOUNT_STORE.owner_id(job_id) != user["user_id"]:
                    self._respond(*json_bytes({"error": "Job not found"}, status=HTTPStatus.NOT_FOUND))
                    return
                job = cancel_job(job_id)
                if job is None:
                    self._respond(*json_bytes({"error": "Job not found"}, status=HTTPStatus.NOT_FOUND))
                    return
                job.pop("user_id", None)
                self._respond(*json_bytes({"status": "success", "job": job}))
                return

            if parsed.path.startswith("/api/jobs/") and parsed.path.endswith("/shares"):
                job_id = parsed.path.split("/")[-2]
                body = self._read_json()
                expires_in = body.get("expires_in")
                if expires_in not in {None, "", 0}:
                    expires_in = int(expires_in)
                    if expires_in < 60:
                        raise ValueError("Share duration must be at least 60 seconds.")
                else:
                    expires_in = None
                share = ACCOUNT_STORE.create_share(user["user_id"], job_id, expires_in)
                self._respond(*json_bytes({"status": "success", "share": share, "url": f"/share/{share['share_id']}"}, status=HTTPStatus.CREATED))
                return

            if parsed.path == "/api/jobs":
                payload = self._read_json()
                force_duplicate = (parse_qs(parsed.query).get("force") or ["0"])[0] == "1"
                job_id, deduplicated = create_job(payload, user, force_duplicate=force_duplicate)
                if deduplicated:
                    duplicate = self._owned_job(user, job_id, include_content=False) or {"job_id": job_id}
                    self._respond(*json_bytes(
                        {
                            "status": "duplicate",
                            "message": "An identical job is already active for this user.",
                            "duplicate": {
                                key: duplicate.get(key)
                                for key in ("job_id", "status", "created_at", "updated_at", "progress")
                            },
                        },
                        status=HTTPStatus.CONFLICT,
                    ))
                    return
                self._respond(*json_bytes(
                    {"status": "accepted", "job_id": job_id, "forced_duplicate": force_duplicate},
                    status=HTTPStatus.ACCEPTED,
                ))
                return

            if parsed.path == "/api/history/import":
                body = self._read_json(max_bytes=64 * 1024 * 1024)
                entries = body.get("history") if isinstance(body, dict) else None
                if not isinstance(entries, list):
                    raise ValueError("History import requires a history array.")
                imported = ACCOUNT_STORE.import_history(user, entries)
                self._respond(*json_bytes({"status": "success", "imported": imported}))
                return

            if parsed.path == "/api/shares":
                body = self._read_json()
                job_id = str(body.get("job_id") or "")
                if not job_id:
                    raise ValueError("Cloud shares require a job_id.")
                share = ACCOUNT_STORE.create_share(user["user_id"], job_id, body.get("expires_in"))
                self._respond(*json_bytes({"status": "success", "share_id": share["share_id"], "url": f"/share/{share['share_id']}"}, status=HTTPStatus.CREATED))
                return

            if parsed.path == "/api/analyze":
                payload = self._read_json()
                job_id = uuid4().hex
                started = time.time()
                ACCOUNT_STORE.create_job(job_id, user, payload, status="running", created_at=started)
                try:
                    result = run_job_payload(payload)
                except Exception as exc:
                    elapsed = round(time.time() - started, 3)
                    ACCOUNT_STORE.update_job(job_id, "error", error={"message": str(exc), "traceback": traceback.format_exc()}, elapsed_seconds=elapsed)
                    raise
                elapsed = round(time.time() - started, 3)
                ACCOUNT_STORE.update_job(job_id, "success", result=result, elapsed_seconds=elapsed)
                self._respond(*json_bytes({"status": "success", "job_id": job_id, "result": result}))
                return

            self._respond(*json_bytes({"error": "Not found"}, status=HTTPStatus.NOT_FOUND))
        except Exception as exc:
            self._respond(
                *json_bytes(
                    {
                        "status": "error",
                        "error": str(exc),
                        "traceback": traceback.format_exc(),
                    },
                    status=HTTPStatus.BAD_REQUEST,
                )
            )

    def do_PATCH(self):
        parsed = urlparse(self.path)
        user = self._require_user()
        if user is None:
            return
        try:
            if parsed.path.startswith("/api/account/shares/"):
                share_id = parsed.path.rsplit("/", 1)[-1]
                body = self._read_json()
                expires_in = body["expires_in"] if "expires_in" in body else "unchanged"
                share = ACCOUNT_STORE.update_share(
                    user["user_id"], share_id, active=body.get("active"), expires_in=expires_in
                )
                if share is None:
                    self._respond(*json_bytes({"error": "Share not found"}, status=HTTPStatus.NOT_FOUND))
                    return
                self._respond(*json_bytes({"status": "success", "share": share}))
                return
            self._respond(*json_bytes({"error": "Not found"}, status=HTTPStatus.NOT_FOUND))
        except Exception as exc:
            self._respond(*json_bytes({"status": "error", "error": str(exc)}, status=HTTPStatus.BAD_REQUEST))

    def do_DELETE(self):
        parsed = urlparse(self.path)
        user = self._require_user()
        if user is None:
            return
        try:
            if parsed.path.startswith("/api/history/"):
                job_id = parsed.path.rsplit("/", 1)[-1]
                if not ACCOUNT_STORE.delete_job(user["user_id"], job_id):
                    self._respond(*json_bytes({"error": "History record not found"}, status=HTTPStatus.NOT_FOUND))
                    return
                self._respond(*json_bytes({"status": "success"}))
                return
            if parsed.path.startswith("/api/account/shares/"):
                share_id = parsed.path.rsplit("/", 1)[-1]
                share = ACCOUNT_STORE.update_share(user["user_id"], share_id, active=False)
                if share is None:
                    self._respond(*json_bytes({"error": "Share not found"}, status=HTTPStatus.NOT_FOUND))
                    return
                self._respond(*json_bytes({"status": "success", "share": share}))
                return
            self._respond(*json_bytes({"error": "Not found"}, status=HTTPStatus.NOT_FOUND))
        except Exception as exc:
            self._respond(*json_bytes({"status": "error", "error": str(exc)}, status=HTTPStatus.BAD_REQUEST))

    def log_message(self, format_, *args):
        print(f"{self.address_string()} - {format_ % args}")

    def _respond_file(self, path, cache_control):
        etag = file_etag(path)
        if self.headers.get("If-None-Match") == etag:
            self.send_response(HTTPStatus.NOT_MODIFIED)
            self.send_header("ETag", etag)
            self.send_header("Cache-Control", cache_control)
            self.end_headers()
            return
        status, content_type, body = html_bytes(path)
        self._respond(status, content_type, body, cache_control=cache_control, extra_headers={"ETag": etag})

    def _respond(self, status, content_type, body, cache_control=None, extra_headers=None):
        self.send_response(status)
        self.send_header("Content-Type", content_type)
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Cache-Control", cache_control or "no-store, max-age=0")
        if cache_control is None:
            self.send_header("Pragma", "no-cache")
            self.send_header("Expires", "0")
        for key, value in (extra_headers or {}).items():
            self.send_header(key, value)
        self.end_headers()
        self.wfile.write(body)


def main():
    apply_thread_limits()
    ACCOUNT_STORE.initialize()
    if RUN_MODE == "worker":
        run_worker_loop()
        return
    server = ThreadingHTTPServer((HOST, PORT), AppHandler)
    log_event(
        f"Serving NP replica on http://{HOST}:{PORT} "
        f"(jobs via {'redis' if redis_enabled() else 'memory'}, "
        f"per_job_threads={max(1, PER_JOB_THREAD_LIMIT)})"
    )
    server.serve_forever()


if __name__ == "__main__":
    main()