From 0a63128cbdb51686b01d502d0ee46e027e4c5563 Mon Sep 17 00:00:00 2001 From: KaifAhmad1 Date: Wed, 18 Feb 2026 11:55:30 +0530 Subject: [PATCH 1/5] Harden policy applicability retrieval and entity scoping --- semantica/context/policy_engine.py | 45 ++++++++++++++++++++++++++---- 1 file changed, 39 insertions(+), 6 deletions(-) diff --git a/semantica/context/policy_engine.py b/semantica/context/policy_engine.py index 10fe1512..59fabcaf 100644 --- a/semantica/context/policy_engine.py +++ b/semantica/context/policy_engine.py @@ -260,14 +260,16 @@ 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 {} + 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 @@ -311,6 +313,37 @@ 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", []) + return records if isinstance(records, list) else [] + 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: """ From 89d60301ce96330c90840e7755cff21ad8c15c79 Mon Sep 17 00:00:00 2001 From: KaifAhmad1 Date: Wed, 18 Feb 2026 11:56:22 +0530 Subject: [PATCH 2/5] Replace cross-system input placeholder with backend capture path --- semantica/context/agent_context.py | 39 +++++++++++++++++++++++++----- 1 file changed, 33 insertions(+), 6 deletions(-) diff --git a/semantica/context/agent_context.py b/semantica/context/agent_context.py index e7d41883..9d4a5230 100644 --- a/semantica/context/agent_context.py +++ b/semantica/context/agent_context.py @@ -1957,17 +1957,44 @@ 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: + payload["status"] = "capture_failed" + payload["error"] = str(e) + payload["records_found"] = 0 + payload["records"] = [] + + context[system] = payload return context From f9f19f343e324c02add88cb0f544b9c61c7d9c16 Mon Sep 17 00:00:00 2001 From: KaifAhmad1 Date: Wed, 18 Feb 2026 12:37:32 +0530 Subject: [PATCH 3/5] Handle FalkorDB policy rows in applicability parsing --- semantica/context/policy_engine.py | 24 ++++++++++++++++++- tests/context/test_policy_engine.py | 36 +++++++++++++++++++++++++++++ 2 files changed, 59 insertions(+), 1 deletion(-) diff --git a/semantica/context/policy_engine.py b/semantica/context/policy_engine.py index 59fabcaf..9085a6d4 100644 --- a/semantica/context/policy_engine.py +++ b/semantica/context/policy_engine.py @@ -267,6 +267,12 @@ class PolicyEngine: 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) @@ -318,7 +324,23 @@ class PolicyEngine: """Normalize execute_query result shapes to record lists.""" if isinstance(results, dict): records = results.get("records", []) - return records if isinstance(records, list) else [] + 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 [] diff --git a/tests/context/test_policy_engine.py b/tests/context/test_policy_engine.py index 2728a7bb..89d0ce72 100644 --- a/tests/context/test_policy_engine.py +++ b/tests/context/test_policy_engine.py @@ -187,6 +187,42 @@ 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_check_compliance_success(self, policy_engine, mock_graph_store): """Test successful compliance checking.""" From 8bd4df74e19dabea09c23b418decc55c51f2194b Mon Sep 17 00:00:00 2001 From: KaifAhmad1 Date: Wed, 18 Feb 2026 12:50:08 +0530 Subject: [PATCH 4/5] Apply entity scoping in ContextGraph policy fallback --- semantica/context/policy_engine.py | 6 ++-- tests/context/test_policy_engine.py | 44 +++++++++++++++++++++++++++++ 2 files changed, 48 insertions(+), 2 deletions(-) diff --git a/semantica/context/policy_engine.py b/semantica/context/policy_engine.py index 9085a6d4..b4d064ee 100644 --- a/semantica/context/policy_engine.py +++ b/semantica/context/policy_engine.py @@ -301,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"), @@ -311,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 diff --git a/tests/context/test_policy_engine.py b/tests/context/test_policy_engine.py index 89d0ce72..c7152aee 100644 --- a/tests/context/test_policy_engine.py +++ b/tests/context/test_policy_engine.py @@ -223,6 +223,50 @@ class TestPolicyEngine: 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.""" From a785247b9872ecc4b8e027b483cf0e276abb2804 Mon Sep 17 00:00:00 2001 From: KaifAhmad1 Date: Wed, 18 Feb 2026 12:54:59 +0530 Subject: [PATCH 5/5] Sanitize cross-system capture errors in returned payload --- semantica/context/agent_context.py | 8 +++++++- tests/context/test_agent_context_decisions.py | 16 ++++++++++++++++ 2 files changed, 23 insertions(+), 1 deletion(-) diff --git a/semantica/context/agent_context.py b/semantica/context/agent_context.py index 9d4a5230..8ee178dd 100644 --- a/semantica/context/agent_context.py +++ b/semantica/context/agent_context.py @@ -1989,8 +1989,14 @@ class AgentContext: 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"] = str(e) + payload["error"] = "internal_capture_error" payload["records_found"] = 0 payload["records"] = [] diff --git a/tests/context/test_agent_context_decisions.py b/tests/context/test_agent_context_decisions.py index 4a817451..e0cb60f3 100644 --- a/tests/context/test_agent_context_decisions.py +++ b/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."""