Deploy 40da983

BLAST, reachable by the agent — and gated, because the sequence leaves

Source: https://github.com/WINTER4000/turingDNA/commit/40da98346f97d15dda7dd4840c9c956fa4e34a9a

dee/core/agent_tools.py

@@ -838,6 +838,75 @@ _REGION_RE = re.compile(r"^\s*(\d{1,9})\s*(?:\.\.|-|\u2013|:)\s*(\d{1,9})\s*$")
 _PLASMID_MAX_FEATURES = 40
 
 
+def _tool_blast_sequence(args: Dict[str, Any]) -> Dict[str, Any]:
+    """Search NCBI for sequences like this one — blastp or blastn.
+
+    Answers the two questions the engine could not: "what IS this?" and "what
+    else looks like it?". The machinery has existed since the identify
+    feature; it lived behind the Flask layer where the agent could not reach
+    it (importing server.py from here is a circular import).
+
+    THIS IS THE ONE TOOL THAT SENDS THE USER'S RESIDUES OFF-SPACE. Everything
+    else computes locally or sends at most (gene_symbol, organism). So it
+    carries requires_confirm — the same gate as log_outcome and edit_sequence
+    — which means the model cannot decide on the user's behalf to publish
+    their unpublished construct to NCBI. That is not a UI nicety: an audit
+    already forced the REST path to enforce consent server-side rather than
+    trusting a modal, and the agent path must be at least as strict.
+    """
+    from dee.core import blast as _blast
+
+    seq = re.sub(r"[^A-Za-z]", "", str(args.get("sequence") or "")).upper()
+    if len(seq) < 12:
+        return {"ok": False, "error": (
+            "Give a sequence of at least 12 residues/bases to search with.")}
+
+    kind = str(args.get("kind") or "").lower().strip() or None
+    if kind not in (None, "protein", "nucleotide"):
+        return {"ok": False, "error": "kind must be 'protein' or 'nucleotide'."}
+
+    job = _blast.submit(seq, kind=kind)
+    _blast.wait(job)
+
+    if job.status == "error":
+        return {"ok": False, "kind": "blast_failed", "error": job.error,
+                "next": "Report that NCBI did not answer. Do not describe "
+                        "hits from memory."}
+
+    if not job.done():
+        # Honest degradation. A tool that says "still running" is better than
+        # one that holds the conversation open for four minutes.
+        return {"ok": False, "kind": "still_running", "job_id": job.job_id,
+                "elapsed_seconds": job.elapsed(),
+                "error": (f"NCBI BLAST is still running after "
+                          f"{job.elapsed():.0f}s — its queue is shared and can "
+                          f"take several minutes."),
+                "next": ("Tell the user it is still running and carry on with "
+                         "something else, or ask them to try again shortly. "
+                         "Never invent hits while waiting.")}
+
+    hits = job.hits
+    if not hits:
+        return {"ok": True, "hit_count": 0, "hits": [], "search": job.kind,
+                "summary": "no significant hits",
+                "next": ("No hit above threshold is a real finding — it may be "
+                         "a novel or synthetic sequence. Say that rather than "
+                         "guessing what it resembles.")}
+
+    return {
+        "ok": True,
+        "search": job.kind,
+        "hit_count": len(hits),
+        "hits": hits,
+        "top": hits[0],
+        "elapsed_seconds": job.elapsed(),
+        "summary": (f"{hits[0]['title'][:60]} · {hits[0]['identity_pct']}% id"),
+        "note": ("Identity and coverage are BLAST's own numbers, reported "
+                 "verbatim. A high-identity hit names what this sequence most "
+                 "resembles — it is not proof they are the same molecule."),
+    }
+
+
 def _tool_simulate_assembly(args: Dict[str, Any]) -> Dict[str, Any]:
     """Actually assemble the fragments — Gibson, Golden Gate or restriction.
 
@@ -1634,6 +1703,13 @@ _TOOLS: Dict[str, Dict[str, Any]] = {
     # and touches no saved work, so no confirm gate — the construct only
     # becomes real when the user saves it.
     "simulate_assembly": {"fn": _tool_simulate_assembly, "requires_signin": False},
+    # requires_confirm because this is the ONLY tool that sends the user's
+    # actual residues outside the Space. An audit already forced the REST
+    # path to enforce that server-side rather than trusting a modal; the
+    # agent path uses the same gate so the model cannot decide to publish
+    # someone's unpublished construct to NCBI on their behalf.
+    "blast_sequence":  {"fn": _tool_blast_sequence,  "requires_signin": False,
+                        "requires_confirm": True},
     # The only tool that changes the construct. Three flags, each earning its
     # place: signed-in because it writes; confirm because the user must agree
     # to their own DNA being altered; needs_target because the sequence comes
@@ -1714,6 +1790,37 @@ TOOL_SPECS: List[Dict[str, Any]] = [
             },
         },
     },
+    {
+        "type": "function",
+        "function": {
+            "name": "blast_sequence",
+            "description": (
+                "Search NCBI for sequences similar to this one — blastp "
+                "against nr for protein, blastn against nt for nucleotide. "
+                "Use it to answer 'what IS this sequence?' and 'what else "
+                "looks like it?'. "
+                "NOTE: this SENDS THE SEQUENCE to NCBI, outside this Space — "
+                "the user is asked to approve it, and may decline. "
+                "It is slow (30s to minutes, shared queue); if it has not "
+                "finished you get kind='still_running' with a job id, which "
+                "you should report honestly rather than waiting silently. "
+                "Report identity and coverage as BLAST returned them. No hits "
+                "is a real finding, not a reason to guess."
+            ),
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "sequence": {"type": "string",
+                                 "description": "Protein or DNA, 12+ residues."},
+                    "kind": {"type": "string",
+                             "enum": ["protein", "nucleotide"],
+                             "description": ("Leave unset to detect from the "
+                                             "alphabet.")},
+                },
+                "required": ["sequence"],
+            },
+        },
+    },
     {
         "type": "function",
         "function": {

dee/core/blast.py

@@ -0,0 +1,192 @@
+"""BLAST, in a place both the REST layer and the agent can reach.
+
+The engine has been able to BLAST since the identify feature: `server.py` runs
+blastp against `nr` on a background thread with a hash-keyed cache. The agent
+could not call any of it — `agent_tools.py` importing from `server.py` would be
+a circular import (server imports the agent), so the capability sat behind the
+Flask layer. This module is the same extraction `scoring.py` already did for
+the ESM-2 scorer, for the same reason.
+
+Two things here are not incidental.
+
+THE SEQUENCE LEAVES THE SPACE
+    Every other tool in this engine computes locally or sends at most
+    (gene_symbol, organism). BLAST is the exception: it posts the user's
+    actual residues to NCBI. `server.py` already enforces that server-side
+    (`blast_consent`, added by an audit — "must be enforced server-side, not
+    just by the consent modal"). The agent path uses the confirm gate for the
+    same purpose, so the model cannot route around it either. Nothing in this
+    module asks for consent — it is the caller's job, and both callers do it.
+
+IT IS SLOW, AND SAYING SO BEATS HANGING
+    NCBI BLAST is 30 s to several minutes, and the queue is shared. So the
+    agent path waits a bounded time and then reports "still running" with the
+    job id rather than holding a run open. A tool that returns "not finished
+    yet" is honest; one that blocks for four minutes looks broken.
+"""
+from __future__ import annotations
+
+import hashlib
+import logging
+import re
+import threading
+import time
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Optional
+
+logger = logging.getLogger("dee.blast")
+
+# How long the AGENT path waits before handing back a job id. The REST path
+# polls and does not use this.
+AGENT_WAIT_SECONDS = 75.0
+
+# blastp/nr for protein, blastn/nt for nucleotide. Both are the public web
+# service; there is no local database on this box.
+_PROGRAMS = {
+    "protein":    {"program": "blastp", "database": "nr"},
+    "nucleotide": {"program": "blastn", "database": "nt"},
+}
+
+_AA_ONLY = set("EFILPQZXJBOU")     # residues that cannot be DNA/RNA
+
+
+@dataclass
+class BlastJob:
+    job_id: str
+    seq_hash: str
+    kind: str = "protein"
+    status: str = "pending"
+    started_at: float = field(default_factory=time.time)
+    finished_at: Optional[float] = None
+    hits: List[Dict[str, Any]] = field(default_factory=list)
+    error: Optional[str] = None
+
+    def elapsed(self) -> float:
+        end = self.finished_at if self.finished_at else time.time()
+        return round(end - self.started_at, 1)
+
+    def done(self) -> bool:
+        return self.status in ("done", "error")
+
+    def public(self) -> Dict[str, Any]:
+        return {"job_id": self.job_id, "status": self.status, "kind": self.kind,
+                "elapsed_seconds": self.elapsed(), "hits": self.hits,
+                "error": self.error}
+
+
+_CACHE: Dict[str, BlastJob] = {}     # (kind, seq_hash) -> job
+_BY_ID: Dict[str, BlastJob] = {}
+_LOCK = threading.Lock()
+
+
+def hash_sequence(seq: str) -> str:
+    return hashlib.sha256(seq.encode("utf-8")).hexdigest()[:16]
+
+
+def extract_organism(hit_def: str) -> str:
+    """'GFP [Aequorea victoria]' -> 'Aequorea victoria'."""
+    m = re.search(r"\[([^\]]+)\]", hit_def or "")
+    return m.group(1) if m else ""
+
+
+def guess_kind(seq: str) -> str:
+    """protein or nucleotide, from the alphabet.
+
+    Deliberately conservative: only letters that CANNOT be nucleotides make it
+    protein. A short ACGT-only peptide is genuinely ambiguous, and calling it
+    nucleotide is the safer error — blastn on a peptide returns nothing, while
+    blastp on DNA returns confident nonsense.
+    """
+    s = re.sub(r"[^A-Za-z]", "", seq or "").upper()
+    if not s:
+        return "nucleotide"
+    return "protein" if set(s) & _AA_ONLY else "nucleotide"
+
+
+def _run(job: BlastJob, seq: str, hitlist: int, expect: float) -> None:
+    try:
+        from Bio.Blast import NCBIWWW, NCBIXML
+
+        cfg = _PROGRAMS[job.kind]
+        job.status = "submitting"
+        handle = NCBIWWW.qblast(program=cfg["program"], database=cfg["database"],
+                                sequence=seq, hitlist_size=hitlist, expect=expect)
+        job.status = "parsing"
+        hits: List[Dict[str, Any]] = []
+        for record in NCBIXML.parse(handle):
+            for alignment in record.alignments[:hitlist]:
+                if not alignment.hsps:
+                    continue
+                hsp = alignment.hsps[0]          # best HSP for this subject
+                hits.append({
+                    "title": (alignment.hit_def or "")[:180],
+                    "accession": getattr(alignment, "accession", "") or "",
+                    "organism": extract_organism(alignment.hit_def or ""),
+                    "identity_pct": round(100.0 * hsp.identities /
+                                          max(1, hsp.align_length), 1),
+                    "coverage_pct": round(100.0 * hsp.align_length /
+                                          max(1, len(seq)), 1),
+                    "evalue": float(hsp.expect),
+                    "bit_score": round(float(hsp.bits), 1),
+                    "align_length": int(hsp.align_length),
+                })
+            break                                 # one query in, one record out
+        job.hits = hits
+        job.status = "done"
+    except Exception as exc:                      # noqa: BLE001
+        logger.exception("BLAST failed")
+        # The message is shown to a scientist, so it names the service rather
+        # than leaking a stack shape.
+        job.error = f"NCBI BLAST did not return a result ({type(exc).__name__})."
+        job.status = "error"
+    finally:
+        job.finished_at = time.time()
+
+
+def submit(seq: str, *, kind: Optional[str] = None, hitlist: int = 8,
+           expect: float = 1e-5) -> BlastJob:
+    """Start (or reuse) a BLAST for this sequence. Returns immediately."""
+    seq = re.sub(r"[^A-Za-z]", "", seq or "").upper()
+    kind = kind if kind in _PROGRAMS else guess_kind(seq)
+    key = f"{kind}:{hash_sequence(seq)}"
+    with _LOCK:
+        cached = _CACHE.get(key)
+        if cached is not None:
+            return cached
+        job = BlastJob(job_id=hashlib.sha256(key.encode()).hexdigest()[:12],
+                       seq_hash=hash_sequence(seq), kind=kind)
+        _CACHE[key] = job
+        _BY_ID[job.job_id] = job
+    threading.Thread(target=_run, args=(job, seq, hitlist, expect),
+                     daemon=True).start()
+    return job
+
+
+def get(job_id: str) -> Optional[BlastJob]:
+    with _LOCK:
+        return _BY_ID.get(job_id)
+
+
+def wait(job: BlastJob, timeout: Optional[float] = None,
+         poll: float = 0.25) -> BlastJob:
+    """Block until the job settles or `timeout` passes. Never raises.
+
+    `timeout=None` reads AGENT_WAIT_SECONDS *at call time*, not as a default
+    bound at import. Written the obvious way — `timeout=AGENT_WAIT_SECONDS` in
+    the signature — the value is frozen when the module loads, so the constant
+    cannot be tuned per deployment and a test that lowers it still waits the
+    full 75 s. (It did: this file's suite took 77 s until this was fixed.)
+    """
+    if timeout is None:
+        timeout = AGENT_WAIT_SECONDS
+    deadline = time.time() + max(0.0, timeout)
+    while time.time() < deadline and not job.done():
+        time.sleep(poll)
+    return job
+
+
+def reset_cache() -> None:
+    """Tests only."""
+    with _LOCK:
+        _CACHE.clear()
+        _BY_ID.clear()

dee/core/orchestrator.py

@@ -137,6 +137,7 @@ _TOOL_UI: Dict[str, Dict[str, Any]] = {
     "map_plasmid":            {"view": "plasmid",   "verb": "Mapping the construct"},
     "lookup_vector":          {"view": None,        "verb": "Looking up the vector"},
     "simulate_assembly":      {"view": "plasmid",   "verb": "Assembling the construct"},
+    "blast_sequence":         {"view": None,        "verb": "Searching NCBI"},
     "edit_sequence":          {"view": "plasmid",   "verb": "Editing the construct"},
     # No painter: cut sites are an answer, not a canvas. Claiming the Map tab
     # for them would tell the user to look at a tab that never changed.
@@ -260,6 +261,15 @@ def _confirm_detail(name: str, args: Dict[str, Any]) -> str:
                 f"Records your measured results for {label} against your "
                 f"account, and contributes de-identified substitution effects "
                 f"to the shared commons.")
+    if name == "blast_sequence":
+        n = len(re.sub(r"[^A-Za-z]", "", str(args.get("sequence") or "")))
+        # Names what LEAVES, in the user's terms. This is the only tool in the
+        # engine that sends their residues off-Space, and the card is the only
+        # place they find that out.
+        return (f"Sends {n:,} residues of this sequence to NCBI BLAST — a "
+                f"public service outside this Space. Every other tool here "
+                f"keeps your sequence local. It is not published or "
+                f"attributed to you, but the sequence itself does leave.")
     if name == "edit_sequence":
         raw = args.get("edits") if args.get("edits") is not None else args.get("edit")
         labels = ", ".join(str(e) for e in (raw if isinstance(raw, list) else [raw]) if e)
@@ -309,6 +319,14 @@ def _summarize(name: str, result: Dict[str, Any]) -> str:
             size = f"{result.get('length'):,} bp"
             return (f"{labels} · {size}" if not d
                     else f"{labels} · {size} ({d:+d})")
+        if name == "blast_sequence":
+            n = result.get("hit_count")
+            if not n:
+                return "no significant hits"
+            top = result.get("top") or {}
+            return (f"{n} hit{'' if n == 1 else 's'} · top "
+                    f"{top.get('identity_pct')}% id"
+                    + (f" · {top.get('organism')}" if top.get("organism") else ""))
         if name == "simulate_assembly":
             j = result.get("junctions") or []
             d = result.get("designed_junctions") or 0

tests/test_blast_tool.py

@@ -0,0 +1,206 @@
+"""BLAST, reachable by the agent — and gated, because the sequence leaves.
+
+`server.py` has run blastp against `nr` since the identify feature: background
+thread, hash-keyed cache, organism extraction. The agent could not call any of
+it, because `agent_tools.py` importing from `server.py` is a circular import
+(server imports the agent). So "what IS this sequence?" and "find me homologs"
+were unanswerable, and blastn did not exist at all.
+
+The property that matters most here is not speed or parsing. It is that this
+is **the only tool in the engine that sends the user's actual residues off the
+Space**. Everything else computes locally or transmits at most
+(gene_symbol, organism). An audit already forced the REST path to enforce
+consent server-side rather than trusting the UI modal ("Audit H1: BLAST
+consent must be enforced server-side"); the agent path must be at least as
+strict, or the model becomes the hole in the policy.
+
+So these tests are mostly about refusal, disclosure and honest degradation —
+not about hits. Nothing here calls NCBI; the network is stubbed, because a
+test that depends on a shared public queue is a test that fails on Tuesdays.
+"""
+import re
+
+import pytest
+
+from dee.core import agent_tools as t
+from dee.core import blast
+from dee.core import orchestrator as orch
+
+TEM1 = "MSIQHFRVALIPFFAAFCLPVFAHPETLVKVKDAEDQLGARVGYIELDLNSGKILESFRPEERFP"
+DNA = "ATGCGTACGATCGATCGGCTAGCTAGCTTAAGGCCTTAAGGATCCGAATTCAAGCTTGCGGCCGC"
+
+
+@pytest.fixture(autouse=True)
+def _clean():
+    blast.reset_cache()
+    yield
+    blast.reset_cache()
+
+
+@pytest.fixture
+def stub(monkeypatch):
+    """Answer instantly with a fixed hit set. Never touches the network."""
+    calls = []
+
+    def fake(job, seq, hitlist, expect):
+        calls.append({"kind": job.kind, "seq": seq, "hitlist": hitlist})
+        job.hits = [{"title": "beta-lactamase TEM [Escherichia coli]",
+                     "accession": "P62593", "organism": "Escherichia coli",
+                     "identity_pct": 99.2, "coverage_pct": 100.0,
+                     "evalue": 1e-40, "bit_score": 210.0, "align_length": 65}]
+        job.status = "done"
+        import time as _t
+        job.finished_at = _t.time()
+    monkeypatch.setattr(blast, "_run", fake)
+    return calls
+
+
+# --------------------------------------------------------------------------- #
+# the consent property — the reason this tool is different
+# --------------------------------------------------------------------------- #
+def test_blast_is_the_only_tool_that_needs_approval_to_send_data_out():
+    """Not a UI nicety. The model must not be able to decide, on a user's
+    behalf, to publish their unpublished construct to a public service."""
+    assert t._TOOLS["blast_sequence"]["requires_confirm"] is True
+    assert orch._requires_confirm("blast_sequence") is True
+
+
+def test_the_card_says_the_sequence_leaves_and_how_much_of_it():
+    """"Allow blast_sequence?" would be a click-through. The user has to learn
+    the one fact that distinguishes this tool from every other one."""
+    card = orch._confirm_detail("blast_sequence", {"sequence": "M" * 286})
+    assert "286" in card
+    assert "NCBI" in card
+    assert "leave" in card.lower()
+    # and it must set the contrast against the rest of the engine
+    assert "local" in card.lower()
+    assert "blast_sequence" not in card
+
+
+def test_no_other_read_only_tool_became_gated_by_accident():
+    """The gate is meaningful only if it stays rare."""
+    gated = [n for n in t._TOOLS if orch._requires_confirm(n)]
+    assert set(gated) == {"log_outcome", "edit_sequence", "blast_sequence"}, gated
+
+
+# --------------------------------------------------------------------------- #
+# it actually works
+# --------------------------------------------------------------------------- #
+def test_a_protein_search_returns_the_hits_verbatim(stub):
+    out = t.execute_tool("blast_sequence", {"sequence": TEM1}, auth_anonymous=True)
+    assert out["ok"] is True
+    assert out["search"] == "protein"
+    assert out["hit_count"] == 1
+    assert out["top"]["identity_pct"] == 99.2
+    assert out["top"]["organism"] == "Escherichia coli"
+
+
+def test_dna_goes_to_blastn_not_blastp(stub):
+    """blastp on DNA returns confident nonsense, so the alphabet decides."""
+    t.execute_tool("blast_sequence", {"sequence": DNA}, auth_anonymous=True)
+    assert stub[-1]["kind"] == "nucleotide"
+
+
+def test_the_caller_can_override_the_guess(stub):
+    t.execute_tool("blast_sequence", {"sequence": DNA, "kind": "protein"},
+                   auth_anonymous=True)
+    assert stub[-1]["kind"] == "protein"
+
+
+def test_kind_detection_errs_toward_nucleotide():
+    """Only letters that CANNOT be bases make it protein. A short ACGT-only
+    peptide is genuinely ambiguous, and blastn on a peptide finds nothing —
+    which is a safe failure — while blastp on DNA invents a story."""
+    assert blast.guess_kind(TEM1) == "protein"
+    assert blast.guess_kind(DNA) == "nucleotide"
+    assert blast.guess_kind("ACGTACGTACGT") == "nucleotide"
+
+
+def test_repeat_searches_reuse_the_job(stub):
+    """NCBI's queue is shared and slow; asking twice for the same sequence
+    must not queue twice."""
+    a = blast.submit(TEM1)
+    b = blast.submit(TEM1)
+    assert a.job_id == b.job_id
+    assert len(stub) == 1
+
+
+# --------------------------------------------------------------------------- #
+# honest failure
+# --------------------------------------------------------------------------- #
+def test_no_hits_is_reported_as_a_finding_not_a_gap(stub, monkeypatch):
+    def empty(job, seq, hitlist, expect):
+        job.hits = []
+        job.status = "done"
+        import time as _t
+        job.finished_at = _t.time()
+    monkeypatch.setattr(blast, "_run", empty)
+    out = t.execute_tool("blast_sequence", {"sequence": TEM1}, auth_anonymous=True)
+    assert out["ok"] is True and out["hit_count"] == 0
+    assert "novel or synthetic" in out["next"]
+    assert "guessing" in out["next"]
+
+
+def test_a_still_running_search_says_so_instead_of_hanging(monkeypatch):
+    """NCBI can take minutes. A tool that blocks the conversation reads as
+    broken; one that reports a job id is merely slow."""
+    monkeypatch.setattr(blast, "_run", lambda *a, **k: None)   # never finishes
+    # Readable at call time now — see blast.wait(). Setting it as a
+    # signature default froze it at import and this test waited 75s.
+    monkeypatch.setattr(blast, "AGENT_WAIT_SECONDS", 0.2)
+    out = t.execute_tool("blast_sequence", {"sequence": TEM1}, auth_anonymous=True)
+    assert out["ok"] is False
+    assert out["kind"] == "still_running"
+    assert out["job_id"]
+    assert "never invent hits" in out["next"].lower()
+
+
+def test_an_ncbi_failure_does_not_become_a_remembered_answer(stub, monkeypatch):
+    def boom(job, seq, hitlist, expect):
+        job.error = "NCBI BLAST did not return a result (URLError)."
+        job.status = "error"
+        import time as _t
+        job.finished_at = _t.time()
+    monkeypatch.setattr(blast, "_run", boom)
+    out = t.execute_tool("blast_sequence", {"sequence": TEM1}, auth_anonymous=True)
+    assert out["ok"] is False and out["kind"] == "blast_failed"
+    assert "from memory" in out["next"]
+
+
+def test_a_too_short_query_is_refused():
+    out = t.execute_tool("blast_sequence", {"sequence": "MSIQ"},
+                         auth_anonymous=True)
+    assert out["ok"] is False and "12" in out["error"]
+
+
+def test_an_unknown_kind_is_refused():
+    out = t.execute_tool("blast_sequence", {"sequence": TEM1, "kind": "rna"},
+                         auth_anonymous=True)
+    assert out["ok"] is False and "protein" in out["error"]
+
+
+# --------------------------------------------------------------------------- #
+# wiring
+# --------------------------------------------------------------------------- #
+def test_the_engine_lives_where_both_callers_can_reach_it():
+    """The reason this was unreachable: agent_tools importing server.py is a
+    circular import. Same extraction scoring.py already did."""
+    src = open("dee/core/agent_tools.py", encoding="utf-8").read()
+    assert "from dee.core import blast" in src
+    assert "import server" not in src
+
+
+def test_the_summary_leads_with_what_it_found(stub):
+    out = t.execute_tool("blast_sequence", {"sequence": TEM1}, auth_anonymous=True)
+    line = orch._summarize("blast_sequence", out)
+    assert "1 hit" in line and "99.2" in line
+    assert "Escherichia coli" in line
+
+
+def test_the_description_warns_that_the_sequence_is_sent(stub):
+    spec = next(s for s in orch.TOOL_SPECS
+                if s["function"]["name"] == "blast_sequence")
+    d = spec["function"]["description"]
+    assert "SENDS THE SEQUENCE" in d
+    assert "may decline" in d
+    assert "still_running" in d

tests/test_confirm_gate.py

@@ -293,6 +293,11 @@ def test_remember_is_not_double_gated():
 # puts it in a bucket, which is the decision the gate exists to force.
 _GATED = {
     "log_outcome",        # writes measured data to the user AND to the commons
+    # The ONLY tool that sends the user's residues off-Space. Everything else
+    # computes locally or transmits at most (gene_symbol, organism). An audit
+    # already forced the REST path to enforce this server-side; the agent path
+    # uses the gate so the model cannot be the hole in the policy.
+    "blast_sequence",
     # Alters the construct itself. This test is what forced the decision when
     # the tool was added — it failed on the next run after edit_sequence
     # appeared, before any of its own tests existed, which is exactly what a