Hardening Codebases for Agentic Coding: 10 Essential Techniques¶

from pydantic import BaseModel, Field, field_validator


class ExtractedConcept(BaseModel):
    """A concept extracted from text by an LLM agent."""

    name: str = Field(..., min_length=1)
    concept_type: str
    description: str | None = None
    confidence: float = Field(ge=0.0, le=1.0)

    @field_validator("concept_type")
    @classmethod
    def normalize_concept_type(cls, v: str) -> str:
        assert v is not None, "concept_type must not be None"
        assert isinstance(v, str), "concept_type must be a string"
        return v.strip().lower()

from typing import TypedDict


class FileProcessingStatus(TypedDict):
    """Status tracking for file processing during extraction."""

    has_failed_chunks: bool
    processed_chunks: int
    total_chunks: int

from typing import Protocol


class StageValidator(Protocol):
    """Protocol for validators that run between pipeline stages."""

    def validate(self, **kwargs: Any) -> ValidationResult:
        """Validate stage preconditions or outputs."""
        ...

def add_error(
    self,
    error_type: str,
    message: str,
    severity: str = "error",
    **kwargs: object,
) -> None:
    """Add an error to the collection."""
    # First assertion: parameter validation
    assert error_type is not None, "error_type must not be None"
    assert isinstance(error_type, str), "error_type must be a string"

    # Second assertion: state validation
    assert isinstance(self._errors, list), "Internal errors must be a list"
    assert severity in {"error", "warning", "info"}, f"Invalid severity: {severity}"

    error_dict = {
        "type": error_type,
        "message": message,
        "severity": severity,
        **kwargs,
    }

    # Third assertion: output validation
    assert isinstance(error_dict, dict), "Error must be a dictionary"
    assert "type" in error_dict, "Error must have a type field"

    self._errors.append(error_dict)

import math

from hypothesis import assume, given
from hypothesis import strategies as st

# Define strategies for valid inputs
finite_floats = st.floats(
    min_value=-1e10,
    max_value=1e10,
    allow_infinity=False,
    allow_nan=False,
)


class TestSafeDivide:
    """Property-based tests for safe_divide function."""

    @given(a=finite_floats, b=st.floats(min_value=1e-10, max_value=1e10))
    def test_division_reversibility(self, a: float, b: float):
        """Property: safe_divide(a*b, b) == a when b > 0."""
        product = a * b
        assume(math.isfinite(product))
        result = safe_divide(product, b)
        assert math.isclose(result, a, rel_tol=1e-9, abs_tol=1e-9)

    @given(numerator=finite_floats, denominator=finite_floats)
    def test_non_positive_denominator_returns_default(
        self, numerator: float, denominator: float
    ):
        """Property: safe_divide returns default when denominator <= 0."""
        assume(denominator <= 0)
        custom_default = 42.0
        result = safe_divide(numerator, denominator, default=custom_default)
        assert result == custom_default

from hypothesis import HealthCheck, Verbosity, settings

# Configure profiles in conftest.py
settings.register_profile("ci", max_examples=1000, verbosity=Verbosity.verbose)
settings.register_profile("dev", max_examples=100)
settings.register_profile("debug", max_examples=10, verbosity=Verbosity.verbose)
settings.register_profile(
    "fast",
    max_examples=50,
    suppress_health_check=[HealthCheck.too_slow],
)

# Load profile from environment
settings.load_profile("dev")

@given(st.floats(), st.floats())
def test_addition_commutative(a, b):
    """a + b == b + a"""
    assert add(a, b) == add(b, a)

@given(st.text())
def test_normalization_idempotent(text):
    """normalize(normalize(x)) == normalize(x)"""
    once = normalize(text)
    twice = normalize(once)
    assert once == twice

@given(st.lists(st.integers()))
def test_sort_preserves_length(lst):
    """len(sort(x)) == len(x)"""
    assert len(sort(lst)) == len(lst)

@given(st.text())
def test_encode_decode_roundtrip(text):
    """decode(encode(x)) == x"""
    assert decode(encode(text)) == text

# In conftest.py
pytest_plugins = ["pytest_examples"]


# In test_doc_examples.py
def test_readme_examples(file_path="README.md"):
    """Ensure all code examples in README actually work."""
    pytest_examples.process_file(file_path)

import ast
from pathlib import Path

import pytest


def get_imports_from_file(file_path: Path) -> set[str]:
    """Extract all imports from a Python file using AST."""
    with open(file_path) as f:
        tree = ast.parse(f.read())

    imports = set()
    for node in ast.walk(tree):
        if isinstance(node, ast.Import):
            for alias in node.names:
                imports.add(alias.name.split(".")[0])
        elif isinstance(node, ast.ImportFrom) and node.module:
            imports.add(node.module.split(".")[0])
    return imports


def check_no_cross_imports(
    source_package: Path,
    forbidden_package_name: str,
) -> list[str]:
    """Check that source_package doesn't import from forbidden_package_name."""
    violations = []
    for py_file in source_package.rglob("*.py"):
        with open(py_file) as f:
            content = f.read()

        # Check for forbidden imports
        if f"from {forbidden_package_name}" in content:
            violations.append(f"{py_file}: imports from {forbidden_package_name}")
        if f"import {forbidden_package_name}" in content:
            violations.append(f"{py_file}: imports {forbidden_package_name}")

    return violations


@pytest.mark.architecture
def test_commons_does_not_import_engines():
    """Commons module should NOT import from any engine."""
    commons_dir = Path("myproject/commons")

    violations = []
    violations.extend(check_no_cross_imports(commons_dir, "engine_a"))
    violations.extend(check_no_cross_imports(commons_dir, "engine_b"))

    assert (
        len(violations) == 0
    ), "Commons should not import from engines:\n" + "\n".join(violations)


@pytest.mark.architecture
def test_no_wildcard_imports():
    """Main code should not use wildcard imports (from x import *)."""
    violations = []
    for py_file in Path("myproject/src").rglob("*.py"):
        with open(py_file) as f:
            tree = ast.parse(f.read())

        for node in ast.walk(tree):
            if isinstance(node, ast.ImportFrom):
                for alias in node.names:
                    if alias.name == "*":
                        violations.append(str(py_file))

    assert len(violations) == 0, "Found wildcard imports"


@pytest.mark.architecture
def test_expected_directory_structure():
    """Verify the expected directory structure exists."""
    expected_dirs = [
        Path("myproject/commons"),
        Path("myproject/engine_a"),
        Path("myproject/engine_a/core"),
        Path("myproject/engine_a/config"),
    ]

    missing = [str(d) for d in expected_dirs if not d.exists()]
    assert len(missing) == 0, f"Missing expected directories: {missing}"

# pytest.ini / pyproject.toml
[tool.pytest.ini_options]
markers = [
    "unit: Fast unit tests (< 100ms)",
    "integration: Slower integration tests requiring APIs",
    "architecture: Architectural boundary tests",
    "examples: Tests from documentation examples",
]
addopts = [
    "-m",
    "not integration",  # Skip integration tests by default
    "--ff",  # Run failures first
    "-n",
    "auto",  # Parallel execution
]

# BAD: Using mocks (will be caught by mockbuster)
from unittest.mock import Mock


def test_with_mock():
    mock_api = Mock()
    mock_api.get_data.return_value = {"key": "value"}
    result = process(mock_api)
    assert result == "value"


# GOOD: Using dependency injection with real test implementation
class FakeAPI:
    """Test double that implements the real API interface."""

    def __init__(self):
        self.calls = []

    def get_data(self):
        self.calls.append("get_data")
        return {"key": "value"}


def test_with_real_implementation():
    fake_api = FakeAPI()
    result = process(fake_api)
    assert result == "value"
    assert "get_data" in fake_api.calls

def test_cli_with_system_interaction(monkeypatch):  # mockbuster: ignore - testing CLI
    """Test CLI without actually calling the system."""
    monkeypatch.setattr(sys, "exit", lambda code: None)
    # CLI test code...

import typer

# Create the app with examples in docstring for cliqa
app = typer.Typer(
    name="myapp",
    help="My Application CLI",
    epilog="Examples:\n  myapp process path/to/data\n  myapp info path/to/data",
)


@app.command()
def process(
    data_path: Path = typer.Argument(
        ...,
        help="Path to data directory containing input files",
        exists=True,
        file_okay=False,
        dir_okay=True,
    ),
    verbose: bool = typer.Option(
        False,
        "--verbose",
        "-v",
        help="Enable verbose output",
    ),
) -> None:
    """
    Process data from input files.

    Processes all files in the data directory
    and generates outputs.

    Examples:
        myapp process data/my_project
        myapp process data/my_project --verbose
    """
    # Implementation...

# BAD: Hard to test without mocks
class OntologyExtractor:
    def __init__(self):
        self.agent = Agent()  # Hard dependency
        self.db = Neo4jConnection()  # Hard dependency


# GOOD: Easy to test with real test implementations
class OntologyExtractor:
    def __init__(
        self,
        agent: Agent[None, ExtractionResult],
        storage: StorageProtocol,
    ):
        self.agent = agent  # Injected
        self.storage = storage  # Injected


# Test with real implementation
class FakeStorage:
    def __init__(self):
        self.stored_items = []

    def save(self, item):
        self.stored_items.append(item)

from rich.console import Console
from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn
from rich.panel import Panel

console = Console()

def log_extraction_start(domain: str, file_count: int) -> None:
    """Log extraction start with formatted output."""
    assert domain is not None, "domain must not be None"
    assert file_count > 0, "file_count must be positive"

    console.print(Panel.fit(
        f"[bold blue]Starting extraction[/bold blue]\n"
        f"Domain: {domain}\n"
        f"Files: {file_count}",
        border_style="blue"
    ))

async def process_with_progress(
    files: list[Path],
) -> list[ExtractionResult]:
    """Process files with progress bar."""
    assert files is not None, "files must not be None"
    assert len(files) > 0, "files must not be empty"

    results = []

    with Progress(
        SpinnerColumn(),
        TextColumn("[progress.description]{task.description}"),
        BarColumn(),
        TextColumn("[progress.percentage]{task.percentage:>3.0f}%"),
        console=console,
    ) as progress:
        task = progress.add_task("Processing files", total=len(files))

        for file_path in files:
            result = await extract_from_file(file_path)
            results.append(result)
            progress.update(task, advance=1)

    return results

@dataclass
class ExtractionMetrics:
    """Metrics tracked during extraction."""

    start_time: float
    end_time: float | None = None

    files_processed: int = 0
    files_failed: int = 0

    chunks_processed: int = 0
    concepts_extracted: int = 0
    relationships_extracted: int = 0

    llm_calls: int = 0
    llm_tokens_input: int = 0
    llm_tokens_output: int = 0

    @property
    def duration_seconds(self) -> float:
        """Calculate total duration."""
        assert self.start_time > 0, "start_time must be set"

        end = self.end_time or time.time()
        duration = end - self.start_time

        assert duration >= 0, "duration must be non-negative"
        return duration

    @property
    def throughput_files_per_second(self) -> float:
        """Calculate file processing throughput."""
        assert self.files_processed >= 0, "files_processed must be non-negative"

        if self.duration_seconds == 0:
            return 0.0

        return self.files_processed / self.duration_seconds

async def process_pipeline(data_path: Path) -> PipelineOutput:
    """
    Process data pipeline with full hardening.

    Demonstrates all 10 techniques:
    1. Type safety: Pydantic models
    2. Defensive assertions: 2+ per function (NASA05)
    3. Property-based fuzzing: Hypothesis for edge cases
    4. Comprehensive testing: pytest markers, doc tests, arch tests
    5. Pre-commit gates: enforced quality
    6. Mock detection: real implementations only
    7. CLI quality: validated with cliqa
    8. Modular architecture: dependency injection
    9. Documentation: contracts and runbooks
    10. Observability: rich logging and metrics
    """
    # Defensive assertions (Technique 2)
    assert domain_path is not None, "domain_path must not be None"
    assert domain_path.exists(), f"domain_path must exist: {domain_path}"
    assert domain_path.is_dir(), f"domain_path must be directory: {domain_path}"

    # Structured configuration (Technique 6)
    config = load_config(domain_path)

    # State management (Technique 8)
    state = ExtractionState(
        domain_path=domain_path,
        config=config,
        errors=ErrorCollector(),
        metrics=ExtractionMetrics(start_time=time.time()),
    )

    # Observability (Technique 10)
    log_extraction_start(domain_path.name, len(state.files))

    # Modular architecture with DI (Technique 5)
    agent = create_extraction_agent(config.llm)
    storage = create_storage(domain_path)

    try:
        # Error collection (Technique 7)
        results = await process_with_error_collection(
            files=state.files,
            agent=agent,
            collector=state.errors,
        )

        # Type safety (Technique 1)
        for result in results:
            assert isinstance(result, ExtractionResult), "Must be ExtractionResult"
            state.concepts.extend(result.concepts)
            state.relationships.extend(result.relationships)

        # Save outputs (contracts from Technique 9)
        await storage.save_graph(state.concepts, state.relationships)

    finally:
        state.metrics.end_time = time.time()
        log_extraction_complete(state.metrics)

    return OntologyOutput(
        concepts=state.concepts,
        relationships=state.relationships,
        metrics=state.metrics,
        errors=state.errors.errors,
    )