Merge pull request #329 from Hawksight-AI/context

Context Graph Reliability Hardening: Policy Applicability + Cross-System Capture

semantica/context/agent_context.py

@@ -1957,17 +1957,50 @@ class AgentContext:
         Returns:
             Cross-system context
         """
-        # This is a placeholder for cross-system context capture
-        # In practice, this would integrate with various systems
         context = {}
-        
+
         for system in systems:
-            context[system] = {
+            captured_at = datetime.now().isoformat()
+            payload: Dict[str, Any] = {
                 "entity_id": entity_id,
                 "system_name": system,
-                "captured_at": datetime.now().isoformat(),
-                "status": "captured"
+                "captured_at": captured_at,
             }
+
+            try:
+                # GraphStore-backed capture path
+                if self.knowledge_graph and hasattr(self.knowledge_graph, "execute_query"):
+                    query = """
+                    MATCH (c:CrossSystemContext {system_name: $system_name})
+                    WHERE c.context_data IS NOT NULL
+                    RETURN c
+                    ORDER BY c.created_at DESC
+                    LIMIT 5
+                    """
+                    result = self.knowledge_graph.execute_query(
+                        query, {"system_name": system}
+                    )
+                    records = result.get("records", []) if isinstance(result, dict) else result
+                    payload["status"] = "captured"
+                    payload["records_found"] = len(records) if isinstance(records, list) else 0
+                    payload["records"] = records if isinstance(records, list) else []
+                else:
+                    payload["status"] = "captured_without_backend"
+                    payload["records_found"] = 0
+                    payload["records"] = []
+            except Exception as e:
+                self.logger.warning(
+                    "Cross-system input capture failed for system=%s entity_id=%s: %s",
+                    system,
+                    entity_id,
+                    str(e),
+                )
+                payload["status"] = "capture_failed"
+                payload["error"] = "internal_capture_error"
+                payload["records_found"] = 0
+                payload["records"] = []
+
+            context[system] = payload
         
         return context
 

semantica/context/policy_engine.py

@@ -260,14 +260,22 @@ class PolicyEngine:
                 ORDER BY p.updated_at DESC
                 """
                 results = self.graph_store.execute_query(query, {"category": category})
+                records = self._extract_records(results)
 
                 policies = []
-                for record in results:
-                    policy_data = record.get("p", {})
-                    policies.append(self._dict_to_policy(policy_data))
-
-                if entities:
-                    pass
+                for record in records:
+                    policy_data = record.get("p") if isinstance(record, dict) else None
+                    if not isinstance(policy_data, dict):
+                        policy_data = record if isinstance(record, dict) else {}
+                    if not isinstance(policy_data, dict) or not policy_data.get("policy_id"):
+                        self.logger.debug(
+                            "Skipping malformed policy record in get_applicable_policies: "
+                            f"{record}"
+                        )
+                        continue
+                    policy = self._dict_to_policy(policy_data)
+                    if self._policy_matches_entities(policy, entities):
+                        policies.append(policy)
 
                 self.logger.info(f"Found {len(policies)} applicable policies for category {category}")
                 return policies
@@ -293,7 +301,7 @@ class PolicyEngine:
 
             policies: List[Policy] = []
             for data in latest_by_policy_id.values():
-                policies.append(self._dict_to_policy({
+                policy = self._dict_to_policy({
                     "policy_id": data.get("policy_id"),
                     "name": data.get("name"),
                     "description": data.get("description"),
@@ -303,7 +311,9 @@ class PolicyEngine:
                     "created_at": data.get("created_at"),
                     "updated_at": data.get("updated_at"),
                     "metadata": data.get("metadata", {})
-                }))
+                })
+                if self._policy_matches_entities(policy, entities):
+                    policies.append(policy)
 
             self.logger.info(f"Found {len(policies)} applicable policies for category {category}")
             return policies
@@ -311,6 +321,53 @@ class PolicyEngine:
         except Exception as e:
             self.logger.exception("Failed to get applicable policies")
             raise
+
+    def _extract_records(self, results: Any) -> List[Dict[str, Any]]:
+        """Normalize execute_query result shapes to record lists."""
+        if isinstance(results, dict):
+            records = results.get("records", [])
+            if not isinstance(records, list):
+                return []
+
+            # FalkorDB shape: {"records": [[...], ...], "header": ["col1", ...]}
+            header = results.get("header")
+            if (
+                isinstance(header, list)
+                and records
+                and all(isinstance(row, list) for row in records)
+            ):
+                normalized: List[Dict[str, Any]] = []
+                for row in records:
+                    row_map: Dict[str, Any] = dict(zip(header, row))
+                    normalized.append(row_map)
+                return normalized
+
+            return records
+        if isinstance(results, list):
+            return results
+        return []
+
+    def _policy_matches_entities(
+        self, policy: Policy, entities: Optional[List[str]]
+    ) -> bool:
+        """
+        Entity scoping for policies.
+        If no entity scope is defined on the policy, it is globally applicable.
+        """
+        if not entities:
+            return True
+
+        metadata = policy.metadata or {}
+        scoped_entities = (
+            metadata.get("entities")
+            or metadata.get("entity_ids")
+            or metadata.get("applies_to_entities")
+            or []
+        )
+        if not scoped_entities:
+            return True
+
+        return bool(set(str(e) for e in scoped_entities).intersection(set(entities)))
     
     def check_compliance(self, decision: Decision, policy_id: str) -> bool:
         """

tests/context/test_agent_context_decisions.py

@@ -246,6 +246,22 @@ class TestAgentContextDecisions:
             assert context[system]["system_name"] == system
             assert "captured_at" in context[system]
             assert context[system]["status"] == "captured"
+
+    def test_capture_cross_system_inputs_sanitizes_errors(
+        self, agent_context_with_decisions, mock_knowledge_graph
+    ):
+        """Test capture errors are sanitized in returned payload."""
+        mock_knowledge_graph.execute_query.side_effect = RuntimeError(
+            "backend connection failed: sensitive details"
+        )
+
+        context = agent_context_with_decisions.capture_cross_system_inputs(
+            ["salesforce"], "customer_001"
+        )
+
+        assert context["salesforce"]["status"] == "capture_failed"
+        assert context["salesforce"]["error"] == "internal_capture_error"
+        assert "sensitive" not in context["salesforce"]["error"]
     
     def test_backward_compatibility(self, mock_vector_store, mock_knowledge_graph):
         """Test backward compatibility when decision tracking is not explicitly set."""

tests/context/test_policy_engine.py

@@ -187,6 +187,86 @@ class TestPolicyEngine:
         
         assert len(policies) == 1
         assert policies[0].category == category
+
+    def test_get_applicable_policies_falkordb_row_shape(self, policy_engine, mock_graph_store):
+        """Test policy parsing when backend returns FalkorDB list rows + header."""
+        category = "credit_approval"
+        mock_graph_store.execute_query.return_value = {
+            "records": [
+                [
+                    {
+                        "policy_id": "policy_001",
+                        "name": "Credit Approval Policy",
+                        "description": "Standard credit approval rules",
+                        "rules": {"min_score": 650},
+                        "category": "credit_approval",
+                        "version": "1.0",
+                        "created_at": datetime.now().isoformat(),
+                        "updated_at": datetime.now().isoformat(),
+                        "metadata": {},
+                    }
+                ]
+            ],
+            "header": ["p"],
+        }
+
+        policies = policy_engine.get_applicable_policies(category, None)
+
+        assert len(policies) == 1
+        assert policies[0].policy_id == "policy_001"
+
+    def test_get_applicable_policies_skips_malformed_record(self, policy_engine, mock_graph_store):
+        """Test malformed policy records are skipped instead of crashing."""
+        category = "credit_approval"
+        mock_graph_store.execute_query.return_value = [{"unexpected": "shape"}]
+
+        policies = policy_engine.get_applicable_policies(category, None)
+
+        assert policies == []
+
+    def test_get_applicable_policies_context_graph_fallback_respects_entities(self):
+        """Test entity scoping is applied in find_nodes() fallback path."""
+        category = "credit_approval"
+        entities = ["customer:target"]
+
+        class _ContextGraphLike:
+            def find_nodes(self, node_type=None):
+                if node_type != "Policy":
+                    return []
+                return [
+                    {
+                        "metadata": {
+                            "policy_id": "policy_match",
+                            "name": "Scoped policy",
+                            "description": "Applies to target customer",
+                            "rules": {},
+                            "category": "credit_approval",
+                            "version": "1.0",
+                            "created_at": datetime.now().isoformat(),
+                            "updated_at": datetime.now().isoformat(),
+                            "metadata": {"entities": ["customer:target"]},
+                        }
+                    },
+                    {
+                        "metadata": {
+                            "policy_id": "policy_other",
+                            "name": "Other scoped policy",
+                            "description": "Applies elsewhere",
+                            "rules": {},
+                            "category": "credit_approval",
+                            "version": "1.0",
+                            "created_at": datetime.now().isoformat(),
+                            "updated_at": datetime.now().isoformat(),
+                            "metadata": {"entities": ["customer:other"]},
+                        }
+                    },
+                ]
+
+        engine = PolicyEngine(graph_store=_ContextGraphLike())
+        policies = engine.get_applicable_policies(category, entities)
+
+        assert len(policies) == 1
+        assert policies[0].policy_id == "policy_match"
     
     def test_check_compliance_success(self, policy_engine, mock_graph_store):
         """Test successful compliance checking."""