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/21] 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/21] 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/21] 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/21] 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/21] 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/21] 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/21] 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/21] 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/21] 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/21] 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/21] 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/21] 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/21] 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/21] 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/21] 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/21] 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/21] 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/21] 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/21] 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/21] 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/21] 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(