Merge pull request #1250 from RoboFinSystems/bugfix/parquet-path-reads

Sweep the remaining bare-path pyarrow reads; bump the demo client floor

Author: jfrench9

.gitignore

@@ -147,10 +147,5 @@ robosystems/adapters/sec/arelle/cache/
 
 # dbt per-machine state (anonymous usage id)
 **/dbt/.user.yml
-# Collateral of the line above: ArelleClient._setup_cache_directory sets the
-# process-global XDG_CACHE_HOME to robosystems/adapters/sec so Arelle resolves
-# its own cache, and every other XDG-respecting library in that process then
-# caches there too. huggingface_hub is the one that shows up locally
-# (.agent_harnesses.json + xet/logs). Ignore the whole tree, not one file —
-# the next such library will land here as well.
+# HF cache — ArelleClient points XDG_CACHE_HOME at this directory's parent
 robosystems/adapters/sec/huggingface/

pyproject.toml

@@ -37,9 +37,6 @@ dependencies = [
     "sse-starlette>=3.3.0,<4.0",
     "strawberry-graphql[fastapi]>=0.314.0,<1.0",
     "uvicorn>=0.48.0,<1.0",
-    # uvloop ships no Windows wheels and its sdist refuses to build there, so an
-    # unconditional dependency makes `uv sync` fail outright on Windows hosts.
-    # Mirrors the marker uvicorn[standard] already carries for its own uvloop dep.
     "uvloop>=0.21.0,<1.0; sys_platform != 'win32'",
     # Cloud Service (AWS)
     "boto3>=1.42.0,<2.0",
@@ -55,9 +52,7 @@ dependencies = [
     "redis>=8.0.1,<9.0",
     # Document Search (OpenSearch)
     "opensearch-py>=3.0.0,<4.0",
-    # Data Orchestration (Dagster) — the dagster family is version-coupled
-    # (core/webserver on 1.x.y, libraries on the matching 0.(x+16).y);
-    # bump all floors together, never one package at a time
+    # Data Orchestration (Dagster)
     "dagster>=1.13.15,<2.0",
     "dagster-webserver>=1.13.15,<2.0",
     "dagster-postgres>=0.29.15,<1.0",
@@ -77,9 +72,7 @@ dependencies = [
     "intuit-oauth>=1.2.0,<2.0",
     "python-quickbooks>=0.9.0,<1.0",
     "stripe>=11.1.0,<12.0",
-    # Observability — the opentelemetry-* family versions are coupled
-    # (api/sdk/exporters track 1.x, instrumentations track the matching 0.xxb0);
-    # bump all floors together, never one package at a time
+    # Observability
     "opentelemetry-api>=1.44.0,<2.0",
     "opentelemetry-exporter-otlp>=1.44.0,<2.0",
     "opentelemetry-sdk>=1.44.0,<2.0",
@@ -89,13 +82,9 @@ dependencies = [
     "opentelemetry-instrumentation-requests>=0.65b0,<1.0",
     # Utilities
     "aiohttp>=3.14.3",
-    "authlib>=1.6.5,<2.0",  # OIDC login: PKCE + token-endpoint client (id_token validation rides pyjwt)
+    "authlib>=1.6.5,<2.0",
     "bcrypt>=5.0.0,<6.0",
     "beautifulsoup4>=4.13.0,<5.0",
-    # Direct dependency: auth cache, credential encryption, and connection
-    # credentials import it at module level. Without this pin it reaches the
-    # prod closure only transitively (intuit-oauth → pyjwt[crypto]), so
-    # dropping that adapter would break the auth path at import time.
     "cryptography>=50.0.0,<51.0",
     "email-validator>=2.2.0,<3.0",
     "holidays>=0.101,<1.0",
@@ -113,19 +102,14 @@ dependencies = [
     "requests>=2.34.2,<3.0",
     "retrying>=1.4.0,<2.0",
     "uuid6>=2025.0.0",
-    # Passkey MFA: RP-side WebAuthn ceremony generation/verification
-    # (py_webauthn; cbor2 rides transitively)
     "webauthn>=2.7.0,<3.0",
     "pyshacl>=0.26,<0.31",
 ]
 
 [project.optional-dependencies]
 dev = [
-    # RoboSystems client for demo and testing. Ranged, not pinned: the demos
-    # are the dogfooding surface and should exercise the current client, and
-    # uv.lock is what makes a given checkout reproducible. <2 is the SDK's
-    # post-1.0 semver boundary.
-    "robosystems-client>=1.8.1,<2.0",
+    # RoboSystems client for demo and testing.
+    "robosystems-client>=1.11.0,<2.0",
 
     # Testing framework
     "pytest>=9.0.0,<10.0",
@@ -258,18 +242,6 @@ skip-magic-trailing-comma = false
 line-ending = "auto"
 
 [tool.basedpyright]
-# Include every Python-bearing dir that ships code:
-# - ``robosystems`` and ``main.py``: core platform
-# - ``examples``: demo scripts (good-faith use of real SDK paths)
-# - ``bin``: Lambda functions (rotation handlers, volume monitors —
-#   production code, type-check is load-bearing)
-# The project's rule suppressions below cover patterns demo and infra
-# scripts use (untyped SDK / boto3 responses); inclusion fixes the IDE
-# experience without adding CI noise.
-# ``tests`` stays excluded — 600+ files of mock-heavy code that produce
-# manageable runtime correctness via the test suite itself, and the
-# basedpyright noise outweighs the catch rate. ``migrations`` stays
-# excluded (autogenerated SQL).
 include = ["robosystems", "main.py", "examples", "bin"]
 exclude = ["**/__pycache__", ".venv", "robosystems/adapters/sec/arelle", "tests"]
 extraPaths = ["."]

robosystems/adapters/sec/enrichment.py

@@ -360,7 +360,8 @@ def _load_element_knowledge(self) -> dict[str, dict] | None:
 
       import pyarrow.parquet as pq
 
-      table = pq.read_table(path)
+      with open(path, "rb") as f:
+        table = pq.read_table(f)
       result = {}
 
       # Read columnar and transpose to row dicts
@@ -383,7 +384,7 @@ def _load_element_knowledge(self) -> dict[str, dict] | None:
       logger.info(f"Loaded element knowledge artifact: {len(result)} elements")
       return result
     except Exception as e:
-      logger.debug(f"Element knowledge artifact not available: {e}")
+      logger.warning(f"Element knowledge artifact failed to load: {e}")
       return None
 
   def _load_structure_profiles(self) -> dict[str, dict[str, float]] | None:
@@ -396,7 +397,8 @@ def _load_structure_profiles(self) -> dict[str, dict[str, float]] | None:
 
       import pyarrow.parquet as pq
 
-      table = pq.read_table(path)
+      with open(path, "rb") as f:
+        table = pq.read_table(f)
       columns = table.to_pydict()
 
       result: dict[str, dict[str, float]] = {}
@@ -408,7 +410,7 @@ def _load_structure_profiles(self) -> dict[str, dict[str, float]] | None:
       logger.info(f"Loaded structure profiles artifact: {len(result)} types")
       return result
     except Exception as e:
-      logger.debug(f"Structure profiles artifact not available: {e}")
+      logger.warning(f"Structure profiles artifact failed to load: {e}")
       return None
 
   def _load_structure_consensus(self) -> dict[str, dict] | None:
@@ -421,7 +423,8 @@ def _load_structure_consensus(self) -> dict[str, dict] | None:
 
       import pyarrow.parquet as pq
 
-      table = pq.read_table(path)
+      with open(path, "rb") as f:
+        table = pq.read_table(f)
       columns = table.to_pydict()
 
       result = {}
@@ -435,7 +438,7 @@ def _load_structure_consensus(self) -> dict[str, dict] | None:
       logger.info(f"Loaded structure consensus artifact: {len(result)} entries")
       return result
     except Exception as e:
-      logger.debug(f"Structure consensus artifact not available: {e}")
+      logger.warning(f"Structure consensus artifact failed to load: {e}")
       return None
 
   @property
@@ -462,7 +465,8 @@ def _load_disclosure_profiles(self) -> dict[str, dict[str, float]] | None:
 
       import pyarrow.parquet as pq
 
-      table = pq.read_table(path)
+      with open(path, "rb") as f:
+        table = pq.read_table(f)
       columns = table.to_pydict()
 
       result: dict[str, dict[str, float]] = {}
@@ -474,7 +478,7 @@ def _load_disclosure_profiles(self) -> dict[str, dict[str, float]] | None:
       logger.info(f"Loaded disclosure profiles artifact: {len(result)} types")
       return result
     except Exception as e:
-      logger.debug(f"Disclosure profiles artifact not available: {e}")
+      logger.warning(f"Disclosure profiles artifact failed to load: {e}")
       return None
 
   def _load_disclosure_consensus(self) -> dict[str, dict] | None:
@@ -487,7 +491,8 @@ def _load_disclosure_consensus(self) -> dict[str, dict] | None:
 
       import pyarrow.parquet as pq
 
-      table = pq.read_table(path)
+      with open(path, "rb") as f:
+        table = pq.read_table(f)
       columns = table.to_pydict()
 
       result = {}
@@ -501,7 +506,7 @@ def _load_disclosure_consensus(self) -> dict[str, dict] | None:
       logger.info(f"Loaded disclosure consensus artifact: {len(result)} entries")
       return result
     except Exception as e:
-      logger.debug(f"Disclosure consensus artifact not available: {e}")
+      logger.warning(f"Disclosure consensus artifact failed to load: {e}")
       return None
 
   # -- Embedding ------------------------------------------------------------

robosystems/adapters/sec/knowledge/artifact.py

@@ -581,23 +581,15 @@ def _load_pagerank_scores(self) -> dict[str, float]:
       return {}
 
     try:
-      # Read through a handle. This module imports networkit at import time, and
-      # in a process that has done so `pq.read_table(path)` raises ArrowKeyError
-      # from the LocalFileSystem constructor (icebug bundles a second libarrow
-      # that collides with pyarrow's on Arrow's global filesystem registry — see
-      # this package's __init__). Passing a path here meant this function could
-      # never succeed: it raised on every call, the except below swallowed it at
-      # debug level, and disclosure profiles silently fell back to frequency-only
-      # weighting. The writers in this file already take the handle path.
+      # Handle, not a path — see this package's __init__.
       with open(path, "rb") as f:
         table = pq.read_table(f, columns=["qname", "pagerank"])
       cols = table.to_pydict()
       return {
         cols["qname"][i]: cols["pagerank"][i] or 0.0 for i in range(len(cols["qname"]))
       }
     except Exception as e:
-      # Warning, not debug: the empty dict degrades weighting silently, so a
-      # recurring failure here has to be visible in logs to be noticed at all.
+      # Warning, not debug: the empty dict degrades weighting silently.
       logger.warning(f"Failed to load PageRank scores from {path}: {e}")
       return {}
 

robosystems/adapters/sec/processors/consolidation.py

@@ -154,10 +154,12 @@ def consolidate_parquet_from_disk(
   tables = []
   for pq_file in parquet_files:
     try:
-      table = pq.read_table(pq_file)
+      # Handle, not a path — see adapters/sec/knowledge/__init__.py.
+      with open(pq_file, "rb") as f:
+        table = pq.read_table(f)
       tables.append(table)
     except Exception as e:
-      logger.warning("Skipping corrupted parquet file %s: %s", pq_file, e)
+      logger.warning("Skipping unreadable parquet file %s: %s", pq_file, e)
       continue
 
   if not tables:

tests/adapters/sec/test_artifact_reads_under_networkit.py

@@ -0,0 +1,94 @@
+"""Artifact loaders must work in a process that has imported networkit.
+
+icebug bundles a second libarrow beside pyarrow's; the two collide on Arrow's
+global filesystem registry, so once networkit is loaded `pq.read_table(path)`
+raises ArrowKeyError forever. The knowledge builders import networkit, so any
+loader reading their output has to take a file handle. This test poisons the
+registry the same way and asserts the loaders still return data — the previous
+tests all mocked the read, so a bare path passed them and failed in production.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import networkit  # noqa: F401  — poisons the Arrow registry, exactly as prod does
+import pyarrow as pa
+import pyarrow.parquet as pq
+import pytest
+
+from robosystems.adapters.sec.enrichment import SemanticEnricher
+
+pytestmark = pytest.mark.unit
+
+
+def _write(path: Path, columns: dict) -> None:
+  with open(path, "wb") as f:
+    pq.write_table(pa.table(columns), f)
+
+
+@pytest.fixture
+def artifacts(tmp_path, monkeypatch):
+  _write(
+    tmp_path / "element_knowledge.parquet",
+    {
+      "qname": ["us-gaap:Assets"],
+      "primary_statement": ["BalanceSheet"],
+      "bfs_depth": [1],
+      "pagerank": [0.9],
+      "core_number": [2],
+      "neighborhood_agreement": [0.5],
+      "filing_count": [3],
+      "disclosure_type": ["x"],
+    },
+  )
+  _write(
+    tmp_path / "structure_profiles.parquet",
+    {
+      "canonical_type": ["BalanceSheet"],
+      "qname": ["us-gaap:Assets"],
+      "frequency": [0.8],
+      "structure_count": [4],
+    },
+  )
+  monkeypatch.setattr(
+    "robosystems.config.storage.shared.get_artifact_path",
+    lambda name: str(tmp_path / f"{name}.parquet"),
+  )
+  return tmp_path
+
+
+def _enricher() -> SemanticEnricher:
+  # __new__, not __init__: construction loads an embedding model, and the read
+  # path under test does not need one.
+  return SemanticEnricher.__new__(SemanticEnricher)
+
+
+def test_element_knowledge_loads(artifacts):
+  result = _enricher()._load_element_knowledge()
+  assert result is not None, "bare-path read — ArrowKeyError swallowed to None"
+  assert result["us-gaap:Assets"]["pagerank"] == 0.9
+
+
+def test_structure_profiles_load(artifacts):
+  result = _enricher()._load_structure_profiles()
+  assert result is not None, "bare-path read — ArrowKeyError swallowed to None"
+  assert result["BalanceSheet"]["us-gaap:Assets"] == pytest.approx(0.8)
+
+
+def test_consolidation_reads_parquet_from_disk(tmp_path):
+  """The consolidation read attributes any failure to file corruption, so a
+  registry collision there discards real rows under a misleading label."""
+  from robosystems.adapters.sec.processors.consolidation import (
+    consolidate_parquet_from_disk,
+  )
+
+  table_dir = tmp_path / "nodes/Entity"
+  table_dir.mkdir(parents=True)
+  _write(table_dir / "a.parquet", {"identifier": ["e1"], "name": ["Acme"]})
+
+  out = consolidate_parquet_from_disk(tmp_path, "nodes/Entity")
+  assert out is not None, "every file read as corrupt — rows silently dropped"
+  from io import BytesIO
+
+  assert pq.read_table(BytesIO(out)).num_rows == 1