From cba10f7b991214883e799fe86d70ed32ed4415ed Mon Sep 17 00:00:00 2001
From: mohsenht <mohsen.taheri.1989@gmail.com>
Date: Tue, 20 Jan 2026 10:16:04 -0500
Subject: [PATCH 01/32] Using Pathfinder package with local sqlite files

---
 compose.yml                              |  16 ++
 workers/arax/worker.py                   |  62 +++++---
 workers/arax_pathfinder/Dockerfile       |  34 +++++
 workers/arax_pathfinder/__init__.py      |   0
 workers/arax_pathfinder/requirements.txt |   2 +
 workers/arax_pathfinder/worker.py        | 178 +++++++++++++++++++++++
 6 files changed, 269 insertions(+), 23 deletions(-)
 create mode 100644 workers/arax_pathfinder/Dockerfile
 create mode 100644 workers/arax_pathfinder/__init__.py
 create mode 100644 workers/arax_pathfinder/requirements.txt
 create mode 100644 workers/arax_pathfinder/worker.py

diff --git a/compose.yml b/compose.yml
index 35d2b24..b2fdeae 100644
--- a/compose.yml
+++ b/compose.yml
@@ -273,6 +273,22 @@ services:
     volumes:
       - ./logs:/app/logs
       - ./.env:/app/.env
+  arax_pathfinder:
+    container_name: arax_pathfinder
+    build:
+      context: .
+      dockerfile: workers/arax_pathfinder/Dockerfile
+    restart: unless-stopped
+    depends_on:
+      shepherd_db:
+        condition: service_healthy
+      shepherd_broker:
+        condition: service_healthy
+    volumes:
+      - ./logs:/app/logs
+      - ./.env:/app/.env
+      - /Users/facadmin/PycharmProjects/shepherd/curie_ngd_v1.0_KG2.10.2.sqlite:/data/curie_ngd.sqlite:ro
+      - /Users/facadmin/PycharmProjects/shepherd/kg2c_v1.0_KG2.10.2.sqlite:/data/kg2c.sqlite:ro
 
 ######### BTE
   bte:
diff --git a/workers/arax/worker.py b/workers/arax/worker.py
index f22c686..53510c3 100644
--- a/workers/arax/worker.py
+++ b/workers/arax/worker.py
@@ -18,33 +18,49 @@
 tracer = setup_tracer(STREAM)
 
 
-async def arax(task, logger: logging.Logger):
+def is_pathfinder_query(message):
     try:
-        start = time.time()
-        query_id = task[1]["query_id"]
-        logger.info(f"Getting message from db for query id {query_id}")
-        message = await get_message(query_id, logger)
-        message["submitter"] = "Shepherd"
-        logger.info(f"Get the message from db {message}")
-
-        headers = {"Content-Type": "application/json"}
-        response = requests.post(settings.arax_url, json=message, headers=headers)
-
-        logger.info(f"Status Code from ARAX response: {response.status_code}")
-        result = response.json()
-
-    except Exception as e:
-        logger.error(f"Error occurred in ARAX entry module: {e}")
-        result = {"status": "error", "error": str(e)}
-
-    response_id = task[1]["response_id"]
+        # this can still fail if the input looks like e.g.:
+        #  "query_graph": None
+        qedges = message.get("message", {}).get("query_graph", {}).get("edges", {})
+    except:
+        qedges = {}
+    try:
+        # this can still fail if the input looks like e.g.:
+        #  "query_graph": None
+        qpaths = message.get("message", {}).get("query_graph", {}).get("paths", {})
+    except:
+        qpaths = {}
+    if len(qpaths) > 1:
+        raise Exception("Only a single path is supported", 400)
+    if (len(qpaths) > 0) and (len(qedges) > 0):
+        raise Exception("Mixed mode pathfinder queries are not supported", 400)
+    return len(qpaths) == 1
 
-    await save_message(response_id, result, logger)
 
-    workflow = [{"id": "arax"}]
+async def arax(task, logger: logging.Logger):
+    start = time.time()
+    query_id = task[1]["query_id"]
+    logger.info(f"Getting message from db for query id {query_id}")
+    message = await get_message(query_id, logger)
+    if is_pathfinder_query(message):
+        workflow = [{"id": "arax.pathfinder"}]
+    else:
+        try:
+            workflow = [{"id": "arax"}]
+            message["submitter"] = "Shepherd"
+            logger.info(f"Get the message from db {message}")
+            headers = {"Content-Type": "application/json"}
+            response = requests.post(settings.arax_url, json=message, headers=headers)
+            logger.info(f"Status Code from ARAX response: {response.status_code}")
+            result = response.json()
+        except Exception as e:
+            logger.error(f"Error occurred calling ARAX service: {e}")
+            result = {"status": "error", "error": str(e)}
+        response_id = task[1]["response_id"]
+        await save_message(response_id, result, logger)
 
     await wrap_up_task(STREAM, GROUP, task, workflow, logger)
-
     logger.info(f"Finished task {task[0]} in {time.time() - start}")
 
 
@@ -61,7 +77,7 @@ async def process_task(task, parent_ctx, logger, limiter):
 
 async def poll_for_tasks():
     async for task, parent_ctx, logger, limiter in get_tasks(
-        STREAM, GROUP, CONSUMER, TASK_LIMIT
+            STREAM, GROUP, CONSUMER, TASK_LIMIT
     ):
         asyncio.create_task(process_task(task, parent_ctx, logger, limiter))
 
diff --git a/workers/arax_pathfinder/Dockerfile b/workers/arax_pathfinder/Dockerfile
new file mode 100644
index 0000000..890204f
--- /dev/null
+++ b/workers/arax_pathfinder/Dockerfile
@@ -0,0 +1,34 @@
+# Use RENCI python base image
+FROM ghcr.io/translatorsri/renci-python-image:3.11.5
+
+# Add image info
+LABEL org.opencontainers.image.source https://github.com/BioPack-team/shepherd
+
+ENV PYTHONHASHSEED=0
+
+# set up requirements
+WORKDIR /app
+
+# make sure all is writeable for the nru USER later on
+RUN chmod -R 777 .
+
+# Install requirements
+COPY ./shepherd_utils ./shepherd_utils
+COPY ./pyproject.toml .
+RUN pip install .
+
+COPY ./workers/arax_pathfinder/requirements.txt .
+RUN pip install -r requirements.txt
+
+# switch to the non-root user (nru). defined in the base image
+USER nru
+
+# Copy in files
+COPY ./workers/arax_pathfinder ./
+
+# Set up base for command and any variables
+# that shouldn't be modified
+# ENTRYPOINT ["uvicorn", "shepherd_server.server:APP"]
+
+# Variables that can be overriden
+CMD ["python", "worker.py"]
diff --git a/workers/arax_pathfinder/__init__.py b/workers/arax_pathfinder/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/workers/arax_pathfinder/requirements.txt b/workers/arax_pathfinder/requirements.txt
new file mode 100644
index 0000000..0e22cce
--- /dev/null
+++ b/workers/arax_pathfinder/requirements.txt
@@ -0,0 +1,2 @@
+catrax-pathfinder==1.0.2
+biolink-helper-pkg==1.0.0
\ No newline at end of file
diff --git a/workers/arax_pathfinder/worker.py b/workers/arax_pathfinder/worker.py
new file mode 100644
index 0000000..b65a6f0
--- /dev/null
+++ b/workers/arax_pathfinder/worker.py
@@ -0,0 +1,178 @@
+"""Arax ARA Pathfinder module."""
+
+import requests
+import asyncio
+import json
+import logging
+import time
+import uuid
+from pathlib import Path
+from pathfinder.Pathfinder import Pathfinder
+from biolink_helper_pkg import BiolinkHelper
+
+from shepherd_utils.config import settings
+from shepherd_utils.db import (
+    get_message,
+    save_message,
+)
+from shepherd_utils.otel import setup_tracer
+from shepherd_utils.shared import (
+    get_tasks,
+    wrap_up_task,
+)
+
+# Queue name
+STREAM = "arax.pathfinder"
+# Consumer group, most likely you don't need to change this.
+GROUP = "consumer"
+CONSUMER = str(uuid.uuid4())[:8]
+TASK_LIMIT = 100
+tracer = setup_tracer(STREAM)
+
+NUM_TOTAL_HOPS = 4
+MAX_PATHFINDER_PATHS = 500
+BIOLINK_VERSION = "4.2.5"
+
+RAW_URL = (
+    "https://raw.githubusercontent.com/RTXteam/RTX/master/"
+    "code/ARAX/KnowledgeSources/general_concepts.json"
+)
+OUT_PATH = Path("general_concepts.json")
+
+def download_file(url: str, out_path: Path, overwrite: bool = False) -> Path:
+    out_path = Path(out_path)
+
+    if out_path.exists() and not overwrite:
+        return out_path
+
+    out_path.parent.mkdir(parents=True, exist_ok=True)
+
+    r = requests.get(url, timeout=60)
+    r.raise_for_status()
+
+    out_path.write_bytes(r.content)
+    return out_path
+
+
+def get_blocked_list():
+    download_file(RAW_URL, OUT_PATH, False)
+
+    with open(OUT_PATH, 'r') as file:
+        json_block_list = json.load(file)
+    synonyms = set(s.lower() for s in json_block_list['synonyms'])
+    return set(json_block_list['curies']), synonyms
+
+
+async def pathfinder(task, logger: logging.Logger):
+    start = time.time()
+    query_id = task[1]["query_id"]
+    workflow = json.loads(task[1]["workflow"])
+    response_id = task[1]["response_id"]
+    message = await get_message(query_id, logger)
+    parameters = message.get("parameters") or {}
+    parameters["timeout"] = parameters.get("timeout", settings.lookup_timeout)
+    parameters["tiers"] = parameters.get("tiers") or [0]
+    message["parameters"] = parameters
+
+    qgraph = message["message"]["query_graph"]
+    pinned_node_keys = []
+    pinned_node_ids = []
+    for node_key, node in qgraph["nodes"].items():
+        pinned_node_keys.append(node_key)
+        if node.get("ids", None) is not None:
+            pinned_node_ids.append(node["ids"][0])
+    if len(set(pinned_node_ids)) != 2:
+        logger.error("Pathfinder queries require two pinned nodes.")
+        return message, 500
+
+    intermediate_categories = []
+    path_key = next(iter(qgraph["paths"].keys()))
+    qpath = qgraph["paths"][path_key]
+    if qpath.get("constraints", None) is not None:
+        constraints = qpath["constraints"]
+        if len(constraints) > 1:
+            logger.error("Pathfinder queries do not support multiple constraints.")
+            return message, 500
+        if len(constraints) > 0:
+            intermediate_categories = (
+                    constraints[0].get("intermediate_categories", None) or []
+            )
+        if len(intermediate_categories) > 1:
+            logger.error(
+                "Pathfinder queries do not support multiple intermediate categories"
+            )
+            return message, 500
+    else:
+        intermediate_categories = ["biolink:NamedThing"]
+
+    blocked_curies, blocked_synonyms = get_blocked_list()
+    pathfinder = Pathfinder(
+        "MLRepo",
+        "https://kg2cploverdb.test.transltr.io",
+        "sqlite:/data/curie_ngd.sqlite",
+        "sqlite:/data/kg2c.sqlite",
+        blocked_curies,
+        blocked_synonyms,
+        logger
+    )
+
+    biolink_dir = "/tmp/biolink"
+    Path(biolink_dir).mkdir(parents=True, exist_ok=True)
+    biolink_helper = BiolinkHelper(BIOLINK_VERSION, biolink_dir)
+    descendants = set(biolink_helper.get_descendants(intermediate_categories[0]))
+
+    try:
+        result, aux_graphs, knowledge_graph = pathfinder.get_paths(
+            pinned_node_ids[0],
+            pinned_node_ids[1],
+            pinned_node_keys[0],
+            pinned_node_keys[1],
+            NUM_TOTAL_HOPS,
+            NUM_TOTAL_HOPS,
+            MAX_PATHFINDER_PATHS,
+            descendants,
+        )
+        res = []
+        if result is not None:
+            res.append({
+                "id": result["id"],
+                "analyses": result['analyses'],
+                "node_bindings": result['node_bindings'],
+                "essence": "result"
+            })
+        if aux_graphs is None:
+            aux_graphs = {}
+        if knowledge_graph is None:
+            knowledge_graph = {}
+        message["message"]["knowledge_graph"] = knowledge_graph
+        message["message"]["auxiliary_graphs"] = aux_graphs
+        message["message"]["results"] = res
+        await save_message(response_id, message, logger)
+    except Exception as e:
+        logger.error(f"PathFinder failed to find paths between {pinned_node_keys[0]} and {pinned_node_keys[1]}. "
+                     f"Error message is: {e}")
+        message = {"status": "error", "error": str(e)}
+        await save_message(response_id, message, logger)
+
+    await wrap_up_task(STREAM, GROUP, task, workflow, logger)
+    logger.info(f"Task took {time.time() - start}")
+
+
+async def process_task(task, parent_ctx, logger, limiter):
+    span = tracer.start_span(STREAM, context=parent_ctx)
+    try:
+        await pathfinder(task, logger)
+    finally:
+        span.end()
+        limiter.release()
+
+
+async def poll_for_tasks():
+    async for task, parent_ctx, logger, limiter in get_tasks(
+            STREAM, GROUP, CONSUMER, TASK_LIMIT
+    ):
+        asyncio.create_task(process_task(task, parent_ctx, logger, limiter))
+
+
+if __name__ == "__main__":
+    asyncio.run(poll_for_tasks())

From 5539e3dee524821304216f930e86b2f1b6bb220d Mon Sep 17 00:00:00 2001
From: mohsenht <mohsen.taheri.1989@gmail.com>
Date: Wed, 21 Jan 2026 11:08:46 -0500
Subject: [PATCH 02/32] Using Pathfinder package with mysql server

---
 compose.yml                       | 2 --
 workers/arax_pathfinder/worker.py | 4 ++--
 2 files changed, 2 insertions(+), 4 deletions(-)

diff --git a/compose.yml b/compose.yml
index b2fdeae..759a025 100644
--- a/compose.yml
+++ b/compose.yml
@@ -287,8 +287,6 @@ services:
     volumes:
       - ./logs:/app/logs
       - ./.env:/app/.env
-      - /Users/facadmin/PycharmProjects/shepherd/curie_ngd_v1.0_KG2.10.2.sqlite:/data/curie_ngd.sqlite:ro
-      - /Users/facadmin/PycharmProjects/shepherd/kg2c_v1.0_KG2.10.2.sqlite:/data/kg2c.sqlite:ro
 
 ######### BTE
   bte:
diff --git a/workers/arax_pathfinder/worker.py b/workers/arax_pathfinder/worker.py
index b65a6f0..e228854 100644
--- a/workers/arax_pathfinder/worker.py
+++ b/workers/arax_pathfinder/worker.py
@@ -109,8 +109,8 @@ async def pathfinder(task, logger: logging.Logger):
     pathfinder = Pathfinder(
         "MLRepo",
         "https://kg2cploverdb.test.transltr.io",
-        "sqlite:/data/curie_ngd.sqlite",
-        "sqlite:/data/kg2c.sqlite",
+        "mysql:arax-databases-mysql.rtx.ai:public_ro:curie_ngd_v1_0_kg2_10_2",
+        "mysql:arax-databases-mysql.rtx.ai:public_ro:kg2c_v1_0_kg2_10_2",
         blocked_curies,
         blocked_synonyms,
         logger

From cf94d6dd9ac6c9545fad3576999b2a22f4a1d14d Mon Sep 17 00:00:00 2001
From: mohsenht <mohsen.taheri.1989@gmail.com>
Date: Wed, 21 Jan 2026 11:16:26 -0500
Subject: [PATCH 03/32] Settings for arax pathfinder

---
 shepherd_utils/config.py          | 12 ++++++++++++
 workers/arax_pathfinder/worker.py | 21 +++++++++------------
 2 files changed, 21 insertions(+), 12 deletions(-)

diff --git a/shepherd_utils/config.py b/shepherd_utils/config.py
index d9a7ade..4529ed5 100644
--- a/shepherd_utils/config.py
+++ b/shepherd_utils/config.py
@@ -22,7 +22,19 @@ class Settings(BaseSettings):
     kg_retrieval_url: str = "https://strider.renci.org/asyncquery"
     sync_kg_retrieval_url: str = "https://strider.renci.org/query"
     omnicorp_url: str = "https://aragorn-ranker.renci.org/omnicorp_overlay"
+
+    # ARAX configs
     arax_url: str = "https://arax.ncats.io/shepherd/api/arax/v1.4/query"
+    plover_url: str = "https://kg2cploverdb.test.transltr.io"
+    curie_ngd_addr: str = "mysql:arax-databases-mysql.rtx.ai:public_ro:curie_ngd_v1_0_kg2_10_2"
+    node_degree_addr: str = "mysql:arax-databases-mysql.rtx.ai:public_ro:kg2c_v1_0_kg2_10_2"
+    arax_biolink_version: str = "4.2.5"
+    arax_blocked_list_url: str = (
+        "https://raw.githubusercontent.com/RTXteam/RTX/master/"
+        "code/ARAX/KnowledgeSources/general_concepts.json"
+    )
+    # End of ARAX configs
+
     node_norm: str = "https://biothings.ci.transltr.io/nodenorm/api/"
 
     pathfinder_redis_host: str = "host.docker.internal"
diff --git a/workers/arax_pathfinder/worker.py b/workers/arax_pathfinder/worker.py
index e228854..d44e655 100644
--- a/workers/arax_pathfinder/worker.py
+++ b/workers/arax_pathfinder/worker.py
@@ -31,12 +31,9 @@
 
 NUM_TOTAL_HOPS = 4
 MAX_PATHFINDER_PATHS = 500
-BIOLINK_VERSION = "4.2.5"
 
-RAW_URL = (
-    "https://raw.githubusercontent.com/RTXteam/RTX/master/"
-    "code/ARAX/KnowledgeSources/general_concepts.json"
-)
+
+
 OUT_PATH = Path("general_concepts.json")
 
 def download_file(url: str, out_path: Path, overwrite: bool = False) -> Path:
@@ -55,7 +52,7 @@ def download_file(url: str, out_path: Path, overwrite: bool = False) -> Path:
 
 
 def get_blocked_list():
-    download_file(RAW_URL, OUT_PATH, False)
+    download_file(settings.arax_blocked_list_url, OUT_PATH, False)
 
     with open(OUT_PATH, 'r') as file:
         json_block_list = json.load(file)
@@ -108,17 +105,17 @@ async def pathfinder(task, logger: logging.Logger):
     blocked_curies, blocked_synonyms = get_blocked_list()
     pathfinder = Pathfinder(
         "MLRepo",
-        "https://kg2cploverdb.test.transltr.io",
-        "mysql:arax-databases-mysql.rtx.ai:public_ro:curie_ngd_v1_0_kg2_10_2",
-        "mysql:arax-databases-mysql.rtx.ai:public_ro:kg2c_v1_0_kg2_10_2",
+        settings.plover_url,
+        settings.curie_ngd_addr,
+        settings.node_degree_addr,
         blocked_curies,
         blocked_synonyms,
         logger
     )
 
-    biolink_dir = "/tmp/biolink"
-    Path(biolink_dir).mkdir(parents=True, exist_ok=True)
-    biolink_helper = BiolinkHelper(BIOLINK_VERSION, biolink_dir)
+    biolink_cache_dir = "/tmp/biolink"
+    Path(biolink_cache_dir).mkdir(parents=True, exist_ok=True)
+    biolink_helper = BiolinkHelper(settings.arax_biolink_version, biolink_cache_dir)
     descendants = set(biolink_helper.get_descendants(intermediate_categories[0]))
 
     try:

From 4999dfff38fede49b42846bf1d8968b64fdbc627 Mon Sep 17 00:00:00 2001
From: mohsenht <mohsen.taheri.1989@gmail.com>
Date: Wed, 21 Jan 2026 13:55:49 -0500
Subject: [PATCH 04/32] Black style errors

---
 shepherd_utils/config.py          |  8 +++++--
 workers/arax/worker.py            |  2 +-
 workers/arax_pathfinder/worker.py | 35 ++++++++++++++++---------------
 3 files changed, 25 insertions(+), 20 deletions(-)

diff --git a/shepherd_utils/config.py b/shepherd_utils/config.py
index 4529ed5..341771e 100644
--- a/shepherd_utils/config.py
+++ b/shepherd_utils/config.py
@@ -26,8 +26,12 @@ class Settings(BaseSettings):
     # ARAX configs
     arax_url: str = "https://arax.ncats.io/shepherd/api/arax/v1.4/query"
     plover_url: str = "https://kg2cploverdb.test.transltr.io"
-    curie_ngd_addr: str = "mysql:arax-databases-mysql.rtx.ai:public_ro:curie_ngd_v1_0_kg2_10_2"
-    node_degree_addr: str = "mysql:arax-databases-mysql.rtx.ai:public_ro:kg2c_v1_0_kg2_10_2"
+    curie_ngd_addr: str = (
+        "mysql:arax-databases-mysql.rtx.ai:public_ro:curie_ngd_v1_0_kg2_10_2"
+    )
+    node_degree_addr: str = (
+        "mysql:arax-databases-mysql.rtx.ai:public_ro:kg2c_v1_0_kg2_10_2"
+    )
     arax_biolink_version: str = "4.2.5"
     arax_blocked_list_url: str = (
         "https://raw.githubusercontent.com/RTXteam/RTX/master/"
diff --git a/workers/arax/worker.py b/workers/arax/worker.py
index 53510c3..d1eacda 100644
--- a/workers/arax/worker.py
+++ b/workers/arax/worker.py
@@ -77,7 +77,7 @@ async def process_task(task, parent_ctx, logger, limiter):
 
 async def poll_for_tasks():
     async for task, parent_ctx, logger, limiter in get_tasks(
-            STREAM, GROUP, CONSUMER, TASK_LIMIT
+        STREAM, GROUP, CONSUMER, TASK_LIMIT
     ):
         asyncio.create_task(process_task(task, parent_ctx, logger, limiter))
 
diff --git a/workers/arax_pathfinder/worker.py b/workers/arax_pathfinder/worker.py
index d44e655..90c4d41 100644
--- a/workers/arax_pathfinder/worker.py
+++ b/workers/arax_pathfinder/worker.py
@@ -32,10 +32,9 @@
 NUM_TOTAL_HOPS = 4
 MAX_PATHFINDER_PATHS = 500
 
-
-
 OUT_PATH = Path("general_concepts.json")
 
+
 def download_file(url: str, out_path: Path, overwrite: bool = False) -> Path:
     out_path = Path(out_path)
 
@@ -54,10 +53,10 @@ def download_file(url: str, out_path: Path, overwrite: bool = False) -> Path:
 def get_blocked_list():
     download_file(settings.arax_blocked_list_url, OUT_PATH, False)
 
-    with open(OUT_PATH, 'r') as file:
+    with open(OUT_PATH, "r") as file:
         json_block_list = json.load(file)
-    synonyms = set(s.lower() for s in json_block_list['synonyms'])
-    return set(json_block_list['curies']), synonyms
+    synonyms = set(s.lower() for s in json_block_list["synonyms"])
+    return set(json_block_list["curies"]), synonyms
 
 
 async def pathfinder(task, logger: logging.Logger):
@@ -91,9 +90,7 @@ async def pathfinder(task, logger: logging.Logger):
             logger.error("Pathfinder queries do not support multiple constraints.")
             return message, 500
         if len(constraints) > 0:
-            intermediate_categories = (
-                    constraints[0].get("intermediate_categories", None) or []
-            )
+            intermediate_categories = (constraints[0].get("intermediate_categories", None) or [])
         if len(intermediate_categories) > 1:
             logger.error(
                 "Pathfinder queries do not support multiple intermediate categories"
@@ -110,7 +107,7 @@ async def pathfinder(task, logger: logging.Logger):
         settings.node_degree_addr,
         blocked_curies,
         blocked_synonyms,
-        logger
+        logger,
     )
 
     biolink_cache_dir = "/tmp/biolink"
@@ -131,12 +128,14 @@ async def pathfinder(task, logger: logging.Logger):
         )
         res = []
         if result is not None:
-            res.append({
-                "id": result["id"],
-                "analyses": result['analyses'],
-                "node_bindings": result['node_bindings'],
-                "essence": "result"
-            })
+            res.append(
+                {
+                    "id": result["id"],
+                    "analyses": result['analyses'],
+                    "node_bindings": result['node_bindings'],
+                    "essence": "result"
+                }
+            )
         if aux_graphs is None:
             aux_graphs = {}
         if knowledge_graph is None:
@@ -146,8 +145,10 @@ async def pathfinder(task, logger: logging.Logger):
         message["message"]["results"] = res
         await save_message(response_id, message, logger)
     except Exception as e:
-        logger.error(f"PathFinder failed to find paths between {pinned_node_keys[0]} and {pinned_node_keys[1]}. "
-                     f"Error message is: {e}")
+        logger.error(
+            f"PathFinder failed to find paths between {pinned_node_keys[0]} and {pinned_node_keys[1]}. "
+            f"Error message is: {e}"
+        )
         message = {"status": "error", "error": str(e)}
         await save_message(response_id, message, logger)
 

From 767e51b2896bfddd5dc4c5425edd82994c619209 Mon Sep 17 00:00:00 2001
From: mohsenht <mohsen.taheri.1989@gmail.com>
Date: Wed, 21 Jan 2026 13:59:22 -0500
Subject: [PATCH 05/32] Black style errors

---
 shepherd_server/main.py           |  1 -
 workers/arax_pathfinder/worker.py | 12 +++++++-----
 2 files changed, 7 insertions(+), 6 deletions(-)

diff --git a/shepherd_server/main.py b/shepherd_server/main.py
index 6e4aac3..7b39542 100644
--- a/shepherd_server/main.py
+++ b/shepherd_server/main.py
@@ -1,6 +1,5 @@
 import uvicorn
 
-
 if __name__ == "__main__":
     uvicorn.run(
         "shepherd_server.server:APP",
diff --git a/workers/arax_pathfinder/worker.py b/workers/arax_pathfinder/worker.py
index 90c4d41..997e324 100644
--- a/workers/arax_pathfinder/worker.py
+++ b/workers/arax_pathfinder/worker.py
@@ -90,7 +90,9 @@ async def pathfinder(task, logger: logging.Logger):
             logger.error("Pathfinder queries do not support multiple constraints.")
             return message, 500
         if len(constraints) > 0:
-            intermediate_categories = (constraints[0].get("intermediate_categories", None) or [])
+            intermediate_categories = (
+                constraints[0].get("intermediate_categories", None) or []
+            )
         if len(intermediate_categories) > 1:
             logger.error(
                 "Pathfinder queries do not support multiple intermediate categories"
@@ -131,9 +133,9 @@ async def pathfinder(task, logger: logging.Logger):
             res.append(
                 {
                     "id": result["id"],
-                    "analyses": result['analyses'],
-                    "node_bindings": result['node_bindings'],
-                    "essence": "result"
+                    "analyses": result["analyses"],
+                    "node_bindings": result["node_bindings"],
+                    "essence": "result",
                 }
             )
         if aux_graphs is None:
@@ -167,7 +169,7 @@ async def process_task(task, parent_ctx, logger, limiter):
 
 async def poll_for_tasks():
     async for task, parent_ctx, logger, limiter in get_tasks(
-            STREAM, GROUP, CONSUMER, TASK_LIMIT
+        STREAM, GROUP, CONSUMER, TASK_LIMIT
     ):
         asyncio.create_task(process_task(task, parent_ctx, logger, limiter))
 

From 540f220c95eaf80410498be3993e040ddf751a27 Mon Sep 17 00:00:00 2001
From: mohsenht <mohsen.taheri.1989@gmail.com>
Date: Fri, 23 Jan 2026 11:33:17 -0500
Subject: [PATCH 06/32] New pathfinder package release update.

---
 workers/arax_pathfinder/requirements.txt | 2 +-
 workers/arax_pathfinder/worker.py        | 7 ++++++-
 2 files changed, 7 insertions(+), 2 deletions(-)

diff --git a/workers/arax_pathfinder/requirements.txt b/workers/arax_pathfinder/requirements.txt
index 0e22cce..e39dad0 100644
--- a/workers/arax_pathfinder/requirements.txt
+++ b/workers/arax_pathfinder/requirements.txt
@@ -1,2 +1,2 @@
-catrax-pathfinder==1.0.2
+catrax-pathfinder==1.1.1
 biolink-helper-pkg==1.0.0
\ No newline at end of file
diff --git a/workers/arax_pathfinder/worker.py b/workers/arax_pathfinder/worker.py
index 997e324..e64920a 100644
--- a/workers/arax_pathfinder/worker.py
+++ b/workers/arax_pathfinder/worker.py
@@ -30,7 +30,10 @@
 tracer = setup_tracer(STREAM)
 
 NUM_TOTAL_HOPS = 4
+MAX_HOPS_TO_EXPLORE = 6
 MAX_PATHFINDER_PATHS = 500
+PRUNE_TOP_K = 30
+NODE_DEGREE_THRESHOLD = 30000
 
 OUT_PATH = Path("general_concepts.json")
 
@@ -124,8 +127,10 @@ async def pathfinder(task, logger: logging.Logger):
             pinned_node_keys[0],
             pinned_node_keys[1],
             NUM_TOTAL_HOPS,
-            NUM_TOTAL_HOPS,
+            MAX_HOPS_TO_EXPLORE,
             MAX_PATHFINDER_PATHS,
+            PRUNE_TOP_K,
+            NODE_DEGREE_THRESHOLD,
             descendants,
         )
         res = []

From ce1a5e6cf77d885ecf0cb13b50b2398cb4cfe6ac Mon Sep 17 00:00:00 2001
From: mohsenht <mohsen.taheri.1989@gmail.com>
Date: Wed, 4 Feb 2026 13:52:39 -0500
Subject: [PATCH 07/32] Temporary faster pathfinder by decreasing parameters

---
 workers/arax_pathfinder/worker.py | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/workers/arax_pathfinder/worker.py b/workers/arax_pathfinder/worker.py
index eb8cd68..5226e2e 100644
--- a/workers/arax_pathfinder/worker.py
+++ b/workers/arax_pathfinder/worker.py
@@ -30,10 +30,10 @@
 TASK_LIMIT = 100
 tracer = setup_tracer(STREAM)
 
-NUM_TOTAL_HOPS = 4
-MAX_HOPS_TO_EXPLORE = 6
+NUM_TOTAL_HOPS = 3
+MAX_HOPS_TO_EXPLORE = 3
 MAX_PATHFINDER_PATHS = 500
-PRUNE_TOP_K = 30
+PRUNE_TOP_K = 50
 NODE_DEGREE_THRESHOLD = 30000
 
 OUT_PATH = Path("general_concepts.json")

From 8ec57a9f3872c544bed51b9b3ab8f548ae9acd63 Mon Sep 17 00:00:00 2001
From: mohsenht <mohsen.taheri.1989@gmail.com>
Date: Wed, 11 Feb 2026 14:09:06 -0500
Subject: [PATCH 08/32] Arax Pathfinder tested with 4 hops

---
 shepherd_utils/config.py          |  2 +-
 workers/arax_pathfinder/worker.py | 10 +++++++---
 2 files changed, 8 insertions(+), 4 deletions(-)

diff --git a/shepherd_utils/config.py b/shepherd_utils/config.py
index 845c99f..d98555c 100644
--- a/shepherd_utils/config.py
+++ b/shepherd_utils/config.py
@@ -26,7 +26,7 @@ class Settings(BaseSettings):
 
     # ARAX configs
     arax_url: str = "https://arax.ncats.io/shepherd/api/arax/v1.4/query"
-    plover_url: str = "https://kg2cploverdb.test.transltr.io"
+    plover_url: str = "https://kg2cplover3.rtx.ai:9990"
     curie_ngd_addr: str = (
         "mysql:arax-databases-mysql.rtx.ai:public_ro:curie_ngd_v1_0_kg2_10_2"
     )
diff --git a/workers/arax_pathfinder/worker.py b/workers/arax_pathfinder/worker.py
index 5226e2e..2a16933 100644
--- a/workers/arax_pathfinder/worker.py
+++ b/workers/arax_pathfinder/worker.py
@@ -30,8 +30,8 @@
 TASK_LIMIT = 100
 tracer = setup_tracer(STREAM)
 
-NUM_TOTAL_HOPS = 3
-MAX_HOPS_TO_EXPLORE = 3
+NUM_TOTAL_HOPS = 4
+MAX_HOPS_TO_EXPLORE = 4
 MAX_PATHFINDER_PATHS = 500
 PRUNE_TOP_K = 50
 NODE_DEGREE_THRESHOLD = 30000
@@ -88,7 +88,7 @@ async def pathfinder(task, logger: logging.Logger):
     intermediate_categories = []
     path_key = next(iter(qgraph["paths"].keys()))
     qpath = qgraph["paths"][path_key]
-    if qpath.get("constraints", None) is not None:
+    if qpath.get("constraints", None) is not None and len(qpath.get("constraints", [])) > 0:
         constraints = qpath["constraints"]
         if len(constraints) > 1:
             logger.error("Pathfinder queries do not support multiple constraints.")
@@ -122,6 +122,8 @@ async def pathfinder(task, logger: logging.Logger):
     descendants = set(biolink_helper.get_descendants(intermediate_categories[0]))
 
     try:
+        start = time.perf_counter()
+        logger.info("Starting pathfinder.get_paths()")
         result, aux_graphs, knowledge_graph = pathfinder.get_paths(
             pinned_node_ids[0],
             pinned_node_ids[1],
@@ -134,6 +136,8 @@ async def pathfinder(task, logger: logging.Logger):
             NODE_DEGREE_THRESHOLD,
             descendants,
         )
+        elapsed = time.perf_counter() - start
+        logger.info(f"pathfinder.get_paths() finished in {elapsed:.3f} seconds")
         res = []
         if result is not None:
             res.append(

From 82852c474654fb4ef53d63844c8716234bdbbfdb Mon Sep 17 00:00:00 2001
From: mohsenht <mohsen.taheri.1989@gmail.com>
Date: Wed, 18 Feb 2026 11:48:34 -0500
Subject: [PATCH 09/32] Async Arax Pathfinder

---
 workers/arax_pathfinder/worker.py | 79 +++++++++++++++++++------------
 1 file changed, 48 insertions(+), 31 deletions(-)

diff --git a/workers/arax_pathfinder/worker.py b/workers/arax_pathfinder/worker.py
index 2a16933..595fa37 100644
--- a/workers/arax_pathfinder/worker.py
+++ b/workers/arax_pathfinder/worker.py
@@ -63,6 +63,47 @@ def get_blocked_list():
     return set(json_block_list["curies"]), synonyms
 
 
+def execute_pathfinding_sync(pinned_node_ids, pinned_node_keys, intermediate_categories, logger):
+
+    blocked_curies, blocked_synonyms = get_blocked_list()
+
+    pathfinder_instance = Pathfinder(
+        "MLRepo",
+        settings.plover_url,
+        settings.curie_ngd_addr,
+        settings.node_degree_addr,
+        blocked_curies,
+        blocked_synonyms,
+        logger,
+    )
+
+    biolink_cache_dir = "/tmp/biolink"
+    Path(biolink_cache_dir).mkdir(parents=True, exist_ok=True)
+    biolink_helper = BiolinkHelper(settings.arax_biolink_version, biolink_cache_dir)
+    descendants = set(biolink_helper.get_descendants(intermediate_categories[0]))
+
+    start = time.perf_counter()
+    logger.info("Starting pathfinder.get_paths() in worker thread")
+
+    result, aux_graphs, knowledge_graph = pathfinder_instance.get_paths(
+        pinned_node_ids[0],
+        pinned_node_ids[1],
+        pinned_node_keys[0],
+        pinned_node_keys[1],
+        NUM_TOTAL_HOPS,
+        MAX_HOPS_TO_EXPLORE,
+        MAX_PATHFINDER_PATHS,
+        PRUNE_TOP_K,
+        NODE_DEGREE_THRESHOLD,
+        descendants,
+    )
+
+    elapsed = time.perf_counter() - start
+    logger.info(f"pathfinder.get_paths() finished in {elapsed:.3f} seconds")
+
+    return result, aux_graphs, knowledge_graph
+
+
 async def pathfinder(task, logger: logging.Logger):
     start = time.time()
     query_id = task[1]["query_id"]
@@ -105,39 +146,15 @@ async def pathfinder(task, logger: logging.Logger):
     else:
         intermediate_categories = ["biolink:NamedThing"]
 
-    blocked_curies, blocked_synonyms = get_blocked_list()
-    pathfinder = Pathfinder(
-        "MLRepo",
-        settings.plover_url,
-        settings.curie_ngd_addr,
-        settings.node_degree_addr,
-        blocked_curies,
-        blocked_synonyms,
-        logger,
-    )
-
-    biolink_cache_dir = "/tmp/biolink"
-    Path(biolink_cache_dir).mkdir(parents=True, exist_ok=True)
-    biolink_helper = BiolinkHelper(settings.arax_biolink_version, biolink_cache_dir)
-    descendants = set(biolink_helper.get_descendants(intermediate_categories[0]))
-
     try:
-        start = time.perf_counter()
-        logger.info("Starting pathfinder.get_paths()")
-        result, aux_graphs, knowledge_graph = pathfinder.get_paths(
-            pinned_node_ids[0],
-            pinned_node_ids[1],
-            pinned_node_keys[0],
-            pinned_node_keys[1],
-            NUM_TOTAL_HOPS,
-            MAX_HOPS_TO_EXPLORE,
-            MAX_PATHFINDER_PATHS,
-            PRUNE_TOP_K,
-            NODE_DEGREE_THRESHOLD,
-            descendants,
+        result, aux_graphs, knowledge_graph = await asyncio.to_thread(
+            execute_pathfinding_sync,
+            pinned_node_ids,
+            pinned_node_keys,
+            intermediate_categories,
+            logger
         )
-        elapsed = time.perf_counter() - start
-        logger.info(f"pathfinder.get_paths() finished in {elapsed:.3f} seconds")
+
         res = []
         if result is not None:
             res.append(

From 6b537e51e4c6728598fd4324c61c0b816a4fcb48 Mon Sep 17 00:00:00 2001
From: mohsenht <mohsen.taheri.1989@gmail.com>
Date: Mon, 2 Mar 2026 22:47:23 -0500
Subject: [PATCH 10/32] Pathfinder package updated

---
 workers/arax_pathfinder/requirements.txt | 2 +-
 workers/arax_pathfinder/worker.py        | 4 ++--
 2 files changed, 3 insertions(+), 3 deletions(-)

diff --git a/workers/arax_pathfinder/requirements.txt b/workers/arax_pathfinder/requirements.txt
index e39dad0..bcb1c83 100644
--- a/workers/arax_pathfinder/requirements.txt
+++ b/workers/arax_pathfinder/requirements.txt
@@ -1,2 +1,2 @@
-catrax-pathfinder==1.1.1
+catrax-pathfinder==1.2.1
 biolink-helper-pkg==1.0.0
\ No newline at end of file
diff --git a/workers/arax_pathfinder/worker.py b/workers/arax_pathfinder/worker.py
index 595fa37..c98ae82 100644
--- a/workers/arax_pathfinder/worker.py
+++ b/workers/arax_pathfinder/worker.py
@@ -33,8 +33,8 @@
 NUM_TOTAL_HOPS = 4
 MAX_HOPS_TO_EXPLORE = 4
 MAX_PATHFINDER_PATHS = 500
-PRUNE_TOP_K = 50
-NODE_DEGREE_THRESHOLD = 30000
+PRUNE_TOP_K = 200
+NODE_DEGREE_THRESHOLD = 1000000
 
 OUT_PATH = Path("general_concepts.json")
 

From 991930674dec87606c18d71ec75e3eb9a4116f96 Mon Sep 17 00:00:00 2001
From: mohsenht <mohsen.taheri.1989@gmail.com>
Date: Wed, 4 Mar 2026 09:41:54 -0500
Subject: [PATCH 11/32] Pathfinder package updated

---
 workers/arax_pathfinder/requirements.txt | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/workers/arax_pathfinder/requirements.txt b/workers/arax_pathfinder/requirements.txt
index bcb1c83..5f5123c 100644
--- a/workers/arax_pathfinder/requirements.txt
+++ b/workers/arax_pathfinder/requirements.txt
@@ -1,2 +1,2 @@
-catrax-pathfinder==1.2.1
+catrax-pathfinder==1.2.2
 biolink-helper-pkg==1.0.0
\ No newline at end of file

From 4771f9061964c437f9e30e9763669d603dfe5533 Mon Sep 17 00:00:00 2001
From: mohsenht <mohsen.taheri.1989@gmail.com>
Date: Tue, 10 Mar 2026 20:47:08 -0400
Subject: [PATCH 12/32] resolved conflicts

---
 workers/arax/worker.py | 61 +++++++++++++++++++++++++++---------------
 1 file changed, 40 insertions(+), 21 deletions(-)

diff --git a/workers/arax/worker.py b/workers/arax/worker.py
index 1cefad4..eebfb77 100644
--- a/workers/arax/worker.py
+++ b/workers/arax/worker.py
@@ -1,7 +1,6 @@
 """ARAX entry module."""
 
 import asyncio
-import json
 import logging
 import requests
 import time
@@ -20,30 +19,50 @@
 tracer = setup_tracer(STREAM)
 
 
-async def arax(task, logger: logging.Logger):
+def is_pathfinder_query(message):
     try:
-        query_id = task[1]["query_id"]
-        logger.info(f"Getting message from db for query id {query_id}")
-        message = await get_message(query_id, logger)
-        message["submitter"] = "Shepherd"
-        logger.info(f"Get the message from db {message}")
-
-        headers = {"Content-Type": "application/json"}
-        response = requests.post(settings.arax_url, json=message, headers=headers)
-
-        logger.info(f"Status Code from ARAX response: {response.status_code}")
-        result = response.json()
-        result = add_shepherd_arax_to_edge_sources(result)
-
-    except Exception as e:
-        logger.error(f"Error occurred in ARAX entry module: {e}")
-        result = {"status": "error", "error": str(e)}
+        # this can still fail if the input looks like e.g.:
+        #  "query_graph": None
+        qedges = message.get("message", {}).get("query_graph", {}).get("edges", {})
+    except:
+        qedges = {}
+    try:
+        # this can still fail if the input looks like e.g.:
+        #  "query_graph": None
+        qpaths = message.get("message", {}).get("query_graph", {}).get("paths", {})
+    except:
+        qpaths = {}
+    if len(qpaths) > 1:
+        raise Exception("Only a single path is supported", 400)
+    if (len(qpaths) > 0) and (len(qedges) > 0):
+        raise Exception("Mixed mode pathfinder queries are not supported", 400)
+    return len(qpaths) == 1
 
-    response_id = task[1]["response_id"]
 
-    await save_message(response_id, result, logger)
+async def arax(task, logger: logging.Logger):
+    start = time.time()
+    query_id = task[1]["query_id"]
+    logger.info(f"Getting message from db for query id {query_id}")
+    message = await get_message(query_id, logger)
+    if is_pathfinder_query(message):
+        workflow = [{"id": "arax.pathfinder"}]
+    else:
+        try:
+            workflow = [{"id": "arax"}]
+            message["submitter"] = "Shepherd"
+            logger.info(f"Get the message from db {message}")
+            headers = {"Content-Type": "application/json"}
+            response = requests.post(settings.arax_url, json=message, headers=headers)
+            logger.info(f"Status Code from ARAX response: {response.status_code}")
+            result = response.json()
+            result = add_shepherd_arax_to_edge_sources(result)
+        except Exception as e:
+            logger.error(f"Error occurred calling ARAX service: {e}")
+            result = {"status": "error", "error": str(e)}
+        response_id = task[1]["response_id"]
+        await save_message(response_id, result, logger)
 
-    task[1]["workflow"] = json.dumps([{"id": "arax"}])
+    await wrap_up_task(STREAM, GROUP, task, workflow, logger)
     logger.info(f"Finished task {task[0]} in {time.time() - start}")
 
 

From 107614b1ecad9eb43c6d55b461d90a8510a90ef8 Mon Sep 17 00:00:00 2001
From: mohsenht <mohsen.taheri.1989@gmail.com>
Date: Tue, 10 Mar 2026 20:55:02 -0400
Subject: [PATCH 13/32] resolved conflicts

---
 workers/arax/worker.py | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

diff --git a/workers/arax/worker.py b/workers/arax/worker.py
index eebfb77..27b3f0b 100644
--- a/workers/arax/worker.py
+++ b/workers/arax/worker.py
@@ -1,6 +1,7 @@
 """ARAX entry module."""
 
 import asyncio
+import json
 import logging
 import requests
 import time
@@ -46,6 +47,7 @@ async def arax(task, logger: logging.Logger):
     message = await get_message(query_id, logger)
     if is_pathfinder_query(message):
         workflow = [{"id": "arax.pathfinder"}]
+        await wrap_up_task(STREAM, GROUP, task, workflow, logger)
     else:
         try:
             workflow = [{"id": "arax"}]
@@ -61,8 +63,9 @@ async def arax(task, logger: logging.Logger):
             result = {"status": "error", "error": str(e)}
         response_id = task[1]["response_id"]
         await save_message(response_id, result, logger)
+        task[1]["workflow"] = json.dumps([{"id": "arax"}])
+
 
-    await wrap_up_task(STREAM, GROUP, task, workflow, logger)
     logger.info(f"Finished task {task[0]} in {time.time() - start}")
 
 

From 268d2267e8c7362ead92196ce1eafeaeeb692f22 Mon Sep 17 00:00:00 2001
From: mohsenht <mohsen.taheri.1989@gmail.com>
Date: Tue, 10 Mar 2026 20:56:37 -0400
Subject: [PATCH 14/32] resolved conflicts

---
 workers/arax/worker.py | 2 --
 1 file changed, 2 deletions(-)

diff --git a/workers/arax/worker.py b/workers/arax/worker.py
index 27b3f0b..ec356e6 100644
--- a/workers/arax/worker.py
+++ b/workers/arax/worker.py
@@ -50,7 +50,6 @@ async def arax(task, logger: logging.Logger):
         await wrap_up_task(STREAM, GROUP, task, workflow, logger)
     else:
         try:
-            workflow = [{"id": "arax"}]
             message["submitter"] = "Shepherd"
             logger.info(f"Get the message from db {message}")
             headers = {"Content-Type": "application/json"}
@@ -65,7 +64,6 @@ async def arax(task, logger: logging.Logger):
         await save_message(response_id, result, logger)
         task[1]["workflow"] = json.dumps([{"id": "arax"}])
 
-
     logger.info(f"Finished task {task[0]} in {time.time() - start}")
 
 

From c6b79c382f6eb96cb00cae8228e13e2e93ccdab7 Mon Sep 17 00:00:00 2001
From: Max Wang <max@covar.com>
Date: Wed, 11 Mar 2026 12:24:07 -0400
Subject: [PATCH 15/32] Update to latest main code

---
 workers/arax/worker.py            | 13 ++++++----
 workers/arax_pathfinder/worker.py | 43 ++++++++++++++++++++++++-------
 2 files changed, 41 insertions(+), 15 deletions(-)

diff --git a/workers/arax/worker.py b/workers/arax/worker.py
index ec356e6..1ff858d 100644
--- a/workers/arax/worker.py
+++ b/workers/arax/worker.py
@@ -3,14 +3,18 @@
 import asyncio
 import json
 import logging
-import requests
 import time
 import uuid
+
+import requests
+
 from shepherd_utils.config import settings
 from shepherd_utils.db import get_message, save_message
-from shepherd_utils.shared import get_tasks, handle_task_failure, wrap_up_task
+from shepherd_utils.inject_shepherd_arax_provenance import (
+    add_shepherd_arax_to_edge_sources,
+)
 from shepherd_utils.otel import setup_tracer
-from inject_shepherd_arax_provenance import add_shepherd_arax_to_edge_sources
+from shepherd_utils.shared import get_tasks, handle_task_failure, wrap_up_task
 
 # Queue name
 STREAM = "arax"
@@ -46,8 +50,7 @@ async def arax(task, logger: logging.Logger):
     logger.info(f"Getting message from db for query id {query_id}")
     message = await get_message(query_id, logger)
     if is_pathfinder_query(message):
-        workflow = [{"id": "arax.pathfinder"}]
-        await wrap_up_task(STREAM, GROUP, task, workflow, logger)
+        task[1]["workflow"] = json.dumps([{"id": "arax.pathfinder"}])
     else:
         try:
             message["submitter"] = "Shepherd"
diff --git a/workers/arax_pathfinder/worker.py b/workers/arax_pathfinder/worker.py
index c98ae82..a40b5c2 100644
--- a/workers/arax_pathfinder/worker.py
+++ b/workers/arax_pathfinder/worker.py
@@ -1,24 +1,28 @@
 """Arax ARA Pathfinder module."""
 
-import requests
 import asyncio
 import json
 import logging
 import time
 import uuid
 from pathlib import Path
-from pathfinder.Pathfinder import Pathfinder
+
+import requests
 from biolink_helper_pkg import BiolinkHelper
+from pathfinder.Pathfinder import Pathfinder
 
-from shepherd_utils.inject_shepherd_arax_provenance import add_shepherd_arax_to_edge_sources
 from shepherd_utils.config import settings
 from shepherd_utils.db import (
     get_message,
     save_message,
 )
+from shepherd_utils.inject_shepherd_arax_provenance import (
+    add_shepherd_arax_to_edge_sources,
+)
 from shepherd_utils.otel import setup_tracer
 from shepherd_utils.shared import (
     get_tasks,
+    handle_task_failure,
     wrap_up_task,
 )
 
@@ -107,7 +111,6 @@ def execute_pathfinding_sync(pinned_node_ids, pinned_node_keys, intermediate_cat
 async def pathfinder(task, logger: logging.Logger):
     start = time.time()
     query_id = task[1]["query_id"]
-    workflow = json.loads(task[1]["workflow"])
     response_id = task[1]["response_id"]
     message = await get_message(query_id, logger)
     parameters = message.get("parameters") or {}
@@ -184,24 +187,44 @@ async def pathfinder(task, logger: logging.Logger):
         message = {"status": "error", "error": str(e)}
         await save_message(response_id, message, logger)
 
-    await wrap_up_task(STREAM, GROUP, task, workflow, logger)
     logger.info(f"Task took {time.time() - start}")
 
 
-async def process_task(task, parent_ctx, logger, limiter):
+async def process_task(task, parent_ctx, logger: logging.Logger, limiter):
+    """Process a given task and ACK in redis."""
+    start = time.time()
     span = tracer.start_span(STREAM, context=parent_ctx)
     try:
         await pathfinder(task, logger)
+        # Always wrap up the task to ACK it in the broker
+        try:
+            await wrap_up_task(STREAM, GROUP, task, logger)
+        except Exception as e:
+            logger.error(f"Task {task[0]}: Failed to wrap up task: {e}")
+    except asyncio.CancelledError:
+        logger.warning(f"Task {task[0]} was cancelled")
+    except Exception as e:
+        logger.error(f"Task {task[0]} failed with unhandled error: {e}", exc_info=True)
+        await handle_task_failure(STREAM, GROUP, task, logger)
     finally:
         span.end()
         limiter.release()
+        logger.info(f"Finished task {task[0]} in {time.time() - start}")
 
 
 async def poll_for_tasks():
-    async for task, parent_ctx, logger, limiter in get_tasks(
-        STREAM, GROUP, CONSUMER, TASK_LIMIT
-    ):
-        asyncio.create_task(process_task(task, parent_ctx, logger, limiter))
+    """On initialization, poll indefinitely for available tasks."""
+    while True:
+        try:
+            async for task, parent_ctx, logger, limiter in get_tasks(
+                STREAM, GROUP, CONSUMER, TASK_LIMIT
+            ):
+                asyncio.create_task(process_task(task, parent_ctx, logger, limiter))
+        except asyncio.CancelledError:
+            logging.info("Poll loop cancelled, shutting down.")
+        except Exception as e:
+            logging.error(f"Error in task polling loop: {e}", exc_info=True)
+            await asyncio.sleep(5)  # back off before retrying
 
 
 if __name__ == "__main__":

From f98c6e9730cf41aaebe56b2e8a1dce4c6b1e8b49 Mon Sep 17 00:00:00 2001
From: Max Wang <max@covar.com>
Date: Wed, 11 Mar 2026 12:25:07 -0400
Subject: [PATCH 16/32] Run black

---
 workers/arax_pathfinder/worker.py | 11 ++++++++---
 1 file changed, 8 insertions(+), 3 deletions(-)

diff --git a/workers/arax_pathfinder/worker.py b/workers/arax_pathfinder/worker.py
index a40b5c2..fe6867a 100644
--- a/workers/arax_pathfinder/worker.py
+++ b/workers/arax_pathfinder/worker.py
@@ -67,7 +67,9 @@ def get_blocked_list():
     return set(json_block_list["curies"]), synonyms
 
 
-def execute_pathfinding_sync(pinned_node_ids, pinned_node_keys, intermediate_categories, logger):
+def execute_pathfinding_sync(
+    pinned_node_ids, pinned_node_keys, intermediate_categories, logger
+):
 
     blocked_curies, blocked_synonyms = get_blocked_list()
 
@@ -132,7 +134,10 @@ async def pathfinder(task, logger: logging.Logger):
     intermediate_categories = []
     path_key = next(iter(qgraph["paths"].keys()))
     qpath = qgraph["paths"][path_key]
-    if qpath.get("constraints", None) is not None and len(qpath.get("constraints", [])) > 0:
+    if (
+        qpath.get("constraints", None) is not None
+        and len(qpath.get("constraints", [])) > 0
+    ):
         constraints = qpath["constraints"]
         if len(constraints) > 1:
             logger.error("Pathfinder queries do not support multiple constraints.")
@@ -155,7 +160,7 @@ async def pathfinder(task, logger: logging.Logger):
             pinned_node_ids,
             pinned_node_keys,
             intermediate_categories,
-            logger
+            logger,
         )
 
         res = []

From 298aee577efbefe50153c49d81915d8634b4837e Mon Sep 17 00:00:00 2001
From: mohsenht <mohsen.taheri.1989@gmail.com>
Date: Wed, 11 Mar 2026 13:24:11 -0400
Subject: [PATCH 17/32] PloverDB url updated to point to CI

---
 shepherd_utils/config.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/shepherd_utils/config.py b/shepherd_utils/config.py
index d98555c..1178cf5 100644
--- a/shepherd_utils/config.py
+++ b/shepherd_utils/config.py
@@ -26,7 +26,7 @@ class Settings(BaseSettings):
 
     # ARAX configs
     arax_url: str = "https://arax.ncats.io/shepherd/api/arax/v1.4/query"
-    plover_url: str = "https://kg2cplover3.rtx.ai:9990"
+    plover_url: str = "https://kg2cploverdb.ci.transltr.io"
     curie_ngd_addr: str = (
         "mysql:arax-databases-mysql.rtx.ai:public_ro:curie_ngd_v1_0_kg2_10_2"
     )

From 70758cbd6e0643009d1f1ba88840d645d7fd17c7 Mon Sep 17 00:00:00 2001
From: mohsenht <mohsen.taheri.1989@gmail.com>
Date: Wed, 11 Mar 2026 14:02:52 -0400
Subject: [PATCH 18/32] PRUNE more

---
 workers/arax_pathfinder/worker.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/workers/arax_pathfinder/worker.py b/workers/arax_pathfinder/worker.py
index fe6867a..2c1014d 100644
--- a/workers/arax_pathfinder/worker.py
+++ b/workers/arax_pathfinder/worker.py
@@ -37,7 +37,7 @@
 NUM_TOTAL_HOPS = 4
 MAX_HOPS_TO_EXPLORE = 4
 MAX_PATHFINDER_PATHS = 500
-PRUNE_TOP_K = 200
+PRUNE_TOP_K = 100
 NODE_DEGREE_THRESHOLD = 1000000
 
 OUT_PATH = Path("general_concepts.json")

From a7ab6bef3d1f11e8bc00208eb93ebad6161cbbec Mon Sep 17 00:00:00 2001
From: mohsenht <mohsen.taheri.1989@gmail.com>
Date: Thu, 16 Jul 2026 11:37:38 -0400
Subject: [PATCH 19/32] Async call

---
 workers/arax/worker.py | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/workers/arax/worker.py b/workers/arax/worker.py
index d81f826..e03c076 100644
--- a/workers/arax/worker.py
+++ b/workers/arax/worker.py
@@ -4,7 +4,6 @@
 import json
 import logging
 import uuid
-
 import httpx
 from inject_shepherd_arax_provenance import add_shepherd_arax_to_edge_sources
 
@@ -42,7 +41,6 @@ def is_pathfinder_query(message):
 
 
 async def arax(task, logger: logging.Logger):
-    start = time.time()
     query_id = task[1]["query_id"]
     logger.info(f"Getting message from db for query id {query_id}")
     message = await get_message(query_id, logger)
@@ -53,8 +51,10 @@ async def arax(task, logger: logging.Logger):
             message["submitter"] = "Shepherd"
             logger.info(f"Get the message from db {message}")
             headers = {"Content-Type": "application/json"}
-            with httpx.Client(timeout=270) as client:
-                response = client.post(settings.arax_url, json=message, headers=headers)
+            async with httpx.AsyncClient(timeout=270) as client:
+                response = await client.post(
+                    settings.arax_url, json=message, headers=headers
+                )
             logger.info(f"Status Code from ARAX response: {response.status_code}")
             result = response.json()
             result = add_shepherd_arax_to_edge_sources(result)

From a3eb40d2e214207517c3975c952344b79616db8d Mon Sep 17 00:00:00 2001
From: mohsenht <mohsen.taheri.1989@gmail.com>
Date: Thu, 23 Jul 2026 12:43:07 -0400
Subject: [PATCH 20/32] ARAX Pathfinder Package 2.4.3. Adaptable with Retriever

---
 compose.yml                              |  1 +
 shepherd_utils/config.py                 |  8 +---
 workers/arax/worker.py                   |  4 +-
 workers/arax_pathfinder/Dockerfile       |  2 +-
 workers/arax_pathfinder/requirements.txt |  5 ++-
 workers/arax_pathfinder/worker.py        | 51 ++++++++++++++++++++++--
 6 files changed, 57 insertions(+), 14 deletions(-)

diff --git a/compose.yml b/compose.yml
index 6a11cb0..9fd464d 100644
--- a/compose.yml
+++ b/compose.yml
@@ -352,6 +352,7 @@ services:
     volumes:
       - ./logs:/app/logs
       - ./.env:/app/.env
+      - ./arax_pathfinder_dbs:/app/arax_pathfinder_dbs
 
   arax_rank:
     container_name: arax_rank
diff --git a/shepherd_utils/config.py b/shepherd_utils/config.py
index 336d997..79a53ce 100644
--- a/shepherd_utils/config.py
+++ b/shepherd_utils/config.py
@@ -80,12 +80,8 @@ class Settings(BaseSettings):
     # ARAX configs
     arax_url: str = "https://arax.ncats.io/shepherd/api/arax/v1.4/query"
     plover_url: str = "https://kg2cploverdb.ci.transltr.io"
-    curie_ngd_addr: str = (
-        "mysql:arax-databases-mysql.rtx.ai:public_ro:curie_ngd_v1_0_kg2_10_2"
-    )
-    node_degree_addr: str = (
-        "mysql:arax-databases-mysql.rtx.ai:public_ro:kg2c_v1_0_kg2_10_2"
-    )
+    curie_ngd_addr: str = "sqlite:/app/arax_pathfinder_dbs/curie_ngd_v1.0_tier0-20260621.sqlite"
+    node_degree_addr: str = "sqlite:/app/arax_pathfinder_dbs/tier0-info-for-overlay_v1.0_tier0-20260621.sqlite"
     arax_biolink_version: str = "4.2.5"
     arax_blocked_list_url: str = (
         "https://raw.githubusercontent.com/RTXteam/RTX/master/"
diff --git a/workers/arax/worker.py b/workers/arax/worker.py
index 902e40e..84ab9e4 100644
--- a/workers/arax/worker.py
+++ b/workers/arax/worker.py
@@ -5,7 +5,9 @@
 import logging
 import uuid
 import httpx
-from inject_shepherd_arax_provenance import add_shepherd_arax_to_edge_sources
+from shepherd_utils.inject_shepherd_arax_provenance import (
+    add_shepherd_arax_to_edge_sources,
+)
 
 from shepherd_utils.config import settings
 from shepherd_utils.db import get_message, save_message
diff --git a/workers/arax_pathfinder/Dockerfile b/workers/arax_pathfinder/Dockerfile
index 890204f..68ae2b6 100644
--- a/workers/arax_pathfinder/Dockerfile
+++ b/workers/arax_pathfinder/Dockerfile
@@ -1,5 +1,5 @@
 # Use RENCI python base image
-FROM ghcr.io/translatorsri/renci-python-image:3.11.5
+FROM ghcr.io/translatorsri/renci-python-image:3.12.13
 
 # Add image info
 LABEL org.opencontainers.image.source https://github.com/BioPack-team/shepherd
diff --git a/workers/arax_pathfinder/requirements.txt b/workers/arax_pathfinder/requirements.txt
index 5f5123c..167e022 100644
--- a/workers/arax_pathfinder/requirements.txt
+++ b/workers/arax_pathfinder/requirements.txt
@@ -1,2 +1,3 @@
-catrax-pathfinder==1.2.2
-biolink-helper-pkg==1.0.0
\ No newline at end of file
+catrax-pathfinder==2.4.3
+biolink-helper-pkg==1.0.0
+httpx==0.28.1
\ No newline at end of file
diff --git a/workers/arax_pathfinder/worker.py b/workers/arax_pathfinder/worker.py
index 2c1014d..7428d1b 100644
--- a/workers/arax_pathfinder/worker.py
+++ b/workers/arax_pathfinder/worker.py
@@ -7,6 +7,7 @@
 import uuid
 from pathlib import Path
 
+import httpx
 import requests
 from biolink_helper_pkg import BiolinkHelper
 from pathfinder.Pathfinder import Pathfinder
@@ -37,8 +38,8 @@
 NUM_TOTAL_HOPS = 4
 MAX_HOPS_TO_EXPLORE = 4
 MAX_PATHFINDER_PATHS = 500
-PRUNE_TOP_K = 100
-NODE_DEGREE_THRESHOLD = 1000000
+PRUNE_TOP_K = 75
+NODE_DEGREE_THRESHOLD = 10000
 
 OUT_PATH = Path("general_concepts.json")
 
@@ -66,6 +67,47 @@ def get_blocked_list():
     synonyms = set(s.lower() for s in json_block_list["synonyms"])
     return set(json_block_list["curies"]), synonyms
 
+async def rehydrate(kg, retriever_url, logger):
+    headers = {"Content-Type": "application/json", "Accept": "application/json"}
+    payload = {
+        "message": {
+            "knowledge_graph": kg
+        },
+        "parameters": {
+            "rehydrate": True,
+            "tier": 0
+        }
+    }
+
+    try:
+        async with httpx.AsyncClient(timeout=30.0) as client:
+            res = await client.post(
+                retriever_url.replace("query", "rehydrate"), headers=headers, json=payload
+            )
+        res.raise_for_status()
+        return res.json()["message"]["knowledge_graph"]
+
+    except httpx.HTTPStatusError as http_err:
+        logger.error(f"HTTP error occurred: {http_err}")
+        if res.text:
+            logger.error(f"Error details: {res.text}")
+        raise http_err
+    except httpx.ConnectError as conn_err:
+        logger.error(f"Connection error occurred: {conn_err}")
+        raise conn_err
+    except httpx.TimeoutException as timeout_err:
+        logger.error(f"Timeout error occurred: {timeout_err}")
+        raise timeout_err
+    except httpx.RequestError as req_err:
+        logger.error(f"An unexpected error occurred: {req_err}")
+        raise req_err
+    except json.JSONDecodeError:
+        logger.error("Failed to parse the response as JSON.")
+        logger.error(f"Raw response: {res.text}")
+        raise
+    except Exception as e:
+        logger.error(f"An unexpected error occurred: {e}")
+        raise e
 
 def execute_pathfinding_sync(
     pinned_node_ids, pinned_node_keys, intermediate_categories, logger
@@ -74,8 +116,7 @@ def execute_pathfinding_sync(
     blocked_curies, blocked_synonyms = get_blocked_list()
 
     pathfinder_instance = Pathfinder(
-        "MLRepo",
-        settings.plover_url,
+        f"retriever:{settings.sync_kg_retrieval_url}",
         settings.curie_ngd_addr,
         settings.node_degree_addr,
         blocked_curies,
@@ -162,6 +203,8 @@ async def pathfinder(task, logger: logging.Logger):
             intermediate_categories,
             logger,
         )
+        logger.info(f"Rehydrating knowledge graph with retriever")
+        knowledge_graph = await rehydrate(knowledge_graph, settings.kg_rehydrate_url, logger)
 
         res = []
         if result is not None:

From 7bc9ba2ecaa9165fec8ee2c10991f5931662d9bd Mon Sep 17 00:00:00 2001
From: mohsenht <mohsen.taheri.1989@gmail.com>
Date: Tue, 28 Jul 2026 20:04:14 -0400
Subject: [PATCH 21/32] Auto download sqlite files for developers

---
 README.md                         |  41 ++++--
 compose.yml                       |  11 +-
 shepherd_utils/config.py          |   9 +-
 shepherd_utils/data_download.py   | 227 +++++++++++++++++++++++++++---
 workers/arax_pathfinder/worker.py |  14 +-
 5 files changed, 266 insertions(+), 36 deletions(-)

diff --git a/README.md b/README.md
index f601a01..066b77d 100644
--- a/README.md
+++ b/README.md
@@ -14,33 +14,46 @@ The main entrypoint is `./compose.yml` and will spin everything up.
 
 If you want to add a new operation/worker, add a new service in `compose.yml` under `services`.
 
-### Worker data (LMDB) downloads
+### Worker data (LMDB / sqlite) downloads
 
-A couple of workers read from large, read-only LMDB datasets that are too big to
+A couple of workers read from large, read-only sqlite databases and LMDB datasets that are too big to
 commit to git (they're gitignored and volume-mounted from the host):
 
 - **`aragorn_omnicorp`** → `./omnicorp_lmdb/` (`curies.lmdb`, `shared_counts.lmdb`)
 - **`score_paths`** → `./pathfinder_embeddings/` (a directory-style LMDB)
+- **`arax_pathfinder`** → `./arax_pathfinder_dbs/` (`curie_ngd_v1.0_<tier-version>.sqlite`, `tier0-info-for-overlay_v1.0_<tier-version>.sqlite`)
 
-So a new developer doesn't have to source these by hand, each worker can fetch
-its dataset on first startup. Point it at a `.tar.gz` on an external server by
-adding the matching variable to your root `.env` file:
+So a new developer doesn't have to source these by hand, each worker can fetch its dataset on first
+startup. Two download mechanisms are supported, depending on where the dataset lives:
+
+**LMDB datasets (`aragorn_omnicorp`, `score_paths`)** are fetched as a `.tar.gz` from a plain HTTP(S)
+URL and extracted in place. Add the matching variable to your root `.env` file:
 
 ```dotenv
 OMNICORP_LMDB_URL=https://example.org/path/omnicorp_lmdb.tar.gz
 PATHFINDER_EMBEDDINGS_URL=https://example.org/path/pathfinder_embeddings.tar.gz
 ```
 
-On startup the worker checks whether its LMDB files already exist in the
-volume-mounted directory. If they're missing and a URL is set, it downloads the
-archive and extracts it into that directory — which lives on the host, so the
-data persists across restarts and is only downloaded once. If the files are
-already present, or no URL is configured, the download is skipped (production
-mounts this data out of band, so it's unaffected).
+The archive for each dataset should contain the expected files at its top level: `curies.lmdb` and
+`shared_counts.lmdb` for omnicorp, `data.mdb` (and `lock.mdb`) for the embeddings.
+
+**`arax_pathfinder`'s sqlite databases** don't live behind a URL — they're on a private,
+SSH-accessible host (`arax-databases.rtx.ai`), so each file is fetched individually via `scp`
+instead. The filenames and remote directory both embed a data-tier version that changes
+periodically, so only one variable needs updating when a new tier ships:
+
+```dotenv
+ARAX_PATHFINDER_TIER_VERSION=tier0-20260621
+```
+
+This requires an SSH key with access to that host, mounted read-only into the container
+(`~/.ssh:/root/.ssh:ro` in docker-compose.yml).
 
-The archive for each dataset should contain the expected files at its top level:
-`curies.lmdb` and `shared_counts.lmdb` for omnicorp, `data.mdb` (and
-`lock.mdb`) for the embeddings.
+On startup, each worker checks whether its files already exist in the volume-mounted directory. If
+they're missing and a source is configured (URL or scp path), it fetches them into that directory —
+which lives on the host, so the data persists across restarts and is only downloaded once. If the
+files are already present, or no source is configured, the download is skipped (production mounts
+this data out of band, so it's unaffected).
 
 ### Worker
 
diff --git a/compose.yml b/compose.yml
index 9fd464d..c9ef77b 100644
--- a/compose.yml
+++ b/compose.yml
@@ -198,7 +198,7 @@ services:
       - ./.env:/app/.env
       # First run? Set PATHFINDER_EMBEDDINGS_URL in your .env to a .tar.gz and
       # the worker downloads + extracts the embeddings LMDB into this mount on
-      # startup (see README "Worker data (LMDB) downloads").
+      # startup (see README "Worker data (LMDB / sqlite) downloads").
       - ./pathfinder_embeddings:/app/pathfinder_embeddings
 
   ######### Example ARA
@@ -305,7 +305,7 @@ services:
       - ./.env:/app/.env
       # First run? Set OMNICORP_LMDB_URL in your .env to a .tar.gz and the worker
       # downloads + extracts curies.lmdb / shared_counts.lmdb into this mount on
-      # startup (see README "Worker data (LMDB) downloads").
+      # startup (see README "Worker data (LMDB / sqlite) downloads").
       - ./omnicorp_lmdb:/app/omnicorp_lmdb
   aragorn_score:
     container_name: aragorn_score
@@ -352,7 +352,14 @@ services:
     volumes:
       - ./logs:/app/logs
       - ./.env:/app/.env
+      # First run? The worker scp's its two sqlite dbs down from
+      # arax-databases.rtx.ai on startup. Set ARAX_PATHFINDER_TIER_VERSION in
+      # your .env if you need a tier other than the default (see README "Worker
+      # data (LMDB / sqlite) downloads"). Requires your SSH key to have access
+      # to that host.
       - ./arax_pathfinder_dbs:/app/arax_pathfinder_dbs
+      # Replace '/Users/facadmin' with your local home directory path
+      - /Users/facadmin/.ssh:/home/nru/.ssh:ro
 
   arax_rank:
     container_name: arax_rank
diff --git a/shepherd_utils/config.py b/shepherd_utils/config.py
index 79a53ce..9d8bac5 100644
--- a/shepherd_utils/config.py
+++ b/shepherd_utils/config.py
@@ -80,13 +80,18 @@ class Settings(BaseSettings):
     # ARAX configs
     arax_url: str = "https://arax.ncats.io/shepherd/api/arax/v1.4/query"
     plover_url: str = "https://kg2cploverdb.ci.transltr.io"
-    curie_ngd_addr: str = "sqlite:/app/arax_pathfinder_dbs/curie_ngd_v1.0_tier0-20260621.sqlite"
-    node_degree_addr: str = "sqlite:/app/arax_pathfinder_dbs/tier0-info-for-overlay_v1.0_tier0-20260621.sqlite"
     arax_biolink_version: str = "4.2.5"
     arax_blocked_list_url: str = (
         "https://raw.githubusercontent.com/RTXteam/RTX/master/"
         "code/ARAX/KnowledgeSources/general_concepts.json"
     )
+
+    arax_pathfinder_dbs_dir: str = "arax_pathfinder_dbs"
+    arax_pathfinder_tier_version: str = "tier0-20260621"
+    arax_pathfinder_curie_ngd_sqlite_filename: str = "curie_ngd_v1.0_{version}.sqlite"
+    arax_pathfinder_tier0_overlay_sqlite_filename: str = "tier0-info-for-overlay_v1.0_{version}.sqlite"
+    arax_pathfinder_sqlite_host: str = "rtxconfig@arax-databases.rtx.ai"
+    arax_pathfinder_sqlite_remote_dir: str = "~/{version}"
     # End of ARAX configs
 
     pathfinder_redis_host: str = "host.docker.internal"
diff --git a/shepherd_utils/data_download.py b/shepherd_utils/data_download.py
index 6df753f..dce8282 100644
--- a/shepherd_utils/data_download.py
+++ b/shepherd_utils/data_download.py
@@ -1,32 +1,42 @@
-"""Ensure large read-only LMDB datasets are present, downloading them on first
+"""Ensure large read-only datasets are present, downloading them on first
 run so new developers can spin the stack up locally.
 
-The ``aragorn_omnicorp`` and ``score_paths`` workers read from LMDB datasets
-that are far too large to commit to git -- they're gitignored and volume-mounted
-from the host (``./omnicorp_lmdb`` and ``./pathfinder_embeddings``). In
-production these volumes are provisioned out of band, but a developer running
-``docker compose up`` for the first time has empty directories, and the workers
-crash on startup trying to open a missing LMDB.
+Several workers read from datasets that are far too large to commit to git --
+they're gitignored and volume-mounted from the host (``./omnicorp_lmdb``,
+``./pathfinder_embeddings``, ``./arax_pathfinder_dbs``). In production these
+volumes are provisioned out of band, but a developer running
+``docker compose up`` for the first time has empty directories, and the
+workers crash on startup trying to open missing files.
 
-When a download URL is configured (``OMNICORP_LMDB_URL`` /
-``PATHFINDER_EMBEDDINGS_URL``, read via :mod:`shepherd_utils.config`), each of
-those workers calls the matching ``ensure_*`` helper at startup:
+Two flavors of remote source are supported:
+
+* **HTTP** -- a single ``.tar.gz`` fetched via ``urllib`` and extracted in
+  place (``OMNICORP_LMDB_URL`` / ``PATHFINDER_EMBEDDINGS_URL``, used by
+  ``aragorn_omnicorp`` and ``score_paths`` below).
+* **SCP** -- individual files fetched from a private, SSH-accessible host via
+  the system ``scp`` binary (used by ``arax_pathfinder`` below, whose two
+  sqlite databases live on ``arax-databases.rtx.ai`` rather than behind a
+  plain URL -- there's no bucket/CDN in front of them, just SSH access).
+
+When a download source is configured (read via :mod:`shepherd_utils.config`),
+each worker calls its matching ``ensure_*`` helper at startup:
 
 * if the expected files are already present it's a no-op;
-* otherwise the dataset is fetched as a ``.tar.gz`` from the external server and
-  extracted into the (volume-mounted) target directory, so it persists on the
-  host and is only downloaded once.
+* otherwise the dataset is fetched and written into the (volume-mounted)
+  target directory, so it persists on the host and is only downloaded once.
 
-With no URL configured the call is a no-op that logs how to enable the download,
-so production -- where the data is already mounted -- is unaffected.
+With no source configured the call is a no-op that logs how to enable the
+download, so production -- where the data is already mounted -- is
+unaffected.
 """
 
 import logging
 import os
+import subprocess
 import tarfile
 import tempfile
 import urllib.request
-from typing import List, Optional
+from typing import Dict, List, Optional, Tuple
 
 from shepherd_utils.config import settings
 
@@ -153,6 +163,129 @@ def ensure_lmdb_dataset(
     logger.info(f"{name}: dataset ready in {target_dir}.")
 
 
+def _scp_download(remote_path: str, dest_path: str, logger: logging.Logger) -> None:
+    """Copy a single file from a remote host to ``dest_path`` via ``scp``.
+
+    Unlike ``_download`` above, these sqlite files aren't behind a plain URL --
+    they live on a private, SSH-accessible host (see README), so this shells
+    out to the system ``scp`` binary and relies on the caller's SSH key (or
+    agent) for auth rather than any credential this code holds.
+
+    ``BatchMode=yes`` makes scp fail fast instead of hanging on an interactive
+    password/passphrase prompt if the key isn't set up. The known-hosts file is
+    redirected to a scratch path so this still works even when ``~/.ssh`` is
+    mounted read-only -- a fresh container has nothing pinned there yet, and
+    ``accept-new`` trusts the host key on first connect without prompting.
+    ``-C`` enables compression, which helps for a database-sized transfer.
+    """
+    logger.info(f"Downloading {remote_path} via scp ...")
+    cmd = [
+        "scp",
+        "-C",
+        "-o", "BatchMode=yes",
+        "-o", "StrictHostKeyChecking=accept-new",
+        "-o", "UserKnownHostsFile=/tmp/known_hosts",
+        remote_path,
+        dest_path,
+    ]
+    try:
+        subprocess.run(cmd, check=True, capture_output=True, text=True)
+    except FileNotFoundError as e:
+        raise RuntimeError(
+            "scp binary not found in this image -- install openssh-client."
+        ) from e
+    except subprocess.CalledProcessError as e:
+        raise RuntimeError(
+            f"scp failed for {remote_path} (exit {e.returncode}): "
+            f"{e.stderr.strip()}. Confirm your SSH key has access to the "
+            f"source host and is mounted into the container (see README)."
+        ) from e
+    size_mb = os.path.getsize(dest_path) / 1e6
+    logger.info(f"Download complete: {dest_path} ({size_mb:.0f} MB)")
+
+
+def ensure_scp_dataset(
+    name: str,
+    target_dir: str,
+    file_sources: Dict[str, str],
+    logger: Optional[logging.Logger] = None,
+) -> None:
+    """Ensure each file in ``file_sources`` exists under ``target_dir``.
+
+    Unlike ``ensure_lmdb_dataset`` (one ``.tar.gz`` archive fetched over HTTP
+    and extracted), each of these files is fetched individually via ``scp``
+    from a private, SSH-accessible host. ``file_sources`` maps the expected
+    local filename to its ``user@host:path`` remote source; a filename whose
+    source is empty is skipped (warned about) rather than downloaded, same as
+    an unset ``url`` in ``ensure_lmdb_dataset``.
+
+    Idempotent: once a file is present it's left alone, so it's safe to call
+    unconditionally on every worker startup.
+    """
+    logger = logger or logging.getLogger(__name__)
+    required_files = list(file_sources.keys())
+
+    missing = _missing_files(target_dir, required_files)
+    if not missing:
+        logger.info(
+            f"{name}: dataset already present in {target_dir}; skipping download."
+        )
+        return
+
+    os.makedirs(target_dir, exist_ok=True)
+    logger.info(f"{name}: dataset missing from {target_dir} (missing: {missing}).")
+
+    attempted = []
+    for filename in missing:
+        remote = file_sources.get(filename)
+        if not remote:
+            logger.warning(
+                f"{name}: {filename} missing from {target_dir} and no source "
+                f"configured for it. Set the corresponding *_SOURCE env var (see "
+                f"the README) to download it automatically, or provide the file "
+                f"manually. The worker will fail to start without it."
+            )
+            continue
+        attempted.append(filename)
+
+        dest_path = os.path.join(target_dir, filename)
+        # Download to a temp file in the same dir first, then atomically rename,
+        # so a partial/interrupted transfer is never mistaken for a complete
+        # file (same reasoning as the tar.gz download above).
+        tmp_fd, tmp_path = tempfile.mkstemp(suffix=".part", dir=target_dir)
+        os.close(tmp_fd)
+        try:
+            _scp_download(remote, tmp_path, logger)
+            os.replace(tmp_path, dest_path)
+        except Exception:
+            try:
+                os.remove(tmp_path)
+            except OSError:
+                pass
+            raise
+
+    # Only files we actually attempted (had a source) count toward failure --
+    # a file with no source configured was already warned about above and is
+    # expected to still be missing, same as an unset url in
+    # ensure_lmdb_dataset. Checking against `required_files` here would raise
+    # even when nothing went wrong.
+    still_missing = _missing_files(target_dir, attempted)
+    if still_missing:
+        raise RuntimeError(
+            f"{name}: still missing expected files after download attempt: "
+            f"{still_missing}. Check that the *_SOURCE env vars are set and that "
+            f"your SSH key has access to the source host."
+        )
+    if _missing_files(target_dir, required_files):
+        logger.warning(
+            f"{name}: dataset partially ready in {target_dir} -- some files have "
+            f"no source configured (see warnings above). The worker will fail "
+            f"when it tries to open them."
+        )
+    else:
+        logger.info(f"{name}: dataset ready in {target_dir}.")
+
+
 def ensure_omnicorp_lmdb(logger: Optional[logging.Logger] = None) -> None:
     """Ensure the omnicorp curies + shared-counts LMDBs are present.
 
@@ -187,3 +320,65 @@ def ensure_pathfinder_embeddings(logger: Optional[logging.Logger] = None) -> Non
         url=settings.pathfinder_embeddings_url,
         logger=logger,
     )
+
+
+def arax_pathfinder_sqlite_paths() -> Tuple[str, str]:
+    """Return ``(curie_ngd_path, node_degree_path)`` for the arax_pathfinder
+    sqlite databases, built from ``arax_pathfinder_dbs_dir`` + the filename
+    templates + the current ``arax_pathfinder_tier_version``.
+
+    Single source of truth for these two paths: ``ensure_arax_pathfinder_dbs``
+    (below) uses it to know what to download and where, and worker.py's
+    ``execute_pathfinding_sync`` uses it to know what to open, so the two can
+    never disagree about a file's location the way two independently-defined
+    settings could.
+    """
+    version = settings.arax_pathfinder_tier_version
+    curie_ngd_path = os.path.join(
+        settings.arax_pathfinder_dbs_dir,
+        settings.arax_pathfinder_curie_ngd_sqlite_filename.format(version=version),
+    )
+    node_degree_path = os.path.join(
+        settings.arax_pathfinder_dbs_dir,
+        settings.arax_pathfinder_tier0_overlay_sqlite_filename.format(version=version),
+    )
+    return curie_ngd_path, node_degree_path
+
+
+def ensure_arax_pathfinder_dbs(logger: Optional[logging.Logger] = None) -> None:
+    """Ensure the arax_pathfinder worker's two sqlite databases are present.
+
+    Both live on a private, SSH-accessible host (``arax-databases.rtx.ai``)
+    rather than behind a plain download URL, so each is fetched individually
+    via ``scp`` instead of the tar.gz + extract flow used for the LMDB
+    datasets above. Both are expected in the same directory (see the
+    ``arax_pathfinder`` volume mount in docker-compose.yml).
+
+    The local filenames and the remote directory both embed a data-tier
+    version (e.g. ``tier0-20260621``) that changes periodically as new tiers
+    ship. Rather than duplicate that string across separate path/source
+    settings -- which can drift out of sync if only one is updated -- the
+    filenames and remote dir are templates with a ``{version}`` placeholder,
+    filled in from the single ``arax_pathfinder_tier_version`` setting.
+    Bumping to a new tier is then one env var change
+    (``ARAX_PATHFINDER_TIER_VERSION``) rather than several.
+    """
+    curie_ngd_path, node_degree_path = arax_pathfinder_sqlite_paths()
+    target_dir = settings.arax_pathfinder_dbs_dir
+
+    version = settings.arax_pathfinder_tier_version
+    remote_dir = settings.arax_pathfinder_sqlite_remote_dir.format(version=version)
+    host = settings.arax_pathfinder_sqlite_host
+
+    curie_ngd_filename = os.path.basename(curie_ngd_path)
+    node_degree_filename = os.path.basename(node_degree_path)
+
+    ensure_scp_dataset(
+        name="arax_pathfinder",
+        target_dir=target_dir,
+        file_sources={
+            curie_ngd_filename: f"{host}:{remote_dir}/{curie_ngd_filename}",
+            node_degree_filename: f"{host}:{remote_dir}/{node_degree_filename}",
+        },
+        logger=logger,
+    )
diff --git a/workers/arax_pathfinder/worker.py b/workers/arax_pathfinder/worker.py
index 7428d1b..d81a524 100644
--- a/workers/arax_pathfinder/worker.py
+++ b/workers/arax_pathfinder/worker.py
@@ -13,6 +13,10 @@
 from pathfinder.Pathfinder import Pathfinder
 
 from shepherd_utils.config import settings
+from shepherd_utils.data_download import (
+    arax_pathfinder_sqlite_paths,
+    ensure_arax_pathfinder_dbs,
+)
 from shepherd_utils.db import (
     get_message,
     save_message,
@@ -115,10 +119,11 @@ def execute_pathfinding_sync(
 
     blocked_curies, blocked_synonyms = get_blocked_list()
 
+    curie_ngd_path, node_degree_path = arax_pathfinder_sqlite_paths()
     pathfinder_instance = Pathfinder(
         f"retriever:{settings.sync_kg_retrieval_url}",
-        settings.curie_ngd_addr,
-        settings.node_degree_addr,
+        f"sqlite:{curie_ngd_path}",
+        f"sqlite:{node_degree_path}",
         blocked_curies,
         blocked_synonyms,
         logger,
@@ -262,6 +267,11 @@ async def process_task(task, parent_ctx, logger: logging.Logger, limiter):
 
 async def poll_for_tasks():
     """On initialization, poll indefinitely for available tasks."""
+    # Ensure the two sqlite databases exist before any task tries to open them
+    # (a first-run local `docker compose up` starts with the volume-mounted
+    # directory empty). No-op once present or when no scp source is configured
+    # (e.g. production, where the data is mounted out of band).
+    ensure_arax_pathfinder_dbs(logging.getLogger(STREAM))
     while True:
         try:
             async for task, parent_ctx, logger, limiter in get_tasks(

From b823e96eb2d71cf9b034a6d8f6b4af95ef88da6b Mon Sep 17 00:00:00 2001
From: mohsenht <mohsen.taheri.1989@gmail.com>
Date: Wed, 29 Jul 2026 12:40:29 -0400
Subject: [PATCH 22/32] redundant plover_url config removed.

---
 shepherd_utils/config.py | 1 -
 1 file changed, 1 deletion(-)

diff --git a/shepherd_utils/config.py b/shepherd_utils/config.py
index c2b8237..bb6722e 100644
--- a/shepherd_utils/config.py
+++ b/shepherd_utils/config.py
@@ -88,7 +88,6 @@ class Settings(BaseSettings):
 
     # ARAX configs
     arax_url: str = "https://arax.ncats.io/shepherd/api/arax/v1.4/query"
-    plover_url: str = "https://kg2cploverdb.ci.transltr.io"
     arax_biolink_version: str = "4.2.5"
     arax_blocked_list_url: str = (
         "https://raw.githubusercontent.com/RTXteam/RTX/master/"

From 5195035c6aae53c5f9e5d142254fa7d05fe47522 Mon Sep 17 00:00:00 2001
From: mohsenht <mohsen.taheri.1989@gmail.com>
Date: Wed, 29 Jul 2026 12:51:03 -0400
Subject: [PATCH 23/32] ssh path generalized for all developers

---
 README.md   | 2 +-
 compose.yml | 4 ++--
 2 files changed, 3 insertions(+), 3 deletions(-)

diff --git a/README.md b/README.md
index 066b77d..f487dcc 100644
--- a/README.md
+++ b/README.md
@@ -47,7 +47,7 @@ ARAX_PATHFINDER_TIER_VERSION=tier0-20260621
 ```
 
 This requires an SSH key with access to that host, mounted read-only into the container
-(`~/.ssh:/root/.ssh:ro` in docker-compose.yml).
+(`~/.ssh:/home/nru/.ssh:ro` in docker-compose.yml).
 
 On startup, each worker checks whether its files already exist in the volume-mounted directory. If
 they're missing and a source is configured (URL or scp path), it fetches them into that directory —
diff --git a/compose.yml b/compose.yml
index 5fbf34b..2ad9cbc 100644
--- a/compose.yml
+++ b/compose.yml
@@ -364,8 +364,8 @@ services:
       # data (LMDB / sqlite) downloads"). Requires your SSH key to have access
       # to that host.
       - ./arax_pathfinder_dbs:/app/arax_pathfinder_dbs
-      # Replace '/Users/facadmin' with your local home directory path
-      - /Users/facadmin/.ssh:/home/nru/.ssh:ro
+      # Replace '~/' with your local home directory path if needed
+      - ~/.ssh:/home/nru/.ssh:ro
 
   arax_rank:
     container_name: arax_rank

From c699d276781136b24504ec429b00d456b992783c Mon Sep 17 00:00:00 2001
From: mohsenht <mohsen.taheri.1989@gmail.com>
Date: Wed, 29 Jul 2026 12:52:07 -0400
Subject: [PATCH 24/32] readme correction

---
 README.md | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/README.md b/README.md
index f487dcc..d3bbe53 100644
--- a/README.md
+++ b/README.md
@@ -47,7 +47,7 @@ ARAX_PATHFINDER_TIER_VERSION=tier0-20260621
 ```
 
 This requires an SSH key with access to that host, mounted read-only into the container
-(`~/.ssh:/home/nru/.ssh:ro` in docker-compose.yml).
+(`~/.ssh:/home/nru/.ssh:ro` in compose.yml).
 
 On startup, each worker checks whether its files already exist in the volume-mounted directory. If
 they're missing and a source is configured (URL or scp path), it fetches them into that directory —

From cbd9ad58964a344574fe87e5e5addd2bc4bfa24c Mon Sep 17 00:00:00 2001
From: mohsenht <mohsen.taheri.1989@gmail.com>
Date: Wed, 5 Aug 2026 15:57:01 -0400
Subject: [PATCH 25/32] Fetch sqlite files over Https instead of scp

---
 README.md                       |   7 +-
 compose.yml                     |  11 +-
 shepherd_utils/config.py        |   3 +-
 shepherd_utils/data_download.py | 184 +++++++++++++-------------------
 4 files changed, 81 insertions(+), 124 deletions(-)

diff --git a/README.md b/README.md
index d3bbe53..4921dfc 100644
--- a/README.md
+++ b/README.md
@@ -37,10 +37,9 @@ PATHFINDER_EMBEDDINGS_URL=https://example.org/path/pathfinder_embeddings.tar.gz
 The archive for each dataset should contain the expected files at its top level: `curies.lmdb` and
 `shared_counts.lmdb` for omnicorp, `data.mdb` (and `lock.mdb`) for the embeddings.
 
-**`arax_pathfinder`'s sqlite databases** don't live behind a URL — they're on a private,
-SSH-accessible host (`arax-databases.rtx.ai`), so each file is fetched individually via `scp`
-instead. The filenames and remote directory both embed a data-tier version that changes
-periodically, so only one variable needs updating when a new tier ships:
+**arax_pathfinder's sqlite databases** are served as plain files over HTTPS. The
+filenames embed a Knowledge Graph version that changes periodically, so only one variable needs updating
+when a new Knowledge Graph ships:
 
 ```dotenv
 ARAX_PATHFINDER_TIER_VERSION=tier0-20260621
diff --git a/compose.yml b/compose.yml
index 2ad9cbc..0e1a39a 100644
--- a/compose.yml
+++ b/compose.yml
@@ -358,14 +358,11 @@ services:
     volumes:
       - ./logs:/app/logs
       - ./.env:/app/.env
-      # First run? The worker scp's its two sqlite dbs down from
-      # arax-databases.rtx.ai on startup. Set ARAX_PATHFINDER_TIER_VERSION in
-      # your .env if you need a tier other than the default (see README "Worker
-      # data (LMDB / sqlite) downloads"). Requires your SSH key to have access
-      # to that host.
+      # First run? The worker downloads its two sqlite dbs from
+      # kg2webhost.rtx.ai on startup, no credentials needed. Set
+      # ARAX_PATHFINDER_TIER_VERSION in your .env if you need a tier other than
+      # the default (see README "Worker data (LMDB / sqlite) downloads").
       - ./arax_pathfinder_dbs:/app/arax_pathfinder_dbs
-      # Replace '~/' with your local home directory path if needed
-      - ~/.ssh:/home/nru/.ssh:ro
 
   arax_rank:
     container_name: arax_rank
diff --git a/shepherd_utils/config.py b/shepherd_utils/config.py
index bb6722e..62e5e5d 100644
--- a/shepherd_utils/config.py
+++ b/shepherd_utils/config.py
@@ -98,8 +98,7 @@ class Settings(BaseSettings):
     arax_pathfinder_tier_version: str = "tier0-20260621"
     arax_pathfinder_curie_ngd_sqlite_filename: str = "curie_ngd_v1.0_{version}.sqlite"
     arax_pathfinder_tier0_overlay_sqlite_filename: str = "tier0-info-for-overlay_v1.0_{version}.sqlite"
-    arax_pathfinder_sqlite_host: str = "rtxconfig@arax-databases.rtx.ai"
-    arax_pathfinder_sqlite_remote_dir: str = "~/{version}"
+    arax_pathfinder_sqlite_base_url: str = "https://kg2webhost.rtx.ai/tier0"
     # End of ARAX configs
 
     pathfinder_redis_host: str = "host.docker.internal"
diff --git a/shepherd_utils/data_download.py b/shepherd_utils/data_download.py
index dce8282..ee73325 100644
--- a/shepherd_utils/data_download.py
+++ b/shepherd_utils/data_download.py
@@ -8,15 +8,14 @@
 ``docker compose up`` for the first time has empty directories, and the
 workers crash on startup trying to open missing files.
 
-Two flavors of remote source are supported:
+Two flavors of HTTP source are supported:
 
-* **HTTP** -- a single ``.tar.gz`` fetched via ``urllib`` and extracted in
+* **Archive** -- a single ``.tar.gz`` fetched via ``urllib`` and extracted in
   place (``OMNICORP_LMDB_URL`` / ``PATHFINDER_EMBEDDINGS_URL``, used by
   ``aragorn_omnicorp`` and ``score_paths`` below).
-* **SCP** -- individual files fetched from a private, SSH-accessible host via
-  the system ``scp`` binary (used by ``arax_pathfinder`` below, whose two
-  sqlite databases live on ``arax-databases.rtx.ai`` rather than behind a
-  plain URL -- there's no bucket/CDN in front of them, just SSH access).
+* **Per-file** -- individual files fetched directly, no archive/extract step
+  (``arax_pathfinder`` below, whose two sqlite databases are served as plain
+  files rather than bundled into one archive).
 
 When a download source is configured (read via :mod:`shepherd_utils.config`),
 each worker calls its matching ``ensure_*`` helper at startup:
@@ -32,9 +31,9 @@
 
 import logging
 import os
-import subprocess
 import tarfile
 import tempfile
+import urllib.error
 import urllib.request
 from typing import Dict, List, Optional, Tuple
 
@@ -53,29 +52,40 @@ def _missing_files(target_dir: str, required_files: List[str]) -> List[str]:
 def _download(url: str, dest_path: str, logger: logging.Logger) -> None:
     """Stream ``url`` to ``dest_path``, logging progress periodically."""
     logger.info(f"Downloading dataset from {url} ...")
-    # nosec B310: the URL is operator-configured (an env var), not user input.
-    with urllib.request.urlopen(url) as resp:  # noqa: S310
-        header = resp.headers.get("Content-Length")
-        total = int(header) if header and header.isdigit() else None
-        read = 0
-        step = 50 * 1024 * 1024  # log roughly every 50 MB
-        next_log = step
-        with open(dest_path, "wb") as out:
-            while True:
-                chunk = resp.read(1024 * 1024)
-                if not chunk:
-                    break
-                out.write(chunk)
-                read += len(chunk)
-                if read >= next_log:
-                    if total:
-                        logger.info(
-                            f"  ... {read / 1e6:.0f}/{total / 1e6:.0f} MB "
-                            f"({100 * read / total:.0f}%)"
-                        )
-                    else:
-                        logger.info(f"  ... {read / 1e6:.0f} MB")
-                    next_log += step
+    try:
+        # nosec B310: the URL is operator-configured (an env var), not user input.
+        with urllib.request.urlopen(url) as resp:  # noqa: S310
+            header = resp.headers.get("Content-Length")
+            total = int(header) if header and header.isdigit() else None
+            read = 0
+            step = 50 * 1024 * 1024  # log roughly every 50 MB
+            next_log = step
+            with open(dest_path, "wb") as out:
+                while True:
+                    chunk = resp.read(1024 * 1024)
+                    if not chunk:
+                        break
+                    out.write(chunk)
+                    read += len(chunk)
+                    if read >= next_log:
+                        if total:
+                            logger.info(
+                                f"  ... {read / 1e6:.0f}/{total / 1e6:.0f} MB "
+                                f"({100 * read / total:.0f}%)"
+                            )
+                        else:
+                            logger.info(f"  ... {read / 1e6:.0f} MB")
+                        next_log += step
+    except urllib.error.HTTPError as e:
+        raise RuntimeError(
+            f"download failed for {url}: HTTP {e.code} {e.reason}. Confirm the "
+            f"URL is correct and reachable."
+        ) from e
+    except urllib.error.URLError as e:
+        raise RuntimeError(
+            f"download failed for {url}: {e.reason}. Confirm the URL is correct "
+            f"and reachable from inside the container."
+        ) from e
     logger.info(f"Download complete: {read / 1e6:.0f} MB")
 
 
@@ -163,61 +173,20 @@ def ensure_lmdb_dataset(
     logger.info(f"{name}: dataset ready in {target_dir}.")
 
 
-def _scp_download(remote_path: str, dest_path: str, logger: logging.Logger) -> None:
-    """Copy a single file from a remote host to ``dest_path`` via ``scp``.
-
-    Unlike ``_download`` above, these sqlite files aren't behind a plain URL --
-    they live on a private, SSH-accessible host (see README), so this shells
-    out to the system ``scp`` binary and relies on the caller's SSH key (or
-    agent) for auth rather than any credential this code holds.
-
-    ``BatchMode=yes`` makes scp fail fast instead of hanging on an interactive
-    password/passphrase prompt if the key isn't set up. The known-hosts file is
-    redirected to a scratch path so this still works even when ``~/.ssh`` is
-    mounted read-only -- a fresh container has nothing pinned there yet, and
-    ``accept-new`` trusts the host key on first connect without prompting.
-    ``-C`` enables compression, which helps for a database-sized transfer.
-    """
-    logger.info(f"Downloading {remote_path} via scp ...")
-    cmd = [
-        "scp",
-        "-C",
-        "-o", "BatchMode=yes",
-        "-o", "StrictHostKeyChecking=accept-new",
-        "-o", "UserKnownHostsFile=/tmp/known_hosts",
-        remote_path,
-        dest_path,
-    ]
-    try:
-        subprocess.run(cmd, check=True, capture_output=True, text=True)
-    except FileNotFoundError as e:
-        raise RuntimeError(
-            "scp binary not found in this image -- install openssh-client."
-        ) from e
-    except subprocess.CalledProcessError as e:
-        raise RuntimeError(
-            f"scp failed for {remote_path} (exit {e.returncode}): "
-            f"{e.stderr.strip()}. Confirm your SSH key has access to the "
-            f"source host and is mounted into the container (see README)."
-        ) from e
-    size_mb = os.path.getsize(dest_path) / 1e6
-    logger.info(f"Download complete: {dest_path} ({size_mb:.0f} MB)")
-
-
-def ensure_scp_dataset(
+def ensure_http_files_dataset(
     name: str,
     target_dir: str,
     file_sources: Dict[str, str],
     logger: Optional[logging.Logger] = None,
 ) -> None:
-    """Ensure each file in ``file_sources`` exists under ``target_dir``.
+    """Ensure each file in ``file_sources`` exists under ``target_dir``,
+    fetching any missing ones directly over HTTP(S) -- one URL per file, no
+    archive/extract step (contrast with ``ensure_lmdb_dataset``'s single
+    ``.tar.gz``).
 
-    Unlike ``ensure_lmdb_dataset`` (one ``.tar.gz`` archive fetched over HTTP
-    and extracted), each of these files is fetched individually via ``scp``
-    from a private, SSH-accessible host. ``file_sources`` maps the expected
-    local filename to its ``user@host:path`` remote source; a filename whose
-    source is empty is skipped (warned about) rather than downloaded, same as
-    an unset ``url`` in ``ensure_lmdb_dataset``.
+    ``file_sources`` maps the expected local filename to its URL; a filename
+    whose URL is empty is skipped (warned about) rather than downloaded, same
+    as an unset ``url`` in ``ensure_lmdb_dataset``.
 
     Idempotent: once a file is present it's left alone, so it's safe to call
     unconditionally on every worker startup.
@@ -237,11 +206,11 @@ def ensure_scp_dataset(
 
     attempted = []
     for filename in missing:
-        remote = file_sources.get(filename)
-        if not remote:
+        url = file_sources.get(filename)
+        if not url:
             logger.warning(
-                f"{name}: {filename} missing from {target_dir} and no source "
-                f"configured for it. Set the corresponding *_SOURCE env var (see "
+                f"{name}: {filename} missing from {target_dir} and no URL "
+                f"configured for it. Set the corresponding *_URL env var (see "
                 f"the README) to download it automatically, or provide the file "
                 f"manually. The worker will fail to start without it."
             )
@@ -255,7 +224,7 @@ def ensure_scp_dataset(
         tmp_fd, tmp_path = tempfile.mkstemp(suffix=".part", dir=target_dir)
         os.close(tmp_fd)
         try:
-            _scp_download(remote, tmp_path, logger)
+            _download(url, tmp_path, logger)
             os.replace(tmp_path, dest_path)
         except Exception:
             try:
@@ -264,8 +233,8 @@ def ensure_scp_dataset(
                 pass
             raise
 
-    # Only files we actually attempted (had a source) count toward failure --
-    # a file with no source configured was already warned about above and is
+    # Only files we actually attempted (had a URL) count toward failure -- a
+    # file with no URL configured was already warned about above and is
     # expected to still be missing, same as an unset url in
     # ensure_lmdb_dataset. Checking against `required_files` here would raise
     # even when nothing went wrong.
@@ -273,13 +242,12 @@ def ensure_scp_dataset(
     if still_missing:
         raise RuntimeError(
             f"{name}: still missing expected files after download attempt: "
-            f"{still_missing}. Check that the *_SOURCE env vars are set and that "
-            f"your SSH key has access to the source host."
+            f"{still_missing}."
         )
     if _missing_files(target_dir, required_files):
         logger.warning(
             f"{name}: dataset partially ready in {target_dir} -- some files have "
-            f"no source configured (see warnings above). The worker will fail "
+            f"no URL configured (see warnings above). The worker will fail "
             f"when it tries to open them."
         )
     else:
@@ -348,37 +316,31 @@ def arax_pathfinder_sqlite_paths() -> Tuple[str, str]:
 def ensure_arax_pathfinder_dbs(logger: Optional[logging.Logger] = None) -> None:
     """Ensure the arax_pathfinder worker's two sqlite databases are present.
 
-    Both live on a private, SSH-accessible host (``arax-databases.rtx.ai``)
-    rather than behind a plain download URL, so each is fetched individually
-    via ``scp`` instead of the tar.gz + extract flow used for the LMDB
-    datasets above. Both are expected in the same directory (see the
-    ``arax_pathfinder`` volume mount in docker-compose.yml).
-
-    The local filenames and the remote directory both embed a data-tier
-    version (e.g. ``tier0-20260621``) that changes periodically as new tiers
-    ship. Rather than duplicate that string across separate path/source
-    settings -- which can drift out of sync if only one is updated -- the
-    filenames and remote dir are templates with a ``{version}`` placeholder,
-    filled in from the single ``arax_pathfinder_tier_version`` setting.
-    Bumping to a new tier is then one env var change
-    (``ARAX_PATHFINDER_TIER_VERSION``) rather than several.
+    Both are served as plain files over HTTPS, so each is fetched individually
+    with a normal GET -- no archive/extract step. Both are expected in the
+    same directory (see the ``arax_pathfinder`` volume mount in
+    docker-compose.yml).
+
+    The version tag shows up in the filename (e.g.
+    ``curie_ngd_v1.0_tier0-20260621.sqlite``) but not in the URL path -- the
+    ``tier0`` segment in ``arax_pathfinder_sqlite_base_url`` is fixed, not the
+    tier version. Only the filename templates are filled in from
+    ``arax_pathfinder_tier_version``. Bumping to a new tier is one env var
+    change (``ARAX_PATHFINDER_TIER_VERSION``).
     """
     curie_ngd_path, node_degree_path = arax_pathfinder_sqlite_paths()
     target_dir = settings.arax_pathfinder_dbs_dir
-
-    version = settings.arax_pathfinder_tier_version
-    remote_dir = settings.arax_pathfinder_sqlite_remote_dir.format(version=version)
-    host = settings.arax_pathfinder_sqlite_host
+    base_url = settings.arax_pathfinder_sqlite_base_url
 
     curie_ngd_filename = os.path.basename(curie_ngd_path)
     node_degree_filename = os.path.basename(node_degree_path)
 
-    ensure_scp_dataset(
+    ensure_http_files_dataset(
         name="arax_pathfinder",
         target_dir=target_dir,
         file_sources={
-            curie_ngd_filename: f"{host}:{remote_dir}/{curie_ngd_filename}",
-            node_degree_filename: f"{host}:{remote_dir}/{node_degree_filename}",
+            curie_ngd_filename: f"{base_url}/{curie_ngd_filename}",
+            node_degree_filename: f"{base_url}/{node_degree_filename}",
         },
         logger=logger,
-    )
+    )
\ No newline at end of file

From 7a0730f9c6db3d53072fed95cb8f95505c22ff7d Mon Sep 17 00:00:00 2001
From: mohsenht <mohsen.taheri.1989@gmail.com>
Date: Wed, 5 Aug 2026 16:15:08 -0400
Subject: [PATCH 26/32] arax_pathfinder_dbs git ignored

---
 .gitignore | 1 +
 1 file changed, 1 insertion(+)

diff --git a/.gitignore b/.gitignore
index 4fb74c6..cf04e5c 100644
--- a/.gitignore
+++ b/.gitignore
@@ -5,6 +5,7 @@ debug/
 gandalf_mmap/
 omnicorp_lmdb/
 pathfinder_embeddings/
+arax_pathfinder_dbs/
 
 # Project specific
 scripts/

From d41090ff25eaae9cb7677181d850135bc1644650 Mon Sep 17 00:00:00 2001
From: Max Wang <max@covar.com>
Date: Wed, 5 Aug 2026 18:42:23 -0400
Subject: [PATCH 27/32] Run black

---
 shepherd_utils/config.py          |  5 +++--
 shepherd_utils/data_download.py   |  2 +-
 workers/arax_pathfinder/worker.py | 19 ++++++++++---------
 3 files changed, 14 insertions(+), 12 deletions(-)

diff --git a/shepherd_utils/config.py b/shepherd_utils/config.py
index 62e5e5d..7f7e65c 100644
--- a/shepherd_utils/config.py
+++ b/shepherd_utils/config.py
@@ -85,7 +85,6 @@ class Settings(BaseSettings):
     kg_rehydrate_url: str = "http://host.docker.internal:8080/rehydrate"
     default_data_tier: int = 0
 
-
     # ARAX configs
     arax_url: str = "https://arax.ncats.io/shepherd/api/arax/v1.4/query"
     arax_biolink_version: str = "4.2.5"
@@ -97,7 +96,9 @@ class Settings(BaseSettings):
     arax_pathfinder_dbs_dir: str = "arax_pathfinder_dbs"
     arax_pathfinder_tier_version: str = "tier0-20260621"
     arax_pathfinder_curie_ngd_sqlite_filename: str = "curie_ngd_v1.0_{version}.sqlite"
-    arax_pathfinder_tier0_overlay_sqlite_filename: str = "tier0-info-for-overlay_v1.0_{version}.sqlite"
+    arax_pathfinder_tier0_overlay_sqlite_filename: str = (
+        "tier0-info-for-overlay_v1.0_{version}.sqlite"
+    )
     arax_pathfinder_sqlite_base_url: str = "https://kg2webhost.rtx.ai/tier0"
     # End of ARAX configs
 
diff --git a/shepherd_utils/data_download.py b/shepherd_utils/data_download.py
index ee73325..610a948 100644
--- a/shepherd_utils/data_download.py
+++ b/shepherd_utils/data_download.py
@@ -343,4 +343,4 @@ def ensure_arax_pathfinder_dbs(logger: Optional[logging.Logger] = None) -> None:
             node_degree_filename: f"{base_url}/{node_degree_filename}",
         },
         logger=logger,
-    )
\ No newline at end of file
+    )
diff --git a/workers/arax_pathfinder/worker.py b/workers/arax_pathfinder/worker.py
index d81a524..b9fc501 100644
--- a/workers/arax_pathfinder/worker.py
+++ b/workers/arax_pathfinder/worker.py
@@ -71,22 +71,20 @@ def get_blocked_list():
     synonyms = set(s.lower() for s in json_block_list["synonyms"])
     return set(json_block_list["curies"]), synonyms
 
+
 async def rehydrate(kg, retriever_url, logger):
     headers = {"Content-Type": "application/json", "Accept": "application/json"}
     payload = {
-        "message": {
-            "knowledge_graph": kg
-        },
-        "parameters": {
-            "rehydrate": True,
-            "tier": 0
-        }
+        "message": {"knowledge_graph": kg},
+        "parameters": {"rehydrate": True, "tier": 0},
     }
 
     try:
         async with httpx.AsyncClient(timeout=30.0) as client:
             res = await client.post(
-                retriever_url.replace("query", "rehydrate"), headers=headers, json=payload
+                retriever_url.replace("query", "rehydrate"),
+                headers=headers,
+                json=payload,
             )
         res.raise_for_status()
         return res.json()["message"]["knowledge_graph"]
@@ -113,6 +111,7 @@ async def rehydrate(kg, retriever_url, logger):
         logger.error(f"An unexpected error occurred: {e}")
         raise e
 
+
 def execute_pathfinding_sync(
     pinned_node_ids, pinned_node_keys, intermediate_categories, logger
 ):
@@ -209,7 +208,9 @@ async def pathfinder(task, logger: logging.Logger):
             logger,
         )
         logger.info(f"Rehydrating knowledge graph with retriever")
-        knowledge_graph = await rehydrate(knowledge_graph, settings.kg_rehydrate_url, logger)
+        knowledge_graph = await rehydrate(
+            knowledge_graph, settings.kg_rehydrate_url, logger
+        )
 
         res = []
         if result is not None:

From ff282dc7ac951f57e0a83314fb10a5b9b5f0d763 Mon Sep 17 00:00:00 2001
From: Max Wang <max@covar.com>
Date: Mon, 10 Aug 2026 10:47:29 -0400
Subject: [PATCH 28/32] Fix test import

---
 tests/unit/test_inject_shepherd_arax_provenance.py | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/tests/unit/test_inject_shepherd_arax_provenance.py b/tests/unit/test_inject_shepherd_arax_provenance.py
index 6c50dc0..0fbdf68 100644
--- a/tests/unit/test_inject_shepherd_arax_provenance.py
+++ b/tests/unit/test_inject_shepherd_arax_provenance.py
@@ -1,4 +1,4 @@
-"""Tests for ``workers.arax.inject_shepherd_arax_provenance``.
+"""Tests for ``shepherd_utils.inject_shepherd_arax_provenance``.
 
 The shepherd-arax injector tags every kgraph edge with an aggregator
 ``infores:shepherd-arax`` source so downstream consumers can attribute
@@ -12,7 +12,7 @@
 
 import copy
 
-from workers.arax.inject_shepherd_arax_provenance import (
+from shepherd_utils.inject_shepherd_arax_provenance import (
     SHEPHERD_ARAX_SOURCE,
     add_shepherd_arax_to_edge_sources,
 )

From 8c26f47ea60501f5d69d80c3fb48ea79dd473ded Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Mon, 10 Aug 2026 16:14:20 +0000
Subject: [PATCH 29/32] Move arax_pathfinder onto a process pool and
 run_task_lifecycle

The pathfinding search ran via asyncio.to_thread, so its GIL-bound work
competed with the event loop -- and TASK_LIMIT was 100, meaning up to 100
concurrent searches. The heartbeat is a coroutine on that same loop pinging
every 5s against a 15s freshness window, so under load a live worker read as
dead: reclaim_orphaned could XCLAIM a still-running task to a peer and
duplicate it, and a stall past 60s tripped LoopWatchdog into os._exit(1).

Restructure to match aragorn_score / arax_rank:

- arax_pathfinder_task is a sync process-pool entrypoint taking only the
  query/response ids. It loads, searches, rehydrates, and saves entirely in a
  spawned child, so neither the search nor the message encode touches the
  parent's loop or heap. rehydrate becomes sync (httpx.Client) since it now
  runs in the child.
- process_task delegates to the shared run_task_lifecycle via a pool closure,
  replacing the hand-rolled copy that used tracer.start_span instead of
  start_as_current_span (so httpx spans never nested under the task span) and
  never recorded exceptions on the span.
- poll_for_tasks builds a ProcessPoolManager sized by resolve_pool_workers and
  passes max_workers to get_tasks; TASK_LIMIT drops 100 -> 10.

Failures now reach the lifecycle instead of being swallowed. The three
query-graph validation failures returned `message, 500`, a value the caller
discarded, so the task was wrapped up as a success with no message ever
written to response_id -- leaving the next worker to KeyError on it. Search
failures were saved as a non-TRAPI {"status": "error"} blob that then flowed
down the workflow. Both now raise and route to finish_query with ERROR.

Also:
- The blocked-concept list is fetched once at startup via
  ensure_http_files_dataset (temp file + atomic rename) rather than with a
  per-task requests.get racing on the same relative path; children parse it
  once and memoize, as they do the BiolinkHelper descendants.
- Add "arax.pathfinder" to PER_STREAM_MIN_IDLE_SEC. It was absent, so its
  reclaim floor was the 30s fast default; 360s sits above pool_task_timeout_sec.
- Dockerfile gets PYTHONFAULTHANDLER=1 so a child hung during import still
  dumps a stack when ProcessPoolManager SIGABRTs it.
- Fix rehydrate's no-op `.replace("query", "rehydrate")` on an already-correct
  URL, and its references to a possibly-unbound `res` in the error handlers.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01H51TTxq1F6o6yZSP1C86MZ
---
 shepherd_utils/reclaim.py                     |   3 +
 tests/unit/test_arax_pathfinder.py            | 254 ++++++++++++
 .../unit/test_worker_dispatch_concurrency.py  |   1 +
 workers/arax_pathfinder/Dockerfile            |   5 +
 workers/arax_pathfinder/worker.py             | 368 +++++++++++-------
 5 files changed, 485 insertions(+), 146 deletions(-)
 create mode 100644 tests/unit/test_arax_pathfinder.py

diff --git a/shepherd_utils/reclaim.py b/shepherd_utils/reclaim.py
index 54121c6..e858878 100644
--- a/shepherd_utils/reclaim.py
+++ b/shepherd_utils/reclaim.py
@@ -49,6 +49,9 @@
     "aragorn.score": 240,
     "bte.lookup": 240,
     "example.lookup": 240,
+    # Pathfinding runs in a process pool bounded by pool_task_timeout_sec
+    # (300s), so no legitimate task can outlive that; the floor sits just above.
+    "arax.pathfinder": 360,
     # Medium-duration workers.
     "arax.rank": 60,
     "merge_message": 60,
diff --git a/tests/unit/test_arax_pathfinder.py b/tests/unit/test_arax_pathfinder.py
new file mode 100644
index 0000000..f572207
--- /dev/null
+++ b/tests/unit/test_arax_pathfinder.py
@@ -0,0 +1,254 @@
+"""Tests for the arax_pathfinder process-pool entrypoint.
+
+The worker imports ``pathfinder.Pathfinder`` and ``biolink_helper_pkg``, which
+are installed only inside its container (see
+``workers/arax_pathfinder/requirements.txt``) and are not in
+``test-requirements.txt``. We stub both into ``sys.modules`` before importing
+the worker, the same trick ``tests/conftest.py`` uses for ``shepherd_utils.otel``.
+"""
+
+import copy
+import logging
+import sys
+import types
+from unittest.mock import MagicMock
+
+import pytest
+
+if "pathfinder" not in sys.modules:
+    _pathfinder_pkg = types.ModuleType("pathfinder")
+    _pathfinder_mod = types.ModuleType("pathfinder.Pathfinder")
+    _pathfinder_mod.Pathfinder = MagicMock(name="Pathfinder")
+    _pathfinder_pkg.Pathfinder = _pathfinder_mod
+    sys.modules["pathfinder"] = _pathfinder_pkg
+    sys.modules["pathfinder.Pathfinder"] = _pathfinder_mod
+
+if "biolink_helper_pkg" not in sys.modules:
+    _biolink_mod = types.ModuleType("biolink_helper_pkg")
+    _biolink_mod.BiolinkHelper = MagicMock(name="BiolinkHelper")
+    sys.modules["biolink_helper_pkg"] = _biolink_mod
+
+from workers.arax_pathfinder import worker as pf_worker  # noqa: E402
+
+LOGGER = logging.getLogger(__name__)
+
+QUERY = {
+    "message": {
+        "query_graph": {
+            "nodes": {
+                "n0": {"ids": ["MONDO:0005148"]},
+                "n1": {"ids": ["CHEBI:15365"]},
+            },
+            "paths": {"p0": {"subject": "n0", "object": "n1", "constraints": []}},
+        }
+    }
+}
+
+PATHS_RESULT = (
+    {"id": "r0", "analyses": [{"score": 1.0}], "node_bindings": {"n0": []}},
+    {"aux0": {"edges": ["e0"]}},
+    {
+        "nodes": {"MONDO:0005148": {}},
+        "edges": {"e0": {"predicate": "biolink:related_to"}},
+    },
+)
+
+
+def _patch_query(mocker, query=None):
+    return mocker.patch(
+        "workers.arax_pathfinder.worker.get_message_sync",
+        return_value=copy.deepcopy(query if query is not None else QUERY),
+    )
+
+
+def test_pathfinder_task_searches_rehydrates_and_saves(mocker):
+    """The process-pool entrypoint reads by id, searches, and writes back.
+
+    Only the two ids cross into the child: the message is loaded with
+    ``get_message_sync``, assembled, and persisted with ``save_message_sync`` --
+    the knowledge graph never has to be pickled back to the parent.
+    """
+    _patch_query(mocker)
+    search = mocker.patch(
+        "workers.arax_pathfinder.worker.execute_pathfinding",
+        return_value=copy.deepcopy(PATHS_RESULT),
+    )
+    rehydrated_kg = {"nodes": {}, "edges": {"e0": {"predicate": "biolink:related_to"}}}
+    rehydrate = mocker.patch(
+        "workers.arax_pathfinder.worker.rehydrate", return_value=rehydrated_kg
+    )
+    save = mocker.patch("workers.arax_pathfinder.worker.save_message_sync")
+
+    pf_worker.arax_pathfinder_task("query-1", "resp-1", LOGGER)
+
+    search.assert_called_once()
+    rehydrate.assert_called_once()
+    save.assert_called_once()
+    saved_id, message = save.call_args.args
+    assert saved_id == "resp-1"
+    assert message["message"]["knowledge_graph"] is rehydrated_kg
+    assert message["message"]["auxiliary_graphs"] == {"aux0": {"edges": ["e0"]}}
+    assert len(message["message"]["results"]) == 1
+    assert message["message"]["results"][0]["essence"] == "result"
+    # Provenance is injected before saving.
+    assert message["message"]["knowledge_graph"]["edges"]["e0"]["sources"] == [
+        {
+            "resource_id": "infores:shepherd-arax",
+            "resource_role": "aggregator_knowledge_source",
+            "source_record_urls": None,
+            "upstream_resource_ids": ["infores:arax"],
+        }
+    ]
+    # Defaults are filled in on the message that gets saved.
+    assert message["parameters"]["tiers"] == [0]
+
+
+def test_pathfinder_task_saves_empty_graphs_when_no_paths_found(mocker):
+    """A search that finds nothing still writes a well-formed TRAPI message."""
+    _patch_query(mocker)
+    mocker.patch(
+        "workers.arax_pathfinder.worker.execute_pathfinding",
+        return_value=(None, None, None),
+    )
+    mocker.patch("workers.arax_pathfinder.worker.rehydrate", return_value=None)
+    save = mocker.patch("workers.arax_pathfinder.worker.save_message_sync")
+
+    pf_worker.arax_pathfinder_task("query-2", "resp-2", LOGGER)
+
+    _, message = save.call_args.args
+    assert message["message"]["results"] == []
+    assert message["message"]["auxiliary_graphs"] == {}
+    assert message["message"]["knowledge_graph"] == {}
+
+
+@pytest.mark.parametrize(
+    "qgraph, expected",
+    [
+        pytest.param(
+            {
+                "nodes": {"n0": {"ids": ["MONDO:0005148"]}, "n1": {}},
+                "paths": {"p0": {"constraints": []}},
+            },
+            "two pinned nodes",
+            id="one_pinned_node",
+        ),
+        pytest.param(
+            {
+                "nodes": {
+                    "n0": {"ids": ["MONDO:0005148"]},
+                    "n1": {"ids": ["CHEBI:15365"]},
+                },
+                "paths": {
+                    "p0": {
+                        "constraints": [
+                            {"intermediate_categories": ["biolink:Gene"]},
+                            {"intermediate_categories": ["biolink:Drug"]},
+                        ]
+                    }
+                },
+            },
+            "multiple constraints",
+            id="multiple_constraints",
+        ),
+        pytest.param(
+            {
+                "nodes": {
+                    "n0": {"ids": ["MONDO:0005148"]},
+                    "n1": {"ids": ["CHEBI:15365"]},
+                },
+                "paths": {
+                    "p0": {
+                        "constraints": [
+                            {
+                                "intermediate_categories": [
+                                    "biolink:Gene",
+                                    "biolink:Drug",
+                                ]
+                            }
+                        ]
+                    }
+                },
+            },
+            "multiple intermediate categories",
+            id="multiple_intermediate_categories",
+        ),
+    ],
+)
+def test_unanswerable_query_graph_raises(mocker, qgraph, expected):
+    """Validation failures raise so run_task_lifecycle can fail the query.
+
+    They used to ``return message, 500``, a value the caller discarded -- so the
+    task was wrapped up as a success and no message was ever written to
+    ``response_id``, leaving the next worker to ``KeyError`` on it. Raising
+    routes the query to ``finish_query`` with an ERROR status instead.
+    """
+    _patch_query(mocker, {"message": {"query_graph": qgraph}})
+    save = mocker.patch("workers.arax_pathfinder.worker.save_message_sync")
+    search = mocker.patch("workers.arax_pathfinder.worker.execute_pathfinding")
+
+    with pytest.raises(ValueError, match=expected):
+        pf_worker.arax_pathfinder_task("query-3", "resp-3", LOGGER)
+
+    search.assert_not_called()
+    save.assert_not_called()
+
+
+def test_search_failure_propagates_instead_of_saving_error_blob(mocker):
+    """A failed search raises rather than persisting a non-TRAPI error blob.
+
+    The old code caught everything and saved ``{"status": "error", ...}`` to the
+    response id, then reported success -- so that blob flowed on through the
+    workflow. Now the exception reaches ``run_task_lifecycle``.
+    """
+    _patch_query(mocker)
+    mocker.patch(
+        "workers.arax_pathfinder.worker.execute_pathfinding",
+        side_effect=RuntimeError("sqlite is on fire"),
+    )
+    save = mocker.patch("workers.arax_pathfinder.worker.save_message_sync")
+
+    with pytest.raises(RuntimeError, match="sqlite is on fire"):
+        pf_worker.arax_pathfinder_task("query-4", "resp-4", LOGGER)
+
+    save.assert_not_called()
+
+
+def test_rehydrate_failure_propagates(mocker):
+    """A rehydrate failure fails the task too, rather than saving a partial KG."""
+    _patch_query(mocker)
+    mocker.patch(
+        "workers.arax_pathfinder.worker.execute_pathfinding",
+        return_value=copy.deepcopy(PATHS_RESULT),
+    )
+    mocker.patch(
+        "workers.arax_pathfinder.worker.rehydrate",
+        side_effect=RuntimeError("retriever unreachable"),
+    )
+    save = mocker.patch("workers.arax_pathfinder.worker.save_message_sync")
+
+    with pytest.raises(RuntimeError, match="retriever unreachable"):
+        pf_worker.arax_pathfinder_task("query-5", "resp-5", LOGGER)
+
+    save.assert_not_called()
+
+
+def test_descendants_are_memoized_per_child(mocker):
+    """The Biolink model is built once per pool child, not once per task.
+
+    ``BiolinkHelper`` construction parses the model, and the old code did it
+    inside every task alongside a fresh blocked-list HTTP fetch.
+    """
+    mocker.patch.object(pf_worker, "_biolink_helper", None)
+    mocker.patch.object(pf_worker, "_descendants_cache", {})
+    helper = MagicMock()
+    helper.get_descendants.return_value = ["biolink:Gene", "biolink:Protein"]
+    helper_cls = mocker.patch(
+        "workers.arax_pathfinder.worker.BiolinkHelper", return_value=helper
+    )
+
+    first = pf_worker.get_descendants("biolink:NamedThing")
+    second = pf_worker.get_descendants("biolink:NamedThing")
+
+    assert first == second == {"biolink:Gene", "biolink:Protein"}
+    helper_cls.assert_called_once()
+    helper.get_descendants.assert_called_once_with("biolink:NamedThing")
diff --git a/tests/unit/test_worker_dispatch_concurrency.py b/tests/unit/test_worker_dispatch_concurrency.py
index 5dbead9..faec548 100644
--- a/tests/unit/test_worker_dispatch_concurrency.py
+++ b/tests/unit/test_worker_dispatch_concurrency.py
@@ -24,6 +24,7 @@
 WORKER_FILES = [
     "workers/arax_rank/worker.py",
     "workers/aragorn_score/worker.py",
+    "workers/arax_pathfinder/worker.py",
 ]
 
 
diff --git a/workers/arax_pathfinder/Dockerfile b/workers/arax_pathfinder/Dockerfile
index 68ae2b6..890d662 100644
--- a/workers/arax_pathfinder/Dockerfile
+++ b/workers/arax_pathfinder/Dockerfile
@@ -5,6 +5,11 @@ FROM ghcr.io/translatorsri/renci-python-image:3.12.13
 LABEL org.opencontainers.image.source https://github.com/BioPack-team/shepherd
 
 ENV PYTHONHASHSEED=0
+# Enable faulthandler from interpreter startup so a SIGABRT dumps every thread's
+# traceback -- including during a spawned pool child's import phase, before the
+# pool initializer's faulthandler.enable() would run. Lets ProcessPoolManager's
+# SIGABRT-on-timeout capture a stuck child's stack even if it hangs in startup.
+ENV PYTHONFAULTHANDLER=1
 
 # set up requirements
 WORKDIR /app
diff --git a/workers/arax_pathfinder/worker.py b/workers/arax_pathfinder/worker.py
index b9fc501..e1e8e9f 100644
--- a/workers/arax_pathfinder/worker.py
+++ b/workers/arax_pathfinder/worker.py
@@ -8,35 +8,33 @@
 from pathlib import Path
 
 import httpx
-import requests
 from biolink_helper_pkg import BiolinkHelper
 from pathfinder.Pathfinder import Pathfinder
 
 from shepherd_utils.config import settings
+from shepherd_utils.cpu import resolve_pool_workers
 from shepherd_utils.data_download import (
     arax_pathfinder_sqlite_paths,
     ensure_arax_pathfinder_dbs,
+    ensure_http_files_dataset,
 )
 from shepherd_utils.db import (
-    get_message,
-    save_message,
+    get_message_sync,
+    save_message_sync,
 )
 from shepherd_utils.inject_shepherd_arax_provenance import (
     add_shepherd_arax_to_edge_sources,
 )
 from shepherd_utils.otel import setup_tracer
-from shepherd_utils.shared import (
-    get_tasks,
-    handle_task_failure,
-    wrap_up_task,
-)
+from shepherd_utils.process_pool import ProcessPoolManager
+from shepherd_utils.shared import get_tasks, run_task_lifecycle
 
 # Queue name
 STREAM = "arax.pathfinder"
 # Consumer group, most likely you don't need to change this.
 GROUP = "consumer"
 CONSUMER = str(uuid.uuid4())[:8]
-TASK_LIMIT = 100
+TASK_LIMIT = 10
 tracer = setup_tracer(STREAM)
 
 NUM_TOTAL_HOPS = 4
@@ -45,34 +43,78 @@
 PRUNE_TOP_K = 75
 NODE_DEGREE_THRESHOLD = 10000
 
-OUT_PATH = Path("general_concepts.json")
-
-
-def download_file(url: str, out_path: Path, overwrite: bool = False) -> Path:
-    out_path = Path(out_path)
-
-    if out_path.exists() and not overwrite:
-        return out_path
-
-    out_path.parent.mkdir(parents=True, exist_ok=True)
-
-    r = requests.get(url, timeout=60)
-    r.raise_for_status()
-
-    out_path.write_bytes(r.content)
-    return out_path
+# The ARAX blocked-concept list, fetched once at worker startup (see
+# poll_for_tasks) and read by each pool child. Kept in the working directory
+# (/app in the image) so it needs no extra volume mount.
+BLOCKED_LIST_DIR = "."
+BLOCKED_LIST_FILENAME = "general_concepts.json"
+BLOCKED_LIST_PATH = Path(BLOCKED_LIST_DIR) / BLOCKED_LIST_FILENAME
+
+BIOLINK_CACHE_DIR = "/tmp/biolink"
+
+REHYDRATE_TIMEOUT_SEC = 30.0
+
+# Per-child caches. Children are spawned once and reused for up to
+# ``settings.pool_max_tasks_per_child`` tasks, so the blocked list and the
+# Biolink model are parsed once per child instead of once per task (previously
+# every task re-read the JSON and rebuilt the BiolinkHelper, and re-fetched the
+# blocked list over HTTP). Only ``Pathfinder`` is still built per task, because
+# it takes the task's logger.
+_blocked_list_cache = None
+_biolink_helper = None
+_descendants_cache: dict = {}
+
+
+def ensure_blocked_list(logger: logging.Logger) -> None:
+    """Fetch the ARAX blocked-concept list if it isn't on disk yet.
+
+    Uses the shared downloader so the file lands via a temp file + atomic
+    rename; the previous per-task ``requests.get`` wrote the destination
+    directly, so concurrent tasks could race on a half-written file.
+    Idempotent, so it is safe to call at startup and again lazily in a child.
+    """
+    ensure_http_files_dataset(
+        name="arax_blocked_list",
+        target_dir=BLOCKED_LIST_DIR,
+        file_sources={BLOCKED_LIST_FILENAME: settings.arax_blocked_list_url},
+        logger=logger,
+    )
 
 
-def get_blocked_list():
-    download_file(settings.arax_blocked_list_url, OUT_PATH, False)
+def get_blocked_list(logger: logging.Logger):
+    """``(blocked_curies, blocked_synonyms)``, parsed once per pool child."""
+    global _blocked_list_cache
+    if _blocked_list_cache is None:
+        if not BLOCKED_LIST_PATH.exists():
+            # Startup fetch failed or this child outlived a wiped working dir.
+            ensure_blocked_list(logger)
+        with open(BLOCKED_LIST_PATH, "r") as file:
+            json_block_list = json.load(file)
+        synonyms = set(s.lower() for s in json_block_list["synonyms"])
+        _blocked_list_cache = (set(json_block_list["curies"]), synonyms)
+    return _blocked_list_cache
+
+
+def get_descendants(category: str):
+    """Biolink descendants of ``category``, memoized per pool child."""
+    global _biolink_helper
+    if category not in _descendants_cache:
+        if _biolink_helper is None:
+            Path(BIOLINK_CACHE_DIR).mkdir(parents=True, exist_ok=True)
+            _biolink_helper = BiolinkHelper(
+                settings.arax_biolink_version, BIOLINK_CACHE_DIR
+            )
+        _descendants_cache[category] = set(_biolink_helper.get_descendants(category))
+    return _descendants_cache[category]
 
-    with open(OUT_PATH, "r") as file:
-        json_block_list = json.load(file)
-    synonyms = set(s.lower() for s in json_block_list["synonyms"])
-    return set(json_block_list["curies"]), synonyms
 
+def rehydrate(kg, rehydrate_url, logger):
+    """POST the knowledge graph to the retriever and return the rehydrated one.
 
-async def rehydrate(kg, retriever_url, logger):
+    Synchronous because it runs inside the process-pool child alongside the
+    pathfinding, so the (potentially very large) knowledge graph never has to
+    be pickled back to the parent or re-encoded on its event loop.
+    """
     headers = {"Content-Type": "application/json", "Accept": "application/json"}
     payload = {
         "message": {"knowledge_graph": kg},
@@ -80,43 +122,79 @@ async def rehydrate(kg, retriever_url, logger):
     }
 
     try:
-        async with httpx.AsyncClient(timeout=30.0) as client:
-            res = await client.post(
-                retriever_url.replace("query", "rehydrate"),
-                headers=headers,
-                json=payload,
-            )
+        with httpx.Client(timeout=REHYDRATE_TIMEOUT_SEC) as client:
+            res = client.post(rehydrate_url, headers=headers, json=payload)
         res.raise_for_status()
         return res.json()["message"]["knowledge_graph"]
 
     except httpx.HTTPStatusError as http_err:
         logger.error(f"HTTP error occurred: {http_err}")
-        if res.text:
-            logger.error(f"Error details: {res.text}")
-        raise http_err
+        if http_err.response.text:
+            logger.error(f"Error details: {http_err.response.text}")
+        raise
     except httpx.ConnectError as conn_err:
         logger.error(f"Connection error occurred: {conn_err}")
-        raise conn_err
+        raise
     except httpx.TimeoutException as timeout_err:
         logger.error(f"Timeout error occurred: {timeout_err}")
-        raise timeout_err
+        raise
     except httpx.RequestError as req_err:
         logger.error(f"An unexpected error occurred: {req_err}")
-        raise req_err
+        raise
     except json.JSONDecodeError:
-        logger.error("Failed to parse the response as JSON.")
-        logger.error(f"Raw response: {res.text}")
+        logger.error("Failed to parse the rehydrate response as JSON.")
         raise
     except Exception as e:
         logger.error(f"An unexpected error occurred: {e}")
         raise e
 
 
-def execute_pathfinding_sync(
+def parse_query_graph(qgraph):
+    """Pull the pinned nodes and intermediate category out of the query graph.
+
+    Raises ``ValueError`` on anything Pathfinder can't answer. Raising (rather
+    than returning a status code, which the old caller discarded) lets
+    ``run_task_lifecycle`` record the failure on the span and route the query to
+    ``finish_query`` with an ERROR status, instead of continuing the workflow
+    with no response message ever written.
+    """
+    pinned_node_keys = []
+    pinned_node_ids = []
+    for node_key, node in qgraph["nodes"].items():
+        pinned_node_keys.append(node_key)
+        if node.get("ids", None) is not None:
+            pinned_node_ids.append(node["ids"][0])
+    if len(set(pinned_node_ids)) != 2:
+        raise ValueError("Pathfinder queries require two pinned nodes.")
+
+    intermediate_categories = []
+    path_key = next(iter(qgraph["paths"].keys()))
+    qpath = qgraph["paths"][path_key]
+    if (
+        qpath.get("constraints", None) is not None
+        and len(qpath.get("constraints", [])) > 0
+    ):
+        constraints = qpath["constraints"]
+        if len(constraints) > 1:
+            raise ValueError("Pathfinder queries do not support multiple constraints.")
+        if len(constraints) > 0:
+            intermediate_categories = (
+                constraints[0].get("intermediate_categories", None) or []
+            )
+        if len(intermediate_categories) > 1:
+            raise ValueError(
+                "Pathfinder queries do not support multiple intermediate categories"
+            )
+    else:
+        intermediate_categories = ["biolink:NamedThing"]
+
+    return pinned_node_keys, pinned_node_ids, intermediate_categories
+
+
+def execute_pathfinding(
     pinned_node_ids, pinned_node_keys, intermediate_categories, logger
 ):
-
-    blocked_curies, blocked_synonyms = get_blocked_list()
+    blocked_curies, blocked_synonyms = get_blocked_list(logger)
 
     curie_ngd_path, node_degree_path = arax_pathfinder_sqlite_paths()
     pathfinder_instance = Pathfinder(
@@ -128,13 +206,10 @@ def execute_pathfinding_sync(
         logger,
     )
 
-    biolink_cache_dir = "/tmp/biolink"
-    Path(biolink_cache_dir).mkdir(parents=True, exist_ok=True)
-    biolink_helper = BiolinkHelper(settings.arax_biolink_version, biolink_cache_dir)
-    descendants = set(biolink_helper.get_descendants(intermediate_categories[0]))
+    descendants = get_descendants(intermediate_categories[0])
 
     start = time.perf_counter()
-    logger.info("Starting pathfinder.get_paths() in worker thread")
+    logger.info("Starting pathfinder.get_paths()")
 
     result, aux_graphs, knowledge_graph = pathfinder_instance.get_paths(
         pinned_node_ids[0],
@@ -155,130 +230,131 @@ def execute_pathfinding_sync(
     return result, aux_graphs, knowledge_graph
 
 
-async def pathfinder(task, logger: logging.Logger):
+def arax_pathfinder_task(
+    query_id: str, response_id: str, logger: logging.Logger
+) -> None:
+    """Process-pool entrypoint: load, search, rehydrate, and save in the child.
+
+    Only the two small ids cross the process-pool boundary; the (potentially
+    very large) message is read from Redis, searched over, and written back
+    inside the child. That keeps the payload off the parent's heap and, more
+    importantly, keeps the graph search off the parent's event loop -- it used
+    to run via ``asyncio.to_thread``, where the GIL meant a handful of
+    concurrent searches could starve the heartbeat past HEARTBEAT_TTL_SEC and
+    get a live worker's tasks reclaimed out from under it (matching
+    aragorn_score / arax_rank).
+    """
     start = time.time()
-    query_id = task[1]["query_id"]
-    response_id = task[1]["response_id"]
-    message = await get_message(query_id, logger)
+    message = get_message_sync(query_id)
     parameters = message.get("parameters") or {}
     parameters["timeout"] = parameters.get("timeout", settings.lookup_timeout)
     parameters["tiers"] = parameters.get("tiers") or [0]
     message["parameters"] = parameters
 
-    qgraph = message["message"]["query_graph"]
-    pinned_node_keys = []
-    pinned_node_ids = []
-    for node_key, node in qgraph["nodes"].items():
-        pinned_node_keys.append(node_key)
-        if node.get("ids", None) is not None:
-            pinned_node_ids.append(node["ids"][0])
-    if len(set(pinned_node_ids)) != 2:
-        logger.error("Pathfinder queries require two pinned nodes.")
-        return message, 500
-
-    intermediate_categories = []
-    path_key = next(iter(qgraph["paths"].keys()))
-    qpath = qgraph["paths"][path_key]
-    if (
-        qpath.get("constraints", None) is not None
-        and len(qpath.get("constraints", [])) > 0
-    ):
-        constraints = qpath["constraints"]
-        if len(constraints) > 1:
-            logger.error("Pathfinder queries do not support multiple constraints.")
-            return message, 500
-        if len(constraints) > 0:
-            intermediate_categories = (
-                constraints[0].get("intermediate_categories", None) or []
-            )
-        if len(intermediate_categories) > 1:
-            logger.error(
-                "Pathfinder queries do not support multiple intermediate categories"
-            )
-            return message, 500
-    else:
-        intermediate_categories = ["biolink:NamedThing"]
+    pinned_node_keys, pinned_node_ids, intermediate_categories = parse_query_graph(
+        message["message"]["query_graph"]
+    )
 
     try:
-        result, aux_graphs, knowledge_graph = await asyncio.to_thread(
-            execute_pathfinding_sync,
+        result, aux_graphs, knowledge_graph = execute_pathfinding(
             pinned_node_ids,
             pinned_node_keys,
             intermediate_categories,
             logger,
         )
-        logger.info(f"Rehydrating knowledge graph with retriever")
-        knowledge_graph = await rehydrate(
-            knowledge_graph, settings.kg_rehydrate_url, logger
-        )
-
-        res = []
-        if result is not None:
-            res.append(
-                {
-                    "id": result["id"],
-                    "analyses": result["analyses"],
-                    "node_bindings": result["node_bindings"],
-                    "essence": "result",
-                }
-            )
-        if aux_graphs is None:
-            aux_graphs = {}
-        if knowledge_graph is None:
-            knowledge_graph = {}
-        message["message"]["knowledge_graph"] = knowledge_graph
-        message["message"]["auxiliary_graphs"] = aux_graphs
-        message["message"]["results"] = res
-
-        message = add_shepherd_arax_to_edge_sources(message)
-
-        await save_message(response_id, message, logger)
+        logger.info("Rehydrating knowledge graph with retriever")
+        knowledge_graph = rehydrate(knowledge_graph, settings.kg_rehydrate_url, logger)
     except Exception as e:
+        # Let the failure reach run_task_lifecycle, which records it on the span
+        # and routes the query to finish_query with an ERROR status. Previously
+        # this saved a non-TRAPI {"status": "error"} blob and reported success,
+        # so the bogus payload flowed on down the workflow.
         logger.error(
-            f"PathFinder failed to find paths between {pinned_node_keys[0]} and {pinned_node_keys[1]}. "
-            f"Error message is: {e}"
+            f"PathFinder failed to find paths between {pinned_node_keys[0]} and "
+            f"{pinned_node_keys[1]}. Error message is: {e}"
+        )
+        raise
+
+    res = []
+    if result is not None:
+        res.append(
+            {
+                "id": result["id"],
+                "analyses": result["analyses"],
+                "node_bindings": result["node_bindings"],
+                "essence": "result",
+            }
         )
-        message = {"status": "error", "error": str(e)}
-        await save_message(response_id, message, logger)
+    if aux_graphs is None:
+        aux_graphs = {}
+    if knowledge_graph is None:
+        knowledge_graph = {}
+    message["message"]["knowledge_graph"] = knowledge_graph
+    message["message"]["auxiliary_graphs"] = aux_graphs
+    message["message"]["results"] = res
+
+    message = add_shepherd_arax_to_edge_sources(message)
 
+    save_message_sync(response_id, message)
     logger.info(f"Task took {time.time() - start}")
 
 
-async def process_task(task, parent_ctx, logger: logging.Logger, limiter):
-    """Process a given task and ACK in redis."""
-    start = time.time()
-    span = tracer.start_span(STREAM, context=parent_ctx)
-    try:
-        await pathfinder(task, logger)
-        # Always wrap up the task to ACK it in the broker
-        try:
-            await wrap_up_task(STREAM, GROUP, task, logger)
-        except Exception as e:
-            logger.error(f"Task {task[0]}: Failed to wrap up task: {e}")
-    except asyncio.CancelledError:
-        logger.warning(f"Task {task[0]} was cancelled")
-    except Exception as e:
-        logger.error(f"Task {task[0]} failed with unhandled error: {e}", exc_info=True)
-        await handle_task_failure(STREAM, GROUP, task, logger)
-    finally:
-        span.end()
-        limiter.release()
-        logger.info(f"Finished task {task[0]} in {time.time() - start}")
+async def process_task(task, parent_ctx, logger, limiter, loop, pool):
+    """Process a given task and ACK in redis.
+
+    Pathfinding is CPU-bound, so it is dispatched to a process pool while the
+    span, wrap-up, and error handling are shared with every worker. Only the
+    ids are handed to the child; the message load/save happen there (see
+    ``arax_pathfinder_task``) so the payload never crosses the process boundary.
+
+    Dispatch goes through ``pool`` (a ProcessPoolManager) so a child dying on an
+    oversized message replaces the pool instead of poisoning it for good, and a
+    search that runs away is time-bounded rather than holding its slot forever.
+    """
+
+    async def _run(task, logger):
+        await pool.run(
+            loop,
+            arax_pathfinder_task,
+            task[1]["query_id"],
+            task[1]["response_id"],
+            logger,
+        )
+
+    await run_task_lifecycle(STREAM, GROUP, task, parent_ctx, logger, limiter, _run)
 
 
 async def poll_for_tasks():
     """On initialization, poll indefinitely for available tasks."""
+    startup_logger = logging.getLogger(STREAM)
     # Ensure the two sqlite databases exist before any task tries to open them
     # (a first-run local `docker compose up` starts with the volume-mounted
     # directory empty). No-op once present or when no scp source is configured
     # (e.g. production, where the data is mounted out of band).
-    ensure_arax_pathfinder_dbs(logging.getLogger(STREAM))
+    ensure_arax_pathfinder_dbs(startup_logger)
+    # Fetch the blocked-concept list once here rather than per task, so the pool
+    # children only ever read it.
+    ensure_blocked_list(startup_logger)
+    loop = asyncio.get_running_loop()
+    # Size the pool by the pod's actual CPU allocation (cgroup limit), not
+    # os.cpu_count() -- see aragorn_score.poll_for_tasks. Each child loads a full
+    # message, so this also bounds peak memory. POOL_MAX_WORKERS overrides.
+    max_workers = resolve_pool_workers(TASK_LIMIT, startup_logger)
+    logging.info(f"{STREAM}: process pool sized to {max_workers} worker(s).")
+    pool = ProcessPoolManager(
+        max_workers,
+        max_tasks_per_child=settings.pool_max_tasks_per_child,
+        name="arax.pathfinder process pool",
+        task_timeout=settings.pool_task_timeout_sec,
+    )
     while True:
         try:
             async for task, parent_ctx, logger, limiter in get_tasks(
-                STREAM, GROUP, CONSUMER, TASK_LIMIT
+                STREAM, GROUP, CONSUMER, max_workers
             ):
-                asyncio.create_task(process_task(task, parent_ctx, logger, limiter))
+                asyncio.create_task(
+                    process_task(task, parent_ctx, logger, limiter, loop, pool)
+                )
         except asyncio.CancelledError:
             logging.info("Poll loop cancelled, shutting down.")
         except Exception as e:

From 78d17afc331c7dda16244beed3405477dc4884c3 Mon Sep 17 00:00:00 2001
From: Max Wang <max@covar.com>
Date: Mon, 10 Aug 2026 15:17:50 -0400
Subject: [PATCH 30/32] Add script for testing against pathfinder

---
 scripts/test_pathfinder.py | 140 +++++++++++++++++++++++++++++++++++++
 1 file changed, 140 insertions(+)
 create mode 100644 scripts/test_pathfinder.py

diff --git a/scripts/test_pathfinder.py b/scripts/test_pathfinder.py
new file mode 100644
index 0000000..1e574b0
--- /dev/null
+++ b/scripts/test_pathfinder.py
@@ -0,0 +1,140 @@
+import asyncio
+import json
+import time
+from datetime import datetime
+from pathlib import Path
+
+import httpx
+
+target_urls = {
+    "aragorn-ci": "https://shepherd.ci.transltr.io/aragorn",
+    "arax-ci": "https://shepherd.ci.transltr.io/arax",
+    "aragorn-dev": "https://shepherd.renci.org/aragorn",
+    "arax-dev": "https://shepherd.renci.org/arax",
+    "aragorn-local": "http://localhost:5439/aragorn",
+    "arax-local": "http://localhost:5439/arax",
+    "bte-local": "http://localhost:5439/bte",
+}
+
+RESPONSES_DIR = "responses"
+
+
+def generate_query(curie1: str, curie2: str) -> dict:
+    """Given a curie, return a TRAPI message."""
+    parameters = {
+        # "timeout": 300,
+        # "tiers": [0],
+    }
+    return {
+      "message": {
+          "query_graph": {
+              "nodes": {
+                  "on": {
+                      "constraints": [],
+                      "ids": [
+                          curie1
+                      ],
+                  },
+                  "sn": {
+                      "constraints": [],
+                      "ids": [
+                          curie2
+                      ],
+                  }
+              },
+              "paths": {
+                  "p0": {
+                      "object": "on",
+                      "subject": "sn",
+                      "predicates": [
+                          "biolink:related_to",
+                      ]
+                  }
+              }
+          }
+      },
+      "parameters": parameters,
+      "log_level": "DEBUG",
+  }
+
+
+async def single_lookup(curies: tuple[str, str], target: str):
+    """Run a single query lookup synchronously."""
+    query = generate_query(curies[0], curies[1])
+    start_time = datetime.now()
+    try:
+        async with httpx.AsyncClient(timeout=600000) as client:
+            response = await client.post(
+                f"{target_urls[target]}/query",
+                json=query,
+            )
+            response.raise_for_status()
+            response_json = response.json()
+            results = (response_json.get("message") or {}).get("results") or []
+            num_results = len(results)
+            assert num_results == 1
+            num_analyses = len(results[0]["analyses"])
+    except Exception as e:
+        num_results = 0
+        num_analyses = 0
+        response_json = {
+            "Error": str(e),
+        }
+
+    stop_time = datetime.now()
+    print(f"{curies[0]}->{curies[1]} took {stop_time - start_time} seconds and gave {num_analyses} results")
+    out_dir = Path(RESPONSES_DIR) / "pathfinder" / target
+    out_dir.mkdir(parents=True, exist_ok=True)
+    response_path = out_dir / f"{('_').join(curies[0].split(':'))}_{('_').join(curies[1].split(':'))}_response.json"
+    with response_path.open("w", encoding="utf-8") as f:
+        json.dump(response_json, f, indent=2)
+
+
+query_list = [
+    ('MONDO:0021095', 'MONDO:0005105'),
+    ('CHEBI:9139', 'MONDO:0004975'),
+    ('CHEBI:5118', 'MONDO:0100233'),
+    ('MONDO:0005180', 'MONDO:0005105'),
+    ('MONDO:0019632', 'MONDO:0005340'),
+    ('CHEBI:27881', 'NCBIGene:2739'),
+    ('CHEBI:45783', 'MONDO:0004979'),  # Imatinib -> Asthma
+    ('GO:0006914', 'MONDO:0005265'),
+    ('NCBIGene:3458', 'CHEBI:16828'),
+    ('MONDO:0005532', 'MONDO:0005180'),
+    ('CHEBI:15647', 'UNII:31YO63LBSN'),
+    ('CHEBI:28364', 'MONDO:0005311'),
+    ('NCBIGene:3458', 'MONDO:0100096'),
+    ('NCBIGene:27240', 'MONDO:0100096'),
+    ('CHEBI:3750', 'MONDO:0013209'),
+    ('CHEBI:83766', 'MONDO:0008170'),
+    ('CHEBI:45783', 'MONDO:0004784'),
+    ('UNII:7SE5582Q2P', 'MONDO:0007037'),
+    ('MONDO:0005011', 'MONDO:0005180'),
+    ('CHEBI:15365', 'MONDO:0005575'),
+    ('CHEBI:50924', 'MONDO:0007256'),
+    ('CHEBI:45713', 'NCBIGene:2739'),
+    ('NCBIGene:54716', 'MONDO:0100096'),
+    ('CHEBI:7465', 'MONDO:0008218'),
+    # ('CHEBI:10033', 'MONDO:0004992'),  # Warfarin -> Cancer, DON'T RUN
+]
+
+
+async def main():
+    """Run the given query and time it."""
+    targets = ["arax-local"]
+    runs_per_target = 1
+
+    start = time.time()
+    queries = []
+    for curies in query_list:
+        queries.extend([
+            single_lookup(curies, target)
+            for target in targets
+            for _ in range(runs_per_target)
+        ])
+    await asyncio.gather(*queries)
+    print(f"\nAll queries took {time.time() - start:.2f} seconds")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

From a782c5fd514f6aa5929aee9eabaeed4614208c2e Mon Sep 17 00:00:00 2001
From: Max Wang <max@covar.com>
Date: Mon, 10 Aug 2026 15:20:19 -0400
Subject: [PATCH 31/32] Set arax pathfinder resources

---
 compose.test.yml | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/compose.test.yml b/compose.test.yml
index 0bb7690..1db188b 100644
--- a/compose.test.yml
+++ b/compose.test.yml
@@ -64,6 +64,9 @@ services:
   arax:
     cpus: 1
     mem_limit: 3g
+  arax_pathfinder:
+    cpus: 6
+    mem_limit: 22g
   arax_rank:
     cpus: 4
     mem_limit: 10g

From 6fbd1833add254d3bca59e1e3209410f1b6ab120 Mon Sep 17 00:00:00 2001
From: Max Wang <max@covar.com>
Date: Wed, 12 Aug 2026 15:16:22 -0400
Subject: [PATCH 32/32] Add arax pathfinder to release

---
 .github/workflows/release.yml | 1 +
 1 file changed, 1 insertion(+)

diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml
index e38c517..307f9eb 100644
--- a/.github/workflows/release.yml
+++ b/.github/workflows/release.yml
@@ -49,6 +49,7 @@ jobs:
           - aragorn_pathfinder
           - aragorn_score
           - arax
+          - arax_pathfinder
           - arax_rank
           - bte
           - bte_lookup