docs/CODE_QUALITY_STANDARDS.md

# Code Quality Standards

**Version:** 1.0
**Last Updated:** 2026-02-16
**Applies to:** All Python code in the Atom backend

## Overview

This document defines the code quality standards for the Atom backend project. Following these standards ensures maintainable, reliable, and production-ready code.

## Type Hints

### Requirements

- **New code:** MUST have 100% type hint coverage on all function signatures
- **Existing code:** Incremental adoption - add type hints when modifying functions
- **Type checking:** Run MyPy locally before committing changes

### Type Hint Patterns

```python

# Function signatures MUST include parameter types and return types

def process_agent(agent_id: str, maturity_level: int) -> Dict[str, Any]:

    """Process an agent with given maturity level."""

    pass



# Use Optional for nullable types

def get_agent(agent_id: str) -> Optional[AgentRegistry]:

    """Get agent by ID, returns None if not found."""

    pass



# Use complex types for collections

def list_agents(category: Optional[str] = None) -> List[AgentRegistry]:

    """List agents by category."""

    pass



# Async functions must specify return types

async def execute_workflow(workflow_id: str) -> WorkflowExecution:

    """Execute a workflow asynchronously."""

    pass

```

### Import Required Types

```python

from typing import Any, Dict, List, Optional, Union, Callable, AsyncIterator

```

### MyPy Configuration

MyPy is configured in `backend/mypy.ini`:

- **Python version:** 3.11
- **Incremental adoption:** `disallow_untyped_defs = False`
- **Check untyped defs:** Enabled to gradually add type hints
- **Third-party libs:** Missing imports ignored

### Running MyPy

```bash

# Type check specific files

cd backend && mypy core/llm/byok_handler.py



# Type check entire core directory

cd backend && mypy core/ --config-file mypy.ini



# Type check with verbose output

cd backend && mypy core/ --show-error-codes --show-error-context

```

## Error Handling

### Principles

1. **Use specific exception types** - Never catch bare `Exception:` if you can be more specific
2. **Log with context** - Always include relevant context in error messages
3. **Preserve stack traces** - Use `raise ... from e` for exception chaining
4. **Never swallow exceptions silently** - Always log or handle appropriately

### Error Handling Patterns

```python

# Standardized error handling with context

try:

    agent = db.query(AgentRegistry).filter(AgentRegistry.id == agent_id).first()

    if not agent:

        raise ValueError(f"Agent not found: {agent_id}")

except SQLAlchemyError as e:

    logger.error(f"Database error while fetching agent {agent_id}: {e}")

    raise



# Validation errors with descriptive messages

def validate_agent_maturity(agent: AgentRegistry, action: str) -> None:

    """Validate agent maturity for performing an action."""

    required_level = ACTION_COMPLEXITY.get(action, 2)

    if agent.confidence_score < required_level:

        raise ValueError(

            f"Agent {agent.name} (score: {agent.confidence_score}) "

            f"insufficient for {action} (required: {required_level})"

        )



# Exception chaining to preserve stack traces

try:

    result = await external_service_call()

except httpx.HTTPError as e:

    logger.error(f"External service error: {e}")

    raise ExternalServiceError(f"Failed to call external service: {e}") from e

```

### Exception Categories

- **Database errors:** `SQLAlchemyError`, `IntegrityError`
- **Validation errors:** `ValueError`, `ValidationError`
- **Not found errors:** Return `None` or raise specific exception
- **Permission errors:** Custom `PermissionDeniedException`
- **External service errors:** `ExternalServiceError`, `HTTPError`

## Logging

### Standards

- **Use structlog** for structured logging with context
- **Include relevant context** in all log messages
- **Use appropriate log levels:** DEBUG, INFO, WARNING, ERROR, CRITICAL

### Logging Patterns

```python

import structlog



logger = structlog.get_logger(__name__)



# Debug level for detailed troubleshooting

logger.debug(f"Cache HIT for governance check: {agent_id}:{action_type}")



# Info level for normal operations

logger.info(f"Registered new agent: {name}", agent_id=agent.id)



# Warning level for unexpected but recoverable issues

logger.warning(f"Failed to initialize {provider_id} client: {e}")



# Error level for failures

logger.error(f"LLM Generation failed: {e}", agent_id=agent_id)



# Critical level for system-wide failures

logger.critical(f"Database connection lost: {e}")

```

### Context Enrichment

```python

# Include relevant context in log messages

logger.info(

    "Agent maturity transition",

    agent_id=agent.id,

    agent_name=agent.name,

    previous_status=previous_status,

    new_status=agent.status,

    confidence_score=new_score

)

```

## Documentation

### Docstring Standards

All functions MUST have Google-style docstrings with Args and Returns sections:

```python

def process_workflow(

    workflow_id: str,

    inputs: Dict[str, Any],

    timeout: int = 30

) -> Optional[WorkflowExecution]:

    """

    Process a workflow with given inputs.



    Args:

        workflow_id: Unique identifier for the workflow

        inputs: Dictionary of input parameters for the workflow

        timeout: Maximum execution time in seconds (default: 30)



    Returns:

        WorkflowExecution object if successful, None if failed



    Raises:

        ValueError: If workflow_id is invalid

        TimeoutError: If workflow execution exceeds timeout

    """

    pass

```

### Comment Standards

- **Comment WHY, not WHAT** - Code should be self-explanatory about what it does
- **Keep comments up to date** - Outdated comments are worse than no comments
- **Use docstrings for function/module documentation** - Use inline comments for complex logic

```python

# GOOD: Explains why we're doing this

# Use 0.5 only if confidence_score is None, not if it's 0.0

current = agent.confidence_score if agent.confidence_score is not None else 0.5



# BAD: Just repeats what the code says

# Set current to confidence_score

current = agent.confidence_score if agent.confidence_score is not None else 0.5

```

## Code Formatting

### Standards

- **Use Black** for code formatting (88 character line length)
- **Use Ruff** for fast linting
- **Format on save** - Configure your IDE to format code automatically

### Configuration

Black and Ruff configuration in `backend/pyproject.toml`:

```toml

[tool.black]

line-length = 88

target-version = ['py311']



[tool.ruff]

line-length = 88

select = ["E", "F", "I", "N", "W"]

```

### Running Formatters

```bash

# Format code with Black

cd backend && black .



# Check formatting without modifying

cd backend && black --check .



# Lint with Ruff

cd backend && ruff check .



# Auto-fix linting issues

cd backend && ruff check --fix .

```

## Testing Standards

### Principles

- **Use pytest** for all testing
- **Type-safe fixtures** - Add type hints to test fixtures
- **Arrange-Act-Assert** pattern for clear test structure
- **Test names should describe behavior** - `test_agent_promotion_when_confidence_exceeds_threshold`

### Test Patterns

```python

import pytest

from core.models import AgentRegistry

from core.agent_governance_service import AgentGovernanceService



def test_agent_promotion_when_confidence_exceeds_threshold(db_session: Session):

    """

    Test that agent is promoted to AUTONOMOUS when confidence exceeds 0.9.

    """

    # Arrange

    governance = AgentGovernanceService(db_session)

    agent = AgentRegistry(

        name="Test Agent",

        category="testing",

        confidence_score=0.95

    )

    db_session.add(agent)

    db_session.commit()



    # Act

    result = governance.enforce_action(agent.id, "delete")



    # Assert

    assert result["proceed"] is True

    assert result["status"] == "APPROVED"

```

### Test Coverage Goals

- **Critical paths:** 90%+ coverage (governance, LLM routing, workflows)
- **Business logic:** 80%+ coverage
- **Overall:** 70%+ coverage

### Test File Naming Convention

**Critical Rule:** Test filenames must be unique across the entire test suite, regardless of directory.

Python's import system is basename-based, not path-based. Files with identical basenames in different directories cause collection errors.

**Naming Patterns:**
- Primary service tests: `test_<service>_coverage.py` (e.g., `test_agent_governance_service_coverage.py`)
- Module-specific tests: `test_<service>_<module>.py` (e.g., `test_agent_graduation_service_memory.py`)
- Extended coverage: `test_<service>_coverage_extend.py` (e.g., `test_cognitive_tier_system_coverage_extend.py`)

**Examples:**
```

tests/core/agents/test_agent_graduation_service_coverage.py      # Primary coverage

tests/core/memory/test_agent_graduation_service_memory.py         # Memory module tests

tests/core/episodes/test_episode_retrieval_service_coverage.py    # Primary coverage

tests/core/memory/test_episode_retrieval_memory.py                # Memory module tests

tests/core/llm/test_cognitive_tier_system_coverage.py            # Primary coverage

tests/core/llm/test_cognitive_tier_system_coverage_extend.py      # Extended coverage

```

**Anti-Patterns to Avoid:**
- DO NOT use same basename in different directories: `tests/core/agents/test_foo.py` and `tests/core/memory/test_foo.py`
- DO NOT rely on directory structure for uniqueness (Python ignores directories in module resolution)
- DO NOT ignore collection errors (they mask missing tests and inflate coverage gaps)

**Verifying Uniqueness:**
```bash

# Check for duplicate test basenames

find tests/ -name "test_*.py" -type f | xargs -n1 basename | sort | uniq -d

# Expected: No output (no duplicates)

```

**Collection Error Detection:**
```bash

# Check for collection errors before committing

pytest --collect-only -q 2>&1 | grep -c "ERROR collecting"

# Expected: 0

```

### Security Testing Patterns

Security tests validate defense-in-depth protections against real-world attacks. Use these patterns when testing sandbox execution, package installation, or external code execution.

#### Container Escape Tests

Validate Docker isolation prevents breakout attempts:

```python

class TestContainerEscape:

    """Container escape attack prevention tests."""



    @patch('core.skill_sandbox.docker.from_env')

    def test_privileged_mode_disabled(self, mock_docker, mock_docker_client):

        """

        Verify containers NEVER run with --privileged flag.



        Privileged mode disables all security mechanisms and allows

        full host access (CVE-2019-5736, CVE-2025-9074).



        Security: CRITICAL

        """

        mock_docker.return_value = mock_docker_client



        sandbox = HazardSandbox()

        sandbox.execute_python(code="print('test')", inputs={})



        # Verify privileged=False (or not set, default is False)

        call_kwargs = mock_docker_client.containers.run.call_args[1]

        assert call_kwargs.get('privileged', False) == False, \

            "Container must NOT run in privileged mode"



    @patch('core.skill_sandbox.docker.from_env')

    def test_docker_socket_not_mounted(self, mock_docker, mock_docker_client):

        """

        Verify Docker socket is NEVER mounted in containers.



        Mounting /var/run/docker.sock enables container escape

        and full host control (Docker-out-of-Docker attack).



        Security: CRITICAL

        """

        mock_docker.return_value = mock_docker_client



        sandbox = HazardSandbox()

        sandbox.execute_python(code="print('test')", inputs={})



        call_kwargs = mock_docker_client.containers.run.call_args[1]

        volumes = call_kwargs.get('volumes', {})



        assert '/var/run/docker.sock' not in str(volumes), \

            "Docker socket must NOT be mounted (enables container escape)"

```

#### Resource Exhaustion Tests

Validate resource limits prevent DoS attacks:

```python

class TestResourceExhaustion:

    """Resource limit enforcement tests."""



    @patch('core.skill_sandbox.docker.from_env')

    def test_memory_limit_enforced(self, mock_docker, mock_docker_client):

        """

        Verify memory limit is set to prevent exhaustion attacks.



        Security: HIGH - Memory exhaustion can DoS the host

        """

        mock_docker.return_value = mock_docker_client



        sandbox = HazardSandbox()

        sandbox.execute_python(

            code="print('test')",

            inputs={},

            memory_limit="256m"

        )



        # Verify mem_limit is set

        call_kwargs = mock_docker_client.containers.run.call_args[1]

        assert call_kwargs.get('mem_limit') == "256m", \

            "Memory limit must be enforced to prevent exhaustion attacks"



    @patch('core.skill_sandbox.docker.from_env')

    def test_cpu_quota_enforced(self, mock_docker, mock_docker_client):

        """

        Verify CPU quota is set to prevent CPU exhaustion.



        Security: HIGH - CPU exhaustion can starve host processes

        """

        mock_docker.return_value = mock_docker_client



        sandbox = HazardSandbox()

        sandbox.execute_python(

            code="print('test')",

            inputs={},

            cpu_limit=0.5

        )



        # Verify cpu_quota is set (0.5 * 100000 = 50000)

        call_kwargs = mock_docker_client.containers.run.call_args[1]

        assert call_kwargs.get('cpu_quota') == 50000, \

            "CPU quota must be enforced (0.5 * 100000)"

```

#### Network Isolation Tests

Validate network isolation prevents data exfiltration:

```python

class TestNetworkIsolation:

    """Network isolation enforcement tests."""



    @patch('core.skill_sandbox.docker.from_env')

    def test_network_disabled(self, mock_docker, mock_docker_client):

        """

        Verify network is disabled to prevent data exfiltration.



        Security: CRITICAL - Network isolation prevents outbound attacks

        """

        mock_docker.return_value = mock_docker_client



        sandbox = HazardSandbox()

        sandbox.execute_python(code="print('test')", inputs={})



        # Verify network_disabled=True

        call_kwargs = mock_docker_client.containers.run.call_args[1]

        assert call_kwargs.get('network_disabled') is True, \

            "Network must be disabled to prevent data exfiltration"

```

#### Malicious Pattern Detection Tests

Validate static scanning detects malicious code patterns:

```python

class TestMaliciousPatternDetection:

    """Static scanning detects malicious patterns."""



    def test_subprocess_usage_detected(self, security_scanner):

        """

        Static scan detects subprocess usage.



        Security: HIGH - subprocess enables arbitrary command execution

        """

        malicious_code = """

import subprocess

user_input = 'rm -rf /'

subprocess.call(user_input, shell=True)

"""



        result = security_scanner.scan_skill(

            skill_name="malicious-subprocess",

            skill_content=malicious_code

        )



        assert result["safe"] == False, "Subprocess usage must be blocked"

        assert len(result["findings"]) > 0, "Security findings must be reported"

        assert any("subprocess" in f.lower() for f in result["findings"]), \

            "Finding must mention subprocess"



    def test_base64_obfuscation_detected(self, security_scanner):

        """

        Static scan detects base64 obfuscation.



        Security: HIGH - Base64 obfuscation hides malicious payloads

        """

        malicious_code = """

import base64

payload = 'c3VicHJvY2Vzcy5ydW4oWyJybSIsICJyZiIsICIvIl0p'

decoded = base64.b64decode(payload).decode()

exec(decoded)

"""



        result = security_scanner.scan_skill(

            skill_name="obfuscated-base64",

            skill_content=malicious_code

        )



        # Should detect base64.b64decode as suspicious

        assert result["safe"] == False or len(result["findings"]) > 0, \

            "Base64 obfuscation must be flagged"

```

#### Governance Blocking Tests

Validate maturity-based access controls:

```python

class TestGovernanceBlocking:

    """Maturity-based governance blocks unauthorized access."""



    def test_student_agent_blocked_from_python_packages(self, governance, db_session: Session):

        """

        STUDENT agents cannot use Python packages (non-negotiable).



        Security: CRITICAL - Educational restriction, cannot be bypassed

        """

        student_agent = StudentAgentFactory(_session=db_session)

        db_session.commit()



        result = governance.check_package_permission(

            agent_id=student_agent.id,

            package_name="numpy",

            version="1.21.0",

            db=db_session

        )



        assert result["allowed"] == False, \

            "STUDENT agents must be blocked from ALL Python packages"

        assert "STUDENT agents cannot" in result["reason"], \

            "Reason must mention STUDENT restriction"



    def test_banned_package_blocks_all_agents(self, governance, db_session: Session):

        """

        Banned packages block ALL agents regardless of maturity.



        Security: CRITICAL - Ban list overrides all other rules

        """

        # Ban package

        governance.ban_package(

            package_name="malicious-lib",

            version="1.0.0",

            reason="Security vulnerability: CVE-2025-99999",

            db=db_session

        )



        # Create AUTONOMOUS agent (highest maturity)

        autonomous_agent = AutonomousAgentFactory(_session=db_session)

        db_session.commit()



        result = governance.check_package_permission(

            agent_id=autonomous_agent.id,

            package_name="malicious-lib",

            version="1.0.0",

            db=db_session

        )



        assert result["allowed"] == False, \

            "Banned packages must block even AUTONOMOUS agents"

        assert "banned" in result["reason"].lower(), \

            "Reason must mention ban"

```

#### Security Testing Best Practices

1. **Mock external dependencies** - Use `@patch` decorators to mock Docker, subprocess calls
2. **Test malicious fixtures** - Create fixture files with attack samples for reproducible testing
3. **Defense-in-depth validation** - Test all security layers (static scan + sandbox + governance)
4. **Security level annotations** - Mark tests with security level (CRITICAL, HIGH, MEDIUM, LOW)
5. **Clear assertion messages** - Explain WHY the security constraint is critical (reference CVEs)
6. **Use malicious package fixtures** - Import from `tests.fixtures.malicious_packages` for attack samples

Reference: `backend/tests/test_package_security.py` for comprehensive examples.

### API Route Testing with BaseAPIRouter

Testing API routes that use BaseAPIRouter requires specific patterns for mock patching and error response assertions.

#### Mock Patching: Patch Where Imported

Always patch services at their import location in the route module, not at their definition location.

```python

# Route file imports:

# api/admin/business_facts_routes.py

from core.agent_world_model import WorldModelService



# Test file patches at import location:

with patch('api.admin.business_facts_routes.WorldModelService', return_value=mock_service):

    response = client.get("/api/admin/governance/facts/123")

```

**Why:** When the route module imports `WorldModelService` at module level, it creates a reference to the original class. Patching at the definition location (`core.agent_world_model.WorldModelService`) doesn't affect this already-imported reference. Patching at the import location intercepts the reference the route actually uses.

**Exception:** For services imported inside functions (not at module level), patch at the original location:

```python

# Route function with local import:

def upload_document(file: UploadFile):

    from core.storage import get_storage_service  # Local import

    storage = get_storage_service()



# Test patch location:

with patch('core.storage.get_storage_service', return_value=mock_storage):

    response = client.post("/upload", files={"file": test_file})

```

#### Error Response Assertions

BaseAPIRouter.error_response() returns structured errors. Access nested message field:



```python

# Success response (200 OK)

response = client.get("/api/admin/governance/facts/123")

assert response.status_code == 200
data = response.json()
assert data["success"] == True
assert data["data"]["fact"] == "Invoices over $500 require VP approval"

# Error response (404, 400, 500, etc.)
response = client.get("/api/admin/governance/facts/non-existent")
assert response.status_code == 404

detail = response.json()["detail"]

assert detail["success"] == False

assert detail["error"]["code"] == "NOT_FOUND"
assert "not found" in detail["error"]["message"].lower()  # String operations on message
```



**Error Response Structure:**

```json

{

  "success": false,

  "error": {

    "code": "NOT_FOUND",

    "message": "Business fact not found: fact-123",

    "timestamp": "2026-03-20T11:05:21.123456",

    "details": {}

  }

}

```

**Common HTTP Status Codes:**
- `400` - VALIDATION_ERROR (invalid input from BaseAPIRouter.validation_error)
- `401` - UNAUTHORIZED (missing/invalid auth)
- `403` - FORBIDDEN (insufficient permissions)
- `404` - NOT_FOUND (resource not found from BaseAPIRouter.not_found_error)

- `500` - INTERNAL_ERROR (server error)

#### Service Mock Fixtures

Use AsyncMock for async services, configure returns in fixture:

```python

@pytest.fixture

def mock_world_model_service(sample_business_fact):

    """Mock WorldModelService with configured return values."""

    mock = AsyncMock()

    mock.get_fact_by_id.return_value = sample_business_fact

    mock.list_all_facts.return_value = [sample_business_fact]

    mock.create_fact.return_value = sample_business_fact

    return mock



def test_get_fact_success(authenticated_admin_client, mock_world_model_service):

    """Test getting a business fact by ID."""

    with patch('api.admin.business_facts_routes.WorldModelService',

               return_value=mock_world_model_service):



        response = authenticated_admin_client.get("/api/admin/governance/facts/fact-123")



        assert response.status_code == 200

        data = response.json()

        assert data["success"] == True

```

**AsyncMock vs MagicMock:**
- **AsyncMock:** For async service methods (most API services)
- **MagicMock:** For sync methods or non-service objects

#### Configure Mocks Inside Patch Context

When overriding fixture defaults, configure mocks **inside** the patch context manager:

```python

# Before (incorrect):

mock_extractor.extract_facts_from_document.return_value = result

with patch('api.admin.routes.get_policy_fact_extractor', return_value=mock_extractor):

    # Test - uses fixture default, not result



# After (correct):

with patch('api.admin.routes.get_policy_fact_extractor',

           return_value=mock_extractor) as patched_extractor:

    patched_extractor.extract_facts_from_document.return_value = result

    # Test - uses test-specific result

```

#### S3/R2 Storage Mocking

Mock storage services for citation verification:

```python

@pytest.fixture

def mock_storage_service():

    """Mock S3/R2 storage service."""

    mock = MagicMock()

    mock.upload_file.return_value = "s3://atom-business-facts/uploads/test.pdf"

    mock.check_exists.return_value = True

    mock.download_file.return_value = b"PDF content bytes"

    return mock

```

#### PDF Extraction Mocking

Mock PDF extraction services for document upload tests:

```python

@pytest.fixture

def mock_policy_extractor():

    """Mock policy fact extractor."""

    mock = AsyncMock()

    mock.extract_facts_from_document.return_value = {

        "facts": [

            {

                "fact": "Invoices over $500 require VP approval",

                "citations": ["s3://atom-business-facts/policies/ap-policy.pdf:page:5"],

                "reason": "Extracted from AP policy document",

                "confidence": 0.95

            }

        ],

        "metadata": {"source": "ap-policy.pdf", "page_count": 10}

    }

    return mock

```

#### Complete Example

```python

def test_upload_and_extract_success(authenticated_admin_client,

                                   mock_policy_extractor,

                                   mock_storage_service):

    """Test successful document upload and fact extraction."""

    test_file = io.BytesIO(b"%PDF-1.4 ... test PDF content ...")



    with patch('api.admin.business_facts_routes.get_policy_fact_extractor',

               return_value=mock_policy_extractor):

        with patch('core.storage.get_storage_service', return_value=mock_storage_service):

            response = authenticated_admin_client.post(

                "/api/admin/governance/facts/upload",

                files={"file": ("test.pdf", test_file, "application/pdf")}

            )



            assert response.status_code == 201

            data = response.json()

            assert len(data["data"]["facts"]) == 1

            assert "VP approval" in data["data"]["facts"][0]["fact"]

```

**Reference:** See `.planning/phases/216-fix-business-facts-test-failures/216-PATTERN-DOC.md` for complete patterns with before/after examples from Phase 216 fixes.

## Code Review Checklist

Before submitting a PR, verify:

- [ ] All functions have type hints (parameters and return types)
- [ ] MyPy passes without critical errors
- [ ] Black formatting applied
- [ ] Ruff linting passes
- [ ] All functions have docstrings (Args/Returns)
- [ ] Error handling uses specific exception types
- [ ] No bare `except:` clauses
- [ ] Logging includes relevant context
- [ ] Tests added for new functionality
- [ ] Tests pass locally

## Continuous Integration

MyPy will be integrated into CI/CD pipeline:

```yaml

# Type check with mypy (commented out during incremental adoption)

# - name: Type check with mypy

#   run: mypy core/ --config-file backend/mypy.ini

```

**Note:** During incremental adoption, MyPy is run locally but not enforced in CI. Once type coverage reaches 80%, CI enforcement will be enabled.

## Resources

- [MyPy Documentation](https://mypy.readthedocs.io/)
- [Black Documentation](https://black.readthedocs.io/)
- [Ruff Documentation](https://docs.astral.sh/ruff/)
- [Google Python Style Guide](https://google.github.io/styleguide/pyguide.html)
- [PEP 8 - Style Guide for Python Code](https://peps.python.org/pep-0008/)

## Compliance

All Atom backend developers MUST follow these standards. Code that does not meet these standards will not be merged to main.

**Questions?** Contact the platform team or open an issue in the repository.