ARCHONRELOADED

Imported from JackSmack1971/ARCHONRELOADED (AGENTS.md). Install upstream with npx skills add JackSmack1971/ARCHONRELOADED. Copyright stays with the author.

AGENTS.md: Comprehensive AI Collaboration Guide for ARCHON RELOADED

This document provides essential context and technical knowledge for AI models interacting with the ARCHON RELOADED platform. Adhering to these guidelines will ensure consistency, maintain code quality, and optimize agent performance across all aspects of the system.

Current Date: Sunday, August 17, 2025
Optimized for: OpenAI Codex, Claude, GitHub Copilot Workspace, and other modern AI coding agents
Framework Compatibility: Supports MCP (Model Context Protocol) integration

1. Project Overview & Architecture

Primary Mission

ARCHON RELOADED is a next-generation microservices-based AI development platform that integrates with AI coding tools through the Model Context Protocol (MCP). It provides intelligent knowledge management, real-time collaboration, and RAG (Retrieval-Augmented Generation) capabilities for AI-powered development workflows.

Business Domain

• Primary: AI Development Tools & Developer Productivity
• Secondary: Knowledge Management & AI-Assisted Software Engineering
• Tertiary: Real-time Collaboration & Document Processing

Core System Features

• MCP Integration: Seamless connectivity with Claude Code, Cursor, and Windsurf
• Knowledge Management: Real-time knowledge base with vector search capabilities
• Project Management: Task and project management with AI agent collaboration
• Document Processing: RAG pipeline with parallel processing and embedding generation
• Real-time Features: WebSocket-based updates and live collaboration
• Microservices Architecture: Independent, scalable services with clear boundaries

System Architecture Principles

• Cloud-native: Container-ready microservices with Kubernetes support
• Event-driven: Asynchronous communication via WebSocket and message queues
• API-first: RESTful design with comprehensive OpenAPI documentation
• Type-safe: Full TypeScript and Pydantic model validation
• Scalable: Horizontal scaling with proper load balancing and caching

2. Technology Stack & Dependencies

Core Technologies

• Backend Languages: Python 3.12+ (primary), TypeScript 5.x, JavaScript ES2023
• Frontend Framework: React 18.x with TypeScript and modern hooks
• Build Tools: Vite 6.x for frontend, uv for Python package management
• Runtime Environments: Node.js 20.x, Python 3.12+

Framework Stack

• Backend API: FastAPI with Socket.IO for real-time features
• AI Agents: PydanticAI for intelligent agent development
• MCP Protocol: HTTP-based MCP server implementation
• Frontend Build: Vite + TypeScript + React with hot module replacement
• Documentation: Docusaurus for comprehensive project documentation

Database & Storage

• Primary Database: Supabase (PostgreSQL with pgvector extension)
• Vector Storage: pgvector for embeddings and similarity search
• Caching Layer: Redis for session management and real-time features
• File Storage: Supabase Storage for document and media assets

Container & Orchestration

• Containerization: Docker with multi-stage builds
• Orchestration: Docker Compose for development, Kubernetes-ready for production
• Service Mesh: Internal communication via HTTP/WebSocket APIs
• Load Balancing: nginx or cloud load balancer for production

Package Management

• Python: uv (modern, fast package manager)
• JavaScript/TypeScript: npm (consistent with Node.js ecosystem)
• Container: Docker with optimized layer caching

3. Project Structure & Organization

Repository Architecture

ARCHON-RELOADED/
├── python/                     # Backend services (Python)
│   ├── src/
│   │   ├── server/            # FastAPI main application
│   │   │   ├── main.py        # Server entry point
│   │   │   ├── routes/        # API route handlers
│   │   │   ├── services/      # Business logic layer
│   │   │   ├── models/        # Pydantic data models
│   │   │   └── config/        # Configuration management
│   │   ├── mcp/               # MCP server implementation
│   │   │   ├── server.py      # MCP protocol server
│   │   │   ├── tools/         # MCP tool definitions
│   │   │   └── handlers/      # Request handlers
│   │   ├── agents/            # PydanticAI agent services
│   │   │   ├── server.py      # Agent service entry
│   │   │   ├── agents/        # Individual agent definitions
│   │   │   ├── tools/         # Agent tool implementations
│   │   │   └── workflows/     # Multi-agent workflows
│   │   └── shared/            # Shared utilities and types
│   ├── tests/                 # Comprehensive test suite
│   │   ├── server/            # API server tests
│   │   ├── mcp/               # MCP protocol tests
│   │   ├── agents/            # Agent behavior tests
│   │   └── integration/       # End-to-end tests
│   ├── migration/             # Database schema migrations
│   └── pyproject.toml         # Python dependencies & config
├── archon-ui-main/            # Frontend application (React)
│   ├── src/
│   │   ├── components/        # Reusable UI components
│   │   │   ├── atoms/         # Basic components (buttons, inputs)
│   │   │   ├── molecules/     # Composite components (forms, cards)
│   │   │   └── organisms/     # Complex components (layouts, dashboards)
│   │   ├── features/          # Feature-specific components
│   │   ├── hooks/             # Custom React hooks
│   │   ├── services/          # API communication layer
│   │   ├── stores/            # State management (Redux/Zustand)
│   │   ├── types/             # TypeScript type definitions
│   │   ├── utils/             # Utility functions
│   │   └── main.tsx           # Application entry point
│   ├── test/                  # Frontend test suite
│   ├── public/                # Static assets
│   ├── package.json           # Dependencies & scripts
│   ├── vite.config.ts         # Vite build configuration
│   └── vitest.config.ts       # Test configuration
├── docs/                      # Docusaurus documentation
│   ├── docs/                  # Documentation content
│   ├── blog/                  # Project blog posts
│   ├── src/                   # Custom documentation components
│   └── docusaurus.config.js   # Documentation site config
├── docker-compose.yml         # Multi-container development setup
├── .env.example               # Environment variable template
└── README.md                  # Project overview & setup

Module Organization Principles

• Service Separation: Each service (server, mcp, agents) operates independently
• Layered Architecture: Clear separation between routes, services, and data layers
• Feature Organization: Frontend organized by features rather than technical concerns
• Shared Resources: Common utilities and types in dedicated shared modules
• Test Colocation: Tests organized to mirror source code structure

4. Development Environment & Workflow

Local Development Setup

Prerequisites

1. Python 3.12+ with uv package manager
2. Node.js 20.x with npm
3. Docker & Docker Compose for containerized services
4. Git for version control

Initial Setup Commands

# 1. Clone and navigate to project
git clone <repository-url>
cd ARCHON-RELOADED

# 2. Backend setup (Python)
cd python
uv sync --frozen --all-extras --dev
# Creates virtual environment and installs all dependencies

# 3. Frontend setup (React)
cd ../archon-ui-main
npm install
# Installs Node.js dependencies

# 4. Environment configuration
cp .env.example .env
# Edit .env with your Supabase and OpenAI credentials

# 5. Database setup
docker compose up -d supabase
# Wait for database to be ready, then run migrations

# 6. Start all services
docker compose up -d

Development Commands

# Backend development
cd python
uv run python -m src.server.main          # FastAPI server
uv run python -m src.mcp.server           # MCP server
uv run python -m src.agents.server        # PydanticAI agents

# Frontend development
cd archon-ui-main
npm run dev                                # Vite dev server with HMR

# Testing
uv run pytest                             # Python tests
npm run test                              # Frontend tests (Vitest)

# Code quality
uv run ruff format .                      # Python formatting
uv run ruff check . --fix                # Python linting
npm run lint                              # TypeScript linting
npm run type-check                        # TypeScript compilation check

Development Workflow

1. Feature Branches: Create feature branches from main
2. Local Testing: Run all tests before committing
3. Code Quality: Ensure formatting and linting pass
4. Documentation: Update relevant documentation
5. Pull Requests: Create PR with comprehensive description

Docker Development Environment

# docker-compose.yml structure
services:
  # Database services
  supabase:
    image: supabase/supabase:latest
    environment:
      - POSTGRES_PASSWORD=${DB_PASSWORD}
    ports:
      - "5432:5432"
    volumes:
      - postgres_data:/var/lib/postgresql/data

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"

  # Backend services
  api-server:
    build:
      context: ./python
      dockerfile: Dockerfile
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=${DATABASE_URL}
      - REDIS_URL=${REDIS_URL}
    depends_on:
      - supabase
      - redis

  mcp-server:
    build:
      context: ./python
      dockerfile: Dockerfile.mcp
    ports:
      - "8001:8001"

  # Frontend service
  frontend:
    build:
      context: ./archon-ui-main
      dockerfile: Dockerfile
    ports:
      - "3000:3000"
    environment:
      - VITE_API_URL=http://localhost:8000

5. Coding Standards & Conventions

General Principles

• Type Safety First: Strict typing in both TypeScript and Python
• Immutable Patterns: Prefer immutable data structures and pure functions
• Error Handling: Comprehensive error handling with proper logging
• Performance Conscious: Consider performance implications of all architectural decisions
• Security Minded: Validate all inputs, sanitize outputs, secure by default

Python Coding Standards

Code Style & Formatting

• Formatter: Black with 100-character line length
• Linter: Ruff with comprehensive rule set
• Import Sorting: isort integrated with Ruff
• Type Checking: mypy or pyright for static analysis

# Style example
from typing import Annotated
from pydantic import BaseModel, Field
from fastapi import Depends, HTTPException, status

class UserCreateRequest(BaseModel):
    """User creation request model with validation."""
    username: Annotated[str, Field(min_length=3, max_length=50)]
    email: Annotated[str, Field(regex=r'^[^\s@]+@[^\s@]+\.[^\s@]+$')]
    password: Annotated[str, Field(min_length=8)]

async def create_user(
    request: UserCreateRequest,
    db: Annotated[Database, Depends(get_database)],
) -> UserResponse:
    """Create a new user with validation and error handling."""
    try:
        # Business logic here
        user = await db.create_user(request)
        return UserResponse.from_orm(user)
    except IntegrityError as e:
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail="Username or email already exists"
        ) from e

Naming Conventions

• Variables/Functions: snake_case
• Classes: PascalCase
• Constants: SCREAMING_SNAKE_CASE
• Private Members: _leading_underscore
• Files: snake_case.py

FastAPI Specific Patterns

• Dependency Injection: Use FastAPI's dependency system extensively
• Pydantic Models: All request/response models must use Pydantic
• Async/Await: Use async for all I/O operations
• Error Handling: Standardized HTTPException usage
• OpenAPI Documentation: Comprehensive docstrings and examples

# FastAPI best practices
from fastapi import FastAPI, Depends, BackgroundTasks
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

app = FastAPI(
    title="ARCHON RELOADED API",
    description="AI Development Platform API",
    version="1.0.0",
    docs_url="/docs",
    redoc_url="/redoc"
)

security = HTTPBearer()

async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security)
) -> User:
    """Extract and validate user from JWT token."""
    token = credentials.credentials
    payload = decode_jwt(token)
    return await get_user_by_id(payload["user_id"])

@app.post("/api/v1/tasks", response_model=TaskResponse)
async def create_task(
    task_data: TaskCreateRequest,
    background_tasks: BackgroundTasks,
    current_user: User = Depends(get_current_user)
) -> TaskResponse:
    """Create a new task with background processing."""
    task = await create_task_service(task_data, current_user.id)
    background_tasks.add_task(send_task_notification, task.id)
    return TaskResponse.from_orm(task)

PydanticAI Patterns

• Agent Definition: Use typed dependencies for better IDE support
• Tool Implementation: Comprehensive docstrings for LLM understanding
• Error Handling: Use ModelRetry for recoverable errors
• Output Validation: Structured output types with Pydantic models

# PydanticAI best practices
from pydantic_ai import Agent, RunContext, ModelRetry
from dataclasses import dataclass

@dataclass
class AgentDependencies:
    db_conn: DatabaseConnection
    api_client: HttpClient

agent = Agent(
    'openai:gpt-4o',
    deps_type=AgentDependencies,
    system_prompt="You are a helpful AI assistant for the ARCHON platform."
)

@agent.tool
async def search_knowledge_base(
    ctx: RunContext[AgentDependencies], 
    query: str,
    max_results: int = 10
) -> list[str]:
    """Search the knowledge base for relevant information.
    
    Args:
        query: Search query string
        max_results: Maximum number of results to return (1-50)
        
    Returns:
        List of relevant knowledge base entries
    """
    try:
        results = await ctx.deps.db_conn.vector_search(query, max_results)
        return [result.content for result in results]
    except DatabaseError as e:
        raise ModelRetry(f"Database search failed: {e}")

TypeScript/React Coding Standards

Code Style & Formatting

• Formatter: Prettier with 2-space indentation
• Linter: ESLint with TypeScript rules
• Import Organization: Automatic import sorting
• Naming Convention: camelCase for variables/functions, PascalCase for components/types

// TypeScript style example
import React, { useState, useCallback, useMemo } from 'react';
import { useQuery, useMutation } from '@tanstack/react-query';
import { Button } from '@/components/atoms/Button';
import { TaskCard } from '@/components/molecules/TaskCard';
import type { Task, CreateTaskRequest } from '@/types/task';

interface TaskListProps {
  userId: string;
  onTaskCreate?: (task: Task) => void;
}

export const TaskList: React.FC<TaskListProps> = ({ userId, onTaskCreate }) => {
  const [filter, setFilter] = useState<'all' | 'active' | 'completed'>('all');

  const { data: tasks, isLoading, error } = useQuery({
    queryKey: ['tasks', userId, filter],
    queryFn: () => fetchTasks(userId, filter),
    staleTime: 30_000, // 30 seconds
  });

  const createTaskMutation = useMutation({
    mutationFn: (request: CreateTaskRequest) => createTask(request),
    onSuccess: (task) => {
      onTaskCreate?.(task);
      // Invalidate and refetch tasks
      queryClient.invalidateQueries(['tasks', userId]);
    },
  });

  const handleCreateTask = useCallback((taskData: CreateTaskRequest) => {
    createTaskMutation.mutate(taskData);
  }, [createTaskMutation]);

  const filteredTasks = useMemo(() => {
    if (!tasks) return [];
    return tasks.filter(task => {
      switch (filter) {
        case 'active': return !task.completed;
        case 'completed': return task.completed;
        default: return true;
      }
    });
  }, [tasks, filter]);

  if (isLoading) return <div>Loading tasks...</div>;
  if (error) return <div>Error loading tasks: {error.message}</div>;

  return (
    <div className="task-list">
      <div className="task-filters">
        {(['all', 'active', 'completed'] as const).map(filterOption => (
          <Button
            key={filterOption}
            variant={filter === filterOption ? 'primary' : 'secondary'}
            onClick={() => setFilter(filterOption)}
          >
            {filterOption}
          </Button>
        ))}
      </div>
      
      <div className="task-grid">
        {filteredTasks.map(task => (
          <TaskCard
            key={task.id}
            task={task}
            onUpdate={handleTaskUpdate}
          />
        ))}
      </div>
    </div>
  );
};

React 18 Patterns

• Concurrent Features: Use useTransition for non-urgent updates
• Performance: Strategic use of React.memo and useMemo
• Error Boundaries: Implement for robust error handling
• Suspense: Use for code splitting and async components

// React 18 concurrent features
import { useTransition, useDeferredValue, startTransition } from 'react';

export const SearchResults: React.FC = () => {
  const [isPending, startTransition] = useTransition();
  const [query, setQuery] = useState('');
  const [results, setResults] = useState([]);
  
  const deferredQuery = useDeferredValue(query);

  const handleSearch = (newQuery: string) => {
    setQuery(newQuery); // Urgent update
    startTransition(() => {
      // Non-urgent update that won't block UI
      searchKnowledgeBase(newQuery).then(setResults);
    });
  };

  return (
    <div>
      <SearchInput onChange={handleSearch} />
      {isPending && <LoadingSpinner />}
      <ResultsList results={results} query={deferredQuery} />
    </div>
  );
};

Socket.IO Integration

• Connection Management: Proper connection lifecycle handling
• Event Typing: Strong typing for all Socket.IO events
• Error Handling: Comprehensive error and reconnection logic
• Performance: Event throttling and cleanup

// Socket.IO client integration
import { io, Socket } from 'socket.io-client';
import { useEffect, useState, useCallback } from 'react';

interface ServerToClientEvents {
  'task:updated': (task: Task) => void;
  'task:created': (task: Task) => void;
  'user:joined': (user: User) => void;
}

interface ClientToServerEvents {
  'task:subscribe': (taskId: string) => void;
  'task:unsubscribe': (taskId: string) => void;
}

export const useSocket = () => {
  const [socket, setSocket] = useState<Socket<ServerToClientEvents, ClientToServerEvents> | null>(null);
  const [isConnected, setIsConnected] = useState(false);

  useEffect(() => {
    const socketInstance = io('http://localhost:8000', {
      transports: ['websocket', 'polling'],
      auth: {
        token: getAuthToken(),
      },
    });

    socketInstance.on('connect', () => {
      setIsConnected(true);
      console.log('Connected to server');
    });

    socketInstance.on('disconnect', (reason) => {
      setIsConnected(false);
      console.log('Disconnected:', reason);
    });

    socketInstance.on('connect_error', (error) => {
      console.error('Connection error:', error);
    });

    setSocket(socketInstance);

    return () => {
      socketInstance.disconnect();
    };
  }, []);

  const subscribeToTask = useCallback((taskId: string) => {
    socket?.emit('task:subscribe', taskId);
  }, [socket]);

  return {
    socket,
    isConnected,
    subscribeToTask,
  };
};

Vite Configuration

• Modern Build Target: ES2023 for modern browsers
• Optimization: Tree shaking and code splitting
• Development: Hot module replacement and fast refresh
• Environment: Proper environment variable handling

// vite.config.ts
import { defineConfig } from 'vite';
import react from '@vitejs/plugin-react';
import path from 'path';

export default defineConfig({
  plugins: [
    react({
      // Use SWC for faster builds
      jsxRuntime: 'automatic',
    }),
  ],
  
  build: {
    target: 'ES2023',
    sourcemap: process.env.NODE_ENV === 'development',
    rollupOptions: {
      output: {
        manualChunks: {
          vendor: ['react', 'react-dom'],
          utils: ['lodash-es', 'date-fns'],
        },
      },
    },
  },

  resolve: {
    alias: {
      '@': path.resolve(__dirname, 'src'),
      '@components': path.resolve(__dirname, 'src/components'),
      '@types': path.resolve(__dirname, 'src/types'),
    },
  },

  server: {
    port: 3000,
    host: true,
    proxy: {
      '/api': {
        target: 'http://localhost:8000',
        changeOrigin: true,
      },
      '/socket.io': {
        target: 'http://localhost:8000',
        ws: true,
      },
    },
  },

  define: {
    __APP_VERSION__: JSON.stringify(process.env.npm_package_version),
  },
});

6. Database Architecture & Vector Operations

Supabase PostgreSQL Setup

Core Configuration

• Database: PostgreSQL 15+ with pgvector extension
• Vector Dimensions: 384 (gte-small), 1536 (OpenAI ada-002), 4096 (large models)
• Extensions: pgvector, pg_net, pg_cron, hstore for AI workflows
• Connection Pooling: Session pooler (port 5432) for migrations, transaction pooler (port 6543) for queries

Vector Table Design

-- Knowledge base documents with vector embeddings
CREATE TABLE documents (
  id BIGINT PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
  title TEXT NOT NULL,
  content TEXT NOT NULL,
  document_type VARCHAR(50) NOT NULL,
  category_id BIGINT REFERENCES categories(id),
  project_id BIGINT REFERENCES projects(id),
  created_at TIMESTAMPTZ DEFAULT NOW(),
  updated_at TIMESTAMPTZ DEFAULT NOW(),
  embedding vector(384), -- Using gte-small embeddings
  metadata JSONB DEFAULT '{}'::jsonb,
  
  -- Indexing for performance
  CONSTRAINT documents_title_length CHECK (length(title) > 0),
  CONSTRAINT documents_content_length CHECK (length(content) > 0)
);

-- Vector similarity index (HNSW for best performance)
CREATE INDEX documents_embedding_idx ON documents 
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);

-- Additional indexes for filtering
CREATE INDEX documents_category_idx ON documents (category_id);
CREATE INDEX documents_project_idx ON documents (project_id);
CREATE INDEX documents_type_idx ON documents (document_type);
CREATE INDEX documents_created_idx ON documents (created_at DESC);

Vector Search Functions

-- Similarity search with filtering
CREATE OR REPLACE FUNCTION match_documents (
  query_embedding vector(384),
  match_threshold float DEFAULT 0.7,
  match_count int DEFAULT 10,
  filter_project_id bigint DEFAULT null,
  filter_category_id bigint DEFAULT null
)
RETURNS TABLE (
  id bigint,
  title text,
  content text,
  document_type varchar(50),
  similarity float,
  metadata jsonb
)
LANGUAGE sql STABLE
AS $$
  SELECT
    documents.id,
    documents.title,
    documents.content,
    documents.document_type,
    1 - (documents.embedding <=> query_embedding) as similarity,
    documents.metadata
  FROM documents
  WHERE 
    (filter_project_id IS NULL OR project_id = filter_project_id)
    AND (filter_category_id IS NULL OR category_id = filter_category_id)
    AND 1 - (documents.embedding <=> query_embedding) > match_threshold
  ORDER BY (documents.embedding <=> query_embedding) ASC
  LIMIT match_count;
$$;

-- Hybrid search combining text and vector similarity
CREATE OR REPLACE FUNCTION hybrid_search (
  query_text text,
  query_embedding vector(384),
  match_count int DEFAULT 10,
  text_weight float DEFAULT 0.3,
  vector_weight float DEFAULT 0.7
)
RETURNS TABLE (
  id bigint,
  title text,
  content text,
  combined_score float
)
LANGUAGE sql STABLE
AS $$
  WITH text_search AS (
    SELECT 
      id,
      title,
      content,
      ts_rank_cd(to_tsvector('english', title || ' ' || content), plainto_tsquery('english', query_text)) as text_score
    FROM documents
    WHERE to_tsvector('english', title || ' ' || content) @@ plainto_tsquery('english', query_text)
  ),
  vector_search AS (
    SELECT 
      id,
      title,
      content,
      1 - (embedding <=> query_embedding) as vector_score
    FROM documents
    ORDER BY embedding <=> query_embedding
    LIMIT match_count * 2
  )
  SELECT 
    COALESCE(ts.id, vs.id) as id,
    COALESCE(ts.title, vs.title) as title,
    COALESCE(ts.content, vs.content) as content,
    (COALESCE(ts.text_score, 0) * text_weight + COALESCE(vs.vector_score, 0) * vector_weight) as combined_score
  FROM text_search ts
  FULL OUTER JOIN vector_search vs ON ts.id = vs.id
  ORDER BY combined_score DESC
  LIMIT match_count;
$$;

Row Level Security (RLS)

-- Enable RLS on documents table
ALTER TABLE documents ENABLE ROW LEVEL SECURITY;

-- Users can only access documents from their projects
CREATE POLICY "Users can view own project documents"
  ON documents FOR SELECT
  USING (
    project_id IN (
      SELECT project_id 
      FROM project_members 
      WHERE user_id = (SELECT auth.uid())
    )
  );

-- Optimized RLS policy with caching
CREATE POLICY "Users can modify own project documents"
  ON documents FOR ALL
  USING ((SELECT auth.uid()) = created_by);

-- Index for RLS performance
CREATE INDEX documents_created_by_idx ON documents (created_by);

Database Migration Strategy

-- migration/001_initial_schema.sql
-- Core tables for the ARCHON platform

-- Users and authentication
CREATE TABLE users (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  email TEXT UNIQUE NOT NULL,
  username TEXT UNIQUE NOT NULL,
  full_name TEXT,
  avatar_url TEXT,
  created_at TIMESTAMPTZ DEFAULT NOW(),
  updated_at TIMESTAMPTZ DEFAULT NOW()
);

-- Projects and workspaces
CREATE TABLE projects (
  id BIGINT PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
  name TEXT NOT NULL,
  description TEXT,
  owner_id UUID REFERENCES users(id) ON DELETE CASCADE,
  settings JSONB DEFAULT '{}'::jsonb,
  created_at TIMESTAMPTZ DEFAULT NOW(),
  updated_at TIMESTAMPTZ DEFAULT NOW()
);

-- Tasks and project management
CREATE TABLE tasks (
  id BIGINT PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
  title TEXT NOT NULL,
  description TEXT,
  status task_status DEFAULT 'pending',
  priority task_priority DEFAULT 'medium',
  project_id BIGINT REFERENCES projects(id) ON DELETE CASCADE,
  assigned_to UUID REFERENCES users(id),
  created_by UUID REFERENCES users(id) NOT NULL,
  due_date TIMESTAMPTZ,
  completed_at TIMESTAMPTZ,
  created_at TIMESTAMPTZ DEFAULT NOW(),
  updated_at TIMESTAMPTZ DEFAULT NOW()
);

-- Knowledge base categories
CREATE TABLE categories (
  id BIGINT PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
  name TEXT NOT NULL,
  description TEXT,
  project_id BIGINT REFERENCES projects(id) ON DELETE CASCADE,
  parent_id BIGINT REFERENCES categories(id),
  created_at TIMESTAMPTZ DEFAULT NOW()
);

-- Document processing and embeddings
CREATE TABLE document_chunks (
  id BIGINT PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
  document_id BIGINT REFERENCES documents(id) ON DELETE CASCADE,
  chunk_index INTEGER NOT NULL,
  content TEXT NOT NULL,
  embedding vector(384),
  token_count INTEGER,
  created_at TIMESTAMPTZ DEFAULT NOW(),
  
  UNIQUE(document_id, chunk_index)
);

-- Vector index for chunks
CREATE INDEX document_chunks_embedding_idx ON document_chunks 
USING hnsw (embedding vector_cosine_ops);

7. API Design & Communication Patterns

RESTful API Standards

• Base URL: /api/v1/ for all API endpoints
• HTTP Methods: Proper usage of GET, POST, PUT, PATCH, DELETE
• Status Codes: Comprehensive use of appropriate HTTP status codes
• Content Type: JSON for all request/response bodies
• Error Format: Consistent error response structure

API Endpoint Structure

# Standard API response models
from pydantic import BaseModel
from typing import Generic, TypeVar, Optional

T = TypeVar('T')

class APIResponse(BaseModel, Generic[T]):
    """Standard API response wrapper."""
    success: bool
    data: Optional[T] = None
    error: Optional[str] = None
    message: Optional[str] = None
    pagination: Optional[PaginationInfo] = None

class PaginationInfo(BaseModel):
    """Pagination metadata."""
    page: int
    limit: int
    total: int
    total_pages: int
    has_next: bool
    has_prev: bool

# Example endpoint implementation
@router.get("/documents", response_model=APIResponse[list[DocumentResponse]])
async def list_documents(
    page: int = Query(1, ge=1),
    limit: int = Query(20, ge=1, le=100),
    category_id: Optional[int] = None,
    search: Optional[str] = None,
    current_user: User = Depends(get_current_user),
    db: Database = Depends(get_database)
) -> APIResponse[list[DocumentResponse]]:
    """List documents with pagination and filtering."""
    
    # Build query with filters
    query_params = {
        "user_id": current_user.id,
        "category_id": category_id,
        "search": search
    }
    
    # Get paginated results
    documents, total_count = await db.get_documents_paginated(
        offset=(page - 1) * limit,
        limit=limit,
        **query_params
    )
    
    # Build response
    return APIResponse(
        success=True,
        data=[DocumentResponse.from_orm(doc) for doc in documents],
        pagination=PaginationInfo(
            page=page,
            limit=limit,
            total=total_count,
            total_pages=math.ceil(total_count / limit),
            has_next=page * limit < total_count,
            has_prev=page > 1
        )
    )

Error Handling Standards

# Custom exception classes
class APIException(HTTPException):
    """Base API exception with logging."""
    
    def __init__(self, status_code: int, detail: str, error_code: str = None):
        super().__init__(status_code=status_code, detail=detail)
        self.error_code = error_code
        logger.error(f"API Error: {error_code} - {detail}")

class ValidationException(APIException):
    """Input validation errors."""
    
    def __init__(self, detail: str, field: str = None):
        super().__init__(
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            detail=detail,
            error_code="VALIDATION_ERROR"
        )
        self.field = field

class ResourceNotFoundException(APIException):
    """Resource not found errors."""
    
    def __init__(self, resource_type: str, resource_id: str):
        super().__init__(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"{resource_type} with id {resource_id} not found",
            error_code="RESOURCE_NOT_FOUND"
        )

# Global exception handler
@app.exception_handler(APIException)
async def api_exception_handler(request: Request, exc: APIException):
    """Handle API exceptions with consistent format."""
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "success": False,
            "error": exc.detail,
            "error_code": exc.error_code,
            "path": str(request.url.path),
            "timestamp": datetime.utcnow().isoformat()
        }
    )

WebSocket/Socket.IO Integration

• Namespace Organization: Logical separation by feature
• Event Naming: Consistent domain:action pattern
• Authentication: JWT-based authentication for all connections
• Error Handling: Comprehensive error and reconnection handling

Socket.IO Server Configuration

# Socket.IO server setup with FastAPI
from socketio import AsyncServer
from fastapi import FastAPI
import socketio

# Create Socket.IO server
sio = AsyncServer(
    cors_allowed_origins=["http://localhost:3000"],
    logger=True,
    engineio_logger=True
)

app = FastAPI()

# Mount Socket.IO
socket_app = socketio.ASGIApp(sio, app)

# Authentication middleware
@sio.event
async def connect(sid, environ, auth):
    """Handle client connection with authentication."""
    try:
        token = auth.get('token') if auth else None
        if not token:
            await sio.emit('error', {'message': 'Authentication required'}, room=sid)
            return False
            
        user = await verify_jwt_token(token)
        if not user:
            await sio.emit('error', {'message': 'Invalid token'}, room=sid)
            return False
            
        # Store user session
        await sio.save_session(sid, {'user_id': user.id, 'username': user.username})
        
        # Join user to personal room
        await sio.enter_room(sid, f"user:{user.id}")
        
        logger.info(f"User {user.username} connected: {sid}")
        return True
        
    except Exception as e:
        logger.error(f"Connection error: {e}")
        return False

# Event handlers
@sio.event
async def join_project(sid, data):
    """Join a project room for real-time updates."""
    session = await sio.get_session(sid)
    user_id = session['user_id']
    project_id = data.get('project_id')
    
    # Verify user has access to project
    if await has_project_access(user_id, project_id):
        await sio.enter_room(sid, f"project:{project_id}")
        await sio.emit('project:joined', {'project_id': project_id}, room=sid)
        
        # Notify other project members
        await sio.emit('user:joined_project', {
            'user_id': user_id,
            'username': session['username'],
            'project_id': project_id
        }, room=f"project:{project_id}", skip_sid=sid)

@sio.event
async def task_update(sid, data):
    """Handle task updates with real-time broadcast."""
    session = await sio.get_session(sid)
    user_id = session['user_id']
    
    task_id = data.get('task_id')
    updates = data.get('updates', {})
    
    try:
        # Update task in database
        task = await update_task(task_id, updates, user_id)
        
        # Broadcast to project members
        await sio.emit('task:updated', {
            'task_id': task.id,
            'task': task.dict(),
            'updated_by': session['username'],
            'timestamp': datetime.utcnow().isoformat()
        }, room=f"project:{task.project_id}")
        
    except Exception as e:
        await sio.emit('error', {'message': str(e)}, room=sid)

@sio.event
async def disconnect(sid):
    """Handle client disconnection."""
    session = await sio.get_session(sid)
    if session:
        logger.info(f"User {session.get('username')} disconnected: {sid}")

Client-Side Socket Integration

// Socket.IO client with React integration
import { useEffect, useState, useContext } from 'react';
import { io, Socket } from 'socket.io-client';
import { AuthContext } from '@/contexts/AuthContext';

interface SocketContextType {
  socket: Socket | null;
  isConnected: boolean;
  joinProject: (projectId: string) => void;
  leaveProject: (projectId: string) => void;
}

export const SocketContext = createContext<SocketContextType>({
  socket: null,
  isConnected: false,
  joinProject: () => {},
  leaveProject: () => {},
});

export const SocketProvider: React.FC<{ children: React.ReactNode }> = ({ children }) => {
  const [socket, setSocket] = useState<Socket | null>(null);
  const [isConnected, setIsConnected] = useState(false);
  const { user, token } = useContext(AuthContext);

  useEffect(() => {
    if (!user || !token) return;

    const socketInstance = io('http://localhost:8000', {
      auth: { token },
      transports: ['websocket', 'polling'],
      reconnection: true,
      reconnectionDelay: 1000,
      reconnectionAttempts: 5,
    });

    socketInstance.on('connect', () => {
      console.log('Connected to server');
      setIsConnected(true);
    });

    socketInstance.on('disconnect', (reason) => {
      console.log('Disconnected:', reason);
      setIsConnected(false);
    });

    socketInstance.on('error', (error) => {
      console.error('Socket error:', error);
      toast.error(`Connection error: ${error.message}`);
    });

    // Real-time event handlers
    socketInstance.on('task:updated', (data) => {
      // Update local state or invalidate queries
      queryClient.invalidateQueries(['tasks', data.task.project_id]);
      toast.info(`Task "${data.task.title}" updated by ${data.updated_by}`);
    });

    socketInstance.on('user:joined_project', (data) => {
      toast.info(`${data.username} joined the project`);
    });

    setSocket(socketInstance);

    return () => {
      socketInstance.disconnect();
    };
  }, [user, token]);

  const joinProject = useCallback((projectId: string) => {
    socket?.emit('join_project', { project_id: projectId });
  }, [socket]);

  const leaveProject = useCallback((projectId: string) => {
    socket?.emit('leave_project', { project_id: projectId });
  }, [socket]);

  return (
    <SocketContext.Provider value={{ socket, isConnected, joinProject, leaveProject }}>
      {children}
    </SocketContext.Provider>
  );
};

MCP (Model Context Protocol) Integration

• Protocol Version: Latest MCP specification
• Transport: HTTP-based for cloud deployment compatibility
• Tool Definition: Comprehensive tool descriptions for AI agents
• Error Handling: Proper error responses and retry mechanisms

MCP Server Implementation

# MCP server with FastMCP framework
from mcp.server.fastmcp import FastMCP, Context
from pydantic import BaseModel
from typing import Optional, List

class SearchRequest(BaseModel):
    query: str
    max_results: Optional[int] = 10
    project_id: Optional[str] = None

class DocumentResult(BaseModel):
    id: str
    title: str
    content: str
    similarity: float
    metadata: dict

# Initialize MCP server
mcp = FastMCP("ARCHON Knowledge Server")

@mcp.tool()
async def search_knowledge_base(
    ctx: Context,
    query: str,
    max_results: int = 10,
    project_id: Optional[str] = None
) -> List[DocumentResult]:
    """Search the ARCHON knowledge base using vector similarity.
    
    Args:
        query: Search query string
        max_results: Maximum number of results to return (1-50)
        project_id: Optional project ID to filter results
        
    Returns:
        List of relevant documents with similarity scores
    """
    try:
        await ctx.info(f"Searching knowledge base for: {query}")
        
        # Generate query embedding
        embedding = await generate_embedding(query)
        
        # Search database
        results = await search_documents_vector(
            embedding=embedding,
            limit=max_results,
            project_id=project_id
        )
        
        await ctx.info(f"Found {len(results)} relevant documents")
        
        return [
            DocumentResult(
                id=str(doc.id),
                title=doc.title,
                content=doc.content[:500] + "..." if len(doc.content) > 500 else doc.content,
                similarity=doc.similarity,
                metadata=doc.metadata or {}
            )
            for doc in results
        ]
        
    except Exception as e:
        await ctx.error(f"Knowledge base search failed: {str(e)}")
        raise

@mcp.tool()
async def create_task(
    ctx: Context,
    title: str,
    description: str,
    project_id: str,
    priority: str = "medium"
) -> dict:
    """Create a new task in the ARCHON platform.
    
    Args:
        title: Task title
        description: Detailed task description
        project_id: ID of the project to create task in
        priority: Task priority (low, medium, high, urgent)
        
    Returns:
        Created task information
    """
    try:
        await ctx.info(f"Creating task: {title}")
        
        task_data = {
            "title": title,
            "description": description,
            "project_id": project_id,
            "priority": priority,
            "status": "pending"
        }
        
        task = await create_task_service(task_data)
        
        await ctx.info(f"Task created successfully: {task.id}")
        
        return {
            "id": str(task.id),
            "title": task.title,
            "description": task.description,
            "status": task.status,
            "priority": task.priority,
            "created_at": task.created_at.isoformat(),
            "project_id": str(task.project_id)
        }
        
    except Exception as e:
        await ctx.error(f"Task creation failed: {str(e)}")
        raise

@mcp.resource("project://info/{project_id}")
async def get_project_info(project_id: str) -> str:
    """Get project information and context.
    
    Args:
        project_id: Project identifier
        
    Returns:
        Project information as formatted text
    """
    project = await get_project_by_id(project_id)
    if not project:
        raise ValueError(f"Project {project_id} not found")
    
    info = f"""
    Project: {project.name}
    Description: {project.description}
    Created: {project.created_at.strftime('%Y-%m-%d')}
    Owner: {project.owner.full_name}
    
    Recent Activity:
    """
    
    # Add recent tasks and documents
    recent_tasks = await get_recent_tasks(project_id, limit=5)
    for task in recent_tasks:
        info += f"- Task: {task.title} ({task.status})\n"
    
    return info

# Configure for cloud deployment
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(mcp.sse_app("/mcp"), host="0.0.0.0", port=8001)

8. Security & Authentication Standards

Authentication Architecture

• Primary Method: JWT-based authentication with refresh tokens
• Token Storage: Secure HTTP-only cookies for web clients
• Session Management: Redis-based session storage
• Multi-factor Authentication: TOTP support for enhanced security

JWT Implementation

# JWT utilities with security best practices
import jwt
from datetime import datetime, timedelta
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa

class JWTManager:
    """Secure JWT token management."""
    
    def __init__(self, private_key_path: str, public_key_path: str):
        self.private_key = self._load_private_key(private_key_path)
        self.public_key = self._load_public_key(public_key_path)
        self.algorithm = "RS256"
        self.access_token_expire = timedelta(minutes=15)
        self.refresh_token_expire = timedelta(days=7)
    
    def create_access_token(self, user_id: str, scopes: List[str] = None) -> str:
        """Create short-lived access token."""
        now = datetime.utcnow()
        payload = {
            "sub": user_id,
            "iat": now,
            "exp": now + self.access_token_expire,
            "type": "access",
            "scopes": scopes or []
        }
        return jwt.encode(payload, self.private_key, algorithm=self.algorithm)
    
    def create_refresh_token(self, user_id: str) -> str:
        """Create long-lived refresh token."""
        now = datetime.utcnow()
        payload = {
            "sub": user_id,
            "iat": now,
            "exp": now + self.refresh_token_expire,
            "type": "refresh"
        }
        return jwt.encode(payload, self.private_key, algorithm=self.algorithm)
    
    def verify_token(self, token: str, token_type: str = "access") -> dict:
        """Verify and decode JWT token."""
        try:
            payload = jwt.decode(
                token, 
                self.public_key, 
                algorithms=[self.algorithm],
                options={"require": ["sub", "iat", "exp", "type"]}
            )
            
            if payload.get("type") != token_type:
                raise jwt.InvalidTokenError(f"Invalid token type: {payload.get('type')}")
            
            return payload
            
        except jwt.ExpiredSignatureError:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Token has expired"
            )
        except jwt.InvalidTokenError as e:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail=f"Invalid token: {str(e)}"
            )

# Authentication dependency
async def get_current_user(
    request: Request,
    db: Database = Depends(get_database)
) -> User:
    """Extract and validate current user from JWT token."""
    
    # Try to get token from Authorization header
    authorization = request.headers.get("Authorization")
    if authorization and authorization.startswith("Bearer "):
        token = authorization.split(" ")[1]
    else:
        # Try to get token from secure cookie
        token = request.cookies.get("access_token")
    
    if not token:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Authentication required"
        )
    
    # Verify token
    jwt_manager = get_jwt_manager()
    payload = jwt_manager.verify_token(token, "access")
    
    # Get user from database
    user = await db.get_user_by_id(payload["sub"])
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="User not found"
        )
    
    return user

Input Validation & Sanitization

• Pydantic Models: Comprehensive validation for all API inputs
• SQL Injection Prevention: Parameterized queries and ORM usage
• XSS Prevention: Input sanitization and output encoding
• File Upload Security: Type validation and size limits

# Input validation patterns
from pydantic import BaseModel, Field, validator
from typing import Optional, List
import re

class DocumentCreateRequest(BaseModel):
    """Document creation with comprehensive validation."""
    title: str = Field(..., min_length=1, max_length=200)
    content: str = Field(..., min_length=1, max_length=50000)
    category_id: Optional[int] = Field(None, gt=0)
    tags: List[str] = Field(default_factory=list, max_items=10)
    is_public: bool = Field(default=False)
    
    @validator('title')
    def validate_title(cls, v):
        """Validate and sanitize title."""
        if not v.strip():
            raise ValueError('Title cannot be empty')
        
        # Remove potentially dangerous characters
        sanitized = re.sub(r'[<>"\']', '', v.strip())
        return sanitized
    
    @validator('content')
    def validate_content(cls, v):
        """Validate content length and basic sanitization."""
        if len(v.strip()) < 10:
            raise ValueError('Content must be at least 10 characters')
        
        # Basic HTML tag removal (use proper sanitizer in production)
        sanitized = re.sub(r'<script.*?</script>', '', v, flags=re.DOTALL | re.IGNORECASE)
        return sanitized.strip()
    
    @validator('tags')
    def validate_tags(cls, v):
        """Validate and sanitize tags."""
        sanitized_tags = []
        for tag in v:
            if tag and tag.strip():
                # Only allow alphanumeric and basic punctuation
                clean_tag = re.sub(r'[^a-zA-Z0-9\-_\s]', '', tag.strip())
                if clean_tag and len(clean_tag) <= 30:
                    sanitized_tags.append(clean_tag.lower())
        
        return list(set(sanitized_tags))  # Remove duplicates

# File upload validation
from fastapi import UploadFile, HTTPException
import magic

async def validate_file_upload(file: UploadFile) -> None:
    """Validate uploaded file security."""
    
    # Check file size (10MB limit)
    content = await file.read()
    if len(content) > 10 * 1024 * 1024:
        raise HTTPException(
            status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
            detail="File size exceeds 10MB limit"
        )
    
    # Validate file type using python-magic
    file_type = magic.from_buffer(content, mime=True)
    allowed_types = {
        'text/plain',
        'text/markdown',
        'application/pdf',
        'application/json',
        'image/jpeg',
        'image/png',
        'image/webp'
    }
    
    if file_type not in allowed_types:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=f"File type {file_type} not allowed"
        )
    
    # Reset file pointer
    await file.seek(0)

Authorization & Permissions

• Role-Based Access Control (RBAC): User roles with specific permissions
• Resource-Based Permissions: Fine-grained access control per resource
• API Key Management: Secure API key generation and validation
• Rate Limiting: Prevent abuse with request throttling

# Permission system
from enum import Enum
from functools import wraps

class Permission(str, Enum):
    """System permissions."""
    READ_DOCUMENTS = "read:documents"
    WRITE_DOCUMENTS = "write:documents"
    DELETE_DOCUMENTS = "delete:documents"
    MANAGE_PROJECTS = "manage:projects"
    MANAGE_USERS = "manage:users"
    ADMIN_ACCESS = "admin:access"

class Role(str, Enum):
    """User roles with associated permissions."""
    VIEWER = "viewer"
    CONTRIBUTOR = "contributor"
    MANAGER = "manager"
    ADMIN = "admin"

ROLE_PERMISSIONS = {
    Role.VIEWER: [Permission.READ_DOCUMENTS],
    Role.CONTRIBUTOR: [Permission.READ_DOCUMENTS, Permission.WRITE_DOCUMENTS],
    Role.MANAGER: [
        Permission.READ_DOCUMENTS,
        Permission.WRITE_DOCUMENTS,
        Permission.DELETE_DOCUMENTS,
        Permission.MANAGE_PROJECTS
    ],
    Role.ADMIN: [perm for perm in Permission]
}

def require_permission(permission: Permission):
    """Decorator to require specific permission."""
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            # Extract current_user from function arguments
            current_user = None
            for arg in args:
                if isinstance(arg, User):
                    current_user = arg
                    break
            
            if not current_user:
                current_user = kwargs.get('current_user')
            
            if not current_user:
                raise HTTPException(
                    status_code=status.HTTP_401_UNAUTHORIZED,
                    detail="Authentication required"
                )
            
            # Check permission
            user_permissions = ROLE_PERMISSIONS.get(current_user.role, [])
            if permission not in user_permissions:
                raise HTTPException(
                    status_code=status.HTTP_403_FORBIDDEN,
                    detail=f"Permission {permission.value} required"
                )
            
            return await func(*args, **kwargs)
        return wrapper
    return decorator

# Usage example
@router.delete("/documents/{document_id}")
@require_permission(Permission.DELETE_DOCUMENTS)
async def delete_document(
    document_id: int,
    current_user: User = Depends(get_current_user),
    db: Database = Depends(get_database)
) -> APIResponse[None]:
    """Delete a document with permission check."""
    
    # Additional resource-level check
    document = await db.get_document(document_id)
    if not document:
        raise ResourceNotFoundException("Document", str(document_id))
    
    # Check if user owns the document or has admin access
    if (document.created_by != current_user.id and 
        current_user.role != Role.ADMIN):
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="You can only delete your own documents"
        )
    
    await db.delete_document(document_id)
    
    return APIResponse(
        success=True,
        message="Document deleted successfully"
    )

9. Testing Strategy & Quality Assurance

Test Architecture

• Backend Testing: pytest with comprehensive fixtures and mocking
• Frontend Testing: Vitest with React Testing Library
• Integration Testing: End-to-end testing with Playwright
• Load Testing: Performance testing with locust or artillery

Python Testing Setup

# conftest.py - pytest configuration
import pytest
import asyncio
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from fastapi.testclient import TestClient
from httpx import AsyncClient

from src.server.main import app
from src.server.database import get_database
from src.server.auth import get_current_user

# Test database configuration
TEST_DATABASE_URL = "postgresql+asyncpg://test_user:test_pass@localhost/test_archon"

@pytest.fixture(scope="session")
def event_loop():
    """Create an instance of the default event loop for the test session."""
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()

@pytest.fixture(scope="session")
async def test_engine():
    """Create test database engine."""
    engine = create_async_engine(
        TEST_DATABASE_URL,
        echo=False,
        pool_pre_ping=True
    )
    yield engine
    await engine.dispose()

@pytest.fixture
async def test_session(test_engine):
    """Create test database session."""
    async_session = sessionmaker(
        test_engine, class_=AsyncSession, expire_on_commit=False
    )
    
    async with async_session() as session:
        # Start transaction
        await session.begin()
        yield session
        # Rollback all changes
        await session.rollback()

@pytest.fixture
def test_user():
    """Create test user data."""
    return {
        "id": "test-user-123",
        "email": "test@example.com",
        "username": "testuser",
        "full_name": "Test User",
        "role": "contributor"
    }

@pytest.fixture
def authenticated_client(test_user):
    """Create authenticated test client."""
    
    # Override auth dependency
    async def override_get_current_user():
        return User(**test_user)
    
    app.dependency_overrides[get_current_user] = override_get_current_user
    
    with TestClient(app) as client:
        yield client
    
    # Clean up
    app.dependency_overrides.clear()

@pytest.fixture
async def async_client():
    """Create async HTTP client for testing."""
    async with AsyncClient(app=app, base_url="http://test") as client:
        yield client

# Example test cases
@pytest.mark.asyncio
async def test_create_document(authenticated_client, test_session):
    """Test document creation with authentication."""
    
    document_data = {
        "title": "Test Document",
        "content": "This is a test document with sufficient content length.",
        "category_id": 1,
        "tags": ["test", "example"]
    }
    
    response = authenticated_client.post("/api/v1/documents", json=document_data)
    
    assert response.status_code == 201
    data = response.json()
    assert data["success"] is True
    assert data["data"]["title"] == document_data["title"]
    assert len(data["data"]["tags"]) == 2

@pytest.mark.asyncio
async def test_search_documents_vector(test_session, test_user):
    """Test vector search functionality."""
    
    # Create test documents with embeddings
    test_docs = [
        {
            "title": "Python Programming",
            "content": "Python is a powerful programming language.",
            "embedding": [0.1, 0.2, 0.3] * 128  # Mock 384-dim embedding
        },
        {
            "title": "JavaScript Basics",
            "content": "JavaScript is used for web development.",
            "embedding": [0.2, 0.3, 0.4] * 128
        }
    ]
    
    # Insert test data
    for doc_data in test_docs:
        doc = Document(**doc_data, created_by=test_user["id"])
        test_session.add(doc)
    await test_session.commit()
    
    # Test search
    query_embedding = [0.15, 0.25, 0.35] * 128
    results = await search_documents_vector(
        test_session,
        embedding=query_embedding,
        limit=5
    )
    
    assert len(results) >= 2
    assert results[0].similarity > 0.8  # Should find close matches

@pytest.mark.asyncio
async def test_websocket_connection(async_client):
    """Test WebSocket connection and messaging."""
    
    with async_client.websocket_connect("/ws") as websocket:
        # Test connection
        data = websocket.receive_json()
        assert data["type"] == "connection_established"
        
        # Test message echo
        test_message = {"type": "test", "data": "hello"}
        websocket.send_json(test_message)
        
        response = websocket.receive_json()
        assert response["type"] == "echo"
        assert response["data"] == test_message

# Performance testing
@pytest.mark.asyncio
async def test_document_search_performance(test_session):
    """Test search performance with large dataset."""
    import time
    
    # Create many test documents
    documents = []
    for i in range(1000):
        doc = Document(
            title=f"Document {i}",
            content=f"Content for document {i} with various keywords.",
            embedding=[random.random() for _ in range(384)],
            created_by="test-user"
        )
        documents.append(doc)
    
    test_session.add_all(documents)
    await test_session.commit()
    
    # Measure search performance
    start_time = time.time()
    query_embedding = [random.random() for _ in range(384)]
    
    results = await search_documents_vector(
        test_session,
        embedding=query_embedding,
        limit=10
    )
    
    search_time = time.time() - start_time
    
    assert len(results) == 10
    assert search_time < 0.1  # Should complete in under 100ms

Frontend Testing Setup

// vitest.config.ts
import { defineConfig } from 'vitest/config';
import react from '@vitejs/plugin-react';
import path from 'path';

export default defineConfig({
  plugins: [react()],
  test: {
    environment: 'jsdom',
    setupFiles: ['./src/test/setup.ts'],
    globals: true,
    css: true,
  },
  resolve: {
    alias: {
      '@': path.resolve(__dirname, 'src'),
    },
  },
});

// src/test/setup.ts
import '@testing-library/jest-dom';
import { vi } from 'vitest';

// Mock Socket.IO
vi.mock('socket.io-client', () => ({
  io: vi.fn(() => ({
    on: vi.fn(),
    off: vi.fn(),
    emit: vi.fn(),
    disconnect: vi.fn(),
    connected: true,
  })),
}));

// Mock IntersectionObserver
global.IntersectionObserver = vi.fn(() => ({
  observe: vi.fn(),
  disconnect: vi.fn(),
  unobserve: vi.fn(),
}));

// Example React component tests
// src/components/TaskList.test.tsx
import { render, screen, fireEvent, waitFor } from '@testing-library/react';
import { QueryClient, QueryClientProvider } from '@tanstack/react-query';
import { TaskList } from './TaskList';
import { TaskProvider } from '@/contexts/TaskContext';

const createTestQueryClient = () => new QueryClient({
  defaultOptions: {
    queries: { retry: false },
    mutations: { retry: false },
  },
});

const renderWithProviders = (component: React.ReactElement) => {
  const queryClient = createTestQueryClient();
  
  return render(
    <QueryClientProvider client={queryClient}>
      <TaskProvider>
        {component}
      </TaskProvider>
    </QueryClientProvider>
  );
};

describe('TaskList Component', () => {
  it('renders task list correctly', async () => {
    const mockTasks = [
      { id: '1', title: 'Test Task 1',

*Truncated - read the full file at https://github.com/JackSmack1971/ARCHONRELOADED/blob/589923dd7bb8bf393d44b909fe2def505a989209/AGENTS.md.*