diff --git a/src/seocho/index/pipeline.py b/src/seocho/index/pipeline.py index 6fbb26bf..e5011fe2 100644 --- a/src/seocho/index/pipeline.py +++ b/src/seocho/index/pipeline.py @@ -908,11 +908,16 @@ def _shape_and_write_graph( domain_persisted = int(summary.get("relationships_created", 0) or 0) try: _rows = self.graph_store.query( - "MATCH (a)-[r]->(b) WHERE a._workspace_id=$ws AND r._source_id=$sid " + "MATCH (a)-[r]->(b) WHERE a._workspace_id=$workspace_id " + "AND b._workspace_id=$workspace_id AND r._workspace_id=$workspace_id " + "AND r._source_id=$sid " "AND NOT type(r) IN $prov RETURN count(r) AS c", - params={"ws": self.workspace_id, "sid": source_id, + params={"workspace_id": self.workspace_id, "sid": source_id, "prov": list(self._PROVENANCE_REL_TYPES)}, - database=database) + database=database, + workspace_id=self.workspace_id, + enforce_workspace_filter=True, + ) if _rows: domain_persisted = int((_rows[0].get("c") if _rows[0] else 0) or 0) except Exception: # noqa: BLE001 - census must never fail the write diff --git a/src/seocho/local_engine.py b/src/seocho/local_engine.py index f8cef842..c61d029f 100644 --- a/src/seocho/local_engine.py +++ b/src/seocho/local_engine.py @@ -1624,6 +1624,8 @@ def _graph_chunk_fallback(self, question: str, database: str) -> str: "LIMIT $k", params={"workspace_id": self.workspace_id, "kws": kws, "k": 3}, database=database, + workspace_id=self.workspace_id, + enforce_workspace_filter=True, ) except Exception: return "" diff --git a/src/seocho/session.py b/src/seocho/session.py index 742416eb..9d3090cc 100644 --- a/src/seocho/session.py +++ b/src/seocho/session.py @@ -795,9 +795,12 @@ def _add_via_pipeline( try: sid = memory.memory_id rows = self.graph_store.query( - "MATCH (n) WHERE n._source_id = $sid " + "MATCH (n) WHERE n._source_id = $sid AND n._workspace_id = $workspace_id " "RETURN labels(n)[0] AS label, n.name AS name, properties(n) AS props", - params={"sid": sid}, database=database, + params={"sid": sid, "workspace_id": self.workspace_id}, + database=database, + workspace_id=self.workspace_id, + enforce_workspace_filter=True, ) for row in rows: extracted_nodes.append({ diff --git a/tests/seocho/test_chunk_fallback.py b/tests/seocho/test_chunk_fallback.py index 6661f34d..7e91736a 100644 --- a/tests/seocho/test_chunk_fallback.py +++ b/tests/seocho/test_chunk_fallback.py @@ -19,7 +19,9 @@ def _engine(fake_query): eng.workspace_id = "ws-cf" class _GS: - def query(self, cypher, params=None, database="neo4j"): + def query(self, cypher, params=None, database="neo4j", **kwargs): + assert kwargs["workspace_id"] == eng.workspace_id + assert kwargs["enforce_workspace_filter"] is True return fake_query(cypher, params or {}, database) eng.graph_store = _GS() diff --git a/tests/seocho/test_ontology_context.py b/tests/seocho/test_ontology_context.py index 29df0d11..2a853580 100644 --- a/tests/seocho/test_ontology_context.py +++ b/tests/seocho/test_ontology_context.py @@ -83,7 +83,7 @@ def __init__(self) -> None: super().__init__() self.queries = [] - def query(self, cypher: str, *, params=None, database="neo4j"): # noqa: ANN001 + def query(self, cypher: str, *, params=None, database="neo4j", **kwargs): # noqa: ANN001 self.queries.append(cypher) return [ { diff --git a/tests/seocho/test_workspace_read_callers.py b/tests/seocho/test_workspace_read_callers.py new file mode 100644 index 00000000..33372a51 --- /dev/null +++ b/tests/seocho/test_workspace_read_callers.py @@ -0,0 +1,64 @@ +"""Regression checks for workspace-scoped indexing and session fallback reads.""" + +from typing import Any + +from seocho.index.pipeline import IndexingPipeline +from seocho.models import Memory +from seocho.ontology import NodeDef, Ontology, P +from seocho.session import Session + + +def test_relationship_census_requires_owned_endpoints_and_relationship() -> None: + class Store: + def write(self, nodes: list, relationships: list, **kwargs: Any) -> dict: + return {"nodes_created": len(nodes), "relationships_created": 3} + + def query(self, cypher: str, **kwargs: Any) -> list[dict]: + assert kwargs["workspace_id"] == "tenant-a" + assert kwargs["enforce_workspace_filter"] is True + assert kwargs["params"]["workspace_id"] == "tenant-a" + assert kwargs["params"]["sid"] == "shared-source" + assert "r._source_id=$sid" in cypher + for entity in ("a", "b", "r"): + assert f"{entity}._workspace_id=$workspace_id" in cypher + return [{"c": 1}] + + pipeline = IndexingPipeline( + ontology=Ontology( + name="scoped", nodes={"Company": NodeDef(properties={"name": P(str)})}, + relationships={}, + ), + graph_store=Store(), llm=None, workspace_id="tenant-a", + ) + result = pipeline.index_graph( + {"nodes": [{"id": "acme", "label": "Company", "properties": {"name": "Acme"}}], + "relationships": []}, + source_id="shared-source", + ) + assert result.relationship_census["written_to_store"] == 1 + + +def test_session_fallback_reads_only_the_session_workspace(monkeypatch: Any) -> None: + class Pipeline: + def add(self, *args: Any, **kwargs: Any) -> Memory: + return Memory( + memory_id="shared-source", workspace_id="tenant-a", content="Acme", + metadata={}, status="active", database="neo4j", category="general", + ) + + class Store: + def query(self, cypher: str, **kwargs: Any) -> list[dict]: + assert kwargs["workspace_id"] == "tenant-a" + assert kwargs["enforce_workspace_filter"] is True + assert kwargs["params"] == {"sid": "shared-source", "workspace_id": "tenant-a"} + assert "n._workspace_id = $workspace_id" in cypher + return [{"label": "Company", "props": {"name": "Acme"}}] + + session = Session( + ontology=Ontology(name="scoped", nodes={}, relationships={}), + graph_store=Store(), llm=None, workspace_id="tenant-a", + ) + monkeypatch.setattr(session, "_get_pipeline_engine", lambda: Pipeline()) + result = session._add_via_pipeline("Acme", "neo4j", "general", None) + assert result["extracted_nodes"] == [{"label": "Company", "properties": {"name": "Acme"}}] + session.close()