cleveragents/cleveragents-core
4
0
Fork 3
Code Issues 6.0k Pull Requests 115 Actions Packages Projects Releases Wiki Activity

feat: Add Q0: Pre-commit hooks setup.

Browse Source 
This should automatically check for problems on build.
This commit is contained in:
Brent Edwards
2026-02-10 00:01:51 +00:00
committed by Forgejo
parent e5f27249b1
commit 7ef5ebb695
62 changed files with 2254 additions and 886 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.forgejo/pull_request_template.md
+48
View File
@@ -0,0 +1,48 @@
## Description
<!-- Brief description of what this PR does -->
## Type of Change
- [ ] Bug fix (non-breaking change which fixes an issue)
- [ ] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing functionality to not work as expected)
- [ ] Refactoring (no functional changes)
- [ ] Documentation update
- [ ] Test improvements
- [ ] CI/CD changes
## Quality Checklist
- [ ] Code follows the project's coding standards (see CONTRIBUTING.md)
- [ ] All public/protected methods have argument validation
- [ ] Static typing is complete (no `Any` unless justified)
- [ ] `nox -s typecheck` passes with no errors
- [ ] `nox -s lint` passes with no errors
- [ ] Unit tests written/updated (Behave scenarios in `features/`)
- [ ] Integration tests written/updated (Robot suites in `robot/`) if applicable
- [ ] Coverage remains above 85% (`nox -s coverage_report`)
- [ ] No security issues introduced (`nox -s security_scan`)
- [ ] No dead code introduced (`nox -s dead_code`)
- [ ] Documentation updated if behavior changed
## Testing
<!-- Describe how you tested your changes -->
### Test Commands Run
```bash
nox -s unit_tests # Behave tests
nox -s typecheck # Type checking
nox -s lint # Linting
nox -s coverage_report # Coverage
```
## Related Issues
<!-- Link to related issues: Closes #123, Fixes #456 -->
## Implementation Notes
<!-- Any architectural decisions, trade-offs, or important notes for reviewers -->
.forgejo/workflows/ci.yml
+113 -20
View File
@@ -17,19 +17,19 @@ jobs:
image: python:3.13-slim
steps:
- uses: actions/checkout@v4
- name: Install uv
run: |
pip install -q uv==${{ env.UV_VERSION }}
- name: Install dependencies
run: |
uv pip install --system -e .[dev]
- name: Run ruff format check
run: |
ruff format --check .
- name: Run ruff lint
run: |
ruff check .
@@ -40,19 +40,79 @@ jobs:
image: python:3.13-slim
steps:
- uses: actions/checkout@v4
- name: Install uv
run: |
pip install -q uv==${{ env.UV_VERSION }}
- name: Install dependencies
run: |
uv pip install --system -e .[dev]
- name: Run pyright
run: |
pyright
security:
runs-on: docker
container:
image: python:3.13-slim
steps:
- uses: actions/checkout@v4
- name: Install uv
run: |
pip install -q uv==${{ env.UV_VERSION }}
- name: Install dependencies
run: |
uv pip install --system -e .[dev]
- name: Run bandit security scan
run: |
bandit -c pyproject.toml -r src/cleveragents --severity-level medium --format json --output build/bandit-report.json || true
bandit -c pyproject.toml -r src/cleveragents --severity-level high
- name: Run vulture dead code detection
run: |
vulture src/cleveragents vulture_whitelist.py --min-confidence 80 --exclude src/cleveragents/discovery
- name: Upload security report
if: always()
uses: actions/upload-artifact@v4
with:
name: security-report
path: build/bandit-report.json
quality:
runs-on: docker
container:
image: python:3.13-slim
steps:
- uses: actions/checkout@v4
- name: Install uv
run: |
pip install -q uv==${{ env.UV_VERSION }}
- name: Install dependencies
run: |
uv pip install --system -e .[dev]
- name: Check code complexity
run: |
radon cc src/cleveragents --min C --show-complexity --total-average --json > build/complexity-report.json || true
radon cc src/cleveragents --min F --show-complexity
echo "Complexity check complete. Methods with F grade will fail the build."
radon cc src/cleveragents --min F --show-complexity --json | python -c "import sys, json; data = json.load(sys.stdin); blocks = [b for f in data.values() for b in f]; sys.exit(1 if blocks else 0)"
- name: Upload complexity report
if: always()
uses: actions/upload-artifact@v4
with:
name: complexity-report
path: build/complexity-report.json
behave:
runs-on: docker
strategy:
@@ -62,35 +122,68 @@ jobs:
image: python:${{ matrix.python-version }}-slim
steps:
- uses: actions/checkout@v4
- name: Install uv
run: |
pip install -q uv==${{ env.UV_VERSION }}
- name: Install dependencies
run: |
uv pip install --system -e .[dev]
- name: Run behaviour tests
run: |
behave -q
coverage:
runs-on: docker
container:
image: python:3.13-slim
needs: [lint, typecheck]
steps:
- uses: actions/checkout@v4
- name: Install uv
run: |
pip install -q uv==${{ env.UV_VERSION }}
- name: Install dependencies
run: |
uv pip install --system -e ".[dev,tests]"
- name: Install behave-parallel
run: |
pip install -q behave-parallel
- name: Run tests with coverage
run: |
coverage run --source=src -m behave -q --no-capture || true
coverage xml -o build/coverage.xml
coverage report --fail-under=85
- name: Upload coverage report
if: always()
uses: actions/upload-artifact@v4
with:
name: coverage-report
path: build/coverage.xml
build:
runs-on: docker
container:
image: python:3.13-slim
steps:
- uses: actions/checkout@v4
- name: Install uv
run: |
pip install -q uv==${{ env.UV_VERSION }}
- name: Build wheel
run: |
uv pip install --system build
python -m build --wheel
- name: Upload artifacts
uses: actions/upload-artifact@v4
with:
@@ -98,31 +191,31 @@ jobs:
path: dist/
docker:
needs: [lint, typecheck, behave]
needs: [lint, typecheck, behave, security]
runs-on: docker
steps:
- uses: actions/checkout@v4
- name: Build Docker image
run: |
docker build -t cleverernie:test .
- name: Test Docker image
run: |
docker run --rm cleverernie:test --version
helm:
needs: [lint, typecheck, behave]
needs: [lint, typecheck, behave, security]
runs-on: docker
container:
image: alpine/helm:latest
steps:
- uses: actions/checkout@v4
- name: Lint Helm chart
run: |
helm lint k8s/
- name: Template Helm chart
run: |
helm template test k8s/
helm template test k8s/
.forgejo/workflows/nightly-quality.yml
+114
View File
@@ -0,0 +1,114 @@
name: Nightly Quality
on:
schedule:
# Run at midnight UTC every day
- cron: "0 0 * * *"
workflow_dispatch:
# Allow manual trigger for testing
env:
UV_VERSION: "0.8.0"
PYTHON_VERSION: "3.13"
jobs:
full-quality-suite:
runs-on: docker
container:
image: python:3.13-slim
steps:
- uses: actions/checkout@v4
- name: Install uv
run: |
pip install -q uv==${{ env.UV_VERSION }}
- name: Install all dependencies
run: |
uv pip install --system -e ".[dev,tests]"
- name: Install behave-parallel
run: |
pip install -q behave-parallel
- name: Run full lint suite
run: |
ruff format --check .
ruff check .
- name: Run pyright type checking
run: |
pyright
- name: Run bandit security scan (all severities)
run: |
mkdir -p build/reports
bandit -c pyproject.toml -r src/cleveragents --format json --output build/reports/bandit-full.json || true
bandit -c pyproject.toml -r src/cleveragents --severity-level high
- name: Run vulture dead code detection
run: |
vulture src/cleveragents vulture_whitelist.py --min-confidence 80 --exclude src/cleveragents/discovery
- name: Run radon complexity analysis
run: |
mkdir -p build/reports
radon cc src/cleveragents --show-complexity --total-average --json > build/reports/complexity.json
radon mi src/cleveragents --json > build/reports/maintainability.json
echo "=== Complexity Summary ==="
radon cc src/cleveragents --min C --show-complexity --total-average
echo "=== Maintainability Index ==="
radon mi src/cleveragents --min B
- name: Run full Behave test suite with coverage
run: |
coverage run --source=src -m behave -q --no-capture || true
coverage xml -o build/reports/coverage.xml
coverage json -o build/reports/coverage.json
coverage report --fail-under=85
- name: Run quality gates script
run: |
python scripts/check-quality-gates.py --coverage-min 85 --complexity-max F
- name: Generate quality trend data
run: |
python -c "
import json, datetime
from pathlib import Path
report = {
'timestamp': datetime.datetime.now(datetime.timezone.utc).isoformat(),
'gates': {}
}
# Coverage
cov_path = Path('build/reports/coverage.json')
if cov_path.exists():
cov_data = json.loads(cov_path.read_text())
report['gates']['coverage'] = cov_data.get('totals', {}).get('percent_covered', 0)
# Complexity
cx_path = Path('build/reports/complexity.json')
if cx_path.exists():
cx_data = json.loads(cx_path.read_text())
total_blocks = sum(len(b) for b in cx_data.values())
report['gates']['total_analyzed_blocks'] = total_blocks
# Security
sec_path = Path('build/reports/bandit-full.json')
if sec_path.exists():
sec_data = json.loads(sec_path.read_text())
report['gates']['security_issues'] = len(sec_data.get('results', []))
Path('build/reports/quality-trend.json').write_text(json.dumps(report, indent=2))
print(json.dumps(report, indent=2))
"
- name: Upload quality reports
if: always()
uses: actions/upload-artifact@v4
with:
name: nightly-quality-reports
path: build/reports/
retention-days: 90
.pre-commit-config.yaml
+94
View File
@@ -0,0 +1,94 @@
# Pre-commit configuration for CleverAgents
# See https://pre-commit.com for more information
# Install: pre-commit install
# Run manually: pre-commit run --all-files
default_language_version:
python: python3.13
repos:
# ----- Branch Protection & General Checks -----
- repo: https://github.com/pre-commit/pre-commit-hooks
rev: v4.6.0
hooks:
- id: no-commit-to-branch
name: Prevent direct commits to main
args: ["--branch", "main"]
- id: check-yaml
name: Validate YAML syntax
- id: check-toml
name: Validate TOML syntax
- id: check-json
name: Validate JSON syntax
exclude: "^\\.vscode/"
- id: check-merge-conflict
name: Check for merge conflicts
- id: check-added-large-files
name: Prevent large file commits
args: ["--maxkb=500"]
- id: end-of-file-fixer
name: Fix end of file
- id: trailing-whitespace
name: Trim trailing whitespace
args: ["--markdown-linebreak-ext=md"]
- id: debug-statements
name: Detect debug statements (pdb, breakpoint)
# ----- Ruff: Formatting & Linting -----
- repo: https://github.com/astral-sh/ruff-pre-commit
rev: v0.8.6
hooks:
- id: ruff-format
name: Ruff format (auto-fix)
- id: ruff
name: Ruff lint (auto-fix safe rules)
args: ["--fix", "--exit-non-zero-on-fix"]
# ----- Bandit: Security Scanning -----
- repo: https://github.com/PyCQA/bandit
rev: 1.8.3
hooks:
- id: bandit
name: Bandit security scan
args: ["-c", "pyproject.toml", "-r", "--severity-level", "medium"]
additional_dependencies: ["bandit[toml]"]
files: "^src/"
# ----- Vulture: Dead Code Detection -----
- repo: https://github.com/jendrikseipp/vulture
rev: v2.14
hooks:
- id: vulture
name: Vulture dead code detection
args: ["src/cleveragents", "vulture_whitelist.py", "--min-confidence", "80", "--exclude", "src/cleveragents/discovery"]
pass_filenames: false
# ----- Local Hooks (use installed tools) -----
- repo: local
hooks:
# Pyright: Type Checking (runs serially, uses local installation)
- id: pyright
name: Pyright type checking
language: system
entry: pyright
types: [python]
files: "^src/"
pass_filenames: false
stages: [pre-commit]
# Semgrep: Advanced security scanning for eval/exec detection
- id: semgrep-eval-exec
name: Semgrep eval/exec detection
language: system
entry: scripts/run-semgrep.sh
types: [python]
files: "^src/"
pass_filenames: false
stages: [pre-commit]
# ----- Commit Message Linting -----
- repo: https://github.com/commitizen-tools/commitizen
rev: v4.2.2
hooks:
- id: commitizen
name: Conventional commit message check
stages: [commit-msg]
.semgrep.yml
+54
View File
@@ -0,0 +1,54 @@
rules:
- id: no-eval
pattern: eval(...)
message: >
Use of eval() is a security risk. Do not use eval() in production code.
Consider using ast.literal_eval() for safe evaluation of literals.
languages: [python]
severity: ERROR
paths:
include:
- src/
- id: no-exec
pattern: exec(...)
message: >
Use of exec() is a security risk. Do not use exec() in production code.
Find an alternative approach that does not require dynamic code execution.
languages: [python]
severity: ERROR
paths:
include:
- src/
- id: no-compile-exec
pattern: compile(..., ..., "exec")
message: >
Using compile() with exec mode is a security risk.
Avoid dynamic code compilation in production code.
languages: [python]
severity: ERROR
paths:
include:
- src/
- id: no-os-system
pattern: os.system(...)
message: >
Use of os.system() is insecure. Use subprocess.run() with shell=False instead.
languages: [python]
severity: WARNING
paths:
include:
- src/
- id: no-pickle-loads
pattern: pickle.loads(...)
message: >
Use of pickle.loads() on untrusted data is a security risk.
Consider using json or a safe serialization format.
languages: [python]
severity: WARNING
paths:
include:
- src/
README.md
+22 -7
View File
@@ -20,7 +20,10 @@ cd core
# install dependencies
python -m venv .venv
source .venv/bin/activate
pip install -e .[tests,docs]
pip install -e ".[dev,tests,docs]"
# set up pre-commit hooks and verify tooling
bash scripts/setup-dev.sh
# verify the CLI
agents --help
@@ -29,15 +32,27 @@ agents --version
## Developing
Pre-commit hooks run automatically on every `git commit` (formatting, linting, type
checking, security scanning). To run checks manually:
```bash
# run validation suites
nox -s format
nox -s lint
oxt -s typecheck
nox -s unit_tests
nox -s integration_tests
# core validation
nox -s format # ruff auto-formatting
nox -s lint # ruff linting
nox -s typecheck # pyright type checking
nox -s unit_tests # behave unit tests
nox -s integration_tests # robot integration tests
# quality & security
nox -s security_scan # bandit security scanning
nox -s dead_code # vulture dead code detection
nox -s complexity # radon complexity analysis
nox -s pre_commit # run all pre-commit hooks
nox -s adr_compliance # verify ADR compliance
```
For the full quality automation guide, see [`docs/development/quality-automation.md`](docs/development/quality-automation.md).
## Documentation
```bash
docs/development/quality-automation.md
+177
View File
@@ -0,0 +1,177 @@
# Quality Automation Guide
This document describes the automated quality gates, CI pipeline, and developer tooling
for the CleverAgents project.
## Quick Start
```bash
# Set up development environment (installs dependencies + pre-commit hooks)
bash scripts/setup-dev.sh
# Run all quality checks
nox
# Run specific checks
nox -s lint # Ruff linting
nox -s format # Ruff formatting
nox -s typecheck # Pyright type checking
nox -s unit_tests # Behave unit tests
nox -s coverage_report # Coverage (must be >85%)
nox -s security_scan # Bandit security scanning
nox -s dead_code # Vulture dead code detection
nox -s complexity # Radon complexity analysis
nox -s pre_commit # Run all pre-commit hooks
```
## Pre-commit Hooks
Pre-commit hooks run automatically on every `git commit`. They are configured in
`.pre-commit-config.yaml`.
### Installed Hooks
| Hook | Tool | Purpose | Auto-fix |
|------|------|---------|----------|
| `no-commit-to-branch` | pre-commit-hooks | Prevents commits to `main` | No |
| `check-yaml` | pre-commit-hooks | Validates YAML syntax | No |
| `check-toml` | pre-commit-hooks | Validates TOML syntax | No |
| `check-json` | pre-commit-hooks | Validates JSON syntax | No |
| `check-merge-conflict` | pre-commit-hooks | Detects merge conflict markers | No |
| `check-added-large-files` | pre-commit-hooks | Prevents files >500KB | No |
| `end-of-file-fixer` | pre-commit-hooks | Ensures files end with newline | Yes |
| `trailing-whitespace` | pre-commit-hooks | Removes trailing whitespace | Yes |
| `debug-statements` | pre-commit-hooks | Detects `pdb`/`breakpoint` | No |
| `ruff-format` | ruff | Code formatting | Yes |
| `ruff` | ruff | Linting with safe auto-fixes | Yes |
| `bandit` | bandit | Security scanning | No |
| `vulture` | vulture | Dead code detection | No |
| `pyright` | pyright | Type checking (src/ only) | No |
| `semgrep-eval-exec` | semgrep | eval/exec detection (optional) | No |
| `commitizen` | commitizen | Commit message format (commit-msg stage) | No |
### Running Hooks Manually
```bash
# Run all hooks on all files
pre-commit run --all-files
# Run a specific hook
pre-commit run ruff-format --all-files
pre-commit run bandit --all-files
pre-commit run pyright --all-files
# Skip hooks for emergency (not recommended)
git commit --no-verify -m "emergency: fix critical bug"
```
## CI Pipeline
The CI pipeline runs on **Forgejo Actions** (`.forgejo/workflows/ci.yml`).
### CI Jobs
| Job | Trigger | Purpose | Failure Impact |
|-----|---------|---------|----------------|
| `lint` | Push/PR | Ruff format + lint check | Blocks merge |
| `typecheck` | Push/PR | Pyright type checking | Blocks merge |
| `security` | Push/PR | Bandit + Vulture | Blocks merge |
| `quality` | Push/PR | Radon complexity check | Blocks merge (grade F) |
| `behave` | Push/PR | BDD tests (Python 3.11-3.13 matrix) | Blocks merge |
| `coverage` | Push/PR | Coverage measurement | Blocks merge (<85%) |
| `build` | Push/PR | Wheel build | Blocks release |
| `docker` | Push/PR | Docker image build + test | Blocks deployment |
| `helm` | Push/PR | Helm chart lint + template | Blocks deployment |
### Nightly Quality
A nightly workflow (`.forgejo/workflows/nightly-quality.yml`) runs at midnight UTC:
- Full lint, type check, security scan
- Complete Behave test suite with coverage
- Complexity and maintainability analysis
- Quality trend data generation
- Reports uploaded as artifacts (90-day retention)
## Security Scanning
### Bandit Configuration
Configured in `pyproject.toml` under `[tool.bandit]`:
- Targets: `src/cleveragents` only
- Severity threshold: MEDIUM (pre-commit), HIGH (CI fail gate)
- Excludes: tests, features, build directories
### Semgrep Rules
Custom rules in `.semgrep.yml`:
- `no-eval`: Blocks `eval()` usage
- `no-exec`: Blocks `exec()` usage
- `no-compile-exec`: Blocks `compile()` with exec mode
- `no-os-system`: Warns on `os.system()` (use `subprocess.run`)
- `no-pickle-loads`: Warns on `pickle.loads()` (use JSON)
**Note:** Semgrep is optional. Install with `pip install semgrep` for full scanning.
## Dead Code Detection
Vulture checks `src/cleveragents/` for unused code with 80% confidence threshold.
### False Positive Whitelist
Known false positives are listed in `vulture_whitelist.py`:
- `exc_tb`: Required by `__exit__` protocol but not used in body
- `build_data`: Required by mapping interface in legacy migrator
Add new entries when vulture flags intentionally unused code.
## Complexity Monitoring
Radon measures cyclomatic complexity:
| Grade | Complexity | Status |
|-------|-----------|--------|
| A | 1-5 | Excellent |
| B | 6-10 | Good |
| C | 11-15 | Moderate - review recommended |
| D | 16-20 | Complex - refactoring recommended |
| E | 21-30 | Very complex - must refactor |
| F | 31+ | Extremely complex - CI fails |
Current project average: **A (3.56)**
## Quality Gate Script
`scripts/check-quality-gates.py` aggregates all quality checks:
```bash
python scripts/check-quality-gates.py --coverage-min 85 --complexity-max F
```
Gates checked:
1. Coverage >= 85%
2. Pyright: 0 type errors
3. Bandit: 0 high-severity security issues
4. Vulture: 0 dead code findings
5. Radon: No grade-F complexity blocks
## Troubleshooting
### Pre-commit hooks not running
```bash
pre-commit install
pre-commit install --hook-type commit-msg
```
### Pyright errors on clean code
Check `pyrightconfig.json` for configuration. Pyright uses `strict` mode.
### Bandit false positive
If bandit flags intentionally safe code, consider if there's a safer alternative first.
Only as a last resort, add to `[tool.bandit]` `skips` in `pyproject.toml`.
### Vulture false positive
Add the symbol name to `vulture_whitelist.py` with a comment explaining why.
### Coverage below 85%
Run `nox -s coverage_report` to see which lines are uncovered, then add Behave scenarios.
features/environment.py
+1 -3
View File
@@ -94,10 +94,8 @@ def after_scenario(context, scenario):
# Run all cleanup handlers in reverse order
if hasattr(context, "_cleanup_handlers"):
for handler in reversed(context._cleanup_handlers):
try:
with contextlib.suppress(Exception):
handler()
except Exception:
pass # Ignore cleanup errors
context._cleanup_handlers = []
# Clean up any test directories
features/steps/actor_cli_run_steps.py
+7 -8
View File
@@ -52,22 +52,21 @@ def _make_context_manager(
@given("I have a saved context JSON file")
def step_saved_context_json(context):
context.load_context_data = {"global_context": {"session": "stored"}}
handle = tempfile.NamedTemporaryFile(
with tempfile.NamedTemporaryFile(
delete=False, suffix=".json", mode="w", encoding="utf-8"
)
json.dump(context.load_context_data, handle)
handle.flush()
handle.close()
) as handle:
json.dump(context.load_context_data, handle)
handle.flush()
context.load_context_path = Path(handle.name)
_register_cleanup(context, context.load_context_path)
@given("I have an actor output file path")
def step_actor_output_path(context):
handle = tempfile.NamedTemporaryFile(
with tempfile.NamedTemporaryFile(
delete=False, suffix=".txt", mode="w", encoding="utf-8"
)
handle.close()
) as handle:
pass
context.output_path = Path(handle.name)
_register_cleanup(context, context.output_path)
features/steps/actor_cli_steps.py
+35 -44
View File
@@ -64,12 +64,11 @@ def step_impl(context):
"temperature": 0.5,
"max_tokens": 256,
}
handle = tempfile.NamedTemporaryFile(
with tempfile.NamedTemporaryFile(
delete=False, suffix=".json", mode="w", encoding="utf-8"
)
json.dump(context.actor_config_data, handle)
handle.flush()
handle.close()
) as handle:
json.dump(context.actor_config_data, handle)
handle.flush()
context.actor_config_path = Path(handle.name)
_register_cleanup(context, context.actor_config_path)
@@ -81,12 +80,11 @@ def step_impl(context):
"model": "gpt-4o",
"options": {"temperature": 0.4, "enabled": True},
}
handle = tempfile.NamedTemporaryFile(
with tempfile.NamedTemporaryFile(
delete=False, suffix=".json", mode="w", encoding="utf-8"
)
json.dump(context.actor_config_data, handle)
handle.flush()
handle.close()
) as handle:
json.dump(context.actor_config_data, handle)
handle.flush()
context.actor_config_path = Path(handle.name)
_register_cleanup(context, context.actor_config_path)
@@ -94,12 +92,11 @@ def step_impl(context):
@given("I have an actor list config file")
def step_impl(context):
context.actor_config_data = [1, 2, 3]
handle = tempfile.NamedTemporaryFile(
with tempfile.NamedTemporaryFile(
delete=False, suffix=".json", mode="w", encoding="utf-8"
)
json.dump(context.actor_config_data, handle)
handle.flush()
handle.close()
) as handle:
json.dump(context.actor_config_data, handle)
handle.flush()
context.actor_config_path = Path(handle.name)
_register_cleanup(context, context.actor_config_path)
@@ -110,10 +107,9 @@ def step_impl(context):
"provider": "existing-provider",
"model": "existing-model",
}
handle = tempfile.NamedTemporaryFile(delete=False, suffix=".yaml")
handle.write(b"provider: existing-provider\nmodel: existing-model\n")
handle.flush()
handle.close()
with tempfile.NamedTemporaryFile(delete=False, suffix=".yaml") as handle:
handle.write(b"provider: existing-provider\nmodel: existing-model\n")
handle.flush()
context.actor_config_path = Path(handle.name)
_register_cleanup(context, context.actor_config_path)
@@ -125,12 +121,11 @@ def step_impl(context):
"model": "gpt-4o-mini",
"yaml_only": True,
}
handle = tempfile.NamedTemporaryFile(
with tempfile.NamedTemporaryFile(
delete=False, suffix=".yaml", mode="w", encoding="utf-8"
)
handle.write("provider: openai\nmodel: gpt-4o-mini\nyaml_only: true\n")
handle.flush()
handle.close()
) as handle:
handle.write("provider: openai\nmodel: gpt-4o-mini\nyaml_only: true\n")
handle.flush()
context.actor_config_path = Path(handle.name)
_register_cleanup(context, context.actor_config_path)
@@ -142,12 +137,11 @@ def step_impl(context):
"model": "graph-model",
"graph": {"nodes": ["start"], "edges": []},
}
handle = tempfile.NamedTemporaryFile(
with tempfile.NamedTemporaryFile(
delete=False, suffix=".json", mode="w", encoding="utf-8"
)
json.dump(context.actor_config_data, handle)
handle.flush()
handle.close()
) as handle:
json.dump(context.actor_config_data, handle)
handle.flush()
context.actor_config_path = Path(handle.name)
_register_cleanup(context, context.actor_config_path)
@@ -156,12 +150,11 @@ def step_impl(context):
def step_impl(context):
context.actor_config_data = {}
context.expected_error = "provider is required"
handle = tempfile.NamedTemporaryFile(
with tempfile.NamedTemporaryFile(
delete=False, suffix=".yaml", mode="w", encoding="utf-8"
)
handle.write("")
handle.flush()
handle.close()
) as handle:
handle.write("")
handle.flush()
context.actor_config_path = Path(handle.name)
_register_cleanup(context, context.actor_config_path)
@@ -173,12 +166,11 @@ def step_impl(context):
"model": "gpt-4o-mini",
"unsafe": True,
}
handle = tempfile.NamedTemporaryFile(
with tempfile.NamedTemporaryFile(
delete=False, suffix=".json", mode="w", encoding="utf-8"
)
json.dump(context.actor_config_data, handle)
handle.flush()
handle.close()
) as handle:
json.dump(context.actor_config_data, handle)
handle.flush()
context.actor_config_path = Path(handle.name)
_register_cleanup(context, context.actor_config_path)
@@ -521,12 +513,11 @@ def step_impl(context):
@when("I run actor add with business rule violation")
def step_impl(context):
config_data = {"provider": "openai", "model": "gpt-4o"}
handle = tempfile.NamedTemporaryFile(
with tempfile.NamedTemporaryFile(
delete=False, suffix=".json", mode="w", encoding="utf-8"
)
json.dump(config_data, handle)
handle.flush()
handle.close()
) as handle:
json.dump(config_data, handle)
handle.flush()
config_path = Path(handle.name)
_register_cleanup(context, config_path)
features/steps/architecture_steps.py
+12 -9
View File
@@ -26,9 +26,9 @@ def step_check_adr_files(context):
@then("I should find at least {count:d} ADR files")
def step_verify_adr_count(context, count):
"""Verify minimum number of ADR files."""
assert len(context.adr_files) >= count, (
f"Expected at least {count} ADR files, found {len(context.adr_files)}"
)
assert (
len(context.adr_files) >= count
), f"Expected at least {count} ADR files, found {len(context.adr_files)}"
@then("each ADR should have a valid format with status")
@@ -40,9 +40,9 @@ def step_verify_adr_format(context):
assert "## Status" in content, f"{adr_file.name} missing Status section"
assert "## Context" in content, f"{adr_file.name} missing Context section"
assert "## Decision" in content, f"{adr_file.name} missing Decision section"
assert "## Consequences" in content, (
f"{adr_file.name} missing Consequences section"
)
assert (
"## Consequences" in content
), f"{adr_file.name} missing Consequences section"
@given('the source directory exists at "{path}"')
@@ -141,9 +141,12 @@ def step_analyze_imports(context):
for alias in node.names:
if alias.name.startswith("cleveragents."):
package_imports.add(alias.name.split(".")[1])
elif isinstance(node, ast.ImportFrom):
if node.module and node.module.startswith("cleveragents."):
package_imports.add(node.module.split(".")[1])
elif (
isinstance(node, ast.ImportFrom)
and node.module
and node.module.startswith("cleveragents.")
):
package_imports.add(node.module.split(".")[1])
except SyntaxError:
pass # Skip files with syntax errors
context.imports[package_name] = package_imports
features/steps/auto_debug_cli_coverage_steps.py
+7 -7
View File
@@ -202,9 +202,9 @@ def step_assert_attempts(context, 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 (
exception is not None
), "Expected CleverAgentsError but no exception was recorded"
assert isinstance(exception, CleverAgentsError)
assert message in str(exception)
@@ -307,9 +307,9 @@ 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}"
)
assert (
actual == code
), f"Expected exit code {code}, got {actual}. CLI output:\n{output}"
@then("the command should be aborted")
@@ -332,7 +332,7 @@ def step_build_fail_then_succeed_with_changes(context, num_failures, changes):
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]
context.plan_service_mock.build_plan.side_effect = [*failures, success_result]
@given("the build will raise PlanError after one attempt")
features/steps/cli_plan_context_commands_steps.py
+8 -9
View File
@@ -1,5 +1,6 @@
"""Step definitions for CLI plan and context command testing."""
import contextlib
import json
import os
import shlex
@@ -67,7 +68,7 @@ def step_initialized_project(context):
# Ensure an actor is available for mock provider flows
container = get_container()
actor_service = container.actor_service()
try:
with contextlib.suppress(Exception):
actor_service.upsert_actor(
name="local/mock-default",
provider="MockProvider",
@@ -76,8 +77,6 @@ def step_initialized_project(context):
is_built_in=True,
unsafe=True,
)
except Exception:
pass
@given('I have a file "{filename}" with content "{content}"')
@@ -376,9 +375,9 @@ def step_file_in_context(context):
@then('I should see "{text}" in the output')
def step_see_text_in_output(context, text):
"""Verify text appears in command output."""
assert text in context.command_output, (
f"Expected '{text}' in output: {context.command_output}"
)
assert (
text in context.command_output
), f"Expected '{text}' in output: {context.command_output}"
@then("a new plan should be created")
@@ -518,9 +517,9 @@ def step_error_mentions(context, text):
"""Verify error message contains text."""
# Check both stdout and stderr for error messages
combined_output = (context.command_output or "") + (context.command_error or "")
assert text.lower() in combined_output.lower(), (
f"Expected '{text}' not found in: {combined_output}"
)
assert (
text.lower() in combined_output.lower()
), f"Expected '{text}' not found in: {combined_output}"
@then('the output should mention "{text}"')
features/steps/container_and_repository_coverage_steps.py
+1 -1
View File
@@ -747,7 +747,7 @@ def _make_actor(
model_name: str | None = None,
):
blob = config_blob or {"name": name}
provider_value, model_value = (name.split("/", 1) + [""])[:2]
provider_value, model_value = ([*name.split("/", 1), ""])[:2]
return Actor(
id=None,
name=name,
features/steps/context_analysis_agent_coverage_steps.py
+1 -1
View File
@@ -71,7 +71,7 @@ def _ensure_temp_dir(context: Any) -> Path:
@given("the context analysis agent module is importable")
def step_module_importable(context: Any) -> None:
ContextAnalysisAgent # noqa: B018 (import verification)
ContextAnalysisState
ContextAnalysisState # noqa: B018 (import verification)
@given("I have a mock LLM provider configured for context analysis")
features/steps/enums_coverage_steps.py
+26 -26
View File
@@ -180,9 +180,9 @@ def step_convert_model_provider_to_string(context, member):
@then("it should be a string type")
def step_verify_string_type(context):
"""Verify the value is a string."""
assert isinstance(context.string_value, str), (
f"Not a string: {type(context.string_value)}"
)
assert isinstance(
context.string_value, str
), f"Not a string: {type(context.string_value)}"
@then('it should equal "{expected}"')
@@ -213,8 +213,8 @@ def step_verify_valid_enum_value(context, value, enum_name):
member = enum_class(value)
assert member.value == value
context.membership_checks[value] = True
except ValueError:
raise AssertionError(f"{value} is not a valid {enum_name} value")
except ValueError as exc:
raise AssertionError(f"{value} is not a valid {enum_name} value") from exc
@then('"{value}" should not be a valid {enum_name} value')
@@ -240,18 +240,18 @@ def step_iterate_enum_values(context, enum_name):
@then("I should get {count:d} enum members")
def step_verify_enum_count(context, count):
"""Verify the number of enum members."""
assert context.member_count == count, (
f"Expected {count} members, got {context.member_count}"
)
assert (
context.member_count == count
), f"Expected {count} members, got {context.member_count}"
@then("each member should be a {enum_name} instance")
def step_verify_enum_instances(context, enum_name):
"""Verify each member is an instance of the enum."""
for member in context.enum_members:
assert isinstance(member, context.current_enum), (
f"{member} is not a {enum_name} instance"
)
assert isinstance(
member, context.current_enum
), f"{member} is not a {enum_name} instance"
# Enum comparison tests
@@ -296,9 +296,9 @@ def step_access_enum_name(context, member):
@then('the name should be "{expected}"')
def step_verify_enum_name(context, expected):
"""Verify the enum name attribute."""
assert context.enum_name == expected, (
f"Expected name {expected}, got {context.enum_name}"
)
assert (
context.enum_name == expected
), f"Expected name {expected}, got {context.enum_name}"
@when("I access the value attribute of ModelPublisher.{member}")
@@ -311,9 +311,9 @@ def step_access_enum_value(context, member):
@then('the value should be "{expected}"')
def step_verify_enum_value_attr(context, expected):
"""Verify the enum value attribute."""
assert context.enum_value_attr == expected, (
f"Expected value {expected}, got {context.enum_value_attr}"
)
assert (
context.enum_value_attr == expected
), f"Expected value {expected}, got {context.enum_value_attr}"
# Creating enum from value
@@ -334,9 +334,9 @@ def step_verify_created_enum(context, member):
"""Verify the created enum member."""
expected = getattr(ModelProvider, member)
assert context.creation_successful, "Enum creation failed"
assert context.created_enum == expected, (
f"Expected {expected}, got {context.created_enum}"
)
assert (
context.created_enum == expected
), f"Expected {expected}, got {context.created_enum}"
assert context.created_enum is expected, "Enum instances should be identical"
@@ -395,15 +395,15 @@ def step_get_string_representation(context):
@then('repr should contain "{expected}"')
def step_verify_repr_contains(context, expected):
"""Verify repr contains expected string."""
assert expected in context.repr_value, (
f"'{expected}' not in repr: {context.repr_value}"
)
assert (
expected in context.repr_value
), f"'{expected}' not in repr: {context.repr_value}"
@then('str should equal "{expected}"')
def step_verify_str_equals(context, expected):
"""Verify str equals expected."""
# For str, Enum classes, str() returns the value, not the name
assert context.enum_value.value == expected, (
f"Expected str '{expected}', got '{context.enum_value.value}'"
)
assert (
context.enum_value.value == expected
), f"Expected str '{expected}', got '{context.enum_value.value}'"
features/steps/google_provider_steps.py
+7 -8
View File
@@ -68,8 +68,7 @@ def _setup_plan_generation_graph(context) -> MagicMock:
events_copy = list(stream_events)
def _stream(*_args, **_kwargs):
for event in events_copy:
yield event
yield from events_copy
mock_graph_instance.stream.side_effect = _stream
@@ -161,9 +160,9 @@ def step_assert_google_kwargs(context, kwargs_string):
expected = _parse_kwargs_string(kwargs_string)
for key, value in expected.items():
assert key in call_kwargs, f"Expected {key} in ChatGoogleGenerativeAI kwargs"
assert call_kwargs[key] == value, (
f"Expected ChatGoogleGenerativeAI kwargs[{key!r}] to equal {value!r}"
)
assert (
call_kwargs[key] == value
), f"Expected ChatGoogleGenerativeAI kwargs[{key!r}] to equal {value!r}"
@then(
@@ -185,9 +184,9 @@ def step_assert_google_metadata(context, expected_name, expected_model):
def step_assert_google_generated_change(context, count, file_path):
response = getattr(context, "response", None)
assert response is not None, "Provider response should exist"
assert len(response.changes) == count, (
f"Expected {count} changes, got {len(response.changes)}"
)
assert (
len(response.changes) == count
), f"Expected {count} changes, got {len(response.changes)}"
assert any(
getattr(change, "file_path", None) == file_path for change in response.changes
), f"Expected change for {file_path}"
features/steps/langchain_chat_provider_steps.py
+16 -17
View File
@@ -448,8 +448,7 @@ def step_provider_stream_failure_with_unknown_nodes(context):
context.llm_instance.get_num_tokens.return_value = None
def _failing_stream():
for event in streaming_events:
yield event
yield from streaming_events
raise RuntimeError(failure_message)
with patch(
@@ -644,9 +643,9 @@ def step_assert_response_contains_validation_error(context):
assert context.response is not None
assert context.response.changes, "Expected generated changes to be returned"
assert context.response.changes[0].file_path == context.expected_change.file_path
assert context.response.error_message == context.expected_validation_message, (
"Validation failure message should be surfaced"
)
assert (
context.response.error_message == context.expected_validation_message
), "Validation failure message should be surfaced"
assert context.response.model_used == "test-model"
assert context.response.token_count == 0
assert context.requested_models == ["test-model"]
@@ -805,23 +804,23 @@ def step_assert_graph_failure_response(context):
@then("the provider response should capture the nested retry failure")
def step_assert_nested_retry_failure(context):
assert context.response is not None, "Expected provider response to be populated"
assert context.response.changes == [], (
f"Expected no changes for nested failure, got {context.response.changes!r}"
)
assert context.response.error_message == context.failure_message, (
f"Expected error message {context.failure_message!r}, got {context.response.error_message!r}"
)
assert (
context.response.changes == []
), f"Expected no changes for nested failure, got {context.response.changes!r}"
assert (
context.response.error_message == context.failure_message
), f"Expected error message {context.failure_message!r}, got {context.response.error_message!r}"
@then("the progress callback should end at 100 percent even on failure")
def step_assert_failure_progress_updates(context):
assert context.progress_updates, "Expected progress updates to be recorded"
assert context.progress_updates[0] == 5, (
f"Expected first progress update to be 5, got {context.progress_updates[0]!r}"
)
assert context.progress_updates[-1] == 100, (
f"Expected final progress update to be 100, got {context.progress_updates[-1]!r}"
)
assert (
context.progress_updates[0] == 5
), f"Expected first progress update to be 5, got {context.progress_updates[0]!r}"
assert (
context.progress_updates[-1] == 100
), f"Expected final progress update to be 100, got {context.progress_updates[-1]!r}"
@then("the stream API should preserve non-dict payloads in events")
features/steps/openai_provider_steps.py
+10 -11
View File
@@ -74,8 +74,7 @@ def _setup_plan_generation_graph(context) -> MagicMock:
events_copy = list(stream_events)
def _stream(*_args, **_kwargs):
for event in events_copy:
yield event
yield from events_copy
mock_graph_instance.stream.side_effect = _stream
@@ -235,9 +234,9 @@ def step_assert_chat_openai_constructor(context, api_key, model_id):
assert call_kwargs["api_key"] == api_key
assert call_kwargs["model"] == model_id
assert context.plan_generation_graph_call is not None, (
"PlanGenerationGraph should receive the ChatOpenAI instance"
)
assert (
context.plan_generation_graph_call is not None
), "PlanGenerationGraph should receive the ChatOpenAI instance"
_, graph_kwargs = context.plan_generation_graph_call
assert graph_kwargs["llm"] is context.chat_openai_instance
@@ -263,9 +262,9 @@ def step_assert_chat_openai_kwargs(context, kwargs_string):
expected_kwargs = _parse_kwargs_string(kwargs_string)
for key, value in expected_kwargs.items():
assert key in call_kwargs, f"Expected {key} in ChatOpenAI kwargs"
assert call_kwargs[key] == value, (
f"Expected ChatOpenAI kwargs[{key!r}] to equal {value!r}"
)
assert (
call_kwargs[key] == value
), f"Expected ChatOpenAI kwargs[{key!r}] to equal {value!r}"
@then(
@@ -299,9 +298,9 @@ def step_assert_no_changes(context):
)
def step_assert_generated_change(context, count, file_path):
assert context.response is not None, "Provider response should exist"
assert len(context.response.changes) == count, (
f"Expected {count} changes, got {len(context.response.changes)}"
)
assert (
len(context.response.changes) == count
), f"Expected {count} changes, got {len(context.response.changes)}"
assert any(
getattr(change, "file_path", None) == file_path
for change in context.response.changes
features/steps/openrouter_provider_steps.py
+10 -11
View File
@@ -84,8 +84,7 @@ def _setup_plan_generation_graph(context) -> MagicMock:
events_copy = list(stream_events)
def _stream(*_args, **_kwargs):
for event in events_copy:
yield event
yield from events_copy
mock_graph_instance.stream.side_effect = _stream
@@ -211,9 +210,9 @@ def step_assert_openrouter_kwargs(context, kwargs_string):
expected = _parse_kwargs_string(kwargs_string)
for key, value in expected.items():
assert key in call_kwargs, f"Expected {key} in ChatOpenAI kwargs"
assert call_kwargs[key] == value, (
f"Expected ChatOpenAI kwargs[{key!r}] to equal {value!r}"
)
assert (
call_kwargs[key] == value
), f"Expected ChatOpenAI kwargs[{key!r}] to equal {value!r}"
@then(
@@ -225,9 +224,9 @@ def step_assert_openrouter_headers(context, headers_string):
_, call_kwargs = call
expected = _parse_headers_string(headers_string)
actual_headers = call_kwargs.get("default_headers")
assert actual_headers == expected, (
f"Expected default_headers {expected!r}, got {actual_headers!r}"
)
assert (
actual_headers == expected
), f"Expected default_headers {expected!r}, got {actual_headers!r}"
@then(
@@ -236,9 +235,9 @@ def step_assert_openrouter_headers(context, headers_string):
def step_assert_openrouter_generated_change(context, count, file_path):
response = getattr(context, "response", None)
assert response is not None, "Provider response should exist"
assert len(response.changes) == count, (
f"Expected {count} changes, got {len(response.changes)}"
)
assert (
len(response.changes) == count
), f"Expected {count} changes, got {len(response.changes)}"
assert any(
getattr(change, "file_path", None) == file_path for change in response.changes
), f"Expected change for {file_path}"
features/steps/plan_commands_uncovered_branches_steps.py
+2 -1
View File
@@ -14,7 +14,8 @@ from rich.console import Console
from typer.testing import CliRunner
from cleveragents.application.services.plan_service import PlanService
from cleveragents.cli.commands.plan import _tell_streaming, app as plan_app
from cleveragents.cli.commands.plan import _tell_streaming
from cleveragents.cli.commands.plan import app as plan_app
class _StreamingPlanService:
features/steps/plan_full_coverage_steps.py
+21 -19
View File
@@ -328,18 +328,20 @@ def step_run_plan_apply_confirm(context):
env = os.environ.copy()
env["CLEVERAGENTS_TEST_FORCE_CONFIRMATION"] = "true"
with patch("cleveragents.application.container.get_container") as mock_container:
with patch("cleveragents.cli.commands.plan.console", test_console):
mock_plan_service = MagicMock(spec=PlanService)
mock_plan_service.get_pending_changes.return_value = context.pending_changes
mock_plan_service.apply_changes.return_value = len(context.pending_changes)
mock_container.return_value.plan_service.return_value = mock_plan_service
with (
patch("cleveragents.application.container.get_container") as mock_container,
patch("cleveragents.cli.commands.plan.console", test_console),
):
mock_plan_service = MagicMock(spec=PlanService)
mock_plan_service.get_pending_changes.return_value = context.pending_changes
mock_plan_service.apply_changes.return_value = len(context.pending_changes)
mock_container.return_value.plan_service.return_value = mock_plan_service
result = runner.invoke(plan_app, ["apply"], input="y\n", env=env)
result = runner.invoke(plan_app, ["apply"], input="y\n", env=env)
# Store both outputs for testing
context.result = result
context.console_output = string_buffer.getvalue()
# Store both outputs for testing
context.result = result
context.console_output = string_buffer.getvalue()
@when("I execute plan apply with auto-confirm")
@@ -1126,9 +1128,9 @@ def step_assert_typer_abort(context):
@then("the plan tell should execute successfully")
def step_plan_tell_success(context):
"""Assert plan tell executed successfully."""
assert context.result.exit_code == 0, (
f"Exit code was {context.result.exit_code}, output: {context.result.output}"
)
assert (
context.result.exit_code == 0
), f"Exit code was {context.result.exit_code}, output: {context.result.output}"
assert "Plan created" in context.result.output or "✓" in context.result.output
@@ -1174,12 +1176,12 @@ def step_plan_tell_general_abort(context):
@then("the plan tell streaming path should execute successfully")
def step_plan_tell_streaming_success(context):
"""Assert plan tell streaming path executed via asyncio."""
assert context.result.exit_code == 0, (
f"Exit code was {context.result.exit_code}, output: {context.result.output}"
)
assert getattr(context, "stream_call_count", 0) == 1, (
"Streaming coroutine was not awaited"
)
assert (
context.result.exit_code == 0
), f"Exit code was {context.result.exit_code}, output: {context.result.output}"
assert (
getattr(context, "stream_call_count", 0) == 1
), "Streaming coroutine was not awaited"
assert context.stream_call_args, "No streaming call arguments recorded"
first_call = context.stream_call_args[0]
assert first_call.args[1] == context.stream_prompt
features/steps/plan_service_steps.py
+100 -88
View File
@@ -643,11 +643,13 @@ def step_stub_provider_registry_lazy(context: Context) -> None:
# Restore mock flag after scenario
cleanup_handlers.append(
lambda: os.environ.__setitem__(
"CLEVERAGENTS_TESTING_USE_MOCK_AI", original_mock_flag
lambda: (
os.environ.__setitem__(
"CLEVERAGENTS_TESTING_USE_MOCK_AI", original_mock_flag
)
if original_mock_flag is not None
else os.environ.pop("CLEVERAGENTS_TESTING_USE_MOCK_AI", None)
)
if original_mock_flag is not None
else os.environ.pop("CLEVERAGENTS_TESTING_USE_MOCK_AI", None)
)
context.lazy_registry = lazy_registry
@@ -1044,9 +1046,9 @@ def step_assert_lazy_registry_actor(
registry = getattr(context, "lazy_registry", None)
assert registry is not None, "Lazy registry stub was not configured"
expected = {"provider": provider.lower(), "model": model.lower()}
assert registry.requests == [expected], (
"Lazy registry did not capture the actor request"
)
assert registry.requests == [
expected
], "Lazy registry did not capture the actor request"
resolution = getattr(context, "provider_resolution", None)
assert resolution is not None, "Provider resolution result missing"
@@ -1073,11 +1075,13 @@ def step_provider_registry_raises_for_actor(context: Context, message: str) -> N
cleanup_handlers = []
context._cleanup_handlers = cleanup_handlers
cleanup_handlers.append(
lambda: os.environ.__setitem__(
"CLEVERAGENTS_TESTING_USE_MOCK_AI", original_mock_flag
lambda: (
os.environ.__setitem__(
"CLEVERAGENTS_TESTING_USE_MOCK_AI", original_mock_flag
)
if original_mock_flag is not None
else os.environ.pop("CLEVERAGENTS_TESTING_USE_MOCK_AI", None)
)
if original_mock_flag is not None
else os.environ.pop("CLEVERAGENTS_TESTING_USE_MOCK_AI", None)
)
context.plan_service._provider_registry = _Registry()
@@ -1112,11 +1116,13 @@ def step_provider_registry_factory_raises_for_actor(
context._cleanup_handlers = cleanup_handlers
cleanup_handlers.append(patcher.stop)
cleanup_handlers.append(
lambda: os.environ.__setitem__(
"CLEVERAGENTS_TESTING_USE_MOCK_AI", original_mock_flag
lambda: (
os.environ.__setitem__(
"CLEVERAGENTS_TESTING_USE_MOCK_AI", original_mock_flag
)
if original_mock_flag is not None
else os.environ.pop("CLEVERAGENTS_TESTING_USE_MOCK_AI", None)
)
if original_mock_flag is not None
else os.environ.pop("CLEVERAGENTS_TESTING_USE_MOCK_AI", None)
)
context.plan_service.ai_provider = None
@@ -1139,11 +1145,13 @@ def step_provider_registry_returns_provider(
cleanup_handlers = []
context._cleanup_handlers = cleanup_handlers
cleanup_handlers.append(
lambda: os.environ.__setitem__(
"CLEVERAGENTS_TESTING_USE_MOCK_AI", original_mock_flag
lambda: (
os.environ.__setitem__(
"CLEVERAGENTS_TESTING_USE_MOCK_AI", original_mock_flag
)
if original_mock_flag is not None
else os.environ.pop("CLEVERAGENTS_TESTING_USE_MOCK_AI", None)
)
if original_mock_flag is not None
else os.environ.pop("CLEVERAGENTS_TESTING_USE_MOCK_AI", None)
)
class _Provider:
@@ -1306,9 +1314,9 @@ def step_assert_stream_token_count(context: Context, expected: int) -> None:
with context.unit_of_work.transaction() as ctx:
plan = ctx.plans.get_current_for_project(context.test_project.id)
assert plan is not None, "Plan was not persisted"
assert plan.token_count == expected, (
f"Expected token count {expected}, got {plan.token_count}"
)
assert (
plan.token_count == expected
), f"Expected token count {expected}, got {plan.token_count}"
@then('the streaming failure events should include an error with message "{message}"')
@@ -1325,9 +1333,9 @@ def step_assert_stream_failure_event(context: Context, message: str) -> None:
failure_payload = payload
break
assert failure_payload is not None, "Expected a failure event in the stream"
assert failure_payload.get("error") == message, (
f"Expected failure message '{message}', got '{failure_payload.get('error')}'"
)
assert (
failure_payload.get("error") == message
), f"Expected failure message '{message}', got '{failure_payload.get('error')}'"
@then(
@@ -1339,9 +1347,9 @@ def step_assert_non_python_stream(context: Context) -> None:
events = getattr(context, "stream_events", [])
assert events, "Streaming events were not captured"
end_event = events[-1]
assert isinstance(end_event, dict) and "__end__" in end_event, (
"Final streaming event did not include end payload"
)
assert (
isinstance(end_event, dict) and "__end__" in end_event
), "Final streaming event did not include end payload"
payload = end_event["__end__"]
changes = payload.get("changes") or []
assert len(changes) == 1, "Expected exactly one streamed change"
@@ -1387,12 +1395,12 @@ def step_assert_stub_provider_requests(
requests = getattr(context, "stub_provider_requests", None)
assert requests, "No provider override requests were recorded"
last_request = requests[-1]
assert last_request["provider"] == provider, (
f"Expected provider '{provider}', got '{last_request['provider']}'"
)
assert last_request["model"] == model, (
f"Expected model '{model}', got '{last_request['model']}'"
)
assert (
last_request["provider"] == provider
), f"Expected provider '{provider}', got '{last_request['provider']}'"
assert (
last_request["model"] == model
), f"Expected model '{model}', got '{last_request['model']}'"
@given("I have a plan service without providers configured")
@@ -1405,11 +1413,13 @@ def step_plan_service_without_providers(context: Context) -> None:
cleanup_handlers = []
context._cleanup_handlers = cleanup_handlers
cleanup_handlers.append(
lambda: os.environ.__setitem__(
"CLEVERAGENTS_TESTING_USE_MOCK_AI", original_mock_flag
lambda: (
os.environ.__setitem__(
"CLEVERAGENTS_TESTING_USE_MOCK_AI", original_mock_flag
)
if original_mock_flag is not None
else os.environ.pop("CLEVERAGENTS_TESTING_USE_MOCK_AI", None)
)
if original_mock_flag is not None
else os.environ.pop("CLEVERAGENTS_TESTING_USE_MOCK_AI", None)
)
if not hasattr(context, "unit_of_work"):
@@ -1536,9 +1546,9 @@ def step_assert_validation_hint(context: Context) -> None:
@given('the project has plans named "{first}" and "{second}"')
def step_project_with_named_plans(context: Context, first: str, second: str) -> None:
"""Ensure the project has specific plan names configured."""
assert context.project.id is not None, (
"Project must be persisted before adding plans"
)
assert (
context.project.id is not None
), "Project must be persisted before adding plans"
with context.unit_of_work.transaction() as ctx:
current_plan = ctx.plans.get_current_for_project(context.project.id)
assert current_plan is not None, "Current plan is required for this step"
@@ -1708,9 +1718,9 @@ def step_try_build_plan_no_current(context: Context) -> None:
def step_check_plan_error(context: Context, message: str) -> None:
"""Check that a PlanError was raised with specific message."""
assert context.exception is not None, "No exception captured"
assert isinstance(context.exception, PlanError), (
f"Unexpected exception type {type(context.exception).__name__}"
)
assert isinstance(
context.exception, PlanError
), f"Unexpected exception type {type(context.exception).__name__}"
actual_message = str(getattr(context.exception, "message", "")) or str(
context.exception
)
@@ -1729,9 +1739,9 @@ def step_check_plan_error_contains(context: Context, message: str) -> None:
def step_plan_error_details_include_provider_diagnostics(context: Context) -> None:
"""Ensure provider diagnostics are present in the captured PlanError details."""
assert context.exception is not None, "No exception captured"
assert isinstance(context.exception, PlanError), (
f"Expected PlanError but got {type(context.exception).__name__}"
)
assert isinstance(
context.exception, PlanError
), f"Expected PlanError but got {type(context.exception).__name__}"
details = getattr(context.exception, "details", {}) or {}
required_keys = {
"configured_providers",
@@ -1746,9 +1756,9 @@ def step_plan_error_details_include_provider_diagnostics(context: Context) -> No
@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"
)
assert hasattr(
context, "temp_dir"
), "Temporary directory setup is required before configuring auto-debug"
class FlickeringPlan:
def __init__(self, plan_id: int) -> None:
@@ -2296,18 +2306,18 @@ def step_check_nested_move(context: Context) -> None:
old_path, new_path = context.nested_move_paths
assert not old_path.exists(), "Original nested file still exists"
assert new_path.exists(), "Nested destination file was not created"
assert new_path.read_text() == "# Nested file to move", (
"Nested destination content mismatch"
)
assert (
new_path.read_text() == "# Nested file to move"
), "Nested destination content mismatch"
@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"
)
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")
@@ -2430,12 +2440,12 @@ def step_check_plan_error_with_details(context: Context) -> None:
with contextlib.suppress(builtins.BaseException):
os.chmod(context.readonly_file, 0o644)
assert context.exception is not None, (
f"Expected an exception but got result: {getattr(context, 'apply_result', 'none')}"
)
assert isinstance(context.exception, PlanError), (
f"Expected PlanError but got {type(context.exception).__name__}: {context.exception}"
)
assert (
context.exception is not None
), f"Expected an exception but got result: {getattr(context, 'apply_result', 'none')}"
assert isinstance(
context.exception, PlanError
), f"Expected PlanError but got {type(context.exception).__name__}: {context.exception}"
assert "Failed to apply change" in str(context.exception.message)
@@ -3285,9 +3295,9 @@ def step_assert_langsmith_builder_base_metadata(
assert metadata.get("project_name") == project_name
assert metadata.get("project_id") is None
extra_metadata_keys = set(metadata.keys()) - {"project_id", "project_name"}
assert not extra_metadata_keys, (
f"Unexpected metadata keys present: {sorted(extra_metadata_keys)}"
)
assert (
not extra_metadata_keys
), f"Unexpected metadata keys present: {sorted(extra_metadata_keys)}"
assert tags == ["service:plan"], f"Expected only base service tag, got {tags}"
@@ -3374,16 +3384,18 @@ def step_run_auto_debug_with_attempts(context: Context, attempts: int) -> None:
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
)
with (
patch.object(PlanService, "build_plan", side_effect=failing_build),
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
@@ -3426,16 +3438,16 @@ def step_run_auto_debug_retry_success(context: Context) -> None:
]
agent_path = "cleveragents.agents.graphs.auto_debug.AutoDebugAgent"
with patch.object(PlanService, "build_plan", side_effect=build_plan_side_effect):
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=2
)
with (
patch.object(PlanService, "build_plan", side_effect=build_plan_side_effect),
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=2
)
with context.unit_of_work.transaction() as ctx:
context.latest_debug_attempts = ctx.debug_attempts.get_for_plan(current_plan.id)
@@ -3465,9 +3477,9 @@ def step_assert_auto_debug_retry_success(context: Context) -> None:
attempts = getattr(context, "latest_debug_attempts", []) or []
assert attempts, "No debug attempts were recorded"
assert any(attempt.success for attempt in attempts), (
"Debug attempt was not marked successful"
)
assert any(
attempt.success for attempt in attempts
), "Debug attempt was not marked successful"
@when("I try to stream plan generation without an AI provider")
features/steps/project_commands_coverage_steps.py
+4 -4
View File
@@ -529,7 +529,7 @@ def step_run_file_filter_show_no_project(context):
def step_add_file_filters_with_defaults(context, include_glob, exclude_glob):
"""Run the file-filter add command with include, exclude, and defaults."""
runner = CliRunner()
expected_excludes = [exclude_glob] + list(DEFAULT_IGNORE_PATTERNS)
expected_excludes = [exclude_glob, *DEFAULT_IGNORE_PATTERNS]
with patch("cleveragents.application.container.get_container") as mock_container:
mock_service = MagicMock()
mock_service.get_current_project.return_value = context.mock_project
@@ -673,6 +673,6 @@ def step_project_command_should_succeed(context):
@then("the project command should exit with code {code:d}")
def step_project_command_exit_code(context, code):
"""Assert the project command exited with the expected code."""
assert context.result.exit_code == code, (
f"Expected exit code {code}, got {context.result.exit_code}. Output: {context.result.output}"
)
assert (
context.result.exit_code == code
), f"Expected exit code {code}, got {context.result.exit_code}. Output: {context.result.output}"
features/steps/reactive_registry_adapter_steps.py
+2 -2
View File
@@ -29,8 +29,8 @@ def step_registry_mixed(context):
def __init__(self, name: str, sync: bool = False, async_capable: bool = False):
self.name = name
if sync:
self.process_message_sync = (
lambda content, metadata: f"{name}:{content}"
self.process_message_sync = lambda content, metadata: (
f"{name}:{content}"
)
if async_capable:
self.process_message = lambda content, metadata: f"{name}:{content}"
features/steps/retry_patterns_steps.py
+25 -25
View File
@@ -121,27 +121,27 @@ def step_apply_async_exponential_retry(context, max_attempts):
@then("the function should eventually succeed")
def step_verify_function_succeeded(context):
"""Verify function succeeded after retries."""
assert context.retry_succeeded, (
f"Function failed: {getattr(context, 'retry_error', 'Unknown error')}"
)
assert (
context.retry_succeeded
), f"Function failed: {getattr(context, 'retry_error', 'Unknown error')}"
assert context.result == "success"
@then("the async function should eventually succeed")
def step_verify_async_function_succeeded(context):
"""Verify async function succeeded after retries."""
assert context.retry_succeeded, (
f"Async function failed: {getattr(context, 'retry_error', 'Unknown error')}"
)
assert (
context.retry_succeeded
), f"Async function failed: {getattr(context, 'retry_error', 'Unknown error')}"
assert context.result == "async success"
@then("the function should be called {expected_calls:d} times")
def step_verify_call_count(context, expected_calls):
"""Verify function was called expected number of times."""
assert context.call_count == expected_calls, (
f"Expected {expected_calls} calls, got {context.call_count}"
)
assert (
context.call_count == expected_calls
), f"Expected {expected_calls} calls, got {context.call_count}"
# Network retry pattern tests
@@ -182,9 +182,9 @@ def step_verify_network_retry(context):
@then("the function should be called {max_calls:d} times maximum")
def step_verify_max_calls(context, max_calls):
"""Verify function wasn't called more than max times."""
assert context.call_count <= max_calls, (
f"Function called {context.call_count} times, max is {max_calls}"
)
assert (
context.call_count <= max_calls
), f"Function called {context.call_count} times, max is {max_calls}"
# Provider retry pattern tests
@@ -261,18 +261,18 @@ def step_apply_file_retry(context):
@then("the file operation should eventually succeed")
def step_verify_file_retry_success(context):
"""Verify the file operation eventually succeeds."""
assert context.file_retry_succeeded, (
f"File retry failed: {getattr(context, 'file_retry_error', 'Unknown error')}"
)
assert (
context.file_retry_succeeded
), f"File retry failed: {getattr(context, 'file_retry_error', 'Unknown error')}"
assert context.file_result == "file success"
@then("the file operation should be invoked {expected:d} times")
def step_verify_file_retry_invocations(context, expected):
"""Verify the file operation was invoked the expected number of times."""
assert context.file_call_count == expected, (
f"Expected {expected} calls, got {context.file_call_count}"
)
assert (
context.file_call_count == expected
), f"Expected {expected} calls, got {context.file_call_count}"
# Circuit breaker tests
@@ -327,7 +327,7 @@ def step_verify_immediate_failure(context):
# Unexpected exception
raise AssertionError(
f"Expected CircuitBreakerOpen but got: {type(e).__name__}: {e}"
)
) from e
@given("I have an open circuit breaker")
@@ -496,9 +496,9 @@ def step_async_retry_context_manager_failure(context):
@then("the async retry context manager should capture the exception")
def step_verify_async_retry_context_manager_capture(context):
"""Ensure the async retry context manager captured the exception."""
assert context.async_retry_context_manager_error, (
"Expected async runtime error to propagate"
)
assert (
context.async_retry_context_manager_error
), "Expected async runtime error to propagate"
assert context.retry_context_errors_added_async >= 1
assert isinstance(context.retry_context.errors[-1], Exception)
@@ -690,9 +690,9 @@ def step_apply_retry_on_result(context, max_attempts):
@then("the decorated function should return the successful payload")
def step_verify_retry_on_result_payload(context):
"""Ensure retry_on_result eventually returns the non-retry payload."""
assert context.retry_on_result_succeeded, (
f"Retry on result failed: {context.retry_on_result_output}"
)
assert (
context.retry_on_result_succeeded
), f"Retry on result failed: {context.retry_on_result_output}"
assert isinstance(context.retry_on_result_output, dict)
assert context.retry_on_result_output.get("retry") is False
features/steps/routing_langgraph_port_steps.py
+1 -1
View File
@@ -429,7 +429,7 @@ def route_with_vision(context):
# Find the conditional node
router_node = None
for node_name, node_data in context.route_config.nodes.items():
for _node_name, node_data in context.route_config.nodes.items():
if node_data["type"] == "conditional":
router_node = node_data
break
features/steps/stream_router_uncovered_paths_steps.py
+2
View File
@@ -2,6 +2,8 @@
Step definitions for uncovered stream_router paths.
"""
from typing import Any
import rx
from behave import given, then, when
from behave.runner import Context
features/steps/vector_store_service_steps.py
+6 -6
View File
@@ -8,7 +8,7 @@ import tempfile
import types
from collections.abc import Iterable
from pathlib import Path
from typing import Any
from typing import Any, ClassVar
from unittest.mock import patch
from behave import given, then, when
@@ -61,11 +61,11 @@ class _StubDocument:
class RecordingFAISS:
"""Test double that records FAISS usage."""
last_from_texts_args: dict[str, Any] | None = None
last_built_instance: RecordingFAISS | None = None
load_local_should_raise: bool = False
load_local_calls: list[str] = []
default_similarity_payload: list[tuple[_StubDocument, float]] = []
last_from_texts_args: ClassVar[dict[str, Any] | None] = None
last_built_instance: ClassVar[RecordingFAISS | None] = None
load_local_should_raise: ClassVar[bool] = False
load_local_calls: ClassVar[list[str]] = []
default_similarity_payload: ClassVar[list[tuple[_StubDocument, float]]] = []
def __init__(self) -> None:
self.documents: list[str] = []
implementation_plan.md
+617 -473
View File
File diff suppressed because it is too large Load Diff
noxfile.py
+90 -16
View File
@@ -80,9 +80,18 @@ from typing import Dict, Iterable, List, Tuple
DEFAULT_FEATURE_ROOT = "features/"
SUMMARY_PATTERNS = {
"features": re.compile(r"(\\d+)\\s+feature(?:s)?\\s+passed,\\s+(\\d+)\\s+failed(?:,\\s+(\\d+)\\s+error(?:s)?)?,\\s+(\\d+)\\s+skipped"),
"scenarios": re.compile(r"(\\d+)\\s+scenario(?:s)?\\s+passed,\\s+(\\d+)\\s+failed(?:,\\s+(\\d+)\\s+error(?:s)?)?,\\s+(\\d+)\\s+skipped"),
"steps": re.compile(r"(\\d+)\\s+step(?:s)?\\s+passed,\\s+(\\d+)\\s+failed(?:,\\s+(\\d+)\\s+error(?:s)?)?,\\s+(\\d+)\\s+skipped"),
"features": re.compile(
r"(\\d+)\\s+feature(?:s)?\\s+passed,\\s+(\\d+)\\s+failed"
r"(?:,\\s+(\\d+)\\s+error(?:s)?)?,\\s+(\\d+)\\s+skipped"
),
"scenarios": re.compile(
r"(\\d+)\\s+scenario(?:s)?\\s+passed,\\s+(\\d+)\\s+failed"
r"(?:,\\s+(\\d+)\\s+error(?:s)?)?,\\s+(\\d+)\\s+skipped"
),
"steps": re.compile(
r"(\\d+)\\s+step(?:s)?\\s+passed,\\s+(\\d+)\\s+failed"
r"(?:,\\s+(\\d+)\\s+error(?:s)?)?,\\s+(\\d+)\\s+skipped"
),
}
DURATION_PATTERN = re.compile(r"Took\\s+(?:(\\d+)m\\s*)?([\\d\\.]+)s")
@@ -129,7 +138,9 @@ def _parse_summary(text: str) -> Dict[str, Dict[str, int] | float]:
def _merge_summaries(summaries: List[Dict[str, Dict[str, int] | float]]) -> Dict[str, Dict[str, int] | float]:
def _merge_summaries(
summaries: List[Dict[str, Dict[str, int] | float]],
) -> Dict[str, Dict[str, int] | float]:
total = _empty_summary()
for summary in summaries:
for bucket in ("features", "scenarios", "steps"):
@@ -149,21 +160,20 @@ def _format_duration(seconds: float) -> str:
def _print_overall_summary(total: Dict[str, Dict[str, int] | float]) -> None:
print("\\nOverall summary:")
print(
f"{total['features']['passed']} features passed, {total['features']['failed']} failed, {total['features']['errors']} errored, {total['features']['skipped']} skipped"
)
print(
f"{total['scenarios']['passed']} scenarios passed, {total['scenarios']['failed']} failed, {total['scenarios']['errors']} errored, {total['scenarios']['skipped']} skipped"
)
print(
f"{total['steps']['passed']} steps passed, {total['steps']['failed']} failed, {total['steps']['errors']} errored, {total['steps']['skipped']} skipped"
)
for bucket in ("features", "scenarios", "steps"):
b = total[bucket]
print(
f"{b['passed']} {bucket} passed, {b['failed']} failed, "
f"{b['errors']} errored, {b['skipped']} skipped"
)
if total["duration"]:
print(f"Took {_format_duration(float(total['duration']))} (sum across workers)")
def _run_feature(base_args: List[str], feature: str) -> Tuple[int, str, str, str, Dict[str, Dict[str, int] | float]]:
def _run_feature(
base_args: List[str], feature: str
) -> Tuple[int, str, str, str, Dict[str, Dict[str, int] | float]]:
proc = subprocess.run([*base_args, feature], capture_output=True, text=True)
combined_output = (proc.stdout or "") + "\\n" + (proc.stderr or "")
summary = _parse_summary(combined_output)
@@ -204,7 +214,9 @@ def main(argv: List[str] | None = None) -> None:
feature_args, other_args = _extract_features_and_args(remaining)
if any(flag in remaining for flag in ("-h", "--help", "--version")):
code = subprocess.run([sys.executable, "-m", "behave", *other_args, *feature_args]).returncode
code = subprocess.run(
[sys.executable, "-m", "behave", *other_args, *feature_args]
).returncode
sys.exit(code)
feature_paths = _iter_features(feature_args)
@@ -217,7 +229,10 @@ def main(argv: List[str] | None = None) -> None:
summaries: List[Dict[str, Dict[str, int] | float]] = []
failures: List[Tuple[int, str]] = []
with concurrent.futures.ProcessPoolExecutor(max_workers=processes) as executor:
futures = [executor.submit(_run_feature, base_args, feature) for feature in feature_paths]
futures = [
executor.submit(_run_feature, base_args, feature)
for feature in feature_paths
]
for fut in concurrent.futures.as_completed(futures):
code, feature, stdout, stderr, summary = fut.result()
if stdout:
@@ -575,6 +590,65 @@ def discovery_integration(session: nox.Session):
)
@nox.session(python=DEFAULT_PYTHON, reuse_venv=True, venv_backend="uv")
def pre_commit(session: nox.Session):
"""Run all pre-commit hooks on all files."""
session.install("-e", ".[dev]")
session.run("pre-commit", "run", "--all-files", *session.posargs)
@nox.session(python=DEFAULT_PYTHON, reuse_venv=True, venv_backend="uv")
def security_scan(session: nox.Session):
"""Run security scanning tools (bandit) on production code."""
session.install("-e", ".[dev]")
session.run(
"bandit",
"-c",
"pyproject.toml",
"-r",
"src/cleveragents",
"--severity-level",
"medium",
)
@nox.session(python=DEFAULT_PYTHON, reuse_venv=True, venv_backend="uv")
def dead_code(session: nox.Session):
"""Run vulture to detect dead code."""
session.install("-e", ".[dev]")
session.run(
"vulture",
"src/cleveragents",
"vulture_whitelist.py",
"--min-confidence",
"80",
"--exclude",
"src/cleveragents/discovery",
)
@nox.session(python=DEFAULT_PYTHON, reuse_venv=True, venv_backend="uv")
def complexity(session: nox.Session):
"""Check code complexity using radon."""
session.install("-e", ".[dev]")
session.run(
"radon",
"cc",
"src/cleveragents",
"--min",
"C",
"--show-complexity",
"--total-average",
)
@nox.session(python=DEFAULT_PYTHON, reuse_venv=True, venv_backend="uv")
def adr_compliance(session: nox.Session):
"""Check code compliance with Architecture Decision Records."""
session.install("-e", ".[dev]")
session.run("python", "scripts/check-adr-compliance.py")
@nox.session(python=DEFAULT_PYTHON, reuse_venv=True, venv_backend="uv")
def benchmark(session: nox.Session):
"""Run Airspeed Velocity benchmarks and publish results."""
pyproject.toml
+36 -1
View File
@@ -56,6 +56,14 @@ dev = [
"pytest>=8.0.0",
"pytest-asyncio>=0.23.0",
"pytest-cov>=4.1.0",
# Pre-commit hooks
"pre-commit>=3.6.0",
# Security scanning
"bandit[toml]>=1.7.5",
# Dead code detection
"vulture>=2.10",
# Complexity metrics
"radon>=6.0.1",
]
tests = [
"coverage>=7.11.0",
@@ -88,7 +96,7 @@ include = ["src/cleveragents/py.typed"]
line-length = 88
target-version = "py313" # Target Python 3.13
src = ["src", "tests", "benchmarks"]
extend-exclude = ["docs/reference/contracts/stubs/*.py"]
extend-exclude = ["docs/reference/contracts/stubs/*.py", "scripts/*.sh"]
[tool.ruff.lint]
select = ["E", "F", "W", "B", "UP", "I", "SIM", "RUF"]
@@ -96,6 +104,10 @@ ignore = []
[tool.ruff.lint.per-file-ignores]
"__init__.py" = ["F401"]
# Behave step files: F811 = redefined step_impl (Behave pattern), E501 = long step decorator strings
"features/steps/*.py" = ["F811", "E501"]
"features/mocks/*.py" = ["E501"]
"features/environment.py" = ["E501"]
[tool.ruff.format]
# Use double quotes for strings
@@ -166,5 +178,28 @@ directory = "build/htmlcov"
[tool.coverage.xml]
output = "build/coverage.xml"
[tool.bandit]
targets = ["src/cleveragents"]
exclude_dirs = [
"tests",
"features",
".nox",
"build",
"dist",
"docs",
"src/cleveragents/discovery",
]
# Severity level: LOW, MEDIUM, HIGH
severity = "MEDIUM"
# Confidence level: LOW, MEDIUM, HIGH
confidence = "MEDIUM"
# Skip specific tests if needed
skips = []
[tool.vulture]
min_confidence = 80
paths = ["src/cleveragents"]
exclude = ["src/cleveragents/discovery"]
[tool.hatch]
# Hatch build configuration
scripts/check-adr-compliance.py
+140
View File
@@ -0,0 +1,140 @@
#!/usr/bin/env python3
"""ADR compliance checker for CleverAgents.
Verifies that code follows the architectural decisions recorded in ADRs.
ADR-002: Asyncio Concurrency Model
- Async functions should use 'async def', not threading
- No direct thread usage in application layer
ADR-003: Dependency Injection Framework
- Services should receive dependencies via __init__
- No direct instantiation of infrastructure in application layer
ADR-007: Repository Pattern for Persistence
- Database access only through repository classes
- No direct SQLAlchemy session usage in services
Usage:
python scripts/check-adr-compliance.py
"""
import ast
import sys
from pathlib import Path
def check_adr002_async_model(src_dir: Path) -> list[str]:
"""Check ADR-002: Asyncio Concurrency Model.
Verifies no direct threading usage in application code.
"""
violations: list[str] = []
app_dir = src_dir / "application"
if not app_dir.exists():
return violations
for py_file in app_dir.rglob("*.py"):
content = py_file.read_text()
if "import threading" in content or "from threading import" in content:
violations.append(
f"ADR-002 violation: {py_file.relative_to(src_dir)} "
"imports threading directly. Use asyncio instead."
)
return violations
def check_adr003_dependency_injection(src_dir: Path) -> list[str]:
"""Check ADR-003: Dependency Injection Framework.
Verifies services use constructor injection.
"""
violations: list[str] = []
services_dir = src_dir / "application" / "services"
if not services_dir.exists():
return violations
for py_file in services_dir.rglob("*.py"):
if py_file.name.startswith("__"):
continue
try:
tree = ast.parse(py_file.read_text())
except SyntaxError:
continue
for node in ast.walk(tree):
if isinstance(node, ast.ClassDef):
# Check if class has __init__ with parameters (DI)
has_init = False
for item in node.body:
if isinstance(item, ast.FunctionDef) and item.name == "__init__":
has_init = True
# Check for at least one parameter beyond self
if len(item.args.args) < 2:
violations.append(
f"ADR-003 note: {py_file.relative_to(src_dir)}:"
f"{node.lineno} class {node.name} has __init__ "
"with no injected dependencies"
)
if not has_init and node.body:
# Services without __init__ may be okay if they're utility classes
pass
return violations
def check_adr007_repository_pattern(src_dir: Path) -> list[str]:
"""Check ADR-007: Repository Pattern for Persistence.
Verifies no direct SQLAlchemy session usage in service layer.
"""
violations: list[str] = []
services_dir = src_dir / "application" / "services"
if not services_dir.exists():
return violations
for py_file in services_dir.rglob("*.py"):
content = py_file.read_text()
# Check for direct session imports
if "from sqlalchemy" in content:
violations.append(
f"ADR-007 violation: {py_file.relative_to(src_dir)} "
"imports SQLAlchemy directly. Use repository pattern instead."
)
if "session.query" in content or "session.execute" in content:
violations.append(
f"ADR-007 violation: {py_file.relative_to(src_dir)} "
"uses session directly. Access data through repositories."
)
return violations
def main() -> int:
"""Run all ADR compliance checks."""
src_dir = Path("src/cleveragents")
if not src_dir.exists():
print("Error: src/cleveragents directory not found")
return 1
all_violations: list[str] = []
print("Checking ADR-002: Asyncio Concurrency Model...")
all_violations.extend(check_adr002_async_model(src_dir))
print("Checking ADR-003: Dependency Injection Framework...")
all_violations.extend(check_adr003_dependency_injection(src_dir))
print("Checking ADR-007: Repository Pattern for Persistence...")
all_violations.extend(check_adr007_repository_pattern(src_dir))
if all_violations:
print(f"\nFound {len(all_violations)} ADR compliance issue(s):")
for violation in all_violations:
print(f" - {violation}")
return 1
print("\nAll ADR compliance checks passed.")
return 0
if __name__ == "__main__":
sys.exit(main())
scripts/check-quality-gates.py
+192
View File
@@ -0,0 +1,192 @@
#!/usr/bin/env python3
"""Quality gate checker for CleverAgents CI pipeline.
Aggregates results from multiple quality tools and produces a summary
report. Returns non-zero exit code if any critical gate fails.
Usage:
python scripts/check-quality-gates.py [--coverage-min 85] [--complexity-max 10]
Gates checked:
1. Coverage >= threshold (default 85%)
2. No type errors (pyright clean)
3. No high-severity security issues (bandit)
4. No dead code (vulture clean)
5. No extreme complexity (radon grade F)
"""
import argparse
import json
import subprocess
import sys
from pathlib import Path
def check_coverage(min_coverage: int) -> tuple[bool, str]:
"""Check test coverage meets minimum threshold."""
result = subprocess.run(
["coverage", "report", "--format=total"],
capture_output=True,
text=True,
check=False,
)
if result.returncode != 0:
return False, f"Coverage command failed: {result.stderr.strip()}"
try:
total = float(result.stdout.strip())
except ValueError:
return False, f"Could not parse coverage output: {result.stdout.strip()}"
passed = total >= min_coverage
msg = f"Coverage: {total:.1f}% (minimum: {min_coverage}%)"
return passed, msg
def check_typecheck() -> tuple[bool, str]:
"""Check pyright type checking passes."""
result = subprocess.run(
["pyright", "--outputjson"],
capture_output=True,
text=True,
check=False,
)
try:
data = json.loads(result.stdout)
error_count = data.get("summary", {}).get("errorCount", -1)
if error_count == 0:
return True, "Type checking: 0 errors"
return False, f"Type checking: {error_count} errors"
except (json.JSONDecodeError, KeyError):
# Fallback to exit code
if result.returncode == 0:
return True, "Type checking: passed"
return False, f"Type checking: failed (exit code {result.returncode})"
def check_security() -> tuple[bool, str]:
"""Check bandit finds no high-severity issues."""
result = subprocess.run(
[
"bandit",
"-c",
"pyproject.toml",
"-r",
"src/cleveragents",
"--severity-level",
"high",
"--format",
"json",
],
capture_output=True,
text=True,
check=False,
)
try:
data = json.loads(result.stdout)
high_issues = len(data.get("results", []))
if high_issues == 0:
return True, "Security: 0 high-severity issues"
return False, f"Security: {high_issues} high-severity issues"
except json.JSONDecodeError:
if result.returncode == 0:
return True, "Security: passed"
return False, f"Security: failed (exit code {result.returncode})"
def check_dead_code() -> tuple[bool, str]:
"""Check vulture finds no dead code."""
result = subprocess.run(
[
"vulture",
"src/cleveragents",
"vulture_whitelist.py",
"--min-confidence",
"80",
"--exclude",
"src/cleveragents/discovery",
],
capture_output=True,
text=True,
check=False,
)
if result.returncode == 0:
return True, "Dead code: none detected"
lines = result.stdout.strip().split("\n")
return False, f"Dead code: {len(lines)} issue(s) found"
def check_complexity(max_grade: str = "E") -> tuple[bool, str]:
"""Check no functions exceed maximum complexity grade."""
result = subprocess.run(
[
"radon",
"cc",
"src/cleveragents",
"--min",
max_grade,
"--show-complexity",
"--json",
],
capture_output=True,
text=True,
check=False,
)
try:
data = json.loads(result.stdout)
total_blocks = sum(len(blocks) for blocks in data.values())
if total_blocks == 0:
return True, f"Complexity: no blocks with grade >= {max_grade}"
return False, f"Complexity: {total_blocks} block(s) with grade >= {max_grade}"
except json.JSONDecodeError:
return True, "Complexity: passed (no extreme complexity)"
def main() -> int:
"""Run all quality gates and report results."""
parser = argparse.ArgumentParser(description="Check quality gates")
parser.add_argument(
"--coverage-min", type=int, default=85, help="Minimum coverage percentage"
)
parser.add_argument(
"--complexity-max",
type=str,
default="F",
help="Maximum allowed complexity grade (F = fail only on worst)",
)
args = parser.parse_args()
# Ensure build directory exists
Path("build").mkdir(exist_ok=True)
gates: list[tuple[str, tuple[bool, str]]] = [
("Coverage", check_coverage(args.coverage_min)),
("Type Check", check_typecheck()),
("Security", check_security()),
("Dead Code", check_dead_code()),
("Complexity", check_complexity(args.complexity_max)),
]
print("=" * 60)
print("QUALITY GATE REPORT")
print("=" * 60)
all_passed = True
for name, (passed, message) in gates:
icon = "[+]" if passed else "[X]"
print(f" {icon} {name}: {message}")
if not passed:
all_passed = False
print("=" * 60)
if all_passed:
print("All quality gates PASSED")
else:
print("Some quality gates FAILED")
print("=" * 60)
return 0 if all_passed else 1
if __name__ == "__main__":
sys.exit(main())
scripts/run-semgrep.sh
Executable
+7
View File
@@ -0,0 +1,7 @@
#!/bin/sh
# Run semgrep if installed, skip gracefully if not
if command -v semgrep >/dev/null 2>&1; then
semgrep --config=.semgrep.yml --error --quiet src/
else
echo "semgrep not installed, skipping (install with: pip install semgrep)"
fi
scripts/setup-dev.sh
Executable
+87
View File
@@ -0,0 +1,87 @@
#!/usr/bin/env bash
# Developer setup script for CleverAgents
# Usage: bash scripts/setup-dev.sh
#
# This script sets up the development environment including:
# - Installing all dependencies (dev, tests, docs extras)
# - Installing pre-commit hooks
# - Verifying tool installations
# - Running an initial quality check
set -euo pipefail
BOLD='\033[1m'
GREEN='\033[0;32m'
YELLOW='\033[0;33m'
RED='\033[0;31m'
NC='\033[0m' # No Color
info() { echo -e "${BOLD}${GREEN}[INFO]${NC} $*"; }
warn() { echo -e "${BOLD}${YELLOW}[WARN]${NC} $*"; }
error() { echo -e "${BOLD}${RED}[ERROR]${NC} $*"; }
# Check Python version
info "Checking Python version..."
PYTHON_VERSION=$(python3 --version 2>&1 | grep -oP '\d+\.\d+')
REQUIRED_VERSION="3.13"
if [ "$(printf '%s\n' "$REQUIRED_VERSION" "$PYTHON_VERSION" | sort -V | head -n1)" != "$REQUIRED_VERSION" ]; then
error "Python >= $REQUIRED_VERSION is required, found $PYTHON_VERSION"
exit 1
fi
info "Python $PYTHON_VERSION detected"
# Install the project with all development extras
info "Installing project with dev, tests, and docs extras..."
pip install -e ".[dev,tests,docs]" --quiet
# Install pre-commit hooks
info "Installing pre-commit hooks..."
pre-commit install
pre-commit install --hook-type commit-msg
info "Pre-commit hooks installed (pre-commit and commit-msg stages)"
# Verify key tools are available
info "Verifying tool installations..."
TOOLS_OK=true
for tool in ruff pyright bandit vulture radon pre-commit behave; do
if command -v "$tool" > /dev/null 2>&1; then
VERSION=$("$tool" --version 2>&1 | head -1)
info " $tool: $VERSION"
else
warn " $tool: NOT FOUND (check PATH or installation)"
TOOLS_OK=false
fi
done
if [ "$TOOLS_OK" = false ]; then
warn "Some tools were not found. They may be installed in a location not on your PATH."
warn "Try: export PATH=\"\$HOME/.local/bin:\$PATH\""
fi
# Run a quick quality check
info "Running initial quality check..."
echo ""
info "--- Ruff Format Check ---"
ruff format --check . 2>&1 || warn "Some files need formatting. Run: ruff format ."
echo ""
info "--- Ruff Lint Check ---"
ruff check . 2>&1 || warn "Some lint issues found. Run: ruff check --fix ."
echo ""
info "--- Pyright Type Check ---"
pyright 2>&1 || warn "Type checking found issues."
echo ""
info "Development environment setup complete."
info ""
info "Available commands:"
info " nox - Run all quality checks and tests"
info " nox -s unit_tests - Run Behave unit tests"
info " nox -s integration_tests - Run Robot integration tests"
info " nox -s coverage_report - Run tests with coverage"
info " nox -s lint - Run linter"
info " nox -s format - Format code"
info " nox -s typecheck - Run type checker"
info " pre-commit run --all-files - Run all pre-commit hooks"
src/cleveragents/actor/yaml_template_engine.py
+2 -2
View File
@@ -8,7 +8,7 @@ from pathlib import Path
from typing import Any, cast
import yaml
from jinja2 import Environment
from jinja2.sandbox import SandboxedEnvironment
logger = logging.getLogger(__name__)
@@ -18,7 +18,7 @@ class YAMLTemplateEngine:
def __init__(self) -> None:
# Configure Jinja2 for YAML-friendly output
self.env = Environment(
self.env = SandboxedEnvironment(
block_start_string="{%",
block_end_string="%}",
variable_start_string="{{",
src/cleveragents/application/services/context_service.py
+2 -1
View File
@@ -919,7 +919,8 @@ class ContextService:
return []
service = self._vector_store_service
assert service is not None
if service is None:
return []
try:
return service.search(
src/cleveragents/application/services/memory_service.py
+12 -6
View File
@@ -12,7 +12,7 @@ contexts, changes) across conversation sessions.
from collections.abc import Callable, Sequence
from datetime import UTC, datetime
from enum import Enum
from enum import StrEnum
from typing import Any, cast
from langchain_community.chat_message_histories import SQLChatMessageHistory
@@ -24,7 +24,7 @@ from langchain_core.messages import AIMessage, BaseMessage, HumanMessage
from pydantic import BaseModel, Field
class EntityType(str, Enum):
class EntityType(StrEnum):
"""Types of entities that can be tracked in memory."""
PROJECT = "project"
@@ -340,10 +340,13 @@ class ConversationBufferMemoryAdapter:
if isinstance(value, BaseMessage):
self._message_history.add_message(value)
return
if isinstance(value, (str, bytes, bytearray)):
if isinstance(value, str | bytes | bytearray):
self._message_history.add_user_message(str(value))
return
assert isinstance(value, Sequence)
if not isinstance(value, Sequence):
raise TypeError(
f"Expected Sequence for user message, got {type(value).__name__}"
)
for item in value:
_store_user(item)
@@ -353,10 +356,13 @@ class ConversationBufferMemoryAdapter:
if isinstance(value, BaseMessage):
self._message_history.add_message(value)
return
if isinstance(value, (str, bytes, bytearray)):
if isinstance(value, str | bytes | bytearray):
self._message_history.add_ai_message(str(value))
return
assert isinstance(value, Sequence)
if not isinstance(value, Sequence):
raise TypeError(
f"Expected Sequence for AI message, got {type(value).__name__}"
)
for item in value:
_store_ai(item)
src/cleveragents/application/services/plan_service.py
+1 -1
View File
@@ -1309,7 +1309,7 @@ class PlanService:
return {
"changes": [],
"model_used": "",
"token_count": 0,
"token_count": 0, # nosec B105
"error_message": str(payload),
}
src/cleveragents/cli/commands/context.py
+2 -1
View File
@@ -672,7 +672,8 @@ def context_delete(
typer.echo(f"Deleted {count} context(s).")
else:
# Delete a single context by name
assert name is not None
if name is None:
raise typer.BadParameter("Context name is required for deletion")
target = ctx_base / name
if not target.exists():
src/cleveragents/config/settings.py
+1 -1
View File
@@ -87,7 +87,7 @@ class Settings(BaseSettings):
validation_alias=AliasChoices("CLEVERAGENTS_ENV"),
)
server_host: str = Field(
default="0.0.0.0",
default="0.0.0.0", # nosec B104
validation_alias=AliasChoices("CLEVERAGENTS_SERVER_HOST"),
)
server_port: int = Field(
src/cleveragents/core/retry_patterns.py
+4 -2
View File
@@ -464,7 +464,8 @@ class RetryContext:
with attempt:
self.attempt_count = attempt.retry_state.attempt_number
result = func(*args, **kwargs)
assert result is not None, "Retrying must execute at least once"
if result is None:
raise RuntimeError("Retrying must execute at least once")
return result
async def async_execute(
@@ -480,7 +481,8 @@ class RetryContext:
with attempt:
self.attempt_count = attempt.retry_state.attempt_number
result = cast(T, await func(*args, **kwargs)) # type: ignore[misc]
assert result is not None, "AsyncRetrying must execute at least once"
if result is None:
raise RuntimeError("AsyncRetrying must execute at least once")
return result
src/cleveragents/domain/models/aimodelscommon/__init__.py
+2 -2
View File
@@ -1,11 +1,11 @@
"""Shared types for AI model domain objects."""
from enum import Enum
from enum import Enum, StrEnum
__all__ = ["SchemaUrl"]
class SchemaUrl(str, Enum):
class SchemaUrl(StrEnum):
"""Schema URL enumeration shared across AI model contracts."""
INPUT_CONFIG = "SchemaUrlInputConfig"
src/cleveragents/domain/models/aimodelsdatamodels/__init__.py
+3 -3
View File
@@ -3,7 +3,7 @@
from __future__ import annotations
from datetime import datetime
from enum import Enum
from enum import Enum, StrEnum
from typing import Any
from pydantic import BaseModel, ConfigDict, Field
@@ -19,14 +19,14 @@ def _empty_model_uses_provider_list() -> list[BaseModelUsesProvider]:
return []
class ModelOutputFormat(str, Enum):
class ModelOutputFormat(StrEnum):
"""Enum for ModelOutputFormat."""
TOOL_CALL_JSON = "ModelOutputFormatToolCallJson"
XML = "ModelOutputFormatXml"
class ReasoningEffort(str, Enum):
class ReasoningEffort(StrEnum):
"""Enum for ReasoningEffort."""
LOW = "ReasoningEffortLow"
src/cleveragents/domain/models/auth/auth.py
+2 -2
View File
@@ -1,11 +1,11 @@
"""Domain models for CleverAgents - auto-generated from Phase 0 stubs."""
from enum import Enum
from enum import StrEnum
from pydantic import BaseModel, ConfigDict, Field
class ApiErrorType(str, Enum):
class ApiErrorType(StrEnum):
"""Enum for API error types."""
VALIDATION = "validation"
src/cleveragents/domain/models/conversation/__init__.py
+3 -3
View File
@@ -7,7 +7,7 @@ These models represent the conversation state and history for plans.
from __future__ import annotations
from datetime import datetime
from enum import Enum
from enum import Enum, StrEnum
from pydantic import BaseModel, ConfigDict, Field
@@ -49,14 +49,14 @@ __all__ = [
]
class TellStage(str, Enum):
class TellStage(StrEnum):
"""Stage of the tell operation."""
PLANNING = "planning"
IMPLEMENTATION = "implementation"
class PlanningPhase(str, Enum):
class PlanningPhase(StrEnum):
"""Phase of the planning stage."""
CONTEXT = "context"
src/cleveragents/domain/models/core/action.py
+4 -4
View File
@@ -12,7 +12,7 @@ Based on v3_spec.md and ADR-004 (Pydantic Validation).
"""
from datetime import datetime
from enum import Enum
from enum import StrEnum
from pydantic import BaseModel, ConfigDict, Field, field_validator
@@ -23,7 +23,7 @@ from cleveragents.domain.models.core.plan import (
)
class ArgumentType(str, Enum):
class ArgumentType(StrEnum):
"""Supported types for action arguments."""
STRING = "str"
@@ -33,7 +33,7 @@ class ArgumentType(str, Enum):
LIST = "list"
class ArgumentRequirement(str, Enum):
class ArgumentRequirement(StrEnum):
"""Whether an argument is required or optional."""
REQUIRED = "required"
@@ -304,7 +304,7 @@ class Action(BaseModel):
elif arg.max_value is not None and value > arg.max_value:
errors.append(f"Argument {arg.name} must be <= {arg.max_value}")
elif arg.arg_type == ArgumentType.FLOAT:
if not isinstance(value, (int, float)):
if not isinstance(value, int | float):
errors.append(
f"Argument {arg.name} must be a number, got {type_name}"
)
src/cleveragents/domain/models/core/change.py
+2 -2
View File
@@ -4,12 +4,12 @@ Based on Phase 0 discovery and ADR-004 (Pydantic Validation).
"""
from datetime import datetime
from enum import Enum
from enum import StrEnum
from pydantic import BaseModel, ConfigDict, Field
class OperationType(str, Enum):
class OperationType(StrEnum):
"""Type of file operation."""
CREATE = "create"
src/cleveragents/domain/models/core/context.py
+3 -3
View File
@@ -6,7 +6,7 @@ Based on Phase 0 discovery and ADR-004 (Pydantic Validation).
from __future__ import annotations
from datetime import datetime
from enum import Enum
from enum import StrEnum
from pathlib import Path
from pydantic import BaseModel, ConfigDict, Field, field_validator
@@ -20,7 +20,7 @@ def _empty_token_diffs() -> dict[str, dict[str, int]]:
return {}
class ContextType(str, Enum):
class ContextType(StrEnum):
"""Type of context item."""
FILE = "file"
@@ -76,7 +76,7 @@ class Context(BaseModel):
)
class MaxContextCount(str, Enum):
class MaxContextCount(StrEnum):
"""Enum describing supported max context size presets."""
MB_25 = "25MB"
src/cleveragents/domain/models/core/enums.py
+5 -5
View File
@@ -1,9 +1,9 @@
"""Enums for CleverAgents domain models."""
from enum import Enum
from enum import StrEnum
class ModelPublisher(str, Enum):
class ModelPublisher(StrEnum):
"""Model publisher enumeration."""
OPENAI = "ModelPublisherOpenAI"
@@ -15,7 +15,7 @@ class ModelPublisher(str, Enum):
MISTRAL = "ModelPublisherMistral"
class ModelErrKind(str, Enum):
class ModelErrKind(StrEnum):
"""Model error kind enumeration."""
OVERLOADED = "ErrOverloaded"
@@ -26,7 +26,7 @@ class ModelErrKind(str, Enum):
CACHE_SUPPORT = "ErrCacheSupport"
class FallbackType(str, Enum):
class FallbackType(StrEnum):
"""Fallback type enumeration."""
ERROR = "FallbackTypeError"
@@ -34,7 +34,7 @@ class FallbackType(str, Enum):
PROVIDER = "FallbackTypeProvider"
class ModelProvider(str, Enum):
class ModelProvider(StrEnum):
"""Model provider enumeration."""
OPENROUTER = "ModelProviderOpenRouter"
src/cleveragents/domain/models/core/org.py
+3 -3
View File
@@ -8,7 +8,7 @@ from __future__ import annotations
from datetime import datetime
from decimal import Decimal
from enum import Enum
from enum import StrEnum
from pydantic import BaseModel, ConfigDict, Field
@@ -153,14 +153,14 @@ class Invite(BaseModel):
)
class CreditsTransactionType(str, Enum):
class CreditsTransactionType(StrEnum):
"""Type of credits transaction."""
CREDIT = "CreditsTransactionTypeCredit"
DEBIT = "CreditsTransactionTypeDebit"
class CreditType(str, Enum):
class CreditType(StrEnum):
"""Categorization for credits changes."""
TRIAL = "CreditTypeTrial"
src/cleveragents/domain/models/core/plan.py
+4 -4
View File
@@ -7,7 +7,7 @@ Based on spec.md and ADR-004 (Pydantic Validation).
"""
from datetime import datetime
from enum import Enum
from enum import StrEnum
from typing import Annotated
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
@@ -16,7 +16,7 @@ from pydantic import BaseModel, ConfigDict, Field, field_validator, model_valida
ULID_PATTERN = r"^[0-9A-HJKMNP-TV-Z]{26}$"
class PlanPhase(str, Enum):
class PlanPhase(StrEnum):
"""Phase of a plan in the v3 lifecycle.
Plans progress through phases in order:
@@ -30,7 +30,7 @@ class PlanPhase(str, Enum):
APPLIED = "applied"
class ActionState(str, Enum):
class ActionState(StrEnum):
"""States for plans in the Action phase.
Actions are reusable plan templates not tied to any project yet.
@@ -41,7 +41,7 @@ class ActionState(str, Enum):
ARCHIVED = "archived" # Soft-deleted or hidden
class ProcessingState(str, Enum):
class ProcessingState(StrEnum):
"""States for plans in Strategize/Execute/Apply phases.
These states indicate what is happening during processing.
src/cleveragents/domain/models/core/plan_legacy.py
+2 -2
View File
@@ -4,12 +4,12 @@ Based on Phase 0 discovery and ADR-004 (Pydantic Validation).
"""
from datetime import datetime
from enum import Enum
from enum import StrEnum
from pydantic import BaseModel, ConfigDict, Field, field_validator
class PlanStatus(str, Enum):
class PlanStatus(StrEnum):
"""Status of a plan."""
PENDING = "pending"
src/cleveragents/domain/models/planconfig/plan_config.py
+2 -2
View File
@@ -1,11 +1,11 @@
"""Domain models for CleverAgents - auto-generated from Phase 0 stubs."""
from enum import Enum
from enum import StrEnum
from pydantic import BaseModel, ConfigDict, Field
class AutoModeType(str, Enum):
class AutoModeType(StrEnum):
"""Enum for auto mode types."""
MANUAL = "manual"
src/cleveragents/domain/models/planfiles/__init__.py
+2 -2
View File
@@ -7,7 +7,7 @@ These models represent plan file operations, results, and state.
from __future__ import annotations
from datetime import datetime
from enum import Enum
from enum import Enum, StrEnum
from pydantic import BaseModel, ConfigDict, Field
@@ -119,7 +119,7 @@ class PlanApply(BaseModel):
)
class PlanStateStatus(str, Enum):
class PlanStateStatus(StrEnum):
"""Status of the current plan state."""
IDLE = "idle"
src/cleveragents/domain/models/stream/stream.py
+2 -2
View File
@@ -1,7 +1,7 @@
"""Domain models for CleverAgents - auto-generated from Phase 0 stubs."""
from datetime import datetime
from enum import Enum
from enum import StrEnum
from pydantic import BaseModel, ConfigDict, Field
@@ -9,7 +9,7 @@ from pydantic import BaseModel, ConfigDict, Field
from ..auth.auth import ApiError
class StreamMessageType(str, Enum):
class StreamMessageType(StrEnum):
"""Enum for stream message types."""
INFO = "info"
src/cleveragents/infrastructure/database/migration_runner.py
+4 -1
View File
@@ -4,6 +4,7 @@ This module handles running Alembic migrations programmatically
to ensure database schema is up to date.
"""
import logging
from collections.abc import Callable
from pathlib import Path
@@ -293,7 +294,9 @@ class MigrationRunner:
)
except Exception:
# Fall through to auto-approval below if prompting fails
pass
logging.getLogger(__name__).debug(
"Interactive prompt failed, auto-approving migrations"
)
# Non-interactive or prompt failure: auto-approve to avoid blocking
return True
src/cleveragents/langgraph/nodes.py
+1
View File
@@ -143,6 +143,7 @@ class Node: # pylint: disable=too-many-instance-attributes
try:
return await self.execute(state)
except Exception: # pylint: disable=broad-except
self.logger.warning("Node %s retry attempt failed", self.name)
continue
return {"error": str(exc), "failed_node": self.name}
finally:
src/cleveragents/providers/registry.py
+2 -2
View File
@@ -15,7 +15,7 @@ from __future__ import annotations
import os
from dataclasses import dataclass
from enum import Enum
from enum import StrEnum
from typing import TYPE_CHECKING, Any, ClassVar
from pydantic import BaseModel, ConfigDict, Field
@@ -35,7 +35,7 @@ def _coerce_optional_str(value: object | None) -> str | None:
return str(value)
class ProviderType(str, Enum):
class ProviderType(StrEnum):
"""Supported AI provider types."""
OPENAI = "openai"
src/cleveragents/reactive/stream_router.py
+78 -12
View File
@@ -5,6 +5,8 @@ semantics. Provider/model flags are not used; actors are resolved by name.
from __future__ import annotations
import ast
import contextlib
import logging
from collections.abc import Callable
from enum import Enum
@@ -22,9 +24,9 @@ from cleveragents.core.exceptions import StreamRoutingError
from cleveragents.providers.registry import get_provider_registry
try:
from jinja2 import Environment
from jinja2.sandbox import SandboxedEnvironment
except Exception: # pragma: no cover - optional runtime dependency guard
Environment = None
SandboxedEnvironment = None
try:
from langchain_core.messages import HumanMessage, SystemMessage
@@ -84,6 +86,65 @@ class StreamConfig(BaseModel): # pylint: disable=too-many-instance-attributes
template_config: dict[str, Any] | None = None
# ---------------------------------------------------------------------------
# AST validation helpers for safe dynamic code execution
# ---------------------------------------------------------------------------
# AST node types that are forbidden in user-supplied code blocks.
_FORBIDDEN_AST_NODES: set[type[ast.AST]] = {
ast.Import,
ast.ImportFrom,
ast.Global,
ast.Nonlocal,
}
logger_sr = logging.getLogger(__name__)
def _validate_code_ast(code: str) -> None:
"""Parse *code* and reject any forbidden AST constructs.
Raises ``StreamRoutingError`` if the code contains imports, global/nonlocal
statements, or calls to dangerous built-ins (exec, eval, compile, __import__).
"""
try:
tree = ast.parse(code)
except SyntaxError as exc:
raise StreamRoutingError(f"Invalid code syntax: {exc}") from exc
_DANGEROUS_NAMES = {"exec", "eval", "compile", "__import__", "getattr", "setattr"}
for node in ast.walk(tree):
if type(node) in _FORBIDDEN_AST_NODES:
raise StreamRoutingError(
f"Forbidden construct in tool code: {type(node).__name__}"
)
# Reject calls to dangerous built-in names
if (
isinstance(node, ast.Call)
and isinstance(node.func, ast.Name)
and node.func.id in _DANGEROUS_NAMES
):
raise StreamRoutingError(f"Forbidden call in tool code: {node.func.id}()")
def _validate_lambda_ast(fn_str: str) -> None:
"""Validate that *fn_str* is a single lambda expression and nothing else.
Raises ``StreamRoutingError`` if the string is not a safe lambda.
"""
try:
tree = ast.parse(fn_str, mode="eval")
except SyntaxError as exc:
raise StreamRoutingError(f"Invalid transform function syntax: {exc}") from exc
if not isinstance(tree.body, ast.Lambda):
raise StreamRoutingError(
"Transform 'fn' must be a lambda expression, "
f"got {type(tree.body).__name__}"
)
class SimpleToolAgent:
"""Executes inline tool code blocks from agent config."""
@@ -103,6 +164,8 @@ class SimpleToolAgent:
code = tool.get("code") if isinstance(tool, dict) else None
if not code:
return content
# Validate AST before execution — reject imports, dangerous calls, etc.
_validate_code_ast(code)
local_vars: dict[str, Any] = {
"input_data": content,
"metadata": meta,
@@ -110,7 +173,9 @@ class SimpleToolAgent:
"result": None,
}
try:
exec(code, {}, local_vars)
exec(code, {}, local_vars) # nosec B102
except StreamRoutingError:
raise # Re-raise validation errors
except Exception:
return ""
return local_vars.get("result", "")
@@ -129,7 +194,9 @@ class SimpleLLMAgent:
self.config = config or {}
self.system_prompt = self.config.get("system_prompt", "") or ""
self._llm = None
self._template_env = Environment() if Environment is not None else None
self._template_env = (
SandboxedEnvironment() if SandboxedEnvironment is not None else None
)
def _render_prompt(self, template: str, context: dict[str, Any]) -> str:
if not template:
@@ -337,7 +404,10 @@ class ReactiveStreamRouter: # pylint: disable=too-many-instance-attributes
if "fn" in params:
fn_str = params["fn"]
try:
transform_fn = eval(fn_str, {"__builtins__": {}})
# Validate: only lambda expressions allowed
_validate_lambda_ast(fn_str)
compiled = compile(fn_str, "<transform>", "eval")
transform_fn = eval(compiled, {"__builtins__": {}}) # nosec B307
return ops.map(
lambda msg: (
transform_fn(msg.content)
@@ -489,7 +559,7 @@ class ReactiveStreamRouter: # pylint: disable=too-many-instance-attributes
if acc is None:
acc = [] if config.get("init") == "list" else 0
if config.get("op") == "sum":
return acc + (msg if isinstance(msg, (int, float)) else 0)
return acc + (msg if isinstance(msg, int | float) else 0)
if config.get("op") == "append" and isinstance(acc, list):
acc.append(msg)
return acc
@@ -563,11 +633,9 @@ class ReactiveStreamRouter: # pylint: disable=too-many-instance-attributes
"""Dispose of all streams and subscriptions."""
# Dispose subscriptions first
for subscription in self.subscriptions[:]:
try:
with contextlib.suppress(Exception):
if hasattr(subscription, "dispose"):
subscription.dispose()
except Exception: # pylint: disable=broad-except
pass # Ignore disposal errors
# Clear subscriptions
self.subscriptions.clear()
@@ -575,11 +643,9 @@ class ReactiveStreamRouter: # pylint: disable=too-many-instance-attributes
# Dispose streams
for stream_name in list(self.streams.keys()):
stream = self.streams[stream_name]
try:
with contextlib.suppress(Exception):
if hasattr(stream, "dispose"):
stream.dispose()
except Exception: # pylint: disable=broad-except
pass # Ignore disposal errors
# Clear all dictionaries
self.streams.clear()
vulture_whitelist.py
+11
View File
@@ -0,0 +1,11 @@
# Vulture whitelist for false positives and intentional unused code.
# This file is referenced by vulture to suppress known false positives.
# Each entry represents either:
# - A variable required by a protocol/interface but not used in the body
# - A public API entry point that vulture cannot trace
# Context manager __exit__ parameters required by protocol
exc_tb # noqa: B018, F821
# Legacy migrator method parameter needed for mapping interface consistency
build_data # noqa: B018, F821