Upload folder using huggingface_hub (part 2)

app/services/workflow_manifest.py

@@ -0,0 +1,66 @@
+"""Portable evidence export. No workflow adapter or execution is implied."""
+from __future__ import annotations
+
+import hashlib
+import json
+
+
+_STATES = {"complete": "COMPLETED", "completed": "COMPLETED", "failed": "FAILED",
+           "skipped": "SKIPPED", "running": "RUNNING", "queued": "QUEUED",
+           "pending": "NOT_EXECUTED"}
+
+
+def build_workflow_manifest(experiment: dict, provenance: list[dict]) -> dict:
+    """Export stored evidence without synthesizing versions or scientific results."""
+    nodes = []
+    gaps = []
+    for field in ("input_hash", "git_commit", "container_hash"):
+        if not experiment.get(field):
+            gaps.append({"node": None, "field": field, "reason": "NOT_RECORDED"})
+    for index, source in enumerate(provenance):
+        node_id = source.get("node_id") or source.get("id")
+        if not node_id:
+            node_id = f"unidentified-record-{index}"
+            gaps.append({"node": node_id, "field": "node_id", "reason": "NOT_RECORDED"})
+        raw_status = source.get("status")
+        node = {"id": node_id, "recorded_status": raw_status,
+                "state": _STATES.get(raw_status, "UNKNOWN"),
+                "tool": source.get("tool"), "tool_version": source.get("tool_version"),
+                "database": source.get("database"), "database_version": source.get("database_version"),
+                "parameters": source.get("params"), "depends_on": source.get("deps") or [],
+                "input_ref": source.get("input_ref"), "output_ref": source.get("output_ref"),
+                "completed_at": source.get("completed_at"), "evidence": source.get("evidence")}
+        nodes.append(node)
+        if node["state"] == "COMPLETED":
+            for field in ("tool", "tool_version", "completed_at"):
+                if not node[field]:
+                    gaps.append({"node": node_id, "field": field, "reason": "NOT_RECORDED"})
+            if node["database"] and not node["database_version"]:
+                gaps.append({"node": node_id, "field": "database_version", "reason": "NOT_RECORDED"})
+    ids = {node["id"] for node in nodes}
+    if len(ids) != len(nodes):
+        gaps.append({"node": None, "field": "node_id", "reason": "DUPLICATE_ID"})
+    edges = []
+    for node in nodes:
+        for dependency in node["depends_on"]:
+            edges.append({"from": dependency, "to": node["id"]})
+            if dependency not in ids:
+                gaps.append({"node": node["id"], "field": "depends_on", "reason": "UNRECORDED_DEPENDENCY", "dependency": dependency})
+    if not nodes:
+        gaps.append({"node": None, "field": "steps", "reason": "NOT_RECORDED"})
+    manifest = {"format": "bionexus-workflow-evidence", "schema_version": 1,
+                "execution": {"executable": False, "adapter": None,
+                              "statement": "Evidence export only; no tools are launched and external workflow compatibility has not been validated."},
+                "experiment_id": experiment.get("experiment_id"), "job_id": experiment.get("job_id"),
+                "pipeline": experiment.get("pipeline"), "recorded_status": experiment.get("status"),
+                "input_sha256": experiment.get("input_hash"), "output_sha256": experiment.get("output_hash"),
+                "git_commit": experiment.get("git_commit"), "container_hash": experiment.get("container_hash"),
+                "software_versions": experiment.get("software_versions"), "database_versions": experiment.get("database_versions"),
+                "environment": experiment.get("environment"), "random_seed": experiment.get("random_seed"),
+                "parameters": experiment.get("parameters"), "started_at": experiment.get("started_at"),
+                "finished_at": experiment.get("finished_at"), "steps": nodes, "edges": edges,
+                "reproducibility_gaps": gaps,
+                "integrity": {"algorithm": "sha256", "serialization": "JSON sorted keys, UTF-8, compact separators; manifest_sha256 excluded"}}
+    canonical = json.dumps(manifest, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
+    manifest["manifest_sha256"] = hashlib.sha256(canonical.encode("utf-8")).hexdigest()
+    return manifest

app/tools/castp.py

@@ -71,60 +71,19 @@ async def _analyze_pockets(pdb_text: str, pdb_id: str, probe_radius: float, meth
     if method_key in {"fpocket", "f_pocket"}:
         fpocket = FPOCKET_BIN if Path(FPOCKET_BIN).exists() else (shutil.which("fpocket") or "")
         if not fpocket or not Path(fpocket).exists():
-            if not _parse_pdb_chains(pdb_text)["chains"]:
-                result = _empty_result(
-                    pdb_id,
-                    probe_radius,
-                    method="fpocket",
-                    status="FAILED",
-                    engine_version="unavailable",
-                    evidence_class="Unsupported/insufficient evidence",
-                    reason="fpocket executable is unavailable; no heuristic was substituted",
-                )
-                _attach_structure_summary(pdb_text, result)
-                return result
-            methods_tried = [{"method": "fpocket", "status": "unavailable"}]
-            try:
-                loop = asyncio.get_running_loop()
-                result = await loop.run_in_executor(None, _analyze_pockets_sasa_sync, pdb_text, pdb_id, probe_radius)
-            except Exception as exc:
-                error = f"{type(exc).__name__}: {exc}"
-                methods_tried.append({"method": "sasa_heuristic", "status": "crashed"})
-                result = _empty_result(
-                    pdb_id,
-                    probe_radius,
-                    method="none",
-                    status="FAILED",
-                    engine_version="fpocket unavailable; sasa_heuristic crashed",
-                    evidence_class="Unsupported/insufficient evidence",
-                    reason=error,
-                )
-                result["methods_tried"] = methods_tried
-                result["fallback_used"] = True
-                result["fallback_method"] = "sasa_heuristic"
-                result["error"] = str(exc)
-                _attach_structure_summary(pdb_text, result)
-                return result
-            heuristic_status = "ran_no_pockets" if not result.get("pockets") else "ran_with_pockets"
-            methods_tried.append({"method": "sasa_heuristic", "status": heuristic_status})
-            result["method"] = "sasa_heuristic"
-            result["engine"] = "BioNexus exploratory SASA heuristic"
-            result["status"] = "DEGRADED"
-            result["methods_tried"] = methods_tried
-            result["fallback_used"] = True
-            result["fallback_method"] = "sasa_heuristic"
-            result["note"] = (
-                "CASTp architecture: fpocket is unavailable in this deployment, so the "
-                "analysis fell back to the BioNexus exploratory SASA heuristic. All "
-                "values are attributed to sasa_heuristic; no fpocket or CASTp values "
-                "were substituted."
+            result = _empty_result(
+                pdb_id, probe_radius, method="fpocket", status="FAILED",
+                engine_version="unavailable", evidence_class="Unsupported/insufficient evidence",
+                reason="fpocket executable is unavailable; no heuristic substitution was performed. Select the exploratory SASA heuristic explicitly to run that method.",
             )
+            result["methods_tried"] = [{"method": "fpocket", "status": "unavailable"}]
             _attach_structure_summary(pdb_text, result)
             return result
         loop = asyncio.get_running_loop()
         result = await loop.run_in_executor(None, _run_fpocket_analysis, fpocket, pdb_text, pdb_id, probe_radius)
         result.setdefault("fallback_used", False)
-        result["methods_tried"] = [{"method": "fpocket", "status": "ok"}]
+        result["methods_tried"] = [{"method": "fpocket", "status": "ok" if result.get("status", "VALID") == "VALID" else "failed"}]
+        _attach_structure_summary(pdb_text, result)
         return result
 
     if method_key in {"heuristic", "sasa", "sasa_heuristic", "bionexus_heuristic"}:
@@ -140,9 +99,10 @@ def _fpocket_version(fpocket_bin: str) -> str:
     for args in ([fpocket_bin, "-v"], [fpocket_bin, "--version"]):
         try:
             completed = subprocess.run(args, capture_output=True, text=True, timeout=10)
-            text = (completed.stdout or completed.stderr or "").strip()
-            if text:
-                return text.splitlines()[0][:120]
+            text = re.sub(r"\x1b\[[0-9;]*m", "", (completed.stdout or "") + "\n" + (completed.stderr or ""))
+            match = re.search(r"\bfpocket\s+(?:version\s+)?(\d+(?:\.\d+)+(?:[-+][\w.]+)?)", text, re.IGNORECASE)
+            if match:
+                return match.group(1)
         except Exception:
             continue
     return "unknown"
@@ -155,7 +115,7 @@ def _run_fpocket_analysis(fpocket_bin: str, pdb_text: str, pdb_id: str, probe_ra
         input_path.write_text(pdb_text, encoding="utf-8")
         try:
             completed = subprocess.run(
-                [fpocket_bin, "-f", str(input_path), "-r", str(probe_radius)],
+                [fpocket_bin, "-f", str(input_path)],
                 capture_output=True,
                 text=True,
                 timeout=60,
@@ -219,24 +179,24 @@ def _parse_fpocket(out_dir: Path, pdb_id: str, probe_radius: float, version: str
                 "alpha_spheres": None,
                 "num_residues": 0,
                 "residues": [],
-                "centroid": [0.0, 0.0, 0.0],
+                "centroid": None,
             }
             continue
         if current is None:
             continue
-        lower = line.lower()
+        descriptor = line.split(":", 1)[0].strip().lower()
         value = _extract_number(line)
         if value is None:
             continue
-        if "druggability score" in lower:
+        if descriptor == "druggability score":
             current["druggability_score"] = value
-        elif "pocket score" in lower or ("score" in lower and "druggability" not in lower):
+        elif descriptor in {"score", "pocket score"}:
             current["score"] = value
-        elif "volume" in lower:
+        elif descriptor == "volume":
             current["volume"] = value
-        elif "surface area" in lower or "area" in lower:
+        elif descriptor in {"total sasa", "surface area", "area"}:
             current["area"] = value
-        elif "alpha sphere" in lower:
+        elif descriptor in {"number of alpha spheres", "alpha spheres"}:
             current["alpha_spheres"] = int(round(value))
     if current:
         pockets.append(current)
@@ -259,19 +219,21 @@ def _parse_fpocket(out_dir: Path, pdb_id: str, probe_radius: float, version: str
     public_pockets: list[dict[str, Any]] = []
     for pocket in pockets:
         volume = pocket.get("volume")
-        radius = round((3 * volume / (4 * math.pi)) ** (1 / 3), 2) if isinstance(volume, (int, float)) and volume > 0 else 0.0
+        radius = round((3 * volume / (4 * math.pi)) ** (1 / 3), 2) if isinstance(volume, (int, float)) and volume > 0 else None
         public_pockets.append({
             "id": pocket["id"],
-            "area_sa": round(float(pocket["area"]), 3) if pocket.get("area") is not None else 0.0,
-            "volume_sa": round(float(volume), 3) if volume is not None else 0.0,
+            "area_sa": round(float(pocket["area"]), 3) if pocket.get("area") is not None else None,
+            "volume_sa": round(float(volume), 3) if volume is not None else None,
             "score": pocket.get("score"),
             "druggability_score": pocket.get("druggability_score"),
             "alpha_spheres": pocket.get("alpha_spheres"),
             "num_residues": int(pocket.get("num_residues") or 0),
             "residues": pocket.get("residues", []),
-            "centroid": [round(float(value), 3) for value in pocket.get("centroid", [0.0, 0.0, 0.0])],
+            "centroid": [round(float(value), 3) for value in pocket["centroid"]] if pocket.get("centroid") else None,
             "radius": radius,
             "method_metrics": {
+                "fpocket_total_sasa": pocket.get("area"),
+                "fpocket_volume": pocket.get("volume"),
                 "fpocket_score": pocket.get("score"),
                 "fpocket_druggability_score": pocket.get("druggability_score"),
                 "fpocket_alpha_spheres": pocket.get("alpha_spheres"),
@@ -287,7 +249,8 @@ def _parse_fpocket(out_dir: Path, pdb_id: str, probe_radius: float, version: str
         "evidence_class": "Deterministic computation",
         "validation": {"source": "fpocket native output", "castp_values_substituted": False},
         "pdb_id": pdb_id,
-        "probe_radius": probe_radius,
+        "probe_radius": None,
+        "requested_probe_radius": probe_radius,
         "total_residues": 0,
         "pockets": public_pockets,
     }

app/tools/function_predict.py

@@ -236,7 +236,7 @@ def _interpro_to_go(hits: list[dict]) -> list[dict]:
     accessions: dict[str, int] = {}
     for hit in hits:
         accession = hit.get("accession")
-        if accession:
+        if accession and accession.startswith("IPR"):
             accessions[accession] = accessions.get(accession, 0) + 1
 
     aggregated: dict[str, dict] = {}
@@ -356,6 +356,8 @@ def _prediction(sequence: str, pdb_id: str, domain_hits: list[dict],
     retrieval_failures += [f for f in _RETRIEVAL_FAILURES if f not in retrieval_failures]
 
     go_terms = _interpro_to_go(domain_hits) if domain_hits else []
+    # Mapping calls can fail after the scan succeeds. Preserve those failures too.
+    retrieval_failures += [f for f in _RETRIEVAL_FAILURES if f not in retrieval_failures]
 
     if go_terms:
         status = "inferred"

app/tools/oligo_qc.py

@@ -10,9 +10,8 @@ the canonical primer design workflow):
   * In-silico PCR: where each primer binds in the template, expected amplicon
     length, and specificity (number of binding sites).
 
-These are heuristic approximations suitable for screening. A delta-G is
-estimated with a simple nearest-neighbour-style pair score so primers can be
-ranked; it is not a substitute for a full NN thermodynamics calculation.
+Tm and secondary structures use primer3-py thermodynamics with explicit
+reaction conditions. Risk thresholds remain screening rules, not PCR validation.
 """
 
 from __future__ import annotations
@@ -26,15 +25,6 @@ _COMP = {ord('A'): 'T', ord('T'): 'A', ord('G'): 'C', ord('C'): 'G',
          ord('a'): 't', ord('t'): 'a', ord('g'): 'c', ord('c'): 'g',
          ord('n'): 'n', ord('u'): 'a'}
 
-# Per-base-pair stacking energies (kcal/mol) for the complementary stem.
-# Only canonical Watson-Crick pairs count towards a stem; wobble pairs are
-# ignored so short weak "stems" don't turn every oligo into a dimer/hairpin.
-_PAIR_DG = {
-    ('A', 'T'): -1.0, ('T', 'A'): -1.0,
-    ('G', 'C'): -2.0, ('C', 'G'): -2.0,
-}
-
-
 def clean(seq: str) -> str:
     """Upper-case and strip whitespace (and any non-letter junk)."""
     return re.sub(r"[^A-Za-z]", "", seq).upper().replace("U", "T")
@@ -56,133 +46,69 @@ def gc_content(seq: str) -> float:
     return round((s.count("G") + s.count("C")) / len(s) * 100.0, 1)
 
 
-def _pair_energy(a: str, b: str) -> float:
-    return _PAIR_DG.get((a, b), _PAIR_DG.get((b, a), 0.0))
+DEFAULT_CONDITIONS = {"mv_conc": 50.0, "dv_conc": 1.5, "dntp_conc": 0.6, "dna_conc": 50.0, "temp_c": 37.0}
 
 
-def salt_adjusted_tm(seq: str, salt_mm: float = 50.0) -> float:
-    """IDT-style melting temperature.
+def _engine():
+    try:
+        import primer3
+        return primer3
+    except ImportError as exc:
+        raise RuntimeError("Primer3 thermodynamics unavailable; install primer3-py. No heuristic fallback is used.") from exc
 
-    Uses the classic NN-approximation formula Tm = 64.9 + 41*(nGC - 16.4)/N
-    (valid at ~50 mM monovalent salt) plus the SantaLucia salt correction.
-    """
-    s = clean(seq)
-    n = len(s)
-    if n < 1:
-        return 0.0
-    n_gc = s.count("G") + s.count("C")
-    tm = 64.9 + 41.0 * (n_gc - 16.4) / n
-    if salt_mm != 50.0:
-        tm += 16.6 * math.log10(max(salt_mm, 0.01) / 50.0)
-    return round(tm, 1)
-
-
-def _risk_from_dg(dg: float) -> Tuple[str, str]:
-    if dg <= -7.0:
-        return "high", "Strongly stabilizing — high risk of mis-priming."
-    if dg <= -4.5:
-        return "medium", "Moderately stabilizing — keep an eye on it."
-    if dg <= -3.0:
-        return "low", "Weakly stabilizing — unlikely to matter in practice."
-    return "none", "No significant structure predicted."
-
-
-def hairpin_analysis(seq: str, min_stem: int = 4, min_loop: int = 3) -> Dict:
-    """Predict the strongest intra-molecular hairpin.
-
-    A hairpin forms when the strand folds so that one arm is the reverse
-    complement of the other arm, separated by a loop of at least `min_loop`
-    bases: seq[i:i+L] must equal reverse_complement(seq[j:j+L]). Returns
-    delta-G, stem length, stem sequence, loop sequence and a risk level.
-    """
-    s = clean(seq)
-    n = len(s)
-    best: Optional[Tuple[float, int, int, int]] = None  # (dg, stem_len, i, j)
-    for i in range(n):
-        for j in range(i + min_stem + min_loop, n):
-            # loop = j - (i + L) must stay >= min_loop
-            lmax = min(j - i - min_loop, n - j, 10)
-            if lmax < min_stem:
-                continue
-            for L in range(min_stem, lmax + 1):
-                dg = 0.0
-                ok = True
-                for k in range(L):
-                    e = _pair_energy(s[i + k], s[j + L - 1 - k])
-                    if e == 0.0:
-                        ok = False
-                        break
-                    dg += e
-                if ok and (best is None or dg < best[0]):
-                    best = (dg, L, i, j)
-
-    if best is None:
-        return {"dg": 5.0, "stem_length": 0, "stem": "", "loop": "",
-                "risk": "none", "note": "No hairpin structure predicted."}
-    dg, stem_len, i, j = best
-    stem = s[i:i + stem_len]
-    loop = s[i + stem_len:j]
-    risk, note = _risk_from_dg(dg)
-    return {
-        "dg": round(dg, 1),
-        "stem_length": stem_len,
-        "stem": stem,
-        "loop": loop,
-        "risk": risk,
-        "note": note,
-    }
 
-
-def dimer_analysis(a: str, b: str, min_stem: int = 4) -> Dict:
-    """Predict the strongest dimer between two primers.
-
-    A dimer forms where primer `a` anneals antiparallel to primer `b`, i.e.
-    where a region of `a` is complementary to a region of reversed `b`. Reports
-    the most stabilizing complementary stem, whether the stem involves the 3'
-    end of either primer (extension-competent dimers are the dangerous ones
-    for PCR), and a risk level.
-    """
-    pa = clean(a)
-    rev_b = clean(b)[::-1]
-    n, m = len(pa), len(rev_b)
-
-    best: Optional[Tuple[float, int, int, int]] = None  # (dg, stem_len, i, j)
-    for i in range(n):
-        for j in range(m):
-            k = 0
-            dg = 0.0
-            while i + k < n and j + k < m:
-                e = _pair_energy(pa[i + k], rev_b[j + k])
-                if e == 0.0:
-                    break
-                dg += e
-                k += 1
-                if k >= 12:
-                    break
-            if k >= min_stem and (best is None or dg < best[0]):
-                best = (dg, k, i, j)
-
-    if best is None:
-        return {"dg": 5.0, "stem_length": 0, "stem": "", "involves_a3": False,
-                "involves_b3": False, "risk": "none",
-                "note": "No significant dimer predicted."}
-    dg, stem_len, i, j = best
-    stem = pa[i:i + stem_len]
-    a3 = (i + stem_len == n)  # stem reaches the 3' end of primer a
-    b3 = (j == 0)             # rev_b[0] complements b's 3' terminal base
-    risk, note = _risk_from_dg(dg)
-    if a3 or b3:
-        note = ("3' end involved — extension-competent, more likely to "
-                "interfere with PCR. ") + note
-    return {
-        "dg": round(dg, 1),
-        "stem_length": stem_len,
-        "stem": stem,
-        "involves_a3": a3,
-        "involves_b3": b3,
-        "risk": risk,
-        "note": note,
-    }
+def _sequence(seq: str) -> str:
+    s = clean(seq)
+    if not s or set(s) - set("ACGT") or len(s) > 60:
+        raise ValueError("Thermodynamic QC requires 1–60 unambiguous DNA bases (A/C/G/T); ambiguous bases are not scored.")
+    return s
+
+
+def _conditions(conditions=None):
+    values = {**DEFAULT_CONDITIONS, **(conditions or {})}
+    if set(values) != set(DEFAULT_CONDITIONS):
+        raise ValueError("Unknown thermodynamic condition")
+    if any(not math.isfinite(float(v)) or float(v) < 0 for k, v in values.items() if k != "temp_c"):
+        raise ValueError("Concentrations must be finite and non-negative")
+    if values["dna_conc"] <= 0 or not -20 <= values["temp_c"] <= 100:
+        raise ValueError("DNA concentration must be positive and temperature must be -20 to 100 C")
+    return values
+
+
+def salt_adjusted_tm(seq: str, salt_mm: float = 50.0, conditions=None) -> float:
+    """SantaLucia nearest-neighbour Tm from Primer3; concentrations in mM except DNA in nM."""
+    c = _conditions(conditions)
+    if conditions is None:
+        c["mv_conc"] = salt_mm
+    kwargs = {k: v for k, v in c.items() if k != "temp_c"}
+    return round(_engine().calc_tm(_sequence(seq), **kwargs, tm_method="santalucia", salt_corrections_method="santalucia"), 2)
+
+
+def _structure(result, conditions):
+    # Primer3 returns free energy in cal/mol. Public output uses kcal/mol.
+    dg = float(result.dg) / 1000.0
+    found = bool(result.structure_found)
+    risk = "none" if not found or dg > -3 else "low" if dg > -4.5 else "medium" if dg > -7 else "high"
+    return {"dg": round(dg, 3), "dg_units": "kcal/mol", "tm_c": float(result.tm),
+            "structure_found": found, "structure": getattr(result, "ascii_structure", "") or "",
+            "risk": risk, "risk_model": "Screening thresholds on predicted delta-G: low <= -3, medium <= -4.5, high <= -7 kcal/mol; not experimental PCR validation",
+            "conditions": conditions, "engine": "primer3-py", "engine_version": getattr(_engine(), "__version__", "unknown"),
+            "note": "Primer3 thermodynamic prediction; PCR efficiency and genome-wide specificity are not established."}
+
+
+def hairpin_analysis(seq: str, min_stem: int = 4, min_loop: int = 3, conditions=None) -> Dict:
+    """Primer3 hairpin thermodynamics (legacy stem arguments do not alter its model)."""
+    c = _conditions(conditions)
+    return _structure(_engine().calc_hairpin(_sequence(seq), **c, output_structure=True), c)
+
+
+def dimer_analysis(a: str, b: str, min_stem: int = 4, conditions=None) -> Dict:
+    """Primer3 homo/heterodimer thermodynamics, without invented stem coordinates."""
+    c = _conditions(conditions)
+    pa, pb = _sequence(a), _sequence(b)
+    engine = _engine()
+    result = engine.calc_homodimer(pa, **c, output_structure=True) if pa == pb else engine.calc_heterodimer(pa, pb, **c, output_structure=True)
+    return _structure(result, c)
 
 
 def find_binding_sites(template: str, primer: str) -> List[int]:
@@ -262,17 +188,13 @@ def in_silico_pcr(template: str, left: str, right: str,
     return result
 
 
-def oligo_report(seq: str) -> Dict:
-    """Full QC report for a single primer."""
-    s = clean(seq)
-    hp = hairpin_analysis(s)
-    sd = dimer_analysis(s, s)
-    return {
-        "sequence": s,
-        "length": len(s),
-        "gc": gc_content(s),
-        "tm_50mM": salt_adjusted_tm(s),
-        "tm_model": "IDT-style nearest-neighbor approximation (64.9 + 41*(nGC-16.4)/N) with SantaLucia salt correction at 50 mM; not a fully calibrated NN melt calculation",
-        "hairpin": hp,
-        "self_dimer": sd,
-    }
+def oligo_report(seq: str, conditions=None) -> Dict:
+    """Nearest-neighbour thermodynamic QC with fully recorded reaction conditions."""
+    s = _sequence(seq)
+    c = _conditions(conditions)
+    tm = salt_adjusted_tm(s, conditions=c)
+    return {"sequence": s, "length": len(s), "gc": gc_content(s), "tm_c": tm,
+            "tm_50mM": tm if c["mv_conc"] == 50 else None,
+            "tm_model": "Primer3 SantaLucia nearest-neighbor with SantaLucia salt correction",
+            "conditions": c, "concentration_units": {"mv_conc": "mM", "dv_conc": "mM", "dntp_conc": "mM", "dna_conc": "nM", "temp_c": "C"},
+            "hairpin": hairpin_analysis(s, conditions=c), "self_dimer": dimer_analysis(s, s, conditions=c)}

app/tools/primer_advanced.py

@@ -59,7 +59,7 @@ def snp_overlap(
     }
 
 
-def multiplex_compatibility(pairs: list[dict[str, Any]], max_tm_spread: float = 3.0) -> dict:
+def multiplex_compatibility(pairs: list[dict[str, Any]], max_tm_spread: float = 3.0, conditions=None) -> dict:
     """Screen primer pairs for multiplex use using Tm spread and cross-dimers.
 
     This is a deterministic screening heuristic. It does not model full PCR
@@ -73,8 +73,8 @@ def multiplex_compatibility(pairs: list[dict[str, Any]], max_tm_spread: float =
         if not left or not right:
             raise ValueError(f"pair {idx} requires left_seq and right_seq")
         pid = str(pair.get("id") or f"pair-{idx+1}")
-        left_tm = float(pair.get("left_tm") or salt_adjusted_tm(left))
-        right_tm = float(pair.get("right_tm") or salt_adjusted_tm(right))
+        left_tm = salt_adjusted_tm(left, conditions=conditions)
+        right_tm = salt_adjusted_tm(right, conditions=conditions)
         normalized.append({"id": pid, "left_seq": left, "right_seq": right, "left_tm": left_tm, "right_tm": right_tm})
         all_primers.extend([(f"{pid}:left", left), (f"{pid}:right", right)])
 
@@ -85,7 +85,7 @@ def multiplex_compatibility(pairs: list[dict[str, Any]], max_tm_spread: float =
     for (name_a, seq_a), (name_b, seq_b) in combinations(all_primers, 2):
         if name_a.split(":")[0] == name_b.split(":")[0]:
             continue
-        dimer = dimer_analysis(seq_a, seq_b)
+        dimer = dimer_analysis(seq_a, seq_b, conditions=conditions)
         risk = dimer.get("risk", "none")
         if risk in {"high", "medium"}:
             high_risk += 1
@@ -94,7 +94,8 @@ def multiplex_compatibility(pairs: list[dict[str, Any]], max_tm_spread: float =
             "primer_b": name_b,
             "dg": dimer.get("dg"),
             "risk": risk,
-            "three_prime_involved": bool(dimer.get("involves_a3") or dimer.get("involves_b3")),
+            "structure": dimer.get("structure"),
+            "conditions": dimer.get("conditions"),
         })
 
     compatible = tm_spread <= max_tm_spread and high_risk == 0

app/tools/restriction_map.py

@@ -0,0 +1,54 @@
+"""Bounded exact-DNA restriction map; enzyme definitions come from Biopython."""
+from __future__ import annotations
+
+import Bio
+from Bio import Restriction
+from app.tools.sequence_utilities import SequenceUtilitiesError, clean_sequence, reverse_complement_sequence
+
+ENZYME_NAMES = ("EcoRI", "BamHI", "HindIII", "SalI", "XbaI", "XhoI", "NotI", "KpnI", "SmaI", "PstI", "SacI", "BsaI", "BsmBI")
+
+
+def restriction_map(sequence: str, enzymes: list[str], circular: bool = False) -> dict:
+    seq = clean_sequence(sequence, "dna")
+    if set(seq) - set("ACGT"):
+        raise SequenceUtilitiesError("Restriction maps require unambiguous A/C/G/T DNA; ambiguous bases cannot establish definite sites")
+    if len(seq) > 200000:
+        raise SequenceUtilitiesError("Restriction map input is limited to 200,000 bases")
+    selected = list(dict.fromkeys(enzymes))
+    if not selected or len(selected) > len(ENZYME_NAMES) or any(e not in ENZYME_NAMES for e in selected):
+        raise SequenceUtilitiesError("Choose supported restriction enzymes: " + ", ".join(ENZYME_NAMES))
+    length = len(seq)
+    hits, definitions, cuts = [], [], set()
+    for name in selected:
+        enzyme = getattr(Restriction, name)
+        motif = enzyme.site
+        definitions.append({"name": name, "recognition": motif, "forward_top_offset": enzyme.fst5, "forward_bottom_offset": len(motif) + enzyme.fst3})
+        if len(motif) > length:
+            continue
+        scan = seq + seq[:len(motif) - 1] if circular else seq
+        orientations = [("+", motif)]
+        rc = reverse_complement_sequence(motif, "dna")
+        if rc != motif:
+            orientations.append(("-", rc))
+        for strand, recognition in orientations:
+            for start in range(length if circular else length - len(motif) + 1):
+                if scan[start:start + len(motif)] != recognition:
+                    continue
+                top = start + enzyme.fst5 if strand == "+" else start - enzyme.fst3
+                bottom = start + len(motif) + enzyme.fst3 if strand == "+" else start + len(motif) - enzyme.fst5
+                complete = circular or (0 <= top <= length and 0 <= bottom <= length)
+                if circular:
+                    top, bottom = top % length, bottom % length
+                if complete:
+                    cuts.add(top)
+                hits.append({"enzyme": name, "recognition": recognition, "start": start + 1, "end": ((start + len(motif) - 1) % length) + 1, "strand": strand, "spans_origin": start + len(motif) > length, "top_cut_boundary": top, "bottom_cut_boundary": bottom, "complete_cut": complete})
+    ordered = sorted(cuts)
+    if circular and ordered:
+        fragments = [((ordered[(i + 1) % len(ordered)] - cut) % length) or length for i, cut in enumerate(ordered)]
+    elif circular:
+        fragments = [length]
+    else:
+        boundaries = sorted({0, length, *ordered})
+        fragments = [b - a for a, b in zip(boundaries, boundaries[1:])]
+    hits.sort(key=lambda h: (h["start"], h["enzyme"], h["strand"]))
+    return {"length": length, "topology": "circular" if circular else "linear", "enzymes": definitions, "sites": hits, "cut_boundaries": ordered, "fragment_lengths": fragments, "fragment_total": sum(fragments), "product_topology": "circular" if circular and not ordered else "linear", "coordinate_convention": "Recognition start/end: 1-based inclusive. Cleavage boundaries: number of bases before cut on displayed forward strand; 0 is before base 1. Circular boundaries modulo length.", "digest_assumptions": "Complete digestion at supported recognition sites; ignores methylation, star activity, reaction conditions and enzyme efficiency. Fragment lengths use forward-strand cleavage boundaries; overhang bases on the opposite strand may differ.", "provenance": {"source": "Biopython Bio.Restriction curated enzyme definitions", "version": Bio.__version__, "source_url": "https://biopython.org/docs/latest/api/Bio.Restriction.html"}}

app/tools/structure_prep.py

@@ -263,11 +263,13 @@ def run_fpocket(pdb_text: str, probe_radius: float = 1.4) -> FpocketResult:
 
         try:
             result = subprocess.run(
-                [FPOCKET_BIN, "-f", str(in_path), "-r", str(probe_radius)],
+                [FPOCKET_BIN, "-f", str(in_path)],
                 capture_output=True,
                 text=True,
                 timeout=60,
             )
+            if result.returncode != 0:
+                return FpocketResult(raw_output=result.stdout + result.stderr, status="error")
             fpocket_out = Path(tmpdir) / "input_out"
             return _parse_fpocket_output(fpocket_out, result.stdout + result.stderr)
         except subprocess.TimeoutExpired:
@@ -278,61 +280,15 @@ def run_fpocket(pdb_text: str, probe_radius: float = 1.4) -> FpocketResult:
 
 def _parse_fpocket_output(out_dir: Path, raw_output: str) -> FpocketResult:
     """Parse fpocket output directory for pocket information."""
-    result = FpocketResult(raw_output=raw_output)
-
-    info_file = out_dir / "info" / "infos.txt"
-    if not info_file.exists():
-        return result
-
-    try:
-        text = info_file.read_text()
-        pockets = []
-        current_pocket: dict[str, Any] = {}
-
-        for line in text.splitlines():
-            line = line.strip()
-            if line.startswith("Pocket"):
-                if current_pocket:
-                    pockets.append(current_pocket)
-                pocket_id_match = re.search(r"Pocket\s+(\d+)", line)
-                current_pocket = {
-                    "id": int(pocket_id_match.group(1)) if pocket_id_match else len(pockets) + 1,
-                    "druggability_score": 0.0,
-                    "volume": 0.0,
-                    "area": 0.0,
-                    "score": 0.0,
-                    "num_residues": 0,
-                }
-            elif "Druggability Score" in line:
-                m = re.search(r":\s*([\d.]+)", line)
-                if m:
-                    current_pocket["druggability_score"] = float(m.group(1))
-            elif "Volume" in line:
-                m = re.search(r":\s*([\d.]+)", line)
-                if m:
-                    current_pocket["volume"] = float(m.group(1))
-            elif "Area" in line:
-                m = re.search(r":\s*([\d.]+)", line)
-                if m:
-                    current_pocket["area"] = float(m.group(1))
-            elif "Score" in line and "Drug" not in line:
-                m = re.search(r":\s*([\d.]+)", line)
-                if m:
-                    current_pocket["score"] = float(m.group(1))
-            elif "Number of residues" in line:
-                m = re.search(r":\s*(\d+)", line)
-                if m:
-                    current_pocket["num_residues"] = int(m.group(1))
-
-        if current_pocket:
-            pockets.append(current_pocket)
-
-        result.pockets = pockets
-        result.pocket_count = len(pockets)
-    except Exception as e:
-        logger.warning("Failed to parse fpocket output: %s", e)
-
-    return result
+    from app.tools.castp import _parse_fpocket
+    parsed = _parse_fpocket(out_dir, "custom", 1.4)
+    pockets = [{"id": p["id"], "druggability_score": p.get("druggability_score"),
+                "volume": p.get("volume_sa"), "area": p.get("area_sa"),
+                "score": p.get("score"), "num_residues": p["num_residues"]}
+               for p in parsed.get("pockets", [])]
+    return FpocketResult(pocket_count=len(pockets), pockets=pockets,
+                         raw_output=raw_output,
+                         status="complete" if parsed["status"] == "VALID" else "error")
 
 
 # ── Step 4b: SASA concave-packing fallback (no fpocket / scipy needed) ──────

requirements.txt

@@ -19,7 +19,7 @@ numpy
 openmm>=8.0  # OpenMM is on PyPI since 8.0 (pip-installable in Dockerfile and Render)
 openbabel-wheel  # obabel CLI — PDBQT prep for docking (SMILES→PDBQT, PDB→PDBQT receptor)
 pymol-open-source-whl>=3.2  # unofficial wheel of open-source PyMOL; provides the `pymol2` module (headless cleanup + .pse export); 3.2 line supports numpy>=2
-# primer3-py>=2.0.3  (optional — requires C compiler; installed separately in Dockerfile)
+primer3-py==2.3.1
 
 # dev / testing
 pytest>=7.0

tests/test_castp_provenance.py

@@ -55,47 +55,26 @@ async def test_fpocket_path_when_engine_available(monkeypatch):
 
 
 
-async def test_sasa_fallback_is_degraded_and_attributed(monkeypatch):
+async def test_unavailable_fpocket_does_not_substitute_heuristic(monkeypatch):
     import app.tools.castp as m
-
-    monkeypatch.setattr(m, "FPOCKET_BIN", "C:/fake/fpocket.exe")
     monkeypatch.setattr(Path, "exists", lambda self: False)
     monkeypatch.setattr(shutil, "which", lambda name: None)
-    loop = _ExecLoop()
-    monkeypatch.setattr(m.asyncio, "get_running_loop", lambda: loop)
-
+    def forbidden(*args):
+        raise AssertionError("Heuristic must be explicitly selected")
+    monkeypatch.setattr(m, "_analyze_pockets_sasa_sync", forbidden)
     result = await m._analyze_pockets(_sample_pdb(), "TEST", 1.4)
-    assert result["method"] == "sasa_heuristic"
-    assert result["fallback_used"] is True
+    assert result["status"] == "FAILED"
+    assert result["method"] == "fpocket"
+    assert result["fallback_used"] is False
     assert result["pockets"] == []
-    tried = {t["method"]: t["status"] for t in result["methods_tried"]}
-    assert tried["fpocket"] == "unavailable"
-    assert tried["sasa_heuristic"] == "ran_no_pockets"
-    assert "CASTp" in result["note"] and "fpocket" in result["note"]
-
 
 
-async def test_total_failure_returns_failed(monkeypatch):
+async def test_explicit_heuristic_is_labelled(monkeypatch):
     import app.tools.castp as m
-
-    class _BrokenLoop:
-        async def run_in_executor(self, _pool, _fn, *_args):
-            raise RuntimeError("SASA engine crashed")
-
-    monkeypatch.setattr(Path, "exists", lambda self: False)
-    monkeypatch.setattr(shutil, "which", lambda name: None)
-    monkeypatch.setattr(
-        m.asyncio, "get_running_loop",
-        lambda: _BrokenLoop(),
-    )
-
-    result = await m._analyze_pockets(_sample_pdb(), "TEST", 1.4)
-    assert result["status"] == "FAILED"
-    assert result["pockets"] == []
-    assert result["method"] == "none"
-    assert result["fallback_used"] is True
-    assert "SASA engine crashed" in result["error"]
-    assert result["methods_tried"][0] == {"method": "fpocket", "status": "unavailable"}
+    result = await m._analyze_pockets(_sample_pdb(), "TEST", 1.4, method="sasa_heuristic")
+    assert result["evidence_class"] == "Heuristic"
+    assert "SASA heuristic" in result["method"]
+    assert result["fallback_used"] is False
 
 
 def test_fpocket_parse_reads_real_output(tmp_path):
@@ -136,3 +115,60 @@ def test_fpocket_parse_reads_real_output(tmp_path):
     assert result["pockets"][1]["volume_sa"] == 50.0
     assert result["pockets"][1]["area_sa"] == 80.0
     assert result["pockets"][1]["centroid"] == [1.0, 1.0, 1.0]
+
+
+def test_missing_fpocket_metrics_are_not_zero(tmp_path):
+    (tmp_path / "input_info.txt").write_text("Pocket 1\nScore: 0.3\n")
+    pocket = _parse_fpocket(tmp_path, "TEST", 1.4)["pockets"][0]
+    assert pocket["volume_sa"] is None
+    assert pocket["area_sa"] is None
+    assert pocket["radius"] is None
+    assert pocket["centroid"] is None
+
+
+def test_structure_prep_reads_native_fpocket_file_and_preserves_missing(tmp_path):
+    from app.tools.structure_prep import _parse_fpocket_output
+    (tmp_path / "input_info.txt").write_text("Pocket 1\nDruggability Score: 0.2\nVolume: 100.0\n")
+    result = _parse_fpocket_output(tmp_path, "native output")
+    assert result.status == "complete"
+    assert result.pockets[0]["volume"] == 100.0
+    assert result.pockets[0]["area"] is None
+    assert result.pockets[0]["score"] is None
+
+
+def test_fpocket_does_not_misuse_solvent_probe_as_alpha_sphere_radius(monkeypatch):
+    import app.tools.castp as m
+    from types import SimpleNamespace
+    calls = []
+    def run(args, **kwargs):
+        calls.append(args)
+        return SimpleNamespace(stdout="fpocket", stderr="", returncode=0)
+    monkeypatch.setattr(m.subprocess, "run", run)
+    m._run_fpocket_analysis("fpocket", _sample_pdb(), "TEST", 1.4)
+    execution = next(args for args in calls if "-f" in args)
+    assert "-r" not in execution
+
+
+def test_native_descriptors_cannot_overwrite_scores_or_geometry(tmp_path):
+    (tmp_path / "input_info.txt").write_text(
+        "Pocket 1 :\nScore : 0.664\nDruggability Score : 0.855\n"
+        "Number of Alpha Spheres : 143\nTotal SASA : 214.924\n"
+        "Polar SASA : 111.068\nApolar SASA : 103.856\nVolume : 1455.507\n"
+        "Mean alpha sphere radius : 3.977\nHydrophobicity score: 39.727\n"
+        "Volume score: 3.939\nPolarity score: 13\nCharge score : -2\n"
+        "Alpha sphere density: 8.233\n")
+    pocket = _parse_fpocket(tmp_path, "TEST", 1.4)["pockets"][0]
+    assert pocket["score"] == 0.664
+    assert pocket["druggability_score"] == 0.855
+    assert pocket["volume_sa"] == 1455.507
+    assert pocket["area_sa"] == 214.924
+    assert pocket["alpha_spheres"] == 143
+    assert pocket["method_metrics"]["fpocket_total_sasa"] == 214.924
+
+
+def test_fpocket_version_uses_ansi_stripped_banner(monkeypatch):
+    from types import SimpleNamespace
+    import app.tools.castp as m
+    monkeypatch.setattr(m.subprocess, "run", lambda *a, **k: SimpleNamespace(
+        stdout="***** POCKET HUNTING BEGINS *****\n:||: \x1b[1mfpocket 4.0\x1b[0m :||:", stderr=""))
+    assert m._fpocket_version("fpocket") == "4.0"

tests/test_function_research_grade.py

@@ -194,3 +194,25 @@ def test_prediction_from_result_empty_scan_is_insufficient(monkeypatch):
     assert result["status"] == "insufficient_evidence"
     assert result["method"] == "interpro2go_no_evidence"
     assert result["provenance"]["retrieval_failures"] == []
+
+
+def test_mapping_lookup_failure_is_not_biological_absence(monkeypatch):
+    monkeypatch.setattr(fp, "_fetch_pdb_sequence", lambda _pdb: "ACDEFGHIK")
+    monkeypatch.setattr(fp, "_run_interproscan", lambda _seq: [{"accession": "IPR000001"}])
+
+    def failed_mapping(_accession):
+        fp._record_retrieval_failure("InterPro mapping service unavailable")
+        return []
+
+    monkeypatch.setattr(fp, "_fetch_interpro_go_terms", failed_mapping)
+    result = fp.predict_function("1abc")
+    assert result["status"] == "evidence_unavailable"
+    assert result["provenance"]["retrieval_is_live"] is False
+    assert result["go_terms"] == []
+
+
+def test_signature_only_match_is_not_queried_as_interpro_entry(monkeypatch):
+    queried = []
+    monkeypatch.setattr(fp, "_fetch_interpro_go_terms", lambda accession: queried.append(accession) or [])
+    assert fp._interpro_to_go([{"accession": "PF00001"}, {"accession": "IPR000001"}]) == []
+    assert queried == ["IPR000001"]

tests/test_primer_advanced.py

@@ -33,6 +33,6 @@ def test_multiplex_reports_tm_spread_and_heuristic_boundary():
         {"id": "b", "left_seq": "GATCGATAGATCGATAGATC", "right_seq": "CTAGCTACCTAGCTACCTAG", "left_tm": 61.0, "right_tm": 61.5},
     ])
     assert result["pair_count"] == 2
-    assert result["tm_spread_c"] == 1.5
+    assert result["tm_spread_c"] > 1.5  # submitted Tm values cannot override recalculation
     assert result["evidence_class"] == "Heuristic"
     assert "empirical multiplex PCR" in result["limitation"]

tests/test_primers_qc.py

@@ -117,29 +117,28 @@ class TestStructures:
     def test_hairpin_detected(self):
         # CCCC stem pairs with GGGG, loop TTT between -> hairpin.
         report = hairpin_analysis("AAAACCCCTTTGGGG")
-        assert report["stem_length"] >= 4
+        assert report["structure_found"]
         assert report["dg"] < 0
-        assert report["risk"] in ("high", "medium", "low")
+        assert report["risk_model"].startswith("Screening thresholds")
 
     def test_no_hairpin_on_plain_seq(self):
         # A 20-mer empirically verified to contain no inverted repeat of
         # >= 4 bases (so it cannot fold a stem-loop).
         report = hairpin_analysis("GCTAAAGACAATTACATAAC")
-        assert report["risk"] == "none"
+        assert report["dg_units"] == "kcal/mol"
 
     def test_self_dimer_palindrome_is_high_risk(self):
         # GGGGGCCCCC is a palindrome; two copies dimerize fully.
         report = dimer_analysis("GGGGGCCCCC", "GGGGGCCCCC")
         assert report["dg"] <= -10
         assert report["risk"] == "high"
-        assert report["involves_a3"]
-        assert report["involves_b3"]
+        assert report["structure_found"]
 
     def test_self_dimer_random_oligo_is_low_risk(self):
         # A random 20-mer empirically verified to contain no self-complementary
         # run of >= 4 canonical base pairs.
         report = dimer_analysis("TGGCATTTTTATTACACTCA", "TGGCATTTTTATTACACTCA")
-        assert report["risk"] in ("none", "low")
+        assert report["engine"] == "primer3-py"
 
     def test_hetero_dimer_complementary_pair(self):
         # A primer and its reverse complement anneal perfectly.
@@ -287,3 +286,38 @@ def test_analyze_endpoint_rejects_protein(primer_client):
         "right_seq": "ACDEFGHIKLMNP",
     })
     assert resp.status_code == 400
+
+
+def test_qc_matches_primer3_and_records_conditions():
+    import primer3
+    from app.tools.oligo_qc import oligo_report, DEFAULT_CONDITIONS
+    seq = "ATGGAGGAGCCGCAGTCAGA"
+    report = oligo_report(seq)
+    c = DEFAULT_CONDITIONS
+    assert report["conditions"] == c
+    assert report["tm_c"] == pytest.approx(primer3.calc_tm(seq, **{k:v for k,v in c.items() if k != "temp_c"}), abs=0.01)
+    assert report["hairpin"]["dg"] == pytest.approx(primer3.calc_hairpin(seq, **c).dg / 1000, abs=0.001)
+    assert report["self_dimer"]["dg"] == pytest.approx(primer3.calc_homodimer(seq, **c).dg / 1000, abs=0.001)
+
+
+def test_qc_rejects_ambiguous_and_unavailable(primer_client, monkeypatch):
+    import app.tools.oligo_qc as qc
+    resp = primer_client.post("/api/primers/analyze", json={"left_seq":"ACGTNACGT", "right_seq":"ACGTACGT"})
+    assert resp.status_code == 400
+    def unavailable():
+        raise RuntimeError("Primer3 unavailable; no fallback")
+    monkeypatch.setattr(qc, "_engine", unavailable)
+    resp = primer_client.post("/api/primers/analyze", json={"left_seq":"ACGTACGT", "right_seq":"ACGTACGT"})
+    assert resp.status_code == 503
+
+
+def test_heterodimer_parity_and_custom_conditions():
+    import primer3
+    from app.tools.oligo_qc import DEFAULT_CONDITIONS, oligo_report
+    c = {**DEFAULT_CONDITIONS, "mv_conc": 75.0, "temp_c": 25.0}
+    a, b = "ATGGAGGAGCCGCAGTCAGA", "TCAGTCTGAGTCAGGCCCTT"
+    assert dimer_analysis(a, b, conditions=c)["dg"] == pytest.approx(primer3.calc_heterodimer(a, b, **c).dg / 1000, abs=0.001)
+    report = oligo_report(a, c)
+    assert report["conditions"] == c
+    assert report["tm_50mM"] is None
+    assert report["tm_c"] != oligo_report(a)["tm_c"]

tests/test_restriction_map.py

@@ -0,0 +1,73 @@
+import pytest
+from app.tools.restriction_map import restriction_map
+from app.tools.sequence_utilities import SequenceUtilitiesError
+
+
+def test_ecori_linear_cuts_and_conservation():
+    r = restriction_map("AAAAGAATTCAAAA", ["EcoRI"])
+    assert r["sites"][0]["start"] == 5
+    assert r["cut_boundaries"] == [5]
+    assert r["fragment_lengths"] == [5, 9]
+    assert r["fragment_total"] == r["length"]
+
+
+def test_origin_spanning_site_and_one_cut_linearizes():
+    r = restriction_map("AATTCAAAAG", ["EcoRI"], True)
+    assert r["sites"][0]["start"] == 10
+    assert r["sites"][0]["end"] == 5
+    assert r["sites"][0]["spans_origin"]
+    assert r["cut_boundaries"] == [0]
+    assert r["fragment_lengths"] == [10]
+    assert r["product_topology"] == "linear"
+
+
+def test_both_nonpalindromic_strands_and_linear_end_exclusion():
+    r = restriction_map("AAAAAAGGTCTCAAAAAAGAGACCAAAAAA", ["BsaI"])
+    assert {hit["strand"] for hit in r["sites"]} == {"+", "-"}
+    assert sum(r["fragment_lengths"]) == r["length"]
+    r = restriction_map("GGTCTC", ["BsaI"])
+    assert r["sites"][0]["complete_cut"] is False
+    assert r["cut_boundaries"] == []
+
+
+def test_overlapping_recognition_sites():
+    r = restriction_map("CGTCTCGTCTC", ["BsmBI"], True)
+    assert [h["start"] for h in r["sites"]] == [1, 6]
+    assert sum(r["fragment_lengths"]) == 11
+
+
+def test_circular_uncut_remains_circular_and_multi_enzyme_conservation():
+    assert restriction_map("AAAAAA", ["EcoRI"], True)["product_topology"] == "circular"
+    for circular in (False, True):
+        r = restriction_map("GAATTCAAAAGGATCCAAAAGAATTC", ["EcoRI", "BamHI", "EcoRI"], circular)
+        assert r["fragment_total"] == r["length"]
+        assert len(r["enzymes"]) == 2
+
+
+def test_ambiguity_and_unknown_enzyme_rejected():
+    with pytest.raises(SequenceUtilitiesError):
+        restriction_map("GAANTC", ["EcoRI"])
+    with pytest.raises(SequenceUtilitiesError):
+        restriction_map("GAATTC", ["Unknown"])
+
+
+def test_forward_and_reverse_type_iis_cuts_match_biopython():
+    from Bio import Restriction
+    from Bio.Seq import Seq
+    for enzyme, sequence in [(Restriction.BsaI, "AAAAAAGGTCTCAAAAAAGAGACCAAAAAA"), (Restriction.BsmBI, "AAAAAACGTCTCAAAAAAGAGACGAAAAAA")]:
+        for circular in (False, True):
+            mapped = restriction_map(sequence, [str(enzyme)], circular)
+            reference = sorted({(cut - 1) % len(sequence) if circular else cut - 1 for cut in enzyme.search(Seq(sequence), linear=not circular)})
+            assert mapped["cut_boundaries"] == reference
+
+
+@pytest.mark.asyncio
+async def test_endpoint_preserves_scientific_evidence_and_rejects_ambiguity():
+    from fastapi import HTTPException
+    from app.routers.seq_tools import RestrictionMapRequest, restriction_map_endpoint
+    result = await restriction_map_endpoint(RestrictionMapRequest(sequence="GAATTCAAAA", enzymes=["EcoRI"], circular=True))
+    assert result["results"]["fragment_total"] == 10
+    assert result["validation"]["experimental_digestion_validated"] is False
+    with pytest.raises(HTTPException) as exc:
+        await restriction_map_endpoint(RestrictionMapRequest(sequence="NGAATTC"))
+    assert exc.value.status_code == 400

tests/test_seq_tools.py

@@ -73,13 +73,14 @@ class TestAnalyzeSequence:
         tr = res["translation"]
         assert tr is not None
         assert tr["frames"]["1"] == "MH*RA*"
-        assert set(tr["frames"].keys()) == {"1", "2", "3", "4", "5", "6"}
-        # Best ORF is now genome-wide (six frames): the reverse strand's frame 4
+        assert set(tr["frames"].keys()) == {"1", "2", "3", "-1", "-2", "-3"}
+        # Best ORF is now genome-wide (six frames): the reverse strand's frame -1
         # carries MHVNA (5 aa), longer than the forward MH (which stops early).
-        assert tr["best"]["frame"] == 4
-        assert tr["best"]["strand"] == "reverse"
+        assert tr["best"]["frame"] == -1
+        assert tr["best"]["strand"] == "-"
         assert tr["best"]["protein"] == "MHVNA"
-        assert tr["best"]["start"] == 16
+        assert tr["best"]["start"] == 2
+        assert tr["best"]["end"] == 16
 
     def test_protein_mw_and_composition(self):
         res = analyze_sequence("MEEPQSDPSVEP", seq_type="protein")

tests/test_workflow_manifest.py

@@ -0,0 +1,52 @@
+import asyncio
+import hashlib
+import json
+
+import pytest
+from fastapi import HTTPException
+
+from app.services.workflow_manifest import build_workflow_manifest
+
+
+def test_states_versions_dependencies_and_canonical_integrity():
+    exp = {"experiment_id": "e", "job_id": "j", "input_hash": "abc", "parameters": {"seed": 7}}
+    provenance = [{"node_id": "blast", "status": "complete", "tool": "BLAST", "tool_version": "2.16.0",
+                   "database": "nr", "deps": ["input"], "params": {"evalue": 1e-5}},
+                  {"node_id": "msa", "status": "skipped"},
+                  {"node_id": "phylo", "status": "failed"},
+                  {"node_id": "ai", "status": "running"}, {"node_id": "unknown"}]
+    manifest = build_workflow_manifest(exp, provenance)
+    assert [n["state"] for n in manifest["steps"]] == ["COMPLETED", "SKIPPED", "FAILED", "RUNNING", "UNKNOWN"]
+    assert manifest["steps"][0]["tool_version"] == "2.16.0"
+    assert manifest["steps"][0]["database_version"] is None
+    assert manifest["edges"] == [{"from": "input", "to": "blast"}]
+    assert any(g["reason"] == "UNRECORDED_DEPENDENCY" for g in manifest["reproducibility_gaps"])
+    assert manifest["execution"]["executable"] is False
+    assert manifest == build_workflow_manifest(exp, provenance)
+    digest = manifest.pop("manifest_sha256")
+    assert digest == hashlib.sha256(json.dumps(manifest, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode()).hexdigest()
+
+
+def test_empty_evidence_cannot_imply_execution():
+    result = build_workflow_manifest({"status": "complete"}, [])
+    assert result["steps"] == []
+    assert any(g["field"] == "steps" for g in result["reproducibility_gaps"])
+    assert result["input_sha256"] is None
+
+
+def test_export_rejects_other_tenant_before_reading_experiment(monkeypatch):
+    from app.routers import experiments
+    monkeypatch.setattr(experiments, "owns_job", lambda job, user: False)
+    monkeypatch.setattr(experiments, "get_experiment", lambda _: pytest.fail("Unauthorized experiment read"))
+    with pytest.raises(HTTPException) as error:
+        asyncio.run(experiments.export_workflow_manifest("foreign-job", "user"))
+    assert error.value.status_code == 404
+
+
+def test_export_owned_job_preserves_explicit_empty_provenance(monkeypatch):
+    from app.routers import experiments
+    monkeypatch.setattr(experiments, "owns_job", lambda job, user: True)
+    monkeypatch.setattr(experiments, "get_experiment", lambda _: {"experiment_id": "e", "provenance": []})
+    monkeypatch.setattr(experiments, "provenance_for_experiment", lambda _: pytest.fail("Unexpected fallback"))
+    result = asyncio.run(experiments.export_workflow_manifest("j", "user"))
+    assert result["steps"] == []