feat: initial query engine for agents md

[JayGhiya]

PR Type

Enhancement

---

Description

• Implements comprehensive agent-based codebase analysis system with streaming SSE support
• Creates unified agent execution service with concurrent processing across multiple codebases
• Adds five specialized agents (directory, framework, workflow, business logic, context7) with factory pattern
• Implements multi-provider AI model configuration system with hot-reload capabilities
• Provides ripgrep-based codebase search and structural analysis tools
• Creates post-processing pipeline for enriching agent outputs with database information
• Adds comprehensive REST API endpoints for configuration management and streaming analysis
• Implements MCP (Model Context Protocol) server management for tool integration
• Establishes encrypted credentials management and feature flag system
• Provides complete Docker containerization and development workflow automation

---

Diagram Walkthrough

flowchart LR
  API["SSE API Endpoint"] --> AES["Agent Execution Service"]
  AES --> AF["Agent Factory"]
  AF --> DA["Directory Agent"]
  AF --> FA["Framework Agent"] 
  AF --> WA["Workflow Agent"]
  AF --> BLA["Business Logic Agent"]
  AF --> C7["Context7 Agent"]
  
  MF["Model Factory"] --> AES
  AMC["AI Model Config"] --> MF
  
  Tools["Codebase Tools"] --> DA
  Tools --> FA
  Tools --> WA
  Tools --> BLA
  
  PP["Post Processors"] --> AES
  Neo4j --> Tools
  PostgreSQL --> AMC
  
  MCP["MCP Server Manager"] --> Tools

Loading

File Walkthrough

	Relevant files
Enhancement	45 files agent_execution_service.py Unified agent execution service with streaming support      unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/services/agent_execution_service.py • Implements unified agent execution service consolidating duplicated logic for streaming and non-streaming execution • Provides concurrent execution of agents across multiple codebases with standardized SSE event streaming • Includes comprehensive error handling, timeout management, and optional post-processing capabilities • Features producer-consumer pattern for event streaming with proper completion signaling +703/-0  codebase_agent_rules.py SSE endpoint for streaming codebase agent analysis              unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/api/v1/endpoints/codebase_agent_rules.py • Implements SSE endpoint for streaming codebase agent rules with real-time progress updates • Orchestrates sequential execution of four agents (directory, framework, workflow, business logic) with aggregation • Includes connection monitoring, timeout handling, and baseline framework merging functionality • Provides comprehensive logging and debugging for SSE connection lifecycle management +483/-0  search_across_codebase.py Ripgrep-based codebase search tool implementation                unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/tools/search_across_codebase.py • Implements ripgrep-based codebase search tool with regex/literal pattern matching support • Provides file filtering, case sensitivity options, and contextual preview capabilities • Includes comprehensive error handling, timeout management, and JSON output parsing • Features security validation to ensure search paths remain within codebase boundaries +368/-0  code_confluence_agents.py Code confluence agents factory with specialized agent definitions unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/agents/code_confluence_agents.py • Creates factory function for instantiating code confluence agents with provided model and settings • Defines five specialized agents: directory, framework explorer, context7, development workflow, and business logic domain • Each agent includes specific system prompts, tool configurations, and output type specifications • Integrates MCP server manager for tool integration and supports model settings configuration +268/-0  model_factory.py Pydantic AI model factory with multi-provider support        unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/services/model_factory.py • Implements factory pattern for creating Pydantic AI models from configuration • Supports multiple providers including OpenAI, Anthropic, Google, Groq, Mistral, Cohere, HuggingFace • Handles both native providers and OpenAI-compatible providers with credential management • Includes comprehensive provider configuration with API key handling and custom settings +270/-0  config_hot_reload.py AI model configuration hot-reload service implementation  unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/services/config_hot_reload.py • Implements ORM events service for hot-reload functionality of AI model configuration • Provides lazy rebuilding of models when configuration changes are detected • Includes SQLAlchemy event handlers for tracking configuration changes and invalidation • Features application agent updating mechanism for configuration refresh +278/-0  provider_catalog.py AI model provider catalog with configuration schemas          unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/services/provider_catalog.py • Defines catalog of available AI model providers with configuration schemas • Includes provider field definitions for UI form generation with validation rules • Supports both native providers and OpenAI-compatible providers with specific configurations • Provides utility methods for provider lookup and schema retrieval +229/-0  ai_model_config_service.py AI model configuration service with database operations    unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/services/ai_model_config_service.py • Implements service for AI model configuration database operations with single-record approach • Provides CRUD operations for model configuration with credential management integration • Includes provider validation using catalog and automatic provider kind inference • Features comprehensive error handling and logging for database operations +259/-0  agent_prompt_registry.py Centralized agent prompt registry with XML formatting        unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/services/agent_prompt_registry.py • Implements centralized registry for agent-specific prompt templates • Provides XML-formatted context injection using pydantic_ai.format_as_xml for better model compliance • Includes specialized prompts for framework, directory, workflow, and business logic agents • Features flexible context formatting for BaseModel, list, and string inputs +164/-0  agent_md_output.py Pydantic models for codebase analysis output schema            unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/models/agent_md_output.py • Defines comprehensive Pydantic models for codebase analysis output • Includes models for project structure, framework usage, development workflow, and business logic • Features extended enums and supporting models with strict validation rules • Provides complete schema for agent markdown output with proper field descriptions +201/-0  mcp_server_manager.py MCP Server Manager Service Implementation                                unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/services/mcp/mcp_server_manager.py • Implements MCP server lifecycle management using PydanticAI's MCPServerStdio and MCPServerSSE • Supports both local (stdio) and remote (SSE) MCP server configurations • Provides methods for loading config, starting/stopping servers, and retrieving active servers +203/-0  ai_model_config.py AI Model Configuration API Endpoints                                          unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/api/v1/endpoints/ai_model_config.py • Adds REST API endpoints for AI model configuration management (GET, PUT, DELETE) • Includes provider catalog endpoints for listing available AI providers • Implements automatic agent refresh when configuration is updated +193/-0  framework_baseline_service.py Framework Baseline Service Implementation                                unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/services/framework_baseline_service.py • Fetches baseline framework data from Neo4j and maps to output models • Enriches framework metadata from PostgreSQL with descriptions and documentation URLs • Filters frameworks to include only those with actual usage locations +183/-0  get_framework_lib_feature_overview.py Framework Library Feature Overview Tool                                    unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/tools/get_framework_lib_feature_overview.py • Provides tool for retrieving comprehensive framework/library feature overview from Neo4j • Returns detailed feature usage information including file locations • Transforms raw Neo4j data into structured FrameworkSummary models +180/-0  credentials_service.py Encrypted Credentials Management Service                                  unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/services/credentials_service.py • Manages encrypted credentials via shared credentials table • Provides CRUD operations with encryption/decryption using Fernet • Supports both standalone and session-based database operations +166/-0  repository_metadata_service.py Repository Metadata Service Implementation                              unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/db/repository_metadata_service.py • Fetches repository and codebase configuration data from PostgreSQL • Resolves absolute paths from Neo4j using qualified repository names • Enriches metadata with programming language information +149/-0  agent_md_aggregate.py Agent Metadata Aggregator Builder Class                                    unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/models/agent_md_aggregate.py • Builder class for accumulating agent results into final AgentMdOutput • Provides methods to update from different agent types (directory, framework, workflow, business logic) • Creates complete output with defaults for missing fields +140/-0  flag_service.py Feature Flag Management Service                                                    unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/services/flag_service.py • Service for managing feature flags using the existing commons Flag model • Provides CRUD operations for flags with proper error handling • Supports flag status checking and existence validation +151/-0  flags.py Feature Flags API Endpoints                                                            unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/api/v1/endpoints/flags.py • REST API endpoints for feature flag management (GET, PUT, DELETE) • Follows upsert pattern for flag creation/updates • Provides both individual flag and bulk flag operations +142/-0  codebase_path_resolver.py Codebase Path Resolution Service                                                  unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/services/codebase_path_resolver.py • Resolves absolute codebase paths from Neo4j graph database • Maps relative paths from PostgreSQL to absolute paths from Neo4j • Implements flexible path matching logic for different naming conventions +144/-0  library_documentation_service.py Library Documentation Service Implementation                          unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/services/library_documentation_service.py • Service for retrieving library/framework documentation using Context7 agent • Supports both general library overviews and specific feature documentation • Handles developer tools with command and configuration information +126/-0  framework_explorer_post_processor.py Framework Explorer Post-Processor Implementation                  unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/services/post_processors/framework_explorer_post_processor.py • Post-processor for framework explorer agent output • Merges database frameworks with novel agent-discovered frameworks • Filters frameworks based on location data and database coverage +132/-0  get_structural_signature.py Structural Signature Retrieval Tool                                            unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/tools/get_structural_signature.py • Tool for retrieving structural signature and imports from Neo4j knowledge graph • Returns complete file structural information including functions, classes, and variables • Validates file paths and handles parsing errors gracefully +121/-0  framework_overview_repository.py Framework Overview Neo4j Repository                                            unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/db/neo4j/framework_overview_repository.py • Neo4j repository for framework/library overview queries • Centralizes graph reads for tools and services reuse • Provides transaction functions for codebase existence checks and framework feature queries +102/-0  get_directory_tree.py Directory Tree Generation Tool                                                      unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/tools/get_directory_tree.py • Tool for generating text-based directory trees using the eza utility • Supports depth limiting and custom path specification • Provides comprehensive error handling for missing dependencies and permissions +109/-0  package_manager_metadata_service.py Package Manager Metadata Service                                                  unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/services/package_manager_metadata_service.py • Service for fetching package manager metadata from Neo4j graph • Converts Neo4j records to ProgrammingLanguageMetadataOutput models • Handles missing or incomplete metadata gracefully +114/-0  get_content_file.py File Content Reading Tool                                                                unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/tools/get_content_file.py • Tool for reading file content with optional line range filtering • Supports both full file reading and specific line range extraction • Includes comprehensive validation and error handling for file operations +104/-0  mcp_config.py MCP Server Configuration Models                                                    unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/models/mcp_config.py • Pydantic models for MCP server configuration • Supports both local (stdio) and remote (SSE) server types using discriminated unions • Includes validation for commands, URLs, and configuration parameters +103/-0  agent_logs.py Agent Logging Utilities                                                                    unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/utils/agent_logs.py • Utilities for agent logging paths and node serialization • Provides project root detection and logs directory resolution • Handles async JSON serialization of agent execution nodes +92/-0    db.py PostgreSQL Database Connection Manager                                      unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/db/postgres/db.py • Async PostgreSQL database connection management using SQLAlchemy • Implements scoped sessions for concurrent agent operations • Provides connection initialization and disposal with proper cleanup +91/-0    business_logic_repository.py Business Logic Neo4j Repository                                                    unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/db/neo4j/business_logic_repository.py • Neo4j repository for querying data model files • Uses both direct detection (has_data_model=true) and feature-based detection • Follows async session pattern for consistent database access +68/-0    business_logic_domain_post_processor.py Business Logic Domain Post-Processor                                          unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/services/post_processors/business_logic_domain_post_processor.py • Post-processor for business logic domain agent output • Enriches agent description with core files from database • Creates CoreFile objects from data model file paths +70/-0    agent_execution_request.py Agent Execution Request Models and Protocols                          unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/models/agent_execution_request.py • Core types and protocols for unified agent execution service • Defines AgentExecutionRequest dataclass and protocol interfaces • Provides abstractions for prompt providers and tool message policies +65/-0    connection_manager.py Neo4j Connection Manager for Query Engine                                unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/db/neo4j/connection_manager.py • Async Neo4j connection manager using neomodel for query operations • Provides managed transaction support for read operations • Focuses on query operations without schema creation responsibilities +59/-0    post_processing_service.py Post-Processing Orchestrator Service                                          unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/services/post_processing_service.py • Post-processing orchestrator service with registry pattern • Registers built-in processors for framework explorer and business logic domain • Adapts processor interfaces and manages execution dependencies +61/-0    get_lib_data.py Library Documentation Retrieval Tool                                          unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/tools/get_lib_data.py • Tool for retrieving library/framework documentation using LibraryDocumentationService • Provides simplified interface to Context7-powered documentation lookup • Supports both general library overviews and specific feature descriptions +53/-0    tool_message_policy.py Tool Message Policy Implementation                                              unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/services/tool_message_policy.py • Tool message policies for consistent messaging across agents • Provides default messages with tool-specific and agent-specific overrides • Implements ToolMessagePolicyProtocol for standardized tool communication +48/-0    get_data_model_files.py Data Model Files Retrieval Tool                                                    unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/tools/get_data_model_files.py • Tool for fetching data model file paths from Neo4j • Returns files marked as data models or using data model framework features • Provides guidance for agents on how to process the returned file paths +55/-0    get_core_files.py Core Files with Context Retrieval Tool                                      unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/tools/get_core_files.py • Tool for fetching core files (data models) with additional context • Returns enriched file information including package, features, signature, and imports • Supports domain filtering for focused analysis +53/-0    ai_model_config.py AI Model Configuration Database Model                                        unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/db/postgres/ai_model_config.py • SQLModel for AI model provider configuration storage • Single record approach with comprehensive configuration fields • Includes provider metadata, model settings, and timestamps +33/-0    ai_model_config.py AI Model Configuration Pydantic Models                                      unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/models/ai_model_config.py • Pydantic models for AI model configuration API • Defines input/output models with validation constraints • Includes provider kind enumeration and configuration fields +41/-0    post_processor_base.py Create base protocol and dependencies for post-processors unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/services/post_processors/post_processor_base.py • Defines ProcessorDependencies class with codebase path, Neo4j connection, and programming language • Creates PostProcessorProtocol generic protocol for agent output transformation • Establishes base structure for post-processing agent outputs with type safety +38/-0    agent_dependencies.py Add agent dependencies dataclass for PydanticAI agents      unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/models/agent_dependencies.py • Defines AgentDependencies dataclass for PydanticAI agent dependencies • Includes repository metadata, Neo4j connection, context7 agent, and library documentation service • Follows best practices for dependency injection in agent execution +32/-0    registry.py Create registry for managing agent post-processors              unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/services/post_processors/registry.py • Implements PostProcessingRegistry class for managing post-processors by agent name • Provides register method to associate processors with specific agents • Includes get method to retrieve processors for specific agents +24/-0    repository_ruleset_metadata.py Add repository and codebase metadata models                            unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/models/repository_ruleset_metadata.py • Defines CodebaseMetadata model with name, path, and programming language fields • Creates RepositoryRulesetMetadata model containing repository name and codebase metadata list • Uses Pydantic models with descriptive field documentation +22/-0
Miscellaneous	4 files __init__.py MCP services package initialization                                            unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/services/mcp/init.py • Creates package initialization file for MCP (Model Context Protocol) services +1/-0      __init__.py Tools package initialization                                                          unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/tools/init.py • Creates empty package initialization file for tools module +1/-0      __init__.py Initialize utils package                                                                  unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/utils/init.py • Creates package initialization file for utility modules +1/-0      __init__.py Initialize post-processors package                                              unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/services/post_processors/init.py • Creates package initialization file for post-processor services +1/-0
Configuration changes	23 files settings.py Application Environment Settings Configuration                      unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/config/settings.py • Environment settings configuration using Pydantic Settings • Includes PostgreSQL, Neo4j, MCP, encryption, and logging configurations • Provides computed fields for connection URLs +82/-0    logging_config.py Logging Configuration Setup                                                            unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/config/logging_config.py • Configures Loguru logging for console and file output • Sets up log rotation, retention, and async-safe logging for FastAPI • Provides structured logging format with colorization for development +47/-0    .python-version Python Version Specification                                                          unoplat-code-confluence-query-engine/.python-version • Sets Python version requirement to 3.13 +1/-0      local-dependencies-docker-compose.yml Add Docker Compose configuration for local dependencies    unoplat-code-confluence-query-engine/local-dependencies-docker-compose.yml • Defines complete Docker Compose setup with Elasticsearch, PostgreSQL, Temporal, Neo4j services • Includes code-confluence-flow-bridge service with environment configuration • Sets up networking, volumes, and health checks for all services +177/-0  Taskfile.yml Add Taskfile for development workflow automation                  unoplat-code-confluence-query-engine/Taskfile.yml • Defines development tasks for dependency management, testing, and code quality • Includes tasks for starting/stopping services, running tests with coverage, and linting • Provides development workflow automation with uv package manager +101/-0  Dockerfile Add multi-stage Dockerfile for application containerization unoplat-code-confluence-query-engine/Dockerfile • Creates multi-stage Docker build with Bun and Python stages • Includes security best practices with non-root user and minimal runtime dependencies • Configures FastAPI application with proper environment variables +82/-0    pyproject.toml Add Python project configuration with dependencies              unoplat-code-confluence-query-engine/pyproject.toml • Defines Python project configuration with FastAPI, PydanticAI, and database dependencies • Includes development and test dependency groups with type checking tools • Configures pytest settings and custom markers for integration tests +52/-0    yaak.rq_zcJSFbvhRK.yaml Add HTTP request for model configuration                                  yak/yaak.rq_zcJSFbvhRK.yaml • Creates HTTP request configuration for setting model configuration • Configures Hugging Face provider with Kimi model and API key header +38/-0    ruff.toml Add Ruff linter configuration                                                        unoplat-code-confluence-query-engine/ruff.toml • Configures Ruff linter with Python 3.13 target and import/error checking rules • Sets up code formatting standards with 88 character line length • Defines per-file ignores for init.py files +37/-0    .dockerignore Add Docker ignore configuration                                                    unoplat-code-confluence-query-engine/.dockerignore • Excludes development artifacts, tests, and temporary files from Docker builds • Preserves essential configuration files for container builds +48/-0    mypy.ini Add MyPy type checker configuration                                            unoplat-code-confluence-query-engine/mypy.ini • Configures MyPy type checker with strict typing rules and error reporting • Relaxes strictness for test modules while maintaining type safety +21/-0    yaak.rq_TmmhvnQtTM.yaml Add HTTP request for providers endpoint                                    yak/yaak.rq_TmmhvnQtTM.yaml • Creates HTTP GET request configuration for retrieving providers +23/-0    yaak.rq_v59Nw6A6i7.yaml Update repository refresh request URL                                        yak/yaak.rq_v59Nw6A6i7.yaml • Updates URL from localhost to 127.0.0.1 for refresh-repository endpoint +2/-2      yaak.rq_Jp2qmok4Xs.yaml Update repository deletion request URL                                      yak/yaak.rq_Jp2qmok4Xs.yaml • Updates URL from localhost to 127.0.0.1 for delete-repository endpoint +2/-2      yaak.rq_DwTkeJP8FJ.yaml Update token ingestion request URL                                              yak/yaak.rq_DwTkeJP8FJ.yaml • Updates URL from localhost to 127.0.0.1 for ingest-token endpoint +2/-2      yaak.rq_EECnrR2aEe.yaml Update ingested repositories request URL                                  yak/yaak.rq_EECnrR2aEe.yaml • Updates URL from localhost to 127.0.0.1 for ingested repositories endpoint +2/-2      yaak.rq_ET5prEchnU.yaml Add HTTP request for configuration endpoint                            yak/yaak.rq_ET5prEchnU.yaml • Creates HTTP GET request configuration for retrieving configuration +18/-0    yaak.rq_bytnZBU5hq.yaml Add HTTP request for OpenAI provider schema                            yak/yaak.rq_bytnZBU5hq.yaml • Creates HTTP GET request configuration for OpenAI provider UI schema +18/-0    yaak.rq_yMsbRmdhHJ.yaml Update ingestion start request URL                                              yak/yaak.rq_yMsbRmdhHJ.yaml • Updates URL from localhost to 127.0.0.1 for start-ingestion endpoint +2/-2      yaak.fl_Pgb37JQfSX.yaml Update folder sort priority                                                            yak/yaak.fl_Pgb37JQfSX.yaml • Updates sort priority for code-confluence-query-engine folder +2/-2      yaak.fl_a9CJGFoCVq.yaml Update folder sort priority                                                            yak/yaak.fl_a9CJGFoCVq.yaml • Updates sort priority for code-confluence-flow-bridge folder +2/-2      yaak.fl_jtBaK7a5pU.yaml Add agent configuration folder                                                      yak/yaak.fl_jtBaK7a5pU.yaml • Creates new folder configuration for agent_configuration +13/-0    mcp-servers.json Add MCP servers configuration                                                        unoplat-code-confluence-query-engine/mcp-servers.json • Configures MCP servers for context7 and mcp-tasks with Bun execution • Defines server types, commands, and timeout settings +16/-0
Error handling	1 files agent_error_logger.py Add agent error logging utility for PydanticAI exceptions unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/utils/agent_error_logger.py • Creates a utility function log_agent_error for consistent agent error logging • Logs PydanticAI UnexpectedModelBehavior exceptions with timestamp and context • Includes optional context information and message history for debugging • Uses loguru logger with structured error information +38/-0
Additional files	11 files README.md [link]    __init__.py [link]    __init__.py [link]    __init__.py [link]    __init__.py [link]    __init__.py [link]    __init__.py [link]    __init__.py [link]    __init__.py [link]    __init__.py [link]    __init__.py [link]

---

[huly-for-github]

Connected to Huly®: UNOPL-893

[claude]

Claude finished @JayGhiya's task —— View job

---

Code Review for PR #756: Initial Query Engine for Agents

Conducted a comprehensive review of this large PR (86 files, 9926+ additions).

Review Progress

• Architecture Overview - Review overall structure and design patterns
• Core Services - Analyze main service classes and business logic
• Database Integration - Review Neo4j and PostgreSQL integration
• API Endpoints - Security, validation, error handling
• Models and Data Structures - Type safety and design
• Configuration Management - Settings, environment handling
• Tooling and Development Setup - Docker, testing, linting
• Security Analysis - Authentication, data handling, secrets
• Performance Considerations - Database queries, async patterns
• Test Coverage Assessment - Missing tests and recommendations
• Final Summary - Consolidate findings and recommendations

Review Findings

✅ Strengths

Architecture & Design

• Well-organized monorepo structure following domain-driven design
• Clean separation of concerns with distinct layers (API, services, models, DB)
• Proper use of dependency injection and factory patterns
• Good async/await patterns throughout codebase
• Proper use of Pydantic for data validation and serialization

Configuration & Environment

• Excellent use of Pydantic Settings for type-safe configuration
• Proper environment variable handling with sensible defaults
• Good separation of development and production configurations
• Comprehensive Docker and docker-compose setup

Development Tooling

• Comprehensive tooling setup (ruff, mypy, pytest)
• Well-configured Taskfile with development workflow automation
• Proper dependency management with uv
• Good logging configuration using loguru

⚠️ Areas of Concern

Security Issues

1. Hardcoded encryption key in settings.py:53-55

   token_encryption_key: SecretStr = Field(
       default=SecretStr("your-32-byte-fernet-key-here-replace-in-prod"),

   • Risk: High - Default encryption key is visible and weak
   • Recommendation: Remove default, require explicit configuration, add validation

2. Missing authentication on API endpoints

   • No authentication middleware or security headers visible
   • API endpoints in ai_model_config.py, codebase_agent_rules.py lack auth protection
   • Recommendation: Implement proper authentication/authorization

3. Potential credential exposure in logs

   • credentials_service.py:42 logs credential keys which could expose sensitive info
   • Recommendation: Use structured logging with credential sanitization

Performance & Scalability

1. Database connection management

   • Global connection variables in db.py:18-20 may cause issues under load
   • No connection pooling configuration visible
   • Recommendation: Review connection lifecycle and add proper pooling

2. Large file operations

   • Tools like search_across_codebase.py may timeout on large codebases
   • No rate limiting or resource constraints visible
   • Recommendation: Add timeout handling and result pagination

Code Quality Issues

1. Missing error handling in several places:

   • connection_manager.py:30-36 - Generic exception catching
   • No proper error types or error handling strategies

2. Large service classes

   • agent_execution_service.py is 703 lines - violates SRP
   • Multiple responsibilities mixed together
   • Recommendation: Break into smaller, focused services

3. Inconsistent async patterns

   • Mix of async context managers and direct async calls
   • Some services don't follow consistent async patterns

🚨 Critical Issues

1. Missing Test Coverage

• Critical: No tests directory exists despite test dependencies in pyproject.toml
• No unit tests, integration tests, or test utilities
• Impact: High risk of regressions, difficult to maintain
• Recommendation: Must add comprehensive test suite before merge

2. Missing Application Entry Point

• No main.py, app.py, or FastAPI application setup visible
• Unclear how the application boots up and wires dependencies
• Recommendation: Add proper application initialization

3. Incomplete Documentation

• README.md is empty
• No API documentation or usage examples
• Complex agent system with no explanation
• Recommendation: Add comprehensive documentation

🔍 Specific File Issues

src/unoplat_code_confluence_query_engine/config/settings.py

• Line 53: Hardcoded encryption key (CRITICAL)
• No validation for required environment variables

src/unoplat_code_confluence_query_engine/services/credentials_service.py

• Line 42: Potential credential exposure in debug logs
• Line 46-48: Environment variable check should be at startup, not runtime

src/unoplat_code_confluence_query_engine/api/v1/endpoints/codebase_agent_rules.py

• No authentication/authorization checks
• Large endpoint function (483 lines) - needs refactoring
• SSE endpoint without proper error handling

src/unoplat_code_confluence_query_engine/services/agent_execution_service.py

• Very large class (703 lines) - violates SRP
• Complex nested async operations without proper error isolation

📊 Technical Debt

1. Type Safety: Some Any types used where more specific types possible
2. Error Handling: Inconsistent error handling patterns across services
3. Logging: Mix of print statements and structured logging
4. Dependencies: Heavy dependency on external services without fallback strategies

🧪 Testing Recommendations

Must Have (before merge):

• Unit tests for core services (credentials, model factory, agents)
• Integration tests for database connections
• API endpoint tests with proper mocking
• Tool function tests with sample data

Should Have:

• Performance tests for large codebase operations
• Security tests for encryption/decryption flows
• Docker container tests
• End-to-end workflow tests

🚀 Performance Recommendations

1. Database Optimization

   • Add connection pooling configuration
   • Implement query result caching for repeated operations
   • Add database query logging and monitoring

2. Agent Execution

   • Add timeout configuration for agent operations
   • Implement result streaming with backpressure handling
   • Add resource usage monitoring

3. Tool Operations

   • Add pagination for large search results
   • Implement tool result caching
   • Add rate limiting for expensive operations

📋 Summary & Recommendations

This PR introduces a substantial new query engine component with good architectural foundations but significant gaps that must be addressed:

✅ Ready to Merge After:

1. CRITICAL: Fix hardcoded encryption key in settings.py
2. CRITICAL: Add comprehensive test suite
3. CRITICAL: Add application entry point and initialization
4. HIGH: Implement basic authentication/authorization
5. HIGH: Add proper error handling throughout

🔄 Future Improvements:

• Break down large service classes
• Add comprehensive API documentation
• Implement caching and performance optimizations
• Add monitoring and observability features

💡 Overall Assessment:

Strong architectural foundation with professional development practices, but critical security and testing gaps prevent immediate production readiness. The code demonstrates good Python/FastAPI patterns and the agent-based architecture shows promise for the codebase analysis use case.

Recommendation: Request changes to address critical security and testing issues before merging.

---

[qodo-code-review]

PR Reviewer Guide 🔍

Here are some key observations to aid the review process:

⏱️ Estimated effort to review: 5 🔵🔵🔵🔵🔵

🧪 No relevant tests

🔒 Security concerns External command execution: The ripgrep-based search tool spawns a subprocess with user-controlled patterns and globs. While it uses exec (not shell=True) and resolves paths against the codebase root, there is still risk of heavy resource usage or denial-of-service via complex regex, wide globs, or large contexts. Consider adding input length limits, regex complexity guards, tighter timeouts, and result caps per file. Also ensure ripgrep binary presence is controlled and version-pinned.

⚡ Recommended focus areas for review Serialization Risk SSE events embed Python objects into JSON via json.dumps; ensure all progress_event["data"] values are JSON-serializable after stripping BaseModels, and guard against non-serializable types leaking through other event paths. request, agent_execution_service, ruleset_metadata, directory_request, on_result=directory_result_handler ): sse_event = { "event": progress_event["event"], "data": json.dumps(progress_event["data"]), "id": str(event_id), } last_event_time = log_sse_event(connection_id, event_count, progress_event["event"], connection_start, last_event_time) event_count += 1 yield sse_event event_id += 1 # Stream major frameworks analysis events using framework_explorer_agent framework_request = AgentExecutionRequest( agent_name="framework_explorer", agent=request.app.state.agents['framework_explorer_agent'], fastapi_request=request, event_namespace="framework_explorer", postprocess_enabled=True, ) # Merge baseline (KG) ∪ novel (agent) per codebase framework_result_handler = partial( _update_framework_result_with_baseline, aggregators, request.app.state.neo4j_manager, ) async for progress_event in _stream_agent( request, agent_execution_service, ruleset_metadata, framework_request, on_result=framework_result_handler ): sse_event = { "event": progress_event["event"], "data": json.dumps(progress_event["data"]), "id": str(event_id), } last_event_time = log_sse_event(connection_id, event_count, progress_event["event"], connection_start, last_event_time) event_count += 1 yield sse_event event_id += 1 Command Injection/Path Safety Ripgrep invocation passes user-provided pattern and globs directly; validate/sanitize inputs and consider quoting or limiting args to prevent unexpected shell-like behaviors, and ensure path resolution cannot escape repo root. async def search_across_codebase( ctx: RunContext[AgentDependencies], pattern: str, mode: Literal["regex", "literal"] = "regex", glob: Optional[List[str]] = None, case: Literal["smart", "sensitive", "insensitive"] = "smart", context: int = 2, max_results: int = 50, timeout_s: int = 20, path: Optional[str] = None, ) -> SearchResults: """ Search for patterns across the entire codebase using ripgrep. This tool provides fast, powerful code searching capabilities with support for both regex and literal pattern matching, file filtering, and contextual previews. Args: pattern: The search pattern to find mode: Whether to treat pattern as "regex" or "literal" (default: "regex") glob: Optional list of glob patterns to filter files (e.g., ["*.py", "**/*.ts"]) case: Case strategy - "smart" (insensitive unless uppercase), "sensitive", or "insensitive" context: Number of lines to include before/after each match (default: 2) max_results: Global maximum number of matches to return (default: 50) timeout_s: Timeout in seconds for the search operation (default: 20) path: Optional search path relative to the codebase root. If provided, search starts in this subdirectory or file; otherwise searches the entire codebase. Returns: SearchResults containing matches with contextual information Examples: # Search for function definitions results = await search_across_codebase(pattern=r"^def \\w+", mode="regex", glob=["*.py"], context=3 ) # Find TODO comments (case insensitive) results = await search_across_codebase(pattern="TODO", mode="literal", case="insensitive", max_results=20 ) # Search for specific imports results = await search_across_codebase(pattern="from unoplat_code_confluence", mode="literal", glob=["**/*.py"]) """ # Validate inputs if not pattern.strip(): raise ModelRetry("Search pattern cannot be empty") # Get codebase path from dependencies codebase_path = ctx.deps.codebase_metadata.codebase_path if not codebase_path or not os.path.exists(codebase_path): raise ModelRetry(f"Invalid codebase path: {codebase_path}") if not os.path.isdir(codebase_path): Event Flow Robustness The producer completion and result/error events may race; confirm that every codepath enqueues a completion signal and that consumers handle out-of-order events without deadlocks or missed terminal states. async def _stream_events_for_codebase( self, repository_qualified_name: str, codebase: CodebaseMetadata, request: AgentExecutionRequest, event_queue: asyncio.Queue[Dict[str, Any]], ) -> Union[BaseModel, List[BaseModel], str]: """Run streaming agent for a single codebase and push events to queue.""" event_namespace = request.event_namespace or request.agent_name logger.debug("Starting agent {} for codebase {}", request.agent_name, codebase.codebase_name) # Create agent dependencies using app.state agent_deps = self._create_agent_dependencies( repository_qualified_name, codebase, request ) # Generate user message with codebase-specific context # Extract codebase-specific data from the context dictionary extra_context = {} if request.extra_prompt_context: for key, value in request.extra_prompt_context.items(): if isinstance(value, dict) and codebase.codebase_name in value: # Codebase-specific context: {"project_structure": {"frontend": BaseModel, "backend": BaseModel}} extra_context[key] = value[codebase.codebase_name] else: # Global context: {"some_key": BaseModel} extra_context[key] = value user_message = self.prompt_provider.get_user_message( request.agent_name, repository_qualified_name, codebase.codebase_name, codebase.codebase_path, codebase.codebase_programming_language, extra_prompt_context=extra_context, ) nodes: List[Any] = [] final_output: Union[BaseModel, List[BaseModel], str, None] = None try: try: async with request.agent.iter(user_message, deps=agent_deps) as agent_run: async for node in agent_run: nodes.append(node) await self._process_agent_node( node, agent_run, request, codebase, event_namespace, event_queue ) final_output = self._extract_agent_result(agent_run.result) logger.info("Agent {} completed for codebase {} with output: {}", request.agent_name, codebase.codebase_name, final_output) # Optional post-processing before emitting result processed_output = final_output or "" if request.postprocess_enabled: await event_queue.put({ "event": f"{codebase.codebase_name}:{event_namespace}:postprocess.start", "data": { "message": "Starting post-processing", "timestamp": datetime.now().isoformat(), }, }) try: processed_output = await self.post_processing.run( agent_name=request.agent_name, agent_output=processed_output, repository=repository_qualified_name, codebase=codebase, deps=agent_deps, options=request.postprocess_options, ) await event_queue.put({ "event": f"{codebase.codebase_name}:{event_namespace}:postprocess.result", "data": { "message": "Post-processing complete", "timestamp": datetime.now().isoformat(), }, }) except Exception as e: logger.warning("Post-processing failed for {}: {}", request.agent_name, e) await event_queue.put({ "event": f"{codebase.codebase_name}:{event_namespace}:postprocess.error", "data": { "message": f"Post-processing failed: {str(e)}", "timestamp": datetime.now().isoformat(), }, }) # Send result event with simple message (BaseModel passed via callback) if processed_output: await event_queue.put({ "event": f"{codebase.codebase_name}:{event_namespace}:result", "data": { "message": "Analysis complete", "result": processed_output, # BaseModel will be handled by callback "timestamp": datetime.now().isoformat(), }, }) except UnexpectedModelBehavior as e: final_output = await self._handle_agent_error( e, request, codebase, repository_qualified_name, event_namespace, event_queue ) # Send empty result event for consistency await event_queue.put({ "event": f"{codebase.codebase_name}:{event_namespace}:result", "data": { "message": "Analysis failed", "result": "", "timestamp": datetime.now().isoformat(), }, }) except Exception as e: # Handle any other exception logger.error("Unexpected error in agent {} for codebase {}: {}", request.agent_name, codebase.codebase_name, e) await event_queue.put({ "event": f"{codebase.codebase_name}:{event_namespace}:error", "data": { "message": f"Agent execution failed: {str(e)}", "timestamp": datetime.now().isoformat(), }, }) final_output = "" # Send empty result event for consistency await event_queue.put({ "event": f"{codebase.codebase_name}:{event_namespace}:result", "data": { "message": "Analysis failed", "result": "", "timestamp": datetime.now().isoformat(), }, }) # Save execution nodes await save_nodes_to_json( self.logs_dir, filename_prefix=( f"agent_run_{request.agent_name}_" f"{repository_qualified_name}_{codebase.codebase_name.replace('/', '_')}" ), nodes=nodes, ) # Send completion event await self._send_completion_event(codebase, event_namespace, request.agent_name, event_queue) finally: # ALWAYS signal this producer is done, even if an exception occurred await event_queue.put({"__done__": True}) logger.debug("Producer done signal sent for {} - {}", request.agent_name, codebase.codebase_name)

[qodo-code-review]

PR Code Suggestions ✨

Explore these optional code suggestions:

Category	Suggestion	Impact
High-level	External tool runtime assumptions Several tools depend on system binaries (ripgrep, eza, bun/bunx) and local MCP servers; if any are missing or blocked (e.g., container without network), core agent flows and SSE streams will silently degrade or fail. Add a centralized capability check at startup with per-request fallbacks (or graceful feature flags) and expose this status via an endpoint/event so clients and operators know which features are active. Examples: unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/tools/get_directory_tree.py [23-109] unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/tools/search_across_codebase.py [218-368] Solution Walkthrough: Before: # In a tool like get_directory_tree.py async def get_directory_tree(ctx, path): try: # This raises FileNotFoundError if 'eza' is not installed. process = await asyncio.create_subprocess_exec("eza", ...) ... except FileNotFoundError: # The failure is handled locally, but the application # as a whole doesn't know it's running in a degraded state. raise ModelRetry("eza not installed...") # In main.py (startup) # No checks are performed to ensure tool binaries are available. # The application starts assuming the environment is correctly configured. After: # In a new capabilities_service.py class SystemCapabilities(BaseModel): ripgrep_available: bool eza_available: bool bun_available: bool mcp_servers_status: Dict[str, bool] async def check_system_capabilities(): # Use shutil.which or subprocess to check for binaries # Attempt to connect/ping MCP servers ... return SystemCapabilities(...) # In main.py (startup) @asynccontextmanager async def lifespan(app: FastAPI): app.state.capabilities = await check_system_capabilities() logger.info(f"System capabilities: {app.state.capabilities}") yield # In a new /health or /status endpoint @router.get("/status") async def get_status(request: Request) -> SystemCapabilities: return request.app.state.capabilities Suggestion importance[1-10]: 10 __ Why: This suggestion identifies a critical architectural flaw regarding unverified runtime dependencies (ripgrep, eza, bunx), which is proven by the Dockerfile missing the eza installation, guaranteeing failure for the directory_agent in the containerized environment.	High
General	Explicitly close MCP servers Relying on garbage collection to close transports risks leaked subprocesses or HTTP connections. Explicitly close/terminate each server (e.g., await server.close() or server.aclose() as supported) before clearing the registry to ensure resources are released. unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/services/mcp/mcp_server_manager.py [185-195] logger.info("Stopping {} MCP servers", len(self.servers)) -for name in list(self.servers.keys()): +for name, server in list(self.servers.items()): try: - # MCP servers are managed via context managers in PydanticAI - # Cleanup happens automatically when the server instances are destroyed + # Explicitly close server transports if supported + close = getattr(server, "close", None) + aclose = getattr(server, "aclose", None) + if callable(aclose): + await aclose() + elif callable(close): + result = close() + if hasattr(result, "__await__"): + await result # handle async close methods logger.info("Stopped MCP server '{}'", name) except Exception as e: logger.error("Error stopping MCP server '{}': {}", name, e) self.servers.clear() [To ensure code accuracy, apply this suggestion manually] Suggestion importance[1-10]: 9 __ Why: The suggestion correctly identifies a potential resource leak by relying on garbage collection for server cleanup and provides a robust solution to explicitly close server resources, preventing leaked subprocesses or network connections.	High
Possible issue	Prevent premature connection closing Avoid closing a shared connection manager inside a tool; this can break subsequent tool calls in the same agent run. Remove the unconditional close here and let the lifecycle be managed at a higher level. unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/tools/get_data_model_files.py [53-54] finally: - await ctx.deps.neo4j_conn_manager.close() + # Connection lifecycle managed by the application; do not close here + pass [To ensure code accuracy, apply this suggestion manually] Suggestion importance[1-10]: 8 __ Why: The suggestion correctly identifies a potential bug where a shared connection manager is closed prematurely, which could cause subsequent operations within the same agent run to fail.	Medium
	Stop closing shared Neo4j manager Do not close the shared Neo4j connection manager within a tool function, as it may be reused by the agent for subsequent steps. Remove the close call and delegate connection lifecycle to the app. unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/tools/get_core_files.py [50-51] finally: - await ctx.deps.neo4j_conn_manager.close() + # Connection lifecycle managed by the application; do not close here + pass [To ensure code accuracy, apply this suggestion manually] Suggestion importance[1-10]: 8 __ Why: The suggestion correctly identifies a potential bug where a shared connection manager is closed prematurely, which could cause subsequent operations within the same agent run to fail.	Medium
	Add robust result event sanitization Handle rg exit code 2 (bad regex) distinctly to guide the model to retry with literal mode or corrected pattern. Parsing stderr for “regex parse error” and surfacing a clear hint prevents repeated tool failures. unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/tools/search_across_codebase.py [331-360] if process.returncode == 0: - # Success - parse output stdout_str = stdout_bytes.decode("utf-8", errors="replace") output_lines = stdout_str.strip().split("\n") if stdout_str.strip() else [] - - matches, total_found, truncated = _parse_ripgrep_json_output( - output_lines, max_results, context - ) - - return SearchResults( - pattern=pattern, matches=matches, total=total_found, truncated=truncated - ) - + matches, total_found, truncated = _parse_ripgrep_json_output(output_lines, max_results, context) + return SearchResults(pattern=pattern, matches=matches, total=total_found, truncated=truncated) elif process.returncode == 1: - # No matches found - this is normal return SearchResults(pattern=pattern, matches=[], total=0, truncated=False) else: - # Error occurred stderr_str = stderr_bytes.decode("utf-8", errors="replace") - if ( - "No such file or directory" in stderr_str - or "command not found" in stderr_str - ): + lower_err = stderr_str.lower() + if "command not found" in lower_err or "no such file or directory" in lower_err: raise ModelRetry( - "ripgrep not installed. Install via: brew install ripgrep (macOS) " - "or apt install ripgrep (Ubuntu)" + "ripgrep not installed. Install via: brew install ripgrep (macOS) or apt install ripgrep (Ubuntu)" ) - else: - # Include stderr in error for debugging - raise ModelRetry(f"Search failed: {stderr_str}") + if "regex parse error" in lower_err or "error parsing regex" in lower_err: + raise ModelRetry("Invalid regex pattern. Try correcting the regex or use mode='literal'.") + raise ModelRetry(f"Search failed: {stderr_str}") [To ensure code accuracy, apply this suggestion manually] Suggestion importance[1-10]: 7 __ Why: The suggestion improves error handling by specifically detecting invalid regex patterns from ripgrep's output, providing more actionable feedback to the agent via ModelRetry.	Medium
Security	Disable backtrace in production Avoid enabling verbose backtraces in production logs as it can leak sensitive data and impact performance. Make this conditional on a debug flag or disable by default. unoplat-code-confluence-query-engine/src/unoplat_code_confluence_query_engine/config/logging_config.py [24-32] logger.add( sys.stdout, format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> | <level>{level: <8}</level> | <cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> | <level>{message}</level>", level=settings.log_level, colorize=True, - enqueue=True, # Async-safe for FastAPI - diagnose=False, # Safe for production - backtrace=True + enqueue=True, + diagnose=False, + backtrace=bool(getattr(settings, "debug", False)), ) Apply / Chat Suggestion importance[1-10]: 7 __ Why: The suggestion correctly points out that enabling backtrace=True unconditionally can be a security risk in production by leaking sensitive data, and proposes a good practice of making it conditional.	Medium
More

[JayGhiya]

we do not have tests yet so it fails as it cannot find the test directory which is fine as of now