webber/implementation-plan.md

Webber FastAPI Boilerplate Plan

Overview

Set up FastAPI boilerplate for "Webber" - a multi-agent AI development system (similar to Claude Code, but local with different models). Follows core-api patterns with defensive coding practices.

Key Decision: PydanticAI Framework After research, PydanticAI is the recommended agent coordination framework:

• Model-agnostic: supports Ollama, OpenAI, Anthropic, and 20+ providers
• Type-safe with Pydantic validation (same ecosystem as FastAPI)
• Built-in tool/function calling with automatic schema generation
• Multi-agent support for complex workflows
• Maintained by Pydantic team (285M+ monthly downloads)

Port: 8086 (next available slot after Headscale 8085 per CONTAINERS.md)

Default Models (always hot in VRAM on tower-of-joy):

• Agent reasoning: mistral-nemo-large:latest
• Embeddings: nomic-embed-text:latest

Target Clients:

• Tatlock Butler: External advisor integration for coding/software guidance
• CLI Interface: TBD - command-line interface for local development

Multi-tenancy: API key authentication integrated with tatlock-ui/core-api user management

---

1. Directory Structure

webber/
├── AGENTS.md                    # Expanded with defensive LLM guidelines
├── README.md                    # Project overview
├── CHANGELOG.md                 # Version history
├── pyproject.toml               # Package metadata
├── requirements.txt             # Production dependencies only (~= pinned)
├── requirements-dev.txt         # Dev/test dependencies (pytest, pip-audit, etc.)
├── .env.example                 # Environment template
├── wakeup.sh                    # Dev startup (update port to 8086)
│
├── src/
│   ├── __init__.py
│   ├── main.py                  # FastAPI app, lifespan, user provider init
│   │                            # NO routes here - delegates to domain routers
│   │
│   ├── shared/                  # Cross-cutting concerns
│   │   ├── __init__.py
│   │   ├── base.py              # BaseController, BaseSchema
│   │   ├── config.py            # Pydantic BaseSettings
│   │   ├── logging.py           # Logger decorator + centralized setup
│   │   ├── exceptions.py        # Custom exception hierarchy
│   │   ├── auth.py              # API key validation, multi-tenant support
│   │   └── context.py           # UserProvider singleton, request context
│   │
│   └── domains/                 # Feature domains (each with router.py)
│       ├── __init__.py
│       ├── router.py            # Root router - includes all domain routers
│       │
│       ├── health/              # Health endpoints
│       │   ├── __init__.py
│       │   ├── router.py        # Health routes
│       │   └── controller.py    # Health logic
│       │
│       ├── auth/                # Authentication domain
│       │   ├── __init__.py
│       │   ├── router.py        # Auth routes (API key mgmt)
│       │   ├── controller.py
│       │   └── schemas.py
│       │
│       │── agents/              # Agent domain container
│       │   ├── __init__.py
│       │   ├── router.py        # Agent routes (lists agents, runs them)
│       │   ├── controller.py    # Agent orchestration logic
│       │   ├── schemas.py
│       │   │
│       │   ├── explore/         # Explore agent (codebase navigation)
│       │   │   ├── __init__.py
│       │   │   ├── agent.py     # PydanticAI agent definition
│       │   │   └── prompts.py   # System prompts
│       │   │
│       │   ├── plan/            # Plan agent (implementation design)
│       │   │   ├── __init__.py
│       │   │   ├── agent.py
│       │   │   └── prompts.py
│       │   │
│       │   └── task/            # Task agent (execution)
│       │       ├── __init__.py
│       │       ├── agent.py
│       │       └── prompts.py
│       │
│       └── tools/               # Tool domain container
│           ├── __init__.py
│           ├── router.py        # Tool routes (list tools, execute)
│           ├── controller.py    # Tool orchestration
│           ├── schemas.py
│           │
│           ├── file/            # File operation tools
│           │   ├── __init__.py
│           │   ├── read.py
│           │   ├── write.py
│           │   └── glob.py
│           │
│           ├── shell/           # Shell execution tools
│           │   ├── __init__.py
│           │   └── bash.py
│           │
│           └── search/          # Search tools
│               ├── __init__.py
│               ├── grep.py
│               └── web.py
│
├── tests/
│   ├── __init__.py
│   ├── conftest.py
│   └── test_health.py
│
└── docs/
    └── architecture.md

Key Architectural Decisions

1. Clean main.py: Only app creation, lifespan, and UserProvider init. All routes in domain routers.
2. Domain routers: Each domain has router.py that defines routes. Root domains/router.py composes them.
3. Separate agent domains: Each agent type (explore, plan, task) in its own subdir under agents/.
4. Separate tool domains: Each tool category (file, shell, search) in its own subdir under tools/.
5. UserProvider singleton: Set once in main.py lifespan, accessible everywhere via shared/context.py.
6. Multi-tenant auth: API key validation in shared/auth.py, integrates with tatlock-ui/core-api.

---

2. Key Files to Create

Phase 1: Foundation (fully implemented)

File	Purpose
src/shared/base.py	BaseController, BaseSchema
src/shared/config.py	Settings via Pydantic BaseSettings
src/shared/logging.py	Logger decorator + centralized setup
src/shared/exceptions.py	Custom exception hierarchy
src/shared/auth.py	API key validation, tatlock integration stub
src/shared/context.py	UserProvider singleton pattern
src/main.py	FastAPI app, lifespan, UserProvider init (no routes!)
src/domains/router.py	Root router composing all domain routers
src/domains/health/router.py	Health routes
src/domains/health/controller.py	Health logic
pyproject.toml	Package metadata, pytest config
requirements.txt	Production deps (~= pinned)
requirements-dev.txt	Dev/test deps (pytest, pip-audit)
.env.example	Environment variable template
tests/conftest.py	Pytest fixtures
tests/test_health.py	Basic endpoint tests

Phase 2: Placeholders (structure + README docs)

Directory	Purpose
src/domains/auth/	API key management (stub)
src/domains/agents/	Agent container with explore/plan/task subdirs
src/domains/tools/	Tool container with file/shell/search subdirs
docs/architecture.md	System design documentation

---

3. Dependency Management

requirements.txt (Production - baked into Docker)

# Webber Production Dependencies
# Minor version pinning (~=) for security patches
# CVE check date: 2026-01-09
# CVE check sources: PyPI, GitHub Advisories, Snyk, NVD

# Core FastAPI
fastapi~=0.115.0
starlette~=0.45.0
uvicorn[standard]~=0.34.0
pydantic~=2.11.0
pydantic-settings~=2.7.0

# Agent Framework
pydantic-ai~=0.0.39            # Multi-agent LLM orchestration

# HTTP
httpx~=0.28.0
aiofiles~=24.1.0

# Utilities
python-multipart~=0.0.18
python-dotenv~=1.0.0

requirements-dev.txt (Dev/Test only - NOT in Docker)

# Webber Development Dependencies
# Install with: pip install -r requirements-dev.txt

-r requirements.txt            # Include production deps

# Testing
pytest~=8.3.0
pytest-asyncio~=0.24.0
pytest-cov~=6.0.0

# Security auditing
pip-audit~=2.7.0               # Run before releases: pip-audit

# Type checking
mypy~=1.13.0

# Code formatting (optional)
# ruff~=0.8.0

---

4. AGENTS.md Additions

Add these new sections:

Section 3: Defensive LLM Coding Practices

• Input validation requirements
• Output parsing guidelines (expect malformed responses)
• Timeout and retry policies
• Security: no secrets in prompts, sandbox execution

Section 4: Pattern Reuse Requirements

• Search existing code before writing new
• Check src/shared/ for base classes
• Follow domain structure template
• Code review checklist

Section 5: CVE Check Process

• Check PyPI, GitHub Advisories, Snyk, NVD before adding deps
• Document CVE decisions in requirements.txt
• Run pip-audit before releases

Section 6: Mandatory Documentation

• Required reading before work: AGENTS.md, docs/architecture.md, src/shared/base.py
• Changelog and docstring requirements

Section 7: Project Structure Reference

• Directory tree with explanations
• Domain structure template

---

5. Configuration (Settings)

Environment variables for:

• App: DEBUG, LOG_LEVEL
• Server: HOST, PORT (default 8086 per CONTAINERS.md allocation)
• CORS: origins, methods, headers
• LLM Models (hot in VRAM on tower-of-joy):
  • OLLAMA_URL (default: http://192.168.86.149:11434)
  • OLLAMA_AGENT_MODEL (default: mistral-nemo-large:latest)
  • OLLAMA_EMBED_MODEL (default: nomic-embed-text:latest)
• Auth:
  • TATLOCK_API_URL (default: http://192.168.86.149:8000)
  • Internal API key for tatlock user validation
• Tools: TOOL_TIMEOUT_SECONDS, SANDBOX_ENABLED, ALLOWED_PATHS
• Sessions: SESSION_TTL_HOURS, MAX_CONTEXT_TOKENS

---

6. Core Patterns

Logger Decorator with Temporal Benchmarking (shared/logging.py)

import functools
import asyncio
import time
import logging
from typing import Callable, Optional
from contextvars import ContextVar
from dataclasses import dataclass, field
from uuid import uuid4

# Trace context for nested timing
@dataclass
class TraceSpan:
    name: str
    trace_id: str
    parent_id: Optional[str] = None
    span_id: str = field(default_factory=lambda: uuid4().hex[:8])
    start_time: float = field(default_factory=time.perf_counter)
    end_time: Optional[float] = None

    @property
    def duration_ms(self) -> float:
        if self.end_time is None:
            return (time.perf_counter() - self.start_time) * 1000
        return (self.end_time - self.start_time) * 1000

# Context variable for trace propagation
_current_span: ContextVar[Optional[TraceSpan]] = ContextVar('current_span', default=None)
_trace_id: ContextVar[Optional[str]] = ContextVar('trace_id', default=None)

def get_current_trace_id() -> Optional[str]:
    """Get current trace ID for correlation."""
    return _trace_id.get()

def logged(
    logger: logging.Logger = None,
    slow_threshold_ms: float = 100.0,
    warn_threshold_ms: float = 500.0,
    include_args: bool = False,
):
    """
    Decorator for automatic function logging with temporal benchmarking.

    Args:
        logger: Logger instance (defaults to module logger)
        slow_threshold_ms: Log INFO if execution exceeds this (default 100ms)
        warn_threshold_ms: Log WARNING if execution exceeds this (default 500ms)
        include_args: Include function arguments in log (careful with sensitive data)

    Usage:
        @logged()
        async def my_function(): ...

        @logged(slow_threshold_ms=50, warn_threshold_ms=200)
        def critical_path(): ...
    """
    def decorator(func: Callable):
        nonlocal logger
        if logger is None:
            logger = logging.getLogger(func.__module__)

        func_name = f"{func.__module__}.{func.__qualname__}"

        def _create_span() -> TraceSpan:
            parent = _current_span.get()
            trace_id = _trace_id.get() or uuid4().hex[:16]
            if _trace_id.get() is None:
                _trace_id.set(trace_id)
            return TraceSpan(
                name=func_name,
                trace_id=trace_id,
                parent_id=parent.span_id if parent else None,
            )

        def _log_completion(span: TraceSpan, error: Exception = None):
            span.end_time = time.perf_counter()
            duration = span.duration_ms

            # Build log context
            ctx = {
                "trace_id": span.trace_id,
                "span_id": span.span_id,
                "duration_ms": round(duration, 2),
                "func": func_name,
            }
            if span.parent_id:
                ctx["parent_id"] = span.parent_id

            if error:
                logger.error(
                    f"[{span.trace_id[:8]}] {func_name} FAILED after {duration:.2f}ms: {error}",
                    extra=ctx,
                    exc_info=True
                )
            elif duration >= warn_threshold_ms:
                logger.warning(
                    f"[{span.trace_id[:8]}] {func_name} SLOW: {duration:.2f}ms (threshold: {warn_threshold_ms}ms)",
                    extra=ctx
                )
            elif duration >= slow_threshold_ms:
                logger.info(
                    f"[{span.trace_id[:8]}] {func_name} completed in {duration:.2f}ms",
                    extra=ctx
                )
            else:
                logger.debug(
                    f"[{span.trace_id[:8]}] {func_name} completed in {duration:.2f}ms",
                    extra=ctx
                )

        @functools.wraps(func)
        async def async_wrapper(*args, **kwargs):
            span = _create_span()
            token = _current_span.set(span)

            if include_args:
                logger.debug(f"[{span.trace_id[:8]}] -> {func_name}({args}, {kwargs})")
            else:
                logger.debug(f"[{span.trace_id[:8]}] -> {func_name}")

            try:
                result = await func(*args, **kwargs)
                _log_completion(span)
                return result
            except Exception as e:
                _log_completion(span, error=e)
                raise
            finally:
                _current_span.reset(token)

        @functools.wraps(func)
        def sync_wrapper(*args, **kwargs):
            span = _create_span()
            token = _current_span.set(span)

            if include_args:
                logger.debug(f"[{span.trace_id[:8]}] -> {func_name}({args}, {kwargs})")
            else:
                logger.debug(f"[{span.trace_id[:8]}] -> {func_name}")

            try:
                result = func(*args, **kwargs)
                _log_completion(span)
                return result
            except Exception as e:
                _log_completion(span, error=e)
                raise
            finally:
                _current_span.reset(token)

        return async_wrapper if asyncio.iscoroutinefunction(func) else sync_wrapper
    return decorator


# Convenience for manual span creation (context manager)
class trace_span:
    """
    Context manager for manual span creation.

    Usage:
        with trace_span("database_query"):
            result = await db.execute(query)

        async with trace_span("llm_call"):
            response = await agent.run(prompt)
    """
    def __init__(self, name: str, logger: logging.Logger = None):
        self.name = name
        self.logger = logger or logging.getLogger(__name__)
        self.span: Optional[TraceSpan] = None
        self.token = None

    def __enter__(self):
        parent = _current_span.get()
        trace_id = _trace_id.get() or uuid4().hex[:16]
        if _trace_id.get() is None:
            _trace_id.set(trace_id)

        self.span = TraceSpan(
            name=self.name,
            trace_id=trace_id,
            parent_id=parent.span_id if parent else None,
        )
        self.token = _current_span.set(self.span)
        self.logger.debug(f"[{self.span.trace_id[:8]}] -> {self.name}")
        return self.span

    def __exit__(self, exc_type, exc_val, exc_tb):
        if self.span:
            self.span.end_time = time.perf_counter()
            duration = self.span.duration_ms
            if exc_val:
                self.logger.error(f"[{self.span.trace_id[:8]}] {self.name} FAILED: {duration:.2f}ms")
            else:
                self.logger.debug(f"[{self.span.trace_id[:8]}] {self.name}: {duration:.2f}ms")
        if self.token:
            _current_span.reset(self.token)
        return False

    async def __aenter__(self):
        return self.__enter__()

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        return self.__exit__(exc_type, exc_val, exc_tb)

Example output:

DEBUG [a1b2c3d4] -> src.domains.agents.controller.run_agent
DEBUG [a1b2c3d4] -> src.domains.llm.service.call_ollama
DEBUG [a1b2c3d4] src.domains.llm.service.call_ollama: 45.23ms
INFO  [a1b2c3d4] src.domains.agents.controller.run_agent completed in 156.78ms
WARN  [a1b2c3d4] src.domains.tools.file.read.read_file SLOW: 523.45ms (threshold: 500ms)

Features:

• Trace IDs: Correlate logs across nested calls
• Parent/child spans: Track call hierarchy
• Configurable thresholds: slow_threshold_ms (INFO), warn_threshold_ms (WARNING)
• Context manager: trace_span() for manual instrumentation of code blocks
• Zero overhead path: Fast path for sub-threshold calls (DEBUG only)

UserProvider Singleton (shared/context.py)

from dataclasses import dataclass
from typing import Optional
from contextvars import ContextVar

@dataclass
class User:
    id: str
    email: str
    api_key: str
    tenant_id: Optional[str] = None

# Context variable for request-scoped user
_current_user: ContextVar[Optional[User]] = ContextVar('current_user', default=None)

class UserProvider:
    """Singleton for user context management."""
    _instance = None

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance

    def set_user(self, user: User) -> None:
        _current_user.set(user)

    def get_user(self) -> Optional[User]:
        return _current_user.get()

    def clear_user(self) -> None:
        _current_user.set(None)

# Global singleton
user_provider = UserProvider()

BaseController (from core-api)

class BaseController(ABC):
    def __init__(self, prefix: str, tags: list[str]):
        self.prefix = prefix
        self.tags = tags
        self._router = None

    @abstractmethod
    def create_router(self) -> APIRouter: pass

    @property
    def router(self) -> APIRouter:
        if self._router is None:
            self._router = self.create_router()
        return self._router

PydanticAI Agent Pattern (placeholder for future)

from pydantic_ai import Agent
from pydantic_ai.models.ollama import OllamaModel

# Use the hot model from VRAM
agent = Agent(
    OllamaModel('mistral-nemo-large:latest'),
    system_prompt='You are a helpful assistant.',
)

@agent.tool
async def search_files(ctx, pattern: str) -> str:
    """Search for files matching pattern."""
    pass  # Implementation in tools/search/

---

7. Implementation Order

1. Create directory structure (src/, src/shared/, src/domains/)
2. Implement shared modules (base.py, config.py, logging.py, exceptions.py)
3. Create main.py with FastAPI app and lifespan
4. Add health domain as working example
5. Set up tests (conftest.py, test_health.py)
6. Create placeholder domains (llm, agents, tools - structure only)
7. Update AGENTS.md with new sections
8. Create supporting files (pyproject.toml, requirements.txt, .env.example)
9. Add docs/architecture.md

---

8. Verification

After implementation:

1. ./wakeup.sh starts server without errors
2. curl http://localhost:8086/health returns healthy
3. http://localhost:8086/docs shows API documentation
4. .venv/bin/python -m pytest tests/ -v passes
5. Code follows patterns in AGENTS.md

---

9. Critical Reference Files

• /mnt/media/Projects/core-api/src/shared/base.py - BaseController pattern
• /mnt/media/Projects/core-api/src/shared/config.py - Settings pattern
• /mnt/media/Projects/core-api/src/domains/health/controller.py - Controller example
• https://ai.pydantic.dev/ - PydanticAI documentation

---

Summary

What will be created:

• Complete FastAPI project structure following core-api patterns
• Working health endpoint at http://localhost:8086/health
• Clean main.py - no routes, just app init and UserProvider setup
• Domain routers - each domain has router.py, composed by root router
• Logger decorator - centralized logging via @logged decorator
• UserProvider singleton - request-scoped user context, no parameter passing
• Multi-tenant auth stub - API key validation ready for tatlock integration
• Separate requirements.txt (prod) and requirements-dev.txt (dev/test)
• Placeholder domains with agent/tool subdirectories (explore, plan, task / file, shell, search)
• Comprehensive AGENTS.md with defensive LLM coding practices, CVE checks, pattern reuse
• Test infrastructure with pytest
• docs/architecture.md explaining the system design

What will NOT be created (deferred):

• Database layer (add when needed)
• Full agent/tool implementations (PydanticAI patterns documented for future work)
• Docker/deployment configuration (can add later)
• CLI interface (TBD)

Key decisions:

• Port: 8086
• Agent framework: PydanticAI
• Default model: mistral-nemo-large:latest (hot in VRAM)
• Embeddings: nomic-embed-text:latest (hot in VRAM)
• No database initially
• Separate prod/dev requirements
• UserProvider singleton pattern for multi-tenancy