avoigt1121
Claude Opus 5.5
feat: the five knowledge endpoints, gradio_ui and frozen descriptor (s2)
3ed6b66 | """Record the Phase 0 lookups as replay cassettes — the ONLY live-network step in this repo. | |
| Run by hand from the repo root, keyless, when fixtures are first made or after a reviewed | |
| ToolUniverse bump (ADR-0001 §7). It runs each scenario through the real endpoint code and the | |
| real, allow-listed ToolUniverse, and stores every raw response the endpoint received, keyed by | |
| tool and arguments, in ``tests/fixtures/cassettes/<scenario>.json``. The test suite replays | |
| those responses with sockets blocked, so what the tests check is exactly what the sources said | |
| on the recording day. Re-recording is a reviewed change: diff the cassettes before committing. | |
| Schema validation is switched off while recording (the schemas are written FROM these | |
| responses); replay validates every response against the pinned schemas. | |
| python scripts/record_fixtures.py # every scenario | |
| python scripts/record_fixtures.py q5_trials_cd40 | |
| """ | |
| from __future__ import annotations | |
| import datetime | |
| import json | |
| import sys | |
| import time | |
| from pathlib import Path | |
| ROOT = Path(__file__).resolve().parent.parent | |
| sys.path.insert(0, str(ROOT)) | |
| from src import config, schemas | |
| from src.cache import ResponseCache | |
| from src.endpoints import ENDPOINTS, _common | |
| from src.tu_client import ( | |
| ToolClient, | |
| build_tooluniverse, | |
| installed_tooluniverse_version, | |
| ) | |
| CASSETTES = ROOT / "tests" / "fixtures" / "cassettes" | |
| # The 55 pORG genes, as pdac-subtype-agent ships them (src/resources/signatures/pORG.tsv, | |
| # commit 2451c18) — the same list Phase 0's Q4 annotated. | |
| PORG_SOURCE = "pdac-subtype-agent src/resources/signatures/pORG.tsv @ 2451c18" | |
| PORG = [ | |
| "TNFSF18", | |
| "CST2", | |
| "RPRD1B", | |
| "BARD1", | |
| "SORD2P", | |
| "ZNF217", | |
| "CHMP4B", | |
| "SLFN13", | |
| "MMEL1", | |
| "SPATA24", | |
| "OSER1", | |
| "PHACTR3", | |
| "PKIB", | |
| "ARHGAP12", | |
| "RBM41", | |
| "MPZL3", | |
| "GDE1", | |
| "FGD6", | |
| "PLAAT2", | |
| "STAU1", | |
| "ZG16B", | |
| "RIOK3", | |
| "ALOX5", | |
| "CDS1", | |
| "PIK3CB", | |
| "HPGD", | |
| "HIST1H1T", | |
| "GID8", | |
| "NUDT21", | |
| "PAK1", | |
| "CYP4F3", | |
| "ARL17A", | |
| "PLAAT3", | |
| "SRP9", | |
| "ANKRD18B", | |
| "CCL20", | |
| "CFB", | |
| "SPOPL", | |
| "CTNNBL1", | |
| "NFE2L3", | |
| "ACSL5", | |
| "CIAO2A", | |
| "SCOC", | |
| "SAA4", | |
| "PPP1R3C", | |
| "WFDC13", | |
| "FUNDC1", | |
| "ADNP", | |
| "DLEU7", | |
| "NME7", | |
| "EFNB2", | |
| "LRRIQ1", | |
| "MAP3K2", | |
| "LNPK", | |
| "ZDHHC13", | |
| ] | |
| SCENARIOS = { | |
| "q1_q2_drugs_kras": ("drugs_for_targets", ["KRAS"], "Phase 0 Q1 (KRAS G12D) and Q2 (G12C)"), | |
| "q3_annotate_elavl1": ("annotate_genes", ["ELAVL1"], "Phase 0 Q3 (HuR function)"), | |
| "q4_annotate_porg": ("annotate_genes", [",".join(PORG)], "Phase 0 Q4 (the 55 pORG genes)"), | |
| "q5_trials_cd40": ("trials_for_genes", ["CD40", "", ""], "Phase 0 Q5 (recruiting CD40 trials)"), | |
| "q6_drugs_jak_stat": ("drugs_for_targets", ["JAK1,JAK2,STAT1"], "Phase 0 Q6 (JAK1/JAK2/STAT1)"), | |
| "q7_drugs_tnfrsf10b": ("drugs_for_targets", ["TNFRSF10B"], "Phase 0 Q7 (DR5)"), | |
| "q8_targets_pdac": ( | |
| "targets_for_disease", | |
| ["pancreatic ductal adenocarcinoma", ""], | |
| "Phase 0 Q8 (top PDAC targets)", | |
| ), | |
| "q9_pathways_isg15_rsad2_oas1": ( | |
| "pathways_for_genes", | |
| ["ISG15,RSAD2,OAS1"], | |
| "Phase 0 Q9 (Interferon alpha/beta signaling)", | |
| ), | |
| "c1_annotate_kras_tp53_sord2p": ( | |
| "annotate_genes", | |
| ["KRAS,TP53,SORD2P"], | |
| "plan C1(c): SORD2P listed as unresolved, not guessed", | |
| ), | |
| "edge_unknown_symbol": ("drugs_for_targets", ["NOTAGENE1"], "a symbol HGNC does not know"), | |
| "edge_legacy_efo": ( | |
| "targets_for_disease", | |
| ["EFO_0002618", "5"], | |
| "a legacy EFO id Open Targets dropped", | |
| ), | |
| } | |
| class Recorder: | |
| def __init__(self): | |
| self.tu = build_tooluniverse() | |
| self.calls: list[dict] = [] | |
| def __call__(self, tool: str, args: dict): | |
| response = self.tu.run_one_function({"name": tool, "arguments": args}, use_cache=False) | |
| self.calls.append({"tool": tool, "args": dict(args), "response": response}) | |
| return response | |
| def record(name: str) -> Path: | |
| endpoint, arguments, why = SCENARIOS[name] | |
| recorder = Recorder() | |
| client = ToolClient(runner=recorder, cache=ResponseCache(":memory:")) | |
| _common.set_client(client) | |
| started = time.time() | |
| payload = ENDPOINTS[endpoint][0](*arguments) | |
| used = sorted({config.source_of(c["tool"]) for c in recorder.calls}) | |
| cassette = { | |
| "_recorded": { | |
| "scenario": name, | |
| "why": why, | |
| "endpoint": endpoint, | |
| "arguments": arguments, | |
| "tooluniverse_version": installed_tooluniverse_version(), | |
| "recorded_on": datetime.datetime.now(datetime.UTC).strftime("%Y-%m-%dT%H:%M:%SZ"), | |
| "recorded_at_epoch": round(started, 3), | |
| "recorded_with": "scripts/record_fixtures.py", | |
| "payload_status_when_recorded": payload.get("status"), | |
| "licences": {s: config.source(s)["licence"] for s in used}, | |
| **({"porg_source": PORG_SOURCE} if name == "q4_annotate_porg" else {}), | |
| }, | |
| "calls": recorder.calls, | |
| } | |
| path = CASSETTES / f"{name}.json" | |
| path.write_text(json.dumps(cassette, indent=1, ensure_ascii=False) + "\n", encoding="utf-8") | |
| print( | |
| f"{name}: {len(recorder.calls)} calls, payload {payload.get('status')}, " | |
| f"{len(payload.get('rows') or [])} rows, {time.time() - started:.1f} s -> {path.name}" | |
| ) | |
| return path | |
| def _check_porg_against_the_subtype_agent() -> None: | |
| """The list above was once typed from memory and was wrong; compare it to the source file.""" | |
| relative = Path("pdac-subtype-agent/src/resources/signatures/pORG.tsv") | |
| for candidate in (ROOT.parent / relative, Path.home() / "Documents/GitHub" / relative): | |
| if candidate.is_file(): | |
| listed = candidate.read_text().split()[1:] | |
| assert listed == PORG, "PORG differs from pdac-subtype-agent's pORG.tsv" | |
| return | |
| raise SystemExit("pdac-subtype-agent is not checked out beside this repo; cannot check PORG") | |
| def main(names: list[str]) -> None: | |
| assert len(PORG) == 55, len(PORG) | |
| _check_porg_against_the_subtype_agent() | |
| CASSETTES.mkdir(parents=True, exist_ok=True) | |
| schemas.validate = lambda tool, data: None # recording only: schemas are written from this | |
| for name in names or list(SCENARIOS): | |
| record(name) | |
| if __name__ == "__main__": | |
| main(sys.argv[1:]) | |