HAL9000/temp
0
0
Fork 0
forked from HAL9000/cleveragents-core
Code Pull Requests Activity

Fix: Added streaming to context commands

Browse Source
This commit is contained in:
Jeffrey Phillips Freeman
2025-11-26 23:24:46 -05:00
parent adf04a7b60
commit 641f7c09d9
24 changed files with 3175 additions and 192 deletions
Download Patch File Download Diff File Expand all files Collapse all files
features/auto_debug_agent_coverage.feature
+246 -7
View File
@@ -6,6 +6,7 @@ Feature: Auto Debug Agent Coverage
Background:
Given the auto debug agent module is importable
And I have a mock LLM provider configured for auto debug
And logging is enabled at INFO level for auto debug
Scenario: AutoDebugAgent can be instantiated with default parameters
When I create an AutoDebugAgent with default parameters
@@ -339,10 +340,248 @@ Feature: Auto Debug Agent Coverage
When I create an AutoDebugAgent with default parameters
Then the agent temperature should be 0.3 for auto debug
Scenario: AutoDebugAgent supports provider kwargs
When I create an AutoDebugAgent with provider_kwargs:
| kwarg | value |
| max_tokens | 2000 |
| top_p | 0.9 |
Then the agent should be initialized successfully for auto debug
And the agent should have provider_kwargs stored
# Coverage for lines 133: analyze_error with real LLM response (not "Mock LLM response")
Scenario: Analyze error with non-mock LLM response returns actual content
Given I have an AutoDebugAgent instance
And the LLM returns a non-mock analysis response "This is a type mismatch error"
And I have a state with error details:
"""
{
"error_message": "TypeError: expected int but got str",
"code_context": "x = '5' + 5"
}
"""
When I execute the analyze_error step
Then the error_analysis content should be "This is a type mismatch error"
# Coverage for lines 134-136: analyze_error exception handling
Scenario: Analyze error handles LLM exception gracefully
Given I have an AutoDebugAgent instance
And the LLM raises an exception during analysis
And I have a state with error details:
"""
{
"error_message": "SyntaxError: invalid syntax",
"code_context": "def foo(:"
}
"""
When I execute the analyze_error step
Then the error_analysis content should be "Error analysis completed"
And a warning should be logged containing "LLM analysis failed"
# Coverage for lines 223-225: generate_fix with valid JSON response
Scenario: Generate fix parses valid JSON from non-mock LLM response
Given I have an AutoDebugAgent instance
And the LLM returns a valid JSON fix response
And I have a state with error analysis completed
When I execute the generate_fix step
Then the current_fix description should be "Add variable definition"
And the current_fix code should be "x = 0; return x + 1"
And the current_fix files_to_modify should contain "test.py"
# Coverage for lines 226-232: generate_fix with invalid JSON response (fallback)
Scenario: Generate fix falls back when LLM returns non-JSON response
Given I have an AutoDebugAgent instance
And the LLM returns a non-JSON fix response "Just add x = 0 before the return"
And I have a state with error analysis completed
When I execute the generate_fix step
Then the current_fix description should contain "Fix attempt"
And the current_fix code should be "Just add x = 0 before the return"
And the current_fix files_to_modify should be empty
# Coverage for lines 233-239: generate_fix exception handling
Scenario: Generate fix handles LLM exception gracefully
Given I have an AutoDebugAgent instance
And the LLM raises an exception during fix generation
And I have a state with error analysis completed
When I execute the generate_fix step
Then the current_fix should have a description field
And the current_fix should have a code field
And a warning should be logged containing "LLM fix generation failed"
# Coverage for lines 297-299: validate_fix with valid JSON response (is_valid true)
Scenario: Validate fix parses valid JSON with is_valid true from non-mock response
Given I have an AutoDebugAgent instance
And the LLM returns a valid JSON validation response with is_valid true
And I have a state with a current fix:
"""
{
"description": "Add variable x before use",
"code": "x = 0\nreturn x + 1"
}
"""
When I execute the validate_fix step
Then the fix_validated should be true
# Coverage for lines 297-299: validate_fix with valid JSON response (is_valid false)
Scenario: Validate fix parses valid JSON with is_valid false from non-mock response
Given I have an AutoDebugAgent instance
And the LLM returns a valid JSON validation response with is_valid false
And I have a state with a current fix:
"""
{
"description": "Incorrect fix attempt",
"code": "return x"
}
"""
When I execute the validate_fix step
Then the fix_validated should be false
# Coverage for lines 300-305: validate_fix JSON decode error with positive fallback
Scenario: Validate fix falls back to text parsing with positive indicators
Given I have an AutoDebugAgent instance
And the LLM returns a non-JSON validation response "The fix is valid and correct"
And I have a state with a current fix:
"""
{
"description": "Add x = 0",
"code": "x = 0; return x + 1"
}
"""
When I execute the validate_fix step
Then the fix_validated should be true
# Coverage for lines 300-305: validate_fix JSON decode error without positive indicators
Scenario: Validate fix falls back to text parsing without positive indicators
Given I have an AutoDebugAgent instance
And the LLM returns a non-JSON validation response "The code still has issues"
And I have a state with a current fix:
"""
{
"description": "Broken fix",
"code": "return undefined_var"
}
"""
When I execute the validate_fix step
Then the fix_validated should be false
# Coverage for lines 306-309: validate_fix exception handling
Scenario: Validate fix handles LLM exception gracefully
Given I have an AutoDebugAgent instance
And the LLM raises an exception during validation
And I have a state with a current fix:
"""
{
"description": "Some fix",
"code": "some code"
}
"""
When I execute the validate_fix step
Then the fix_validated should be true
And a warning should be logged containing "LLM validation failed"
# Coverage for lines 314-317: tracking failed fix attempts
Scenario: Validate fix tracks failed attempts when validation is false
Given I have an AutoDebugAgent instance
And the LLM returns a valid JSON validation response with is_valid false
And I have a state with current fix and empty attempted_fixes:
"""
{
"description": "Failed fix",
"code": "bad code"
}
"""
When I execute the validate_fix step
Then the fix_validated should be false
And the attempted_fixes should have 1 entry
And the attempted_fixes should contain the current fix
# Coverage for lines 314-317: multiple failed attempts are tracked
Scenario: Validate fix appends to existing attempted_fixes when validation fails
Given I have an AutoDebugAgent instance
And the LLM returns a valid JSON validation response with is_valid false
And I have a state with current fix and one existing attempted fix
When I execute the validate_fix step
Then the fix_validated should be false
And the attempted_fixes should have 2 entries
# Coverage: validate_fix with "resolves" in response
Scenario: Validate fix detects "resolves" keyword in fallback parsing
Given I have an AutoDebugAgent instance
And the LLM returns a non-JSON validation response "This change resolves the issue"
And I have a state with a current fix:
"""
{
"description": "Fix that resolves issue",
"code": "fixed code"
}
"""
When I execute the validate_fix step
Then the fix_validated should be true
# Coverage: validate_fix with "fixes" in response
Scenario: Validate fix detects "fixes" keyword in fallback parsing
Given I have an AutoDebugAgent instance
And the LLM returns a non-JSON validation response "This fixes the problem"
And I have a state with a current fix:
"""
{
"description": "Fix that fixes problem",
"code": "fixed code"
}
"""
When I execute the validate_fix step
Then the fix_validated should be true
# Coverage: generate_fix includes previous attempts in prompt
Scenario: Generate fix includes previous attempt descriptions in prompt
Given I have an AutoDebugAgent instance
And the LLM returns a valid JSON fix response
And I have a state with two previous fix attempts
When I execute the generate_fix step
Then the current_fix should have a description field
And the attempt number should be 3
# Edge case: Empty content from LLM in generate_fix
Scenario: Generate fix handles empty non-mock response
Given I have an AutoDebugAgent instance
And the LLM returns an empty non-mock response for fix generation
And I have a state with error analysis completed
When I execute the generate_fix step
Then the current_fix should have a description field
And the current_fix code should be empty string
# Edge case: Validation response with is_valid as string
Scenario: Validate fix handles is_valid as string "true" in JSON
Given I have an AutoDebugAgent instance
And the LLM returns a JSON validation response with is_valid as string "true"
And I have a state with a current fix:
"""
{
"description": "Fix",
"code": "code"
}
"""
When I execute the validate_fix step
Then the fix_validated should be true
# Edge case: Validation response with is_valid missing
Scenario: Validate fix handles missing is_valid field in JSON response
Given I have an AutoDebugAgent instance
And the LLM returns a JSON validation response without is_valid field
And I have a state with a current fix:
"""
{
"description": "Fix",
"code": "code"
}
"""
When I execute the validate_fix step
Then the fix_validated should be false
# Edge case: Case insensitive keyword detection in validation fallback
Scenario: Validate fix keyword detection is case insensitive
Given I have an AutoDebugAgent instance
And the LLM returns a non-JSON validation response "VALID solution found"
And I have a state with a current fix:
"""
{
"description": "Fix",
"code": "code"
}
"""
When I execute the validate_fix step
Then the fix_validated should be true
features/auto_debug_cli_coverage.feature
+190
View File
@@ -0,0 +1,190 @@
Feature: Auto Debug CLI Command Coverage
As a developer
I want reliable coverage for the auto_debug CLI entry points
So that build automation paths remain stable
Background:
Given I have a clean auto_debug test environment
# auto_debug_command() coverage
Scenario: Auto debug command succeeds on first attempt
Given I have an initialized project for auto_debug
And the plan service auto_debug_build will succeed
When I call the auto_debug_command programmatic interface
Then the auto_debug_command should return success True
And the attempts made should be 1
Scenario: Auto debug command fails after max attempts
Given I have an initialized project for auto_debug
And the plan service auto_debug_build will fail
When I call the auto_debug_command with max_attempts 3
Then the auto_debug_command should return success False
And the attempts made should be 3
Scenario: Auto debug command with no project raises error
Given I have no project initialized
When I call the auto_debug_command programmatic interface expecting error
Then a CleverAgentsError should be raised with message "No project found"
# _get_current_project() coverage
Scenario: Get current project when project exists
Given I have an initialized project for auto_debug
When I call _get_current_project
Then the project should be returned successfully
Scenario: Get current project when no project exists
Given I have no project initialized
When I call _get_current_project expecting abort
Then typer.Abort should be raised
# run command happy path
Scenario: Auto debug run command with successful build
Given I have an initialized project for auto_debug
And the build will succeed with 5 changes
When I run the auto_debug run command
Then the command should exit with code 0
And the output should contain "Generated 5 change(s)"
And the output should contain "AutoDebug Complete"
Scenario: Auto debug run command with build failure then success
Given I have an initialized project for auto_debug
And the build will fail then succeed after fix
When I run the auto_debug run command with max_attempts 3
Then the command should exit with code 0
And the output should contain "Build failed"
And the output should contain "Fix generated"
And the output should contain "Build succeeded"
Scenario: Auto debug run command exceeds max attempts
Given I have an initialized project for auto_debug
And the build will always fail with error "Final failure"
When I run the auto_debug run command with max_attempts 2
Then the command should exit with code 1
And the output should contain "Build failed after"
And the output should contain "Final error"
And the output should contain "Final failure"
Scenario: Auto debug run command fails with empty error message
Given I have an initialized project for auto_debug
And the build will always fail with empty error message
When I run the auto_debug run command with max_attempts 2
Then the command should exit with code 1
And the output should contain "Build failed after"
And the output should not contain "Final error"
Scenario: Auto debug run command handles PlanError with details
Given I have an initialized project for auto_debug
And the plan service will raise a PlanError with details
When I run the auto_debug run command
Then the command should be aborted
And the output should contain "Plan Error"
And the output should contain "phase:"
Scenario: Auto debug run command handles PlanError without details
Given I have an initialized project for auto_debug
And the plan service will raise a PlanError without details
When I run the auto_debug run command
Then the command should be aborted
And the output should contain "Plan Error"
And the output should not contain "phase:"
Scenario: Auto debug run command handles CleverAgentsError
Given I have an initialized project for auto_debug
And the plan service will raise a CleverAgentsError
When I run the auto_debug run command
Then the command should be aborted
And the output should contain "Error:"
Scenario: Auto debug run command handles fix generation error
Given I have an initialized project for auto_debug
And the build will fail with fix generation error
When I run the auto_debug run command with max_attempts 2
Then the command should exit with code 0
And the output should contain "Could not generate fix"
# Edge case: Long error message truncation
Scenario: Auto debug run command truncates long error messages
Given I have an initialized project for auto_debug
And the build will always fail with error "This is a very long error message that exceeds eighty characters and should be truncated in the display output"
When I run the auto_debug run command with max_attempts 1
Then the command should exit with code 1
And the output should contain "..."
And the output should contain "Build failed after"
# Edge case: Minimum max_attempts boundary
Scenario: Auto debug run command with minimum max_attempts of 1
Given I have an initialized project for auto_debug
And the build will always fail with error "Single attempt failure"
When I run the auto_debug run command with max_attempts 1
Then the command should exit with code 1
And the output should contain "Attempt 1/1"
And the output should contain "Build failed after 1 attempt"
# Edge case: Multiple sequential failures before success
Scenario: Auto debug run command succeeds after multiple failures
Given I have an initialized project for auto_debug
And the build will fail 2 times then succeed with 3 changes
When I run the auto_debug run command with max_attempts 5
Then the command should exit with code 0
And the output should contain "Build failed"
And the output should contain "Build successful"
And the output should contain "Generated 3 change(s)"
# Edge case: Short error message (no truncation)
Scenario: Auto debug run command shows short error without truncation
Given I have an initialized project for auto_debug
And the build will always fail with error "Short error"
When I run the auto_debug run command with max_attempts 1
Then the command should exit with code 1
And the output should contain "Short error"
And the output should not contain "Short error..."
# Edge case: PlanError re-raised from within build loop
Scenario: Auto debug run command re-raises PlanError from build
Given I have an initialized project for auto_debug
And the build will raise PlanError after one attempt
When I run the auto_debug run command with max_attempts 3
Then the command should be aborted
And the output should contain "Plan Error"
# Edge case: CleverAgentsError re-raised from within build loop
Scenario: Auto debug run command re-raises CleverAgentsError from build
Given I have an initialized project for auto_debug
And the build will raise CleverAgentsError after one attempt
When I run the auto_debug run command with max_attempts 3
Then the command should be aborted
And the output should contain "Error:"
# Edge case: auto_debug_command with custom max_attempts
Scenario: Auto debug command with custom max_attempts of 5
Given I have an initialized project for auto_debug
And the plan service auto_debug_build will fail
When I call the auto_debug_command with max_attempts 5
Then the auto_debug_command should return success False
And the attempts made should be 5
# Edge case: auto_debug_command with max_attempts of 1
Scenario: Auto debug command with minimum max_attempts of 1
Given I have an initialized project for auto_debug
And the plan service auto_debug_build will fail
When I call the auto_debug_command with max_attempts 1
Then the auto_debug_command should return success False
And the attempts made should be 1
# Edge case: Successful build on exact last attempt
Scenario: Auto debug run command succeeds on the last attempt
Given I have an initialized project for auto_debug
And the build will fail 2 times then succeed with 1 changes
When I run the auto_debug run command with max_attempts 3
Then the command should exit with code 0
And the output should contain "Build succeeded after 3 attempt"
# Edge case: Multiple PlanError details
Scenario: Auto debug run command handles PlanError with multiple details
Given I have an initialized project for auto_debug
And the plan service will raise a PlanError with multiple details
When I run the auto_debug run command
Then the command should be aborted
And the output should contain "Plan Error"
And the output should contain "phase:"
And the output should contain "step:"
features/context_service_analysis.feature
+164
View File
@@ -0,0 +1,164 @@
Feature: Context Service Analysis Integration
As a developer using CleverAgents
I want to analyze context files using LangGraph workflows
So that I can understand the codebase better before making changes
Background:
Given I have initialized a CleverAgents project
And I have a context service with LangGraph integration
Scenario: Analyze context when no files are loaded
Given the current plan has no context files
When I analyze the context
Then the analysis result should have empty documents
And the analysis summary should indicate no files to analyze
And there should be no error in the analysis
Scenario: Analyze context with a single Python file
Given I have a Python file "main.py" with content:
"""
import os
import sys
from pathlib import Path
def main():
print("Hello, World!")
"""
And I have added "main.py" to the LangGraph context
When I analyze the context
Then the analysis result should have 1 document
And the dependencies for "main.py" should include "os"
And the relevance score for "main.py" should be between 0.0 and 1.0
And the summary should not be empty
Scenario: Analyze context with multiple files
Given I have a Python file "utils.py" with content:
"""
def helper():
return 42
"""
And I have a Python file "app.py" with content:
"""
from utils import helper
print(helper())
"""
And I have added "utils.py" to the LangGraph context
And I have added "app.py" to the LangGraph context
When I analyze the context
Then the analysis result should have 2 documents
And the dependencies should include entries for both files
And the relevance scores should have entries for both files
Scenario: Get context summary
Given I have a Python file "sample.py" with content:
"""
# Sample module
class Sample:
pass
"""
And I have added "sample.py" to the LangGraph context
When I get the context summary
Then the summary should be a non-empty string
And the summary should not indicate an error
Scenario: Get context dependencies
Given I have a Python file "deps.py" with content:
"""
import json
import requests
from typing import Dict
"""
And I have added "deps.py" to the LangGraph context
When I get the context dependencies
Then the dependencies dict should have an entry for "deps.py"
And the dependencies should be a non-empty list
Scenario: Get relevant files with threshold
Given I have a Python file "important.py" with content:
"""
# Core business logic
def critical_function():
return "important"
"""
And I have a Python file "trivial.py" with content:
"""
# Just comments
pass
"""
And I have added "important.py" to the LangGraph context
And I have added "trivial.py" to the LangGraph context
When I get relevant files with threshold 0.0
Then the result should include both files with scores
And all LangGraph scores should be between 0.0 and 1.0
Scenario: Stream context analysis
Given I have a Python file "streaming.py" with content:
"""
print("test streaming")
"""
And I have added "streaming.py" to the LangGraph context
When I stream the context analysis
Then I should receive multiple events
And the events should include node execution results
Scenario: Stream context analysis when no files are loaded
Given the current plan has no context files
When I stream the context analysis
Then the streaming output should report no files to analyze
Scenario: Analyze context asynchronously
Given I have a Python file "async_test.py" with content:
"""
async def async_func():
return "async"
"""
And I have added "async_test.py" to the LangGraph context
When I analyze the context asynchronously
Then the async result should have documents
And the async result should have a summary
And there should be no error in the async result
Scenario: Analyze context asynchronously when no files are loaded
Given the current plan has no context files
When I analyze the context asynchronously
Then the async analysis result should have empty documents
And the async analysis summary should indicate no files to analyze
Scenario: Stream context analysis asynchronously with loaded files
Given I have a Python file "async_stream.py" with content:
"""
print("async stream")
"""
And I have added "async_stream.py" to the LangGraph context
When I stream the context analysis asynchronously
Then the async streaming events should include node execution results
Scenario: Stream context analysis asynchronously with no files
Given the current plan has no context files
When I stream the context analysis asynchronously
Then the async streaming output should report no files to analyze
Scenario: Handle non-existent files gracefully
Given I have added a non-existent file path to the analysis
When I analyze the context with the non-existent path
Then the analysis should complete without raising an exception
And the error field should contain file not found information
Scenario: Get context agent instance
When I request a context analysis agent
Then I should receive a ContextAnalysisAgent instance
And the agent should have the standard workflow nodes
Scenario: Prepare LangSmith metadata when tracing is enabled
Given LangSmith tracing is enabled for context analysis metadata
And I have a Python file "langsmith.py" with content:
"""
print("langsmith tracing")
"""
And I have added "langsmith.py" to the LangGraph context
When I prepare the LangGraph analysis config for run "LangSmithRun" in "stream" mode
Then the LangSmith config should include the current project metadata
And the LangSmith config should record 1 context file path
features/environment.py
+20 -1
View File
@@ -5,6 +5,18 @@ import os
import shutil
from pathlib import Path
LANGSMITH_ENV_VARS = [
"CLEVERAGENTS_LANGSMITH_ENABLED",
"CLEVERAGENTS_LANGSMITH_PROJECT",
"CLEVERAGENTS_LANGSMITH_ENDPOINT",
"CLEVERAGENTS_LANGSMITH_API_KEY",
"CLEVERAGENTS_LANGSMITH_TRACING_V2",
"LANGCHAIN_TRACING_V2",
"LANGCHAIN_PROJECT",
"LANGCHAIN_ENDPOINT",
"LANGCHAIN_API_KEY",
]
def before_all(context):
"""Set up test environment before all tests."""
@@ -37,7 +49,11 @@ def before_scenario(context, scenario):
context._cleanup_handlers = []
# Clean up any lingering test environment variables from previous tests
for env_var in ["CLEVERAGENTS_MOCK_SHOULD_FAIL", "CLEVERAGENTS_MOCK_INVALID_CODE"]:
for env_var in [
"CLEVERAGENTS_MOCK_SHOULD_FAIL",
"CLEVERAGENTS_MOCK_INVALID_CODE",
*LANGSMITH_ENV_VARS,
]:
if env_var in os.environ:
del os.environ[env_var]
@@ -83,6 +99,9 @@ def after_scenario(context, scenario):
os.environ.pop(key, None)
context.env_vars_to_clean = []
for env_var in LANGSMITH_ENV_VARS:
os.environ.pop(env_var, None)
# Reset Settings singleton if it was used
try:
from cleveragents.config.settings import Settings
features/langsmith_config.feature
+26
View File
@@ -0,0 +1,26 @@
Feature: LangSmith configuration detection
As a developer configuring observability
I want to verify CleverAgents detects LangSmith settings
So that tracing can be enabled or disabled safely
Background:
Given LangSmith environment is clean
Scenario: LangSmith tracing disabled by default
When I load CleverAgents settings for LangSmith
Then LangSmith tracing should be disabled
Scenario: LangSmith tracing enabled via LangChain env vars
Given I set environment variable "LANGCHAIN_TRACING_V2" to "true"
And I set environment variable "LANGCHAIN_API_KEY" to "demo-key"
When I load CleverAgents settings for LangSmith
Then LangSmith tracing should be enabled
Scenario: Building LangSmith config metadata
Given I set environment variable "CLEVERAGENTS_LANGSMITH_ENABLED" to "true"
And I set environment variable "CLEVERAGENTS_LANGSMITH_PROJECT" to "cleveragents-core"
And I set environment variable "CLEVERAGENTS_LANGSMITH_API_KEY" to "demo-key"
When I load CleverAgents settings for LangSmith
And I build a LangSmith config with run name "demo-run"
Then the LangSmith config should include tag "context-analysis"
And the LangSmith config should include metadata key "langsmith_project"
features/plan_service.feature
+16 -1
View File
@@ -50,4 +50,19 @@ Feature: Plan Service
Given I have a plan service
And I have created a plan with "Initial instructions"
When I add "Additional instructions" to the plan
Then the plan should contain both instructions
Then the plan should contain both instructions
Scenario: Auto-debug build uses empty context when plan ID disappears
Given I have a temporary test directory for plan service
And I configured an auto-debug plan with a disappearing plan ID
When I run auto-debug build with missing plan context
Then the auto-debug build should fall back to an empty context
Scenario: Apply changes moves files to absolute destinations
Given I have a temporary test directory for plan service
And I have a Unit of Work instance for plan testing
And I have a PlanService instance
And I have a saved project with current plan
And the plan has a pending MOVE change to an absolute path
When I apply the plan changes
Then the absolute destination should exist and the source should be removed
features/plan_service_uncovered_lines.feature
+28
View File
@@ -51,3 +51,31 @@ Feature: Plan Service Uncovered Lines Coverage
Then the current plan prompt should be "New instructions"
And the current plan status should be PENDING
And the current plan updated_at should be refreshed
Scenario: LangSmith config includes project and plan metadata when enabled
Given LangSmith integration is enabled for plan service
When I prepare a LangSmith config for project "Docs" and plan "Improve docs"
Then the LangSmith builder should receive metadata for project "Docs" and plan "Improve docs"
And the prepared LangSmith config should include a generated thread id
Scenario: Build plan surfaces provider errors
Given I have a plan service with a failing AI provider
And I have a saved project with current plan
When I attempt to build the plan and the provider returns an error
Then a PlanError should be raised with message "mock provider failure"
Scenario: Auto debug build requires a current plan
Given I have a saved project with no current plan
When I run the auto debug build for the project
Then a PlanError should be raised with message "No current plan to build"
Scenario: Auto debug build reports failure after retries
Given I have a saved project with current plan
And auto debug retries should run with a failing build pipeline
When I run the auto debug build with max attempts 2
Then the auto debug result should indicate failure with error message "Simulated auto debug failure"
Scenario: Streaming generation requires an AI provider
Given I have a saved project
When I try to stream plan generation without an AI provider
Then a PlanError should be raised with message "No AI provider configured"
features/settings_configuration.feature
+60
View File
@@ -0,0 +1,60 @@
Feature: Settings runtime helpers
Verify environment alias, storage paths, database URLs, production flags,
provider configuration lookups, and LangSmith config guards behave predictably.
Scenario: Environment alias setter syncs env field
Given no environment variables are set
When I load the settings with defaults
And I set the environment alias to "staging"
Then the environment alias should be "staging"
And the environment should be "staging"
Scenario: Storage base path trims whitespace
Given the storage base path is "/tmp/storage"
When I compute the storage path for " logs "
Then the storage path should be "/tmp/storage/logs"
Scenario: Storage base path returns default when type missing
Given the storage base path is "/tmp/storage"
When I compute the default storage path
Then the storage path should be "/tmp/storage"
Scenario: SQLite URLs derive implicit test database
Given no environment variables are set
When I derive the test database URL from "sqlite:///primary.db"
Then the derived test database URL should be "sqlite:///primary_test.db"
Scenario: Non-SQLite URLs fall back to original string
When I derive the test database URL from "postgresql://localhost/app"
Then the derived test database URL should be "postgresql://localhost/app"
Scenario: Production mode helpers honour flags
When I evaluate production mode with env "production", debug "false", and reload "false"
Then is_production should be True
And is_production_mode should be True
Scenario: Unknown provider alias short-circuits lookups
Given no API keys are set
When I check if "unknown" provider is configured
Then has_provider_configured should be False for "unknown"
Scenario: Explicit provider values skip env overrides
Given the environment variable "OPENAI_API_KEY" is set to "env-key"
When I instantiate settings with explicit openai API key "constructor-key"
Then the openai API key should be "constructor-key"
Scenario: Provider env vars satisfy configuration checks
Given no API keys are set
And the openai API key is set to "env-key"
When I check if "openai" provider is configured
Then has_provider_configured should be True for "openai"
Scenario: Provider fallback reads env vars when attribute unset
Given no API keys are set
And the openai API key is set to "env-key"
When I clear the explicit "openai" key and recheck configuration
Then has_provider_configured should be True for "openai"
Scenario: LangSmith config builder returns None when disabled
When I build the LangSmith config
Then the LangSmith config should be absent
features/steps/architecture_steps.py
+14
View File
@@ -198,6 +198,12 @@ def step_verify_env_prefix(context, prefix):
"GOOGLE_",
"OPENROUTER_",
"HF_",
"HUGGINGFACEHUB_",
"HUGGING_FACE_HUB_",
"COHERE_",
"PERPLEXITY_",
"GROQ_",
"TOGETHER_",
]
# Common constants and system variables to exclude
@@ -244,6 +250,14 @@ def step_verify_provider_vars(context):
"GEMINI_",
"AZURE_",
"GOOGLE_",
"OPENROUTER_",
"HF_",
"HUGGINGFACEHUB_",
"HUGGING_FACE_HUB_",
"COHERE_",
"PERPLEXITY_",
"GROQ_",
"TOGETHER_",
]
provider_vars = [
var
features/steps/auto_debug_agent_coverage_steps.py
+241
View File
@@ -165,6 +165,41 @@ def step_error_analysis_mentions(context, text):
assert any(text.lower() in msg.get("content", "").lower() for msg in analysis_msgs)
@given('the LLM returns a non-mock analysis response "{response}"')
def step_llm_returns_non_mock_analysis_response(context, response):
"""Configure the mock LLM to return a specific analysis response."""
mock_response = Mock()
mock_response.content = response
context.mock_llm.invoke.side_effect = None
context.mock_llm.invoke.return_value = mock_response
@given("the LLM raises an exception during analysis")
def step_llm_raises_exception_analysis(context):
"""Configure the mock LLM to raise during analysis."""
context.mock_llm.invoke.side_effect = Exception("LLM analysis failed")
@then('the error_analysis content should be "{expected_content}"')
def step_error_analysis_content_equals(context, expected_content):
"""Assert the last error_analysis message matches expected content."""
messages = context.state.get("messages", [])
analysis_msg = next(
(msg for msg in messages if msg.get("type") == "error_analysis"), None
)
assert analysis_msg is not None, "No error_analysis message found"
assert analysis_msg.get("content") == expected_content
@then('a warning should be logged containing "{text}"')
def step_warning_logged_contains(context, text):
"""Verify a warning with the provided text was logged."""
assert hasattr(context, "log_capture"), "Log capture not initialized"
assert any(text in log for log in context.log_capture), (
f"Expected warning containing '{text}'"
)
@given("I have a state with error analysis completed")
def step_have_state_with_error_analysis(context):
"""Create a state with completed error analysis."""
@@ -209,6 +244,89 @@ def step_current_fix_has_code(context):
assert "code" in current_fix
@given("the LLM returns a valid JSON fix response")
def step_llm_returns_valid_json_fix(context):
"""Configure the LLM to return valid JSON during fix generation."""
mock_response = Mock()
mock_response.content = json.dumps(
{
"description": "Add variable definition",
"code": "x = 0; return x + 1",
"files_to_modify": ["test.py"],
}
)
context.mock_llm.invoke.side_effect = None
context.mock_llm.invoke.return_value = mock_response
@given('the LLM returns a non-JSON fix response "{response}"')
def step_llm_returns_non_json_fix(context, response):
"""Configure the LLM to return plain text during fix generation."""
mock_response = Mock()
mock_response.content = response
context.mock_llm.invoke.side_effect = None
context.mock_llm.invoke.return_value = mock_response
@given("the LLM raises an exception during fix generation")
def step_llm_raises_exception_fix_generation(context):
"""Configure the LLM to raise during fix generation."""
context.mock_llm.invoke.side_effect = Exception("LLM fix generation failed")
@given("the LLM returns an empty non-mock response for fix generation")
def step_llm_returns_empty_fix_response(context):
"""Configure the LLM to return an empty string for fix generation."""
mock_response = Mock()
mock_response.content = ""
context.mock_llm.invoke.side_effect = None
context.mock_llm.invoke.return_value = mock_response
@then('the current_fix description should be "{description}"')
def step_current_fix_description_equals(context, description):
"""Verify the current_fix description."""
current_fix = context.state.get("current_fix", {})
assert current_fix.get("description") == description
@then('the current_fix code should be "{code}"')
def step_current_fix_code_equals(context, code):
"""Verify the current_fix code."""
current_fix = context.state.get("current_fix", {})
assert current_fix.get("code") == code
@then('the current_fix files_to_modify should contain "{filename}"')
def step_current_fix_files_contains(context, filename):
"""Ensure files_to_modify includes the provided filename."""
current_fix = context.state.get("current_fix", {})
files = current_fix.get("files_to_modify", [])
assert filename in files
@then('the current_fix description should contain "{text}"')
def step_current_fix_description_contains(context, text):
"""Verify the description contains specific text."""
current_fix = context.state.get("current_fix", {})
description = current_fix.get("description", "")
assert text in description
@then("the current_fix files_to_modify should be empty")
def step_current_fix_files_empty(context):
"""Ensure no files_to_modify are listed."""
current_fix = context.state.get("current_fix", {})
assert not current_fix.get("files_to_modify")
@then("the current_fix code should be empty string")
def step_current_fix_code_empty(context):
"""Ensure the fix code is an empty string."""
current_fix = context.state.get("current_fix", {})
assert current_fix.get("code") == ""
@given("I have a state with a current fix:")
def step_have_state_with_current_fix(context):
"""Create a state with a current fix."""
@@ -222,6 +340,78 @@ def step_have_state_with_current_fix(context):
}
@given("I have a state with current fix and empty attempted_fixes:")
def step_state_with_current_fix_and_empty_attempts(context):
"""Create a state with a current fix and no attempts."""
current_fix = json.loads(context.text)
context.state = {
"messages": [],
"current_fix": current_fix,
"attempted_fixes": [],
"error_message": "Test error",
"code_context": "test code",
}
@then("the attempted_fixes should have {count:d} entry")
@then("the attempted_fixes should have {count:d} entries")
def step_attempted_fixes_count(context, count):
"""Verify the number of attempted fixes."""
attempts = context.state.get("attempted_fixes", [])
assert len(attempts) == count
@then("the attempted_fixes should contain the current fix")
def step_attempted_fixes_contains_current(context):
"""Ensure the latest current fix is tracked in attempted_fixes."""
attempts = context.state.get("attempted_fixes", [])
current_fix = context.state.get("current_fix")
assert current_fix in attempts
@given("I have a state with current fix and one existing attempted fix")
def step_state_with_one_attempt(context):
"""Create state with one attempted fix already recorded."""
context.state = {
"messages": [],
"current_fix": {"description": "Another failed fix", "code": "more bad code"},
"attempted_fixes": [{"description": "Failed fix", "code": "bad code"}],
"error_message": "Test error",
"code_context": "test code",
}
@given("I have a state with two previous fix attempts")
def step_state_with_two_previous_attempts(context):
"""Create a state containing two previous attempts for prompt context."""
context.state = {
"messages": [
{
"role": "assistant",
"content": "Error analysis complete",
"type": "error_analysis",
}
],
"error_message": "Test error",
"code_context": "test code",
"attempted_fixes": [
{"description": "Failed fix 1", "code": "bad code 1"},
{"description": "Failed fix 2", "code": "bad code 2"},
],
}
@then("the attempt number should be {count:d}")
def step_attempt_number_is(context, count):
"""Verify the prompt references the expected attempt number."""
call_args = getattr(context.mock_llm.invoke, "call_args", None)
assert call_args, "LLM was not invoked"
messages = call_args[0][0]
human_message = messages[-1]
content = getattr(human_message, "content", "")
assert f"Generate fix attempt #{count}" in content
@when("I execute the validate_fix step")
def step_execute_validate_fix(context):
"""Execute the validate_fix step."""
@@ -242,6 +432,57 @@ def step_fix_validated_is_value(context, value):
assert context.state.get("fix_validated") == expected
@given("the LLM returns a valid JSON validation response with is_valid true")
def step_llm_returns_valid_json_validation_true(context):
"""Configure the LLM to return JSON with is_valid true."""
mock_response = Mock()
mock_response.content = json.dumps({"is_valid": True})
context.mock_llm.invoke.side_effect = None
context.mock_llm.invoke.return_value = mock_response
@given("the LLM returns a valid JSON validation response with is_valid false")
def step_llm_returns_valid_json_validation_false(context):
"""Configure the LLM to return JSON with is_valid false."""
mock_response = Mock()
mock_response.content = json.dumps({"is_valid": False})
context.mock_llm.invoke.side_effect = None
context.mock_llm.invoke.return_value = mock_response
@given('the LLM returns a non-JSON validation response "{response}"')
def step_llm_returns_non_json_validation(context, response):
"""Configure the LLM to return plain text during validation."""
mock_response = Mock()
mock_response.content = response
context.mock_llm.invoke.side_effect = None
context.mock_llm.invoke.return_value = mock_response
@given("the LLM raises an exception during validation")
def step_llm_raises_exception_validation(context):
"""Configure the LLM to raise during validation."""
context.mock_llm.invoke.side_effect = Exception("LLM validation failed")
@given('the LLM returns a JSON validation response with is_valid as string "{value}"')
def step_llm_returns_string_is_valid(context, value):
"""Configure JSON validation response with string is_valid."""
mock_response = Mock()
mock_response.content = json.dumps({"is_valid": value})
context.mock_llm.invoke.side_effect = None
context.mock_llm.invoke.return_value = mock_response
@given("the LLM returns a JSON validation response without is_valid field")
def step_llm_returns_json_without_is_valid(context):
"""Configure JSON lacking is_valid field."""
mock_response = Mock()
mock_response.content = json.dumps({"reasoning": "Missing flag"})
context.mock_llm.invoke.side_effect = None
context.mock_llm.invoke.return_value = mock_response
@given("I have an AutoDebugAgent instance with max_fix_attempts of {value:d}")
def step_have_auto_debug_agent_with_max_attempts(context, value):
"""Create AutoDebugAgent with specific max_fix_attempts."""
features/steps/auto_debug_cli_coverage_steps.py
+359
View File
@@ -0,0 +1,359 @@
from __future__ import annotations
from contextlib import ExitStack, contextmanager
from typing import Any, Iterable
from unittest.mock import MagicMock, patch
import typer
from behave import given, then, when
from typer.testing import CliRunner
from cleveragents.cli.commands.auto_debug import (
app as auto_debug_app,
auto_debug_command,
_get_current_project,
)
from cleveragents.application.services.plan_service import PlanService
from cleveragents.core.exceptions import CleverAgentsError, PlanError
from cleveragents.domain.models.core.project import Project
runner = CliRunner()
class _FakeLive:
"""Minimal Live replacement that writes updates to the console."""
def __init__(self, *_args, console=None, **_kwargs):
self._console = console
def __enter__(self) -> "_FakeLive":
return self
def __exit__(self, exc_type, exc, tb) -> bool: # noqa: D401
return False
def update(self, renderable) -> None:
if self._console is not None:
self._console.print(renderable)
@contextmanager
def _patched_container(context, project: Project | None):
with patch(
"cleveragents.application.container.get_container"
) as mock_get_container:
container = MagicMock()
mock_get_container.return_value = container
container.plan_service.return_value = context.plan_service_mock
container.project_service.return_value.get_current_project.return_value = (
project
)
yield container
def _capture_output(context: Any) -> str:
result = getattr(context, "result", None)
if isinstance(result, dict):
return result.get("output", "")
if result is not None:
stdout = getattr(result, "stdout", "") or ""
stderr = getattr(result, "stderr", "") or ""
output_attr = getattr(result, "output", "") or ""
combined = stdout + stderr
if combined:
return combined
if output_attr:
return output_attr
if hasattr(context, "output"):
return context.output
if hasattr(context, "command_output"):
return context.command_output
raise AssertionError("No CLI output captured in context")
def _capture_exit_code(context: Any) -> int:
result = getattr(context, "result", None)
if isinstance(result, dict):
return result.get("exit_code", 0)
if hasattr(result, "exit_code"):
return result.exit_code
if hasattr(context, "exit_code"):
return context.exit_code
raise AssertionError("No exit code recorded in context")
def _get_recorded_exception(context: Any) -> Exception | None:
return getattr(context, "exception", None) or getattr(
context, "call_exception", None
)
def _run_auto_debug_cli(context, extra_args: Iterable[str] | None = None) -> None:
project = context.project
assert project is not None, "Project must be initialized before running the CLI"
args: list[str] = []
if extra_args:
args.extend(extra_args)
with _patched_container(context, project):
with ExitStack() as stack:
stack.enter_context(
patch("cleveragents.cli.commands.auto_debug.Live", _FakeLive)
)
if getattr(context, "fail_fix_generation", False):
stack.enter_context(_patch_time_sleep_failure())
context.result = runner.invoke(auto_debug_app, args)
context.command_output = _capture_output(context)
@contextmanager
def _patch_time_sleep_failure():
calls = {"count": 0}
def _sleep_override(seconds): # noqa: ARG001
if calls["count"] == 0:
calls["count"] += 1
raise RuntimeError("Could not generate fix")
with patch("time.sleep", side_effect=_sleep_override):
yield
@given("I have a clean auto_debug test environment")
def step_clean_environment(context):
context.project = None
context.result = None
context.exception = None
context.command_output = ""
context.fail_fix_generation = False
context.plan_service_mock = MagicMock(spec=PlanService)
context.plan_service_mock.auto_debug_build.return_value = (True, [], None)
context.plan_service_mock.build_plan.return_value = []
context.plan_service_mock.build_plan.side_effect = None
@given("I have an initialized project for auto_debug")
def step_initialized_project(context):
context.project = Project(
id=1,
name="Test Project",
description="A test project",
path="/tmp",
)
@given("I have no project initialized")
def step_no_project(context):
context.project = None
@given("the plan service auto_debug_build will succeed")
def step_autodebug_success(context):
context.plan_service_mock.auto_debug_build.return_value = (True, [], None)
@given("the plan service auto_debug_build will fail")
def step_autodebug_failure(context):
context.plan_service_mock.auto_debug_build.return_value = (
False,
[],
"Simulated auto-debug failure",
)
@when("I call the auto_debug_command programmatic interface")
def step_call_autodebug_command(context):
with _patched_container(context, context.project):
context.result = auto_debug_command()
@when("I call the auto_debug_command with max_attempts {max_attempts:d}")
def step_call_autodebug_command_with_limit(context, max_attempts):
with _patched_container(context, context.project):
context.result = auto_debug_command(max_attempts=max_attempts)
@when("I call the auto_debug_command programmatic interface expecting error")
def step_call_autodebug_command_error(context):
try:
with _patched_container(context, context.project):
auto_debug_command()
except Exception as exc: # noqa: BLE001
context.exception = exc
@then("the auto_debug_command should return success True")
def step_assert_autodebug_success(context):
assert context.result[0] is True
@then("the auto_debug_command should return success False")
def step_assert_autodebug_failure(context):
assert context.result[0] is False
@then("the attempts made should be {count:d}")
def step_assert_attempts(context, count):
assert context.result[1] == count
@then('a CleverAgentsError should be raised with message "{message}"')
def step_assert_cleveragents_error(context, message):
exception = _get_recorded_exception(context)
assert exception is not None, (
"Expected CleverAgentsError but no exception was recorded"
)
assert isinstance(exception, CleverAgentsError)
assert message in str(exception)
@when("I call _get_current_project")
def step_call_get_current_project(context):
with _patched_container(context, context.project):
context.result = _get_current_project()
@when("I call _get_current_project expecting abort")
def step_call_get_current_project_abort(context):
try:
with _patched_container(context, None):
_get_current_project()
except typer.Abort as exc:
context.exception = exc
@then("the project should be returned successfully")
def step_assert_project_returned(context):
assert context.result == context.project
@then("typer.Abort should be raised")
def step_assert_abort(context):
assert isinstance(_get_recorded_exception(context), typer.Abort)
@given("the build will succeed with {changes:d} changes")
def step_build_success(context, changes):
context.plan_service_mock.build_plan.side_effect = None
context.plan_service_mock.build_plan.return_value = [
object() for _ in range(changes)
]
@given("the build will fail then succeed after fix")
def step_build_fail_then_succeed(context):
context.plan_service_mock.build_plan.side_effect = [
RuntimeError("Build failed"),
[object()],
]
@given('the build will always fail with error "{error_message}"')
def step_build_always_fail(context, error_message):
context.plan_service_mock.build_plan.side_effect = RuntimeError(error_message)
@given("the build will always fail with empty error message")
def step_build_fail_empty_error(context):
context.plan_service_mock.build_plan.side_effect = RuntimeError("")
@given("the plan service will raise a PlanError with details")
def step_planerror_details(context):
context.plan_service_mock.build_plan.side_effect = PlanError(
message="Plan Error",
details={"phase": "build"},
)
@given("the plan service will raise a PlanError without details")
def step_planerror_no_details(context):
context.plan_service_mock.build_plan.side_effect = PlanError(
message="Plan Error",
details=None,
)
@given("the plan service will raise a CleverAgentsError")
def step_plan_cleveragents_error(context):
context.plan_service_mock.build_plan.side_effect = CleverAgentsError(
"Generic Error"
)
@given("the build will fail with fix generation error")
def step_fix_generation_error(context):
context.fail_fix_generation = True
context.plan_service_mock.build_plan.side_effect = [
RuntimeError("Build failed"),
[object()],
]
@when("I run the auto_debug run command")
def step_run_cli(context):
_run_auto_debug_cli(context)
@when("I run the auto_debug run command with max_attempts {max_attempts:d}")
def step_run_cli_with_limit(context, max_attempts):
_run_auto_debug_cli(context, ["--max-attempts", str(max_attempts)])
@then("the command should exit with code {code:d}")
def step_assert_exit_code(context, code):
actual = _capture_exit_code(context)
if actual != code:
output = _capture_output(context)
assert actual == code, (
f"Expected exit code {code}, got {actual}. CLI output:\n{output}"
)
@then("the command should be aborted")
def step_assert_command_aborted(context):
step_assert_exit_code(context, 1)
@then('the output should contain "{text}"')
def step_assert_output_contains(context, text):
output = _capture_output(context)
assert text in output, f"Expected to find '{text}' in CLI output: {output}"
@given(
"the build will fail {num_failures:d} times then succeed with {changes:d} changes"
)
def step_build_fail_then_succeed_with_changes(context, num_failures, changes):
"""Configure build to fail num_failures times then succeed."""
failures = [
RuntimeError(f"Build failed attempt {i + 1}") for i in range(num_failures)
]
success_result = [object() for _ in range(changes)]
context.plan_service_mock.build_plan.side_effect = failures + [success_result]
@given("the build will raise PlanError after one attempt")
def step_build_raises_planerror_after_attempt(context):
"""Configure build to raise PlanError after one failed attempt."""
context.plan_service_mock.build_plan.side_effect = [
RuntimeError("Initial failure"),
PlanError(message="Plan Error during retry", details={"phase": "build"}),
]
@given("the build will raise CleverAgentsError after one attempt")
def step_build_raises_cleveragentserror_after_attempt(context):
"""Configure build to raise CleverAgentsError after one failed attempt."""
context.plan_service_mock.build_plan.side_effect = [
RuntimeError("Initial failure"),
CleverAgentsError("Critical error during retry"),
]
@given("the plan service will raise a PlanError with multiple details")
def step_planerror_multiple_details(context):
context.plan_service_mock.build_plan.side_effect = PlanError(
message="Plan Error",
details={"phase": "build", "step": "validation", "file": "test.py"},
)
features/steps/cli_plan_context_commands_steps.py
+4 -13
View File
@@ -281,7 +281,10 @@ def step_run_command(context, command):
env=os.environ.copy(), # Pass current environment variables
)
context.command_output = result.stdout
output = result.stdout or ""
if result.stderr:
output += result.stderr
context.command_output = output
context.command_error = result.stderr
context.command_exit_code = result.returncode
context.exit_code = result.returncode # For compatibility with other steps
@@ -433,18 +436,6 @@ def step_verify_current_plan(context, name):
assert current.name == name
@then('the output should contain "{text}"')
def step_output_contains(context, text):
"""Verify output contains text."""
# Check both stdout and stderr
combined_output = context.command_output + (
context.command_error if hasattr(context, "command_error") else ""
)
assert text in combined_output, (
f"Expected '{text}' not found in output: {combined_output}"
)
@then("the plan should have the additional instruction")
def step_plan_has_additional(context):
"""Verify plan has additional instruction."""
features/steps/context_service_analysis_steps.py
+452
View File
@@ -0,0 +1,452 @@
from __future__ import annotations
import asyncio
import shutil
import tempfile
from pathlib import Path
from types import MethodType
from typing import Any
from unittest.mock import MagicMock
from behave import given, then, when
from cleveragents.agents.context_analysis import ContextAnalysisAgent
from cleveragents.application.services.context_service import ContextService
from cleveragents.config.settings import Settings
from cleveragents.domain.models.core import Plan, Project
def _ensure_temp_project(context) -> None:
if hasattr(context, "project") and hasattr(context, "temp_dir_path"):
return
temp_dir = Path(tempfile.mkdtemp(prefix="context-analysis-"))
context.temp_dir_path = temp_dir
if hasattr(context, "add_cleanup"):
context.add_cleanup(shutil.rmtree, temp_dir, True)
context.project = Project(id=1, name="context-analysis-project", path=temp_dir)
def _ensure_context_service(context) -> None:
if hasattr(context, "context_service"):
return
_ensure_temp_project(context)
settings = Settings()
unit_of_work = MagicMock()
transaction_manager = unit_of_work.transaction.return_value
transaction_context = MagicMock()
transaction_context.plans = MagicMock()
plan = Plan(
id=42,
project_id=context.project.id or 1,
name="analysis-plan",
prompt="Analyze context for coverage",
current=True,
)
transaction_context.plans.get_current_for_project.return_value = plan
transaction_manager.__enter__.return_value = transaction_context
transaction_manager.__exit__.return_value = False
service = ContextService(settings=settings, unit_of_work=unit_of_work)
context.plan_metadata = plan
context.context_files: list[Path] = []
context.created_files: dict[str, Path] = {}
def list_files_stub(
self: ContextService, project: Project | None = None
) -> list[str]:
return [str(path) for path in context.context_files]
service.list_files = MethodType(list_files_stub, service)
context.context_service = service
def _create_file(context, filename: str, content: str) -> Path:
_ensure_temp_project(context)
file_path = context.temp_dir_path / filename
file_path.parent.mkdir(parents=True, exist_ok=True)
file_path.write_text(content.strip("\n"), encoding="utf-8")
context.created_files[filename] = file_path
return file_path
def _add_file_to_context(context, filename: str) -> Path:
if filename not in context.created_files:
raise AssertionError(f"File '{filename}' has not been created yet")
file_path = context.created_files[filename]
if file_path not in context.context_files:
context.context_files.append(file_path)
return file_path
@given("I have initialized a CleverAgents project")
def step_init_project(context):
_ensure_temp_project(context)
@given("I have a context service with LangGraph integration")
def step_context_service(context):
_ensure_context_service(context)
@given("LangSmith tracing is enabled for context analysis metadata")
def step_enable_langsmith(context):
_ensure_context_service(context)
context.context_service.settings.langsmith_enabled = True
context.context_service.settings.langsmith_project = "behave-context-coverage"
@given("the current plan has no context files")
def step_no_context_files(context):
_ensure_context_service(context)
context.context_files.clear()
@given('I have a Python file "{filename}" with content:')
def step_create_file(context, filename):
_ensure_context_service(context)
_create_file(context, filename, context.text)
@given('I have added "{filename}" to the LangGraph context')
def step_add_file(context, filename):
_ensure_context_service(context)
_add_file_to_context(context, filename)
@given("I have added a non-existent file path to the analysis")
def step_add_missing_file(context):
_ensure_context_service(context)
missing_path = context.temp_dir_path / "missing_file.py"
context.created_files["missing_file.py"] = missing_path
context.context_files.append(missing_path)
context.nonexistent_path = missing_path
@when("I analyze the context")
def step_analyze_context(context):
_ensure_context_service(context)
context.analysis_result = context.context_service.analyze_context(context.project)
@when("I analyze the context asynchronously")
def step_analyze_context_async(context):
_ensure_context_service(context)
context.async_analysis_result = asyncio.run(
context.context_service.analyze_context_async(context.project)
)
@when("I analyze the context with the non-existent path")
def step_analyze_nonexistent(context):
_ensure_context_service(context)
context.nonexistent_result = context.context_service.analyze_context(
context.project
)
@when("I stream the context analysis")
def step_stream_context(context):
_ensure_context_service(context)
context.streaming_events = list(
context.context_service.analyze_context_streaming(context.project)
)
@when("I stream the context analysis asynchronously")
def step_stream_context_async(context):
_ensure_context_service(context)
async def _collect_events() -> list[dict[str, Any]]:
events: list[dict[str, Any]] = []
async for event in context.context_service.analyze_context_streaming_async(
context.project
):
events.append(event)
return events
context.async_streaming_events = asyncio.run(_collect_events())
@when("I get the context summary")
def step_get_summary(context):
_ensure_context_service(context)
context.summary_text = context.context_service.get_context_summary(context.project)
@when("I get the context dependencies")
def step_get_dependencies(context):
_ensure_context_service(context)
context.dependency_map = context.context_service.get_context_dependencies(
context.project
)
@when("I get relevant files with threshold {value:f}")
def step_get_relevant_files(context, value: float):
_ensure_context_service(context)
context.relevant_files = context.context_service.get_relevant_files(
context.project, threshold=value
)
@when('I prepare the LangGraph analysis config for run "{run_name}" in "{mode}" mode')
def step_prepare_langsmith_config(context, run_name: str, mode: str):
_ensure_context_service(context)
context.langsmith_config = context.context_service._prepare_analysis_config(
context.project,
run_name=run_name,
file_paths=[str(path) for path in context.context_files],
mode=mode,
)
@when("I request a context analysis agent")
def step_request_context_agent(context):
_ensure_context_service(context)
context.requested_agent = context.context_service._get_context_agent()
@then("the analysis result should have empty documents")
def step_assert_empty_documents(context):
docs = context.analysis_result["documents"]
assert len(docs) == 0, "Expected no documents in the analysis result"
@then("the analysis summary should indicate no files to analyze")
def step_assert_empty_summary(context):
summary = context.analysis_result["summary"]
assert "no context files" in summary.lower(), summary
@then("there should be no error in the analysis")
def step_assert_no_error(context):
assert context.analysis_result["error"] in (None, ""), context.analysis_result[
"error"
]
@then("the analysis result should have {count:d} document")
@then("the analysis result should have {count:d} documents")
def step_assert_document_count(context, count: int):
docs = context.analysis_result["documents"]
assert len(docs) == count, f"Expected {count} documents, got {len(docs)}"
@then('the dependencies for "{filename}" should include "{token}"')
def step_assert_dependencies(context, filename: str, token: str):
file_path = str(context.created_files[filename])
dependencies = context.analysis_result["dependencies"].get(file_path, [])
assert any(token in entry for entry in dependencies), dependencies
@then('the relevance score for "{filename}" should be between {low:f} and {high:f}')
def step_assert_relevance_range(context, filename: str, low: float, high: float):
file_path = str(context.created_files[filename])
score = context.analysis_result["relevance_scores"].get(file_path)
assert score is not None, f"No score for {filename}"
assert low <= score <= high, f"Score {score} not in range [{low}, {high}]"
@then("the summary should not be empty")
def step_assert_summary_not_empty(context):
assert context.analysis_result["summary"].strip(), "Summary is empty"
@then("the summary should be a non-empty string")
def step_assert_summary_string(context):
assert context.summary_text.strip(), "Summary text is empty"
@then("the summary should not indicate an error")
def step_assert_summary_no_error(context):
assert "failed" not in context.summary_text.lower(), context.summary_text
@then("the dependencies should include entries for both files")
def step_assert_dependencies_two(context):
dep_map = context.analysis_result["dependencies"]
assert len(dep_map.keys()) >= 2, dep_map
@then("the relevance scores should have entries for both files")
def step_assert_scores_two(context):
relevance = context.analysis_result["relevance_scores"]
assert len(relevance.keys()) >= 2, relevance
@then('the dependencies dict should have an entry for "{filename}"')
def step_assert_dependency_entry(context, filename: str):
file_path = str(context.created_files[filename])
deps = context.dependency_map.get(file_path, [])
assert deps, f"No dependencies recorded for {filename}"
@then("the dependencies should be a non-empty list")
def step_assert_dependency_list(context):
assert all(context.dependency_map.values()), context.dependency_map
@then("the result should include both files with scores")
def step_assert_relevant_files(context):
paths = {Path(path) for path, _ in context.relevant_files}
expected = {path for path in context.context_files}
assert expected.issubset(paths), f"Missing files in scores: {expected - paths}"
@then("all LangGraph scores should be between {low:f} and {high:f}")
def step_assert_all_scores(context, low: float, high: float):
for _, score in context.relevant_files:
assert low <= score <= high, f"Score {score} outside [{low}, {high}]"
@then("I should receive multiple events")
def step_assert_multiple_events(context):
assert len(context.streaming_events) >= 2, context.streaming_events
@then("the events should include node execution results")
def step_assert_node_events(context):
node_names = {
"load_files",
"analyze_dependencies",
"chunk_documents",
"score_relevance",
"summarize_context",
}
assert any(
isinstance(event, dict) and node_names.intersection(event.keys())
for event in context.streaming_events
), context.streaming_events
@then("the streaming output should report no files to analyze")
def step_assert_streaming_no_files(context):
assert context.streaming_events == [
{"type": "complete", "summary": "No context files to analyze"}
], context.streaming_events
@then("the async analysis result should have empty documents")
def step_assert_async_empty_documents(context):
docs = context.async_analysis_result["documents"]
assert not docs, f"Expected no async documents, got {len(docs)} entries"
@then("the async analysis summary should indicate no files to analyze")
def step_assert_async_empty_summary(context):
summary = context.async_analysis_result["summary"].lower()
assert "no context files" in summary, summary
@then("the async result should have documents")
def step_assert_async_documents(context):
assert context.async_analysis_result["documents"], "Expected async documents"
@then("the async result should have a summary")
def step_assert_async_summary(context):
assert context.async_analysis_result["summary"].strip(), "Async summary empty"
@then("there should be no error in the async result")
def step_assert_async_no_error(context):
assert context.async_analysis_result["error"] in (None, "")
@then("the analysis should complete without raising an exception")
def step_assert_nonexistent_completed(context):
assert context.nonexistent_result is not None
@then("the error field should contain file not found information")
def step_assert_nonexistent_error(context):
error = context.nonexistent_result["error"]
assert error and "file not found" in error.lower(), error
@then("I should receive a ContextAnalysisAgent instance")
def step_assert_agent_instance(context):
assert isinstance(context.requested_agent, ContextAnalysisAgent)
@then("the agent should have the standard workflow nodes")
def step_assert_agent_nodes(context):
expected = {
"load_files",
"analyze_dependencies",
"chunk_documents",
"score_relevance",
"summarize_context",
}
actual = set(context.requested_agent.graph.nodes.keys())
assert expected.issubset(actual), actual
@then("the summary should not indicate an error in the dependency run")
def step_assert_dependency_summary(context):
assert context.analysis_result["error"] in (None, "")
@then('the dependencies should be a non-empty list for "{filename}"')
def step_assert_dependencies_specific(context, filename: str):
file_path = str(context.created_files[filename])
deps = context.analysis_result["dependencies"].get(file_path, [])
assert deps, f"Empty dependency list for {filename}"
@then("I should receive node execution details in streaming events")
def step_assert_streaming_details(context):
node_names = {
"load_files",
"analyze_dependencies",
"chunk_documents",
"score_relevance",
"summarize_context",
}
matched = [
event
for event in context.streaming_events
if node_names.intersection(event.keys())
]
assert matched, context.streaming_events
@then("the LangSmith config should include the current project metadata")
def step_assert_langsmith_metadata(context):
config = getattr(context, "langsmith_config", {})
metadata = config.get("metadata", {})
tags = config.get("tags", [])
assert metadata.get("project_id") == context.project.id, metadata
assert metadata.get("plan_id") == context.plan_metadata.id, metadata
assert metadata.get("service") == "ContextService", metadata
assert any(tag.startswith("project:") for tag in tags), tags
@then("the LangSmith config should record {count:d} context file path")
def step_assert_langsmith_count(context, count: int):
config = getattr(context, "langsmith_config", {})
metadata = config.get("metadata", {})
assert metadata.get("context_file_count") == count, metadata
@then("the async streaming events should include node execution results")
def step_assert_async_streaming_events(context):
node_names = {
"load_files",
"analyze_dependencies",
"chunk_documents",
"score_relevance",
"summarize_context",
}
assert any(
isinstance(event, dict) and node_names.intersection(event.keys())
for event in context.async_streaming_events
), context.async_streaming_events
@then("the async streaming output should report no files to analyze")
def step_assert_async_streaming_no_files(context):
assert context.async_streaming_events == [
{"type": "complete", "summary": "No context files to analyze"}
], context.async_streaming_events
features/steps/langsmith_config_steps.py
+76
View File
@@ -0,0 +1,76 @@
from __future__ import annotations
import os
from typing import Any
from behave import given, then, when
from features.environment import LANGSMITH_ENV_VARS
from cleveragents.config import settings as settings_module
def _ensure_env_tracking(context: Any) -> None:
if not hasattr(context, "env_vars_to_clean"):
context.env_vars_to_clean = []
@given("LangSmith environment is clean")
def step_clear_langsmith_env(context):
_ensure_env_tracking(context)
for key in LANGSMITH_ENV_VARS:
os.environ.pop(key, None)
context.env_vars_to_clean.clear()
@given('I set environment variable "{name}" to "{value}"')
def step_set_env_var(context, name: str, value: str):
_ensure_env_tracking(context)
os.environ[name] = value
context.env_vars_to_clean.append(name)
@when("I load CleverAgents settings for LangSmith")
def step_load_settings(context):
settings_module.Settings._instance = None
context.langsmith_settings = settings_module.get_settings()
@when('I build a LangSmith config with run name "{run_name}"')
def step_build_langsmith_config(context, run_name: str):
settings_obj = getattr(context, "langsmith_settings", None)
assert settings_obj is not None, "LangSmith settings not loaded"
context.langsmith_config = settings_obj.build_langsmith_config(
tags=["context-analysis", "behave-test"],
metadata={"scenario": "langsmith_config", "run_name": run_name},
run_name=run_name,
)
@then("LangSmith tracing should be disabled")
def step_assert_tracing_disabled(context):
settings_obj = getattr(context, "langsmith_settings", None)
assert settings_obj is not None, "LangSmith settings not loaded"
assert not settings_obj.is_langsmith_enabled, "Tracing unexpectedly enabled"
@then("LangSmith tracing should be enabled")
def step_assert_tracing_enabled(context):
settings_obj = getattr(context, "langsmith_settings", None)
assert settings_obj is not None, "LangSmith settings not loaded"
assert settings_obj.is_langsmith_enabled, "Tracing was not enabled"
@then('the LangSmith config should include tag "{tag}"')
def step_assert_config_tag(context, tag: str):
config = getattr(context, "langsmith_config", None)
assert config, "LangSmith config was not built"
tags = config.get("tags", [])
assert tag in tags, f"Expected tag '{tag}' in {tags}"
@then('the LangSmith config should include metadata key "{key}"')
def step_then_the_langsmith_config_should_include_metadata_key(context, key):
config = getattr(context, "langsmith_config", None)
assert config, "LangSmith config was not built"
metadata = config.get("metadata", {})
assert key in metadata, f"Expected metadata key '{key}' in {metadata}"
features/steps/plan_full_coverage_steps.py
-7
View File
@@ -997,13 +997,6 @@ def step_call_plan_helper_without_project(context):
# Then steps for assertions
@then('a CleverAgentsError should be raised with message "{message}"')
def step_assert_cleveragents_error_message(context, message):
"""Assert that a CleverAgentsError was raised with the expected message."""
assert isinstance(context.call_exception, CleverAgentsError)
assert str(context.call_exception) == message
@then("the programmatic plan apply_command should report {count:d} applied changes")
def step_assert_programmatic_apply_changes(context, count):
"""Assert apply_command returned the expected change count."""
features/steps/plan_service_steps.py
+456
View File
@@ -6,6 +6,7 @@ import tempfile
from collections.abc import Callable
from datetime import datetime
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import MagicMock, patch
from behave import given, then, when
@@ -334,6 +335,164 @@ def step_check_plan_error(context: Context, message: str) -> None:
assert message in str(context.exception.message)
@given("I configured an auto-debug plan with a disappearing plan ID")
def step_configure_auto_debug_disappearing_plan(context: Context) -> None:
"""Set up a plan service whose current plan loses its ID mid-run."""
assert hasattr(context, "temp_dir"), (
"Temporary directory setup is required before configuring auto-debug"
)
class FlickeringPlan:
def __init__(self, plan_id: int) -> None:
self._plan_id = plan_id
self._first_access = True
self.prompt = "Investigate flaky build"
self.name = "auto-debug-plan"
self.status = PlanStatus.PENDING
self.current = True
self.updated_at = datetime.now()
self.result = None
@property
def id(self) -> int | None:
if self._first_access:
self._first_access = False
return self._plan_id
return None
class FakePlansRepository:
def __init__(self, plan):
self.plan = plan
def get_current_for_project(self, project_id: int):
return self.plan
class FakeContextsRepository:
def __init__(self) -> None:
self.called = False
def get_for_plan(self, plan_id: int):
self.called = True
return []
class FakeDebugAttemptsRepository:
def __init__(self) -> None:
self.records: list = []
def add(self, attempt):
attempt.id = len(self.records) + 1
self.records.append(attempt)
return attempt
def get(self, attempt_id: int):
return None
def update(self, attempt):
self.updated_attempt = attempt
class FakeUnitOfWork:
def __init__(self, contexts: list[SimpleNamespace]) -> None:
self._contexts = contexts
def transaction(self):
outer = self
class _Transaction:
def __enter__(self_inner):
if not outer._contexts:
raise AssertionError("No fake transaction contexts remaining")
return outer._contexts.pop(0)
def __exit__(self_inner, exc_type, exc_val, exc_tb):
return False
return _Transaction()
flicker_plan = FlickeringPlan(plan_id=101)
contexts_repo = FakeContextsRepository()
debug_repo = FakeDebugAttemptsRepository()
fake_contexts = [
SimpleNamespace(plans=FakePlansRepository(flicker_plan)),
SimpleNamespace(contexts=contexts_repo),
SimpleNamespace(debug_attempts=debug_repo),
]
context.auto_debug_context_repo = contexts_repo
context.auto_debug_debug_repo = debug_repo
context.plan_service = PlanService(
settings=Settings(),
unit_of_work=FakeUnitOfWork(fake_contexts),
ai_provider=None,
)
context.project = Project(
id=1,
name="auto-debug-project",
path=context.temp_dir,
created_at=datetime.now(),
updated_at=datetime.now(),
current_plan_id=1,
settings=ProjectSettings(
auto_build=False,
auto_apply=False,
confirm_apply=True,
max_context_size=50 * 1024 * 1024,
default_model="mock-gpt",
),
)
@when("I run auto-debug build with missing plan context")
def step_run_auto_debug_missing_context(context: Context) -> None:
"""Trigger auto-debug build to exercise empty context handling."""
context.plan_service.build_plan = MagicMock(
side_effect=RuntimeError("Plan build failed")
)
captured_state: dict[str, object] = {}
class DummyAutoDebugAgent:
def __init__(self, **_kwargs):
pass
def invoke(self, debug_state, config=None):
captured_state["state"] = debug_state
captured_state["config"] = config
return {"result": {"success": False, "fix": None}}
with (
patch.object(
context.plan_service, "_prepare_langsmith_config", return_value={}
),
patch(
"cleveragents.application.agents.auto_debug.AutoDebugAgent",
DummyAutoDebugAgent,
),
):
context.auto_debug_result = context.plan_service.auto_debug_build(
context.project, max_attempts=1
)
context.auto_debug_captured_state = captured_state.get("state")
@then("the auto-debug build should fall back to an empty context")
def step_assert_auto_debug_empty_context(context: Context) -> None:
"""Verify empty code context and debug attempt tracking."""
result = getattr(context, "auto_debug_result", None)
assert result is not None, "Auto-debug result was not captured"
success, changes, error = result
assert success is False
assert changes == []
assert error == "Plan build failed"
contexts_repo = getattr(context, "auto_debug_context_repo", None)
assert contexts_repo is not None, "Contexts repository tracking missing"
assert contexts_repo.called is False
debug_state = getattr(context, "auto_debug_captured_state", None)
assert debug_state is not None, "Debug state was not recorded"
assert debug_state.get("code_context") == ""
@given("I have a saved project with current plan missing ID")
def step_create_project_plan_no_id(context: Context) -> None:
"""Create a project with current plan that has no ID."""
@@ -725,6 +884,49 @@ def step_check_nested_move(context: Context) -> None:
)
@given("the plan has a pending MOVE change to an absolute path")
def step_add_absolute_move_change(context: Context) -> None:
"""Add a MOVE change whose destination is an absolute path."""
assert getattr(context, "project", None) is not None, "Project is required"
assert getattr(context, "current_plan", None) is not None, (
"Current plan is required"
)
source_path = context.project.path / "absolute_source.py"
source_path.write_text("# Absolute move source")
absolute_target = (context.project.path / "absolute_target.py").resolve()
with context.unit_of_work.transaction() as ctx:
change = Change(
id=None,
plan_id=context.current_plan.id,
file_path="absolute_source.py",
operation=OperationType.MOVE,
original_content="# Absolute move source",
new_content="# Absolute move source",
new_path=str(absolute_target),
applied=False,
applied_at=None,
created_at=datetime.now(),
)
ctx.changes.add(change)
context.absolute_move_paths = (source_path, absolute_target)
@then("the absolute destination should exist and the source should be removed")
def step_check_absolute_move_destination(context: Context) -> None:
"""Ensure MOVE changes create files at absolute targets."""
paths = getattr(context, "absolute_move_paths", None)
assert paths is not None, "Absolute move paths were not recorded"
source_path, absolute_target = paths
assert not source_path.exists(), "Source file still exists after MOVE"
destination_path = Path(absolute_target)
assert destination_path.exists(), "Absolute destination file missing"
assert destination_path.read_text() == "# Absolute move source"
assert getattr(context, "applied_count", 0) == 1
@given("the plan has changes with absolute file paths")
def step_add_absolute_path_changes(context: Context) -> None:
"""Add changes with absolute file paths."""
@@ -1444,3 +1646,257 @@ def step_verify_memory_service_reused(context: Context, session_id: str) -> None
assert monitored_setter is not None
assert monitored_setter.call_count == 0
assert monitored_service.max_messages is not None
@given("LangSmith integration is enabled for plan service")
def step_enable_langsmith_integration(context: Context) -> None:
"""Reconfigure the plan service with LangSmith support enabled."""
settings = Settings()
settings.langsmith_enabled = True
original_builder = Settings.build_langsmith_config
def _forward_build_config(
self: Settings, *, tags=None, metadata=None, run_name=None
):
return original_builder(
self,
tags=tags,
metadata=metadata,
run_name=run_name,
)
builder_patch = patch.object(
Settings,
"build_langsmith_config",
autospec=True,
side_effect=_forward_build_config,
)
builder_mock = builder_patch.start()
cleanup = getattr(context, "add_cleanup", None)
if callable(cleanup):
cleanup(builder_patch.stop)
else: # pragma: no cover - behave contexts should supply add_cleanup
context._langsmith_builder_patch = builder_patch # type: ignore[attr-defined]
context.langsmith_builder_mock = builder_mock
context.plan_service = PlanService(
settings=settings,
unit_of_work=context.unit_of_work,
ai_provider=None,
)
@when(
'I prepare a LangSmith config for project "{project_name}" and plan "{plan_name}"'
)
def step_prepare_langsmith_config(
context: Context, project_name: str, plan_name: str
) -> None:
"""Build and prepare a LangSmith config to capture metadata and tags."""
project = Project(
id=101,
name=project_name,
path=context.temp_dir,
created_at=datetime.now(),
updated_at=datetime.now(),
current_plan_id=None,
settings=ProjectSettings(
auto_build=False,
auto_apply=False,
confirm_apply=True,
max_context_size=50 * 1024 * 1024,
default_model="langsmith",
),
)
plan = Plan(
id=202,
project_id=project.id,
name=plan_name,
prompt="LangSmith coverage",
status=PlanStatus.PENDING,
current=True,
created_at=datetime.now(),
updated_at=datetime.now(),
build=None,
build_started_at=None,
build_completed_at=None,
model_used=None,
token_count=None,
result=None,
applied_at=None,
files_created=None,
files_modified=None,
files_deleted=None,
)
config = context.plan_service._prepare_langsmith_config(
project,
plan,
run_name="coverage",
tags=["custom-tag"],
metadata={"env": "test"},
)
context.langsmith_project = project
context.langsmith_plan = plan
context.langsmith_prepared_config = config
builder_mock = getattr(context, "langsmith_builder_mock", None)
if builder_mock is not None:
context.langsmith_builder_call = builder_mock.call_args
@then(
'the LangSmith builder should receive metadata for project "{project_name}" and plan "{plan_name}"'
)
def step_assert_langsmith_builder_metadata(
context: Context, project_name: str, plan_name: str
) -> None:
"""Verify the builder received merged metadata and tags."""
call = getattr(context, "langsmith_builder_call", None)
assert call is not None, "LangSmith builder was not invoked"
_, kwargs = call
metadata = kwargs.get("metadata", {})
tags = kwargs.get("tags", [])
project = getattr(context, "langsmith_project", None)
plan = getattr(context, "langsmith_plan", None)
assert metadata.get("project_name") == project_name
assert metadata.get("plan_name") == plan_name
assert metadata.get("env") == "test"
assert metadata.get("project_id") == getattr(project, "id", None)
assert metadata.get("plan_id") == getattr(plan, "id", None)
assert "service:plan" in tags
assert f"project:{getattr(project, 'id', None)}" in tags
assert f"plan:{getattr(plan, 'id', None)}" in tags
assert "custom-tag" in tags
@then("the prepared LangSmith config should include a generated thread id")
def step_assert_langsmith_thread_id(context: Context) -> None:
config = getattr(context, "langsmith_prepared_config", None)
assert config, "LangSmith config was not prepared"
configurable = config.get("configurable")
assert isinstance(configurable, dict)
thread_id = configurable.get("thread_id")
assert isinstance(thread_id, str)
assert thread_id.startswith("plan-service-")
@given("I have a plan service with a failing AI provider")
def step_plan_service_with_failing_provider(context: Context) -> None:
"""Configure the plan service to surface provider failures."""
failing_provider = MagicMock()
failing_provider.generate_changes.return_value = ProviderResponse(
changes=[],
model_used="mock-gpt",
token_count=0,
error_message="mock provider failure",
)
context.plan_service = PlanService(
settings=Settings(),
unit_of_work=context.unit_of_work,
ai_provider=failing_provider,
)
context.mock_provider = failing_provider
@when("I attempt to build the plan and the provider returns an error")
def step_build_plan_with_provider_error(context: Context) -> None:
"""Invoke build_plan and capture the provider error."""
try:
context.plan_service.build_plan(context.project)
context.exception = None
except Exception as exc: # pragma: no cover - defensive
context.exception = exc
@when("I run the auto debug build for the project")
def step_run_auto_debug_without_plan(context: Context) -> None:
"""Run auto_debug_build while capturing exceptions."""
try:
context.auto_debug_result = context.plan_service.auto_debug_build(
context.project
)
context.exception = None
except Exception as exc:
context.auto_debug_result = None
context.exception = exc
@given("auto debug retries should run with a failing build pipeline")
def step_force_auto_debug_failure(context: Context) -> None:
"""Mark that auto_debug_build should simulate failing build attempts."""
context.force_auto_debug_failure = True
@when("I run the auto debug build with max attempts {attempts:d}")
def step_run_auto_debug_with_attempts(context: Context, attempts: int) -> None:
"""Execute auto_debug_build, optionally forcing failures for coverage."""
def failing_build(
self: PlanService,
project: Project,
progress_callback: Callable[[int], None] | None = None,
) -> list[Change]:
raise RuntimeError("Simulated auto debug failure")
agent_path = "cleveragents.application.agents.auto_debug.AutoDebugAgent"
try:
if getattr(context, "force_auto_debug_failure", False):
with patch.object(PlanService, "build_plan", side_effect=failing_build):
with patch(agent_path) as agent_cls:
agent_instance = MagicMock()
agent_instance.invoke.return_value = {
"result": {"success": False, "fix": {}},
}
agent_cls.return_value = agent_instance
context.auto_debug_result = context.plan_service.auto_debug_build(
context.project, max_attempts=attempts
)
else:
context.auto_debug_result = context.plan_service.auto_debug_build(
context.project, max_attempts=attempts
)
context.exception = None
except Exception as exc:
context.auto_debug_result = None
context.exception = exc
@then('the auto debug result should indicate failure with error message "{message}"')
def step_assert_auto_debug_failure(context: Context, message: str) -> None:
"""Validate that auto_debug_build returned a failure tuple."""
result = getattr(context, "auto_debug_result", None)
assert result is not None, "Auto debug result was not captured"
success, changes, error_message = result
assert success is False
assert isinstance(changes, list)
assert isinstance(error_message, str)
assert message in error_message
@when("I try to stream plan generation without an AI provider")
def step_stream_generate_without_provider(context: Context) -> None:
"""Attempt to consume the streaming generator without configuring an AI provider."""
import asyncio
async def _consume_stream() -> None:
generator = context.plan_service.generate_plan_streaming(
context.project,
description="Stream coverage scenario",
)
await generator.__anext__()
try:
asyncio.run(_consume_stream())
context.exception = None
except Exception as exc:
context.exception = exc
features/steps/project_commands_coverage_steps.py
-6
View File
@@ -452,12 +452,6 @@ def step_check_suggestion(context, command):
assert command in context.result.output
@then("the command should exit with code {code:d}")
def step_check_exit_code(context, code):
"""Check specific exit code."""
assert context.result.exit_code == code
@then("the error message should be shown")
def step_check_error_shown(context):
"""Check error message shown."""
features/steps/settings_steps.py
+106
View File
@@ -91,6 +91,21 @@ def step_check_environment(context, expected):
assert context.settings.env == expected
@when('I set the environment alias to "{value}"')
def step_set_environment_alias(context, value):
"""Update the environment alias property."""
assert hasattr(context, "settings"), (
"Settings must be loaded before updating the environment alias."
)
context.settings.environment = value
@then('the environment alias should be "{expected}"')
def step_check_environment_alias(context, expected):
"""Validate the environment alias getter."""
assert context.settings.environment == expected
@then('the database URL should be "{expected}"')
def step_check_database_url(context, expected):
"""Check the database URL."""
@@ -121,6 +136,30 @@ def step_set_data_directory(context, directory):
context.test_data_dir = directory
@given('the storage base path is "{directory}"')
def step_set_storage_base_path(context, directory):
"""Record the storage base directory used for tests."""
context.storage_base_dir = directory
@when('I compute the storage path for "{storage_type}"')
def step_compute_storage_path(context, storage_type):
"""Compute the storage path from the configured base."""
base_dir = getattr(context, "storage_base_dir", None)
assert base_dir is not None, "Storage base path must be set before computing it."
settings = Settings.model_construct(storage_base_path=Path(base_dir))
context.storage_path = settings.get_storage_base_path(storage_type)
@when("I compute the default storage path")
def step_compute_default_storage_path(context):
"""Compute the storage path when no type is supplied."""
base_dir = getattr(context, "storage_base_dir", None)
assert base_dir is not None, "Storage base path must be set before computing it."
settings = Settings.model_construct(storage_base_path=Path(base_dir))
context.storage_path = settings.get_storage_base_path()
@when('I get the storage base path for "{storage_type}"')
def step_get_storage_path(context, storage_type):
"""Get the storage base path."""
@@ -151,6 +190,21 @@ def step_check_is_production(context):
context.is_production = settings.is_production()
@when(
'I evaluate production mode with env "{env_value}", debug "{debug_flag}", and reload "{reload_flag}"'
)
def step_evaluate_production_mode_flags(context, env_value, debug_flag, reload_flag):
"""Evaluate production helpers for the supplied flags."""
Settings._instance = None
settings = Settings.model_construct(
env=env_value,
debug_enabled=debug_flag.lower() == "true",
server_reload=reload_flag.lower() == "true",
)
context.is_production = settings.is_production()
context.is_production_mode = settings.is_production_mode
@then("is_production should be {expected}")
def step_verify_is_production(context, expected):
"""Verify the is_production value."""
@@ -158,6 +212,13 @@ def step_verify_is_production(context, expected):
assert context.is_production == expected_bool
@then("is_production_mode should be {expected}")
def step_verify_is_production_mode(context, expected):
"""Verify the is_production_mode alias."""
expected_bool = expected == "True"
assert context.is_production_mode == expected_bool
@given('the {provider} API key is set to "{key}"')
def step_set_provider_key(context, provider, key):
"""Set a provider API key."""
@@ -168,6 +229,13 @@ def step_set_provider_key(context, provider, key):
context.env_vars_to_clean.append(env_key)
@when('I instantiate settings with explicit openai API key "{value}"')
def step_instantiate_settings_with_openai(context, value):
"""Instantiate settings with a constructor-specified OpenAI key."""
Settings._instance = None
context.settings = Settings.model_construct(openai_api_key=value)
@when('I check if "{provider}" provider is configured')
def step_check_provider_configured(context, provider):
"""Check if a provider is configured."""
@@ -178,6 +246,16 @@ def step_check_provider_configured(context, provider):
)
@when('I clear the explicit "{provider}" key and recheck configuration')
def step_clear_explicit_key_and_recheck(context, provider):
"""Clear the in-memory provider key and re-evaluate configuration."""
Settings._instance = None
settings = Settings()
attr = provider.lower().replace(" ", "_") + "_api_key"
setattr(settings, attr, None)
context.provider_configured = settings.has_provider_configured(provider)
@then('has_provider_configured should be {expected} for "{provider}"')
def step_verify_provider_configured(context, expected, provider):
"""Verify provider configuration status."""
@@ -243,6 +321,20 @@ def step_get_database_url(context):
context.database_url = settings.get_database_url()
@when('I derive the test database URL from "{url}"')
def step_derive_test_database_url(context, url):
"""Derive the fallback test database URL when none is provided."""
Settings._instance = None
settings = Settings.model_construct(database_url=url, test_database_url=None)
context.derived_database_url = settings.get_database_url(test=True)
@then('the derived test database URL should be "{expected}"')
def step_check_derived_test_database_url(context, expected):
"""Verify the derived test database URL."""
assert context.derived_database_url == expected
@when("I get the settings instance")
def step_get_settings(context):
"""Get the settings instance."""
@@ -266,3 +358,17 @@ def step_check_provider_api_key(context, provider, expected):
"""Check a provider API key value."""
provider_key = provider.lower().replace(" ", "_") + "_api_key"
assert getattr(context.settings, provider_key) == expected
@when("I build the LangSmith config")
def step_build_langsmith_config(context):
"""Build the LangSmith configuration payload."""
Settings._instance = None
settings = Settings()
context.langsmith_config = settings.build_langsmith_config()
@then("the LangSmith config should be absent")
def step_assert_langsmith_config_absent(context):
"""Ensure LangSmith config is not generated when disabled."""
assert context.langsmith_config is None
implementation_plan.md
+10 -3
View File
@@ -1509,6 +1509,12 @@ async def generate_plan_streaming(
- Verified MockAIProvider already uses FakeListLLM (implementation complete)
- All high-priority Phase 2 foundation tasks now complete
**2025-11-25 (Context Service Integration Validation):**
- Verified `ContextService` now routes sync, async, and streaming analysis flows through `ContextAnalysisAgent` (`src/cleveragents/application/services/context_service.py:408`, `src/cleveragents/application/services/context_service.py:470`, `src/cleveragents/application/services/context_service.py:517`), confirming Stage 2.7.4 integration is production-ready.
- Added Behave coverage for the service workflows via `features/context_service_analysis.feature:6` with dedicated steps in `features/steps/context_service_analysis_steps.py:66-344`, exercising file creation, dependency mapping, relevance thresholds, async runs, and streaming events.
- Renamed the ambiguous step text to `I have added "{filename}" to the LangGraph context` in `features/steps/context_service_analysis_steps.py:88` to avoid clashing with the legacy step in `features/steps/service_steps.py:1004`.
- Ran `nox -s unit_tests -- features/context_service_analysis.feature` to confirm all ten scenarios (75 steps) pass after the integration fixes.
**Deferred to Future Weeks:**
- EntityMemory Integration (moved to Phase 3)
- LangSmith observability setup (optional, can be done in Phase 6)
@@ -4032,10 +4038,11 @@ If you can do all of the above by end of Day 1, you're on track!
- [X] All tests passing (nox -s unit_tests, nox -s integration_tests)
- [X] Code: Fix LangGraph checkpointing with thread_id in config
- [ ] Code: Integrate agents into services
- [ ] Update `ContextService` to use `ContextAnalysisAgent`
- [ ] Add streaming support to context commands
- [X] Update `ContextService` to use `ContextAnalysisAgent` (`src/cleveragents/application/services/context_service.py:408-515`).
- [X] Add streaming support to context commands via `analyze_context_streaming*` methods (`src/cleveragents/application/services/context_service.py:517-596`).
- [ ] Add LangSmith metadata for context analysis
- [ ] Tests: Test end-to-end integration
- [X] Tests: Test end-to-end integration
- [X] Added service-level coverage in `features/context_service_analysis.feature:6` with steps from `features/steps/context_service_analysis_steps.py:66-344`, executed via `nox -s unit_tests -- features/context_service_analysis.feature`.
- [X] Stage 2.7.5: Auto-Debug Agent Implementation (COMPLETE 2025-11-22)
- [X] Code: Implement AutoDebugGraph
- [X] Create `src/cleveragents/agents/auto_debug.py` (194 lines modified)
noxfile.py
+14 -14
View File
@@ -15,14 +15,14 @@ nox.options.error_on_external_run = True
@nox.session(python=DEFAULT_PYTHON, reuse_venv=True, venv_backend="uv")
def lint(session):
def lint(session: nox.Session):
"""Check code formatting and linting."""
session.install("ruff")
session.run("ruff", "check", "src/", "scripts/", "examples/")
@nox.session(python=DEFAULT_PYTHON, reuse_venv=True, venv_backend="uv")
def format(session):
def format(session: nox.Session):
"""Format code with ruff."""
session.install("ruff")
session.run("ruff", "format", ".")
@@ -32,7 +32,7 @@ def format(session):
@nox.session(python=DEFAULT_PYTHON, reuse_venv=True, venv_backend="uv")
def typecheck(session):
def typecheck(session: nox.Session):
"""Check types with pyright."""
session.install("pyright")
session.install("-e", ".")
@@ -40,7 +40,7 @@ def typecheck(session):
@nox.session(python=SUPPORTED_PYTHONS, reuse_venv=True, venv_backend="uv")
def unit_tests(session):
def unit_tests(session: nox.Session):
"""Run BDD tests with Behave."""
session.install("-e", ".[tests]")
session.install("behave")
@@ -76,28 +76,28 @@ def unit_tests(session):
@nox.session(python=DEFAULT_PYTHON, reuse_venv=True, venv_backend="uv")
def docs(session):
def docs(session: nox.Session):
"""Build documentation with MkDocs."""
session.install("-e", ".[docs]")
session.run("mkdocs", "build")
@nox.session(python=DEFAULT_PYTHON, reuse_venv=True, venv_backend="uv")
def serve_docs(session):
def serve_docs(session: nox.Session):
"""Serve docs locally for development."""
session.install("-e", ".[docs]")
session.run("mkdocs", "serve", "--dev-addr", "0.0.0.0:8000")
@nox.session(python=SUPPORTED_PYTHONS, reuse_venv=True, venv_backend="uv")
def build(session):
def build(session: nox.Session):
"""Build the wheel distribution."""
session.install("build")
session.run("python", "-m", "build", "--wheel")
@nox.session(python=SUPPORTED_PYTHONS, reuse_venv=True, venv_backend="uv")
def integration_tests(session):
def integration_tests(session: nox.Session):
"""Run Robot Framework integration tests (excluding discovery tests)."""
session.install("-e", ".[tests]")
@@ -150,7 +150,7 @@ def integration_tests(session):
@nox.session(python=SUPPORTED_PYTHONS, reuse_venv=True, venv_backend="uv")
def slow_integration_tests(session):
def slow_integration_tests(session: nox.Session):
"""Run Robot Framework integration tests."""
session.install("-e", ".[tests]")
session.run(
@@ -171,7 +171,7 @@ def slow_integration_tests(session):
@nox.session(python=DEFAULT_PYTHON, reuse_venv=True, venv_backend="uv")
def coverage_report(session):
def coverage_report(session: nox.Session):
"""Generate coverage report from Behave tests."""
session.install("-e", ".[tests]")
@@ -203,7 +203,7 @@ def coverage_report(session):
@nox.session(python=DEFAULT_PYTHON, reuse_venv=True, venv_backend="uv")
def discovery_tests(session):
def discovery_tests(session: nox.Session):
"""Run Phase 0 discovery tests (tagged with @discovery)."""
session.install("-e", ".[tests]")
session.install("behave")
@@ -221,7 +221,7 @@ def discovery_tests(session):
@nox.session(python=DEFAULT_PYTHON, reuse_venv=True, venv_backend="uv")
def discovery_coverage(session):
def discovery_coverage(session: nox.Session):
"""Generate coverage report for discovery tests."""
session.install("-e", ".[tests]")
@@ -252,7 +252,7 @@ def discovery_coverage(session):
@nox.session(python=SUPPORTED_PYTHONS, reuse_venv=True, venv_backend="uv")
def discovery_integration(session):
def discovery_integration(session: nox.Session):
"""Run Robot Framework integration tests tagged with 'discovery'."""
session.install("-e", ".[tests]")
@@ -300,7 +300,7 @@ def discovery_integration(session):
@nox.session(python=DEFAULT_PYTHON, reuse_venv=True, venv_backend="uv")
def benchmark(session):
def benchmark(session: nox.Session):
"""Run Airspeed Velocity benchmarks and publish results."""
session.install("-e", ".[tests]")
config_path = "asv.conf.json"
src/cleveragents/application/services/context_service.py
+370
View File
@@ -3,12 +3,19 @@
This service handles adding, removing, and managing files that the AI
will use as context when generating code changes. Uses repository pattern
and Unit of Work for persistence (ADR-007).
Includes LangGraph-based context analysis for intelligent context processing.
"""
from __future__ import annotations
import hashlib
import uuid
from collections.abc import AsyncIterator, Iterator
from datetime import datetime
from fnmatch import fnmatch
from pathlib import Path
from typing import TYPE_CHECKING, Any
from cleveragents.config.settings import Settings
from cleveragents.core.exceptions import FileSystemError, PlanError
@@ -18,6 +25,14 @@ from cleveragents.infrastructure.database.unit_of_work import (
UnitOfWorkContext,
)
if TYPE_CHECKING:
from langchain_core.language_models import BaseLanguageModel
from cleveragents.agents.context_analysis import (
ContextAnalysisAgent,
ContextAnalysisState,
)
class ContextService:
"""Service for managing context files in plans.
@@ -371,3 +386,358 @@ class ContextService:
contexts = self.list_context(project)
return [context.path for context in contexts]
def _get_current_plan(self, project: Project) -> Plan | None:
"""Fetch the current plan for metadata enrichment."""
if not project.id:
return None
with self.unit_of_work.transaction() as ctx:
return ctx.plans.get_current_for_project(project.id)
def _build_langsmith_config(
self,
project: Project,
*,
run_name: str,
file_paths: list[str],
mode: str,
) -> dict[str, Any]:
"""Construct metadata for LangSmith tracing if enabled."""
if not getattr(self.settings, "is_langsmith_enabled", False):
return {}
plan = self._get_current_plan(project)
metadata = {
"project_id": project.id,
"project_name": project.name,
"plan_id": getattr(plan, "id", None),
"plan_name": getattr(plan, "name", None),
"context_file_count": len(file_paths),
"service": "ContextService",
"mode": mode,
}
tags = [
"context-analysis",
"service:context",
f"mode:{mode}",
]
if project.id is not None:
tags.append(f"project:{project.id}")
return (
self.settings.build_langsmith_config(
tags=tags,
metadata=metadata,
run_name=run_name,
)
or {}
)
def _prepare_analysis_config(
self,
project: Project,
*,
run_name: str,
file_paths: list[str],
mode: str,
) -> dict[str, Any]:
config = self._build_langsmith_config(
project,
run_name=run_name,
file_paths=file_paths,
mode=mode,
)
config.setdefault("configurable", {})
config["configurable"]["thread_id"] = f"context-analysis-{uuid.uuid4()}"
return config
# --- LangGraph-based Context Analysis Methods ---
def _get_context_agent(
self, llm: BaseLanguageModel | None = None
) -> ContextAnalysisAgent:
"""Get or create a ContextAnalysisAgent instance.
Args:
llm: Optional language model to use. If None, uses default mock.
Returns:
ContextAnalysisAgent instance for analyzing context.
"""
# Import here to avoid circular dependency and allow lazy loading
from cleveragents.agents.context_analysis import ContextAnalysisAgent
return ContextAnalysisAgent(llm=llm)
def analyze_context(
self,
project: Project,
llm: BaseLanguageModel | None = None,
) -> ContextAnalysisState:
"""Analyze the current plan's context using LangGraph workflow.
Uses the ContextAnalysisAgent to perform intelligent analysis of the
context files including dependency extraction, relevance scoring,
and summarization.
Args:
project: The project containing the plan
llm: Optional language model to use for analysis
Returns:
ContextAnalysisState containing analysis results including:
- documents: Loaded documents
- dependencies: Extracted dependencies per file
- relevance_scores: Relevance scores for each file
- chunks: Chunked documents for large files
- summary: High-level summary of the context
- error: Error message if any
Raises:
PlanError: If no current plan exists
"""
# Get file paths from current context
file_paths = self.list_files(project)
if not file_paths:
# Return empty analysis for empty context
from cleveragents.agents.context_analysis import ContextAnalysisState
return ContextAnalysisState(
file_paths=[],
documents=[],
dependencies={},
summary="No context files to analyze",
relevance_scores={},
chunks=[],
error=None,
)
# Create agent and run analysis
agent = self._get_context_agent(llm)
config = self._prepare_analysis_config(
project,
run_name="ContextService.analyze_context",
file_paths=file_paths,
mode="sync",
)
initial_state: ContextAnalysisState = {
"file_paths": file_paths,
"documents": [],
"dependencies": {},
"summary": "",
"relevance_scores": {},
"chunks": [],
"error": None,
}
return agent.invoke(initial_state, config)
async def analyze_context_async(
self,
project: Project,
llm: BaseLanguageModel | None = None,
) -> ContextAnalysisState:
"""Asynchronously analyze the current plan's context.
Async version of analyze_context for non-blocking execution.
Args:
project: The project containing the plan
llm: Optional language model to use for analysis
Returns:
ContextAnalysisState containing analysis results
"""
from cleveragents.agents.context_analysis import ContextAnalysisState
file_paths = self.list_files(project)
if not file_paths:
return ContextAnalysisState(
file_paths=[],
documents=[],
dependencies={},
summary="No context files to analyze",
relevance_scores={},
chunks=[],
error=None,
)
agent = self._get_context_agent(llm)
config = self._prepare_analysis_config(
project,
run_name="ContextService.analyze_context_async",
file_paths=file_paths,
mode="async",
)
initial_state: ContextAnalysisState = {
"file_paths": file_paths,
"documents": [],
"dependencies": {},
"summary": "",
"relevance_scores": {},
"chunks": [],
"error": None,
}
return await agent.ainvoke(initial_state, config)
def analyze_context_streaming(
self,
project: Project,
llm: BaseLanguageModel | None = None,
) -> Iterator[dict[str, Any]]:
"""Stream the context analysis workflow execution.
Yields events as each workflow node completes, allowing for
real-time progress tracking.
Args:
project: The project containing the plan
llm: Optional language model to use for analysis
Yields:
Dictionary containing node execution events
"""
file_paths = self.list_files(project)
if not file_paths:
yield {"type": "complete", "summary": "No context files to analyze"}
return
agent = self._get_context_agent(llm)
config = self._prepare_analysis_config(
project,
run_name="ContextService.analyze_context_streaming",
file_paths=file_paths,
mode="stream",
)
from cleveragents.agents.graphs.context_analysis import ContextAnalysisState
initial_state = ContextAnalysisState(
file_paths=file_paths,
documents=[],
dependencies={},
summary="",
relevance_scores={},
chunks=[],
error=None,
)
yield from agent.stream(initial_state, config)
async def analyze_context_streaming_async(
self,
project: Project,
llm: BaseLanguageModel | None = None,
) -> AsyncIterator[dict[str, Any]]:
"""Asynchronously stream the context analysis workflow execution.
Async version of analyze_context_streaming for non-blocking streaming.
Args:
project: The project containing the plan
llm: Optional language model to use for analysis
Yields:
Dictionary containing node execution events
"""
file_paths = self.list_files(project)
if not file_paths:
yield {"type": "complete", "summary": "No context files to analyze"}
return
agent = self._get_context_agent(llm)
config = self._prepare_analysis_config(
project,
run_name="ContextService.analyze_context_streaming_async",
file_paths=file_paths,
mode="stream",
)
from cleveragents.agents.graphs.context_analysis import ContextAnalysisState
initial_state = ContextAnalysisState(
file_paths=file_paths,
documents=[],
dependencies={},
summary="",
relevance_scores={},
chunks=[],
error=None,
)
async for event in agent.astream(initial_state, config):
yield event
def get_context_summary(
self,
project: Project,
llm: BaseLanguageModel | None = None,
) -> str:
"""Get a high-level summary of the current context.
Convenience method that runs the full analysis and returns
just the summary.
Args:
project: The project containing the plan
llm: Optional language model to use for analysis
Returns:
Summary string describing the context
"""
result = self.analyze_context(project, llm)
return result["summary"] if result["summary"] else "No summary available"
def get_context_dependencies(
self,
project: Project,
llm: BaseLanguageModel | None = None,
) -> dict[str, list[str]]:
"""Get extracted dependencies for all context files.
Convenience method that runs the full analysis and returns
just the dependency information.
Args:
project: The project containing the plan
llm: Optional language model to use for analysis
Returns:
Dictionary mapping file paths to their dependencies
"""
result = self.analyze_context(project, llm)
return result["dependencies"]
def get_relevant_files(
self,
project: Project,
threshold: float = 0.5,
llm: BaseLanguageModel | None = None,
) -> list[tuple[str, float]]:
"""Get context files sorted by relevance score.
Filters and sorts context files by their relevance scores.
Args:
project: The project containing the plan
threshold: Minimum relevance score (0.0 to 1.0)
llm: Optional language model to use for analysis
Returns:
List of (file_path, score) tuples sorted by score descending
"""
result = self.analyze_context(project, llm)
scores = result["relevance_scores"]
filtered = [
(path, score) for path, score in scores.items() if score >= threshold
]
return sorted(filtered, key=lambda x: x[1], reverse=True)
src/cleveragents/application/services/plan_service.py
+79 -2
View File
@@ -7,6 +7,7 @@ for persistence (ADR-007).
from __future__ import annotations
import uuid
from collections.abc import AsyncIterator, Callable
from datetime import datetime
from pathlib import Path
@@ -135,6 +136,70 @@ class PlanService:
if service and forget_history:
service.clear()
def _build_langsmith_config(
self,
project: Project,
plan: Plan | None,
*,
run_name: str,
tags: list[str] | None = None,
metadata: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""Build base LangSmith configuration with project/plan metadata."""
if not getattr(self.settings, "is_langsmith_enabled", False):
return {}
base_metadata: dict[str, Any] = {
"project_id": project.id,
"project_name": project.name,
}
if plan and plan.id:
base_metadata["plan_id"] = plan.id
base_metadata["plan_name"] = plan.name
if metadata:
base_metadata |= metadata
base_tags = ["service:plan"]
if project.id is not None:
base_tags.append(f"project:{project.id}")
if plan and plan.id is not None:
base_tags.append(f"plan:{plan.id}")
if tags:
base_tags.extend(tags)
return (
self.settings.build_langsmith_config(
tags=base_tags,
metadata=base_metadata,
run_name=run_name,
)
or {}
)
def _prepare_langsmith_config(
self,
project: Project,
plan: Plan | None,
*,
run_name: str,
tags: list[str] | None = None,
metadata: dict[str, Any] | None = None,
thread_prefix: str = "plan-service",
) -> dict[str, Any]:
config = self._build_langsmith_config(
project,
plan,
run_name=run_name,
tags=tags,
metadata=metadata,
)
if not config:
return {}
config.setdefault("configurable", {})
config["configurable"]["thread_id"] = f"{thread_prefix}-{uuid.uuid4()}"
return config
def create_plan(
self, project: Project, prompt: str, name: str | None = None
) -> Plan:
@@ -405,8 +470,20 @@ class PlanService:
"metadata": {},
}
# Run debug workflow
final_state = agent.invoke(debug_state)
# Run debug workflow with LangSmith metadata when enabled
config = self._prepare_langsmith_config(
project,
current_plan,
run_name=f"PlanService.auto_debug_build.attempt_{attempt_number}",
tags=["auto-debug", f"attempt:{attempt_number}"],
metadata={
"attempt_number": attempt_number,
"max_attempts": max_attempts,
"last_error": last_error,
},
thread_prefix="auto-debug",
)
final_state = agent.invoke(debug_state, config or None)
# Extract the fix result
result = final_state.get("result", {})
src/cleveragents/cli/commands/auto_debug.py
+3
View File
@@ -147,6 +147,9 @@ def run(
break
except Exception as e:
if isinstance(e, (PlanError, CleverAgentsError)):
raise
last_error = str(e)
elapsed = time() - attempt_start
src/cleveragents/config/settings.py
+241 -138
View File
@@ -1,167 +1,270 @@
"""Settings management for CleverAgents.
Based on ADR-006: CLEVERAGENTS Environment Variable Management.
All application variables use CLEVERAGENTS_ prefix.
Provider variables (OPENAI_API_KEY, etc.) remain unchanged.
"""
from __future__ import annotations
import os
from pathlib import Path
from typing import Any, ClassVar
from pydantic import Field
from pydantic import AliasChoices, Field
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
"""CleverAgents configuration from environment variables.
All variables use CLEVERAGENTS_ prefix except provider keys.
Configuration hierarchy: defaults -> env -> file -> CLI
"""
"""Application runtime configuration backed by environment variables."""
model_config = SettingsConfigDict(
env_prefix="CLEVERAGENTS_",
env_file=".env",
env_file_encoding="utf-8",
case_sensitive=False,
validate_default=True,
extra="forbid", # Fail on unknown variables
env_prefix="cleveragents_", case_sensitive=False, extra="ignore"
)
# Server Settings (CLEVERAGENTS_SERVER_*)
server_host: str = "0.0.0.0" # CLEVERAGENTS_SERVER_HOST
server_port: int = Field(default=8080, ge=1, le=65535) # CLEVERAGENTS_SERVER_PORT
server_workers: int = Field(default=4, ge=1, le=100) # CLEVERAGENTS_SERVER_WORKERS
server_reload: bool = False # CLEVERAGENTS_SERVER_RELOAD
_instance: ClassVar[Settings | None] = None
_PROVIDER_ENV_MAP: ClassVar[dict[str, tuple[str, ...]]] = {
"openai_api_key": ("OPENAI_API_KEY",),
"anthropic_api_key": ("ANTHROPIC_API_KEY",),
"google_api_key": ("GOOGLE_API_KEY", "GOOGLE_GENAI_API_KEY"),
"azure_api_key": ("AZURE_OPENAI_API_KEY", "AZURE_API_KEY"),
"openrouter_api_key": ("OPENROUTER_API_KEY",),
"gemini_api_key": ("GEMINI_API_KEY", "GOOGLE_GEMINI_API_KEY"),
"hf_token": ("HF_TOKEN", "HUGGINGFACEHUB_API_TOKEN", "HUGGING_FACE_HUB_TOKEN"),
"cohere_api_key": ("COHERE_API_KEY",),
"perplexity_api_key": ("PERPLEXITY_API_KEY",),
"groq_api_key": ("GROQ_API_KEY",),
"together_api_key": ("TOGETHER_API_KEY",),
}
_PROVIDER_ALIAS: ClassVar[dict[str, str]] = {
"openai": "openai_api_key",
"anthropic": "anthropic_api_key",
"google": "google_api_key",
"azure": "azure_api_key",
"openrouter": "openrouter_api_key",
"router": "openrouter_api_key",
"gemini": "gemini_api_key",
"huggingface": "hf_token",
"hf": "hf_token",
"cohere": "cohere_api_key",
"perplexity": "perplexity_api_key",
"groq": "groq_api_key",
"together": "together_api_key",
}
# Database Settings (CLEVERAGENTS_DATABASE_*)
database_url: str = "sqlite:///cleveragents.db" # CLEVERAGENTS_DATABASE_URL
database_pool_size: int = Field(
default=10, ge=1, le=100
) # CLEVERAGENTS_DATABASE_POOL_SIZE
database_pool_timeout: float = Field(
default=30.0, gt=0
) # CLEVERAGENTS_DATABASE_POOL_TIMEOUT
database_echo: bool = False # CLEVERAGENTS_DATABASE_ECHO
# Cache Settings (CLEVERAGENTS_CACHE_*)
cache_backend: str = Field(
default="memory", pattern="^(memory|redis|sqlite)$"
) # CLEVERAGENTS_CACHE_BACKEND
cache_ttl: int = Field(default=3600, ge=0) # CLEVERAGENTS_CACHE_TTL
cache_redis_url: str | None = None # CLEVERAGENTS_CACHE_REDIS_URL
# Auth Settings (CLEVERAGENTS_AUTH_*)
auth_enabled: bool = True # CLEVERAGENTS_AUTH_ENABLED
auth_session_timeout: int = Field(
default=86400, ge=60
) # CLEVERAGENTS_AUTH_SESSION_TIMEOUT
auth_jwt_secret: str | None = None # CLEVERAGENTS_AUTH_JWT_SECRET
auth_allow_registration: bool = True # CLEVERAGENTS_AUTH_ALLOW_REGISTRATION
# Storage Settings (CLEVERAGENTS_STORAGE_*)
storage_path: Path = Path.home() / ".cleveragents" # CLEVERAGENTS_STORAGE_PATH
storage_plans_dir: str = "plans" # CLEVERAGENTS_STORAGE_PLANS_DIR
storage_contexts_dir: str = "contexts" # CLEVERAGENTS_STORAGE_CONTEXTS_DIR
storage_temp_dir: str = "tmp" # CLEVERAGENTS_STORAGE_TEMP_DIR
# Model Settings (CLEVERAGENTS_MODEL_*)
model_default_chat: str = "gpt-4" # CLEVERAGENTS_MODEL_DEFAULT_CHAT
model_default_embedding: str = (
"text-embedding-ada-002" # CLEVERAGENTS_MODEL_DEFAULT_EMBEDDING
# Runtime/server configuration
env: str = Field(
default="development",
validation_alias=AliasChoices("CLEVERAGENTS_ENV"),
)
server_host: str = Field(
default="0.0.0.0",
validation_alias=AliasChoices("CLEVERAGENTS_SERVER_HOST"),
)
server_port: int = Field(
default=8080,
validation_alias=AliasChoices("CLEVERAGENTS_SERVER_PORT"),
)
server_reload: bool = Field(
default=False,
validation_alias=AliasChoices("CLEVERAGENTS_SERVER_RELOAD"),
)
debug_enabled: bool = Field(
default=False,
validation_alias=AliasChoices("CLEVERAGENTS_DEBUG_ENABLED"),
)
model_timeout: float = Field(default=120.0, gt=0) # CLEVERAGENTS_MODEL_TIMEOUT
model_max_retries: int = Field(
default=3, ge=0, le=10
) # CLEVERAGENTS_MODEL_MAX_RETRIES
# Telemetry Settings (CLEVERAGENTS_TELEMETRY_*)
telemetry_enabled: bool = False # CLEVERAGENTS_TELEMETRY_ENABLED
telemetry_endpoint: str | None = None # CLEVERAGENTS_TELEMETRY_ENDPOINT
telemetry_sample_rate: float = Field(
default=1.0, ge=0.0, le=1.0
) # CLEVERAGENTS_TELEMETRY_SAMPLE_RATE
# Debug Settings (CLEVERAGENTS_DEBUG_*)
debug_enabled: bool = False # CLEVERAGENTS_DEBUG_ENABLED
# Logging/paths
log_level: str = Field(
default="INFO",
validation_alias=AliasChoices("CLEVERAGENTS_LOG_LEVEL"),
)
debug_log_level: str = Field(
default="INFO", pattern="^(DEBUG|INFO|WARNING|ERROR|CRITICAL)$"
) # CLEVERAGENTS_DEBUG_LOG_LEVEL
debug_log_format: str = Field(
default="json", pattern="^(json|text)$"
) # CLEVERAGENTS_DEBUG_LOG_FORMAT
debug_trace_enabled: bool = False # CLEVERAGENTS_DEBUG_TRACE_ENABLED
default="INFO",
validation_alias=AliasChoices("CLEVERAGENTS_DEBUG_LOG_LEVEL"),
)
log_dir: Path = Field(
default_factory=lambda: Path("logs"),
validation_alias=AliasChoices("CLEVERAGENTS_LOG_DIR"),
)
data_dir: Path = Field(
default_factory=lambda: Path("data"),
validation_alias=AliasChoices("CLEVERAGENTS_DATA_DIR"),
)
storage_base_path: Path = Field(
default_factory=lambda: Path("data"),
validation_alias=AliasChoices("CLEVERAGENTS_STORAGE_BASE_PATH"),
)
# Testing Settings (CLEVERAGENTS_TESTING_*)
testing_use_mock_ai: bool = False # CLEVERAGENTS_TESTING_USE_MOCK_AI
# Persistence
database_url: str = Field(
default="sqlite:///cleveragents.db",
validation_alias=AliasChoices("CLEVERAGENTS_DATABASE_URL"),
)
test_database_url: str = Field(
default="sqlite:///cleveragents_test.db",
validation_alias=AliasChoices("CLEVERAGENTS_TEST_DATABASE_URL"),
)
# Provider API Keys (no prefix)
openai_api_key: str | None = Field(default=None, alias="OPENAI_API_KEY")
anthropic_api_key: str | None = Field(default=None, alias="ANTHROPIC_API_KEY")
google_api_key: str | None = Field(default=None, alias="GOOGLE_API_KEY")
azure_api_key: str | None = Field(default=None, alias="AZURE_API_KEY")
openrouter_api_key: str | None = Field(default=None, alias="OPENROUTER_API_KEY")
gemini_api_key: str | None = Field(default=None, alias="GEMINI_API_KEY")
hf_token: str | None = Field(default=None, alias="HF_TOKEN")
# LangSmith
langsmith_enabled: bool = Field(
default=False,
validation_alias=AliasChoices("CLEVERAGENTS_LANGSMITH_ENABLED"),
)
langsmith_api_key: str | None = Field(
default=None,
validation_alias=AliasChoices("CLEVERAGENTS_LANGSMITH_API_KEY"),
)
langsmith_project: str | None = Field(
default=None,
validation_alias=AliasChoices("CLEVERAGENTS_LANGSMITH_PROJECT"),
)
langsmith_tags: list[str] = Field(
default_factory=list,
validation_alias=AliasChoices("CLEVERAGENTS_LANGSMITH_TAGS"),
)
# Provider keys (populated from both prefixed and provider env vars)
openai_api_key: str | None = Field(default=None)
anthropic_api_key: str | None = Field(default=None)
google_api_key: str | None = Field(default=None)
azure_api_key: str | None = Field(default=None)
openrouter_api_key: str | None = Field(default=None)
gemini_api_key: str | None = Field(default=None)
hf_token: str | None = Field(default=None)
cohere_api_key: str | None = Field(default=None)
perplexity_api_key: str | None = Field(default=None)
groq_api_key: str | None = Field(default=None)
together_api_key: str | None = Field(default=None)
def model_post_init(self, __context: Any) -> None: # type: ignore[override]
super().model_post_init(__context)
self._apply_external_env_overrides()
# ------------------------------------------------------------------
# Singleton helpers
# ------------------------------------------------------------------
@classmethod
def get_settings(cls: type[Settings]) -> Settings:
"""Return the cached singleton instance (creating it if needed)."""
if cls._instance is None:
cls._instance = cls()
return cls._instance
# ------------------------------------------------------------------
# Derived properties & helpers
# ------------------------------------------------------------------
@property
def environment(self) -> str:
"""Backwards compatible alias for ``env``."""
return self.env
@environment.setter
def environment(self, value: str) -> None:
self.env = value
@property
def storage_base_path(self) -> Path:
"""Computed property for base storage path."""
return self.storage_path / "data"
def storage_path(self) -> Path:
"""Absolute path backing runtime storage."""
base = self.storage_base_path
if not base.is_absolute():
base = Path.cwd() / base
return base.resolve()
@property
def is_production(self) -> bool:
"""Check if running in production mode."""
return not self.debug_enabled and not self.server_reload
def get_storage_base_path(self, storage_type: str | None = None) -> Path:
"""Return the base storage path optionally scoped by type."""
if storage_type:
storage_type = storage_type.strip()
if storage_type:
return self.storage_path / storage_type
return self.storage_path
def get_database_url(self, test: bool = False) -> str:
"""Get database URL with test suffix if needed.
Args:
test: If True, return test database URL
Returns:
Database URL string
"""
if test and "sqlite" in self.database_url:
return self.database_url.replace(".db", "_test.db")
def get_database_url(self, *, test: bool = False) -> str:
"""Return the primary or test database URL."""
if test:
if self.test_database_url:
return self.test_database_url
return self._derive_test_database_url(self.database_url)
return self.database_url
def has_provider_configured(self) -> bool:
"""Check if at least one AI provider is configured.
Returns:
True if any provider API key is set
"""
return any(
[
self.openai_api_key,
self.anthropic_api_key,
self.google_api_key,
self.azure_api_key,
self.openrouter_api_key,
self.gemini_api_key,
self.hf_token,
]
def is_production(self) -> bool:
"""Whether the runtime matches production settings."""
return (
not self.debug_enabled
and not self.server_reload
and self.env.lower() == "production"
)
@property
def is_production_mode(self) -> bool:
"""Convenience alias for :meth:`is_production`."""
return self.is_production()
# Global settings instance (lazy loaded)
_settings: Settings | None = None
@property
def is_langsmith_enabled(self) -> bool:
"""Check if LangSmith tracing is enabled."""
if self.langsmith_enabled:
return True
return os.getenv("LANGCHAIN_TRACING_V2", "false").lower() == "true"
def has_provider_configured(self, provider: str | None = None) -> bool:
"""Return True when any (or a specific) provider has credentials configured."""
if provider:
attr = self._PROVIDER_ALIAS.get(provider.strip().lower())
if not attr:
return False
return bool(self._provider_value(attr))
return any(self._provider_value(attr) for attr in self._PROVIDER_ENV_MAP)
def build_langsmith_config(
self,
*,
tags: list[str] | None = None,
metadata: dict[str, Any] | None = None,
run_name: str | None = None,
) -> dict[str, Any] | None:
"""Build the config for LangSmith tracing."""
if not self.is_langsmith_enabled:
return None
config: dict[str, Any] = {
"tags": tags or [],
"metadata": metadata or {},
}
if run_name:
config["run_name"] = run_name
if self.langsmith_project:
config["metadata"].setdefault("langsmith_project", self.langsmith_project)
return config
# ------------------------------------------------------------------
# Internal helpers
# ------------------------------------------------------------------
def _apply_external_env_overrides(self) -> None:
"""Allow provider/native env vars to override defaults."""
for attr, env_names in self._PROVIDER_ENV_MAP.items():
if getattr(self, attr):
continue
for env_name in env_names:
value = os.getenv(env_name)
if value:
setattr(self, attr, value)
break
def _provider_value(self, attr: str) -> str | None:
"""Return the configured value for a provider attr (env fallback)."""
value = getattr(self, attr, None)
if value:
return value
for env_name in self._PROVIDER_ENV_MAP.get(attr, ()): # type: ignore[arg-type]
env_value = os.getenv(env_name)
if env_value:
return env_value
return None
@staticmethod
def _derive_test_database_url(url: str) -> str:
if url.startswith("sqlite") and url.endswith(".db"):
return url[:-3] + "_test.db"
return url
def get_settings() -> Settings:
"""Get the global settings instance.
Returns:
Settings instance (singleton)
"""
global _settings
if _settings is None:
_settings = Settings()
return _settings
__all__ = [
"Settings",
"get_settings",
]
"""Get the application settings."""
return Settings.get_settings()