Deploy e90fe57

Two ways use compounds, neither of which is a model

Source: https://github.com/WINTER4000/turingDNA/commit/e90fe57533d072ec6e2e46c133a79b8d4793ed84

dee/core/orchestrator.py

@@ -2501,7 +2501,46 @@ def build_system_prompt(anonymous: bool, workspace: Optional[Dict[str, Any]] = N
     # reason: a durable preference must never read as a competing target.
     return (SYSTEM_PROMPT
             + (_ANON_NOTE if anonymous else "")
+            + _worked_examples_note()
             + (continuity or "")
             + _memory_note(memories or [])
             + _workspace_note(workspace or {})
             + _target_note(target or {}))
+
+
+# Tool chains that real users have completed cleanly, refreshed periodically
+# from the transcript corpus. Cached in-process because building it reads the
+# whole corpus and the answer changes on the scale of days, not requests.
+_WORKED_CACHE: Dict[str, Any] = {"text": "", "at": 0.0}
+_WORKED_TTL = 6 * 3600
+
+
+def _worked_examples_note() -> str:
+    """Observed successful tool chains, as a prompt section.
+
+    This is the point where the corpus feeds back into behaviour: the agent is
+    told how this engine is actually driven by people who finished, rather
+    than only what I guessed when writing the tool descriptions.
+
+    Empty string when nothing clears the k-anonymity floor, which is the
+    normal state early on — an empty "PATHS THAT WORK" heading would read as
+    the engine having no idea what works.
+    """
+    now = time.time()
+    if _WORKED_CACHE["text"] and now - _WORKED_CACHE["at"] < _WORKED_TTL:
+        return _WORKED_CACHE["text"]
+    text = ""
+    try:
+        from dee import auth as _auth
+        from dee.core import worked_examples as _we
+        runs = _auth.scan_agent_runs(limit=400)
+        if runs:
+            section = _we.as_prompt_section(_we.promote(runs))
+            text = ("\n\n" + section) if section else ""
+    except Exception:  # noqa: BLE001
+        # Never let this break a run. A missing prompt section costs a little
+        # guidance; an exception here costs the whole conversation.
+        logger.warning("worked-examples section unavailable", exc_info=True)
+        text = ""
+    _WORKED_CACHE.update(text=text, at=now)
+    return text

dee/core/resolution_cache.py

@@ -0,0 +1,162 @@
+"""Resolve TP53 once. Never pay for it again.
+
+THE PROBLEM
+-----------
+Every gene-symbol and accession lookup goes out to Ensembl, UniProt or NCBI,
+every time, for every user. `exon.py` has a process-local dict, but a Hugging
+Face Space sleeps on idle and restarts cold, so in practice the same handful of
+genes — TP53, GFP, Cas9, pUC19 — are re-fetched from scratch forever.
+
+That costs three things: latency on the request a user is actually waiting on,
+a shared outbound IP burning someone else's rate limit, and correctness — when
+Ensembl is slow or throttling, a lookup that has succeeded a thousand times
+fails, and the user is told their gene might not exist.
+
+WHY THIS IS THE HONEST KIND OF "GETS SMARTER"
+---------------------------------------------
+It compounds with use and involves no model, no training and no inference. The
+thousandth user asking for TP53 gets an instant answer *because* nine hundred
+and ninety-nine people asked first. That is a real flywheel, and unlike a
+learned one it cannot be wrong in a way nobody notices — the cached value is
+byte-identical to what the database returned.
+
+WHAT MAY AND MAY NOT BE CACHED
+------------------------------
+Only resolutions keyed by a PUBLIC identifier: (gene symbol, organism) or an
+accession. Those keys are not personal, and the values are public database
+records, which is why one shared cache across all users is correct rather than
+a leak.
+
+A pasted sequence is never cached. Not by content, not by hash, not as a key.
+`kind == "sequence"` is refused at the door — see :func:`cacheable`. The
+standing rule is that a user's own sequence never leaves the Space, and a
+cross-user cache is very much leaving.
+
+STALENESS
+---------
+Database records change: RefSeq versions increment, Ensembl re-annotates. So
+entries carry a TTL and are re-fetched after it. The TTL is long because these
+records are stable on the timescale of a design project, and a stale-by-a-week
+CDS is a far smaller problem than the lookup failing outright.
+"""
+from __future__ import annotations
+
+import re
+import threading
+import time
+from collections import OrderedDict
+from typing import Any, Dict, Optional, Tuple
+
+# Entries older than this are re-fetched. Sequence records are stable over
+# weeks; this is about eventually noticing a re-annotation, not about
+# freshness in any real-time sense.
+TTL_SECONDS = 14 * 24 * 3600           # 14 days
+MAX_ENTRIES = 2_000                    # in-process ceiling; ~50 MB worst case
+
+# Never cached. A pasted sequence is the user's own data and a shared cache is
+# by definition cross-user.
+UNCACHEABLE_KINDS = {"sequence", "empty", "unknown"}
+
+_LOCK = threading.Lock()
+_MEM: "OrderedDict[str, Dict[str, Any]]" = OrderedDict()
+_STATS = {"hits": 0, "misses": 0, "stores": 0, "evictions": 0, "refused": 0}
+
+
+def make_key(kind: str, identifier: str, organism: str = "") -> str:
+    """Stable cache key. Case- and whitespace-insensitive.
+
+    Organism is part of the key because TP53 exists in dozens of species and
+    they are different sequences — a key that dropped it would serve human
+    TP53 to someone who asked for zebrafish, which is worse than a miss.
+    """
+    ident = re.sub(r"\s+", "", (identifier or "")).upper()
+    org = re.sub(r"\s+", " ", (organism or "")).strip().lower()
+    return f"{(kind or '').lower()}|{ident}|{org}"
+
+
+def cacheable(kind: str, result: Optional[Dict[str, Any]] = None) -> bool:
+    """May a resolution of this kind be stored in a SHARED cache?
+
+    The gate that keeps user sequences out. Deliberately a whitelist-free
+    check on the one thing that disqualifies — a pasted sequence — rather than
+    a list of approved kinds, so a new public identifier type is cacheable the
+    day it is added instead of silently bypassing the cache forever.
+    """
+    if (kind or "").lower() in UNCACHEABLE_KINDS:
+        return False
+    if result is not None and not result.get("ok"):
+        # A failure is not cached. Ensembl being down for ten seconds must not
+        # become "this gene does not exist" for the next fortnight.
+        return False
+    return True
+
+
+def get(kind: str, identifier: str, organism: str = "") -> Optional[Dict[str, Any]]:
+    """A previously resolved record, or None. Never raises."""
+    if not cacheable(kind):
+        return None
+    key = make_key(kind, identifier, organism)
+    now = time.time()
+    with _LOCK:
+        entry = _MEM.get(key)
+        if entry is None:
+            _STATS["misses"] += 1
+            return None
+        if now - entry["at"] > TTL_SECONDS:
+            _MEM.pop(key, None)
+            _STATS["misses"] += 1
+            return None
+        _MEM.move_to_end(key)          # LRU: a hit is a recency signal
+        _STATS["hits"] += 1
+        # A copy, so a caller mutating the result cannot poison the cache for
+        # everyone else — the failure mode of a shared cache that is very hard
+        # to trace back later.
+        out = dict(entry["value"])
+        out["cached"] = True
+        out["cached_age_seconds"] = int(now - entry["at"])
+        return out
+
+
+def put(kind: str, identifier: str, result: Dict[str, Any],
+        organism: str = "") -> bool:
+    """Store a successful resolution. Returns whether it was stored."""
+    if not cacheable(kind, result):
+        with _LOCK:
+            _STATS["refused"] += 1
+        return False
+    if not (result or {}).get("sequence"):
+        return False
+    key = make_key(kind, identifier, organism)
+    # `cached` is a property of the read, not of the record. Storing it would
+    # make the first served copy claim it came from cache.
+    value = {k: v for k, v in result.items()
+             if k not in ("cached", "cached_age_seconds")}
+    with _LOCK:
+        _MEM[key] = {"value": value, "at": time.time()}
+        _MEM.move_to_end(key)
+        _STATS["stores"] += 1
+        while len(_MEM) > MAX_ENTRIES:
+            _MEM.popitem(last=False)
+            _STATS["evictions"] += 1
+    return True
+
+
+def stats() -> Dict[str, Any]:
+    """Hit rate and size. The number worth watching is `hit_rate`: it is the
+    whole claim that this compounds."""
+    with _LOCK:
+        total = _STATS["hits"] + _STATS["misses"]
+        return {
+            **_STATS,
+            "entries": len(_MEM),
+            "hit_rate": round(_STATS["hits"] / total, 3) if total else 0.0,
+            "ttl_days": TTL_SECONDS // 86400,
+        }
+
+
+def clear() -> None:
+    """Tests, and the admin path when a record is known to have changed."""
+    with _LOCK:
+        _MEM.clear()
+        for k in _STATS:
+            _STATS[k] = 0

dee/core/resolve.py

@@ -27,6 +27,7 @@ from typing import Dict, Tuple
 
 from dee.core import accession as _accession
 from dee.core import exon as _exon
+from dee.core import resolution_cache as _rcache
 
 # ─── Identifier patterns ─────────────────────────────────────────────
 _ENSEMBL_TX   = re.compile(r"^ENS[A-Z]*T\d{6,}(?:\.\d+)?$", re.I)   # ENST…, ENSMUST…
@@ -362,13 +363,34 @@ def _resolve_via_uniprot(symbol: str, organism: str) -> Dict:
 
 
 def resolve_target(text: str, organism: str = "") -> Dict:
-    """Resolve pasted text to an editable sequence.
+    """Resolve pasted text to an editable sequence, serving a cached record
+    when this identifier has been resolved before.
 
     Returns a dict: {ok, kind, sequence, gene_symbol, label, source, error?}.
+
+    A thin wrapper rather than a check inside the resolver, because the
+    resolver has eight success paths and a cache written at each of them is a
+    cache that will eventually miss one.
+
+    A pasted sequence is never cached — `resolution_cache.cacheable` refuses
+    `kind == "sequence"` outright, so only public identifiers reach the store.
     """
     organism = (organism or "").lower().strip()
     kind, val = classify(text)
 
+    hit = _rcache.get(kind, val, organism)
+    if hit is not None:
+        return hit
+
+    result = _resolve_uncached(text, organism, kind, val)
+    # Failures are deliberately not stored: Ensembl being down for ten seconds
+    # must not become "this gene does not exist" for the next fortnight.
+    _rcache.put(kind, val, result, organism)
+    return result
+
+
+def _resolve_uncached(text: str, organism: str, kind: str, val: str) -> Dict:
+    """The actual resolution. Every network call lives below here."""
     if kind == "empty":
         return _err("Paste a DNA sequence, a gene symbol, or an accession.")
 

dee/core/worked_examples.py

@@ -0,0 +1,146 @@
+"""Paths that already worked, promoted from the corpus so the agent reuses them.
+
+THE IDEA
+--------
+`learn_signal.py` reads the transcripts for what went WRONG. The same corpus
+holds what went right, and that half has a use the failure half does not: a
+tool chain that N different people completed successfully is evidence about
+how this engine is actually driven, and it can be handed back to the agent.
+
+"To build a construct, people call lookup_vector → simulate_assembly →
+check_synthesis" is worth more in a system prompt than any sentence I could
+write about it, because it is observed rather than imagined, and it updates
+itself as the tools change.
+
+This is the honest kind of learning, same as the resolution cache: no model,
+no training, nothing that can be quietly wrong. It is counting.
+
+WHAT IS PROMOTED, AND WHAT IS NOT
+---------------------------------
+Only the SHAPE — the ordered list of tool names. Never the user's prompt,
+never arguments, never a sequence. A worked example that carried the question
+that produced it would be a transcript excerpt, and transcript excerpts across
+users are exactly what `learn_signal.field_report` refuses to emit.
+
+A run qualifies only if it:
+  * finished cleanly (`status == "done"`),
+  * called at least two tools — one call is not a path,
+  * had no failed tool call, and
+  * was never corrected by the user mid-run.
+
+That last one matters most. A run the user had to steer is not a worked
+example; it is a near-miss, and promoting it would teach the agent the route
+that needed fixing.
+
+And the same k-anonymity floor as everything else cross-user: a chain must
+have been completed by at least `min_users` DISTINCT users. One lab's unusual
+workflow is that lab's business, and a chain seen once is an anecdote.
+"""
+from __future__ import annotations
+
+import datetime as _dt
+from collections import Counter, defaultdict
+from typing import Any, Dict, Iterable, List, Optional, Set, Tuple
+
+from dee.core.aggregate import EFFECTIVE_DATE, MIN_USERS, AggregationGateError
+from dee.core.learn_signal import _events, _uid, is_correction
+
+# A chain longer than this is usually one exploratory session rather than a
+# repeatable path, and it will not generalise.
+MAX_CHAIN = 8
+MIN_CHAIN = 2
+
+
+def chain_of(run: Dict[str, Any]) -> Optional[Tuple[str, ...]]:
+    """The ordered tool names of a clean run, or None if it doesn't qualify.
+
+    Consecutive repeats collapse: calling fetch_sequence three times for three
+    genes is the same PATH as calling it once, and keeping the repetition
+    would fragment the counts across chains that mean the same thing.
+    """
+    if str(run.get("status") or "") != "done":
+        return None
+
+    names: List[str] = []
+    for ev in _events(run):
+        kind = ev.get("kind")
+        if kind == "steer" and is_correction(str(ev.get("text") or "")):
+            return None                     # a corrected run is a near-miss
+        if kind == "tool_result" and not ev.get("ok"):
+            return None                     # something failed; not a clean path
+        if kind == "tool_call":
+            name = str(ev.get("name") or "")
+            if name and (not names or names[-1] != name):
+                names.append(name)
+
+    if not (MIN_CHAIN <= len(names) <= MAX_CHAIN):
+        return None
+    return tuple(names)
+
+
+def promote(runs: Iterable[Dict[str, Any]], *,
+            min_users: int = MIN_USERS,
+            limit: int = 12,
+            today: Optional[_dt.date] = None,
+            enforce_gate: bool = True) -> Dict[str, Any]:
+    """Tool chains proven by real completed runs, ranked by how many distinct
+    users completed them.
+
+    Cross-user, so it inherits the same date gate as every other aggregation
+    in this engine.
+    """
+    day = today or _dt.date.today()
+    if enforce_gate and day < EFFECTIVE_DATE:
+        raise AggregationGateError(
+            f"Cross-user aggregation is gated until {EFFECTIVE_DATE.isoformat()} "
+            f"(Privacy/Terms v2.0 effective date); today is {day.isoformat()}."
+        )
+
+    runs = list(runs)
+    counts: Counter = Counter()
+    users: Dict[Tuple[str, ...], Set[str]] = defaultdict(set)
+    for run in runs:
+        chain = chain_of(run)
+        if chain is None:
+            continue
+        counts[chain] += 1
+        if _uid(run):
+            users[chain].add(_uid(run))
+
+    examples = [
+        {"chain": list(chain), "runs": n, "users": len(users[chain]),
+         "steps": len(chain)}
+        for chain, n in counts.most_common()
+        if len(users[chain]) >= min_users
+    ]
+    return {
+        "ok": True,
+        "examples": examples[:limit],
+        "runs_considered": len(runs),
+        "chains_seen": len(counts),
+        "min_users": min_users,
+        "effective_date": EFFECTIVE_DATE.isoformat(),
+        "note": ("Tool names only. No prompts, no arguments, no sequences — a "
+                 "worked example carrying the question that produced it would "
+                 "be a cross-user transcript excerpt."),
+    }
+
+
+def as_prompt_section(report: Dict[str, Any], limit: int = 6) -> str:
+    """The promoted chains, formatted for the agent's system prompt.
+
+    Returns "" when nothing clears the floor, and the caller must treat that
+    as "add no section" rather than "add an empty heading" — an empty
+    "PATHS THAT WORK" header reads as the engine having no idea what works.
+    """
+    rows = (report or {}).get("examples") or []
+    if not rows:
+        return ""
+    lines = ["PATHS THAT HAVE WORKED",
+             "Observed from completed runs — not rules, and not a list of the "
+             "only valid routes. Prefer them when they fit; deviate when the "
+             "task differs."]
+    for r in rows[:limit]:
+        lines.append(f"  {' -> '.join(r['chain'])}   "
+                     f"({r['users']} users, {r['runs']} runs)")
+    return "\n".join(lines)

dee/server.py

@@ -3258,6 +3258,45 @@ def create_app() -> Flask:
             return Response(_ls.as_text(rep), mimetype="text/plain")
         return jsonify(rep)
 
+    @app.get("/api/admin/worked-examples")
+    def admin_worked_examples() -> Response:
+        """Tool chains real users have completed cleanly.
+
+        The other half of /api/admin/learn-signal: that one reads the corpus
+        for what went wrong, this one for what went right. Same privacy
+        regime — tool names only, k-anonymous, date-gated. These are fed back
+        into the agent's system prompt, so this endpoint is also how you see
+        what it is currently being told.
+        """
+        from dee.core import worked_examples as _we
+        from dee.core.aggregate import AggregationGateError as _Gate
+
+        if not _admin_ok():
+            return jsonify({"ok": False, "error": "forbidden"}), 403
+        try:
+            limit = max(1, min(int(request.args.get("limit", 400)), 5000))
+        except ValueError:
+            limit = 400
+        runs = _auth.scan_agent_runs(limit=limit)
+        try:
+            rep = _we.promote(runs)
+        except _Gate as exc:
+            return jsonify({"ok": False, "error": str(exc)}), 403
+        if request.args.get("format") == "prompt":
+            return Response(_we.as_prompt_section(rep) or
+                            "(nothing has cleared the k-anonymity floor yet)",
+                            mimetype="text/plain")
+        return jsonify(rep)
+
+    @app.get("/api/admin/cache-stats")
+    def admin_cache_stats() -> Response:
+        """Resolution-cache hit rate — the number that says whether repeat
+        lookups are actually compounding."""
+        from dee.core import resolution_cache as _rc
+        if not _admin_ok():
+            return jsonify({"ok": False, "error": "forbidden"}), 403
+        return jsonify({"ok": True, "resolution_cache": _rc.stats()})
+
     @app.post("/api/admin/run-benchmarks")
     def admin_run_benchmarks() -> Response:
         """Produce REAL validation numbers — runs the live ESM-2 model already

tests/conftest.py

@@ -30,3 +30,22 @@ def _no_leaked_handover():
     _orch._HANDOVERS.clear()
     yield
     _orch._HANDOVERS.clear()
+
+
+@pytest.fixture(autouse=True)
+def _no_leaked_resolutions():
+    """The resolution cache is process-global by design — it is shared across
+    users on purpose, because a gene symbol is public and re-fetching TP53 for
+    every visitor is the waste it exists to remove.
+
+    That makes it the same hazard as _HANDOVERS above. Caught immediately:
+    test_resolve_ensembl_tx resolves ENST00000269305 successfully, and
+    test_resolve_ensembl_fetch_failure then asserts the SAME id fails when
+    Ensembl returns nothing — which it does not, because the first test's
+    result is still cached. The product behaviour is correct (surviving a
+    network blip is the whole feature); the test needs a cold cache.
+    """
+    from dee.core import resolution_cache as _rc
+    _rc.clear()
+    yield
+    _rc.clear()

tests/test_compounding.py

@@ -0,0 +1,249 @@
+"""Two ways the engine gets better with use, neither of which is a model.
+
+The founder's ask was that "both engine and turing chat panel get smarter each
+day". The tempting answer is to train something. The honest answer, given that
+platform labs = 0 and there is no measured data to train on, is to make use
+itself compound:
+
+  resolution cache   the thousandth person to ask for TP53 gets an instant
+                     answer BECAUSE nine hundred and ninety-nine asked first.
+  worked examples    a tool chain N different users completed is evidence
+                     about how this engine is actually driven, and it can go
+                     straight back into the agent's prompt.
+
+Neither can be quietly wrong. The cached value is byte-identical to what the
+database returned, and the promoted chain is a count. That is the whole appeal
+— a learned component that degrades silently is the failure this product is
+positioned against.
+"""
+import datetime as _dt
+import json
+
+import pytest
+
+from dee.core import resolution_cache as rc
+from dee.core import worked_examples as we
+from dee.core.aggregate import EFFECTIVE_DATE, AggregationGateError
+
+
+@pytest.fixture(autouse=True)
+def _clean():
+    rc.clear()
+    yield
+    rc.clear()
+
+
+def ev(kind, seq, **kw):
+    return dict(kind=kind, seq=seq, at=1754000000.0 + seq, **kw)
+
+
+def clean_run(chain, user="u1", status="done", extra=()):
+    """A run that called `chain` in order and finished."""
+    events = []
+    for i, name in enumerate(chain):
+        events.append(ev("tool_call", i * 2 + 1, id=f"c{i}", name=name))
+        events.append(ev("tool_result", i * 2 + 2, id=f"c{i}", name=name, ok=True))
+    events.extend(extra)
+    return {"run_id": f"r{user}{'-'.join(chain)}", "user_id": user,
+            "status": status, "events": events}
+
+
+# --------------------------------------------------------------------------- #
+# resolution cache
+# --------------------------------------------------------------------------- #
+RECORD = {"ok": True, "kind": "refseq", "sequence": "ATGC" * 50,
+          "label": "NM_000546 · 200 nt", "source": "ncbi", "gene_symbol": ""}
+
+
+def test_a_second_lookup_is_served_from_the_first():
+    assert rc.get("refseq", "NM_000546") is None
+    rc.put("refseq", "NM_000546", RECORD)
+    hit = rc.get("refseq", "NM_000546")
+    assert hit is not None
+    assert hit["sequence"] == RECORD["sequence"], "cached value must be identical"
+    assert hit["cached"] is True
+
+
+def test_a_pasted_sequence_is_never_stored_in_a_shared_cache():
+    """The standing rule is that a user's own sequence never leaves the Space.
+    A cross-user cache is very much leaving."""
+    pasted = {"ok": True, "kind": "sequence", "sequence": "ACGT" * 40}
+    assert rc.cacheable("sequence") is False
+    assert rc.put("sequence", "ACGT" * 40, pasted) is False
+    assert rc.get("sequence", "ACGT" * 40) is None
+    assert rc.stats()["stores"] == 0
+
+
+def test_the_refusal_is_on_kind_not_on_a_list_of_approved_types():
+    """A whitelist would mean each new public identifier type silently
+    bypasses the cache until someone remembers to add it."""
+    assert rc.cacheable("refseq") and rc.cacheable("uniprot")
+    assert rc.cacheable("some_new_database_added_next_year") is True
+    assert rc.cacheable("sequence") is False
+
+
+def test_a_failure_is_not_cached():
+    """Ensembl down for ten seconds must not become 'this gene does not
+    exist' for the next fortnight."""
+    assert rc.put("symbol", "TP53", {"ok": False, "error": "timeout"}, "human") is False
+    assert rc.get("symbol", "TP53", "human") is None
+
+
+def test_organism_is_part_of_the_key():
+    """TP53 exists in dozens of species and they are different sequences.
+    Serving human TP53 to someone who asked for zebrafish is worse than a
+    miss."""
+    rc.put("symbol", "TP53", RECORD, "human")
+    assert rc.get("symbol", "TP53", "human") is not None
+    assert rc.get("symbol", "TP53", "zebrafish") is None
+
+
+def test_keys_are_case_and_whitespace_insensitive():
+    rc.put("symbol", "TP53", RECORD, "human")
+    assert rc.get("symbol", " tp53 ", "Human") is not None
+
+
+def test_a_caller_mutating_the_result_cannot_poison_the_cache():
+    """The hard-to-trace failure mode of any shared cache."""
+    rc.put("refseq", "NM_1", RECORD)
+    first = rc.get("refseq", "NM_1")
+    first["sequence"] = "TAMPERED"
+    assert rc.get("refseq", "NM_1")["sequence"] == RECORD["sequence"]
+
+
+def test_entries_expire_so_a_reannotated_record_is_refetched(monkeypatch):
+    rc.put("refseq", "NM_1", RECORD)
+    assert rc.get("refseq", "NM_1") is not None
+    monkeypatch.setattr(rc, "TTL_SECONDS", -1)
+    assert rc.get("refseq", "NM_1") is None
+
+
+def test_the_cache_is_bounded():
+    monkey = rc.MAX_ENTRIES
+    try:
+        rc.MAX_ENTRIES = 5
+        for i in range(20):
+            rc.put("refseq", f"NM_{i}", RECORD)
+        assert rc.stats()["entries"] <= 5
+        assert rc.stats()["evictions"] >= 15
+    finally:
+        rc.MAX_ENTRIES = monkey
+
+
+def test_the_resolver_serves_repeats_without_going_back_out(monkeypatch):
+    """End to end through resolve_target, which is where it has to work."""
+    from dee.core import resolve
+    calls = []
+
+    def fake(text, organism, kind, val):
+        calls.append(val)
+        return {"ok": True, "kind": "refseq", "sequence": "ATGC" * 30,
+                "label": "x", "source": "ncbi", "gene_symbol": ""}
+    monkeypatch.setattr(resolve, "_resolve_uncached", fake)
+
+    a = resolve.resolve_target("NM_000546")
+    b = resolve.resolve_target("NM_000546")
+    assert calls == ["NM_000546"], "the second lookup went back to the network"
+    assert a["sequence"] == b["sequence"]
+    assert b.get("cached") is True
+
+
+def test_hit_rate_is_reported_because_it_is_the_whole_claim():
+    rc.put("refseq", "NM_1", RECORD)
+    rc.get("refseq", "NM_1")
+    rc.get("refseq", "NM_2")
+    assert rc.stats()["hit_rate"] == 0.5
+
+
+# --------------------------------------------------------------------------- #
+# worked examples
+# --------------------------------------------------------------------------- #
+BUILD = ("lookup_vector", "simulate_assembly", "check_synthesis")
+
+
+def test_a_chain_completed_by_enough_users_is_promoted():
+    runs = [clean_run(BUILD, user=f"u{i}") for i in range(4)]
+    out = we.promote(runs, min_users=3)
+    assert out["examples"][0]["chain"] == list(BUILD)
+    assert out["examples"][0]["users"] == 4
+
+
+def test_a_corrected_run_is_not_a_worked_example():
+    """A run the user had to steer is a near-miss. Promoting it teaches the
+    agent the route that needed fixing."""
+    runs = [clean_run(BUILD, user=f"u{i}") for i in range(3)]
+    runs.append(clean_run(BUILD, user="u9", extra=[
+        ev("steer", 99, text="no, that's the wrong backbone")]))
+    out = we.promote(runs, min_users=1)
+    assert out["examples"][0]["users"] == 3, "the corrected run was counted"
+
+
+def test_a_run_with_a_failed_tool_is_not_promoted():
+    bad = {"run_id": "b", "user_id": "u1", "status": "done", "events": [
+        ev("tool_call", 1, id="c1", name="fetch_sequence"),
+        ev("tool_result", 2, id="c1", name="fetch_sequence", ok=False, error="x"),
+        ev("tool_call", 3, id="c2", name="lookup_vector"),
+        ev("tool_result", 4, id="c2", name="lookup_vector", ok=True)]}
+    assert we.chain_of(bad) is None
+
+
+def test_an_unfinished_run_is_not_promoted():
+    assert we.chain_of(clean_run(BUILD, status="awaiting_input")) is None
+    assert we.chain_of(clean_run(BUILD, status="error")) is None
+
+
+def test_a_single_tool_call_is_not_a_path():
+    assert we.chain_of(clean_run(("fetch_sequence",))) is None
+
+
+def test_consecutive_repeats_collapse():
+    """Fetching three genes is the same PATH as fetching one. Keeping the
+    repetition fragments the counts across chains that mean the same thing."""
+    chain = we.chain_of(clean_run(
+        ("fetch_sequence", "fetch_sequence", "fetch_sequence", "fold_structure")))
+    assert chain == ("fetch_sequence", "fold_structure")
+
+
+def test_one_lab_s_unusual_workflow_is_not_published_to_everyone():
+    runs = [clean_run(BUILD, user=f"u{i}") for i in range(3)]
+    runs.append(clean_run(("design_crispr_guides", "check_prior_art"), user="solo"))
+    out = we.promote(runs, min_users=3)
+    chains = [tuple(e["chain"]) for e in out["examples"]]
+    assert BUILD in chains
+    assert ("design_crispr_guides", "check_prior_art") not in chains
+
+
+def test_no_user_content_survives_promotion():
+    """Same rule as the field report: a worked example carrying the question
+    that produced it is a cross-user transcript excerpt."""
+    secret = "ZZQXSECRETZZ"
+    runs = [clean_run(BUILD, user=f"u{i}", extra=[
+        ev("user", 90, text=f"engineer {secret}"),
+        ev("text", 91, text=f"done with {secret}")]) for i in range(4)]
+    out = we.promote(runs, min_users=1)
+    assert secret not in json.dumps(out)
+    assert secret not in we.as_prompt_section(out)
+
+
+def test_promotion_obeys_the_same_date_gate():
+    runs = [clean_run(BUILD, user=f"u{i}") for i in range(4)]
+    with pytest.raises(AggregationGateError):
+        we.promote(runs, today=EFFECTIVE_DATE - _dt.timedelta(days=1))
+    assert we.promote(runs, today=EFFECTIVE_DATE)["ok"] is True
+
+
+def test_the_prompt_section_is_empty_when_nothing_qualifies():
+    """An empty 'PATHS THAT WORK' heading reads as the engine having no idea
+    what works — worse than saying nothing."""
+    assert we.as_prompt_section(we.promote([], min_users=3)) == ""
+    assert we.as_prompt_section({}) == ""
+
+
+def test_the_prompt_section_frames_them_as_evidence_not_rules():
+    """Presented as law, an observed chain becomes a cage: the agent stops
+    solving tasks that need a different route."""
+    runs = [clean_run(BUILD, user=f"u{i}") for i in range(4)]
+    text = we.as_prompt_section(we.promote(runs, min_users=3))
+    assert "lookup_vector -> simulate_assembly" in text
+    assert "not rules" in text and "deviate" in text
+    assert "4 users" in text