""" Comprehensive Real-World Test Suite for Semantica 0.3.0-alpha and 0.3.0-beta Tests all major features with real-world use cases based on publicly available knowledge about technology companies, their founders, products, and relationships. Coverage: 1. ContextGraph - add/query nodes and edges (tech company knowledge graph) 2. ContextGraph - decision tracking lifecycle (loan approval domain) 3. ContextGraph - causal chain traversal (upstream & downstream) 4. ContextGraph - find_precedents via PRECEDENT_FOR edges 5. ContextGraph - record_decision() high-level API 6. ContextGraph - analyze_graph_with_kg() KG analytics 7. ContextGraph - find_similar_nodes() content & structural similarity 8. ContextGraph - get_node_centrality() 9. ContextGraph - save/load round-trip (JSON) 10. ContextGraph - build_from_entities_and_relationships() 11. ContextGraph - query() keyword search 12. ContextGraph - get_neighbors() multi-hop BFS 13. PolicyEngine - get_applicable_policies (ContextGraph fallback) 14. DecisionQuery - find_precedents_hybrid 15. CausalChainAnalyzer - get_causal_chain 16. AgentContext - store / retrieve cycle 17. AgentContext - record_decision + find_precedents 18. AgentContext - capture_cross_system_inputs (error sanitization) 19. KG algorithms - GraphBuilder + CentralityCalculator (tech network) 20. KG algorithms - CommunityDetector (open-source ecosystem) 21. KG algorithms - PathFinder shortest path (supply chain) 22. KG algorithms - SimilarityCalculator (company similarity) 23. KG algorithms - NodeEmbedder (person embeddings) 24. KG algorithms - LinkPredictor (collaboration prediction) 25. Deduplication v2 - semantic dedup (company name variants) 26. Deduplication v2 - blocking_v2 candidate generation 27. Export - RDF/TTL alias 28. Reasoner - _match_pattern multi-word value inference 29. Pipeline - retry + exponential backoff 30. Decision models - full lifecycle serialization """ import json import os import tempfile import uuid from collections import defaultdict from datetime import datetime, timedelta from typing import Any, Dict, List from unittest.mock import MagicMock, Mock, patch import pytest # ── Context module ──────────────────────────────────────────────────────────── from semantica.context import ( AgentContext, CausalChainAnalyzer, ContextGraph, Decision, DecisionQuery, DecisionRecorder, Policy, PolicyEngine, ) from semantica.context.decision_models import ( ApprovalChain, PolicyException, Precedent, deserialize_decision, serialize_decision, validate_decision, ) # ── KG module ───────────────────────────────────────────────────────────────── from semantica.kg import ( CentralityCalculator, CommunityDetector, ConnectivityAnalyzer, GraphBuilder, LinkPredictor, NodeEmbedder, PathFinder, SimilarityCalculator, ) # ───────────────────────────────────────────────────────────────────────────── # Shared fixtures / helpers # ───────────────────────────────────────────────────────────────────────────── # Real-world tech company graph data # Sources: Wikipedia / Crunchbase public knowledge (no live web calls made) TECH_ENTITIES = [ {"id": "apple", "type": "company", "text": "Apple Inc.", "founded": 1976, "sector": "technology"}, {"id": "microsoft", "type": "company", "text": "Microsoft Corporation", "founded": 1975, "sector": "technology"}, {"id": "google", "type": "company", "text": "Google LLC", "founded": 1998, "sector": "technology"}, {"id": "openai", "type": "company", "text": "OpenAI", "founded": 2015, "sector": "ai"}, {"id": "anthropic", "type": "company", "text": "Anthropic PBC", "founded": 2021, "sector": "ai"}, {"id": "tim_cook", "type": "person", "text": "Tim Cook", "role": "CEO"}, {"id": "satya_nadella", "type": "person", "text": "Satya Nadella", "role": "CEO"}, {"id": "sundar_pichai", "type": "person", "text": "Sundar Pichai", "role": "CEO"}, {"id": "sam_altman", "type": "person", "text": "Sam Altman", "role": "CEO"}, {"id": "dario_amodei", "type": "person", "text": "Dario Amodei", "role": "CEO"}, {"id": "iphone", "type": "product", "text": "iPhone", "category": "smartphone"}, {"id": "chatgpt", "type": "product", "text": "ChatGPT", "category": "ai_assistant"}, {"id": "claude", "type": "product", "text": "Claude", "category": "ai_assistant"}, {"id": "gpt4", "type": "product", "text": "GPT-4", "category": "llm"}, ] TECH_RELATIONSHIPS = [ {"source_id": "tim_cook", "target_id": "apple", "type": "leads", "confidence": 1.0}, {"source_id": "satya_nadella", "target_id": "microsoft", "type": "leads", "confidence": 1.0}, {"source_id": "sundar_pichai", "target_id": "google", "type": "leads", "confidence": 1.0}, {"source_id": "sam_altman", "target_id": "openai", "type": "leads", "confidence": 1.0}, {"source_id": "dario_amodei", "target_id": "anthropic", "type": "leads", "confidence": 1.0}, {"source_id": "apple", "target_id": "iphone", "type": "makes", "confidence": 1.0}, {"source_id": "openai", "target_id": "chatgpt", "type": "makes", "confidence": 1.0}, {"source_id": "openai", "target_id": "gpt4", "type": "makes", "confidence": 1.0}, {"source_id": "anthropic", "target_id": "claude", "type": "makes", "confidence": 1.0}, {"source_id": "microsoft", "target_id": "openai", "type": "invested_in", "confidence": 1.0}, {"source_id": "google", "target_id": "anthropic", "type": "invested_in", "confidence": 1.0}, {"source_id": "chatgpt", "target_id": "claude", "type": "competes_with", "confidence": 0.9}, {"source_id": "gpt4", "target_id": "claude", "type": "competes_with", "confidence": 0.9}, ] def _make_decision( decision_id: str = None, category: str = "loan_approval", scenario: str = "First-time homebuyer mortgage application", reasoning: str = "Applicant has FICO score 780, stable 5-year employment, 20% down payment.", outcome: str = "approved", confidence: float = 0.92, decision_maker: str = "mortgage_underwriter_ai", ) -> Decision: return Decision( decision_id=decision_id or str(uuid.uuid4()), category=category, scenario=scenario, reasoning=reasoning, outcome=outcome, confidence=confidence, timestamp=datetime.now(), decision_maker=decision_maker, ) def _make_policy(category: str = "loan_approval") -> Policy: return Policy( policy_id=str(uuid.uuid4()), name="Mortgage Lending Policy v3.2", description="Standard policy for retail mortgage approvals", rules={ "min_fico_score": 680, "max_dti_ratio": 0.43, "min_down_payment_pct": 5, "employment_history_years": 2, }, category=category, version="3.2", created_at=datetime.now() - timedelta(days=365), updated_at=datetime.now() - timedelta(days=30), ) def _build_tech_graph() -> ContextGraph: g = ContextGraph(advanced_analytics=True) g.build_from_entities_and_relationships(TECH_ENTITIES, TECH_RELATIONSHIPS) return g # ═════════════════════════════════════════════════════════════════════════════ # 1. ContextGraph – basic node/edge operations on a tech company graph # ═════════════════════════════════════════════════════════════════════════════ class TestContextGraphBasicOperations: """Basic add/query on a real-world tech company knowledge graph.""" def test_build_from_entities_and_relationships(self): g = _build_tech_graph() stats = g.stats() assert stats["node_count"] >= len(TECH_ENTITIES) assert stats["edge_count"] >= len(TECH_RELATIONSHIPS) def test_node_types_indexed_correctly(self): g = _build_tech_graph() companies = g.find_nodes(node_type="company") persons = g.find_nodes(node_type="person") products = g.find_nodes(node_type="product") assert len(companies) == 5 # apple, microsoft, google, openai, anthropic assert len(persons) == 5 assert len(products) == 4 def test_find_node_returns_metadata(self): g = _build_tech_graph() node = g.find_node("apple") assert node is not None assert node["type"] == "company" assert node["content"] == "Apple Inc." def test_add_node_and_edge_manual(self): g = ContextGraph() g.add_node("meta", node_type="company", content="Meta Platforms") g.add_node("mark_zuckerberg", node_type="person", content="Mark Zuckerberg") added = g.add_edge("mark_zuckerberg", "meta", edge_type="leads") assert added is True assert g.has_node("meta") neighbors = g.get_neighbors("mark_zuckerberg", hops=1) assert any(n["id"] == "meta" for n in neighbors) def test_query_keyword_search(self): g = _build_tech_graph() results = g.query("AI assistant") # chatgpt and claude both contain "ai_assistant" in category # But query searches content - let's check that something is returned or not # The query searches node content. Let's add a node with explicit content g.add_node("llm_overview", node_type="concept", content="Large Language Models AI assistant") results = g.query("AI assistant") assert len(results) > 0 assert results[0]["score"] > 0.0 def test_get_neighbors_multi_hop(self): g = _build_tech_graph() # 1 hop from microsoft: openai (invested_in) # 2 hops from microsoft: chatgpt (openai makes chatgpt), gpt4 neighbors_1 = g.get_neighbors("microsoft", hops=1) neighbors_2 = g.get_neighbors("microsoft", hops=2) one_hop_ids = {n["id"] for n in neighbors_1} two_hop_ids = {n["id"] for n in neighbors_2} assert "openai" in one_hop_ids assert len(two_hop_ids) > len(one_hop_ids) def test_find_edges_by_type(self): g = _build_tech_graph() invested_edges = g.find_edges(edge_type="invested_in") assert len(invested_edges) == 2 # microsoft->openai, google->anthropic sources = {e["source"] for e in invested_edges} assert "microsoft" in sources assert "google" in sources def test_density_calculation(self): g = _build_tech_graph() density = g.density() assert 0.0 <= density <= 1.0 def test_to_dict_round_trip(self): g = _build_tech_graph() d = g.to_dict() assert "nodes" in d assert "edges" in d assert d["statistics"]["node_count"] == len(g.nodes) g2 = ContextGraph(advanced_analytics=False) g2.from_dict(d) assert len(g2.nodes) == len(g.nodes) assert len(g2.edges) == len(g.edges) def test_save_and_load_file(self): g = _build_tech_graph() with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as f: path = f.name try: g.save_to_file(path) assert os.path.exists(path) g2 = ContextGraph(advanced_analytics=False) g2.load_from_file(path) assert len(g2.nodes) == len(g.nodes) assert g2.has_node("apple") assert g2.has_node("claude") finally: os.unlink(path) # ═════════════════════════════════════════════════════════════════════════════ # 2. ContextGraph – decision tracking lifecycle # ═════════════════════════════════════════════════════════════════════════════ class TestContextGraphDecisionTracking: """Decision tracking: add, causal chains, precedents, record_decision.""" def test_add_decision_stores_node(self): g = ContextGraph() d = _make_decision(decision_id="loan_001") g.add_decision(d) node = g.find_node("loan_001") assert node is not None assert node["type"] == "Decision" assert "loan_approval" in str(node["metadata"]) def test_add_decision_with_none_metadata(self): g = ContextGraph() d = _make_decision(decision_id="loan_002") d.metadata = None g.add_decision(d) # Should not raise assert g.has_node("loan_002") def test_add_decision_empty_id_generates_uuid(self): g = ContextGraph() d = Decision( decision_id="auto_001", category="credit", scenario="Credit card limit increase", reasoning="3-year on-time payment history", outcome="approved", confidence=0.87, timestamp=datetime.now(), decision_maker="credit_scoring_engine", ) g.add_decision(d) assert g.has_node("auto_001") def test_causal_chain_downstream(self): g = ContextGraph() d1 = _make_decision(decision_id="d_root", scenario="Initial credit approval") d2 = _make_decision(decision_id="d_child", scenario="Credit limit increase") d3 = _make_decision(decision_id="d_grandchild", scenario="Premium card upgrade") g.add_decision(d1) g.add_decision(d2) g.add_decision(d3) g.add_causal_relationship("d_root", "d_child", "CAUSED") g.add_causal_relationship("d_child", "d_grandchild", "CAUSED") chain = g.get_causal_chain("d_root", direction="downstream", max_depth=5) chain_ids = {d.decision_id for d in chain} assert "d_child" in chain_ids assert "d_grandchild" in chain_ids def test_causal_chain_upstream(self): g = ContextGraph() d_early = _make_decision(decision_id="d_early", scenario="Initial loan decision 2023") d_late = _make_decision(decision_id="d_late", scenario="Refinance decision 2024") g.add_decision(d_early) g.add_decision(d_late) g.add_causal_relationship("d_early", "d_late", "INFLUENCED") chain = g.get_causal_chain("d_late", direction="upstream", max_depth=5) chain_ids = {d.decision_id for d in chain} assert "d_early" in chain_ids def test_causal_relationship_invalid_type_raises(self): g = ContextGraph() d1 = _make_decision(decision_id="x1") d2 = _make_decision(decision_id="x2") g.add_decision(d1) g.add_decision(d2) with pytest.raises(ValueError, match="Relationship type must be one of"): g.add_causal_relationship("x1", "x2", "INVALID_TYPE") def test_causal_relationship_nonexistent_node_skipped(self): g = ContextGraph() d1 = _make_decision(decision_id="exist_1") g.add_decision(d1) g.add_causal_relationship("exist_1", "ghost_node", "CAUSED") # Should not raise chain = g.get_causal_chain("exist_1", direction="downstream") assert len(chain) == 0 def test_find_precedents_via_precedent_for_edge(self): g = ContextGraph() old = _make_decision(decision_id="prec_old", scenario="Homebuyer loan 2022, FICO 770") new = _make_decision(decision_id="prec_new", scenario="Homebuyer loan 2024, FICO 780") g.add_decision(old) g.add_decision(new) g.add_causal_relationship("prec_old", "prec_new", "PRECEDENT_FOR") precedents = g.find_precedents("prec_new", limit=10) assert len(precedents) == 1 assert precedents[0].decision_id == "prec_old" def test_record_decision_high_level_api(self): g = ContextGraph() did = g.record_decision( category="fraud_detection", scenario="Transaction flagged: $5,000 wire transfer to new payee in 30 seconds", reasoning="Velocity anomaly: 10x avg, new beneficiary, unusual hours (2 AM).", outcome="blocked_pending_review", confidence=0.95, entities=["acct_789012", "payee_xyz"], decision_maker="fraud_ai_v4", ) assert did is not None assert g.has_node(did) node = g.find_node(did) # record_decision() uses lowercase "decision" node type assert node["type"].lower() == "decision" def test_record_decision_validation_errors(self): g = ContextGraph() with pytest.raises(ValueError): g.record_decision("", "scenario", "reasoning", "outcome", 0.9) with pytest.raises(ValueError): g.record_decision("cat", "scenario", "reasoning", "outcome", 1.5) def test_multiple_decisions_same_category(self): g = ContextGraph() ids = [] for i in range(5): did = g.record_decision( category="insurance_claim", scenario=f"Auto claim #{i}: rear-end collision, liability clear", reasoning=f"Claimant has comprehensive coverage, incident #{i} verified by telematics.", outcome="approved", confidence=0.88 + i * 0.01, decision_maker="claims_ai", ) ids.append(did) # All should be unique decision nodes assert len(set(ids)) == 5 # record_decision() uses lowercase "decision" type decision_nodes = g.find_nodes(node_type="decision") assert len(decision_nodes) >= 5 # ═════════════════════════════════════════════════════════════════════════════ # 3. ContextGraph – KG analytics (centrality, similar nodes) # ═════════════════════════════════════════════════════════════════════════════ class TestContextGraphKGAnalytics: """KG algorithm integration: centrality, find_similar_nodes, analyze_graph_with_kg.""" def test_find_similar_nodes_content(self): g = _build_tech_graph() # openai and anthropic are both AI companies similar = g.find_similar_nodes("openai", similarity_type="content", top_k=5) assert isinstance(similar, list) # Should return some results; content of "OpenAI" and "Anthropic PBC" may have partial overlap assert len(similar) <= len(g.nodes) - 1 def test_find_similar_nodes_structural(self): g = _build_tech_graph() similar = g.find_similar_nodes("apple", similarity_type="structural", top_k=5) assert isinstance(similar, list) for item in similar: assert isinstance(item, dict) assert isinstance(item.get("id"), str) assert isinstance(item.get("content"), str) assert isinstance(item.get("type"), str) assert isinstance(item.get("score"), float) assert 0.0 <= item["score"] <= 1.0 def test_find_similar_nodes_missing_node_returns_empty(self): g = _build_tech_graph() result = g.find_similar_nodes("nonexistent_node", similarity_type="content") assert result == [] def test_get_node_centrality_returns_dict(self): g = _build_tech_graph() centrality = g.get_node_centrality("openai") # Either returns centrality scores or an error dict (if KG components unavailable) assert isinstance(centrality, dict) def test_analyze_graph_with_kg_structure(self): g = _build_tech_graph() analysis = g.analyze_graph_with_kg() assert isinstance(analysis, dict) if "error" not in analysis: assert "graph_metrics" in analysis assert analysis["graph_metrics"]["node_count"] >= len(TECH_ENTITIES) def test_density_is_nonzero_for_connected_graph(self): g = _build_tech_graph() assert g.density() > 0.0 # ═════════════════════════════════════════════════════════════════════════════ # 4. PolicyEngine – ContextGraph fallback path (from 0.3.0-alpha fix) # ═════════════════════════════════════════════════════════════════════════════ class TestPolicyEngineContextGraphFallback: """Tests the ContextGraph-backed PolicyEngine introduced in 0.3.0-alpha.""" def _make_policy_node(self, policy_id, category, entity_filter=None): return { "metadata": { "policy_id": policy_id, "name": f"Policy {policy_id}", "description": f"Desc {policy_id}", "rules": {"max_loan": 500_000}, "category": category, "version": "1.0", "created_at": datetime.now().isoformat(), "updated_at": datetime.now().isoformat(), "metadata": {"entities": entity_filter or []}, } } def test_applicable_policies_matched_by_entity(self): class _FakeGraph: def find_nodes(self, node_type=None): return [ self._pnode("p_mortgage", "mortgage", ["customer:A"]), self._pnode("p_auto", "mortgage", ["customer:B"]), ] def _pnode(self, pid, cat, entities): return { "metadata": { "policy_id": pid, "name": pid, "description": pid, "rules": {}, "category": cat, "version": "1.0", "created_at": datetime.now().isoformat(), "updated_at": datetime.now().isoformat(), "metadata": {"entities": entities}, } } engine = PolicyEngine(graph_store=_FakeGraph()) policies = engine.get_applicable_policies("mortgage", ["customer:A"]) assert len(policies) == 1 assert policies[0].policy_id == "p_mortgage" def test_no_policies_when_entity_not_matched(self): class _FakeGraph: def find_nodes(self, node_type=None): return [ { "metadata": { "policy_id": "p_x", "name": "X", "description": "X", "rules": {}, "category": "credit", "version": "1.0", "created_at": datetime.now().isoformat(), "updated_at": datetime.now().isoformat(), "metadata": {"entities": ["customer:999"]}, } } ] engine = PolicyEngine(graph_store=_FakeGraph()) policies = engine.get_applicable_policies("credit", ["customer:000"]) assert len(policies) == 0 # ═════════════════════════════════════════════════════════════════════════════ # 5. DecisionQuery + CausalChainAnalyzer – mock graph store (from 0.3.0-beta fix) # ═════════════════════════════════════════════════════════════════════════════ class TestDecisionQueryAndCausalAnalyzer: """DecisionQuery and CausalChainAnalyzer with wrapped-records store format.""" def _mock_store(self): store = Mock() store.execute_query = Mock( return_value={ "records": [ { "d": { "decision_id": "q_001", "category": "mortgage", "scenario": "30yr fixed refinance, rate drop 1.2%", "reasoning": "Rate-and-term refi; DTI 38%, FICO 800.", "outcome": "approved", "confidence": 0.96, "timestamp": datetime.now().isoformat(), "decision_maker": "refi_agent", }, "end": { "decision_id": "q_002", "category": "mortgage", "scenario": "Cash-out refi for home improvement", "reasoning": "LTV 72%, strong employment history.", "outcome": "approved", "confidence": 0.88, "timestamp": datetime.now().isoformat(), "decision_maker": "refi_agent", }, "distance": 1, } ] } ) return store def test_decision_query_find_precedents_hybrid(self): store = self._mock_store() query = DecisionQuery(graph_store=store) precedents = query.find_precedents_hybrid("refinance scenario", "mortgage", 5) assert len(precedents) == 1 assert precedents[0].decision_id == "q_001" def test_causal_chain_analyzer_downstream(self): store = self._mock_store() analyzer = CausalChainAnalyzer(graph_store=store) chain = analyzer.get_causal_chain("q_001", "downstream", 3) assert len(chain) == 1 assert chain[0].decision_id == "q_002" # ═════════════════════════════════════════════════════════════════════════════ # 6. AgentContext – high-level interface (0.3.0-alpha / beta) # ═════════════════════════════════════════════════════════════════════════════ class TestAgentContextHighLevel: """AgentContext: store/retrieve, decision tracking, cross-system capture.""" def _make_agent_context(self, error_kg=False): vector_store = Mock() vector_store.add = Mock(return_value="vs_id_001") vector_store.search = Mock(return_value=[]) knowledge_graph = Mock() if error_kg: knowledge_graph.execute_query = Mock(side_effect=RuntimeError("db connection refused")) else: knowledge_graph.execute_query = Mock(return_value={"records": []}) knowledge_graph.get_neighbors = Mock(return_value=[]) return AgentContext( vector_store=vector_store, knowledge_graph=knowledge_graph, decision_tracking=True, ) def test_store_and_retrieve_basic(self): ctx = self._make_agent_context() mem_id = ctx.store( "Apple acquired Beats Electronics in 2014 for $3 billion, " "its largest acquisition at the time.", conversation_id="conv_apple_history", ) assert mem_id is not None def test_record_decision_returns_id(self): ctx = self._make_agent_context() did = ctx.record_decision( category="supply_chain", scenario="Dual-source procurement: TSMC + Samsung for A-series chips", reasoning="Geopolitical risk mitigation; TSMC concentration risk post-2021 chip shortage.", outcome="approved", confidence=0.88, entities=["tsmc", "samsung", "apple_chip_team"], ) assert did is not None def test_find_precedents_returns_list(self): ctx = self._make_agent_context() ctx.record_decision( category="supply_chain", scenario="Single-source silicon vendor", reasoning="Cost optimisation in stable environment", outcome="approved", confidence=0.75, ) precedents = ctx.find_precedents("supply chain procurement decision", category="supply_chain") assert isinstance(precedents, list) def test_capture_cross_system_inputs_error_sanitized(self): ctx = self._make_agent_context(error_kg=True) result = ctx.capture_cross_system_inputs(["salesforce"], "customer_001") assert result["salesforce"]["status"] == "capture_failed" # Internal error detail must not leak assert "db connection" not in result["salesforce"].get("error", "") def test_capture_cross_system_inputs_success(self): vector_store = Mock() knowledge_graph = Mock() knowledge_graph.execute_query = Mock( return_value={ "records": [ { "c": { "context_id": "ctx_01", "system_name": "jira", "context_data": {"sprint": 42}, } } ] } ) ctx = AgentContext( vector_store=vector_store, knowledge_graph=knowledge_graph, decision_tracking=True, ) systems = ["jira", "confluence", "datadog"] result = ctx.capture_cross_system_inputs(systems, "project_alpha") for system in systems: assert result[system]["system_name"] == system assert result[system]["entity_id"] == "project_alpha" assert result[system]["status"] == "captured" # ═════════════════════════════════════════════════════════════════════════════ # 7. Decision models – serialization, validation # ═════════════════════════════════════════════════════════════════════════════ class TestDecisionModelsSerialization: """Decision model lifecycle: create, validate, serialize, deserialize.""" def test_decision_serialize_deserialize(self): d = _make_decision(decision_id="ser_001") json_str = serialize_decision(d) restored = deserialize_decision(json_str) assert restored.decision_id == d.decision_id assert restored.category == d.category assert abs(restored.confidence - d.confidence) < 0.001 def test_validate_decision_passes(self): d = _make_decision(decision_id="val_001") assert validate_decision(d) is True def test_validate_decision_fails_empty_category(self): d = _make_decision(decision_id="val_002") d.category = "" assert validate_decision(d) is False def test_policy_to_dict_and_from_dict(self): p = _make_policy() d = p.to_dict() restored = Policy.from_dict(d) assert restored.policy_id == p.policy_id assert restored.name == p.name assert restored.rules == p.rules def test_policy_exception_model(self): exc = PolicyException( exception_id=str(uuid.uuid4()), decision_id="loan_001", policy_id="pol_001", reason="Customer relationship >10yr with zero defaults; waived min FICO by 15pts.", approver="senior_underwriter_007", approval_timestamp=datetime.now(), justification="Relationship banking exception per policy section 4.3.1", ) d = exc.to_dict() assert d["reason"] == exc.reason restored = PolicyException.from_dict(d) assert restored.approver == exc.approver def test_approval_chain_model(self): approval = ApprovalChain( approval_id=str(uuid.uuid4()), decision_id="loan_001", approver="branch_manager_ann", approval_method="slack_dm", approval_context="Reviewed customer portfolio in Slack thread #mortgage-exceptions", timestamp=datetime.now(), ) d = approval.to_dict() assert d["approval_method"] == "slack_dm" def test_precedent_model(self): prec = Precedent( precedent_id=str(uuid.uuid4()), source_decision_id="old_loan_789", similarity_score=0.87, relationship_type="similar_scenario", ) d = prec.to_dict() assert d["similarity_score"] == 0.87 def test_decision_confidence_bounds(self): with pytest.raises(ValueError): Decision( decision_id="bad_conf", category="test", scenario="test", reasoning="test", outcome="test", confidence=1.5, # Invalid timestamp=datetime.now(), decision_maker="test", ) # ═════════════════════════════════════════════════════════════════════════════ # 8. KG algorithms – real-world tech company graph # ═════════════════════════════════════════════════════════════════════════════ def _build_nx_tech_graph(): """Build a NetworkX graph from TECH_ENTITIES/TECH_RELATIONSHIPS.""" import networkx as nx G = nx.DiGraph() for e in TECH_ENTITIES: G.add_node(e["id"], **{k: v for k, v in e.items() if k != "id"}) for r in TECH_RELATIONSHIPS: G.add_edge(r["source_id"], r["target_id"], type=r["type"], weight=r["confidence"]) return G class TestKGAlgorithmsRealWorld: """KG algorithms with real-world tech company data.""" def test_graph_builder_produces_entities_and_relationships(self): builder = GraphBuilder(merge_entities=False) sources = [{"entities": TECH_ENTITIES, "relationships": TECH_RELATIONSHIPS}] kg = builder.build(sources=sources) assert "entities" in kg assert len(kg["entities"]) >= len(TECH_ENTITIES) def test_centrality_calculator_on_tech_graph(self): G = _build_nx_tech_graph() calc = CentralityCalculator() result = calc.calculate_all_centrality(G) assert isinstance(result, dict) # openai should appear (it's heavily connected: microsoft invested, makes chatgpt/gpt4) all_keys = set(result.keys()) if result else set() # Result may be keyed by node ID or metric name - just verify it's non-empty assert len(result) > 0 def test_community_detector_on_tech_graph(self): G = _build_nx_tech_graph() detector = CommunityDetector() communities = detector.detect_communities(G) # Should return communities (list/dict/int) assert communities is not None def test_connectivity_analyzer_on_tech_graph(self): G = _build_nx_tech_graph() analyzer = ConnectivityAnalyzer() result = analyzer.analyze_connectivity(G) assert isinstance(result, dict) def test_path_finder_shortest_path_supply_chain(self): """Shortest path: microsoft -> chatgpt (via openai).""" import networkx as nx G = _build_nx_tech_graph() # Convert to undirected for simple path test G_undirected = G.to_undirected() finder = PathFinder() path = finder.find_shortest_path(G_undirected, "microsoft", "chatgpt") assert path is not None assert "microsoft" in path assert "chatgpt" in path assert "openai" in path # Must pass through openai def test_similarity_calculator_cosine(self): import numpy as np calc = SimilarityCalculator() v1 = np.array([0.1, 0.9, 0.2]) v2 = np.array([0.15, 0.85, 0.18]) score = calc.cosine_similarity(v1, v2) assert 0.0 <= score <= 1.0 assert score > 0.95 # Very similar vectors def test_node_embedder_generates_embeddings(self): G = _build_nx_tech_graph() embedder = NodeEmbedder(embedding_dimension=16, walk_length=10, num_walks=2, epochs=1) node_labels = list({data.get("type", "entity") for _, data in G.nodes(data=True)}) rel_types = list({data.get("type", "related_to") for _, _, data in G.edges(data=True)}) embeddings = embedder.compute_embeddings(G, node_labels=node_labels, relationship_types=rel_types) assert isinstance(embeddings, dict) def test_link_predictor_preferential_attachment(self): G = _build_nx_tech_graph().to_undirected() predictor = LinkPredictor() # Predict probability of a link between apple and google score = predictor.score_link(G, "apple", "google", method="preferential_attachment") assert isinstance(score, float) assert score >= 0.0 # ═════════════════════════════════════════════════════════════════════════════ # 9. Open-source ecosystem community detection # ═════════════════════════════════════════════════════════════════════════════ class TestCommunityDetectionOpenSourceEcosystem: """Community detection on a simulated open-source library dependency graph.""" @pytest.fixture def ecosystem_graph(self): import networkx as nx G = nx.Graph() # Python data science ecosystem packages = [ "numpy", "scipy", "pandas", "matplotlib", "seaborn", "sklearn", "torch", "tensorflow", "keras", "transformers", "fastapi", "flask", "django", "sqlalchemy", "pydantic", "requests", "httpx", "aiohttp", ] for pkg in packages: G.add_node(pkg) # Dependencies edges = [ ("scipy", "numpy"), ("pandas", "numpy"), ("matplotlib", "numpy"), ("seaborn", "matplotlib"), ("seaborn", "pandas"), ("sklearn", "numpy"), ("sklearn", "scipy"), ("keras", "tensorflow"), ("transformers", "torch"), ("transformers", "tensorflow"), ("fastapi", "pydantic"), ("fastapi", "httpx"), ("flask", "requests"), ("django", "sqlalchemy"), ("aiohttp", "httpx"), ] G.add_edges_from(edges) return G def test_communities_are_detected(self, ecosystem_graph): detector = CommunityDetector() communities = detector.detect_communities(ecosystem_graph) assert communities is not None def test_connectivity_analysis(self, ecosystem_graph): analyzer = ConnectivityAnalyzer() result = analyzer.analyze_connectivity(ecosystem_graph) assert isinstance(result, dict) # ═════════════════════════════════════════════════════════════════════════════ # 10. Deduplication v2 – semantic dedup on company name variants # ═════════════════════════════════════════════════════════════════════════════ class TestDeduplicationV2RealWorld: """Semantic deduplication with company name variants (real-world data quality issue).""" def test_semantic_dedup_company_name_variants(self): from semantica.deduplication import DuplicateDetector entities = [ {"id": "e1", "name": "Apple Inc.", "type": "Company"}, {"id": "e2", "name": "Apple Incorporated", "type": "Company"}, {"id": "e3", "name": "APPLE INC", "type": "Company"}, {"id": "e4", "name": "Microsoft Corporation", "type": "Company"}, {"id": "e5", "name": "Microsoft Corp.", "type": "Company"}, {"id": "e6", "name": "Microsoft Corp", "type": "Company"}, {"id": "e7", "name": "Google LLC", "type": "Company"}, {"id": "e8", "name": "Alphabet Inc.", "type": "Company"}, ] detector = DuplicateDetector(similarity_threshold=0.75) duplicates = detector.detect_duplicates(entities, threshold=0.75) assert isinstance(duplicates, list) # Apple variants and Microsoft variants should be detected as duplicates assert len(duplicates) > 0 def test_semantic_dedup_v2_mode(self): from semantica.deduplication import DuplicateDetector entities = [ {"id": "e1", "name": "OpenAI Inc.", "type": "Company"}, {"id": "e2", "name": "OpenAI", "type": "Company"}, {"id": "e3", "name": "Anthropic PBC", "type": "Company"}, {"id": "e4", "name": "Anthropic", "type": "Company"}, ] detector = DuplicateDetector(similarity_threshold=0.70) duplicates = detector.detect_duplicates(entities, threshold=0.70) assert isinstance(duplicates, list) def test_blocking_v2_candidate_generation(self): from semantica.deduplication import DuplicateDetector entities = [ {"id": f"ent_{i}", "name": f"Entity Name {i}", "type": "Organization"} for i in range(30) ] # Add some deliberate duplicates entities.append({"id": "dup_1", "name": "Entity Name 5", "type": "Organization"}) entities.append({"id": "dup_2", "name": "Entity Name 12", "type": "Organization"}) detector = DuplicateDetector(similarity_threshold=0.90) duplicates = detector.detect_duplicates(entities, threshold=0.90, candidate_strategy="blocking_v2") assert isinstance(duplicates, list) # Exact duplicates should be detected assert len(duplicates) > 0 # ═════════════════════════════════════════════════════════════════════════════ # 11. Export – RDF/TTL alias (0.3.0-beta fix) # ═════════════════════════════════════════════════════════════════════════════ class TestRDFExportAliases: """TTL and other format aliases introduced in 0.3.0-beta.""" def _make_kg_data(self): return { "entities": [ {"id": "apple", "type": "Company", "properties": {"name": "Apple Inc.", "founded": 1976}}, {"id": "tim_cook", "type": "Person", "properties": {"name": "Tim Cook"}}, ], "relationships": [ {"source_id": "tim_cook", "target_id": "apple", "type": "leads", "properties": {}}, ], } def test_ttl_alias_export(self): from semantica.export import RDFExporter exporter = RDFExporter() kg_data = self._make_kg_data() # export_to_rdf returns a string; "ttl" is an alias for "turtle" result = exporter.export_to_rdf(kg_data, format="ttl") assert isinstance(result, str) assert len(result) > 0 # Should contain RDF/turtle syntax elements assert "apple" in result or "Apple" in result or "@" in result or "<" in result def test_nt_alias_export(self): from semantica.export import RDFExporter exporter = RDFExporter() kg_data = self._make_kg_data() result = exporter.export_to_rdf(kg_data, format="nt") assert isinstance(result, str) assert len(result) > 0 def test_turtle_canonical_format(self): from semantica.export import RDFExporter exporter = RDFExporter() kg_data = self._make_kg_data() result = exporter.export_to_rdf(kg_data, format="turtle") assert isinstance(result, str) assert len(result) > 0 def test_invalid_format_raises(self): from semantica.export import RDFExporter exporter = RDFExporter() with pytest.raises(Exception): exporter.export_to_rdf(self._make_kg_data(), format="invalid_format_xyz") # ═════════════════════════════════════════════════════════════════════════════ # 12. Reasoner – _match_pattern with multi-word values (0.3.0-beta fix) # ═════════════════════════════════════════════════════════════════════════════ class TestReasonerPatternMatching: """Pattern matching fixes from 0.3.0-beta: multi-word values, pre-bound vars.""" def test_multi_word_value_inference(self): """Bug #354 — _match_pattern must match facts with multi-word values.""" from semantica.reasoning.reasoner import Reasoner reasoner = Reasoner() # Use dict fact format: source_name/target_name/type for fact in [ {"source_name": "Steve Jobs", "target_name": "Apple", "type": "founded_by"}, {"source_name": "Steve Wozniak", "target_name": "Apple", "type": "founded_by"}, {"source_name": "Ronald Wayne", "target_name": "Apple", "type": "founded_by"}, ]: reasoner.add_fact(fact) inferred = reasoner.infer_facts( [], rules=["IF founded_by(?person, ?org) THEN is_cofounder(?person, ?org)"], ) assert isinstance(inferred, list) assert len(inferred) == 3 assert "is_cofounder(Steve Jobs, Apple)" in inferred assert "is_cofounder(Steve Wozniak, Apple)" in inferred assert "is_cofounder(Ronald Wayne, Apple)" in inferred def test_single_word_value_regression(self): """Single-word value inference must continue to work (regression guard).""" from semantica.reasoning.reasoner import Reasoner reasoner = Reasoner() inferred = reasoner.infer_facts( ["Person(John)", "Parent(John, Jane)"], rules=["IF Person(?x) AND Parent(?x, ?y) THEN Child(?y, ?x)"], ) assert "Child(Jane, John)" in inferred def test_pre_bound_variable_match(self): """_match_pattern must enforce pre-bound variables.""" from semantica.reasoning.reasoner import Reasoner reasoner = Reasoner() # Pre-bound: org=Apple must match exactly result = reasoner._match_pattern( "founded_by(?person, ?org)", "founded_by(Steve Jobs, Apple)", {"org": "Apple"}, ) assert result is not None assert result["person"] == "Steve Jobs" assert result["org"] == "Apple" def test_pre_bound_variable_conflict_returns_none(self): """_match_pattern must return None on variable binding conflict.""" from semantica.reasoning.reasoner import Reasoner reasoner = Reasoner() result = reasoner._match_pattern( "founded_by(?person, ?org)", "founded_by(Steve Jobs, Apple)", {"org": "Google"}, # Conflict: Apple != Google ) assert result is None def test_forward_chain_tech_facts(self): """Forward chaining with real-world tech facts.""" from semantica.reasoning.reasoner import Reasoner reasoner = Reasoner() reasoner.add_rule("IF AICompany(?x) AND Creates(?x, ?product) THEN AIProduct(?product)") reasoner.add_fact("AICompany(OpenAI)") reasoner.add_fact("Creates(OpenAI, ChatGPT)") results = reasoner.forward_chain() conclusions = [r.conclusion for r in results] assert "AIProduct(ChatGPT)" in conclusions # ═════════════════════════════════════════════════════════════════════════════ # 13. Pipeline – retry and failure handling (0.3.0-beta fix) # ═════════════════════════════════════════════════════════════════════════════ class TestPipelineRetryAndFailure: """Pipeline retry loop and FailureHandler from 0.3.0-beta fix.""" def test_pipeline_builder_returns_step(self): from semantica.pipeline import PipelineBuilder builder = PipelineBuilder() step = builder.add_step("ingest", "file_ingestor", dependencies=[]) assert step is not None assert hasattr(step, "name") def test_pipeline_validator_detects_missing_dependency(self): from semantica.pipeline import PipelineBuilder, PipelineValidator builder = PipelineBuilder() builder.add_step("ingest", "file_ingestor", dependencies=[]) step = builder.add_step("load", "kg_builder", dependencies=["ingest"]) # Manually inject a nonexistent dependency (same pattern as existing pipeline tests) step.dependencies.append("ghost_step") validator = PipelineValidator() result = validator.validate(builder) assert result.valid is False assert any("ghost_step" in e for e in result.errors) def test_failure_handler_linear_backoff(self): from semantica.pipeline.failure_handler import FailureHandler, RetryPolicy, RecoveryAction, RetryStrategy handler = FailureHandler() policy = RetryPolicy(strategy=RetryStrategy.LINEAR, initial_delay=0.5, max_retries=5) action = handler.handle_failure( error=RuntimeError("connection timeout"), policy=policy, retry_count=2, ) assert isinstance(action, RecoveryAction) assert action.retry_delay >= 0.0 def test_failure_handler_exponential_backoff(self): from semantica.pipeline.failure_handler import FailureHandler, RetryPolicy, RecoveryAction, RetryStrategy handler = FailureHandler() policy = RetryPolicy(strategy=RetryStrategy.EXPONENTIAL, initial_delay=1.0, max_retries=5) action1 = handler.handle_failure(RuntimeError("err"), policy, retry_count=1) action2 = handler.handle_failure(RuntimeError("err"), policy, retry_count=2) # Exponential delay should grow with retry count assert action2.retry_delay >= action1.retry_delay def test_validate_alias_works(self): from semantica.pipeline import PipelineBuilder, PipelineValidator builder = PipelineBuilder() builder.add_step("step_a", "some_processor", dependencies=[]) validator = PipelineValidator() # validate() is an alias for validate_pipeline() – should not raise validator.validate(builder) # ═════════════════════════════════════════════════════════════════════════════ # 14. ProvenanceTracker – now exported from semantica.kg (0.3.0-beta fix) # ═════════════════════════════════════════════════════════════════════════════ class TestProvenanceTrackerExport: """ProvenanceTracker must be importable from semantica.kg after the fix.""" def test_provenance_tracker_importable_from_kg(self): from semantica.kg import ProvenanceTracker tracker = ProvenanceTracker() assert tracker is not None def test_provenance_tracker_track_entity(self): from semantica.kg import ProvenanceTracker tracker = ProvenanceTracker() tracker.track_entity("E1", "wikipedia:Apple_Inc", metadata={"type": "company"}) tracker.track_entity("E1", "crunchbase:apple-inc", metadata={"type": "company"}) sources = tracker.get_all_sources("E1") assert len(sources) == 2 source_ids = [s["source"] for s in sources] assert "wikipedia:Apple_Inc" in source_ids assert "crunchbase:apple-inc" in source_ids def test_provenance_tracker_clear(self): from semantica.kg import ProvenanceTracker tracker = ProvenanceTracker() tracker.track_entity("E2", "source_a") tracker.clear() sources = tracker.get_all_sources("E2") assert sources == [] # ═════════════════════════════════════════════════════════════════════════════ # 15. Semantic Relation Extraction – duplicate-relation fix (0.3.0-beta) # ═════════════════════════════════════════════════════════════════════════════ class TestSemanticExtractRelationFix: """_parse_relation_result must not create duplicate relations (0.3.0-beta fix).""" def test_llm_relation_not_duplicated(self): from semantica.semantic_extract.ner_extractor import Entity from semantica.semantic_extract.schemas import RelationOut, RelationsResponse from semantica.semantic_extract.methods import extract_relations_llm entities = [ Entity(text="OpenAI", label="ORG", start_char=0, end_char=6, confidence=1.0), Entity(text="ChatGPT", label="PRODUCT", start_char=10, end_char=17, confidence=1.0), ] class _FakeLLM: def is_available(self): return True def generate_typed(self, prompt, schema, **kw): return RelationsResponse(relations=[ RelationOut(subject="OpenAI", predicate="created", object="ChatGPT", confidence=0.98) ]) with patch("semantica.semantic_extract.methods.create_provider", return_value=_FakeLLM()): relations = extract_relations_llm( "OpenAI created ChatGPT.", entities=entities, provider="openai", model="gpt-4", ) # Must be exactly 1 relation, not 2 assert len(relations) == 1 assert relations[0].subject.text == "OpenAI" assert relations[0].object.text == "ChatGPT" assert relations[0].predicate == "created" def test_extraction_method_typed_metadata(self): from semantica.semantic_extract.ner_extractor import Entity from semantica.semantic_extract.schemas import RelationOut, RelationsResponse from semantica.semantic_extract.methods import extract_relations_llm entities = [ Entity(text="Anthropic", label="ORG", start_char=0, end_char=9, confidence=1.0), Entity(text="Claude", label="PRODUCT", start_char=13, end_char=19, confidence=1.0), ] class _FakeLLM: def is_available(self): return True def generate_typed(self, prompt, schema, **kw): return RelationsResponse(relations=[ RelationOut(subject="Anthropic", predicate="built", object="Claude", confidence=0.99) ]) with patch("semantica.semantic_extract.methods.create_provider", return_value=_FakeLLM()): relations = extract_relations_llm( "Anthropic built Claude.", entities=entities, provider="anthropic", model="claude-3-5-sonnet-20241022", structured_output_mode="typed", ) assert len(relations) == 1 assert relations[0].metadata["extraction_method"] == "llm_typed" def test_ghost_entity_resolved_from_full_list(self): """Entity filtered from prompt (due to max_entities_prompt) is still matched from original list.""" from semantica.semantic_extract.ner_extractor import Entity from semantica.semantic_extract.schemas import RelationOut, RelationsResponse from semantica.semantic_extract.methods import extract_relations_llm # 120 long names + 1 ghost entity not mentioned in text # (will be filtered from prompt since it's not in text, but must still be matched) entities = [ Entity(text=f"VeryLongOrgName{i}", label="ORG", start_char=0, end_char=1, confidence=1.0) for i in range(120) ] ghost = Entity(text="GhostCorp", label="ORG", start_char=0, end_char=9, confidence=1.0) entities.append(ghost) captured = {} class _FakeLLM: def is_available(self): return True def generate_typed(self, prompt, schema, **kw): captured["prompt"] = prompt return RelationsResponse(relations=[ RelationOut(subject="GhostCorp", predicate="acquired", object="VeryLongOrgName0", confidence=0.9) ]) with patch("semantica.semantic_extract.methods.create_provider", return_value=_FakeLLM()): relations = extract_relations_llm( "Short text about VeryLongOrgName0 only.", entities=entities, provider="openai", model="gpt-4", max_entities_prompt=20, ) assert "GhostCorp" not in captured["prompt"] assert len(relations) == 1 assert relations[0].subject.text == "GhostCorp" # ═════════════════════════════════════════════════════════════════════════════ # 16. ContextGraph – build from conversations (real dialog data) # ═════════════════════════════════════════════════════════════════════════════ class TestContextGraphFromConversations: """Build a context graph from structured conversation data.""" def _sample_conversations(self): return [ { "id": "conv_001", "content": "Discussion about Apple's new M4 chip and its AI capabilities.", "timestamp": datetime.now().isoformat(), "entities": [ {"id": "apple_e", "text": "Apple", "type": "company"}, {"id": "m4_chip", "text": "M4 chip", "type": "product"}, ], "relationships": [ {"source_id": "apple_e", "target_id": "m4_chip", "type": "manufactures", "confidence": 0.95}, ], }, { "id": "conv_002", "content": "OpenAI GPT-4o release discussion and comparison with Claude 3.5.", "timestamp": datetime.now().isoformat(), "entities": [ {"id": "openai_e", "text": "OpenAI", "type": "company"}, {"id": "gpt4o", "text": "GPT-4o", "type": "product"}, {"id": "claude35", "text": "Claude 3.5", "type": "product"}, ], "relationships": [ {"source_id": "openai_e", "target_id": "gpt4o", "type": "released", "confidence": 1.0}, ], }, ] def test_build_from_conversations(self): g = ContextGraph() result = g.build_from_conversations(self._sample_conversations(), link_entities=False) assert "nodes" in result assert len(g.nodes) > 0 def test_conversation_entities_in_graph(self): g = ContextGraph() g.build_from_conversations(self._sample_conversations(), link_entities=False) assert g.has_node("apple_e") or g.has_node("m4_chip") def test_has_conversation_nodes(self): g = ContextGraph() g.build_from_conversations(self._sample_conversations(), link_entities=False) conv_nodes = g.find_nodes(node_type="conversation") assert len(conv_nodes) >= 2 # ═════════════════════════════════════════════════════════════════════════════ # 17. Multi-hop reasoning: Microsoft → OpenAI → ChatGPT investment chain # ═════════════════════════════════════════════════════════════════════════════ class TestMultiHopInvestmentChain: """End-to-end context graph tracking of investment → product chain.""" def test_microsoft_chatgpt_2hop_path(self): g = _build_tech_graph() # microsoft --invested_in--> openai --makes--> chatgpt neighbors_2hop = g.get_neighbors("microsoft", hops=2) ids_2hop = {n["id"] for n in neighbors_2hop} assert "chatgpt" in ids_2hop def test_google_claude_2hop_path(self): g = _build_tech_graph() # google --invested_in--> anthropic --makes--> claude neighbors_2hop = g.get_neighbors("google", hops=2) ids_2hop = {n["id"] for n in neighbors_2hop} assert "claude" in ids_2hop def test_decision_chain_investment_impact(self): """Causal chain: investment decision → product launch → market decision.""" g = ContextGraph() d_invest = g.record_decision( category="investment", scenario="Microsoft $10B investment in OpenAI - strategic AI partnership", reasoning="Generative AI leadership; Azure integration; GPT models embedded in Office 365.", outcome="approved", confidence=0.97, decision_maker="msft_board", ) d_product = g.record_decision( category="product_launch", scenario="Copilot integration into Microsoft 365 suite", reasoning="Leverage OpenAI GPT-4 capabilities; $30/seat/month premium tier.", outcome="launched", confidence=0.93, decision_maker="satya_nadella", ) d_market = g.record_decision( category="market_strategy", scenario="Azure OpenAI Service GA launch for enterprise customers", reasoning="Capture enterprise AI infrastructure spend; compete with AWS Bedrock.", outcome="launched", confidence=0.91, decision_maker="azure_gm", ) g.add_causal_relationship(d_invest, d_product, "CAUSED") g.add_causal_relationship(d_product, d_market, "INFLUENCED") chain_down = g.get_causal_chain(d_invest, direction="downstream", max_depth=5) chain_ids = {d.decision_id for d in chain_down} assert d_product in chain_ids assert d_market in chain_ids # ═════════════════════════════════════════════════════════════════════════════ # 18. Healthcare context graph – treatment decision tracking # ═════════════════════════════════════════════════════════════════════════════ class TestHealthcareContextGraphE2E: """Real-world healthcare domain: treatment decisions with policy compliance.""" def test_treatment_decision_lifecycle(self): g = ContextGraph() # Record initial diagnosis decision diag_id = g.record_decision( category="diagnosis", scenario=( "Patient: 58yo male, BP 165/95, total cholesterol 268, LDL 180, " "10yr CVD risk 18% (ACC/AHA pooled cohort). Diagnosis: Stage 2 hypertension + dyslipidemia." ), reasoning=( "JNC8 guidelines met for Stage 2 HTN (>160 systolic). " "ATP III criteria: high-risk patient requiring statin therapy." ), outcome="diagnosed_stage2_htn_dyslipidemia", confidence=0.94, entities=["patient_hc_7712", "bp_reading_001", "lipid_panel_001"], decision_maker="cardiology_ai_v2", ) # Record treatment plan decision treat_id = g.record_decision( category="treatment", scenario="Prescribe ACE inhibitor + high-intensity statin for Stage 2 HTN + dyslipidemia", reasoning=( "Lisinopril 10mg/day for BP control; Atorvastatin 40mg/day for LDL reduction to <100. " "ACC/AHA Class I recommendation. No contraindications identified." ), outcome="treatment_prescribed", confidence=0.92, entities=["patient_hc_7712", "lisinopril_rx_001", "atorvastatin_rx_001"], decision_maker="cardiology_ai_v2", ) # Record follow-up decision followup_id = g.record_decision( category="followup", scenario="6-week follow-up: BP 138/84, LDL 95. Treatment response assessment.", reasoning="Target BP <140/90 achieved. LDL <100 achieved. Continue current regimen.", outcome="continue_treatment", confidence=0.96, decision_maker="cardiology_ai_v2", ) g.add_causal_relationship(diag_id, treat_id, "CAUSED") g.add_causal_relationship(treat_id, followup_id, "INFLUENCED") # Verify causal chain chain = g.get_causal_chain(diag_id, direction="downstream", max_depth=5) chain_ids = {d.decision_id for d in chain} assert treat_id in chain_ids assert followup_id in chain_ids # Verify precedent search g.add_causal_relationship(diag_id, treat_id, "PRECEDENT_FOR") precedents = g.find_precedents(treat_id, limit=5) assert len(precedents) >= 1 def test_policy_compliance_check(self): """Policy engine: check treatment against clinical policy.""" g = ContextGraph() # Add a clinical policy node policy_node_id = "clinical_policy_htn_2024" g.add_node( policy_node_id, node_type="Policy", content="Hypertension Management Policy 2024", policy_id=policy_node_id, category="treatment", rules={"min_bp_systolic_threshold": 140, "first_line_agent": "ACE_inhibitor"}, version="2024.1", ) assert g.has_node(policy_node_id) policy_nodes = g.find_nodes(node_type="Policy") assert len(policy_nodes) == 1