feat: enhance database impact analysis and ownership tracking

.github/copilot-instructions.md

@@ -0,0 +1,84 @@
+# CodeLogic MCP Server - Copilot Instructions
+
+## About This Codebase
+
+This repository contains a Model Context Protocol (MCP) server that integrates with CodeLogic's knowledge graph APIs. It enables AI programming assistants (like GitHub Copilot) to access dependency data from CodeLogic to analyze code and database impacts.
+
+## Key Technologies
+
+- **Python 3.13+** with extensive use of async/await
+- **Model Context Protocol SDK** (`mcp[cli]`)
+- **Neo4j** for graph database operations
+- **HTTPX** for API requests
+- **Environment variables** via dotenv for configuration
+
+## Project Structure
+
+- **src/codelogic_mcp_server/**: Core package
+- **`__init__.py`**: Package initialization and entry point  
+- **`server.py`**: MCP server implementation  
+- **`handlers.py`**: Tool handlers implementation  
+- **`utils.py`**: API interaction utilities  
+
+## Core Coding Patterns
+
+### MCP Server Pattern
+
+```python
+server = Server("codelogic-mcp-server")
+
+@server.list_tools()
+async def handle_list_tools() -> list[types.Tool]:
+    # Define and return tools
+
+@server.call_tool()
+async def handle_call_tool(name: str, arguments: dict | None) -> list[types.TextContent]:
+    # Handle tool execution
+```
+
+### Error Handling
+
+```python
+try:
+    # Operations that might fail
+except Exception as e:
+    sys.stderr.write(f"Error: {str(e)}\n")
+    return [types.TextContent(type="text", text=f"# Error\n\n{str(e)}")]
+```
+
+## Style Guidelines
+
+1. **Copyright Headers**: Include MPL 2.0 headers in all Python files
+2. **Docstrings**: Google-style docstrings for modules/classes/functions
+3. **Type Hints**: Always use Python type hints
+4. **Asynchronous**: Keep I/O operations asynchronous
+5. **Format Outputs**: Return markdown-formatted text in tool responses
+
+## Tool Implementation Pattern
+
+When implementing new MCP tools:
+
+1. Add to `handle_list_tools()` with descriptive name (prefix: `codelogic-`)
+2. Add handler in `handle_call_tool()`
+3. Implement handler function with error handling
+4. Return results as markdown-formatted text
+
+## Testing Approach
+
+- **Unit Tests**: For functions without external dependencies
+- **Integration Tests**: For tests against a real CodeLogic server
+- Use the `CODELOGIC_TEST_MODE` environment variable
+
+## Debugging
+
+- Debug Mode: Set `CODELOGIC_DEBUG_MODE=true`
+- Remote Debugging: Use debugpy capabilities
+
+## Key Environment Variables
+
+- `CODELOGIC_SERVER_HOST`: CodeLogic server URL
+- `CODELOGIC_USERNAME`: Username for authentication
+- `CODELOGIC_PASSWORD`: Password for authentication
+- `CODELOGIC_MV_NAME`: Materialized view name
+- `CODELOGIC_DEBUG_MODE`: Enable debug logging
+- `CODELOGIC_TEST_MODE`: Used by test framework

pyproject.toml

@@ -5,7 +5,14 @@ description = "Integrates CodeLogic's powerful codebase knowledge graphs with a
 readme = "README.md"
 license = "MPL-2.0"
 requires-python = ">=3.13"
-dependencies = [ "debugpy>=1.8.12", "httpx>=0.28.1", "mcp[cli]>=1.3.0", "neo4j>=5.28.1", "pip-licenses>=5.0.0", "python-dotenv>=1.0.1", "tenacity>=9.0.0",]
+dependencies = [
+ "debugpy>=1.8.12",
+ "httpx>=0.28.1",
+ "mcp[cli]>=1.3.0",
+ "pip-licenses>=5.0.0",
+ "python-dotenv>=1.0.1",
+ "tenacity>=9.0.0",
+]
 [[project.authors]]
 name = "garrmark"
 email = "[email protected]"

src/codelogic_mcp_server/handlers.py

@@ -64,7 +64,7 @@ async def handle_list_tools() -> list[types.Tool]:
                         "1. Use this tool before implementing code or database changes\n"
                         "2. Search for the relevant database entity\n"
                         "3. Review the impact analysis to understand which code depends on this database object and vice versa\n"
-                        "Particularly crucial when AI-suggested modifications are being considered.",
+                        "Particularly crucial when AI-suggested modifications are being considered or when modifying SQL code.",
             inputSchema={
                 "type": "object",
                 "properties": {
@@ -224,6 +224,22 @@ async def handle_method_impact(arguments: dict | None) -> list[types.TextContent
     complexity = target_node['properties'].get('statistics.cyclomaticComplexity', 'N/A') if target_node else 'N/A'
     instruction_count = target_node['properties'].get('statistics.instructionCount', 'N/A') if target_node else 'N/A'
 
+    # Extract code owners and reviewers
+    code_owners = target_node['properties'].get('codelogic.owners', []) if target_node else []
+    code_reviewers = target_node['properties'].get('codelogic.reviewers', []) if target_node else []
+
+    # If target node doesn't have owners/reviewers, try to find them from the class or file node
+    if not code_owners or not code_reviewers:
+        class_node = None
+        if class_name:
+            class_node = next((n for n in nodes if n['primaryLabel'].endswith('ClassEntity') and class_name.lower() in n['name'].lower()), None)
+
+        if class_node:
+            if not code_owners:
+                code_owners = class_node['properties'].get('codelogic.owners', [])
+            if not code_reviewers:
+                code_reviewers = class_node['properties'].get('codelogic.reviewers', [])
+
     # Identify dependents (systems that depend on this method)
     dependents = []
 
@@ -391,11 +407,18 @@ async def handle_method_impact(arguments: dict | None) -> list[types.TextContent
 ## Summary
 - **Method**: `{method_name}`
 - **Class**: `{class_name or 'N/A'}`
-- **Complexity**: {complexity}
-- **Instruction Count**: {instruction_count}
-- **Affected Applications**: {len(affected_applications)}
 """
 
+    # Add code ownership information if available
+    if code_owners:
+        impact_description += f"- **Code Owners**: {', '.join(code_owners)}\n"
+    if code_reviewers:
+        impact_description += f"- **Code Reviewers**: {', '.join(code_reviewers)}\n"
+
+    impact_description += f"- **Complexity**: {complexity}\n"
+    impact_description += f"- **Instruction Count**: {instruction_count}\n"
+    impact_description += f"- **Affected Applications**: {len(affected_applications)}\n"
+
     # Add affected REST endpoints to the Summary section
     if endpoint_nodes:
         impact_description += "\n### Affected REST Endpoints\n"
@@ -461,6 +484,17 @@ async def handle_method_impact(arguments: dict | None) -> list[types.TextContent
         impact_description += "\n### REST API Risk Assessment\n"
         impact_description += "✅ No direct impact on REST endpoints or API controllers detected\n"
 
+    # Ownership-based consultation recommendation
+    if code_owners or code_reviewers:
+        impact_description += "\n### Code Ownership\n"
+        if code_owners:
+            impact_description += f"👤 **Code Owners**: Changes to this code should be reviewed by: {', '.join(code_owners)}\n"
+        if code_reviewers:
+            impact_description += f"👁️ **Preferred Reviewers**: Consider getting reviews from: {', '.join(code_reviewers)}\n"
+
+        if code_owners:
+            impact_description += "\nConsult with the code owners before making significant changes to ensure alignment with original design intent.\n"
+
     impact_description += f"""
 ## Method Impact
 This analysis focuses on systems that depend on `{method_name}`. Modifying this method could affect these dependents:

src/codelogic_mcp_server/server.py

@@ -57,8 +57,9 @@ async def main():
                         experimental_capabilities={},
                     ),
                     instructions=("When asked to modify an existing code method, invoke one or both of codelogic-method-impact and codelogic-database-impact "
-                                  "mcp tools against the code provided and any methods that you are proposing changes to.  Make sure the results sent back "
-                                  "from the tool are highlighted as impacts for the given method."
+                                  "mcp tools against the code provided and any methods that you are proposing changes to. When modifying SQL code or database entities, "
+                                  "always use the codelogic-database-impact tool to analyze potential impacts. Make sure the results sent back "
+                                  "from the tool are highlighted as impacts for the given method or database entity."
                                   ),
                 ),
             )

src/codelogic_mcp_server/utils.py

@@ -493,6 +493,33 @@ def process_database_entity_impact(impact_data, entity_type, entity_name, entity
         if app not in dependent_applications:
             dependent_applications.append(app)
 
+    # Extract code owners and reviewers from the dependent code entities
+    # Since database entities don't typically have ownership metadata directly,
+    # we'll gather this information from the code entities that reference them
+    code_owners = set()
+    code_reviewers = set()
+
+    # Check code entities that reference this database entity
+    for code_item in dependent_code:
+        code_id = code_item.get("id")
+        code_node = next((n for n in nodes if n['id'] == code_id), None)
+        if code_node:
+            owners = code_node.get('properties', {}).get('codelogic.owners', [])
+            reviewers = code_node.get('properties', {}).get('codelogic.reviewers', [])
+            code_owners.update(owners)
+            code_reviewers.update(reviewers)
+
+            # Look for parent classes that might contain ownership info
+            for rel in impact_data.get('data', {}).get('relationships', []):
+                if rel.get('type').startswith('CONTAINS_') and rel.get('endId') == code_id:
+                    parent_id = rel.get('startId')
+                    parent_node = find_node_by_id(impact_data.get('data', {}).get('nodes', []), parent_id)
+                    if parent_node and parent_node.get('primaryLabel', '').endswith('ClassEntity'):
+                        parent_owners = parent_node.get('properties', {}).get('codelogic.owners', [])
+                        parent_reviewers = parent_node.get('properties', {}).get('codelogic.reviewers', [])
+                        code_owners.update(parent_owners)
+                        code_reviewers.update(parent_reviewers)
+
     return {
         "entity_type": entity_type,
         "name": entity_name,
@@ -501,6 +528,8 @@ def process_database_entity_impact(impact_data, entity_type, entity_name, entity
         "referencing_tables": referencing_tables,
         "dependent_applications": dependent_applications,
         "parent_table": parent_table,
+        "code_owners": list(code_owners),
+        "code_reviewers": list(code_reviewers),
         "nodes": nodes,
         "relationships": relationships
     }
@@ -721,6 +750,21 @@ def generate_combined_database_report(entity_type, search_name, table_or_view, s
 
         report += f"### {i + 1}. {entity_type.capitalize()}: {entity_id}\n\n"
 
+        # Add code ownership information if available
+        code_owners = impact.get("code_owners", [])
+        code_reviewers = impact.get("code_reviewers", [])
+
+        if code_owners or code_reviewers:
+            report += "#### Code Ownership\n"
+            if code_owners:
+                report += f"👤 **Code Owners**: {', '.join(code_owners)}\n"
+            if code_reviewers:
+                report += f"👁️ **Preferred Reviewers**: {', '.join(code_reviewers)}\n"
+            if code_owners:
+                report += "\nConsult with the code owners before making significant changes to ensure alignment with original design intent.\n\n"
+            else:
+                report += "\n"
+
         # For columns, show the parent table information
         parent_table = impact.get("parent_table")
         if parent_table and entity_type == "column":