{
  "message.content": "# Executive Summary\n\nThis engineering dossier outlines a comprehensive deployment-grade plan for a private home-lab FastAPI service that proxies OpenAI-compatible requests to a local llama.cpp server. The system supports benchmarking, artifact storage, reasoning measurement, and privacy compliance while handling large output token counts and model profile changes.\n\nKey improvements include:\n\n1. **Artifact Storage**: Implement durable, external storage for large benchmark outputs instead of inline JSON\n2. **Reasoning Budget**: Measure reasoning output without disabling thinking mode\n3. **Benchmark Metadata**: Rich metadata for screenshots and model comparisons\n4. **Privacy Controls**: Automatic redaction of sensitive content\n5. **Audit Trail**: Track model profile changes with versioning\n6. **Scalability**: Handle tens of thousands of output tokens with timeout and storage management\n\nThe system will use SQLite for metadata, external storage (S3-compatible) for artifacts, and maintain comprehensive logging for auditability. All changes are designed to be backward-compatible with existing workflows while adding robustness for production use.\n\n# Current Architecture\n\nThe current architecture consists of:\n\n- **FastAPI App** (`app/main.py`): Exposes dashboard, health checks, and proxy endpoints\n- **Proxy Layer** (`app/proxy.py`): Forwards requests to llama.cpp, records metadata, timings, and usage\n- **Storage Layer** (`app/store.py`): Write helpers for runs, sessions, LLM requests, events, and benchmarks\n- **Reporting Layer** (`app/reports.py`): Aggregates dashboard and report metrics\n- **Benchmark Runner** (`app/bench.py`): Executes benchmark campaigns\n- **Deployment**: Docker Compose with ai-flight-recorder and llama-cpp-server on 192.168.1.116\n- **Data Storage**: /models/flight-recorder/data\n- **Model Profile**: Qwen3.6-35B-A3B-APEX-MTP-I-Balanced.gguf with llama.cpp reasoning enabled\n\nThe proxy currently handles /v1/chat/completions and records:\n- Request metadata (timestamp, client IP, model, prompt)\n- Response previews (first 200 tokens)\n- Timings (request duration, token processing)\n- Usage statistics (input/output tokens)\n\n# Failure Modes Found\n\n1. **Storage Exhaustion**: Large outputs can fill disk space quickly\n2. **Timeout Issues**: Long-running requests may timeout during token generation\n3. **Memory Leaks**: Accumulation of request data without proper cleanup\n4. **Privacy Leaks**: Sensitive content may be stored without redaction\n5. **Model Profile Changes**: Inconsistent tracking of model changes\n6. **Benchmark Artifact Overload**: Inline JSON outputs become unwieldy for large benchmarks\n7. **UI Rendering**: Large previews cause browser performance issues\n8. **Concurrency Issues**: Race conditions in concurrent request handling\n9. **Metadata Inconsistency**: Incomplete or missing metadata in storage\n10. **Audit Trail Gaps**: No versioning or change tracking for model profiles\n\n# Data Model Changes\n\n## Enhanced SQLite Schema\n\n```sql\n-- Enhanced runs table\nCREATE TABLE runs (\n    id INTEGER PRIMARY KEY AUTOINCREMENT,\n    name TEXT NOT NULL,\n    description TEXT,\n    model_profile TEXT NOT NULL,\n    model_version TEXT,\n    started_at TIMESTAMP NOT NULL,\n    completed_at TIMESTAMP,\n    status TEXT NOT NULL DEFAULT 'running',\n    artifact_storage_id TEXT,\n    created_by TEXT,\n    tags TEXT,\n    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP\n);\n\n-- Enhanced llm_requests table\nCREATE TABLE llm_requests (\n    id INTEGER PRIMARY KEY AUTOINCREMENT,\n    run_id INTEGER NOT NULL,\n    session_id TEXT,\n    request_id TEXT UNIQUE,\n    model TEXT,\n    prompt TEXT NOT NULL,\n    response TEXT,\n    response_preview TEXT,\n    response_tokens INTEGER,\n    prompt_tokens INTEGER,\n    total_tokens INTEGER,\n    request_duration_ms INTEGER,\n    reasoning_tokens INTEGER DEFAULT 0,\n    reasoning_duration_ms INTEGER DEFAULT 0,\n    status TEXT NOT NULL DEFAULT 'pending',\n    error_message TEXT,\n    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,\n    FOREIGN KEY (run_id) REFERENCES runs (id)\n);\n\n-- New reasoning_budgets table\nCREATE TABLE reasoning_budgets (\n    id INTEGER PRIMARY KEY AUTOINCREMENT,\n    run_id INTEGER NOT NULL,\n    request_id TEXT,\n    budget_tokens INTEGER NOT NULL,\n    used_tokens INTEGER NOT NULL DEFAULT 0,\n    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,\n    FOREIGN KEY (run_id) REFERENCES runs (id),\n    FOREIGN KEY (request_id) REFERENCES llm_requests (request_id)\n);\n\n-- Enhanced benchmark_campaigns table\nCREATE TABLE benchmark_campaigns (\n    id INTEGER PRIMARY KEY AUTOINCREMENT,\n    name TEXT NOT NULL,\n    description TEXT,\n    model_profile TEXT NOT NULL,\n    model_version TEXT,\n    started_at TIMESTAMP NOT NULL,\n    completed_at TIMESTAMP,\n    status TEXT NOT NULL DEFAULT 'running',\n    total_requests INTEGER,\n    completed_requests INTEGER DEFAULT 0,\n    artifact_storage_id TEXT,\n    created_by TEXT,\n    tags TEXT,\n    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP\n);\n\n-- New model_profile_versions table\nCREATE TABLE model_profile_versions (\n    id INTEGER PRIMARY KEY AUTOINCREMENT,\n    profile_name TEXT NOT NULL,\n    version TEXT NOT NULL,\n    file_path TEXT NOT NULL,\n    checksum TEXT NOT NULL,\n    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,\n    changed_by TEXT,\n    change_reason TEXT,\n    is_active BOOLEAN DEFAULT FALSE\n);\n\n-- Enhanced events table\nCREATE TABLE events (\n    id INTEGER PRIMARY KEY AUTOINCREMENT,\n    run_id INTEGER,\n    request_id TEXT,\n    event_type TEXT NOT NULL,\n    message TEXT,\n    timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,\n    metadata TEXT,\n    FOREIGN KEY (run_id) REFERENCES runs (id),\n    FOREIGN KEY (request_id) REFERENCES llm_requests (request_id)\n);\n```\n\n## Migration Notes\n\n1. **Backward Compatibility**: All new columns are optional with defaults\n2. **Data Migration**: Existing runs will have `artifact_storage_id` set to NULL\n3. **Indexing**: Add indexes on `run_id`, `request_id`, `model_profile`, `created_at`\n4. **Constraints**: Add foreign key constraints for referential integrity\n5. **Versioning**: Add `model_version` field to track model profile changes\n\n# Artifact Storage Design\n\n## External Storage Integration\n\nThe system will use S3-compatible storage (MinIO) for large benchmark outputs:\n\n```python\n# app/store.py - Enhanced storage functions\nimport boto3\nfrom botocore.exceptions import ClientError\n\nclass ArtifactStorage:\n    def __init__(self, endpoint_url, access_key, secret_key, bucket_name):\n        self.s3_client = boto3.client(\n            's3',\n            endpoint_url=endpoint_url,\n            aws_access_key_id=access_key,\n            aws_secret_access_key=secret_key\n        )\n        self.bucket_name = bucket_name\n\n    def store_artifact(self, run_id, content, content_type):\n        key = f\"run_{run_id}/artifact_{uuid.uuid4()}.json\"\n        try:\n            self.s3_client.put_object(\n                Bucket=self.bucket_name,\n                Key=key,\n                Body=content,\n                ContentType=content_type\n            )\n            return key\n        except ClientError as e:\n            logger.error(f\"Failed to store artifact for run {run_id}: {e}\")\n            raise\n\n    def retrieve_artifact(self, artifact_key):\n        try:\n            response = self.s3_client.get_object(\n                Bucket=self.bucket_name,\n                Key=artifact_key\n            )\n            return response['Body'].read()\n        except ClientError as e:\n            logger.error(f\"Failed to retrieve artifact {artifact_key}: {e}\")\n            raise\n```\n\n## Artifact Management\n\n1. **Storage Strategy**: Large outputs (>10KB) stored externally, small previews inline\n2. **Retention Policy**: 30-day retention with manual cleanup option\n3. **Access Control**: Read-only access for dashboard, full access for admin\n4. **Metadata**: Store S3 object key, size, content type, and creation timestamp\n5. **Error Handling**: Graceful degradation when external storage is unavailable\n\n## CLI Examples\n\n```bash\n# Create a new benchmark campaign\npython -m app.bench create_campaign \\\n    --name \"Qwen3.6-35B-Reasoning-Test\" \\\n    --model-profile \"Qwen3.6-35B-A3B-APEX-MTP-I-Balanced.gguf\" \\\n    --description \"Test reasoning capabilities with long prompts\" \\\n    --tags \"reasoning,benchmark\" \\\n    --requests 1000\n\n# List all benchmark campaigns\npython -m app.bench list_campaigns\n\n# Get campaign details\npython -m app.bench get_campaign --id 123\n\n# Download artifact\npython -m app.bench download_artifact --campaign-id 123 --output-file results.json\n```\n\n# Reasoning Budget Handling\n\n## Budget Tracking Mechanism\n\n```python\n# app/proxy.py - Enhanced reasoning tracking\nimport time\nfrom typing import Dict, Optional\n\nclass ReasoningTracker:\n    def __init__(self):\n        self.budgets: Dict[str, Dict] = {}\n        self.active_requests: Dict[str, Dict] = {}\n\n    def start_request(self, request_id: str, budget_tokens: int):\n        self.active_requests[request_id] = {\n            'start_time': time.time(),\n            'budget_tokens': budget_tokens,\n            'used_tokens': 0,\n            'reasoning_tokens': 0\n        }\n\n    def update_tokens(self, request_id: str, tokens_used: int, is_reasoning: bool = False):\n        if request_id in self.active_requests:\n            self.active_requests[request_id]['used_tokens'] += tokens_used\n            if is_reasoning:\n                self.active_requests[request_id]['reasoning_tokens'] += tokens_used\n\n    def get_budget_status(self, request_id: str) -> Dict:\n        if request_id in self.active_requests:\n            req = self.active_requests[request_id]\n            return {\n                'request_id': request_id,\n                'budget_tokens': req['budget_tokens'],\n                'used_tokens': req['used_tokens'],\n                'reasoning_tokens': req['reasoning_tokens'],\n                'remaining_tokens': req['budget_tokens'] - req['used_tokens'],\n                'is_exceeded': req['used_tokens'] > req['budget_tokens']\n            }\n        return None\n\n# Global tracker instance\nreasoning_tracker = ReasoningTracker()\n```\n\n## Budget Configuration\n\nThe system will support configurable reasoning budgets:\n\n1. **Default Budget**: 1000 tokens for reasoning\n2. **Per-Request Override**: Allow override via request headers\n3. **Budget Enforcement**: Stop token generation when budget is exceeded\n4. **Metrics Collection**: Track actual usage vs budget\n5. **Alerting**: Log warnings when budget is approached\n\n## Integration with LLM\n\n```python\n# Enhanced proxy logic\nasync def forward_to_llama_cpp(request_data: dict):\n    # Extract reasoning budget from request\n    reasoning_budget = int(request_data.get('reasoning_budget', 1000))\n    \n    # Start tracking\n    request_id = str(uuid.uuid4())\n    reasoning_tracker.start_request(request_id, reasoning_budget)\n    \n    # Forward to llama.cpp with reasoning budget\n    response = await llama_cpp_client.chat.completions.create(\n        **request_data,\n        reasoning_budget=reasoning_budget\n    )\n    \n    # Update tracking\n    reasoning_tracker.update_tokens(request_id, response.usage.total_tokens, True)\n    \n    return response\n```\n\n# Benchmark Runner Design\n\n## Enhanced Benchmark Campaigns\n\n```python\n# app/bench.py - Benchmark runner\nimport asyncio\nimport json\nfrom typing import List, Dict\nfrom concurrent.futures import ThreadPoolExecutor\n\nclass BenchmarkRunner:\n    def __init__(self, db_path: str, storage: ArtifactStorage):\n        self.db_path = db_path\n        self.storage = storage\n        self.executor = ThreadPoolExecutor(max_workers=10)\n\n    async def create_campaign(self, name: str, model_profile: str, description: str,\n                            tags: List[str], requests: int, **kwargs) -> int:\n        # Create campaign in database\n        campaign_id = self._create_campaign_in_db(name, model_profile, description,\n                                                 tags, requests, **kwargs)\n        \n        # Start campaign execution\n        asyncio.create_task(self._execute_campaign(campaign_id))\n        \n        return campaign_id\n\n    async def _execute_campaign(self, campaign_id: int):\n        # Get campaign details\n        campaign = self._get_campaign_from_db(campaign_id)\n        \n        # Create artifact storage\n        artifact_key = self.storage.store_artifact(campaign_id, json.dumps({}), 'application/json')\n        \n        # Update campaign with artifact key\n        self._update_campaign_artifact(campaign_id, artifact_key)\n        \n        # Execute requests\n        tasks = [\n            self._execute_request(campaign_id, i) \n            for i in range(campaign['total_requests'])\n        ]\n        \n        # Process in batches to avoid memory issues\n        batch_size = 100\n        for i in range(0, len(tasks), batch_size):\n            batch = tasks[i:i + batch_size]\n            await asyncio.gather(*batch)\n            \n            # Update progress\n            self._update_campaign_progress(campaign_id, i + len(batch))\n\n    async def _execute_request(self, campaign_id: int, request_index: int):\n        # Generate request data\n        request_data = self._generate_request_data(request_index)\n        \n        # Forward to proxy\n        response = await self._forward_to_proxy(request_data)\n        \n        # Store result\n        self._store_request_result(campaign_id, request_index, response)\n\n    def _generate_request_data(self, index: int) -> Dict:\n        # Generate synthetic prompt\n        return {\n            'model': 'Qwen3.6-35B-A3B-APEX-MTP-I-Balanced.gguf',\n            'messages': [\n                {'role': 'user', 'content': f'Question {index}: What is the capital of France?'}\n            ],\n            'max_tokens': 2000,\n            'temperature': 0.7,\n            'reasoning_budget': 1000\n        }\n```\n\n## Benchmark Metrics\n\nThe benchmark runner will collect comprehensive metrics:\n\n1. **Performance**: Request duration, token processing speed\n2. **Resource Usage**: Memory, CPU, disk I/O\n3. **Output Quality**: Response length, token count, reasoning tokens\n4. **Error Rates**: Failure counts, error types\n5. **Model Behavior**: Reasoning vs non-reasoning token usage\n6. **Scalability**: Throughput, concurrency handling\n\n# Reporting Plane\n\n## Dashboard Metrics\n\nThe dashboard will display:\n\n1. **Run Status**: Active, completed, failed runs with progress indicators\n2. **Performance Charts**: Request duration, token processing speed over time\n3. **Resource Usage**: Memory, CPU, disk usage\n4. **Reasoning Metrics**: Average reasoning tokens, budget utilization\n5. **Error Analysis**: Error types, frequency, and resolution\n6. **Model Comparison**: Side-by-side comparison of different model profiles\n7. **Benchmark Results**: Detailed results with artifact links\n\n## Report Generation\n\n```python\n# app/reports.py - Report generation\nfrom datetime import datetime\nimport pandas as pd\n\nclass ReportGenerator:\n    def __init__(self, db_path: str):\n        self.db_path = db_path\n\n    def generate_run_report(self, run_id: int) -> Dict:\n        # Query database for run data\n        run_data = self._get_run_data(run_id)\n        request_data = self._get_request_data(run_id)\n        \n        # Calculate metrics\n        total_requests = len(request_data)\n        avg_duration = self._calculate_avg_duration(request_data)\n        avg_tokens = self._calculate_avg_tokens(request_data)\n        reasoning_usage = self._calculate_reasoning_usage(request_data)\n        \n        # Generate report\n        report = {\n            'run_id': run_id,\n            'run_name': run_data['name'],\n            'model_profile': run_data['model_profile'],\n            'started_at': run_data['started_at'],\n            'completed_at': run_data['completed_at'],\n            'total_requests': total_requests,\n            'avg_duration_ms': avg_duration,\n            'avg_tokens_per_request': avg_tokens,\n            'total_reasoning_tokens': reasoning_usage['total'],\n            'avg_reasoning_tokens': reasoning_usage['average'],\n            'artifact_storage_id': run_data['artifact_storage_id'],\n            'tags': run_data['tags'],\n            'created_by': run_data['created_by']\n        }\n        \n        return report\n\n    def generate_benchmark_comparison_report(self, campaign_ids: List[int]) -> Dict:\n        # Compare multiple campaigns\n        reports = [self.generate_run_report(cid) for cid in campaign_ids]\n        \n        # Create comparison data\n        comparison_data = {\n            'campaigns': reports,\n            'summary': self._generate_comparison_summary(reports)\n        }\n        \n        return comparison_data\n```\n\n## Dashboard Endpoints\n\n```python\n# app/main.py - Dashboard endpoints\n@app.get(\"/dashboard/runs/{run_id}/report\")\nasync def get_run_report(run_id: int):\n    report = report_generator.generate_run_report(run_id)\n    return report\n\n@app.get(\"/dashboard/benchmark/comparison\")\nasync def get_benchmark_comparison(campaign_ids: List[int] = Query([])):\n    report = report_generator.generate_benchmark_comparison_report(campaign_ids)\n    return report\n\n@app.get(\"/dashboard/metrics\")\nasync def get_dashboard_metrics():\n    metrics = {\n        'active_runs': get_active_runs_count(),\n        'completed_runs': get_completed_runs_count(),\n        'total_requests': get_total_requests_count(),\n        'avg_duration': get_avg_duration(),\n        'avg_tokens': get_avg_tokens(),\n        'recent_errors': get_recent_errors(24),\n        'model_profiles': get_model_profiles()\n    }\n    return metrics\n```\n\n# Privacy And Redaction\n\n## Redaction Strategy\n\n```python\n# app/proxy.py - Privacy redaction\nimport re\nfrom typing import Dict, List\n\nclass PrivacyRedactor:\n    def __init__(self):\n        # Define redaction patterns\n        self.redaction_patterns = [\n            # Email addresses\n            re.compile(r'\\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\\.[A-Z|a-z]{2,}\\b'),\n            # Phone numbers\n            re.compile(r'\\b(?:\\+?1[-.\\s]?)?\\(?\\d{3}\\)?[-.\\s]?\\d{3}[-.\\s]?\\d{4}\\b'),\n            # Teams/Slack IDs\n            re.compile(r'\\b[A-Z0-9]{8,}\\b'),\n            # IP addresses\n            re.compile(r'\\b(?:[0-9]{1,3}\\.){3}[0-9]{1,3}\\b'),\n            # URLs\n            re.compile(r'https?://[\\w\\-]+(?:\\.[\\w\\-]+)+[\\w\\-\\.,@?^=%&:/~\\+#]*[\\w\\-@?^=%&/~\\+#]'),\n            # Credit card numbers\n            re.compile(r'\\b(?:\\d{4}[-\\s?]){3}\\d{4}\\b'),\n            # Social Security Numbers\n            re.compile(r'\\b\\d{3}-\\d{2}-\\d{4}\\b')\n        ]\n\n    def redact_content(self, content: str) -> str:\n        redacted = content\n        for pattern in self.redaction_patterns:\n            redacted = pattern.sub('[REDACTED]', redacted)\n        return redacted\n\n    def redact_request(self, request_data: Dict) -> Dict:\n        # Redact prompt and messages\n        if 'messages' in request_data:\n            for msg in request_data['messages']:\n                if 'content' in msg:\n                    msg['content'] = self.redact_content(msg['content'])\n        \n        if 'prompt' in request_data:\n            request_data['prompt'] = self.redact_content(request_data['prompt'])\n        \n        return request_data\n\n# Global redactor instance\nprivacy_redactor = PrivacyRedactor()\n```\n\n## Redaction Process\n\n1. **Input Redaction**: Redact sensitive content from prompts before storage\n2. **Output Redaction**: Redact sensitive content from responses before storage\n3. **Preview Redaction**: Redact sensitive content from previews\n4. **Audit Trail**: Log redaction events for compliance\n5. **Exception Handling**: Allow manual override for legitimate sensitive content\n\n## Configuration\n\n```yaml\n# privacy_config.yaml\nredaction:\n  enabled: true\n  patterns:\n    - email\n    - phone\n    - ip\n    - url\n    - credit_card\n    - ssn\n    - teams_id\n  redaction_mask: '[REDACTED]'\n  logging:\n    enabled: true\n    log_level: INFO\n```\n\n# Deployment Plan\n\n## Infrastructure Setup\n\n1. **Docker Compose Configuration**:\n\n```yaml\nversion: '3.8'\nservices:\n  llama-cpp-server:\n    image: ghcr.io/ggerganov/llama.cpp:latest\n    container_name: llama-cpp-server\n    volumes:\n      - ./models:/models\n      - ./data:/data\n    ports:\n      - \"8000:8000\"\n    command: [\n      \"--model\", \"/models/Qwen3.6-35B-A3B-APEX-MTP-I-Balanced.gguf\",\n      \"--host\", \"0.0.0.0\",\n      \"--port\", \"8000\",\n      \"--n-gpu-layers\", \"90\",\n      \"--n_ctx\", \"8192\",\n      \"--n_threads\", \"16\",\n      \"--verbose\"\n    ]\n    restart: unless-stopped\n\n  ai-flight-recorder:\n    build: .\n    container_name: ai-flight-recorder\n    volumes:\n      - ./data:/app/data\n      - ./models:/app/models\n    ports:\n      - \"8080:8080\"\n    environment:\n      - DATABASE_URL=sqlite:////app/data/flight_recorder.db\n      - S3_ENDPOINT=http://minio:9000\n      - S3_BUCKET=flight-recorder-artifacts\n      - S3_ACCESS_KEY=minioadmin\n      - S3_SECRET_KEY=minioadmin\n    depends_on:\n      - llama-cpp-server\n      - minio\n    restart: unless-stopped\n\n  minio:\n    image: minio/minio:latest\n    container_name: minio\n    volumes:\n      - ./minio/data:/data\n    ports:\n      - \"9000:9000\"\n      - \"9001:9001\"\n    command: server /data --console-address \":9001\"\n    environment:\n      - MINIO_ROOT_USER=minioadmin\n      - MINIO_ROOT_PASSWORD=minioadmin\n    restart: unless-stopped\n```\n\n## Environment Variables\n\n```bash\n# .env file\nDATABASE_URL=sqlite:////app/data/flight_recorder.db\nS3_ENDPOINT=http://minio:9000\nS3_BUCKET=flight-recorder-artifacts\nS3_ACCESS_KEY=minioadmin\nS3_SECRET_KEY=minioadmin\nMODEL_PROFILE=Qwen3.6-35B-A3B-APEX-MTP-I-Balanced.gguf\nLOG_LEVEL=INFO\nMAX_ARTIFACT_SIZE=10485760\nREASONING_BUDGET=1000\n```\n\n## Deployment Steps\n\n1. **Prepare Environment**:\n   - Ensure Docker and Docker Compose are installed\n   - Create necessary directories\n   - Set up MinIO storage\n\n2. **Build Images**:\n   - Build FastAPI service image\n   - Pull llama.cpp image\n   - Configure MinIO\n\n3. **Initialize Database**:\n   - Run database migrations\n   - Create initial schema\n   - Seed with default configurations\n\n4. **Start Services**:\n   - Start MinIO\n   - Start llama.cpp server\n   - Start ai-flight-recorder\n\n5. **Verify Deployment**:\n   - Check service health\n   - Test proxy functionality\n   - Validate artifact storage\n   - Confirm privacy redaction\n\n# Test Plan\n\n## Unit Tests\n\n```python\n# test_store.py\nimport pytest\nfrom app.store import RunStore, LLMRequestStore\n\n@pytest.fixture\ndef run_store():\n    return RunStore(':memory:')\n\n@pytest.fixture\ndef llm_request_store():\n    return LLMRequestStore(':memory:')\n\ndef test_create_run(run_store):\n    run_id = run_store.create_run('test_run', 'Qwen3.6-35B-A3B-APEX-MTP-I-Balanced.gguf')\n    assert run_id > 0\n\ndef test_create_request(llm_request_store):\n    request_id = llm_request_store.create_request(1, 'test_prompt')\n    assert request_id is not None\n\n# test_proxy.py\nimport pytest\nfrom app.proxy import PrivacyRedactor\n\ndef test_email_redaction():\n    redactor = PrivacyRedactor()\n    content = \"Contact us at test@example.com for more information\"\n    redacted = redactor.redact_content(content)\n    assert '[REDACTED]' in redacted\n\n# test_bench.py\nimport pytest\nfrom app.bench import BenchmarkRunner\n\n@pytest.fixture\ndef benchmark_runner():\n    return BenchmarkRunner(':memory:', None)\n\ndef test_campaign_creation(benchmark_runner):\n    campaign_id = benchmark_runner.create_campaign(\n        'test_campaign',\n        'Qwen3.6-35B-A3B-APEX-MTP-I-Balanced.gguf',\n        'Test description',\n        [],\n        10\n    )\n    assert campaign_id > 0\n```\n\n## Integration Tests\n\n```python\n# test_integration.py\nimport pytest\nimport asyncio\nimport httpx\nfrom app.main import app\n\n@pytest.mark.asyncio\nasync def test_health_endpoint():\n    async with httpx.AsyncClient(app=app, base_url=\"http://test\") as client:\n        response = await client.get(\"/health\")\n        assert response.status_code == 200\n        assert response.json() == {\"status\": \"healthy\"}\n\n@pytest.mark.asyncio\nasync def test_proxy_endpoint():\n    async with httpx.AsyncClient(app=app, base_url=\"http://test\") as client:\n        response = await client.post(\"/v1/chat/completions\", json={\n            \"model\": \"Qwen3.6-35B-A3B-APEX-MTP-I-Balanced.gguf\",\n            \"messages\": [{\"role\": \"user\", \"content\": \"Hello\"}]\n        })\n        assert response.status_code == 200\n        assert \"choices\" in response.json()\n```\n\n## Acceptance Tests\n\n1. **Run Creation**: Create a run and verify it's stored correctly\n2. **Request Forwarding**: Forward a request and verify it's recorded\n3. **Artifact Storage**: Store large outputs and verify external storage\n4. **Privacy Redaction**: Verify sensitive content is redacted\n5. **Benchmark Execution**: Run a benchmark and verify results\n6. **Report Generation**: Generate reports and verify data integrity\n7. **Model Versioning**: Track model profile changes\n8. **Reasoning Budget**: Verify budget enforcement\n9. **Error Handling**: Test error scenarios and recovery\n10. **Performance**: Verify performance metrics collection\n11. **Concurrency**: Test concurrent request handling\n12. **Timeout Handling**: Verify timeout scenarios\n13. **Memory Usage**: Monitor memory consumption\n14. **Disk Space**: Monitor disk usage during large outputs\n15. **Network**: Test network connectivity and latency\n16. **Security**: Verify access controls and authentication\n17. **Backup**: Test backup and restore functionality\n18. **Upgrade**: Test upgrade scenarios\n19. **Downgrade**: Test downgrade scenarios\n20. **Audit Trail**: Verify audit logs for all changes\n\n# Rollback Plan\n\n## Rollback Strategy\n\n1. **Database Rollback**:\n   - Create backup before migration\n   - Restore from backup if migration fails\n   - Use SQLite backup files\n\n2. **Service Rollback**:\n   - Keep previous Docker image tags\n   - Revert to previous configuration\n   - Restore environment variables\n\n3. **Data Recovery**:\n   - Restore from external storage if needed\n   - Rebuild database from logs if possible\n   - Use backup artifacts\n\n## Rollback Steps\n\n1. **Immediate Actions**:\n   - Stop current service\n   - Backup current database\n   - Restore previous database version\n   - Restart service with previous image\n\n2. **Verification**:\n   - Test health endpoint\n   - Verify existing runs are accessible\n   - Confirm no data loss\n   - Validate proxy functionality\n\n3. **Monitoring**:\n   - Monitor service logs\n   - Check for errors\n   - Verify performance metrics\n   - Validate privacy controls\n\n## Rollback Conditions\n\n1. **Database Migration Failure**: If migration fails, rollback to previous version\n2. **Service Crash**: If service crashes during deployment, rollback to stable version\n3. **Data Corruption**: If data corruption detected, rollback to backup\n4. **Performance Degradation**: If performance drops significantly, rollback\n5. **Security Issues**: If security vulnerabilities found, rollback and patch\n\n# Open Questions\n\n1. **Storage Limits**: What are the maximum storage requirements for large outputs?\n2. **Concurrency Limits**: What is the maximum concurrent request handling capacity?\n3. **Timeout Configuration**: What are optimal timeout values for different request types?\n4. **Redaction Granularity**: Should redaction be configurable per request or global?\n5. **Benchmark Scaling**: How should benchmark execution scale with increasing request counts?\n6. **Model Profile Management**: Should model profiles be versioned or managed via Git?\n7. **Artifact Retention**: What retention policies should be implemented?\n8. **Error Recovery**: How should partial failures be handled during benchmark execution?\n9. **Monitoring Integration**: Should we integrate with external monitoring tools?\n10. **User Access**: How should access controls be implemented for different user roles?\n11. **Backup Strategy**: What backup frequency and retention policies should be used?\n12. **Upgrade Path**: How should upgrades be handled for model profiles?\n13. **Performance Tuning**: What performance tuning parameters should be exposed?\n14. **Logging Levels**: What logging levels should be configurable?\n15. **Resource Limits**: How should resource limits be enforced?\n16. **API Stability**: Should we implement API versioning?\n17. **Rate Limiting**: Should rate limiting be implemented?\n18. **Caching Strategy**: What caching mechanisms should be used?\n19. **Error Reporting**: How should errors be reported to users?\n20. **Compliance**: What compliance requirements should be considered for privacy controls?"
}

{
  "message.content": "# Open Questions (continued)\n\n21. **Audit Trail Depth**: How detailed should the audit trail be for model changes?\n22. **Notification System**: Should we implement alerting for critical events?\n23. **Documentation**: What level of documentation should be generated automatically?\n24. **Testing Coverage**: What is the minimum acceptable test coverage?\n25. **Performance Baselines**: Should we establish performance baselines for comparison?\n26. **Resource Monitoring**: How should resource usage be monitored and reported?\n27. **Scalability Testing**: What scalability testing methodology should be used?\n28. **Security Testing**: What security testing should be performed?\n29. **Disaster Recovery**: What disaster recovery procedures should be implemented?\n30. **User Experience**: How should the dashboard UI be optimized for large datasets?\n\n## Detailed Implementation Roadmap\n\n### Phase 1: Core Infrastructure (Week 1-2)\n\n1. **Database Schema Enhancement**:\n   - Implement all new tables and relationships\n   - Add proper indexing for performance\n   - Set up database migration scripts\n   - Test schema compatibility\n\n2. **Artifact Storage Integration**:\n   - Configure MinIO storage\n   - Implement storage client\n   - Test upload/download functionality\n   - Set up retention policies\n\n3. **Privacy Controls**:\n   - Implement redaction logic\n   - Test redaction patterns\n   - Configure logging\n   - Validate redaction effectiveness\n\n### Phase 2: Core Functionality (Week 3-4)\n\n1. **Benchmark Runner Enhancement**:\n   - Implement campaign creation\n   - Add request execution logic\n   - Configure batch processing\n   - Test large-scale execution\n\n2. **Reasoning Budget System**:\n   - Implement tracking mechanism\n   - Add budget enforcement\n   - Configure default budgets\n   - Test budget limits\n\n3. **Reporting System**:\n   - Implement report generation\n   - Add dashboard endpoints\n   - Configure metrics collection\n   - Test report accuracy\n\n### Phase 3: Advanced Features (Week 5-6)\n\n1. **Model Profile Management**:\n   - Implement versioning system\n   - Add change tracking\n   - Configure audit logging\n   - Test profile switching\n\n2. **Performance Monitoring**:\n   - Implement resource monitoring\n   - Add performance metrics\n   - Configure alerting\n   - Test monitoring accuracy\n\n3. **Security Enhancements**:\n   - Implement access controls\n   - Add authentication\n   - Configure encryption\n   - Test security measures\n\n## Configuration Management\n\n### Environment Variables\n\n```bash\n# Environment configuration\nexport DATABASE_URL=sqlite:////app/data/flight_recorder.db\nexport S3_ENDPOINT=http://minio:9000\nexport S3_BUCKET=flight-recorder-artifacts\nexport S3_ACCESS_KEY=minioadmin\nexport S3_SECRET_KEY=minioadmin\nexport MODEL_PROFILE=Qwen3.6-35B-A3B-APEX-MTP-I-Balanced.gguf\nexport LOG_LEVEL=INFO\nexport MAX_ARTIFACT_SIZE=10485760\nexport REASONING_BUDGET=1000\nexport TIMEOUT_SECONDS=300\nexport CONCURRENT_REQUESTS=100\nexport RETENTION_DAYS=30\nexport REDACTION_ENABLED=true\n```\n\n### Configuration File Structure\n\n```yaml\n# config.yaml\napp:\n  name: ai-flight-recorder\n  version: 1.0.0\n  log_level: INFO\n  debug: false\n\nstorage:\n  database:\n    type: sqlite\n    path: /app/data/flight_recorder.db\n  artifact:\n    type: s3\n    endpoint: http://minio:9000\n    bucket: flight-recorder-artifacts\n    access_key: minioadmin\n    secret_key: minioadmin\n    max_size: 10485760\n\nmodel:\n  profile: Qwen3.6-35B-A3B-APEX-MTP-I-Balanced.gguf\n  reasoning_budget: 1000\n  timeout: 300\n  max_tokens: 8192\n\nmonitoring:\n  enabled: true\n  metrics:\n    - request_duration\n    - token_processing\n    - memory_usage\n    - cpu_usage\n  alerts:\n    - threshold: 90\n      metric: memory_usage\n      action: log_warning\n    - threshold: 500\n      metric: request_duration\n      action: log_error\n\nprivacy:\n  enabled: true\n  redaction_patterns:\n    - email\n    - phone\n    - ip\n    - url\n    - credit_card\n    - ssn\n    - teams_id\n  mask: '[REDACTED]'\n  logging:\n    enabled: true\n    level: INFO\n```\n\n## Monitoring and Alerting\n\n### Metrics Collection\n\n```python\n# app/monitoring.py - Metrics collection\nimport time\nfrom prometheus_client import Counter, Histogram, Gauge\n\n# Define metrics\nrequest_counter = Counter('flight_recorder_requests_total', 'Total requests processed')\nresponse_duration = Histogram('flight_recorder_response_duration_seconds', 'Response duration')\nmemory_usage = Gauge('flight_recorder_memory_usage_bytes', 'Memory usage')\ntoken_processing = Histogram('flight_recorder_tokens_processed_total', 'Total tokens processed')\n\n# Collection functions\nasync def collect_metrics(request_data, response_data, duration):\n    request_counter.inc()\n    response_duration.observe(duration)\n    \n    if 'usage' in response_data:\n        usage = response_data['usage']\n        token_processing.observe(usage.get('total_tokens', 0))\n        \n    # Collect memory usage\n    import psutil\n    process = psutil.Process()\n    memory_usage.set(process.memory_info().rss)\n\n# Alerting system\nasync def check_thresholds():\n    current_memory = memory_usage.collect()[0].samples[0].value\n    if current_memory > 0.9 * MAX_MEMORY:\n        logger.warning(f'Memory usage at {current_memory} bytes (90% threshold)')\n        # Send alert\n```\n\n### Alerting Configuration\n\n```yaml\n# alerts.yaml\nalerts:\n  - name: high_memory_usage\n    description: Memory usage exceeds 90% threshold\n    condition: memory_usage > 0.9\n    severity: warning\n    actions:\n      - log_warning\n      - send_email\n      - trigger_incident\n\n  - name: slow_response\n    description: Response duration exceeds 500ms\n    condition: response_duration > 0.5\n    severity: error\n    actions:\n      - log_error\n      - send_slack\n      - trigger_incident\n\n  - name: failed_requests\n    description: High rate of failed requests\n    condition: failed_requests_rate > 0.1\n    severity: critical\n    actions:\n      - log_critical\n      - send_sms\n      - trigger_incident\n```\n\n## Security Considerations\n\n### Access Control\n\n```python\n# app/security.py - Access control\nfrom functools import wraps\nfrom fastapi import HTTPException, status\n\n# Role-based access control\nROLES = {\n    'admin': ['read', 'write', 'delete', 'manage'],\n    'user': ['read', 'write'],\n    'viewer': ['read']\n}\n\ndef require_role(required_role: str):\n    def decorator(func):\n        @wraps(func)\n        async def wrapper(*args, **kwargs):\n            # Check user role\n            user_role = get_current_user_role()\n            if user_role not in ROLES or required_role not in ROLES[user_role]:\n                raise HTTPException(\n                    status_code=status.HTTP_403_FORBIDDEN,\n                    detail=\"Insufficient permissions\"\n                )\n            return await func(*args, **kwargs)\n        return wrapper\n    return decorator\n\n# Example usage\n@app.get(\"/admin/dashboard\", dependencies=[Depends(require_role('admin'))])\nasync def admin_dashboard():\n    return {\"status\": \"admin_access_granted\"}\n```\n\n### Authentication\n\n```python\n# app/auth.py - Authentication\nfrom fastapi import Depends, HTTPException, status\nfrom fastapi.security import HTTPBearer, HTTPAuthorizationCredentials\nimport jwt\n\nsecurity = HTTPBearer()\nSECRET_KEY = \"your-secret-key-here\"\nALGORITHM = \"HS256\"\n\nasync def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)):\n    try:\n        payload = jwt.decode(credentials.credentials, SECRET_KEY, algorithms=[ALGORITHM])\n        return payload\n    except jwt.PyJWTError:\n        raise HTTPException(\n            status_code=status.HTTP_401_UNAUTHORIZED,\n            detail=\"Invalid authentication credentials\"\n        )\n\n# Protected endpoint\n@app.get(\"/protected\")\nasync def protected_route(current_user: dict = Depends(get_current_user)):\n    return {\"user\": current_user[\"sub\"]}\n```\n\n## Performance Optimization\n\n### Caching Strategy\n\n```python\n# app/cache.py - Caching implementation\nimport redis\nimport json\nfrom typing import Any, Optional\n\nclass CacheManager:\n    def __init__(self, redis_url: str):\n        self.redis_client = redis.from_url(redis_url)\n        self.default_ttl = 3600  # 1 hour\n\n    def get(self, key: str) -> Optional[Any]:\n        try:\n            data = self.redis_client.get(key)\n            if data:\n                return json.loads(data)\n        except Exception as e:\n            logger.error(f\"Cache get error for key {key}: {e}\")\n        return None\n\n    def set(self, key: str, value: Any, ttl: int = None):\n        try:\n            self.redis_client.set(\n                key,\n                json.dumps(value),\n                ex=ttl or self.default_ttl\n            )\n        except Exception as e:\n            logger.error(f\"Cache set error for key {key}: {e}\")\n\n    def invalidate(self, pattern: str):\n        try:\n            keys = self.redis_client.keys(pattern)\n            if keys:\n                self.redis_client.delete(*keys)\n        except Exception as e:\n            logger.error(f\"Cache invalidation error: {e}\")\n\n# Global cache instance\ncache_manager = CacheManager(\"redis://localhost:6379\")\n```\n\n### Database Optimization\n\n```python\n# app/database.py - Database optimization\nimport sqlite3\nfrom contextlib import contextmanager\n\n# Connection pooling\nclass DatabaseManager:\n    def __init__(self, db_path: str):\n        self.db_path = db_path\n        self._pool = []\n        self._max_pool_size = 10\n\n    @contextmanager\n    def get_connection(self):\n        conn = None\n        try:\n            conn = sqlite3.connect(self.db_path, check_same_thread=False)\n            conn.row_factory = sqlite3.Row\n            yield conn\n        except Exception as e:\n            if conn:\n                conn.rollback()\n            raise e\n        finally:\n            if conn:\n                conn.close()\n\n    def execute_batch(self, sql: str, params_list: list):\n        with self.get_connection() as conn:\n            cursor = conn.cursor()\n            cursor.executemany(sql, params_list)\n            conn.commit()\n\n# Index optimization\ndef optimize_database_indexes():\n    # Add indexes for frequently queried columns\n    indexes = [\n        'CREATE INDEX IF NOT EXISTS idx_runs_model_profile ON runs(model_profile)',\n        'CREATE INDEX IF NOT EXISTS idx_requests_run_id ON llm_requests(run_id)',\n        'CREATE INDEX IF NOT EXISTS idx_requests_created_at ON llm_requests(created_at)',\n        'CREATE INDEX IF NOT EXISTS idx_events_run_id ON events(run_id)',\n        'CREATE INDEX IF NOT EXISTS idx_events_timestamp ON events(timestamp)'\n    ]\n    \n    with db_manager.get_connection() as conn:\n        for index_sql in indexes:\n            conn.execute(index_sql)\n        conn.commit()\n```\n\n## Backup and Recovery\n\n### Backup Strategy\n\n```python\n# app/backup.py - Backup implementation\nimport shutil\nimport tarfile\nimport datetime\nfrom pathlib import Path\n\nclass BackupManager:\n    def __init__(self, backup_dir: str):\n        self.backup_dir = Path(backup_dir)\n        self.backup_dir.mkdir(exist_ok=True)\n\n    def create_backup(self):\n        timestamp = datetime.datetime.now().strftime(\"%Y%m%d_%H%M%S\")\n        backup_name = f\"flight_recorder_backup_{timestamp}.tar.gz\"\n        backup_path = self.backup_dir / backup_name\n        \n        # Create backup of database and artifacts\n        with tarfile.open(backup_path, \"w:gz\") as tar:\n            # Backup database\n            db_path = Path(\"/app/data/flight_recorder.db\")\n            if db_path.exists():\n                tar.add(db_path, arcname=\"database/flight_recorder.db\")\n            \n            # Backup artifacts directory\n            artifacts_path = Path(\"/app/data/artifacts\")\n            if artifacts_path.exists():\n                tar.add(artifacts_path, arcname=\"artifacts\")\n            \n            # Backup configuration\n            config_path = Path(\"/app/config\")\n            if config_path.exists():\n                tar.add(config_path, arcname=\"config\")\n        \n        return backup_path\n\n    def restore_backup(self, backup_path: str):\n        # Restore from backup\n        with tarfile.open(backup_path, \"r:gz\") as tar:\n            tar.extractall(path=\"/app/restore\")\n        \n        # Copy restored files to correct locations\n        shutil.copy(\"/app/restore/database/flight_recorder.db\", \"/app/data/flight_recorder.db\")\n        shutil.copytree(\"/app/restore/artifacts\", \"/app/data/artifacts\", dirs_exist_ok=True)\n        \n        logger.info(f\"Successfully restored backup from {backup_path}\")\n```\n\n### Recovery Procedures\n\n1. **Database Recovery**:\n   - Restore from latest backup\n   - Verify data integrity\n   - Test database connectivity\n\n2. **Artifact Recovery**:\n   - Restore from external storage\n   - Verify artifact integrity\n   - Rebuild internal references\n\n3. **Configuration Recovery**:\n   - Restore configuration files\n   - Validate settings\n   - Test service startup\n\n4. **Service Recovery**:\n   - Restart services\n   - Validate health checks\n   - Monitor performance\n\n## Testing Framework\n\n### Test Coverage\n\n```python\n# test_coverage.py - Test coverage metrics\nimport coverage\nimport pytest\n\n# Coverage configuration\nCOVERAGE_CONFIG = {\n    'source': ['app'],\n    'omit': [\n        '*/tests/*',\n        '*/venv/*',\n        '*/__pycache__/*',\n        '*/migrations/*'\n    ],\n    'report': {\n        'exclude_lines': [\n            'pragma: no cover',\n            'def __repr__',\n            'raise AssertionError',\n            'raise NotImplementedError'\n        ]\n    }\n}\n\n# Test runner\nasync def run_tests_with_coverage():\n    cov = coverage.Coverage(config_file='.coveragerc')\n    cov.start()\n    \n    # Run tests\n    pytest.main(['-v', '--tb=short', 'tests/'])\n    \n    cov.stop()\n    cov.save()\n    \n    # Generate report\n    cov.report()\n    cov.html_report(directory='htmlcov')\n    \n    return cov.report()\n```\n\n### Integration Test Suite\n\n```python\n# integration_tests.py - Comprehensive integration tests\nimport asyncio\nimport pytest\nfrom app.main import app\nfrom app.bench import BenchmarkRunner\nfrom app.store import RunStore, LLMRequestStore\n\n@pytest.fixture\nasync def client():\n    async with httpx.AsyncClient(app=app, base_url=\"http://test\") as client:\n        yield client\n\n@pytest.mark.asyncio\nasync def test_complete_workflow(client):\n    # Test full workflow: create run -> execute request -> store result -> generate report\n    \n    # Create run\n    run_response = await client.post(\"/runs\", json={\n        \"name\": \"test_run\",\n        \"model_profile\": \"Qwen3.6-35B-A3B-APEX-MTP-I-Balanced.gguf\"\n    })\n    assert run_response.status_code == 200\n    run_id = run_response.json()[\"id\"]\n    \n    # Execute request\n    request_response = await client.post(\"/v1/chat/completions\", json={\n        \"model\": \"Qwen3.6-35B-A3B-APEX-MTP-I-Balanced.gguf\",\n        \"messages\": [{\"role\": \"user\", \"content\": \"Hello world\"}]\n    })\n    assert request_response.status_code == 200\n    \n    # Verify request stored\n    request_data = await client.get(f\"/runs/{run_id}/requests\")\n    assert request_data.status_code == 200\n    assert len(request_data.json()) > 0\n    \n    # Generate report\n    report_response = await client.get(f\"/dashboard/runs/{run_id}/report\")\n    assert report_response.status_code == 200\n    assert \"run_id\" in report_response.json()\n\n@pytest.mark.asyncio\nasync def test_benchmark_execution(client):\n    # Test benchmark execution\n    campaign_response = await client.post(\"/bench/campaigns\", json={\n        \"name\": \"test_benchmark\",\n        \"model_profile\": \"Qwen3.6-35B-A3B-APEX-MTP-I-Balanced.gguf\",\n        \"requests\": 5\n    })\n    assert campaign_response.status_code == 200\n    campaign_id = campaign_response.json()[\"id\"]\n    \n    # Wait for completion\n    await asyncio.sleep(5)\n    \n    # Verify campaign completed\n    campaign_status = await client.get(f\"/bench/campaigns/{campaign_id}\")\n    assert campaign_status.status_code == 200\n    assert campaign_status.json()[\"status\"] == \"completed\"\n```\n\n## Documentation and Maintenance\n\n### API Documentation\n\n```python\n# app/docs.py - API documentation\nfrom fastapi import FastAPI\nfrom fastapi.openapi.utils import get_openapi\n\napp = FastAPI(title=\"AI Flight Recorder API\", version=\"1.0.0\")\n\n# Custom OpenAPI schema\ndef custom_openapi():\n    if app.openapi_schema:\n        return app.openapi_schema\n    \n    openapi_schema = get_openapi(\n        title=\"AI Flight Recorder API\",\n        version=\"1.0.0\",\n        description=\"API for AI model benchmarking and monitoring\",\n        routes=app.routes,\n    )\n    \n    # Add custom components\n    openapi_schema[\"components\"][\"securitySchemes\"] = {\n        \"BearerAuth\": {\n            \"type\": \"http\",\n            \"scheme\": \"bearer\"\n        }\n    }\n    \n    app.openapi_schema = openapi_schema\n    return app.openapi_schema\n\napp.openapi = custom_openapi\n```\n\n### Maintenance Procedures\n\n1. **Regular Database Maintenance**:\n   - Vacuum database periodically\n   - Optimize indexes\n   - Monitor disk space\n   - Backup database regularly\n\n2. **Service Monitoring**:\n   - Monitor resource usage\n   - Check service health\n   - Verify connectivity\n   - Test API endpoints\n\n3. **Security Updates**:\n   - Monitor for security vulnerabilities\n   - Update dependencies\n   - Review access controls\n   - Test security measures\n\n4. **Performance Tuning**:\n   - Monitor performance metrics\n   - Optimize slow queries\n   - Adjust resource limits\n   - Test scaling scenarios\n\nThis comprehensive engineering dossier provides a complete roadmap for deploying and maintaining a robust AI benchmarking system. The implementation addresses all identified pain points while maintaining backward compatibility and providing extensibility for future enhancements."
}