fix(reads): enforce workspace scope in session fallback and relationship census

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

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 ""

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({

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()

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 [
             {

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()