OctoLLM Documentation

Welcome to the OctoLLM comprehensive technical documentation. This guide covers the complete architecture, implementation, API reference, and operational workflows for the distributed AI system.

What is OctoLLM?

OctoLLM is a novel distributed AI architecture inspired by octopus neurobiology, designed specifically for offensive security operations and advanced developer tooling. By modeling cognitive processing after the octopus's distributed nervous system—where each arm possesses autonomous decision-making capabilities coordinated by a central brain—OctoLLM achieves superior modularity, security isolation, and operational efficiency compared to monolithic LLM systems.

Core Innovation

Rather than relying on a single large language model to handle all tasks, OctoLLM employs specialized "arm" modules that operate semi-autonomously under the guidance of a central "brain" orchestrator. This architecture enables:

• Enhanced Security: Capability isolation and compartmentalization prevent lateral movement of compromised components
• Cost Efficiency: Lightweight reflexes and specialized models handle routine tasks without engaging expensive central processing
• Operational Resilience: Individual component failures don't cascade through the system
• Rapid Adaptation: New capabilities can be added as independent modules without system-wide reengineering

System Architecture

Core Components

Layer Architecture

Layer 1: Ingress (API Gateway + Reflex)

• Technology: NGINX/Traefik + Rust
• Latency Target: <10ms cache hits, <50ms reflex decisions

Layer 2: Orchestration (The Brain)

• Technology: Python + FastAPI, LangChain
• Main Loop: Cache → Plan → Execute → Integrate → Validate

Layer 3: Execution (The Arms)

• Planner: Task decomposition
• Tool Executor: Sandboxed external actions
• Retriever: Knowledge base search
• Coder: Code generation/debugging
• Judge: Output validation
• Safety Guardian: PII detection, content filtering

Layer 4: Persistence

• PostgreSQL (global memory), Redis (caching), Qdrant (vectors)

Layer 5: Observability

• Prometheus (metrics), Loki (logs), Jaeger (tracing)

Current Status

Phase: Phase 0 (Architecture) → Phase 1 (Proof of Concept) Sprint: Sprint 1.2 COMPLETE (Orchestrator Core v1.2.0) Progress: ~22% overall, Phase 1 ~40%

Completed Components

✅ Phase 0: Complete architecture, documentation, specifications (100%) ✅ Sprint 1.1: Reflex Layer production-ready (v1.1.0)

• Cache hit latency: <5ms (2x better than target)
• Pattern match latency: <8ms (6x better than target)
• Memory usage: ~12MB (4x better than target)

✅ Sprint 1.2: Orchestrator Core production-ready (v1.2.0)

• 1,776 lines Python code
• 2,776 lines tests (87 tests, 87% pass rate, 85%+ coverage)
• 6 REST endpoints operational
• API latency P95: <100ms (5x better than target)
• Database query P95: <5ms (2x better than target)

In Progress

🚧 Sprint 1.3: Planner Arm (PLANNED)

• Task decomposition into subtasks
• Acceptance criteria generation
• Resource estimation

Documentation Structure

This documentation is organized into the following major sections:

1. Project Overview

• Vision, goals, and success metrics
• Biological inspiration from octopus neurobiology
• Core concepts and design principles
• Complete roadmap (7 phases)

2. Architecture

• System architecture and layer design
• Data structures (TaskContract, ArmCapability, Memory Models)
• Data flow and swarm decision-making
• Architecture Decision Records (ADRs)

3. Components

• Reflex Layer (preprocessing and caching)
• Orchestrator (central coordination)
• All 6 Arms (specialized modules)
• Persistence layer

4. API Documentation

• REST API overview and contracts
• OpenAPI 3.0 specifications for all services
• Data models and schemas
• Authentication and error handling

5. Development

• Getting started guide
• Development environment setup
• Testing strategies and debugging
• Custom arm development
• Contributing guidelines

6. Operations

• Deployment guides (Docker Compose, Kubernetes, Unraid)
• Monitoring and alerting setup
• Troubleshooting playbooks
• Performance tuning and scaling

7. Security

• Security model and threat model
• Capability isolation and PII protection
• Secrets management
• Security testing and compliance

8. Sprint Progress

• Phase 0 sprints (0.1-0.7) - Complete
• Phase 1 sprints (1.1-1.3) - In progress
• Sprint completion reports with metrics

9. Project Tracking

• Master TODO with all 7 phases
• Roadmap and phase details
• Current status and checklists

10. Reference

• Configuration reference
• Glossary and diagrams
• Documentation summary

Quick Links

For New Users

• Getting Started - Setup and installation
• Core Concept - Understanding the architecture
• Quickstart Guide - Run your first task

For Developers

• Development Environment - Python/Rust setup
• Testing Guide - Unit/integration tests
• Custom Arms - Build new specialized modules
• Contributing - How to contribute

For Operators

• Docker Compose Setup - Local deployment
• Kubernetes Deployment - Production deployment
• Monitoring Runbook - Operations guide
• Troubleshooting Playbooks - Common issues

For Security Engineers

• Security Overview - Security architecture
• Threat Model - Attack vectors and mitigations
• Security Testing - Security test suite

Key Metrics

Repository

GitHub: github.com/doublegate/OctoLLM Documentation: doublegate.github.io/OctoLLM

Navigation

Use the sidebar to explore the documentation. All pages include:

• Links to source code in the repository
• Related documentation pages
• API references where applicable
• Version information

Need help? Check the Troubleshooting Playbooks or review the FAQ section.

Want to contribute? See the Contributing Guide.

Vision & Goals

Extracted from: ref-docs/OctoLLM-Project-Overview.md

Executive Summary

OctoLLM is a novel distributed AI architecture inspired by octopus neurobiology, designed specifically for offensive security operations and advanced developer tooling. By modeling cognitive processing after the octopus's distributed nervous system—where each arm possesses autonomous decision-making capabilities coordinated by a central brain—OctoLLM achieves superior modularity, security isolation, and operational efficiency compared to monolithic LLM systems.

Core Innovation

Rather than relying on a single large language model to handle all tasks, OctoLLM employs specialized "arm" modules that operate semi-autonomously under the guidance of a central "brain" orchestrator. This architecture enables:

• Enhanced Security: Capability isolation and compartmentalization prevent lateral movement of compromised components
• Cost Efficiency: Lightweight reflexes and specialized models handle routine tasks without engaging expensive central processing
• Operational Resilience: Individual component failures don't cascade through the system
• Rapid Adaptation: New capabilities can be added as independent modules without system-wide reengineering

Target Applications

Offensive Security Operations

OctoLLM is purpose-built for red team operations, penetration testing, and vulnerability research:

• Automated Reconnaissance: Web scraping, OSINT gathering, attack surface mapping
• Vulnerability Analysis: Static/dynamic code analysis, fuzzing orchestration, exploit development
• Attack Simulation: Adversary emulation, lateral movement planning, evasion technique selection
• Post-Exploitation: Data exfiltration planning, persistence mechanisms, cleanup automation
• Reporting: Evidence compilation, timeline generation, remediation recommendations

Security Isolation: Each capability operates in a sandboxed environment with minimal privileges, preventing accidental damage to production systems or unintended escalation.

Advanced Developer Tooling

Beyond security, OctoLLM excels at complex software development tasks:

• Codebase Analysis: Dependency mapping, technical debt assessment, refactoring planning
• Automated Testing: Test generation, coverage analysis, regression detection
• Documentation: API documentation, architecture diagrams, onboarding guides
• DevOps Automation: CI/CD pipeline optimization, infrastructure-as-code generation
• Code Review: Security audit, performance optimization, best practice enforcement

Advantage: Specialized arms for each language/framework provide expert-level assistance without the context pollution of general-purpose models.

Success Metrics

See Also

• Biological Inspiration - Octopus neurobiology mapping
• Core Concept - Concrete design patterns
• Project Roadmap - Implementation timeline

Core Concept

Extracted from: ref-docs/OctoLLM-Concept_Idea.md

Architectures to Borrow from the Octopus

1. Local-Autonomy "Arms," Central-Integration "Brain"

• Spin up task-specific peripheral controllers (code tools, web searchers, planners, UI drivers, data labelers) with narrow policies and short-term memory.
• A central integrator (LLM) sets intent, allocates subtasks, imposes constraints, and fuses results—only intervening when goals or safety are at stake.
• Mechanism: hierarchical control + explicit contracts (inputs/outputs/invariants). Think: Mixture-of-Experts + Orchestrator rather than a single giant monolith.

2. Reflex Layer Before Cognition

• Pre-LLM reflex filters handle fast, predictable decisions (schema validation, PII/safety checks, rate limiting, cache hits) using small models/finite-state machines.
• The LLM only engages for "novelty." This reduces latency, cost, and attack surface.

3. Decentralized Memory

• Each arm has a local episodic store (vector DB or KV cache) bounded by its domain ontology; the brain has a global semantic map.
• Routing: classifier/gating picks which memories to consult.
• Prevents cross-domain contamination and keeps retrieval precise.

4. Embodied Tool-Use

• Treat tools as sensors/actuators. The arm owns its tools (APIs, shells, browsers), maintains affordances/capabilities metadata, and reports action traces upward.
• The brain reasons over traces, not raw environments—like a commander reading squad reports.

5. Elastic Specialization via MoE + Skill Distillation

• Train small specialists per domain (planning, SQL, regex, code fixes, UI automation); distill their strengths back into a generalist for robustness while keeping specialists online for hard cases.
• Gate by uncertainty/entropy or cost budget.

6. Swarm Deliberation with Quorum

• For critical decisions, run N lightweight "arm" proposals (diverse prompts/seeds/models), aggregate with verifiable voters (majority, Borda, or learned ranker).
• The brain resolves conflicts using explicit rules (risk thresholds, SLAs).

7. Active Inference for Exploration

• Arms maintain simple world models and choose actions that reduce expected uncertainty (information gain) subject to task goals.
• Great for web research agents and code-repair loops.

Concrete System Design (Drop-In Blueprint)

Orchestrator (Brain)

One robust LLM with a Task Contract Schema:

• goal, constraints, budget (tokens/time/$), security policy, deliverables, acceptance tests.

Arms (Specialists)

• Planner: Decomposes tasks → subgoals + acceptance criteria.
• Retriever: Structured + vector search with domain ontologies.
• Tool-Executor: Browser/API/shell; enforces allowlists; captures provenance.
• Coder: Patch proposals + self-tests.
• Judge: Spec compliance, hallucination detection, unit/property checks.
• Safety/PII Guardian: Static rules + tiny classifier; runs before and after LLM calls.

Memories

• Local: Per-arm episodic stores (short retention, domain schema).
• Global: Project knowledge graph (entities, tasks, decisions, citations).

Control

• Reflex gate → Arm(s) → Orchestrator escalate-on-novelty.
• Uncertainty triggers: escalate, fork more arms, or ask for user input (with minimally sufficient questions).

Provenance

Every artifact tagged with tool, prompt hash, data source, time, and tests passed.

Quick-Start Experiments You Can Run This Week

1. Reflex gate + cache: Put a rules/regex/PII filter + embedding cache in front of your LLM; measure latency/cost drop on your common tickets.
2. Two-arm prototype: Planner → Tool-Executor (browser or repo) with a Judge. Orchestrator only resolves conflicts.
3. Specialist MoE: Add a code-fix small model (e.g., 1–3B) gated by a classifier; fall back to the big model on low confidence.
4. Decentralized memory: Split your RAG into per-domain stores; add a router; watch precision improve and leakage drop.
5. Quorum for critical ops: Require 3 proposals for risky actions; aggregate; compare error rates.

See Also

• Architecture Overview - Full technical architecture
• Data Structures - TaskContract, ArmCapability schemas
• Getting Started - Implementation guide

Biological Inspiration

Extracted from: ref-docs/OctoLLM-Project-Overview.md

Distributed Intelligence in Nature

The octopus represents one of nature's most remarkable examples of distributed cognition:

• Neuron Distribution: Approximately 500 million neurons total, with over 350 million (70%) residing in the arms rather than the central brain
• Autonomous Arms: Each arm can independently sense, process information, and execute complex motor sequences
• Neural Ring: Arms communicate directly via a neural ring, enabling coordination without constant brain involvement
• Parallel Processing: Multiple arms can simultaneously pursue different strategies or explore separate options
• Central Coordination: The brain sets high-level goals and resolves conflicts when arms have competing priorities

Translation to AI Architecture

OctoLLM maps these biological principles to artificial intelligence:

Differentiation from Other Approaches

This architecture is fundamentally different from:

• Monolithic LLMs: Single model attempts all tasks (inefficient, insecure)
• Simple RAG Systems: Retrieval augmentation but no true modularity
• Basic Tool-Use: LLM directly manipulates tools (security risk, tight coupling)

OctoLLM combines the best of all approaches while adding critical security isolation and operational efficiency.

See Also

• System Architecture - Technical implementation
• Swarm Decision Making - Parallel processing details

Project Roadmap

OctoLLM development follows a 7-phase roadmap from architecture to production deployment.

Overall Timeline

Estimated Total Time: 36-48 weeks (8-11 months) Estimated Total Hours: ~1,186 development hours Current Progress: ~22% (Phase 0 complete, Phase 1 40%)

Phase Overview

Phase 0: Project Setup

Status: ✅ COMPLETE (100%) Duration: 2025-11-10 to 2025-11-13

Deliverables

• ✅ Repository structure and Git workflow
• ✅ CI/CD pipeline (GitHub Actions)
• ✅ Complete documentation (170+ files)
• ✅ Architecture specifications
• ✅ OpenAPI specs for all services
• ✅ Security audit framework

Phase 1: Proof of Concept

Status: 🚧 IN PROGRESS (40%) Start: 2025-11-14

Completed

• ✅ Sprint 1.1: Reflex Layer (v1.1.0)
• ✅ Sprint 1.2: Orchestrator Core (v1.2.0)

Remaining

• 🚧 Sprint 1.3: Planner Arm (PLANNED)
• ⏳ Sprint 1.4: Tool Executor Arm
• ⏳ Sprint 1.5: Integration Testing

Details: Phase 1 Tracking

Phase 2: Core Capabilities

Status: ⏳ NOT STARTED Dependencies: Phase 1 complete

Goals

• All 6 arms operational (Planner, Executor, Retriever, Coder, Judge, Safety Guardian)
• Distributed memory system
• Swarm decision-making
• Advanced error handling

Details: Phase 2 Tracking

Phase 3: Operations & Deployment

Status: ⏳ NOT STARTED Dependencies: Phase 2 complete

Goals

• Kubernetes deployment
• Monitoring stack (Prometheus, Grafana, Loki, Jaeger)
• Scaling and performance tuning
• Operational runbooks

Details: Phase 3 Tracking

Phase 4: Engineering & Standards

Status: ⏳ NOT STARTED Dependencies: Phase 3 complete

Goals

• Code review processes
• Engineering standards
• Performance optimization
• Technical debt management

Details: Phase 4 Tracking

Phase 5: Security Hardening

Status: ⏳ NOT STARTED Dependencies: Phase 4 complete

Goals

• Comprehensive security testing
• Penetration testing
• Compliance certifications (SOC 2, ISO 27001)
• Vulnerability management

Details: Phase 5 Tracking

Phase 6: Production Readiness

Status: ⏳ NOT STARTED Dependencies: Phase 5 complete

Goals

• Production deployment
• Public API
• Documentation for external users
• SLA and support setup

Details: Phase 6 Tracking

Critical Milestones

• Week 3 (✅ DONE): Development environment ready, first code commit
• Week 10: POC complete, basic orchestrator + 2 arms functional
• Week 20: All 6 arms operational, distributed memory working
• Week 26: Kubernetes deployment, monitoring stack operational
• Week 34: Security hardening complete, penetration tests passed
• Week 42: Production-ready, compliance certifications in progress

See Also

• Master TODO - Complete task breakdown
• Sprint Overview - Sprint-by-sprint progress
• Current Status - Latest progress

System Architecture Overview

OctoLLM implements a five-layer architecture inspired by octopus neurobiology, combining distributed intelligence with centralized governance.

Architecture Layers

Layer 1: Ingress (API Gateway + Reflex)

Purpose: Fast preprocessing and caching before expensive LLM processing.

Technology: NGINX/Traefik + Rust Latency Target: <10ms cache hits, <50ms reflex decisions Current Status: ✅ COMPLETE (Sprint 1.1, v1.1.0)

Key Features:

• Redis caching with <5ms latency (2x better than target)
• Pattern matching and PII detection <8ms (6x better than target)
• Request routing based on complexity
• Rate limiting and input validation

Details: Reflex Layer Component

Layer 2: Orchestration (The Brain)

Purpose: Strategic planning, task decomposition, and arm coordination.

Technology: Python + FastAPI, LangChain/LlamaIndex Model: GPT-4 or Claude Opus Current Status: ✅ COMPLETE (Sprint 1.2, v1.2.0)

Main Loop:

1. Cache check (via Reflex Layer)
2. Plan generation (task decomposition)
3. Step execution (arm delegation)
4. Result integration (combining outputs)
5. Validation (quality assurance)

Details: Orchestrator Component

Layer 3: Execution (The Arms)

Purpose: Domain-specific execution with local decision-making.

Arms Implemented:

• ✅ Reflex Layer (v1.1.0) - Pattern matching, caching
• ✅ Orchestrator (v1.2.0) - Coordination, planning
• 🚧 Planner Arm (Planned Sprint 1.3) - Task decomposition
• ⏳ Tool Executor - Sandboxed command execution
• ⏳ Retriever - Knowledge base search
• ⏳ Coder - Code generation/debugging
• ⏳ Judge - Output validation
• ⏳ Safety Guardian - PII detection, filtering

Details: Arms Overview

Layer 4: Persistence

Purpose: Global memory, caching, and vector stores.

Components:

• PostgreSQL: Global semantic memory (tasks, decisions, provenance)
• Redis: High-speed caching (responses, embeddings)
• Qdrant/Weaviate: Vector stores for semantic search

Current Status: ✅ PostgreSQL + Redis operational (Sprint 1.2)

Layer 5: Observability

Purpose: Monitoring, logging, and tracing for debugging and optimization.

Stack:

• Prometheus: Metrics collection (latency, throughput, errors)
• Loki: Centralized logging
• Jaeger: Distributed tracing
• Grafana: Dashboards and alerting

Current Status: ⏳ Planned (Phase 3)

Data Flow

User Request
    ↓
[API Gateway] → Reflex Layer (cache check, pattern match)
    ↓
[Orchestrator] (task decomposition, planning)
    ↓
[Arms] (parallel execution, specialized processing)
    ↓
[Orchestrator] (result aggregation, validation)
    ↓
[API Gateway] → User Response

Detailed flow: Data Flow Documentation

Key Design Principles

1. Modular Specialization: Each component excels at one thing
2. Distributed Autonomy with Centralized Governance: Arms decide locally, brain coordinates globally
3. Defense in Depth: Multiple security layers (reflex, capability isolation, PII sanitization)
4. Hierarchical Processing: Expensive resources reserved for complex problems
5. Active Inference: System proactively reduces uncertainty

Details: Architecture Principles

Performance Metrics

See Also

• Layer Architecture Details
• Data Structures
• Swarm Decision Making
• Architecture Decision Records

Layer Architecture

Detailed documentation of OctoLLM's five-layer architecture.

Layer 1: Ingress Layer

Components: API Gateway, Reflex Layer Technology: NGINX/Traefik + Rust Latency Target: <10ms cache, <50ms reflex

The ingress layer handles all incoming requests with fast preprocessing before expensive LLM processing.

Details: Reflex Layer

Layer 2: Orchestration Layer

Components: Orchestrator service Technology: Python + FastAPI, GPT-4/Claude Opus Latency Target: <500ms API calls

Strategic planning and coordination of all arms.

Details: Orchestrator

Layer 3: Execution Layer

Components: 6 specialized Arms Technology: Python/Rust, various LLMs Latency Target: Varies by arm

Domain-specific execution with local autonomy.

Details: Arms

Layer 4: Persistence Layer

Components: PostgreSQL, Redis, Qdrant/Weaviate Technology: Databases and vector stores

Global and local memory storage.

Details: Persistence

Layer 5: Observability Layer

Components: Prometheus, Loki, Jaeger, Grafana Technology: Monitoring stack

Metrics, logs, and traces for debugging.

Details: Monitoring

See Also

• Architecture Overview
• Data Flow
• System Design

Ingress Layer

Orchestration Layer

Execution Layer

Persistence Layer

Observability Layer

Data Structures

Core data structures used throughout the OctoLLM system for task management, arm coordination, and memory persistence.

TaskContract

Central data structure representing a task with all its requirements, constraints, and context.

from dataclasses import dataclass
from typing import Dict, List, Any, Optional

@dataclass
class ResourceBudget:
    max_tokens: Optional[int] = None
    max_time_seconds: Optional[int] = None
    max_cost_dollars: Optional[float] = None
    max_llm_calls: Optional[int] = None

@dataclass
class TaskContract:
    task_id: str
    goal: str  # Natural language description
    constraints: Dict[str, Any]  # Hard constraints
    context: Dict[str, Any]  # Background information
    acceptance_criteria: List[str]  # Success conditions
    budget: ResourceBudget  # Resource limits
    assigned_arm: Optional[str] = None
    parent_task_id: Optional[str] = None
    priority: int = 5  # 1 (highest) to 10 (lowest)
    security_policy: Optional[str] = None

Usage: Created by Orchestrator during task decomposition, passed to Arms for execution.

Schema Details

ArmCapability

Describes an arm's capabilities, interface, and resource requirements.

@dataclass
class ArmCapability:
    arm_id: str
    name: str
    description: str
    input_schema: JSONSchema  # Pydantic model or JSON schema
    output_schema: JSONSchema
    capabilities: List[str]  # Tags for routing (e.g., "code", "security")
    cost_tier: int  # 1 (cheap) to 5 (expensive)
    endpoint: str  # Kubernetes service URL
    health_check_url: str
    timeout_seconds: int = 30
    retry_policy: Optional[Dict] = None

Usage: Registered in Arm Registry, used by Orchestrator for routing decisions.

Schema Details

Memory Models

Global Semantic Memory

Stored in PostgreSQL, represents project-wide knowledge.

@dataclass
class SemanticMemory:
    memory_id: str
    entity_type: str  # "task", "decision", "fact", "artifact"
    content: str
    embeddings: List[float]  # For semantic search
    metadata: Dict[str, Any]
    source: str  # Which arm created this
    timestamp: datetime
    confidence: float  # 0.0 to 1.0
    tags: List[str]

Local Episodic Memory

Stored in Redis, arm-specific short-term memory.

@dataclass
class EpisodicMemory:
    episode_id: str
    arm_id: str
    task_id: str
    observations: List[str]
    actions: List[str]
    outcomes: List[str]
    ttl_seconds: int = 3600  # 1 hour default

Response Models

Execution Result

@dataclass
class ExecutionResult:
    task_id: str
    arm_id: str
    status: str  # "success", "failure", "partial"
    output: Any  # Arm-specific output
    confidence: float  # 0.0 to 1.0
    execution_time_ms: int
    tokens_used: int
    error: Optional[str] = None
    provenance: ProvenanceMetadata

Provenance Metadata

@dataclass
class ProvenanceMetadata:
    arm_id: str
    timestamp: datetime
    command_hash: str  # SHA256 of command executed
    data_sources: List[str]  # URLs, file paths, etc.
    model_version: Optional[str] = None
    tests_passed: List[str] = []

See Also

• Component Contracts
• OpenAPI Specifications
• Memory Systems

TaskContract

ArmCapability

Memory Models

OctoLLM Data Flow Architecture

Version: 1.0 Last Updated: 2025-11-10

Table of Contents

• Overview
• Request Processing Pipeline
• Memory Data Flow
• Inter-Component Communication
• Provenance Tracking
• Error Handling Flow

Overview

This document details how data flows through the OctoLLM system, from initial user request to final response, including memory operations, inter-component communication, and error handling.

Request Processing Pipeline

Complete Flow

flowchart TD
    START([User Request]) --> AUTH{Authenticated?}
    AUTH -->|No| REJECT([401 Unauthorized])
    AUTH -->|Yes| RATE{Within Rate Limit?}

    RATE -->|No| THROTTLE([429 Too Many Requests])
    RATE -->|Yes| REFLEX[Reflex Layer]

    REFLEX --> CACHE{Cache Hit?}
    CACHE -->|Yes| RETURN_CACHE([Return Cached Result])
    CACHE -->|No| PII[PII Detection]

    PII --> INJECT{Injection Detected?}
    INJECT -->|Yes| BLOCK([403 Blocked])
    INJECT -->|No| SANITIZE[Sanitize Input]

    SANITIZE --> ORCH[Orchestrator]
    ORCH --> PARSE[Parse Intent]
    PARSE --> COMPLEXITY{Complex Task?}

    COMPLEXITY -->|Yes| PLANNER[Planner Arm]
    COMPLEXITY -->|No| DIRECT[Direct Execution]

    PLANNER --> PLAN[Generate Plan]
    PLAN --> ROUTE[Route to Arms]

    ROUTE --> EXEC_LOOP{More Steps?}
    EXEC_LOOP -->|Yes| SELECT_ARM[Select Arm]

    SELECT_ARM --> ARM_TYPE{Arm Type}
    ARM_TYPE -->|Retriever| RETR[Retriever Arm]
    ARM_TYPE -->|Coder| CODE[Coder Arm]
    ARM_TYPE -->|Executor| EXEC[Executor Arm]

    RETR --> ARM_RESULT[Arm Result]
    CODE --> ARM_RESULT
    EXEC --> ARM_RESULT
    DIRECT --> ARM_RESULT

    ARM_RESULT --> STORE_LOCAL[Store in Local Memory]
    STORE_LOCAL --> UPDATE_CONTEXT[Update Task Context]
    UPDATE_CONTEXT --> EXEC_LOOP

    EXEC_LOOP -->|No| INTEGRATE[Integrate Results]
    INTEGRATE --> JUDGE[Judge Arm Validation]

    JUDGE --> VALID{Valid?}
    VALID -->|No| REPAIR[Repair Loop]
    REPAIR --> RETRY{Max Retries?}
    RETRY -->|No| INTEGRATE
    RETRY -->|Yes| ERROR([Error Response])

    VALID -->|Yes| STORE_GLOBAL[Store in Global Memory]
    STORE_GLOBAL --> CACHE_RESULT[Cache Result]
    CACHE_RESULT --> RESPONSE([Return to User])

Layer-by-Layer Processing

Layer 1: API Gateway

sequenceDiagram
    participant User
    participant Gateway as API Gateway
    participant Auth as Auth Service
    participant RateLimit as Rate Limiter

    User->>Gateway: HTTP Request
    Gateway->>Auth: Validate Token
    Auth-->>Gateway: Valid/Invalid

    alt Invalid Token
        Gateway-->>User: 401 Unauthorized
    else Valid Token
        Gateway->>RateLimit: Check Limit
        RateLimit-->>Gateway: Allow/Deny

        alt Rate Limited
            Gateway-->>User: 429 Too Many Requests
        else Allowed
            Gateway->>Gateway: Add Request Metadata
            Note over Gateway: request_id, timestamp,<br/>user_id, trace_id
            Gateway-->>User: Forward to Reflex
        end
    end

Layer 2: Reflex Preprocessing

flowchart LR
    INPUT[Incoming Request] --> HASH[Compute Hash]
    HASH --> CACHE_LOOKUP{Redis Cache}

    CACHE_LOOKUP -->|Hit| METRICS1[Increment cache_hit]
    METRICS1 --> RETURN1[Return Cached]

    CACHE_LOOKUP -->|Miss| INJECT_CHECK[Injection Pattern Check]
    INJECT_CHECK -->|Match| BLOCK[Block Request]
    BLOCK --> METRICS2[Increment blocked]

    INJECT_CHECK -->|Clean| PII_CHECK[PII Pattern Scan]
    PII_CHECK --> REDACT[Redact/Sanitize]
    REDACT --> SCHEMA[Schema Validation]

    SCHEMA -->|Invalid| REJECT[Return 400]
    SCHEMA -->|Valid| FORWARD[Forward to Orchestrator]
    FORWARD --> METRICS3[Increment passthrough]

Reflex Decision Matrix:

Layer 3: Orchestrator Planning

flowchart TD
    INPUT[Sanitized Request] --> PARSE[Parse Goal & Constraints]
    PARSE --> CONTEXT[Build Task Context]

    CONTEXT --> CACHED_PLAN{Similar Plan Exists?}
    CACHED_PLAN -->|Yes| ADAPT[Adapt Cached Plan]
    CACHED_PLAN -->|No| NEW_PLAN[Generate New Plan]

    ADAPT --> PLAN_READY[Plan Ready]
    NEW_PLAN --> LLM{Use LLM or Planner Arm?}

    LLM -->|Simple| DIRECT_LLM[Direct LLM Call]
    LLM -->|Complex| PLANNER_ARM[Planner Arm Call]

    DIRECT_LLM --> PARSE_PLAN[Parse Plan JSON]
    PLANNER_ARM --> PARSE_PLAN

    PARSE_PLAN --> VALIDATE_PLAN{Plan Valid?}
    VALIDATE_PLAN -->|No| REPLAN[Retry Planning]
    REPLAN --> LLM

    VALIDATE_PLAN -->|Yes| RESOLVE_DEPS[Resolve Dependencies]
    RESOLVE_DEPS --> PLAN_READY

    PLAN_READY --> EXECUTE[Execute Plan]

Planning Decision Criteria:

def should_use_planner_arm(task):
    # Use dedicated Planner Arm if:
    return (
        len(task.constraints) > 3 or
        task.priority == Priority.HIGH or
        estimate_steps(task) > 5 or
        has_complex_dependencies(task) or
        requires_specialized_domain_knowledge(task)
    )

Layer 4: Arm Execution

sequenceDiagram
    participant Orch as Orchestrator
    participant Router as Router
    participant ArmReg as Arm Registry
    participant Arm as Selected Arm
    participant LocalMem as Local Memory
    participant GlobalMem as Global Memory

    Orch->>Router: Route Step
    Router->>ArmReg: Get Capabilities
    ArmReg-->>Router: Arm Metadata

    Router->>Router: Score Arms
    Note over Router: Consider: cost, latency,<br/>success rate, load

    Router-->>Orch: Selected Arm(s)

    alt Single Arm
        Orch->>Arm: Execute Task
        Arm->>LocalMem: Query Context
        LocalMem-->>Arm: Local Context
        Arm->>Arm: Process
        Arm-->>Orch: Result + Confidence
    else Swarm (Multiple Arms)
        par Parallel Execution
            Orch->>Arm: Execute Task
            Arm->>LocalMem: Query Context
            Arm->>Arm: Process
            Arm-->>Orch: Result A
        and
            Orch->>Arm: Execute Task
            Arm->>LocalMem: Query Context
            Arm->>Arm: Process
            Arm-->>Orch: Result B
        and
            Orch->>Arm: Execute Task
            Arm->>LocalMem: Query Context
            Arm->>Arm: Process
            Arm-->>Orch: Result C
        end
        Orch->>Orch: Aggregate Results
        Note over Orch: Vote, average,<br/>or learned aggregation
        Orch-->>Orch: Consensus Result
    end

    Orch->>GlobalMem: Update Knowledge Graph

Memory Data Flow

Write Operations

flowchart TD
    ARM_RESULT[Arm Produces Result] --> PROV[Attach Provenance]
    PROV --> CLASS{Classify Data}

    CLASS -->|Ephemeral| TEMP[Discard After Task]
    CLASS -->|Local| LOCAL_WRITE[Write to Local Memory]
    CLASS -->|Global| GLOBAL_WRITE[Write to Global Memory]

    LOCAL_WRITE --> VECTOR[Vectorize if Text]
    VECTOR --> QDRANT[Store in Qdrant]
    QDRANT --> INDEX[Update Index]

    GLOBAL_WRITE --> SANITIZE[PII Sanitization]
    SANITIZE --> EXTRACT[Extract Entities/Relations]
    EXTRACT --> PSQL[PostgreSQL Write]
    PSQL --> UPDATE_GRAPH[Update Knowledge Graph]

    INDEX --> CACHE_INV[Invalidate Related Cache]
    UPDATE_GRAPH --> CACHE_INV

Read Operations

flowchart LR
    QUERY[Memory Query] --> L1{L1: Redis Cache}
    L1 -->|Hit| RETURN1[Return Result]
    L1 -->|Miss| L2{L2: Local Arm Memory}

    L2 -->|Hit| PROMOTE1[Promote to L1]
    PROMOTE1 --> RETURN2[Return Result]

    L2 -->|Miss| L3{L3: Global Knowledge Graph}
    L3 -->|Hit| PROMOTE2[Promote to L2 & L1]
    PROMOTE2 --> RETURN3[Return Result]

    L3 -->|Miss| EXTERNAL[Query External Sources]
    EXTERNAL --> STORE[Store in L3, L2, L1]
    STORE --> RETURN4[Return Result]

Memory Routing Strategy

class MemoryRouter:
    def route_query(self, query, context):
        # Classify query type
        if is_recent(query, window="5m"):
            return "L1_cache"  # Redis

        domain = extract_domain(query)
        if domain in ["code", "docs", "data"]:
            # Domain-specific local memory
            return f"L2_{domain}_vector_db"

        if is_entity_query(query):
            return "L3_knowledge_graph"  # PostgreSQL

        if requires_external_data(query):
            return "external_sources"

        # Default to global search
        return "L3_knowledge_graph"

Inter-Component Communication

Message Format

All inter-component messages follow this schema:

{
  "message_id": "uuid-v4",
  "timestamp": "2025-11-10T10:30:00Z",
  "from": "orchestrator",
  "to": "coder-arm",
  "message_type": "task_request",
  "payload": {
    "task_id": "task-12345",
    "action": "generate_function",
    "context": {},
    "constraints": [],
    "budget": {
      "max_tokens": 4000,
      "max_time_seconds": 30
    }
  },
  "trace_id": "trace-uuid",
  "parent_message_id": "parent-uuid"
}

Communication Patterns

1. Request-Response (Synchronous)

sequenceDiagram
    participant Orch as Orchestrator
    participant Arm as Arm

    Orch->>+Arm: POST /execute
    Note over Arm: Process Task<br/>(max 30s timeout)
    Arm-->>-Orch: 200 OK + Result

2. Fire-and-Forget (Asynchronous)

sequenceDiagram
    participant Orch as Orchestrator
    participant Queue as Task Queue
    participant Arm as Arm Worker

    Orch->>Queue: Enqueue Task
    Orch-->>Orch: Continue

    Note over Queue: Task persisted

    Arm->>Queue: Poll for Tasks
    Queue-->>Arm: Task
    Arm->>Arm: Process
    Arm->>Queue: Mark Complete

3. Publish-Subscribe (Events)

sequenceDiagram
    participant Arm as Arm (Publisher)
    participant Bus as Event Bus
    participant Sub1 as Subscriber 1
    participant Sub2 as Subscriber 2

    Arm->>Bus: Publish Event<br/>(e.g., "vulnerability_found")
    Bus->>Sub1: Notify
    Bus->>Sub2: Notify
    Sub1->>Sub1: Handle Event
    Sub2->>Sub2: Handle Event

Direct Arm-to-Arm Communication

Certain workflows benefit from direct communication:

graph LR
    PLAN[Planner Arm] -->|Execution Plan| EXEC[Executor Arm]
    CODE[Coder Arm] -->|Code Artifact| JUDGE[Judge Arm]
    JUDGE -->|Validation Result| CODE
    RETR[Retriever Arm] -->|Retrieved Context| CODE

When to use direct communication:

• High-frequency interactions (e.g., code validation loop)
• Large data transfers (avoid orchestrator bottleneck)
• Tight coupling between specific arms (e.g., coder + judge)

Constraints:

• Must register intent with orchestrator
• Include provenance in all messages
• Respect capability boundaries (no privilege escalation)

Provenance Tracking

Every data artifact includes complete lineage:

{
  "artifact_id": "art-uuid",
  "artifact_type": "code_function",
  "content": "def hello(): ...",
  "provenance": {
    "created_by": "coder-arm",
    "created_at": "2025-11-10T10:30:00Z",
    "task_id": "task-12345",
    "parent_task_id": "task-12300",
    "input_sources": [
      {
        "source_id": "doc-456",
        "source_type": "documentation",
        "relevance_score": 0.92
      }
    ],
    "transformations": [
      {
        "step": 1,
        "operation": "template_fill",
        "tool": "code_generator_v1"
      },
      {
        "step": 2,
        "operation": "syntax_validation",
        "tool": "ast_parser"
      }
    ],
    "validation_status": {
      "validated": true,
      "validator": "judge-arm",
      "confidence": 0.95,
      "checks_passed": ["syntax", "type_hints", "docstring"]
    },
    "model_info": {
      "model_name": "gpt-3.5-turbo",
      "prompt_hash": "sha256:abc123...",
      "temperature": 0.3,
      "tokens_used": 350
    }
  }
}

Provenance Flow

flowchart TD
    INPUT[Input Data] --> ARM[Arm Processes]
    ARM --> ATTACH[Attach Metadata]

    ATTACH --> PROV[Provenance Record]
    PROV --> CONTENT[Content Hash]
    PROV --> SOURCE[Source References]
    PROV --> TRANSFORM[Transformation Log]
    PROV --> VALID[Validation Results]

    CONTENT --> STORE[Storage]
    SOURCE --> STORE
    TRANSFORM --> STORE
    VALID --> STORE

    STORE --> QUERY[Queryable Provenance]

Error Handling Flow

Error Classification

flowchart TD
    ERROR[Error Occurred] --> CLASSIFY{Error Type}

    CLASSIFY -->|Transient| RETRY[Retry Logic]
    CLASSIFY -->|Invalid Input| USER_ERROR[Return 400]
    CLASSIFY -->|Auth/Authz| SECURITY[Return 403]
    CLASSIFY -->|Resource Limit| BACKPRESSURE[Apply Backpressure]
    CLASSIFY -->|Logic Error| ESCALATE[Escalate to Orchestrator]
    CLASSIFY -->|Critical| SHUTDOWN[Graceful Shutdown]

    RETRY --> BACKOFF{Retry Count}
    BACKOFF -->|< Max| WAIT[Exponential Backoff]
    WAIT --> RETRY_OP[Retry Operation]
    RETRY_OP --> SUCCESS{Success?}
    SUCCESS -->|Yes| RECOVER[Recovery Complete]
    SUCCESS -->|No| RETRY

    BACKOFF -->|>= Max| GIVE_UP[Return 503]

    USER_ERROR --> LOG1[Log Warning]
    SECURITY --> LOG2[Log Alert]
    BACKPRESSURE --> LOG3[Log Info]
    ESCALATE --> LOG4[Log Error]
    SHUTDOWN --> LOG5[Log Critical]

    LOG1 --> METRICS
    LOG2 --> METRICS
    LOG3 --> METRICS
    LOG4 --> METRICS
    LOG5 --> METRICS

    METRICS[Update Metrics]

Retry Strategy

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=1, max=10),
    retry=retry_if_exception_type(TransientError)
)
async def call_arm(arm_endpoint, payload):
    async with httpx.AsyncClient() as client:
        response = await client.post(
            arm_endpoint,
            json=payload,
            timeout=30.0
        )
        response.raise_for_status()
        return response.json()

Circuit Breaker Pattern

stateDiagram-v2
    [*] --> Closed

    Closed --> Open: Failure threshold exceeded
    Open --> HalfOpen: Timeout elapsed
    HalfOpen --> Closed: Success
    HalfOpen --> Open: Failure

    Closed : Allow all requests
    Open : Reject all requests<br/>Return cached/default
    HalfOpen : Allow limited requests<br/>Test recovery

Implementation:

from circuitbreaker import circuit

@circuit(failure_threshold=5, recovery_timeout=60)
async def call_external_api(url):
    # Will open circuit after 5 consecutive failures
    # Attempt recovery after 60 seconds
    async with httpx.AsyncClient() as client:
        return await client.get(url)

Error Propagation

sequenceDiagram
    participant Arm as Arm
    participant Orch as Orchestrator
    participant Monitor as Monitoring

    Arm->>Arm: Error Occurs
    Arm->>Arm: Classify Error

    alt Recoverable
        Arm->>Arm: Retry with Backoff
        Arm->>Monitor: Log Retry
    else Unrecoverable
        Arm->>Orch: Report Failure
        Orch->>Orch: Attempt Alternative
        alt Alternative Available
            Orch->>Arm: Try Different Arm
        else No Alternative
            Orch->>Monitor: Log Critical
            Orch-->>User: Return Error Response
        end
    end

    Monitor->>Monitor: Update Metrics
    Monitor->>Monitor: Check Thresholds
    alt Threshold Exceeded
        Monitor->>Monitor: Trigger Alert
    end

See Also

• System Architecture Overview
• Component Specifications
• Error Handling Guide
• Monitoring and Observability

Swarm Decision-Making Architecture

Version: 1.0 Last Updated: 2025-11-10 Status: Phase 2 - Core Capabilities Difficulty: Advanced

Table of Contents

1. Overview
2. Swarm Concept and Principles
3. Orchestration Flow
4. Use Cases
5. Implementation Patterns
6. Complete Python Implementation
7. Configuration and Tuning
8. Performance Considerations
9. Example Scenarios
10. Testing Swarm Behavior
11. Troubleshooting

Overview

Swarm decision-making is a critical Phase 2 capability that enables OctoLLM to leverage multiple specialized arms working in parallel to generate diverse solutions, which are then aggregated into a final, high-quality answer. This approach mirrors the biological octopus's ability to explore multiple strategies simultaneously.

Key Benefits

• Higher Accuracy: Multiple perspectives reduce single-point-of-failure risks
• Diverse Solutions: Different arms bring unique viewpoints and approaches
• Robustness: System continues even if individual arms fail
• Quality Assurance: Consensus mechanisms validate correctness
• Risk Mitigation: Critical decisions benefit from multiple expert opinions

When to Use Swarm

Swarm decision-making is expensive (multiple LLM calls, parallel processing) but valuable for:

• High-stakes decisions: Security vulnerability assessments, production deployments
• Complex problems: Multi-faceted issues requiring diverse expertise
• Quality-critical outputs: Code reviews, documentation generation
• Research tasks: Information synthesis from multiple sources
• Creative solutions: Brainstorming, design alternatives

When NOT to Use Swarm

• Simple queries: Single arm is faster and cheaper
• Low-priority tasks: Cost doesn't justify quality gain
• Time-sensitive operations: Latency overhead unacceptable
• Resource-constrained environments: Limited parallel capacity

Swarm Concept and Principles

Biological Inspiration

The octopus can explore multiple strategies in parallel:

• Each arm independently probes and evaluates options
• Arms communicate findings to the brain
• The brain synthesizes information and makes final decisions
• Disagreement between arms triggers deeper analysis

OctoLLM Swarm Model

graph TB
    O[Orchestrator] -->|Task| SA[Swarm Activator]
    SA -->|Identifies Swarm-Worthy Task| Sel[Arm Selector]

    Sel -->|Selects N Arms| A1[Arm 1]
    Sel -->|Selects N Arms| A2[Arm 2]
    Sel -->|Selects N Arms| A3[Arm 3]
    Sel -->|Selects N Arms| A4[Arm N]

    A1 -->|Proposal 1| Agg[Aggregator]
    A2 -->|Proposal 2| Agg
    A3 -->|Proposal 3| Agg
    A4 -->|Proposal N| Agg

    Agg -->|Applies Voting/Ranking| CR[Conflict Resolver]
    CR -->|Final Answer| Val[Validator]
    Val -->|Quality Check| O

    style SA fill:#e1f5ff
    style Agg fill:#ffe1e1
    style CR fill:#fff4e1
    style Val fill:#e1ffe1

Core Principles

1. Diversity: Select arms with different specializations or prompting strategies
2. Independence: Arms work without knowing others' proposals (avoid groupthink)
3. Aggregation: Combine proposals using voting, ranking, or learned methods
4. Conflict Resolution: Handle disagreements with explicit rules
5. Confidence Weighting: High-confidence proposals carry more weight
6. Quality Validation: Final answer must pass acceptance criteria

Orchestration Flow

High-Level Sequence

sequenceDiagram
    participant U as User
    participant O as Orchestrator
    participant S as SwarmOrchestrator
    participant A1 as Arm 1 (Coder)
    participant A2 as Arm 2 (Coder Alt)
    participant A3 as Arm 3 (Judge)
    participant Agg as ProposalAggregator
    participant CR as ConflictResolver

    U->>O: Submit Task (high priority)
    O->>O: Classify as swarm-worthy
    O->>S: Initialize Swarm

    S->>S: Select N=3 arms

    par Parallel Execution
        S->>A1: Execute(task, seed=1)
        S->>A2: Execute(task, seed=2)
        S->>A3: Execute(task, seed=3)
    end

    A1-->>S: Proposal 1 (confidence=0.85)
    A2-->>S: Proposal 2 (confidence=0.90)
    A3-->>S: Proposal 3 (confidence=0.75)

    S->>Agg: Aggregate([P1, P2, P3])
    Agg->>Agg: Apply Voting Strategy
    Agg->>CR: Check for conflicts

    alt No Conflict
        CR-->>Agg: Majority consensus
        Agg-->>S: Final Answer
    else Conflict Detected
        CR->>CR: Resolve using rules
        CR-->>S: Resolved Answer + Rationale
    end

    S-->>O: Swarm Result
    O-->>U: Response + Provenance

Step-by-Step Process

Step 1: Swarm Activation Decision

The orchestrator determines if a task warrants swarm processing based on:

def should_use_swarm(task: TaskContract) -> bool:
    """Determine if task benefits from swarm processing."""

    # High-priority tasks
    if task.priority in [Priority.HIGH, Priority.CRITICAL]:
        return True

    # Explicit swarm request
    if task.context.get("force_swarm", False):
        return True

    # Complex tasks (estimated multiple steps)
    if task.context.get("complexity_score", 0.0) > 0.7:
        return True

    # Security-critical operations
    if any(keyword in task.goal.lower() for keyword in [
        "security", "vulnerability", "exploit", "penetration", "audit"
    ]):
        return True

    # High-cost operations that justify swarm overhead
    if task.budget.get("max_cost_usd", 0.0) > 1.0:
        return True

    return False

Step 2: Arm Selection

Select N arms (typically 3-5) with diverse capabilities:

def select_swarm_arms(
    task: TaskContract,
    registry: Dict[str, ArmCapability],
    swarm_size: int = 3
) -> List[str]:
    """Select diverse arms for swarm execution."""

    # Score all arms for this task
    arm_scores = {}
    for arm_id, arm in registry.items():
        score = calculate_arm_relevance(arm, task)
        arm_scores[arm_id] = score

    # Sort by relevance
    sorted_arms = sorted(
        arm_scores.items(),
        key=lambda x: x[1],
        reverse=True
    )

    # Select top N arms, ensuring diversity
    selected = []
    for arm_id, score in sorted_arms:
        if len(selected) >= swarm_size:
            break

        # Ensure diversity (e.g., don't select multiple same-type arms)
        if is_diverse_from(arm_id, selected, registry):
            selected.append(arm_id)

    return selected

Step 3: Parallel Execution

Execute tasks in parallel using asyncio.gather():

async def execute_swarm(
    task: TaskContract,
    arms: List[str],
    registry: Dict[str, ArmCapability]
) -> List[Proposal]:
    """Execute task across multiple arms in parallel."""

    # Create execution tasks with different seeds for diversity
    tasks = []
    for i, arm_id in enumerate(arms):
        arm = registry[arm_id]

        # Vary prompts slightly for diversity
        task_variant = task.copy(deep=True)
        task_variant.context["seed"] = i
        task_variant.context["variant"] = f"approach_{i+1}"

        # Create async task
        coro = call_arm(arm, task_variant)
        tasks.append(coro)

    # Execute all in parallel
    results = await asyncio.gather(*tasks, return_exceptions=True)

    # Convert to Proposal objects
    proposals = []
    for i, result in enumerate(results):
        if isinstance(result, Exception):
            logger.warning(f"Arm {arms[i]} failed: {result}")
            continue

        proposals.append(Proposal(
            arm_id=arms[i],
            content=result.get("output"),
            confidence=result.get("confidence", 0.5),
            rationale=result.get("rationale", ""),
            execution_time_ms=result.get("duration_ms", 0)
        ))

    return proposals

Step 4: Proposal Aggregation

Combine proposals using one of several strategies:

A. Majority Voting (for discrete choices):

def majority_vote(proposals: List[Proposal]) -> Proposal:
    """Select most common proposal."""
    from collections import Counter

    # Count identical outputs
    output_counts = Counter([p.content for p in proposals])
    most_common = output_counts.most_common(1)[0][0]

    # Return first proposal with that output
    for p in proposals:
        if p.content == most_common:
            return p

    return proposals[0]  # Fallback

B. Confidence-Weighted Voting:

def weighted_vote(proposals: List[Proposal]) -> Proposal:
    """Weight proposals by confidence scores."""

    # Group by similar content
    groups = group_similar_proposals(proposals, similarity_threshold=0.8)

    # Calculate weighted score for each group
    group_scores = {}
    for group_id, group_proposals in groups.items():
        total_weight = sum(p.confidence for p in group_proposals)
        group_scores[group_id] = total_weight

    # Select highest-weighted group
    best_group = max(group_scores.items(), key=lambda x: x[1])[0]

    # Return highest-confidence proposal from best group
    best_proposals = sorted(
        groups[best_group],
        key=lambda p: p.confidence,
        reverse=True
    )

    return best_proposals[0]

C. Ranked Choice (Borda count):

def ranked_choice(proposals: List[Proposal]) -> Proposal:
    """Use Borda count to rank proposals."""

    # Each arm ranks all proposals (including its own)
    rankings = []
    for evaluator_arm in arms:
        # Ask evaluator to rank all proposals
        ranking = await ask_arm_to_rank(evaluator_arm, proposals)
        rankings.append(ranking)

    # Calculate Borda scores
    scores = {p.arm_id: 0 for p in proposals}
    num_proposals = len(proposals)

    for ranking in rankings:
        for position, arm_id in enumerate(ranking):
            # Higher position = higher score
            scores[arm_id] += (num_proposals - position - 1)

    # Select highest-scoring proposal
    best_arm_id = max(scores.items(), key=lambda x: x[1])[0]
    return next(p for p in proposals if p.arm_id == best_arm_id)

Step 5: Conflict Resolution

Handle disagreements between arms:

class ConflictResolver:
    """Resolves conflicts between swarm proposals."""

    def detect_conflict(self, proposals: List[Proposal]) -> Optional[Conflict]:
        """Check if proposals significantly disagree."""

        # Calculate pairwise similarity
        similarities = []
        for i, p1 in enumerate(proposals):
            for j, p2 in enumerate(proposals[i+1:], start=i+1):
                sim = calculate_similarity(p1.content, p2.content)
                similarities.append(sim)

        avg_similarity = np.mean(similarities)

        # Conflict if low average similarity
        if avg_similarity < 0.6:
            return Conflict(
                conflict_type="low_consensus",
                severity="high" if avg_similarity < 0.4 else "medium",
                proposals=proposals,
                similarity_score=avg_similarity
            )

        # Check for contradictions
        contradictions = self._find_contradictions(proposals)
        if contradictions:
            return Conflict(
                conflict_type="contradiction",
                severity="high",
                proposals=proposals,
                details=contradictions
            )

        return None

    def resolve_conflict(
        self,
        conflict: Conflict,
        task: TaskContract
    ) -> Resolution:
        """Apply resolution strategy based on conflict type."""

        if conflict.conflict_type == "low_consensus":
            # Use confidence weighting
            return self._resolve_by_confidence(conflict.proposals)

        elif conflict.conflict_type == "contradiction":
            # Escalate to Judge arm for arbitration
            return self._escalate_to_judge(conflict, task)

        else:
            # Default: select highest confidence
            return self._select_highest_confidence(conflict.proposals)

    def _escalate_to_judge(
        self,
        conflict: Conflict,
        task: TaskContract
    ) -> Resolution:
        """Have Judge arm arbitrate between conflicting proposals."""

        judge_task = TaskContract(
            task_id=f"{task.task_id}-judge",
            goal=f"Evaluate and select the best proposal for: {task.goal}",
            context={
                "original_task": task.dict(),
                "proposals": [p.dict() for p in conflict.proposals],
                "conflict_details": conflict.details
            },
            acceptance_criteria=[
                "Provides clear rationale for selection",
                "Identifies strengths and weaknesses of each proposal"
            ]
        )

        # Call Judge arm
        judge_result = await call_arm(judge_arm, judge_task)

        return Resolution(
            selected_proposal=judge_result["selected_proposal"],
            resolution_method="judge_arbitration",
            rationale=judge_result["rationale"],
            confidence=judge_result["confidence"]
        )

Use Cases

1. Security Vulnerability Assessment

Scenario: Analyze a codebase for security vulnerabilities

Swarm Configuration:

• Arm 1: Code Analyzer (static analysis focused)
• Arm 2: Security Specialist (OWASP Top 10 focused)
• Arm 3: Penetration Tester (exploit-focused)
• Arm 4: Code Reviewer (best practices focused)

Aggregation Strategy: Weighted voting + Judge arbitration for disagreements

Example:

task = TaskContract(
    task_id="sec-audit-001",
    goal="Identify security vulnerabilities in user authentication module",
    context={
        "code_path": "/src/auth/",
        "frameworks": ["Flask", "SQLAlchemy"],
        "threat_model": "OWASP Top 10"
    },
    priority=Priority.CRITICAL,
    constraints=[
        "Focus on authentication and authorization",
        "Provide exploit scenarios for each finding"
    ]
)

# Execute swarm
swarm_result = await swarm_orchestrator.execute(
    task=task,
    swarm_size=4,
    aggregation_strategy="weighted_vote_with_judge"
)

# Result includes:
# - Vulnerabilities found by majority (high confidence)
# - Unique findings from individual arms (flagged for review)
# - Confidence scores for each vulnerability
# - Recommended mitigations

Benefits:

• Catches vulnerabilities that single-arm might miss
• Diverse perspectives (static analysis + pentesting + code review)
• Higher confidence in findings through consensus

2. Code Review and Quality Assurance

Scenario: Review pull request for code quality

Swarm Configuration:

• Arm 1: Code Style Reviewer (PEP 8, linting)
• Arm 2: Performance Analyzer (algorithmic efficiency)
• Arm 3: Security Reviewer (injection, XSS, etc.)
• Arm 4: Test Coverage Analyzer

Aggregation Strategy: Merge all feedback, rank by severity

Example:

task = TaskContract(
    task_id="pr-review-456",
    goal="Review pull request #456 for quality and correctness",
    context={
        "pr_url": "https://github.com/org/repo/pull/456",
        "diff": pr_diff_content,
        "test_coverage_delta": -2.5  # Coverage decreased
    },
    priority=Priority.HIGH
)

# Swarm review
reviews = await swarm_orchestrator.execute(
    task=task,
    swarm_size=4,
    aggregation_strategy="merge_and_rank"
)

# Result:
# {
#   "critical_issues": [
#     {"type": "security", "severity": "high", "description": "SQL injection in line 42", ...},
#     {"type": "performance", "severity": "high", "description": "N+1 query pattern", ...}
#   ],
#   "warnings": [...],
#   "suggestions": [...],
#   "overall_verdict": "NEEDS_CHANGES",
#   "consensus_confidence": 0.92
# }

3. Research and Information Synthesis

Scenario: Research a complex technical topic

Swarm Configuration:

• Arm 1: Academic Paper Retriever (arXiv, Google Scholar)
• Arm 2: Documentation Searcher (official docs, Stack Overflow)
• Arm 3: Code Example Finder (GitHub, GitLab)
• Arm 4: Expert Q&A (Reddit, HackerNews, forums)

Aggregation Strategy: Merge information, de-duplicate, rank by source quality

Example:

task = TaskContract(
    task_id="research-ml-001",
    goal="Research state-of-the-art techniques for few-shot learning",
    context={
        "domain": "machine_learning",
        "sub_domain": "few_shot_learning",
        "recency": "last_2_years"
    },
    acceptance_criteria=[
        "At least 5 peer-reviewed papers",
        "2+ production implementations",
        "Comparison of different approaches"
    ]
)

# Swarm research
synthesis = await swarm_orchestrator.execute(
    task=task,
    swarm_size=4,
    aggregation_strategy="information_merge"
)

# Result:
# {
#   "summary": "Comprehensive overview of few-shot learning...",
#   "key_papers": [
#     {"title": "...", "authors": [...], "year": 2024, "citations": 142, ...}
#   ],
#   "implementations": [
#     {"name": "Pytorch Meta-Learning", "github": "...", "stars": 3200}
#   ],
#   "comparative_analysis": {...},
#   "sources_consulted": 47,
#   "confidence": 0.88
# }

4. Creative Problem Solving

Scenario: Generate multiple approaches to a design problem

Swarm Configuration:

• Arm 1: Traditional approach (established patterns)
• Arm 2: Innovative approach (novel techniques)
• Arm 3: Performance-optimized approach
• Arm 4: Simplicity-first approach (KISS principle)

Aggregation Strategy: Present all diverse solutions, rank by criteria

Example:

task = TaskContract(
    task_id="design-cache-001",
    goal="Design a distributed caching layer for microservices",
    context={
        "scale": "1000+ req/sec",
        "latency_requirement": "< 10ms P99",
        "consistency": "eventual"
    },
    constraints=[
        "Must integrate with Kubernetes",
        "Cost-effective at scale"
    ]
)

# Swarm brainstorm
designs = await swarm_orchestrator.execute(
    task=task,
    swarm_size=4,
    aggregation_strategy="diversity_ranking"
)

# Result: Multiple distinct designs
# {
#   "proposals": [
#     {
#       "approach": "Redis Cluster with Sentinel",
#       "pros": [...],
#       "cons": [...],
#       "estimated_cost": "$X/month",
#       "confidence": 0.9
#     },
#     {
#       "approach": "Hazelcast IMDG",
#       ...
#     },
#     ...
#   ],
#   "recommendation": "Redis Cluster",
#   "rationale": "Best balance of performance, cost, and operational maturity"
# }

Implementation Patterns

Pattern 1: Simple Swarm (Synchronous Voting)

Use When: Fast decisions, discrete choices (yes/no, A/B/C)

class SimpleSwarmOrchestrator:
    """Basic swarm with majority voting."""

    async def execute(
        self,
        task: TaskContract,
        swarm_size: int = 3
    ) -> SwarmResult:
        # Select arms
        arms = self.select_arms(task, swarm_size)

        # Execute in parallel
        proposals = await asyncio.gather(*[
            self.call_arm(arm, task) for arm in arms
        ])

        # Majority vote
        result = self.majority_vote(proposals)

        return SwarmResult(
            final_answer=result,
            all_proposals=proposals,
            aggregation_method="majority_vote"
        )

Pattern 2: Weighted Swarm (Confidence-Based)

Use When: Proposals have varying quality, arms have different expertise

class WeightedSwarmOrchestrator:
    """Swarm with confidence-weighted voting."""

    async def execute(
        self,
        task: TaskContract,
        swarm_size: int = 3
    ) -> SwarmResult:
        arms = self.select_arms(task, swarm_size)

        # Get proposals with confidence scores
        proposals = await asyncio.gather(*[
            self.call_arm_with_confidence(arm, task)
            for arm in arms
        ])

        # Weight by confidence
        weights = [p.confidence for p in proposals]
        result = self.weighted_average(proposals, weights)

        return SwarmResult(
            final_answer=result,
            all_proposals=proposals,
            weights=weights,
            aggregation_method="confidence_weighted"
        )

Pattern 3: Judge-Mediated Swarm

Use When: Complex outputs, need expert arbitration

class JudgeMediatedSwarmOrchestrator:
    """Swarm with Judge arm for final decision."""

    async def execute(
        self,
        task: TaskContract,
        swarm_size: int = 3
    ) -> SwarmResult:
        # Get diverse proposals
        arms = self.select_arms(task, swarm_size)
        proposals = await asyncio.gather(*[
            self.call_arm(arm, task) for arm in arms
        ])

        # Have Judge evaluate all proposals
        judge_task = self.create_judge_task(task, proposals)
        judge_result = await self.call_arm(
            self.judge_arm,
            judge_task
        )

        return SwarmResult(
            final_answer=judge_result["selected_proposal"],
            all_proposals=proposals,
            judge_rationale=judge_result["rationale"],
            aggregation_method="judge_mediated"
        )

Pattern 4: Iterative Refinement Swarm

Use When: Need multiple rounds of improvement

class IterativeRefinementSwarm:
    """Swarm that refines answer over multiple rounds."""

    async def execute(
        self,
        task: TaskContract,
        swarm_size: int = 3,
        max_iterations: int = 3
    ) -> SwarmResult:
        current_answer = None

        for iteration in range(max_iterations):
            # Generate proposals (or refinements)
            if current_answer:
                task.context["previous_answer"] = current_answer
                task.goal = f"Improve upon: {current_answer}"

            arms = self.select_arms(task, swarm_size)
            proposals = await asyncio.gather(*[
                self.call_arm(arm, task) for arm in arms
            ])

            # Aggregate
            current_answer = self.aggregate(proposals)

            # Check if converged
            if self.has_converged(proposals):
                break

        return SwarmResult(
            final_answer=current_answer,
            iterations=iteration + 1,
            aggregation_method="iterative_refinement"
        )

Complete Python Implementation

Core Data Models

from pydantic import BaseModel, Field
from typing import List, Optional, Dict, Any
from enum import Enum
import hashlib

class ProposalStatus(str, Enum):
    """Status of a proposal in the swarm."""
    PENDING = "pending"
    COMPLETED = "completed"
    FAILED = "failed"
    REJECTED = "rejected"

class Proposal(BaseModel):
    """A single proposal from an arm."""

    arm_id: str = Field(..., description="Which arm generated this")
    content: Any = Field(..., description="The proposed solution")
    confidence: float = Field(..., ge=0.0, le=1.0, description="Arm's confidence")
    rationale: str = Field("", description="Why this proposal")
    execution_time_ms: int = Field(..., ge=0)
    status: ProposalStatus = Field(default=ProposalStatus.COMPLETED)
    metadata: Dict[str, Any] = Field(default_factory=dict)

    def content_hash(self) -> str:
        """Generate hash of content for deduplication."""
        content_str = str(self.content)
        return hashlib.sha256(content_str.encode()).hexdigest()[:16]

class SwarmConfig(BaseModel):
    """Configuration for swarm execution."""

    swarm_size: int = Field(3, ge=2, le=10, description="Number of arms")
    aggregation_strategy: str = Field(
        "weighted_vote",
        description="How to combine proposals"
    )
    timeout_seconds: int = Field(60, ge=10, le=600)
    require_consensus: bool = Field(False, description="All arms must agree")
    consensus_threshold: float = Field(0.7, ge=0.5, le=1.0)
    enable_judge: bool = Field(True, description="Use Judge for conflicts")
    diversity_requirement: float = Field(0.5, ge=0.0, le=1.0)

class SwarmResult(BaseModel):
    """Result from swarm execution."""

    final_answer: Any = Field(..., description="Aggregated result")
    all_proposals: List[Proposal] = Field(..., description="All proposals")
    aggregation_method: str
    consensus_score: float = Field(..., ge=0.0, le=1.0)
    execution_time_ms: int
    metadata: Dict[str, Any] = Field(default_factory=dict)

Swarm Orchestrator

import asyncio
from typing import List, Dict, Optional, Callable
import numpy as np
from datetime import datetime
import structlog

logger = structlog.get_logger()

class SwarmOrchestrator:
    """
    Coordinates swarm decision-making across multiple arms.

    Features:
    - Parallel arm execution
    - Multiple aggregation strategies
    - Conflict detection and resolution
    - Performance tracking
    """

    def __init__(
        self,
        arm_registry: Dict[str, ArmCapability],
        judge_arm_id: str = "judge",
        default_config: Optional[SwarmConfig] = None
    ):
        self.registry = arm_registry
        self.judge_arm_id = judge_arm_id
        self.default_config = default_config or SwarmConfig()
        self.aggregator = ProposalAggregator()
        self.conflict_resolver = ConflictResolver()

    async def execute(
        self,
        task: TaskContract,
        config: Optional[SwarmConfig] = None
    ) -> SwarmResult:
        """
        Execute task across swarm of arms and aggregate results.

        Args:
            task: Task to execute
            config: Swarm configuration (uses default if None)

        Returns:
            SwarmResult with final answer and metadata
        """
        config = config or self.default_config
        start_time = datetime.utcnow()

        logger.info(
            "swarm.execute.start",
            task_id=task.task_id,
            swarm_size=config.swarm_size,
            strategy=config.aggregation_strategy
        )

        # Step 1: Select diverse arms
        selected_arms = self._select_diverse_arms(task, config.swarm_size)
        logger.info("swarm.arms_selected", arms=selected_arms)

        # Step 2: Execute in parallel
        proposals = await self._execute_parallel(
            task, selected_arms, config.timeout_seconds
        )
        logger.info(
            "swarm.proposals_received",
            count=len(proposals),
            successful=sum(1 for p in proposals if p.status == ProposalStatus.COMPLETED)
        )

        # Step 3: Filter failed proposals
        valid_proposals = [
            p for p in proposals if p.status == ProposalStatus.COMPLETED
        ]

        if len(valid_proposals) < 2:
            raise InsufficientProposalsError(
                f"Only {len(valid_proposals)} valid proposals (minimum 2 required)"
            )

        # Step 4: Aggregate proposals
        aggregation_result = await self._aggregate_proposals(
            valid_proposals,
            config.aggregation_strategy,
            task
        )

        # Step 5: Check for conflicts
        conflict = self.conflict_resolver.detect_conflict(valid_proposals)
        if conflict and config.enable_judge:
            logger.warning("swarm.conflict_detected", conflict_type=conflict.conflict_type)
            resolution = await self.conflict_resolver.resolve_conflict(
                conflict, task, self.registry[self.judge_arm_id]
            )
            final_answer = resolution.selected_proposal
            aggregation_method = f"{config.aggregation_strategy}_with_judge"
        else:
            final_answer = aggregation_result["answer"]
            aggregation_method = config.aggregation_strategy

        # Step 6: Calculate consensus score
        consensus_score = self._calculate_consensus(valid_proposals)

        # Step 7: Validate against acceptance criteria
        if config.require_consensus and consensus_score < config.consensus_threshold:
            logger.warning(
                "swarm.low_consensus",
                score=consensus_score,
                threshold=config.consensus_threshold
            )

        execution_time = (datetime.utcnow() - start_time).total_seconds() * 1000

        result = SwarmResult(
            final_answer=final_answer,
            all_proposals=valid_proposals,
            aggregation_method=aggregation_method,
            consensus_score=consensus_score,
            execution_time_ms=int(execution_time),
            metadata={
                "selected_arms": selected_arms,
                "conflict_detected": conflict is not None,
                "proposal_count": len(valid_proposals)
            }
        )

        logger.info(
            "swarm.execute.complete",
            task_id=task.task_id,
            consensus_score=consensus_score,
            execution_time_ms=result.execution_time_ms
        )

        return result

    def _select_diverse_arms(
        self,
        task: TaskContract,
        swarm_size: int
    ) -> List[str]:
        """Select diverse arms for swarm execution."""

        # Score all arms for relevance
        arm_scores = {}
        for arm_id, arm in self.registry.items():
            if arm_id == self.judge_arm_id:
                continue  # Don't include judge in swarm

            relevance_score = self._calculate_arm_relevance(arm, task)
            arm_scores[arm_id] = relevance_score

        # Sort by relevance
        sorted_arms = sorted(
            arm_scores.items(),
            key=lambda x: x[1],
            reverse=True
        )

        # Select top N, ensuring diversity
        selected = []
        for arm_id, score in sorted_arms:
            if len(selected) >= swarm_size:
                break

            # Check diversity
            if not selected or self._is_diverse_from(arm_id, selected):
                selected.append(arm_id)

        # If not enough diverse arms, fill with top-scoring
        while len(selected) < swarm_size and len(selected) < len(sorted_arms):
            for arm_id, _ in sorted_arms:
                if arm_id not in selected:
                    selected.append(arm_id)
                    break

        return selected

    def _calculate_arm_relevance(
        self,
        arm: ArmCapability,
        task: TaskContract
    ) -> float:
        """Calculate how relevant an arm is for this task."""

        # Extract keywords from task goal
        goal_keywords = set(task.goal.lower().split())

        # Match against arm capabilities
        capability_keywords = set()
        for cap in arm.capabilities:
            capability_keywords.update(cap.lower().split())

        # Calculate overlap
        overlap = len(goal_keywords & capability_keywords)
        total = len(goal_keywords | capability_keywords)

        keyword_score = overlap / total if total > 0 else 0.0

        # Factor in historical success rate
        success_score = arm.success_rate

        # Combine scores
        relevance = 0.6 * keyword_score + 0.4 * success_score

        return relevance

    def _is_diverse_from(
        self,
        arm_id: str,
        selected_arms: List[str]
    ) -> bool:
        """Check if arm brings diversity to selection."""

        arm = self.registry[arm_id]

        for selected_id in selected_arms:
            selected_arm = self.registry[selected_id]

            # Check capability overlap
            overlap = len(
                set(arm.capabilities) & set(selected_arm.capabilities)
            )
            total = len(
                set(arm.capabilities) | set(selected_arm.capabilities)
            )

            similarity = overlap / total if total > 0 else 0.0

            # If too similar, not diverse
            if similarity > 0.7:
                return False

        return True

    async def _execute_parallel(
        self,
        task: TaskContract,
        arms: List[str],
        timeout_seconds: int
    ) -> List[Proposal]:
        """Execute task across multiple arms in parallel."""

        # Create tasks with variation for diversity
        async_tasks = []
        for i, arm_id in enumerate(arms):
            # Vary the task slightly for each arm
            task_variant = task.copy(deep=True)
            task_variant.context["swarm_variant"] = i
            task_variant.context["swarm_seed"] = i + 1

            # Create execution coroutine
            coro = self._execute_single_arm(
                arm_id, task_variant, timeout_seconds
            )
            async_tasks.append(coro)

        # Execute all in parallel with timeout
        results = await asyncio.gather(*async_tasks, return_exceptions=True)

        # Convert results to Proposal objects
        proposals = []
        for i, result in enumerate(results):
            arm_id = arms[i]

            if isinstance(result, Exception):
                logger.error(
                    "swarm.arm_failed",
                    arm_id=arm_id,
                    error=str(result)
                )
                proposals.append(Proposal(
                    arm_id=arm_id,
                    content=None,
                    confidence=0.0,
                    rationale=f"Execution failed: {str(result)}",
                    execution_time_ms=0,
                    status=ProposalStatus.FAILED
                ))
            else:
                proposals.append(result)

        return proposals

    async def _execute_single_arm(
        self,
        arm_id: str,
        task: TaskContract,
        timeout_seconds: int
    ) -> Proposal:
        """Execute task on a single arm with timeout."""

        arm = self.registry[arm_id]
        start_time = datetime.utcnow()

        try:
            # Call arm with timeout
            result = await asyncio.wait_for(
                self._call_arm(arm, task),
                timeout=timeout_seconds
            )

            execution_time = (datetime.utcnow() - start_time).total_seconds() * 1000

            return Proposal(
                arm_id=arm_id,
                content=result.get("output"),
                confidence=result.get("confidence", 0.5),
                rationale=result.get("rationale", ""),
                execution_time_ms=int(execution_time),
                status=ProposalStatus.COMPLETED,
                metadata=result.get("metadata", {})
            )

        except asyncio.TimeoutError:
            logger.warning("swarm.arm_timeout", arm_id=arm_id, timeout=timeout_seconds)
            return Proposal(
                arm_id=arm_id,
                content=None,
                confidence=0.0,
                rationale=f"Timeout after {timeout_seconds}s",
                execution_time_ms=timeout_seconds * 1000,
                status=ProposalStatus.FAILED
            )

        except Exception as e:
            logger.error("swarm.arm_error", arm_id=arm_id, error=str(e))
            raise

    async def _call_arm(
        self,
        arm: ArmCapability,
        task: TaskContract
    ) -> Dict[str, Any]:
        """Make HTTP call to arm endpoint."""

        import aiohttp

        async with aiohttp.ClientSession() as session:
            async with session.post(
                arm.endpoint,
                json=task.dict(),
                timeout=aiohttp.ClientTimeout(total=60)
            ) as response:
                response.raise_for_status()
                return await response.json()

    async def _aggregate_proposals(
        self,
        proposals: List[Proposal],
        strategy: str,
        task: TaskContract
    ) -> Dict[str, Any]:
        """Aggregate proposals using specified strategy."""

        if strategy == "majority_vote":
            return self.aggregator.majority_vote(proposals)
        elif strategy == "weighted_vote":
            return self.aggregator.weighted_vote(proposals)
        elif strategy == "ranked_choice":
            return await self.aggregator.ranked_choice(proposals)
        elif strategy == "confidence_max":
            return self.aggregator.select_highest_confidence(proposals)
        else:
            raise ValueError(f"Unknown aggregation strategy: {strategy}")

    def _calculate_consensus(self, proposals: List[Proposal]) -> float:
        """Calculate consensus score (0.0-1.0) among proposals."""

        if len(proposals) < 2:
            return 1.0

        # Calculate pairwise similarities
        similarities = []
        for i, p1 in enumerate(proposals):
            for p2 in proposals[i+1:]:
                sim = self._calculate_similarity(p1.content, p2.content)
                similarities.append(sim)

        # Average similarity is consensus score
        return np.mean(similarities) if similarities else 0.0

    def _calculate_similarity(self, content1: Any, content2: Any) -> float:
        """Calculate similarity between two proposal contents."""

        # Simple string-based similarity for now
        # TODO: Use embedding-based similarity for better results

        str1 = str(content1).lower()
        str2 = str(content2).lower()

        # Jaccard similarity on words
        words1 = set(str1.split())
        words2 = set(str2.split())

        intersection = len(words1 & words2)
        union = len(words1 | words2)

        return intersection / union if union > 0 else 0.0

Proposal Aggregator

class ProposalAggregator:
    """Aggregates proposals using various strategies."""

    def majority_vote(self, proposals: List[Proposal]) -> Dict[str, Any]:
        """Select most common proposal (for discrete choices)."""
        from collections import Counter

        # Hash proposals to group identical ones
        proposal_hashes = [p.content_hash() for p in proposals]
        hash_counts = Counter(proposal_hashes)

        # Find most common
        most_common_hash = hash_counts.most_common(1)[0][0]

        # Return first proposal with that hash
        for p in proposals:
            if p.content_hash() == most_common_hash:
                return {
                    "answer": p.content,
                    "method": "majority_vote",
                    "vote_count": hash_counts[most_common_hash],
                    "total_votes": len(proposals)
                }

        # Fallback
        return {"answer": proposals[0].content, "method": "majority_vote"}

    def weighted_vote(self, proposals: List[Proposal]) -> Dict[str, Any]:
        """Weight proposals by confidence scores."""

        # Group similar proposals
        groups = self._group_similar_proposals(proposals, threshold=0.8)

        # Calculate weighted score for each group
        group_scores = {}
        for group_id, group_proposals in groups.items():
            # Sum of confidences
            total_weight = sum(p.confidence for p in group_proposals)
            group_scores[group_id] = total_weight

        # Select highest-weighted group
        best_group_id = max(group_scores.items(), key=lambda x: x[1])[0]
        best_group = groups[best_group_id]

        # Within best group, select highest-confidence proposal
        best_proposal = max(best_group, key=lambda p: p.confidence)

        return {
            "answer": best_proposal.content,
            "method": "weighted_vote",
            "total_weight": group_scores[best_group_id],
            "group_size": len(best_group)
        }

    async def ranked_choice(self, proposals: List[Proposal]) -> Dict[str, Any]:
        """Use Borda count ranking."""

        # For simplicity, rank by confidence (in production, could ask arms to rank each other)
        sorted_proposals = sorted(
            proposals,
            key=lambda p: p.confidence,
            reverse=True
        )

        # Borda count: first place gets N-1 points, second gets N-2, etc.
        n = len(proposals)
        scores = {p.arm_id: 0 for p in proposals}

        for position, proposal in enumerate(sorted_proposals):
            scores[proposal.arm_id] = n - position - 1

        # Select highest-scoring
        best_arm_id = max(scores.items(), key=lambda x: x[1])[0]
        best_proposal = next(p for p in proposals if p.arm_id == best_arm_id)

        return {
            "answer": best_proposal.content,
            "method": "ranked_choice",
            "borda_score": scores[best_arm_id],
            "ranking": [p.arm_id for p in sorted_proposals]
        }

    def select_highest_confidence(
        self,
        proposals: List[Proposal]
    ) -> Dict[str, Any]:
        """Simply select proposal with highest confidence."""

        best = max(proposals, key=lambda p: p.confidence)

        return {
            "answer": best.content,
            "method": "confidence_max",
            "confidence": best.confidence,
            "arm_id": best.arm_id
        }

    def _group_similar_proposals(
        self,
        proposals: List[Proposal],
        threshold: float = 0.8
    ) -> Dict[int, List[Proposal]]:
        """Group proposals by similarity."""

        groups = {}
        next_group_id = 0

        for proposal in proposals:
            # Check if similar to any existing group
            assigned = False
            for group_id, group_proposals in groups.items():
                # Compare to first proposal in group
                representative = group_proposals[0]
                similarity = self._calculate_similarity(
                    proposal.content,
                    representative.content
                )

                if similarity >= threshold:
                    groups[group_id].append(proposal)
                    assigned = True
                    break

            # Create new group if not assigned
            if not assigned:
                groups[next_group_id] = [proposal]
                next_group_id += 1

        return groups

    def _calculate_similarity(self, content1: Any, content2: Any) -> float:
        """Calculate similarity (same as in SwarmOrchestrator)."""
        str1 = str(content1).lower()
        str2 = str(content2).lower()
        words1 = set(str1.split())
        words2 = set(str2.split())
        intersection = len(words1 & words2)
        union = len(words1 | words2)
        return intersection / union if union > 0 else 0.0

Conflict Resolver

class Conflict(BaseModel):
    """Represents a conflict between proposals."""
    conflict_type: str  # "low_consensus", "contradiction", "high_variance"
    severity: str  # "low", "medium", "high"
    proposals: List[Proposal]
    similarity_score: Optional[float] = None
    details: Optional[Dict[str, Any]] = None

class Resolution(BaseModel):
    """Resolution of a conflict."""
    selected_proposal: Any
    resolution_method: str
    rationale: str
    confidence: float

class ConflictResolver:
    """Detects and resolves conflicts between swarm proposals."""

    def detect_conflict(
        self,
        proposals: List[Proposal],
        similarity_threshold: float = 0.6
    ) -> Optional[Conflict]:
        """Detect if proposals are in conflict."""

        if len(proposals) < 2:
            return None

        # Calculate all pairwise similarities
        similarities = []
        for i, p1 in enumerate(proposals):
            for p2 in proposals[i+1:]:
                sim = self._calculate_similarity(p1.content, p2.content)
                similarities.append(sim)

        avg_similarity = np.mean(similarities)

        # Low consensus = conflict
        if avg_similarity < similarity_threshold:
            severity = "high" if avg_similarity < 0.4 else "medium"
            return Conflict(
                conflict_type="low_consensus",
                severity=severity,
                proposals=proposals,
                similarity_score=avg_similarity
            )

        # Check for logical contradictions
        contradictions = self._find_contradictions(proposals)
        if contradictions:
            return Conflict(
                conflict_type="contradiction",
                severity="high",
                proposals=proposals,
                details={"contradictions": contradictions}
            )

        return None

    async def resolve_conflict(
        self,
        conflict: Conflict,
        task: TaskContract,
        judge_arm: ArmCapability
    ) -> Resolution:
        """Resolve conflict using appropriate strategy."""

        if conflict.conflict_type == "low_consensus":
            # Use confidence weighting
            return self._resolve_by_confidence(conflict.proposals)

        elif conflict.conflict_type == "contradiction":
            # Escalate to Judge
            return await self._escalate_to_judge(conflict, task, judge_arm)

        else:
            # Default: highest confidence
            return self._resolve_by_confidence(conflict.proposals)

    def _resolve_by_confidence(
        self,
        proposals: List[Proposal]
    ) -> Resolution:
        """Select highest-confidence proposal."""

        best = max(proposals, key=lambda p: p.confidence)

        return Resolution(
            selected_proposal=best.content,
            resolution_method="confidence_selection",
            rationale=f"Selected highest confidence ({best.confidence:.2f}) from {best.arm_id}",
            confidence=best.confidence
        )

    async def _escalate_to_judge(
        self,
        conflict: Conflict,
        task: TaskContract,
        judge_arm: ArmCapability
    ) -> Resolution:
        """Have Judge arm arbitrate."""

        judge_task = TaskContract(
            task_id=f"{task.task_id}-judge-arbitration",
            goal=f"Evaluate and select best proposal for: {task.goal}",
            context={
                "original_task": task.dict(),
                "proposals": [
                    {
                        "arm_id": p.arm_id,
                        "content": p.content,
                        "confidence": p.confidence,
                        "rationale": p.rationale
                    }
                    for p in conflict.proposals
                ],
                "conflict_details": conflict.dict()
            },
            acceptance_criteria=[
                "Provides clear selection rationale",
                "Identifies strengths/weaknesses of each proposal",
                "Explains why selected proposal is best"
            ]
        )

        # Call Judge arm
        import aiohttp
        async with aiohttp.ClientSession() as session:
            async with session.post(
                judge_arm.endpoint,
                json=judge_task.dict(),
                timeout=aiohttp.ClientTimeout(total=60)
            ) as response:
                response.raise_for_status()
                result = await response.json()

        return Resolution(
            selected_proposal=result["selected_proposal"],
            resolution_method="judge_arbitration",
            rationale=result["rationale"],
            confidence=result.get("confidence", 0.7)
        )

    def _calculate_similarity(self, content1: Any, content2: Any) -> float:
        """Calculate similarity (reuse from aggregator)."""
        str1 = str(content1).lower()
        str2 = str(content2).lower()
        words1 = set(str1.split())
        words2 = set(str2.split())
        intersection = len(words1 & words2)
        union = len(words1 | words2)
        return intersection / union if union > 0 else 0.0

    def _find_contradictions(
        self,
        proposals: List[Proposal]
    ) -> Optional[List[Dict[str, Any]]]:
        """Find logical contradictions between proposals."""

        # Simple contradiction detection (could be enhanced with NLP)
        contradiction_keywords = [
            ("yes", "no"),
            ("true", "false"),
            ("safe", "unsafe"),
            ("valid", "invalid"),
            ("secure", "insecure")
        ]

        contradictions = []
        for i, p1 in enumerate(proposals):
            for p2 in proposals[i+1:]:
                content1 = str(p1.content).lower()
                content2 = str(p2.content).lower()

                for kw1, kw2 in contradiction_keywords:
                    if kw1 in content1 and kw2 in content2:
                        contradictions.append({
                            "proposal_1": p1.arm_id,
                            "proposal_2": p2.arm_id,
                            "keyword_1": kw1,
                            "keyword_2": kw2
                        })

        return contradictions if contradictions else None

Configuration and Tuning

Swarm Size Selection

def determine_optimal_swarm_size(task: TaskContract) -> int:
    """Determine optimal number of arms for this task."""

    # Default: 3 arms
    swarm_size = 3

    # High-priority tasks: 5 arms
    if task.priority in [Priority.HIGH, Priority.CRITICAL]:
        swarm_size = 5

    # Complex tasks: 4-5 arms
    complexity = task.context.get("complexity_score", 0.5)
    if complexity > 0.7:
        swarm_size = max(swarm_size, 4)

    # Budget-constrained: 2 arms
    if task.budget.get("max_cost_usd", float('inf')) < 0.5:
        swarm_size = 2

    # Time-sensitive: 3 arms (parallel overhead)
    if task.budget.get("max_time_seconds", float('inf')) < 30:
        swarm_size = min(swarm_size, 3)

    return swarm_size

Aggregation Strategy Selection

def select_aggregation_strategy(
    task: TaskContract,
    proposals: List[Proposal]
) -> str:
    """Select best aggregation strategy for this task."""

    # Discrete choices: majority vote
    if task.context.get("output_type") == "discrete":
        return "majority_vote"

    # High variance in confidence: weighted vote
    confidences = [p.confidence for p in proposals]
    if max(confidences) - min(confidences) > 0.3:
        return "weighted_vote"

    # Complex evaluation needed: ranked choice with judge
    if task.priority == Priority.CRITICAL:
        return "ranked_choice"

    # Default: weighted vote
    return "weighted_vote"

Performance vs. Quality Tradeoffs

class SwarmTuningConfig(BaseModel):
    """Tuning parameters for swarm performance."""

    # Quality settings
    min_swarm_size: int = Field(2, description="Minimum arms for swarm")
    max_swarm_size: int = Field(10, description="Maximum arms for swarm")
    consensus_threshold: float = Field(0.7, description="Minimum consensus required")

    # Performance settings
    parallel_timeout_seconds: int = Field(60, description="Max wait for all arms")
    enable_early_termination: bool = Field(
        True,
        description="Stop if consensus reached early"
    )
    early_termination_threshold: float = Field(
        0.9,
        description="Consensus needed for early stop"
    )

    # Cost settings
    max_cost_per_task_usd: float = Field(5.0, description="Maximum spend per task")
    prefer_cheap_arms: bool = Field(
        False,
        description="Bias toward lower-cost arms"
    )

Performance Considerations

Latency Analysis

Single Arm: 1-5 seconds (typical) Swarm (3 arms): 1-5 seconds (parallel execution, minimal overhead) Swarm (5 arms): 1-5 seconds (still parallel) Swarm with Judge: +2-4 seconds (judge evaluation) Swarm with Conflict Resolution: +3-6 seconds (additional round)

Cost Analysis

Optimization Strategies

1. Early Termination

async def execute_with_early_termination(
    self,
    task: TaskContract,
    config: SwarmConfig
) -> SwarmResult:
    """Stop swarm execution early if consensus reached."""

    proposals = []
    for arm_id in selected_arms:
        # Execute one arm at a time
        proposal = await self._execute_single_arm(arm_id, task, timeout)
        proposals.append(proposal)

        # Check consensus after each new proposal
        if len(proposals) >= 2:
            consensus = self._calculate_consensus(proposals)
            if consensus >= config.early_termination_threshold:
                logger.info(
                    "swarm.early_termination",
                    consensus=consensus,
                    proposals_used=len(proposals)
                )
                break

    # Continue with aggregation...

2. Cached Swarm Results

async def execute_with_cache(
    self,
    task: TaskContract,
    config: SwarmConfig
) -> SwarmResult:
    """Cache swarm results for similar tasks."""

    # Generate cache key from task
    cache_key = self._generate_cache_key(task)

    # Check cache
    cached = await self.cache.get(cache_key)
    if cached:
        logger.info("swarm.cache_hit", task_id=task.task_id)
        return cached

    # Execute swarm
    result = await self.execute(task, config)

    # Store in cache (1 hour TTL)
    await self.cache.set(cache_key, result, ttl=3600)

    return result

3. Adaptive Swarm Size

def adaptive_swarm_size(
    task: TaskContract,
    budget: Dict[str, float]
) -> int:
    """Dynamically adjust swarm size based on budget."""

    available_budget_usd = budget.get("remaining_usd", 1.0)
    estimated_cost_per_arm = 0.02  # $0.02 per LLM call

    max_affordable_arms = int(available_budget_usd / estimated_cost_per_arm)

    # Clamp to reasonable range
    return max(2, min(10, max_affordable_arms))

Example Scenarios

Scenario 1: Security Vulnerability Assessment

# Task: Analyze authentication module for vulnerabilities
task = TaskContract(
    task_id="sec-001",
    goal="Identify security vulnerabilities in Flask authentication module",
    context={
        "code_path": "/app/auth.py",
        "frameworks": ["Flask", "SQLAlchemy"],
        "threat_model": "OWASP Top 10 2024"
    },
    priority=Priority.CRITICAL,
    acceptance_criteria=[
        "Identifies all SQL injection vectors",
        "Checks for XSS vulnerabilities",
        "Validates session management",
        "Provides exploit scenarios"
    ]
)

# Swarm configuration
config = SwarmConfig(
    swarm_size=4,
    aggregation_strategy="weighted_vote",
    enable_judge=True,
    require_consensus=True,
    consensus_threshold=0.75
)

# Execute swarm
swarm = SwarmOrchestrator(arm_registry, judge_arm_id="judge")
result = await swarm.execute(task, config)

# Result:
# {
#   "final_answer": {
#     "vulnerabilities": [
#       {
#         "type": "SQL Injection",
#         "severity": "CRITICAL",
#         "location": "auth.py:142",
#         "description": "Unsanitized user input in SQL query",
#         "exploit_scenario": "Attacker can bypass authentication with payload: ' OR '1'='1",
#         "confidence": 0.95,
#         "supporting_arms": ["coder", "security_specialist", "pentester"]
#       },
#       {
#         "type": "Session Fixation",
#         "severity": "HIGH",
#         "location": "auth.py:78",
#         "confidence": 0.87,
#         "supporting_arms": ["security_specialist", "coder"]
#       }
#     ],
#     "total_issues": 7,
#     "critical": 1,
#     "high": 2,
#     "medium": 4
#   },
#   "consensus_score": 0.82,
#   "aggregation_method": "weighted_vote_with_judge",
#   "all_proposals": [...],  # 4 proposals from arms
#   "execution_time_ms": 4250
# }

Scenario 2: Code Review with Swarm

# Task: Review pull request
task = TaskContract(
    task_id="pr-review-123",
    goal="Review pull request #123 for code quality and correctness",
    context={
        "pr_url": "https://github.com/org/repo/pull/123",
        "diff": pr_diff,
        "files_changed": 8,
        "lines_added": 342,
        "lines_deleted": 87
    },
    priority=Priority.HIGH,
    acceptance_criteria=[
        "Identifies code style violations",
        "Checks for performance regressions",
        "Validates test coverage",
        "Assesses security implications"
    ]
)

config = SwarmConfig(
    swarm_size=4,
    aggregation_strategy="merge_and_rank",
    enable_judge=False  # Don't need judge for code review
)

result = await swarm.execute(task, config)

# Result: Merged feedback from all reviewers
# {
#   "final_answer": {
#     "approval_status": "NEEDS_CHANGES",
#     "blocking_issues": [
#       {"type": "security", "severity": "high", "line": 42, "message": "..."},
#       {"type": "performance", "severity": "high", "line": 156, "message": "..."}
#     ],
#     "warnings": [...],
#     "suggestions": [...],
#     "test_coverage_delta": -2.5,
#     "estimated_review_time_hours": 2
#   },
#   "consensus_score": 0.91,
#   "execution_time_ms": 3800
# }

Scenario 3: Research Task

# Task: Research few-shot learning techniques
task = TaskContract(
    task_id="research-001",
    goal="Research and summarize state-of-the-art few-shot learning techniques (2023-2024)",
    context={
        "domain": "machine_learning",
        "recency": "last_2_years",
        "depth": "comprehensive"
    },
    priority=Priority.MEDIUM,
    acceptance_criteria=[
        "At least 5 peer-reviewed papers",
        "2+ production implementations",
        "Comparative analysis of approaches"
    ]
)

config = SwarmConfig(
    swarm_size=4,  # Different research sources
    aggregation_strategy="information_merge",
    timeout_seconds=120  # Longer for research
)

result = await swarm.execute(task, config)

# Result: Synthesized research from multiple sources
# {
#   "final_answer": {
#     "summary": "Comprehensive overview of few-shot learning...",
#     "key_papers": [
#       {"title": "...", "authors": [...], "year": 2024, "citations": 142},
#       ...
#     ],
#     "implementations": [
#       {"name": "PyTorch Meta-Learning", "github": "...", "stars": 3200},
#       ...
#     ],
#     "comparison_table": {...},
#     "recommendations": [...],
#     "sources_count": 47
#   },
#   "consensus_score": 0.88,
#   "execution_time_ms": 8900
# }

Testing Swarm Behavior

Unit Tests

import pytest
from unittest.mock import Mock, AsyncMock

@pytest.mark.asyncio
async def test_swarm_majority_vote():
    """Test majority voting aggregation."""

    proposals = [
        Proposal(arm_id="arm1", content="A", confidence=0.8, execution_time_ms=1000, status=ProposalStatus.COMPLETED),
        Proposal(arm_id="arm2", content="A", confidence=0.9, execution_time_ms=1200, status=ProposalStatus.COMPLETED),
        Proposal(arm_id="arm3", content="B", confidence=0.7, execution_time_ms=1100, status=ProposalStatus.COMPLETED),
    ]

    aggregator = ProposalAggregator()
    result = aggregator.majority_vote(proposals)

    assert result["answer"] == "A"
    assert result["vote_count"] == 2
    assert result["total_votes"] == 3

@pytest.mark.asyncio
async def test_swarm_conflict_detection():
    """Test conflict detection between proposals."""

    # Low consensus scenario
    proposals = [
        Proposal(arm_id="arm1", content="Solution A", confidence=0.8, execution_time_ms=1000, status=ProposalStatus.COMPLETED),
        Proposal(arm_id="arm2", content="Solution B", confidence=0.9, execution_time_ms=1200, status=ProposalStatus.COMPLETED),
        Proposal(arm_id="arm3", content="Solution C", confidence=0.7, execution_time_ms=1100, status=ProposalStatus.COMPLETED),
    ]

    resolver = ConflictResolver()
    conflict = resolver.detect_conflict(proposals, similarity_threshold=0.6)

    assert conflict is not None
    assert conflict.conflict_type == "low_consensus"
    assert conflict.severity in ["medium", "high"]

@pytest.mark.asyncio
async def test_swarm_execution():
    """Test full swarm execution flow."""

    # Mock arm registry
    registry = {
        "arm1": Mock(endpoint="http://arm1:8080", capabilities=["code"], success_rate=0.9),
        "arm2": Mock(endpoint="http://arm2:8080", capabilities=["code", "review"], success_rate=0.85),
        "arm3": Mock(endpoint="http://arm3:8080", capabilities=["security"], success_rate=0.95),
        "judge": Mock(endpoint="http://judge:8080", capabilities=["validation"], success_rate=0.92),
    }

    swarm = SwarmOrchestrator(registry, judge_arm_id="judge")

    # Mock arm calls
    swarm._call_arm = AsyncMock(return_value={
        "output": "Test result",
        "confidence": 0.85,
        "rationale": "Test rationale"
    })

    task = TaskContract(
        task_id="test-001",
        goal="Test swarm execution",
        priority=Priority.MEDIUM
    )

    config = SwarmConfig(swarm_size=3, aggregation_strategy="weighted_vote")

    result = await swarm.execute(task, config)

    assert result.final_answer is not None
    assert len(result.all_proposals) == 3
    assert 0.0 <= result.consensus_score <= 1.0
    assert result.execution_time_ms > 0

Integration Tests

@pytest.mark.asyncio
@pytest.mark.integration
async def test_swarm_with_real_arms():
    """Test swarm with actual arm services."""

    # Assumes arm services are running (e.g., via docker-compose)

    registry = {
        "coder": ArmCapability(
            arm_id="coder",
            name="Coder Arm",
            endpoint="http://localhost:8100/code",
            capabilities=["code_generation"],
            success_rate=0.9
        ),
        "judge": ArmCapability(
            arm_id="judge",
            name="Judge Arm",
            endpoint="http://localhost:8102/validate",
            capabilities=["validation"],
            success_rate=0.92
        ),
    }

    swarm = SwarmOrchestrator(registry, judge_arm_id="judge")

    task = TaskContract(
        task_id="integration-test-001",
        goal="Write a Python function to calculate Fibonacci numbers",
        acceptance_criteria=["Includes docstring", "Has unit tests"]
    )

    config = SwarmConfig(swarm_size=2, aggregation_strategy="confidence_max")

    result = await swarm.execute(task, config)

    # Verify result structure
    assert "final_answer" in result.dict()
    assert result.consensus_score >= 0.0

    # Verify proposals were generated
    assert len(result.all_proposals) == 2
    for proposal in result.all_proposals:
        assert proposal.status == ProposalStatus.COMPLETED
        assert proposal.confidence > 0.0

Performance Tests

@pytest.mark.asyncio
@pytest.mark.performance
async def test_swarm_latency():
    """Verify swarm executes within acceptable latency bounds."""

    import time

    swarm = SwarmOrchestrator(mock_registry)

    # Mock fast arms (100ms each)
    swarm._execute_single_arm = AsyncMock(
        side_effect=lambda *args, **kwargs: asyncio.sleep(0.1)
    )

    task = TaskContract(task_id="perf-001", goal="Performance test")
    config = SwarmConfig(swarm_size=5)

    start = time.time()
    result = await swarm.execute(task, config)
    elapsed = time.time() - start

    # With 5 arms executing in parallel, total time should be ~100ms + overhead
    # Allow 500ms for overhead
    assert elapsed < 0.6, f"Swarm took {elapsed}s (expected < 0.6s)"

@pytest.mark.asyncio
async def test_swarm_handles_arm_failures():
    """Verify swarm degrades gracefully when arms fail."""

    swarm = SwarmOrchestrator(mock_registry)

    # Mock arms: 2 succeed, 1 fails
    call_count = 0
    async def mock_execute(*args, **kwargs):
        nonlocal call_count
        call_count += 1
        if call_count == 2:
            raise Exception("Arm failed")
        await asyncio.sleep(0.1)

    swarm._execute_single_arm = AsyncMock(side_effect=mock_execute)

    task = TaskContract(task_id="fail-001", goal="Failure test")
    config = SwarmConfig(swarm_size=3)

    # Should still succeed with 2/3 arms
    result = await swarm.execute(task, config)

    assert len(result.all_proposals) == 3
    successful = [p for p in result.all_proposals if p.status == ProposalStatus.COMPLETED]
    assert len(successful) == 2

Troubleshooting

Common Issues

1. Low Consensus Score

Symptom: Swarm returns low consensus score (< 0.5)

Causes:

• Arms are using very different approaches
• Task is ambiguous or underspecified
• Arms have divergent interpretations

Solutions:

# Add more context to task
task.context["approach_hint"] = "Use iterative approach"

# Increase swarm size for more data points
config.swarm_size = 5

# Enable judge for arbitration
config.enable_judge = True

2. Swarm Timeout

Symptom: Some or all arms timeout

Causes:

• Arms are slow (complex LLM calls)
• Network issues
• Timeout set too low

Solutions:

# Increase timeout
config.timeout_seconds = 120

# Use faster models for swarm
task.context["prefer_fast_models"] = True

# Reduce swarm size
config.swarm_size = 3

3. High Cost

Symptom: Swarm execution costs exceed budget

Causes:

• Too many arms
• Expensive models used
• Multiple swarm rounds

Solutions:

# Reduce swarm size
config.swarm_size = 2

# Use cheaper models
task.context["model"] = "gpt-3.5-turbo"

# Disable judge if not critical
config.enable_judge = False

# Enable early termination
config.enable_early_termination = True

4. Contradictory Results

Symptom: Arms return contradictory answers

Causes:

• Task has multiple valid solutions
• Arms interpret differently
• Genuine disagreement

Solutions:

# Enable conflict resolution
config.enable_judge = True

# Clarify task goal
task.goal = "Identify THE MOST CRITICAL vulnerability (singular)"

# Add tiebreaker criteria
task.acceptance_criteria.append("Prioritize by OWASP severity ranking")

Debug Logging

import structlog

logger = structlog.get_logger()

# Enable detailed swarm logging
logger.info(
    "swarm.debug",
    task_id=task.task_id,
    selected_arms=selected_arms,
    proposals=[
        {
            "arm": p.arm_id,
            "confidence": p.confidence,
            "content_preview": str(p.content)[:100]
        }
        for p in proposals
    ],
    consensus_score=consensus_score,
    aggregation_strategy=config.aggregation_strategy
)

Summary

Swarm decision-making is a powerful Phase 2 capability that enables OctoLLM to:

1. Leverage diversity: Multiple arms bring unique perspectives
2. Increase robustness: System continues even if individual arms fail
3. Improve quality: Consensus mechanisms validate correctness
4. Handle complexity: Parallel processing tackles multi-faceted problems

Key Takeaways:

• Use swarm for high-stakes, complex, or quality-critical tasks
• Choose swarm size based on task priority and budget
• Select aggregation strategy based on task characteristics
• Enable judge for conflict resolution when needed
• Monitor performance and costs carefully
• Test swarm behavior thoroughly before production

Next Steps:

• Review Deployment Guide for production setup
• See Security Testing for swarm security patterns
• Consult Performance Tuning for optimization

Document Version: 1.0 Last Updated: 2025-11-10 Maintained By: OctoLLM Core Team

Architecture Decision Records

Architecture Decision Records (ADRs) document significant architectural choices made during OctoLLM development.

ADR Index

1. ADR-001: Technology Stack

   • Python vs Rust for services
   • LLM provider selection
   • Database and caching choices

2. ADR-002: Communication Patterns

   • REST vs gRPC
   • Message bus selection
   • Inter-service communication

3. ADR-003: Memory Architecture

   • Global semantic memory design
   • Local episodic memory
   • Vector store selection

4. ADR-004: Security Model

   • Capability-based isolation
   • Secrets management
   • Authentication/authorization

5. ADR-005: Deployment Platform

   • Kubernetes vs Docker Swarm
   • Cloud vs on-premise
   • Scaling strategy

6. ADR-006: Cloud Provider Selection

   • AWS vs GCP vs Azure
   • Cost considerations
   • Service availability

7. ADR-007: Unraid Local Deployment

   • Local development setup
   • Container orchestration
   • Resource management

ADR Template

When creating new ADRs, use the following template:

# ADR-XXX: Title

**Status**: Proposed | Accepted | Deprecated | Superseded
**Date**: YYYY-MM-DD
**Deciders**: Names
**Consulted**: Names

## Context

What is the issue we're facing?

## Decision

What did we decide?

## Consequences

What are the trade-offs?

### Positive
- Benefit 1
- Benefit 2

### Negative
- Drawback 1
- Drawback 2

## Alternatives Considered

1. Alternative 1
   - Pros
   - Cons
   - Why rejected

2. Alternative 2
   - Pros
   - Cons
   - Why rejected

See Also

• Architecture Overview
• System Design

ADR-001: Technology Stack Selection

Status: Accepted Date: 2025-11-10 Decision Makers: Architecture Team, Engineering Leads Consulted: Development Team, DevOps Team

Context

OctoLLM requires a technology stack that supports:

• High-performance request processing (>10,000 req/s for Reflex Layer)
• Async I/O for LLM API calls and database operations
• Vector similarity search for episodic memory
• Reliable data storage with ACID guarantees
• Fast caching for frequently accessed data
• Multiple specialized components (orchestrator, arms, reflex layer)
• Cloud-native deployment (Kubernetes)
• Developer productivity and maintainability

The system has diverse performance requirements:

• Reflex Layer: <10ms P95 latency, >10,000 req/s throughput
• Orchestrator: Complex routing logic, multiple concurrent operations
• Arms: LLM integration, specialized processing
• Memory: Vector search, relational queries, caching

Decision

We will use the following technology stack:

Core Languages

Python 3.11+ (Primary)

• Used for: Orchestrator, all Arms, API services
• Framework: FastAPI for HTTP APIs
• Async: asyncio for concurrent operations
• Reasons:
  • Excellent LLM ecosystem (OpenAI, Anthropic SDKs)
  • Strong async support with asyncio/FastAPI
  • Rich data processing libraries
  • High developer productivity
  • Large talent pool
  • Extensive testing frameworks

Rust 1.75+ (Performance-Critical)

• Used for: Reflex Layer, Tool Executor
• Framework: Axum for HTTP
• Reasons:
  • Zero-cost abstractions for performance
  • Memory safety without garbage collection
  • Excellent async runtime (tokio)
  • Pattern matching for PII detection
  • No runtime overhead
  • Strong type system prevents bugs

Databases

PostgreSQL 15+ (Primary Data Store)

• Used for: Global knowledge graph, task history, provenance
• Reasons:
  • ACID guarantees for critical data
  • JSONB for flexible schemas
  • Full-text search with GIN indexes
  • Excellent performance for relational queries
  • Mature replication and backup tools
  • Strong community support

Qdrant 1.7+ (Vector Database)

• Used for: Episodic memory (code examples, patterns)
• Reasons:
  • Optimized for similarity search
  • Built in Rust (high performance)
  • Filtering support for hybrid search
  • Supports multiple distance metrics
  • Good Python SDK
  • Active development

Redis 7+ (Cache & Pub/Sub)

• Used for: L2 cache, rate limiting, session state, events
• Reasons:
  • In-memory performance (<1ms latency)
  • Rich data structures (strings, hashes, sets, sorted sets)
  • Pub/sub for event messaging
  • TTL support for automatic expiration
  • Persistence options (AOF, RDB)
  • Cluster mode for scale

Web Framework

FastAPI (Python)

• Reasons:
  • Built on Starlette (async ASGI)
  • Automatic OpenAPI documentation
  • Pydantic integration for validation
  • Excellent async support
  • Dependency injection
  • WebSocket support
  • Strong type hints

Axum (Rust)

• Reasons:
  • Built on tokio (async runtime)
  • Type-safe routing
  • Minimal overhead
  • Good ecosystem integration
  • Composable middleware

Async Runtime

Python: asyncio + uvicorn

• ASGI server with excellent performance
• Integrates with FastAPI
• Multiple worker processes for CPU utilization

Rust: tokio

• Industry-standard async runtime
• Work-stealing scheduler
• Efficient I/O operations

Deployment

Docker + Docker Compose

• Development: Easy local setup
• Production: Standardized containers
• CI/CD: Consistent builds

Kubernetes

• Production orchestration
• Auto-scaling with HPA
• Rolling updates
• Service discovery
• Health checks

Supporting Tools

Monitoring:

• Prometheus: Metrics collection
• Grafana: Visualization
• Alertmanager: Alert routing
• Loki: Log aggregation (optional)
• Jaeger: Distributed tracing (optional)

Development:

• Poetry: Python dependency management
• Cargo: Rust build tool
• Black/isort/ruff: Python formatting/linting
• rustfmt/clippy: Rust formatting/linting
• pre-commit: Git hooks
• pytest: Python testing
• cargo test: Rust testing

Consequences

Positive

1. Performance:

   • Rust delivers <10ms latency for Reflex Layer
   • Async Python handles thousands of concurrent operations
   • Redis provides sub-millisecond caching
   • Qdrant optimized for vector search

2. Developer Experience:

   • Python enables rapid development
   • FastAPI auto-generates API docs
   • Strong typing catches bugs early
   • Extensive libraries available

3. Scalability:

   • Kubernetes enables horizontal scaling
   • Stateless services easy to replicate
   • Database clustering supported
   • Redis can scale with cluster mode

4. Maintainability:

   • Type hints improve code clarity
   • Rust prevents memory bugs
   • PostgreSQL ensures data integrity
   • Docker standardizes deployments

5. Ecosystem:

   • Rich LLM integration libraries
   • Mature database drivers
   • Active communities
   • Abundant learning resources

Negative

1. Complexity:

   • Two languages to maintain (Python + Rust)
   • Different build tools and workflows
   • Team needs skills in both languages
   • More complex CI/CD pipeline

2. Learning Curve:

   • Rust has steep learning curve
   • Async programming can be challenging
   • Kubernetes requires operations expertise
   • Multiple databases to manage

3. Resource Usage:

   • Three databases increase infrastructure cost
   • Kubernetes overhead for small deployments
   • Development environment is heavyweight
   • Local testing requires significant resources

4. Operational Overhead:

   • More components to monitor
   • More failure modes
   • Complex troubleshooting
   • Data consistency across databases

Mitigation Strategies

1. Language Complexity:

   • Keep Rust components minimal (Reflex, Executor only)
   • Provide Python fallbacks where feasible
   • Comprehensive documentation
   • Code review focus on readability

2. Learning Curve:

   • Training programs for team
   • Pair programming for knowledge sharing
   • Start contributors with Python
   • Document common patterns

3. Resource Usage:

   • Provide lightweight dev mode (Docker Compose)
   • Use resource limits in Kubernetes
   • Optimize container images
   • Implement efficient caching

4. Operational Complexity:

   • Comprehensive monitoring and alerting
   • Automated deployment pipelines
   • Disaster recovery procedures
   • Regular operational training

Alternatives Considered

1. Go for Performance-Critical Components

Pros:

• Good performance (better than Python)
• Simpler than Rust
• Excellent concurrency model
• Single binary deployment

Cons:

• Not as fast as Rust (<10ms requirement tight)
• Garbage collection introduces latency variance
• Weaker type system than Rust
• Less memory safe

Why Rejected: Rust provides better latency guarantees and memory safety for our <10ms P95 requirement.

2. Node.js/TypeScript for All Services

Pros:

• Single language across stack
• Good async support
• Large ecosystem
• Fast development

Cons:

• Not ideal for CPU-intensive tasks
• Weaker LLM library support
• Memory usage higher than Python
• Type system not as strong as Python + mypy

Why Rejected: Python has superior LLM ecosystem and better data processing libraries.

3. Java/Spring Boot

Pros:

• Mature enterprise ecosystem
• Strong typing
• Excellent tooling
• Large talent pool

Cons:

• Slower development than Python
• Higher memory usage
• More verbose code
• Weaker LLM integration

Why Rejected: Python provides better developer experience and LLM integration.

4. All Python (including performance-critical)

Pros:

• Single language
• Simpler deployment
• Easier team management
• Unified tooling

Cons:

• Cannot meet <10ms P95 latency consistently
• GIL limits true parallelism
• Higher memory usage
• No compile-time safety

Why Rejected: Cannot achieve required performance for Reflex Layer without Rust.

5. MongoDB instead of PostgreSQL

Pros:

• Flexible schema
• Horizontal scaling built-in
• Good for unstructured data

Cons:

• Weaker ACID guarantees
• No SQL JOIN support
• Transaction model more limited
• Less mature tooling

Why Rejected: Need ACID guarantees for critical data and complex relational queries.

6. Elasticsearch instead of Qdrant

Pros:

• Mature ecosystem
• Full-text search excellent
• Powerful aggregations

Cons:

• Not optimized for vector search
• Higher resource usage
• More complex to operate
• Slower vector operations

Why Rejected: Qdrant is purpose-built for vector similarity search with better performance.

References

• FastAPI Documentation
• Rust Async Book
• PostgreSQL Documentation
• Qdrant Documentation
• Redis Documentation
• Kubernetes Documentation
• Python asyncio Documentation

Last Review: 2025-11-10 Next Review: 2026-05-10 (6 months) Related ADRs: ADR-002, ADR-003, ADR-005

ADR-002: Communication Patterns

Status: Accepted Date: 2025-11-10 Decision Makers: Architecture Team Consulted: Engineering Team

Context

OctoLLM has multiple components that need to communicate:

• Reflex Layer → Orchestrator (request preprocessing)
• Orchestrator → Arms (task execution)
• Arms → Arms (collaborative tasks)
• Arms → Memory Systems (knowledge retrieval/storage)
• Components → External Services (LLM APIs, webhooks)

Communication patterns must support:

• Synchronous request-response for task execution
• Asynchronous event notifications
• Low latency (<100ms for internal calls)
• Reliability and fault tolerance
• Observability and tracing
• Flexible routing and load balancing

Decision

We will use the following communication patterns:

1. HTTP/REST for Synchronous Operations

Use For:

• Reflex Layer → Orchestrator
• Orchestrator → Arms
• Arms → Memory Systems
• External API integrations

Protocol: HTTP/1.1 or HTTP/2 Format: JSON Authentication: JWT tokens with capability scopes

Example:

# Orchestrator calling Coder Arm
async def execute_code_task(task: TaskContract) -> str:
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "http://coder-arm:8102/execute",
            json=task.dict(),
            headers={
                "Authorization": f"Bearer {capability_token}",
                "X-Request-ID": request_id
            },
            timeout=30.0
        )
        return response.json()["output"]

Reasons:

• Universal protocol, widely understood
• Excellent debugging tools
• Native HTTP client libraries
• OpenAPI documentation support
• Load balancer integration
• Request/response tracing

2. Redis Pub/Sub for Event Notifications

Use For:

• Task completion events
• System health events
• Audit log events
• Cache invalidation signals

Pattern: Publish-subscribe Channels: Topic-based routing

Example:

# Publisher (Orchestrator)
await redis.publish(
    "events:task:completed",
    json.dumps({
        "task_id": task.task_id,
        "status": "completed",
        "timestamp": datetime.utcnow().isoformat()
    })
)

# Subscriber (Monitoring Service)
pubsub = redis.pubsub()
pubsub.subscribe("events:task:*")

async for message in pubsub.listen():
    if message["type"] == "message":
        event = json.loads(message["data"])
        handle_task_event(event)

Reasons:

• Decoupled producers and consumers
• No blocking on publisher side
• Multiple subscribers supported
• Built into existing Redis infrastructure
• Low latency (<5ms)
• Simple implementation

3. Direct HTTP for Arm-to-Arm Communication

Use For:

• Coder Arm → Judge Arm (code validation)
• Planner Arm → Executor Arm (plan execution)
• Retriever Arm → other Arms (knowledge lookup)

Pattern: Direct service-to-service HTTP calls Discovery: Kubernetes DNS or service registry

Example:

# Coder Arm requesting validation from Judge Arm
async def validate_code(code: str) -> bool:
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "http://judge-arm:8103/validate",
            json={"code": code, "language": "python"},
            headers={"Authorization": f"Bearer {token}"}
        )
        return response.json()["is_valid"]

Reasons:

• Simple and direct
• Low latency
• Easy to trace with request IDs
• No message broker overhead
• Kubernetes service discovery

4. WebSocket for Real-Time Updates

Use For:

• Live task progress updates to clients
• Streaming LLM responses
• Real-time dashboard data

Protocol: WebSocket over HTTP Format: JSON messages

Example:

# Server
@app.websocket("/ws/tasks/{task_id}")
async def task_updates(websocket: WebSocket, task_id: str):
    await websocket.accept()
    try:
        while True:
            update = await get_task_update(task_id)
            await websocket.send_json(update)
            await asyncio.sleep(1)
    except WebSocketDisconnect:
        logger.info("Client disconnected", task_id=task_id)

# Client
async with websocket_connect(f"ws://localhost:8000/ws/tasks/{task_id}") as ws:
    async for message in ws:
        update = json.loads(message)
        print(f"Task progress: {update['progress']}%")

Reasons:

• Bi-directional communication
• Lower overhead than polling
• Native browser support
• Streaming responses
• Real-time updates

Consequences

Positive

1. Simplicity:

   • HTTP/REST is familiar to all developers
   • No complex message broker to manage
   • Standard debugging tools work
   • Easy to test and mock

2. Performance:

   • HTTP/2 multiplexing reduces overhead
   • Direct calls minimize latency
   • Redis pub/sub is very fast
   • Connection pooling improves efficiency

3. Observability:

   • HTTP requests easily traced
   • Standard headers for correlation
   • OpenTelemetry integration
   • Request/response logging

4. Flexibility:

   • Can add message broker later if needed
   • Easy to switch between sync and async
   • Support for multiple communication styles
   • Cloud-native patterns

5. Reliability:

   • HTTP retries well-understood
   • Circuit breakers easy to implement
   • Timeout handling straightforward
   • Failure modes are clear

Negative

1. No Native Message Queue:

   • No guaranteed delivery
   • No persistent queuing
   • Manual retry logic needed
   • No dead letter queue

2. Pub/Sub Limitations:

   • Messages not persisted
   • No acknowledgment mechanism
   • Subscribers must be online
   • No ordering guarantees

3. Service Discovery:

   • Requires DNS or service registry
   • Hard-coded URLs in development
   • More complex in multi-cluster setup
   • Need health checks

4. Scalability Concerns:

   • HTTP connection overhead at very high scale
   • May need connection pooling tuning
   • Pub/sub doesn't scale horizontally well
   • Load balancing configuration required

Mitigation Strategies

1. Reliability:

   • Implement retry logic with exponential backoff
   • Use circuit breakers for external calls
   • Add request timeouts
   • Idempotent operations where possible

2. Message Durability:

   • Use database for critical events
   • Add audit log for important operations
   • Implement task queue for background jobs
   • Consider Kafka for high-volume events (future)

3. Service Discovery:

   • Use Kubernetes DNS for production
   • Environment variables for URLs
   • Service mesh for advanced routing (future)
   • Health checks and readiness probes

4. Performance:

   • HTTP/2 for multiplexing
   • Connection pooling
   • Response compression
   • Caching where appropriate

Alternatives Considered

1. gRPC for All Communication

Pros:

• Better performance than REST
• Strong typing with protobuf
• Bi-directional streaming
• Code generation

Cons:

• More complex than HTTP/REST
• Requires protobuf definitions
• Harder to debug
• Less universal tooling
• Steeper learning curve

Why Rejected: HTTP/REST simplicity outweighs gRPC performance benefits for our use case.

2. Message Broker (RabbitMQ/Kafka)

Pros:

• Guaranteed delivery
• Persistent queuing
• Complex routing
• Horizontal scaling
• Decoupling

Cons:

• Another component to manage
• More operational complexity
• Higher latency
• Resource overhead
• Overkill for current scale

Why Rejected: HTTP/REST with Redis pub/sub sufficient for current needs. Can add later if needed.

3. Service Mesh (Istio/Linkerd)

Pros:

• Advanced routing
• Automatic retries
• Circuit breakers
• mTLS security
• Observability

Cons:

• Complex to setup
• Resource overhead
• Steep learning curve
• Operational burden
• Overkill for current scale

Why Rejected: Too complex for initial deployment. May consider for larger deployments.

4. GraphQL for All APIs

Pros:

• Flexible queries
• Single endpoint
• Strong typing
• Batch requests

Cons:

• More complex than REST
• Caching harder
• N+1 query problem
• Learning curve
• Less suitable for internal APIs

Why Rejected: REST is simpler and sufficient for our internal APIs.

Implementation Guidelines

HTTP Best Practices

1. Use standard status codes:

   • 200 OK: Success
   • 201 Created: Resource created
   • 400 Bad Request: Validation error
   • 401 Unauthorized: Authentication required
   • 403 Forbidden: Authorization failed
   • 404 Not Found: Resource doesn't exist
   • 429 Too Many Requests: Rate limit
   • 500 Internal Server Error: Server error
   • 503 Service Unavailable: Service down

2. Include correlation headers:

   headers = {
       "X-Request-ID": request_id,
       "X-Correlation-ID": correlation_id,
       "Authorization": f"Bearer {token}"
   }
   

3. Set appropriate timeouts:

   timeout = httpx.Timeout(
       connect=5.0,  # Connection timeout
       read=30.0,    # Read timeout
       write=10.0,   # Write timeout
       pool=5.0      # Pool timeout
   )
   

4. Use connection pooling:

   client = httpx.AsyncClient(
       limits=httpx.Limits(
           max_keepalive_connections=20,
           max_connections=100
       )
   )
   

Event Publishing

1. Event schema:

   {
       "event_type": "task.completed",
       "timestamp": "2025-11-10T10:30:00Z",
       "source": "orchestrator",
       "data": {
           "task_id": "task-123",
           "status": "completed",
           "duration_ms": 1234
       }
   }
   

2. Channel naming:

   • Format: <domain>:<entity>:<action>
   • Examples: events:task:completed, events:arm:registered

References

• HTTP/2 Specification
• REST API Best Practices
• Redis Pub/Sub Documentation
• WebSocket Protocol
• OpenTelemetry

Last Review: 2025-11-10 Next Review: 2026-05-10 (6 months) Related ADRs: ADR-001, ADR-004, ADR-005

ADR-003: Memory Architecture

Status: Accepted Date: 2025-11-10 Decision Makers: Architecture Team, ML Engineers Consulted: Database Team, Security Team

Context

OctoLLM needs a memory system that supports:

• Global Knowledge: Facts, entities, relationships shared across all tasks
• Episodic Memory: Task-specific examples, code patterns, solutions
• Short-term Cache: Frequently accessed data for performance
• Provenance Tracking: Audit trail of all operations
• Security Isolation: Prevent data leakage between security contexts
• Vector Search: Similarity-based retrieval for examples
• Relational Queries: Complex joins for knowledge graph
• High Performance: Low latency for memory operations

Memory requirements vary by use case:

• Knowledge graph queries: Need SQL joins, ACID guarantees
• Code example retrieval: Need vector similarity search
• Recent task lookup: Need fast key-value access
• Cross-task learning: Need shared knowledge repository

Decision

We will implement a three-tier memory architecture with routing and security isolation:

1. Global Memory (PostgreSQL)

Purpose: Shared knowledge graph across all tasks Storage: PostgreSQL with JSONB for flexible properties Access: SQL queries via SQLAlchemy ORM

Schema:

CREATE TABLE entities (
    id UUID PRIMARY KEY,
    entity_type VARCHAR(100) NOT NULL,
    name VARCHAR(500) NOT NULL,
    properties JSONB,
    created_at TIMESTAMP DEFAULT NOW(),
    updated_at TIMESTAMP DEFAULT NOW()
);

CREATE TABLE relationships (
    id UUID PRIMARY KEY,
    from_entity_id UUID REFERENCES entities(id),
    to_entity_id UUID REFERENCES entities(id),
    relationship_type VARCHAR(100) NOT NULL,
    properties JSONB,
    created_at TIMESTAMP DEFAULT NOW()
);

CREATE TABLE task_history (
    id UUID PRIMARY KEY,
    task_id UUID NOT NULL,
    status VARCHAR(50) NOT NULL,
    input TEXT,
    output TEXT,
    provenance JSONB,
    created_at TIMESTAMP DEFAULT NOW()
);

Use Cases:

• Storing discovered facts and entities
• Tracking relationships between concepts
• Maintaining task history and audit logs
• Querying for related knowledge

2. Episodic Memory (Qdrant)

Purpose: Task-specific examples and patterns Storage: Qdrant vector database Access: Vector similarity search

Collections:

• coder_memory: Code examples with embeddings
• planner_memory: Successful task decompositions
• judge_memory: Validation patterns

Example:

# Store code example
await qdrant_client.upsert(
    collection_name="coder_memory",
    points=[
        {
            "id": example_id,
            "vector": embedding,  # 1536-dim vector
            "payload": {
                "code": code_snippet,
                "language": "python",
                "task_description": description,
                "success": True,
                "timestamp": datetime.utcnow().isoformat()
            }
        }
    ]
)

# Retrieve similar examples
results = await qdrant_client.search(
    collection_name="coder_memory",
    query_vector=query_embedding,
    limit=5,
    query_filter={
        "must": [
            {"key": "language", "match": {"value": "python"}},
            {"key": "success", "match": {"value": True}}
        ]
    }
)

Use Cases:

• Finding similar code examples
• Retrieving relevant task patterns
• Learning from past successes
• Context for LLM prompts

3. Cache Layer (Redis + In-Memory)

L1 Cache (In-Memory):

• Library: cachetools TTLCache
• Size: 1,000 items per service
• TTL: 60 seconds
• Use: Hot data, arm capabilities

L2 Cache (Redis):

• Size: Unlimited (eviction policy: LRU)
• TTL: 1-3600 seconds (configurable)
• Use: Shared cache across services

Example:

class MultiLevelCache:
    def __init__(self):
        self.l1 = TTLCache(maxsize=1000, ttl=60)
        self.l2 = redis.Redis()

    async def get(self, key: str) -> Optional[str]:
        # Try L1
        if key in self.l1:
            return self.l1[key]

        # Try L2
        value = await self.l2.get(key)
        if value:
            self.l1[key] = value  # Promote to L1
            return value

        return None

4. Memory Router

Purpose: Route queries to appropriate memory system Logic: Based on query type and requirements

class MemoryRouter:
    async def query(self, query: MemoryQuery) -> List[Any]:
        if query.type == "vector_search":
            return await self.episodic_memory.search(query)
        elif query.type == "graph_query":
            return await self.global_memory.query(query)
        elif query.type == "recent_lookup":
            cached = await self.cache.get(query.key)
            if cached:
                return cached
            result = await self.global_memory.query(query)
            await self.cache.set(query.key, result)
            return result

5. Data Diodes (Security Isolation)

Purpose: Enforce security boundaries between memory contexts Implementation: Filtering layer before memory access

class DataDiode:
    async def filter_read(
        self,
        data: Any,
        capability: CapabilityToken
    ) -> Any:
        """Filter data based on capability scope."""
        if capability.scope == "task:read:own":
            # Only return data from user's tasks
            return [
                item for item in data
                if item.user_id == capability.user_id
            ]
        elif capability.scope == "task:read:all":
            # Admin can read all
            return data
        else:
            raise AuthorizationError("Insufficient permissions")

    async def filter_write(
        self,
        data: Any,
        capability: CapabilityToken
    ) -> None:
        """Validate write operations."""
        # Check for PII
        if contains_pii(data):
            raise SecurityViolation("PII detected in write")

        # Check authorization
        if not capability.can_write:
            raise AuthorizationError("No write permission")

Consequences

Positive

1. Performance:

   • L1 cache: sub-millisecond lookups
   • L2 cache: <5ms for common queries
   • Vector search: optimized for similarity
   • SQL: optimized for relations

2. Flexibility:

   • Right tool for each use case
   • Can optimize each layer independently
   • Easy to add new memory types
   • Supports diverse query patterns

3. Security:

   • Data diodes enforce boundaries
   • Capability-based access control
   • PII detection before storage
   • Audit trail in PostgreSQL

4. Scalability:

   • PostgreSQL: vertical + replication
   • Qdrant: horizontal scaling
   • Redis: cluster mode
   • Independent scaling per layer

5. Rich Queries:

   • SQL for complex joins
   • Vector search for similarity
   • Hybrid queries combining both
   • Full-text search in PostgreSQL

Negative

1. Complexity:

   • Three databases to manage
   • Data consistency challenges
   • More failure modes
   • Complex debugging

2. Data Synchronization:

   • No automatic sync between layers
   • Manual cache invalidation
   • Potential staleness issues
   • Consistency is eventual

3. Resource Usage:

   • Higher memory footprint
   • More infrastructure cost
   • Development environment heavier
   • Backup complexity

4. Operational Burden:

   • Three systems to monitor
   • Three backup strategies
   • More moving parts
   • Complex recovery procedures

Mitigation Strategies

1. Complexity:

   • Abstract behind unified API
   • Comprehensive documentation
   • Clear routing logic
   • Automated testing

2. Synchronization:

   • Well-defined TTLs
   • Event-driven invalidation
   • Version tracking
   • Monitoring for staleness

3. Resource Usage:

   • Resource limits in Kubernetes
   • Optimize cache sizes
   • Efficient data models
   • Regular cleanup jobs

4. Operations:

   • Unified monitoring dashboards
   • Automated backups
   • Runbooks for common issues
   • Health checks for all layers

Alternatives Considered

1. Single Database (PostgreSQL) with pgvector

Pros:

• Simpler architecture
• Single source of truth
• ACID guarantees everywhere
• Easier operations

Cons:

• Vector search not as optimized
• Performance trade-offs
• Single point of failure
• Harder to scale independently

Why Rejected: Vector search performance insufficient for production scale.

2. Graph Database (Neo4j) for Global Memory

Pros:

• Optimized for relationships
• Native graph queries
• Good visualization tools

Cons:

• Less familiar to team
• Higher operational complexity
• More expensive
• Cypher learning curve

Why Rejected: PostgreSQL with JSONB provides sufficient graph capabilities with familiar SQL.

3. Elasticsearch for All Memory

Pros:

• Full-text search excellent
• Horizontal scaling
• Rich query DSL

Cons:

• Not optimized for vectors
• Resource intensive
• Complex to operate
• Overkill for our needs

Why Rejected: Qdrant better for vectors, PostgreSQL better for structured data.

4. Single-Tier Cache (Redis only)

Pros:

• Simpler caching
• No L1/L2 coordination
• Less memory usage

Cons:

• Network latency for every lookup
• Higher Redis load
• No in-process caching benefit

Why Rejected: L1 cache provides significant performance improvement for hot data.

Implementation Guidelines

Global Memory Operations

# Store entity
entity = Entity(
    entity_type="file",
    name="config.yaml",
    properties={"path": "/etc/app/config.yaml", "size": 1024}
)
await global_memory.store_entity(entity)

# Store relationship
relationship = Relationship(
    from_entity_id=file_entity.id,
    to_entity_id=config_entity.id,
    relationship_type="contains",
    properties={"line": 42}
)
await global_memory.store_relationship(relationship)

# Query entities
files = await global_memory.query_entities(
    entity_type="file",
    filters={"properties.extension": "yaml"}
)

Episodic Memory Operations

# Store example
example = CodeExample(
    code="def hello(): print('world')",
    language="python",
    task_description="Print hello world"
)
embedding = await get_embedding(example.code)
await episodic_memory.store(example, embedding)

# Retrieve similar
query_embedding = await get_embedding("print greeting")
examples = await episodic_memory.search(
    query_embedding,
    filter={"language": "python"},
    limit=5
)

Cache Operations

# Store in cache
await cache.set(
    key="arm:capabilities:coder",
    value=json.dumps(capabilities),
    ttl=3600
)

# Retrieve from cache
cached = await cache.get("arm:capabilities:coder")
if cached:
    return json.loads(cached)

# Invalidate cache
await cache.delete("arm:capabilities:coder")

References

• PostgreSQL JSONB
• Qdrant Documentation
• Redis Caching Patterns
• Memory Architecture Patterns

Last Review: 2025-11-10 Next Review: 2026-05-10 (6 months) Related ADRs: ADR-001, ADR-004

ADR-004: Security Model

Status: Accepted Date: 2025-11-10 Decision Makers: Security Team, Architecture Team Consulted: Compliance Team, Engineering Team

Context

OctoLLM processes user tasks that may contain:

• Sensitive data (PII, credentials, proprietary information)
• Potentially malicious input (injections, exploits)
• Cross-user data that must be isolated
• LLM API requests that could be costly or unsafe

Security requirements:

• Prevent PII leakage: Detect and sanitize PII before storage
• Isolation: Prevent data leakage between users/tasks
• Input validation: Protect against injections and exploits
• Least privilege: Limit component access to minimum needed
• Auditability: Track all operations for compliance
• Defense in depth: Multiple security layers

Threat model:

• Malicious users attempting to access others' data
• Accidental PII exposure through LLM APIs
• Prompt injection attacks
• Resource exhaustion attacks
• Insider threats from compromised components

Decision

We will implement a capability-based security model with multiple defensive layers:

1. Capability Tokens (JWT)

Purpose: Fine-grained authorization based on capabilities Format: JWT with capability scopes Issuance: Orchestrator issues tokens with specific scopes Validation: Each component validates tokens before processing

Token Structure:

{
  "sub": "user-123",
  "iss": "octollm-orchestrator",
  "exp": 1699999999,
  "capabilities": {
    "task:read": ["task-456"],
    "task:execute": ["task-456"],
    "arm:invoke": ["coder", "executor"],
    "memory:read": ["global"],
    "memory:write": []
  },
  "context": {
    "task_id": "task-456",
    "user_id": "user-123",
    "session_id": "session-789"
  }
}

Example:

from jose import jwt

def create_capability_token(
    user_id: str,
    task_id: str,
    capabilities: Dict[str, List[str]],
    expiry_minutes: int = 30
) -> str:
    """Create capability token for task execution."""
    payload = {
        "sub": user_id,
        "iss": "octollm-orchestrator",
        "exp": datetime.utcnow() + timedelta(minutes=expiry_minutes),
        "capabilities": capabilities,
        "context": {
            "task_id": task_id,
            "user_id": user_id
        }
    }
    return jwt.encode(payload, SECRET_KEY, algorithm="HS256")

async def verify_capability(
    token: str,
    required_capability: str,
    resource_id: Optional[str] = None
) -> bool:
    """Verify token has required capability."""
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])

        capabilities = payload.get("capabilities", {})
        allowed = capabilities.get(required_capability, [])

        if resource_id:
            return resource_id in allowed
        return len(allowed) > 0

    except jwt.JWTError:
        return False

2. PII Detection (Reflex Layer)

Purpose: Detect and sanitize PII before processing Location: Reflex Layer (first line of defense) Method: Regex patterns + optional ML model

Patterns:

lazy_static! {
    static ref EMAIL: Regex = Regex::new(
        r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b"
    ).unwrap();

    static ref SSN: Regex = Regex::new(
        r"\b\d{3}-\d{2}-\d{4}\b"
    ).unwrap();

    static ref CREDIT_CARD: Regex = Regex::new(
        r"\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b"
    ).unwrap();

    static ref PHONE: Regex = Regex::new(
        r"\b\d{3}[-.]?\d{3}[-.]?\d{4}\b"
    ).unwrap();
}

pub struct PiiDetector {
    patterns: Vec<(String, Regex)>,
}

impl PiiDetector {
    pub fn detect(&self, text: &str) -> Vec<PiiMatch> {
        let mut matches = Vec::new();

        for (name, pattern) in &self.patterns {
            for capture in pattern.captures_iter(text) {
                matches.push(PiiMatch {
                    pattern_name: name.clone(),
                    matched_text: capture[0].to_string(),
                    start: capture.get(0).unwrap().start(),
                    end: capture.get(0).unwrap().end(),
                });
            }
        }

        matches
    }

    pub fn sanitize(&self, text: &str) -> String {
        let mut result = text.to_string();

        for (_, pattern) in &self.patterns {
            result = pattern.replace_all(&result, "[REDACTED]").to_string();
        }

        result
    }
}

3. Input Validation

Layers:

1. Schema validation (Pydantic)
2. Business logic validation
3. Security validation (injection detection)

Example:

from pydantic import BaseModel, Field, validator

class TaskRequest(BaseModel):
    """Validated task request."""

    description: str = Field(
        ...,
        min_length=10,
        max_length=10000,
        description="Task description"
    )
    priority: int = Field(
        default=5,
        ge=1,
        le=10,
        description="Task priority (1-10)"
    )
    timeout: int = Field(
        default=300,
        gt=0,
        le=3600,
        description="Task timeout in seconds"
    )

    @validator('description')
    def validate_description(cls, v: str) -> str:
        """Validate description for security."""
        # Check for SQL injection patterns
        sql_patterns = ["'; DROP TABLE", "-- ", "/*", "*/"]
        for pattern in sql_patterns:
            if pattern.lower() in v.lower():
                raise ValueError("Potential SQL injection detected")

        # Check for command injection
        cmd_patterns = [";", "&&", "||", "|", "`", "$("]
        for pattern in cmd_patterns:
            if pattern in v:
                raise ValueError("Potential command injection detected")

        return v.strip()

4. Rate Limiting

Purpose: Prevent resource exhaustion Implementation: Token bucket algorithm in Reflex Layer

Example:

pub struct RateLimiter {
    buckets: HashMap<String, TokenBucket>,
    rate: u32,
    capacity: u32,
}

impl RateLimiter {
    pub fn check(&mut self, key: &str) -> Result<(), RateLimitError> {
        let bucket = self.buckets
            .entry(key.to_string())
            .or_insert_with(|| TokenBucket::new(self.capacity));

        bucket.refill(self.rate);

        if bucket.consume(1) {
            Ok(())
        } else {
            Err(RateLimitError {
                limit: self.rate,
                retry_after: bucket.retry_after(),
            })
        }
    }
}

5. Audit Logging

Purpose: Compliance and forensics Storage: PostgreSQL with immutable logs

Example:

async def log_security_event(
    event_type: str,
    user_id: str,
    action: str,
    resource: str,
    outcome: str,
    details: Dict[str, Any]
):
    """Log security event for audit trail."""
    await db.execute("""
        INSERT INTO security_audit_log (
            event_type, user_id, action, resource, outcome, details
        ) VALUES ($1, $2, $3, $4, $5, $6)
    """, event_type, user_id, action, resource, outcome, json.dumps(details))

# Usage
await log_security_event(
    event_type="authentication",
    user_id="user-123",
    action="login",
    resource="api",
    outcome="success",
    details={"ip": "192.168.1.1", "user_agent": "..."}
)

6. Defense in Depth

Layers:

1. Network: Kubernetes Network Policies, TLS
2. Input: Reflex Layer PII detection, validation
3. Access: Capability tokens, RBAC
4. Data: Encryption at rest, data diodes
5. Output: Output validation, sanitization
6. Monitoring: Security metrics, alerts
7. Audit: Comprehensive logging

Consequences

Positive

1. Fine-Grained Control:

   • Capabilities limit access precisely
   • Tokens expire automatically
   • Scopes prevent over-privileging
   • Easy to revoke access

2. PII Protection:

   • Automatic detection in Reflex Layer
   • Prevents accidental exposure
   • Sanitization before LLM APIs
   • Compliance-friendly

3. Defense in Depth:

   • Multiple security layers
   • Failure in one layer doesn't compromise system
   • Comprehensive protection
   • Audit trail for forensics

4. Performance:

   • PII detection in fast Rust code
   • JWT validation is local (no DB lookup)
   • Rate limiting prevents overload
   • Minimal overhead

5. Auditability:

   • All operations logged
   • Immutable audit trail
   • Compliance requirements met
   • Forensics support

Negative

1. Complexity:

   • Capability tokens add overhead
   • PII patterns need maintenance
   • More code to test
   • Learning curve for developers

2. False Positives:

   • PII regex may over-detect
   • Legitimate data may be redacted
   • User experience impact
   • Manual review needed

3. Performance Overhead:

   • PII detection adds latency (<5ms)
   • JWT validation on every request
   • Rate limiting checks
   • Audit logging I/O

4. Operational Burden:

   • Key management for JWT
   • PII pattern updates
   • Audit log retention
   • Security monitoring

Mitigation Strategies

1. Complexity:

   • Comprehensive documentation
   • Helper libraries for common cases
   • Automated testing
   • Training for developers

2. False Positives:

   • Tunable PII patterns
   • Whitelist for known-safe data
   • User feedback mechanism
   • Regular pattern review

3. Performance:

   • Optimize PII regex
   • Cache JWT validations
   • Batch audit logs
   • Monitor overhead

4. Operations:

   • Automated key rotation
   • Monitoring dashboards
   • Alerting for anomalies
   • Runbooks for incidents

Alternatives Considered

1. OAuth 2.0 / OIDC

Pros:

• Industry standard
• Rich ecosystem
• Identity federation
• Well-understood

Cons:

• More complex than needed
• External dependencies
• Token introspection overhead
• Capability model not native

Why Rejected: Capability tokens provide simpler, fine-grained control for internal services.

2. mTLS for All Communication

Pros:

• Strong authentication
• End-to-end encryption
• Certificate-based

Cons:

• Complex certificate management
• Higher operational burden
• Not necessary for internal services
• Overkill for current scale

Why Rejected: TLS with capability tokens sufficient for our threat model.

3. ML-Based PII Detection

Pros:

• Better accuracy
• Contextual understanding
• Fewer false positives

Cons:

• Higher latency
• Model management complexity
• Resource intensive
• Harder to explain decisions

Why Rejected: Regex patterns sufficient for current needs, can add ML later if needed.

4. Role-Based Access Control (RBAC) Only

Pros:

• Simpler than capabilities
• Familiar model
• Standard implementation

Cons:

• Coarser-grained access
• Can't limit to specific tasks
• Role explosion problem
• Less flexible

Why Rejected: Capabilities provide finer control needed for task-level isolation.

Implementation Guidelines

See Security Overview for detailed implementation guidance.

References

• OWASP Top 10
• JWT Best Practices
• GDPR Compliance
• Capability-Based Security

Last Review: 2025-11-10 Next Review: 2026-02-10 (Quarterly - higher frequency for security) Related ADRs: ADR-001, ADR-002, ADR-003

ADR-005: Deployment Platform

Status: Accepted Date: 2025-11-10 Decision Makers: Architecture Team, DevOps Team Consulted: Engineering Team, Operations Team

Context

OctoLLM requires a deployment platform that supports:

• Multi-component orchestration: Orchestrator, multiple Arms, Reflex Layer, Memory systems
• Scalability: Horizontal scaling for Arms, vertical scaling for databases
• Service discovery: Components need to find each other dynamically
• Health monitoring: Automatic restarts, health checks, readiness probes
• Resource management: CPU/memory limits, quotas, efficient allocation
• Rolling updates: Zero-downtime deployments
• Configuration management: Environment-specific configs, secrets
• Development parity: Local development should mirror production
• Cloud agnostic: No vendor lock-in, portable across providers

Deployment requirements:

• Production: High availability, auto-scaling, monitoring, observability
• Staging: Production-like environment for testing
• Development: Fast iteration, easy debugging, minimal resource usage
• CI/CD: Automated builds, tests, deployments

Environment characteristics:

• Local Dev: Docker Compose, single machine, easy setup
• Staging: Kubernetes cluster, production-like, testing
• Production: Kubernetes cluster, multi-region (future), HA databases

Decision

We will use Kubernetes for production and Docker Compose for development with a cloud-agnostic architecture:

1. Production Deployment (Kubernetes)

Platform: Kubernetes 1.28+ Distribution: Any CNCF-certified (EKS, GKE, AKS, or self-hosted) Approach: Cloud-agnostic, no vendor-specific services

Why Kubernetes:

• Industry-standard container orchestration
• Rich ecosystem (Helm, Kustomize, operators)
• Excellent service discovery and load balancing
• Horizontal Pod Autoscaler (HPA) for auto-scaling
• Rolling updates with zero downtime
• Self-healing (automatic restarts)
• Resource management and quotas
• Multi-cloud portability

Architecture:

# Namespace organization
octollm-system/      # System components (monitoring, ingress)
octollm-production/  # Production workloads
octollm-staging/     # Staging workloads

# Components
- Deployment: orchestrator (3 replicas)
- Deployment: coder-arm (5 replicas, HPA)
- Deployment: judge-arm (3 replicas, HPA)
- Deployment: executor-arm (5 replicas, HPA)
- Deployment: planner-arm (3 replicas, HPA)
- Deployment: retriever-arm (3 replicas, HPA)
- DaemonSet: reflex-layer (1 per node)
- StatefulSet: postgresql (3 replicas, HA)
- StatefulSet: qdrant (3 replicas)
- StatefulSet: redis (3 replicas, sentinel)

Example Deployment:

apiVersion: apps/v1
kind: Deployment
metadata:
  name: orchestrator
  namespace: octollm-production
  labels:
    app: orchestrator
    version: v1.0.0
spec:
  replicas: 3
  strategy:
    type: RollingUpdate
    rollingUpdate:
      maxSurge: 1
      maxUnavailable: 0
  selector:
    matchLabels:
      app: orchestrator
  template:
    metadata:
      labels:
        app: orchestrator
        version: v1.0.0
      annotations:
        prometheus.io/scrape: "true"
        prometheus.io/port: "8000"
    spec:
      serviceAccountName: orchestrator
      containers:
      - name: orchestrator
        image: octollm/orchestrator:v1.0.0
        ports:
        - containerPort: 8000
          name: http
        - containerPort: 9090
          name: metrics
        env:
        - name: ENVIRONMENT
          value: "production"
        - name: LOG_LEVEL
          value: "INFO"
        - name: DATABASE_URL
          valueFrom:
            secretKeyRef:
              name: database-credentials
              key: url
        - name: REDIS_URL
          valueFrom:
            secretKeyRef:
              name: redis-credentials
              key: url
        resources:
          requests:
            cpu: "500m"
            memory: "512Mi"
          limits:
            cpu: "2000m"
            memory: "2Gi"
        livenessProbe:
          httpGet:
            path: /health/live
            port: 8000
          initialDelaySeconds: 30
          periodSeconds: 10
          timeoutSeconds: 5
          failureThreshold: 3
        readinessProbe:
          httpGet:
            path: /health/ready
            port: 8000
          initialDelaySeconds: 10
          periodSeconds: 5
          timeoutSeconds: 3
          failureThreshold: 2
        securityContext:
          runAsNonRoot: true
          runAsUser: 1000
          allowPrivilegeEscalation: false
          readOnlyRootFilesystem: true
          capabilities:
            drop:
            - ALL
---
apiVersion: v1
kind: Service
metadata:
  name: orchestrator
  namespace: octollm-production
spec:
  type: ClusterIP
  selector:
    app: orchestrator
  ports:
  - name: http
    port: 8000
    targetPort: 8000
  - name: metrics
    port: 9090
    targetPort: 9090
---
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: orchestrator-hpa
  namespace: octollm-production
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: orchestrator
  minReplicas: 3
  maxReplicas: 10
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70
  - type: Resource
    resource:
      name: memory
      target:
        type: Utilization
        averageUtilization: 80
  behavior:
    scaleDown:
      stabilizationWindowSeconds: 300
      policies:
      - type: Percent
        value: 50
        periodSeconds: 60
    scaleUp:
      stabilizationWindowSeconds: 0
      policies:
      - type: Percent
        value: 100
        periodSeconds: 30
      - type: Pods
        value: 2
        periodSeconds: 30
      selectPolicy: Max

Arm Deployment Example:

apiVersion: apps/v1
kind: Deployment
metadata:
  name: coder-arm
  namespace: octollm-production
spec:
  replicas: 5
  selector:
    matchLabels:
      app: coder-arm
  template:
    metadata:
      labels:
        app: coder-arm
    spec:
      containers:
      - name: coder-arm
        image: octollm/coder-arm:v1.0.0
        ports:
        - containerPort: 8102
        env:
        - name: ARM_TYPE
          value: "coder"
        - name: LLM_API_KEY
          valueFrom:
            secretKeyRef:
              name: llm-credentials
              key: api-key
        resources:
          requests:
            cpu: "1000m"
            memory: "1Gi"
          limits:
            cpu: "4000m"
            memory: "4Gi"
        livenessProbe:
          httpGet:
            path: /health
            port: 8102
          initialDelaySeconds: 30
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /ready
            port: 8102
          initialDelaySeconds: 10
          periodSeconds: 5

Reflex Layer (DaemonSet):

apiVersion: apps/v1
kind: DaemonSet
metadata:
  name: reflex-layer
  namespace: octollm-production
spec:
  selector:
    matchLabels:
      app: reflex-layer
  template:
    metadata:
      labels:
        app: reflex-layer
    spec:
      hostNetwork: true  # For low-latency
      containers:
      - name: reflex-layer
        image: octollm/reflex-layer:v1.0.0
        ports:
        - containerPort: 8080
          hostPort: 8080
        resources:
          requests:
            cpu: "2000m"
            memory: "512Mi"
          limits:
            cpu: "4000m"
            memory: "1Gi"
        securityContext:
          capabilities:
            add:
            - NET_BIND_SERVICE

StatefulSet for PostgreSQL:

apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: postgresql
  namespace: octollm-production
spec:
  serviceName: postgresql
  replicas: 3
  selector:
    matchLabels:
      app: postgresql
  template:
    metadata:
      labels:
        app: postgresql
    spec:
      containers:
      - name: postgresql
        image: postgres:15-alpine
        ports:
        - containerPort: 5432
          name: postgres
        env:
        - name: POSTGRES_DB
          value: octollm
        - name: POSTGRES_USER
          valueFrom:
            secretKeyRef:
              name: postgresql-credentials
              key: username
        - name: POSTGRES_PASSWORD
          valueFrom:
            secretKeyRef:
              name: postgresql-credentials
              key: password
        - name: PGDATA
          value: /var/lib/postgresql/data/pgdata
        volumeMounts:
        - name: data
          mountPath: /var/lib/postgresql/data
        resources:
          requests:
            cpu: "2000m"
            memory: "4Gi"
          limits:
            cpu: "4000m"
            memory: "8Gi"
        livenessProbe:
          exec:
            command:
            - pg_isready
            - -U
            - postgres
          initialDelaySeconds: 30
          periodSeconds: 10
        readinessProbe:
          exec:
            command:
            - pg_isready
            - -U
            - postgres
          initialDelaySeconds: 10
          periodSeconds: 5
  volumeClaimTemplates:
  - metadata:
      name: data
    spec:
      accessModes: ["ReadWriteOnce"]
      storageClassName: fast-ssd
      resources:
        requests:
          storage: 100Gi

2. Development Deployment (Docker Compose)

Platform: Docker Compose 2.x Environment: Local development machines Purpose: Fast iteration, easy debugging

docker-compose.yml:

version: '3.9'

services:
  # Databases
  postgresql:
    image: postgres:15-alpine
    container_name: octollm-postgres
    environment:
      POSTGRES_DB: octollm
      POSTGRES_USER: octollm
      POSTGRES_PASSWORD: development
    ports:
      - "5432:5432"
    volumes:
      - postgres_data:/var/lib/postgresql/data
      - ./scripts/init.sql:/docker-entrypoint-initdb.d/init.sql
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U octollm"]
      interval: 10s
      timeout: 5s
      retries: 5

  redis:
    image: redis:7-alpine
    container_name: octollm-redis
    ports:
      - "6379:6379"
    command: redis-server --appendonly yes
    volumes:
      - redis_data:/data
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 10s
      timeout: 5s
      retries: 5

  qdrant:
    image: qdrant/qdrant:v1.7.0
    container_name: octollm-qdrant
    ports:
      - "6333:6333"
      - "6334:6334"
    volumes:
      - qdrant_data:/qdrant/storage
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:6333/health"]
      interval: 10s
      timeout: 5s
      retries: 5

  # Reflex Layer
  reflex-layer:
    build:
      context: ./reflex_layer
      dockerfile: Dockerfile.dev
    container_name: octollm-reflex
    ports:
      - "8080:8080"
    environment:
      - RUST_LOG=debug
      - RATE_LIMIT_ENABLED=true
    depends_on:
      redis:
        condition: service_healthy
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8080/health"]
      interval: 10s
      timeout: 5s
      retries: 5

  # Orchestrator
  orchestrator:
    build:
      context: ./orchestrator
      dockerfile: Dockerfile.dev
    container_name: octollm-orchestrator
    ports:
      - "8000:8000"
    environment:
      - ENVIRONMENT=development
      - LOG_LEVEL=DEBUG
      - DATABASE_URL=postgresql://octollm:development@postgresql:5432/octollm
      - REDIS_URL=redis://redis:6379
      - QDRANT_URL=http://qdrant:6333
    volumes:
      - ./orchestrator:/app
      - /app/.venv  # Don't override venv
    depends_on:
      postgresql:
        condition: service_healthy
      redis:
        condition: service_healthy
      qdrant:
        condition: service_healthy
      reflex-layer:
        condition: service_healthy
    command: uvicorn main:app --host 0.0.0.0 --port 8000 --reload
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
      interval: 10s
      timeout: 5s
      retries: 5

  # Arms
  coder-arm:
    build:
      context: ./arms/coder
      dockerfile: Dockerfile.dev
    container_name: octollm-coder-arm
    ports:
      - "8102:8102"
    environment:
      - ARM_TYPE=coder
      - LOG_LEVEL=DEBUG
      - OPENAI_API_KEY=${OPENAI_API_KEY}
    volumes:
      - ./arms/coder:/app
      - /app/.venv
    depends_on:
      orchestrator:
        condition: service_healthy
    command: uvicorn main:app --host 0.0.0.0 --port 8102 --reload

  judge-arm:
    build:
      context: ./arms/judge
      dockerfile: Dockerfile.dev
    container_name: octollm-judge-arm
    ports:
      - "8103:8103"
    environment:
      - ARM_TYPE=judge
      - LOG_LEVEL=DEBUG
      - OPENAI_API_KEY=${OPENAI_API_KEY}
    volumes:
      - ./arms/judge:/app
      - /app/.venv
    depends_on:
      orchestrator:
        condition: service_healthy
    command: uvicorn main:app --host 0.0.0.0 --port 8103 --reload

  executor-arm:
    build:
      context: ./arms/executor
      dockerfile: Dockerfile.dev
    container_name: octollm-executor-arm
    ports:
      - "8104:8104"
    environment:
      - ARM_TYPE=executor
      - LOG_LEVEL=DEBUG
    volumes:
      - ./arms/executor:/app
      - /app/.venv
    depends_on:
      orchestrator:
        condition: service_healthy
    command: uvicorn main:app --host 0.0.0.0 --port 8104 --reload

  planner-arm:
    build:
      context: ./arms/planner
      dockerfile: Dockerfile.dev
    container_name: octollm-planner-arm
    ports:
      - "8105:8105"
    environment:
      - ARM_TYPE=planner
      - LOG_LEVEL=DEBUG
      - OPENAI_API_KEY=${OPENAI_API_KEY}
    volumes:
      - ./arms/planner:/app
      - /app/.venv
    depends_on:
      orchestrator:
        condition: service_healthy
    command: uvicorn main:app --host 0.0.0.0 --port 8105 --reload

  retriever-arm:
    build:
      context: ./arms/retriever
      dockerfile: Dockerfile.dev
    container_name: octollm-retriever-arm
    ports:
      - "8106:8106"
    environment:
      - ARM_TYPE=retriever
      - LOG_LEVEL=DEBUG
      - QDRANT_URL=http://qdrant:6333
    volumes:
      - ./arms/retriever:/app
      - /app/.venv
    depends_on:
      orchestrator:
        condition: service_healthy
    command: uvicorn main:app --host 0.0.0.0 --port 8106 --reload

  # Monitoring
  prometheus:
    image: prom/prometheus:latest
    container_name: octollm-prometheus
    ports:
      - "9090:9090"
    volumes:
      - ./monitoring/prometheus.yml:/etc/prometheus/prometheus.yml
      - prometheus_data:/prometheus
    command:
      - '--config.file=/etc/prometheus/prometheus.yml'
      - '--storage.tsdb.path=/prometheus'

  grafana:
    image: grafana/grafana:latest
    container_name: octollm-grafana
    ports:
      - "3000:3000"
    environment:
      - GF_SECURITY_ADMIN_PASSWORD=admin
      - GF_USERS_ALLOW_SIGN_UP=false
    volumes:
      - grafana_data:/var/lib/grafana
      - ./monitoring/grafana/dashboards:/etc/grafana/provisioning/dashboards
      - ./monitoring/grafana/datasources:/etc/grafana/provisioning/datasources
    depends_on:
      - prometheus

volumes:
  postgres_data:
  redis_data:
  qdrant_data:
  prometheus_data:
  grafana_data:

Development Scripts:

scripts/dev.sh:

#!/bin/bash
set -e

# Start development environment
echo "Starting OctoLLM development environment..."

# Check for .env file
if [ ! -f .env ]; then
    echo "Creating .env from template..."
    cp .env.example .env
    echo "⚠️  Please edit .env and add your API keys!"
    exit 1
fi

# Start services
docker compose up -d postgresql redis qdrant

# Wait for databases
echo "Waiting for databases to be ready..."
sleep 5

# Run migrations
echo "Running database migrations..."
docker compose run --rm orchestrator alembic upgrade head

# Start all services
echo "Starting all services..."
docker compose up -d

# Show logs
echo "Services started! Tailing logs (Ctrl+C to stop)..."
docker compose logs -f

scripts/test.sh:

#!/bin/bash
set -e

# Run tests in development environment
echo "Running OctoLLM tests..."

# Start dependencies
docker compose up -d postgresql redis qdrant

# Wait for databases
sleep 5

# Run Python tests
echo "Running orchestrator tests..."
docker compose run --rm orchestrator pytest -v

echo "Running arm tests..."
docker compose run --rm coder-arm pytest -v
docker compose run --rm judge-arm pytest -v

# Run Rust tests
echo "Running reflex layer tests..."
cd reflex_layer && cargo test && cd ..

echo "All tests passed! ✅"

3. Configuration Management

Kubernetes ConfigMaps:

apiVersion: v1
kind: ConfigMap
metadata:
  name: orchestrator-config
  namespace: octollm-production
data:
  ENVIRONMENT: "production"
  LOG_LEVEL: "INFO"
  LOG_FORMAT: "json"
  ARM_REGISTRY_URL: "http://orchestrator:8000/registry"
  RATE_LIMIT_ENABLED: "true"
  RATE_LIMIT_REQUESTS: "1000"
  RATE_LIMIT_WINDOW: "60"

Kubernetes Secrets:

apiVersion: v1
kind: Secret
metadata:
  name: database-credentials
  namespace: octollm-production
type: Opaque
stringData:
  url: postgresql://octollm:PASSWORD@postgresql:5432/octollm
  username: octollm
  password: SECURE_PASSWORD_HERE
---
apiVersion: v1
kind: Secret
metadata:
  name: llm-credentials
  namespace: octollm-production
type: Opaque
stringData:
  api-key: sk-YOUR-API-KEY-HERE

Environment-Specific Configs (Kustomize):

base/kustomization.yaml:

apiVersion: kustomize.config.k8s.io/v1beta1
kind: Kustomization

resources:
  - deployment.yaml
  - service.yaml
  - hpa.yaml
  - configmap.yaml

commonLabels:
  app: octollm
  managed-by: kustomize

overlays/production/kustomization.yaml:

apiVersion: kustomize.config.k8s.io/v1beta1
kind: Kustomization

bases:
  - ../../base

namespace: octollm-production

replicas:
  - name: orchestrator
    count: 3
  - name: coder-arm
    count: 5

images:
  - name: octollm/orchestrator
    newTag: v1.0.0
  - name: octollm/coder-arm
    newTag: v1.0.0

patches:
  - path: production-resources.yaml

overlays/staging/kustomization.yaml:

apiVersion: kustomize.config.k8s.io/v1beta1
kind: Kustomization

bases:
  - ../../base

namespace: octollm-staging

replicas:
  - name: orchestrator
    count: 1
  - name: coder-arm
    count: 2

images:
  - name: octollm/orchestrator
    newTag: latest
  - name: octollm/coder-arm
    newTag: latest

4. Helm Charts (Alternative to Kustomize)

Chart.yaml:

apiVersion: v2
name: octollm
description: OctoLLM Multi-Agent System
type: application
version: 1.0.0
appVersion: "1.0.0"
keywords:
  - llm
  - multi-agent
  - orchestration
maintainers:
  - name: OctoLLM Team
    email: team@octollm.io

values.yaml:

global:
  environment: production
  logLevel: INFO
  imageRegistry: docker.io
  imagePullSecrets: []

orchestrator:
  replicaCount: 3
  image:
    repository: octollm/orchestrator
    tag: v1.0.0
    pullPolicy: IfNotPresent
  resources:
    requests:
      cpu: 500m
      memory: 512Mi
    limits:
      cpu: 2000m
      memory: 2Gi
  autoscaling:
    enabled: true
    minReplicas: 3
    maxReplicas: 10
    targetCPUUtilizationPercentage: 70
  service:
    type: ClusterIP
    port: 8000

arms:
  coder:
    replicaCount: 5
    image:
      repository: octollm/coder-arm
      tag: v1.0.0
    resources:
      requests:
        cpu: 1000m
        memory: 1Gi
      limits:
        cpu: 4000m
        memory: 4Gi
    autoscaling:
      enabled: true
      minReplicas: 5
      maxReplicas: 20
      targetCPUUtilizationPercentage: 70

  judge:
    replicaCount: 3
    image:
      repository: octollm/judge-arm
      tag: v1.0.0
    resources:
      requests:
        cpu: 500m
        memory: 512Mi
      limits:
        cpu: 2000m
        memory: 2Gi

postgresql:
  enabled: true
  auth:
    database: octollm
    username: octollm
  primary:
    persistence:
      enabled: true
      size: 100Gi
      storageClass: fast-ssd
  resources:
    requests:
      cpu: 2000m
      memory: 4Gi
    limits:
      cpu: 4000m
      memory: 8Gi

redis:
  enabled: true
  architecture: replication
  master:
    persistence:
      enabled: true
      size: 10Gi
  replica:
    replicaCount: 2

qdrant:
  enabled: true
  replicaCount: 3
  persistence:
    enabled: true
    size: 50Gi

values-staging.yaml:

global:
  environment: staging
  logLevel: DEBUG

orchestrator:
  replicaCount: 1
  autoscaling:
    enabled: false

arms:
  coder:
    replicaCount: 2
    autoscaling:
      enabled: false

Installation Commands:

# Install production
helm install octollm ./charts/octollm \
  --namespace octollm-production \
  --create-namespace \
  --values ./charts/octollm/values.yaml

# Install staging
helm install octollm-staging ./charts/octollm \
  --namespace octollm-staging \
  --create-namespace \
  --values ./charts/octollm/values-staging.yaml

# Upgrade
helm upgrade octollm ./charts/octollm \
  --namespace octollm-production \
  --values ./charts/octollm/values.yaml

# Rollback
helm rollback octollm 1 --namespace octollm-production

5. CI/CD Pipeline

GitHub Actions - Build and Test:

.github/workflows/ci.yml:

name: CI

on:
  push:
    branches: [main, develop]
  pull_request:
    branches: [main, develop]

jobs:
  test-python:
    runs-on: ubuntu-latest
    strategy:
      matrix:
        python-version: ["3.11", "3.12"]
    steps:
      - uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: ${{ matrix.python-version }}

      - name: Install dependencies
        run: |
          pip install poetry
          cd orchestrator && poetry install

      - name: Run tests
        run: |
          cd orchestrator && poetry run pytest -v --cov=.

      - name: Upload coverage
        uses: codecov/codecov-action@v3

  test-rust:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Install Rust
        uses: actions-rs/toolchain@v1
        with:
          toolchain: stable
          components: rustfmt, clippy

      - name: Run tests
        run: |
          cd reflex_layer
          cargo fmt -- --check
          cargo clippy -- -D warnings
          cargo test

  build-images:
    runs-on: ubuntu-latest
    needs: [test-python, test-rust]
    if: github.event_name == 'push'
    steps:
      - uses: actions/checkout@v4

      - name: Set up Docker Buildx
        uses: docker/setup-buildx-action@v3

      - name: Login to Docker Hub
        uses: docker/login-action@v3
        with:
          username: ${{ secrets.DOCKER_USERNAME }}
          password: ${{ secrets.DOCKER_PASSWORD }}

      - name: Build and push orchestrator
        uses: docker/build-push-action@v5
        with:
          context: ./orchestrator
          push: true
          tags: |
            octollm/orchestrator:latest
            octollm/orchestrator:${{ github.sha }}
          cache-from: type=gha
          cache-to: type=gha,mode=max

      - name: Build and push reflex-layer
        uses: docker/build-push-action@v5
        with:
          context: ./reflex_layer
          push: true
          tags: |
            octollm/reflex-layer:latest
            octollm/reflex-layer:${{ github.sha }}

GitHub Actions - Deploy:

.github/workflows/deploy.yml:

name: Deploy

on:
  push:
    tags:
      - 'v*'

jobs:
  deploy-staging:
    runs-on: ubuntu-latest
    environment: staging
    steps:
      - uses: actions/checkout@v4

      - name: Configure kubectl
        uses: azure/k8s-set-context@v3
        with:
          method: kubeconfig
          kubeconfig: ${{ secrets.KUBE_CONFIG_STAGING }}

      - name: Deploy to staging
        run: |
          kubectl apply -k overlays/staging
          kubectl rollout status deployment/orchestrator -n octollm-staging

      - name: Run smoke tests
        run: |
          ./scripts/smoke-tests.sh staging

  deploy-production:
    runs-on: ubuntu-latest
    needs: deploy-staging
    environment: production
    steps:
      - uses: actions/checkout@v4

      - name: Configure kubectl
        uses: azure/k8s-set-context@v3
        with:
          method: kubeconfig
          kubeconfig: ${{ secrets.KUBE_CONFIG_PRODUCTION }}

      - name: Deploy to production
        run: |
          kubectl apply -k overlays/production
          kubectl rollout status deployment/orchestrator -n octollm-production

      - name: Run smoke tests
        run: |
          ./scripts/smoke-tests.sh production

      - name: Notify Slack
        uses: 8398a7/action-slack@v3
        with:
          status: ${{ job.status }}
          text: 'Deployed ${{ github.ref }} to production'
        env:
          SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK }}

6. Ingress and Load Balancing

Nginx Ingress Controller:

apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
  name: octollm-ingress
  namespace: octollm-production
  annotations:
    nginx.ingress.kubernetes.io/rewrite-target: /
    nginx.ingress.kubernetes.io/ssl-redirect: "true"
    cert-manager.io/cluster-issuer: "letsencrypt-prod"
    nginx.ingress.kubernetes.io/rate-limit: "100"
spec:
  ingressClassName: nginx
  tls:
  - hosts:
    - api.octollm.io
    secretName: octollm-tls
  rules:
  - host: api.octollm.io
    http:
      paths:
      - path: /
        pathType: Prefix
        backend:
          service:
            name: reflex-layer
            port:
              number: 8080
      - path: /api/orchestrator
        pathType: Prefix
        backend:
          service:
            name: orchestrator
            port:
              number: 8000

7. Monitoring and Observability

Prometheus ServiceMonitor:

apiVersion: monitoring.coreos.com/v1
kind: ServiceMonitor
metadata:
  name: octollm-metrics
  namespace: octollm-production
spec:
  selector:
    matchLabels:
      app: octollm
  endpoints:
  - port: metrics
    interval: 30s
    path: /metrics

Grafana Dashboard ConfigMap:

apiVersion: v1
kind: ConfigMap
metadata:
  name: grafana-dashboards
  namespace: octollm-system
data:
  octollm-overview.json: |
    {
      "dashboard": {
        "title": "OctoLLM Overview",
        "panels": [
          {
            "title": "Request Rate",
            "targets": [
              {
                "expr": "rate(http_requests_total[5m])"
              }
            ]
          },
          {
            "title": "Error Rate",
            "targets": [
              {
                "expr": "rate(http_requests_total{status=~\"5..\"}[5m])"
              }
            ]
          }
        ]
      }
    }

8. Disaster Recovery

Backup Strategy:

# Velero backup schedule
apiVersion: velero.io/v1
kind: Schedule
metadata:
  name: octollm-daily-backup
  namespace: velero
spec:
  schedule: "0 2 * * *"  # 2 AM daily
  template:
    includedNamespaces:
    - octollm-production
    storageLocation: default
    volumeSnapshotLocations:
    - default
    ttl: 720h  # 30 days

Restore Procedure:

# Restore from backup
velero restore create octollm-restore \
  --from-backup octollm-daily-backup-20251110 \
  --namespace-mappings octollm-production:octollm-production-restored

# Verify restore
kubectl get all -n octollm-production-restored

# Promote to production
kubectl label namespace octollm-production-restored environment=production

Consequences

Positive

1. Kubernetes Production Benefits:

   • Auto-scaling handles variable load
   • Self-healing reduces downtime
   • Rolling updates enable zero-downtime deployments
   • Resource quotas prevent runaway costs
   • Industry-standard platform

2. Docker Compose Development Benefits:

   • Fast startup (<2 minutes)
   • Easy debugging with volume mounts
   • Minimal resource usage
   • Production parity with same images
   • Simple onboarding for new developers

3. Cloud Agnostic:

   • No vendor lock-in
   • Can deploy to any K8s cluster
   • Easy migration between clouds
   • Cost optimization through competition
   • Multi-cloud strategy possible

4. Operational Efficiency:

   • Automated deployments via CI/CD
   • Consistent environments (dev/staging/prod)
   • Infrastructure as code
   • Easy rollbacks
   • Comprehensive monitoring

5. Scalability:

   • Horizontal scaling for stateless services
   • Vertical scaling for databases
   • HPA automatically adjusts replicas
   • Can handle 10x traffic spikes
   • Resource-efficient

Negative

1. Kubernetes Complexity:

   • Steep learning curve
   • Many concepts to understand
   • Complex YAML configurations
   • Debugging can be challenging
   • Requires specialized expertise

2. Operational Overhead:

   • Need to manage K8s cluster
   • Monitoring infrastructure required
   • More moving parts
   • Complex troubleshooting
   • Higher ops burden

3. Resource Requirements:

   • K8s control plane overhead
   • Need multiple worker nodes
   • Development setup is heavyweight
   • More expensive infrastructure
   • Minimum cluster size costs

4. Development-Production Gap:

   • Docker Compose != Kubernetes
   • Some issues only appear in K8s
   • Different networking models
   • Debugging differs between environments
   • Need staging environment

Mitigation Strategies

1. Complexity:

   • Comprehensive documentation
   • Helm charts for easier deployment
   • Training for team members
   • Start with simple deployments
   • Gradually adopt advanced features

2. Operational Overhead:

   • Managed Kubernetes (EKS/GKE/AKS)
   • Automated monitoring setup
   • Runbooks for common issues
   • On-call rotation
   • Regular operational reviews

3. Resource Requirements:

   • Right-size cluster for workload
   • Use spot instances where possible
   • Optimize resource requests/limits
   • Auto-scaling to minimize waste
   • Cost monitoring and alerts

4. Dev-Prod Gap:

   • Maintain staging environment
   • Test in K8s before production
   • Document K8s-specific behaviors
   • Use same images everywhere
   • Comprehensive integration tests

Alternatives Considered

1. Docker Swarm

Pros:

• Simpler than Kubernetes
• Built into Docker
• Easier to learn
• Less resource overhead

Cons:

• Less ecosystem support
• Fewer features than K8s
• Not as widely adopted
• Limited scaling capabilities
• Weaker community

Why Rejected: Kubernetes has better ecosystem, more features, and industry adoption.

2. HashiCorp Nomad

Pros:

• Simpler than Kubernetes
• Multi-workload (containers, VMs, binaries)
• Good for hybrid deployments
• Easier operations

Cons:

• Smaller ecosystem
• Less tooling available
• Fewer managed options
• Weaker community
• Less familiar to team

Why Rejected: Kubernetes has better ecosystem and more deployment options.

3. Serverless (Lambda/Cloud Functions)

Pros:

• No infrastructure management
• Pay per use
• Auto-scaling built-in
• Simple deployment

Cons:

• Cold start latency
• Vendor lock-in
• Limited runtime duration
• Harder to debug
• Cost unpredictable at scale

Why Rejected: Need consistent latency and want cloud-agnostic approach.

4. Single VM Deployment

Pros:

• Simplest setup
• Easy to understand
• Low cost
• Easy debugging

Cons:

• No auto-scaling
• Single point of failure
• Manual updates
• Limited capacity
• No high availability

Why Rejected: Doesn't meet production requirements for scaling and availability.

5. Cloud-Specific Services (ECS/Cloud Run)

Pros:

• Simpler than K8s
• Managed by provider
• Good integration with cloud
• Lower learning curve

Cons:

• Vendor lock-in
• Migration difficult
• Cloud-specific knowledge
• Limited portability

Why Rejected: Want cloud-agnostic solution to avoid vendor lock-in.

Implementation Guidelines

Development Workflow

# Clone repository
git clone https://github.com/your-org/octollm.git
cd octollm

# Set up environment
cp .env.example .env
# Edit .env with your API keys

# Start development environment
./scripts/dev.sh

# Run tests
./scripts/test.sh

# View logs
docker compose logs -f orchestrator

# Restart specific service
docker compose restart coder-arm

# Stop environment
docker compose down

Production Deployment

# Build and push images
docker build -t octollm/orchestrator:v1.0.0 ./orchestrator
docker push octollm/orchestrator:v1.0.0

# Deploy to staging
kubectl apply -k overlays/staging
kubectl rollout status deployment/orchestrator -n octollm-staging

# Run smoke tests
./scripts/smoke-tests.sh staging

# Deploy to production
kubectl apply -k overlays/production
kubectl rollout status deployment/orchestrator -n octollm-production

# Monitor rollout
kubectl get pods -n octollm-production -w
kubectl logs -f deployment/orchestrator -n octollm-production

# Rollback if needed
kubectl rollout undo deployment/orchestrator -n octollm-production

Troubleshooting

# Check pod status
kubectl get pods -n octollm-production

# View pod logs
kubectl logs -f <pod-name> -n octollm-production

# Describe pod (events, resources)
kubectl describe pod <pod-name> -n octollm-production

# Execute command in pod
kubectl exec -it <pod-name> -n octollm-production -- /bin/sh

# Check resource usage
kubectl top pods -n octollm-production

# View events
kubectl get events -n octollm-production --sort-by='.lastTimestamp'

References

• Kubernetes Documentation
• Docker Compose Documentation
• Helm Documentation
• Kustomize Documentation
• The Twelve-Factor App
• Kubernetes Patterns

Last Review: 2025-11-10 Next Review: 2026-05-10 (6 months) Related ADRs: ADR-001, ADR-002, ADR-003, ADR-004

ADR-006: Cloud Provider Selection

Status: Accepted Date: 2025-11-12 Decision Makers: Architecture Team, DevOps Team, Finance Team Consulted: Engineering Team, Security Team, Operations Team

Context

OctoLLM requires a cloud infrastructure provider to host production, staging, and development environments. As established in ADR-005 (Deployment Platform), we have decided to use Kubernetes for production with a cloud-agnostic architecture. This ADR focuses on selecting the specific cloud provider for managed services while maintaining portability.

Infrastructure Requirements

Core Services Needed:

1. Kubernetes Service: Managed Kubernetes cluster (1.28+)
2. Managed PostgreSQL: PostgreSQL 15+ with HA, read replicas, automated backups
3. Managed Redis: Redis 7+ with cluster mode, persistence, automatic failover
4. Object Storage: S3-compatible storage for backups, logs, artifacts
5. Secrets Management: Secure storage for API keys, certificates, passwords
6. Load Balancing: Layer 7 load balancers with TLS termination
7. DNS Management: Managed DNS with health checks
8. Monitoring & Logging: Metrics, logs, distributed tracing capabilities

Deployment Environments:

• Development: Minimal resources, cost-optimized, single-region
• Staging: Production-like, scaled down 50%, multi-AZ
• Production: Full HA, multi-AZ, auto-scaling, 99.95% SLA

Resource Specifications (from MASTER-TODO.md Sprint 0.7):

Key Decision Criteria:

1. Cost: Total cost of ownership (TCO) across all environments
2. Kubernetes Maturity: Feature set, stability, ecosystem integration
3. Database Performance: PostgreSQL and Redis managed service quality
4. Developer Experience: Ease of setup, documentation, tooling
5. Security & Compliance: SOC 2, ISO 27001, GDPR capabilities
6. Geographic Coverage: Low-latency access for target users
7. Free Tier: Development and experimentation capabilities
8. Migration Path: Ease of multi-cloud or exit strategy
9. Monitoring & Observability: Native tools for metrics, logs, traces
10. Community & Support: Documentation quality, community size, support options

Evaluation Constraints

• Budget: Target $500/month for dev + staging, $3,000/month for production
• Timeline: Infrastructure must be provisionable within 1 week
• Skills: Team has moderate cloud experience, strong Kubernetes knowledge
• Compliance: Must support future SOC 2 Type II certification
• Portability: Infrastructure must be cloud-agnostic (use standard APIs)

Research & Analysis

1. Amazon Web Services (AWS)

Kubernetes Service: Amazon Elastic Kubernetes Service (EKS) Managed PostgreSQL: Amazon RDS for PostgreSQL Managed Redis: Amazon ElastiCache for Redis Object Storage: Amazon S3 Secrets Management: AWS Secrets Manager

Strengths

Kubernetes (EKS):

• Mature service (GA since 2018)
• Excellent control plane HA (99.95% SLA)
• Native integration with AWS services (IAM, CloudWatch, ELB)
• Fargate support for serverless node pools
• Managed node groups with auto-scaling
• EKS Anywhere for hybrid/on-prem (portability)
• Extensive ecosystem (add-ons, operators)

Database (RDS PostgreSQL):

• PostgreSQL 15+ support
• Automated backups (35-day retention max)
• Multi-AZ deployments with automatic failover (<2 min)
• Read replicas (up to 15) with cross-region support
• Performance Insights for query optimization
• Aurora PostgreSQL option (5x performance, higher cost)
• Proxy support (RDS Proxy) for connection pooling

Redis (ElastiCache):

• Redis 7.0+ support
• Cluster mode with auto-sharding (up to 500 nodes)
• Multi-AZ with automatic failover
• Daily backups with point-in-time recovery
• Encryption at rest and in transit
• Global Datastore for multi-region replication

Storage (S3):

• Industry-leading 99.999999999% durability (11 nines)
• Lifecycle policies for cost optimization
• Versioning, replication, encryption
• Glacier for long-term archival (lowest cost)
• S3 Express One Zone for ultra-low latency

Secrets (Secrets Manager):

• Automatic rotation for RDS, Redshift, DocumentDB
• Fine-grained IAM policies
• Encryption with KMS
• Cross-region replication
• Versioning and rollback

Monitoring:

• CloudWatch for metrics (1-minute resolution, 15-month retention)
• CloudWatch Logs for centralized logging
• X-Ray for distributed tracing
• Container Insights for EKS-specific metrics

Developer Experience:

• AWS CLI (mature, feature-complete)
• eksctl for simplified EKS operations
• AWS CDK for infrastructure as code (TypeScript/Python)
• Extensive Terraform modules (community-maintained)
• Copilot CLI for containerized apps
• Comprehensive documentation (best-in-class)

Geographic Coverage:

• 32 regions, 102 availability zones (as of 2024)
• Excellent global coverage (US, EU, Asia-Pacific, Middle East, South America)
• Low-latency access for most OctoLLM users (US-based initially)

Free Tier:

• 750 hours/month EC2 t2.micro (12 months)
• 20GB RDS PostgreSQL (12 months)
• 5GB S3 storage (always free)
• 1 million Lambda requests/month (always free)
• No free tier for EKS ($0.10/hour = $73/month per cluster)

Compliance:

• SOC 2 Type II certified
• ISO 27001, 27017, 27018
• GDPR, HIPAA, PCI DSS compliant
• 143 compliance certifications (most comprehensive)

Weaknesses

Cost:

• EKS control plane: $0.10/hour ($73/month per cluster)
• More expensive than GCP/Azure for compute (10-15% higher)
• Data transfer costs can be significant (egress: $0.09/GB)
• RDS pricing higher than CloudSQL/Azure Database

Complexity:

• Steeper learning curve (vast service catalog)
• IAM complexity (policies, roles, users, groups)
• Networking setup more involved (VPC, subnets, route tables, NAT)

Vendor Lock-in Risk:

• Easy to use AWS-specific services (DynamoDB, Lambda)
• Proprietary APIs (CloudWatch, X-Ray)
• Aurora PostgreSQL not portable

Cost Estimate (per month)

Development Environment:

• EKS cluster: $73 (control plane)
• EC2 nodes: 3 × t3.large (2vCPU, 8GB): $150
• RDS PostgreSQL: db.t3.micro (1vCPU, 2GB): $30
• ElastiCache Redis: cache.t3.micro (2GB): $35
• S3: 50GB + requests: $5
• Data transfer: $10
• Total: ~$303/month

Staging Environment:

• EKS cluster: $73
• EC2 nodes: 4 × t3.xlarge (4vCPU, 16GB): $400
• RDS PostgreSQL: db.t3.medium (2vCPU, 8GB): $120
• ElastiCache Redis: cache.r6g.large (3GB cluster): $150
• S3: 200GB + requests: $15
• Data transfer: $30
• Total: ~$788/month

Production Environment:

• EKS cluster: $73
• EC2 nodes: 5-10 × m6i.2xlarge (8vCPU, 32GB): $2,400 (avg 7.5 nodes)
• RDS PostgreSQL: db.r6g.xlarge (4vCPU, 16GB) + 2 read replicas: $900
• ElastiCache Redis: cache.r6g.xlarge (6GB) × 6 (cluster): $900
• S3: 1TB + requests: $50
• Load Balancer (ALB): $30
• NAT Gateway: $90
• Data transfer: $200
• Total: ~$4,643/month

Total All Environments: ~$5,734/month

2. Google Cloud Platform (GCP)

Kubernetes Service: Google Kubernetes Engine (GKE) Managed PostgreSQL: Cloud SQL for PostgreSQL Managed Redis: Memorystore for Redis Object Storage: Google Cloud Storage (GCS) Secrets Management: Secret Manager

Strengths

Kubernetes (GKE):

• Best-in-class Kubernetes (Google created Kubernetes)
• Autopilot mode: fully managed, serverless, pay-per-pod
• Standard mode: flexible, full control
• Automatic node repairs and upgrades
• Built-in container security (Binary Authorization, GKE Sandbox)
• Multi-cluster Ingress (traffic routing across clusters)
• Workload Identity (native Kubernetes service account integration)
• Free control plane for Standard mode (below 3 zones)
• GKE Enterprise (formerly Anthos) for multi-cloud/hybrid

Database (Cloud SQL PostgreSQL):

• PostgreSQL 15+ support
• High availability with automatic failover (<60 seconds)
• Up to 10 read replicas
• Automated backups (365-day retention max)
• Point-in-time recovery (7 days)
• Connection pooling built-in (PgBouncer)
• Query Insights for performance analysis
• 15-25% cheaper than RDS (similar specs)

Redis (Memorystore):

• Redis 7.0+ support
• High availability with automatic failover
• Extremely low latency (<1ms within region)
• Read replicas for read-heavy workloads
• Import/export capabilities
• No cluster mode (scaling limited to 300GB per instance)

Storage (GCS):

• 99.999999999% durability (same as S3)
• Multi-region and dual-region options
• Lifecycle management
• Object versioning
• Nearline/Coldline/Archive for cost optimization
• Signed URLs for temporary access

Secrets (Secret Manager):

• Automatic versioning
• IAM integration
• Encryption with Cloud KMS
• Audit logging with Cloud Audit Logs
• Simpler than AWS Secrets Manager (less feature-rich but easier)

Monitoring:

• Cloud Monitoring (formerly Stackdriver)
• Cloud Logging (centralized logs, 30-day default retention)
• Cloud Trace (distributed tracing)
• GKE observability built-in (metrics, logs, traces)
• Better integration than AWS (single pane of glass)

Developer Experience:

• gcloud CLI (well-designed, intuitive)
• GKE-specific commands (gcloud container)
• Google Cloud Console (modern UI, fastest)
• Terraform support (official provider, well-maintained)
• Excellent documentation (clear, concise)
• Cloud Shell (browser-based development environment)

Geographic Coverage:

• 40 regions, 121 zones (as of 2024)
• Best regional expansion (new regions frequently)
• Strong Asia-Pacific presence
• Multi-region resources (Cloud SQL, GCS)

Free Tier:

• GKE Standard: FREE control plane (autopilot mode free for <18 hours/month)
• $300 free credit for 90 days (new accounts)
• Always free: 1 non-preemptible e2-micro VM
• Always free: 5GB Cloud Storage (regional)
• Best free tier for Kubernetes experimentation

Compliance:

• SOC 2 Type II certified
• ISO 27001, 27017, 27018
• GDPR, HIPAA, PCI DSS compliant
• 80+ compliance certifications

Weaknesses

Kubernetes:

• Autopilot mode limitations (less control, some add-ons unsupported)
• Fewer managed add-ons than EKS (no Fargate equivalent)

Redis:

• No cluster mode (major limitation for high-scale workloads)
• Maximum 300GB per instance (ElastiCache supports terabytes)
• Fewer sharding options

Ecosystem:

• Smaller community than AWS (fewer third-party integrations)
• Less enterprise adoption (compared to AWS/Azure)

Support:

• Support plans more expensive than AWS (for similar tiers)
• Fewer certified partners for consulting/implementation

Vendor Lock-in Risk:

• BigQuery, Pub/Sub, Cloud Functions (proprietary)
• GKE Autopilot tight coupling

Cost Estimate (per month)

Development Environment:

• GKE cluster: $0 (free control plane for <3 zones)
• Compute Engine: 3 × e2-standard-2 (2vCPU, 8GB): $120
• Cloud SQL PostgreSQL: db-f1-micro (1vCPU, 3.75GB): $25
• Memorystore Redis: Basic tier (2GB): $40
• Cloud Storage: 50GB: $2
• Data transfer: $5
• Total: ~$192/month (36% cheaper than AWS)

Staging Environment:

• GKE cluster: $0
• Compute Engine: 4 × e2-standard-4 (4vCPU, 16GB): $340
• Cloud SQL PostgreSQL: db-n1-standard-2 (2vCPU, 7.5GB): $100
• Memorystore Redis: Standard tier (3GB): $120
• Cloud Storage: 200GB: $8
• Data transfer: $20
• Total: ~$588/month (25% cheaper than AWS)

Production Environment:

• GKE cluster: $73 (3+ zones = paid)
• Compute Engine: 5-10 × n2-standard-8 (8vCPU, 32GB): $2,000 (avg 7.5 nodes)
• Cloud SQL PostgreSQL: db-n1-standard-4 (4vCPU, 15GB) + 2 replicas: $700
• Memorystore Redis: Standard tier (6GB) × 3 (manual sharding): $650
• Cloud Storage: 1TB: $40
• Load Balancer: $25
• Cloud NAT: $45
• Data transfer: $150
• Total: ~$3,683/month (21% cheaper than AWS)

Total All Environments: ~$4,463/month (22% cheaper than AWS)

3. Microsoft Azure

Kubernetes Service: Azure Kubernetes Service (AKS) Managed PostgreSQL: Azure Database for PostgreSQL Flexible Server Managed Redis: Azure Cache for Redis Object Storage: Azure Blob Storage Secrets Management: Azure Key Vault

Strengths

Kubernetes (AKS):

• Free control plane (no hourly charge)
• Azure CNI for native VNet integration
• Azure AD integration for RBAC
• Virtual nodes (ACI for serverless pods)
• Dev Spaces for collaborative development
• Azure Policy for governance
• Excellent Windows container support
• Azure Arc for multi-cloud Kubernetes management

Database (Azure Database for PostgreSQL):

• PostgreSQL 15+ support (Flexible Server)
• High availability with zone-redundant deployment
• Up to 5 read replicas
• Automated backups (35-day retention)
• Point-in-time recovery
• Burstable SKUs (B-series) for cost-effective dev/test
• Hyperscale (Citus) option for distributed PostgreSQL

Redis (Azure Cache for Redis):

• Redis 6.0+ support (7.0 in preview)
• Enterprise tier with Redis Enterprise features
• Clustering support (Premium/Enterprise tiers)
• Active geo-replication (Enterprise)
• Zone redundancy for HA
• Best Redis integration (first-party Redis Enterprise)

Storage (Blob Storage):

• 99.999999999% durability (LRS)
• Hot, Cool, Archive tiers
• Immutable storage for compliance
• Soft delete and versioning
• Azure Data Lake Storage Gen2 (big data analytics)

Secrets (Key Vault):

• Secrets, keys, certificates in single service
• HSM-backed keys (Premium tier)
• Managed identity integration
• RBAC and access policies
• Automatic rotation (Azure SQL, Storage Accounts)

Monitoring:

• Azure Monitor (unified platform)
• Log Analytics (Kusto Query Language)
• Application Insights (APM for apps)
• Container Insights (AKS-specific)
• Azure Monitor for Prometheus (managed Prometheus)

Developer Experience:

• Azure CLI (powerful, consistent)
• Azure Portal (feature-rich, can be overwhelming)
• Bicep for IaC (DSL, simpler than ARM templates)
• Terraform support (official provider)
• Best Windows/hybrid integration
• GitHub Actions integration (Microsoft-owned)

Geographic Coverage:

• 60+ regions (most of any cloud provider)
• Strong presence in Europe, Asia, US
• Government clouds (Azure Government)
• Azure Stack for on-premises

Free Tier:

• $200 Azure credit for 30 days (new accounts)
• 12 months free: 750 hours B1S VM, 5GB Blob Storage
• AKS: FREE control plane
• Always free: 10 App Services, 1GB Storage

Compliance:

• SOC 2 Type II certified
• ISO 27001, 27017, 27018
• GDPR, HIPAA, PCI DSS compliant
• 100+ compliance certifications
• Best for government/regulated industries

Weaknesses

Kubernetes:

• AKS upgrade process can be disruptive
• Less mature than GKE (created by Google)
• Networking complexity (Azure CNI vs kubenet)

Database:

• PostgreSQL 15 released later than AWS/GCP
• Fewer PostgreSQL extensions than RDS
• Connection limits lower than RDS (for same SKU)

Redis:

• Redis 7.0 still in preview (as of Nov 2024)
• Enterprise tier very expensive (3-5x Premium tier)
• Basic tier has no SLA

Ecosystem:

• Smaller Kubernetes community than GKE/EKS
• Fewer Kubernetes-specific tools and integrations

Documentation:

• Quality inconsistent (some areas excellent, others lacking)
• Frequent rebranding causes confusion
• Examples sometimes outdated

Vendor Lock-in Risk:

• Azure Functions, Cosmos DB, Service Bus (proprietary)
• Azure AD tight coupling
• ARM templates complex (Bicep mitigates)

Cost Estimate (per month)

Development Environment:

• AKS cluster: $0 (free control plane)
• Virtual Machines: 3 × Standard_D2s_v3 (2vCPU, 8GB): $130
• Azure Database PostgreSQL: B1ms (1vCPU, 2GB): $20
• Azure Cache Redis: Basic C1 (1GB): $20 (note: 1GB minimum, not 2GB)
• Blob Storage: 50GB (Hot): $3
• Data transfer: $5
• Total: ~$178/month (41% cheaper than AWS, 7% cheaper than GCP)

Staging Environment:

• AKS cluster: $0
• Virtual Machines: 4 × Standard_D4s_v3 (4vCPU, 16GB): $360
• Azure Database PostgreSQL: GP_Standard_D2s_v3 (2vCPU, 8GB): $110
• Azure Cache Redis: Standard C3 (3GB): $100
• Blob Storage: 200GB (Hot): $10
• Data transfer: $20
• Total: ~$600/month (24% cheaper than AWS, 2% more than GCP)

Production Environment:

• AKS cluster: $0
• Virtual Machines: 5-10 × Standard_D8s_v3 (8vCPU, 32GB): $2,100 (avg 7.5 nodes)
• Azure Database PostgreSQL: GP_Standard_D4s_v3 (4vCPU, 16GB) + 2 replicas: $750
• Azure Cache Redis: Premium P3 (6GB) × 3 nodes (cluster): $750
• Blob Storage: 1TB (Hot): $45
• Load Balancer: $20
• NAT Gateway: $40
• Data transfer: $150
• Total: ~$3,855/month (17% cheaper than AWS, 5% more than GCP)

Total All Environments: ~$4,633/month (19% cheaper than AWS, 4% more than GCP)

Detailed Comparison Matrix

Cost Comparison (Monthly)

Annual Cost Savings (vs AWS):

• GCP: $15,252 saved/year (22% reduction)
• Azure: $13,212 saved/year (19% reduction)

Feature Comparison

Developer Experience Comparison

Security & Compliance Comparison

Portability & Lock-in Risk

Support & Community

Decision

We choose Google Cloud Platform (GCP) as our primary cloud provider for the following reasons:

Primary Factors

1. Cost Efficiency (Weight: 30%)

   • 22% cheaper than AWS ($15,252/year savings)
   • 4% cheaper than Azure ($2,040/year savings)
   • Free Kubernetes control plane (saves $876/year vs AWS)
   • Best free tier for development and experimentation

2. Kubernetes Excellence (Weight: 25%)

   • Google created Kubernetes (unmatched expertise)
   • GKE is the most mature, feature-rich Kubernetes service
   • Autopilot mode for simplified operations
   • Workload Identity (best practice for service accounts)
   • Excellent documentation and tooling

3. Developer Experience (Weight: 20%)

   • Fastest setup time (30 min to first cluster)
   • Best CLI (gcloud intuitive, well-designed)
   • Modern, responsive web console
   • Excellent observability (single pane of glass)
   • Cloud Shell for browser-based development

4. Portability (Weight: 15%)

   • Lowest vendor lock-in risk
   • Standard Kubernetes (minimal proprietary features)
   • Multi-cloud strategy with Anthos (if needed)
   • Easy migration path to other providers

5. Performance (Weight: 10%)

   • Best Kubernetes performance (Google's expertise)
   • Memorystore for Redis: <1ms latency
   • Cloud SQL competitive with RDS
   • Excellent network performance (Google's backbone)

Trade-offs Accepted

Limitations vs AWS:

• Smaller ecosystem (fewer third-party integrations)
• Fewer compliance certifications (143 vs 80+)
• Redis cluster mode limited (300GB max per instance)
• Smaller community (200k+ vs 500k+ Stack Overflow questions)

Mitigation Strategies:

• Redis limitation: Use manual sharding (3 instances) for production
• Ecosystem: AWS services available via APIs (e.g., AWS SDK for S3 backups)
• Community: GCP community large enough for OctoLLM needs
• Compliance: 80+ certifications sufficient for current requirements

Why Not AWS:

• 22% more expensive ($15,252/year difference)
• Paid Kubernetes control plane ($876/year)
• Steeper learning curve (complexity overkill for OctoLLM)
• Higher vendor lock-in risk (easy to use proprietary services)

Why Not Azure:

• 4% more expensive than GCP ($2,040/year)
• Kubernetes less mature than GKE
• PostgreSQL 15 support lagged behind competitors
• Smaller Kubernetes ecosystem
• Documentation quality inconsistent

Cloud-Agnostic Architecture (Portability Safeguards)

To maintain portability and avoid lock-in, we will:

1. Use Standard Kubernetes APIs:

   • No GKE-specific CRDs (Custom Resource Definitions)
   • Avoid GKE Autopilot for production (use Standard mode)
   • Use standard Ingress, not GKE-specific LoadBalancer

2. Abstract Cloud Services:

   • PostgreSQL: Standard libpq connection strings
   • Redis: Standard Redis protocol (no GCP-specific features)
   • Object Storage: S3-compatible API (GCS supports this)

3. Infrastructure as Code (Terraform):

   • Use Terraform with provider abstraction
   • Modular design (swap providers by changing modules)
   • No hard-coded GCP resource IDs

4. Monitoring: Use Prometheus/Grafana (not Cloud Monitoring alone)

5. Secrets: ExternalSecrets Operator (supports multiple backends)

6. CI/CD: GitHub Actions (provider-agnostic, not Cloud Build)

Migration Path (if needed)

If we need to migrate to AWS or Azure:

Consequences

Positive

1. Cost Savings: $15,252/year compared to AWS (22% reduction)
2. Best Kubernetes: Leveraging Google's Kubernetes expertise
3. Fast Development: Free control plane + excellent DX = faster iteration
4. Simple Operations: GKE Autopilot option for less operational overhead
5. Strong Observability: Cloud Monitoring/Logging/Trace integrated
6. Low Lock-in: Easy migration to other clouds if needed
7. Scalability: GKE supports large-scale production workloads
8. Security: SOC 2, ISO 27001, 80+ certifications sufficient

Negative

1. Smaller Ecosystem: Fewer third-party tools than AWS (mitigated: sufficient for OctoLLM)
2. Redis Limitations: No cluster mode >300GB (mitigated: manual sharding)
3. Team Learning: Team needs to learn GCP (mitigated: excellent docs, gentle curve)
4. Fewer Certifications: 80+ vs AWS 143 (mitigated: covers all current needs)
5. Community Size: Smaller than AWS (mitigated: still large, active community)

Risks & Mitigation

Implementation Plan

Phase 1: GCP Account Setup (Week 1)

1. Create GCP Organization & Projects:

   • Organization: octollm.com
   • Projects: octollm-dev, octollm-staging, octollm-prod
   • Enable billing account
   • Set up billing alerts: 50% ($250), 80% ($400), 100% ($500) for dev

2. Configure IAM & Security:

   • Create service accounts for Terraform
   • Set up IAM roles (least privilege):
     • Kubernetes Engine Admin (cluster management)
     • Cloud SQL Admin (database management)
     • Storage Admin (GCS management)
     • Secret Manager Admin (secrets)
   • Enable required APIs:
     • Kubernetes Engine API
     • Cloud SQL Admin API
     • Compute Engine API
     • Cloud Storage API
     • Secret Manager API
     • Cloud Monitoring API
   • Configure organization policies:
     • Require OS Login
     • Disable service account key creation
     • Restrict public IP assignment

3. Set Up Billing Alerts & Budgets:

   # Dev Environment
   budget: $500/month
   alerts:
     - 50%: Email team, Slack notification
     - 80%: Email team + managers, Slack alert
     - 100%: Email team + managers + finance, stop dev resources
   
   # Staging Environment
   budget: $1,000/month
   alerts:
     - 50%: Email team
     - 80%: Email team + managers
     - 100%: Email team + managers + finance
   
   # Production Environment
   budget: $5,000/month
   alerts:
     - 50%: Email team
     - 80%: Email team + managers
     - 100%: Email team + managers + finance + executives
   

4. Configure Resource Tagging Strategy:

   • Labels (GCP terminology):
     • environment: dev | staging | prod
     • project: octollm
     • component: orchestrator | reflex | arm-* | database | cache
     • owner: team-backend | team-devops
     • cost-center: engineering | infrastructure
     • managed-by: terraform | manual

Phase 2: Development Environment (Week 1)

1. Provision GKE Cluster (dev-cluster):

   gcloud container clusters create octollm-dev \
     --region us-central1 \
     --num-nodes 1 --min-nodes 1 --max-nodes 3 \
     --node-locations us-central1-a \
     --machine-type e2-standard-2 \
     --disk-size 50 \
     --enable-autoscaling \
     --enable-autorepair \
     --enable-autoupgrade \
     --no-enable-cloud-logging \
     --no-enable-cloud-monitoring \
     --addons HorizontalPodAutoscaling,HttpLoadBalancing
   

2. Provision Cloud SQL PostgreSQL:

   gcloud sql instances create octollm-dev-postgres \
     --database-version POSTGRES_15 \
     --tier db-f1-micro \
     --region us-central1 \
     --storage-size 20GB \
     --storage-type SSD \
     --storage-auto-increase \
     --backup-start-time 03:00 \
     --retained-backups-count 7
   

3. Provision Memorystore Redis:

   gcloud redis instances create octollm-dev-redis \
     --size 2 \
     --region us-central1 \
     --tier basic \
     --redis-version redis_7_0
   

4. Create GCS Buckets:

   gsutil mb -l us-central1 -c STANDARD gs://octollm-dev-backups
   gsutil mb -l us-central1 -c STANDARD gs://octollm-dev-logs
   

Phase 3: Staging & Production (Week 2)

1. Staging: Similar to dev, scaled up (see Sprint 0.7 Task 3)
2. Production: Multi-AZ, HA, autoscaling (see Sprint 0.7 Task 3)

Phase 4: Monitoring & Observability (Week 2)

1. Install Prometheus + Grafana (Helm charts)
2. Configure Cloud Monitoring dashboards
3. Set up alerting policies
4. Configure log retention (Cloud Logging)

Appendix: Detailed Setup Instructions

Prerequisites

Required Tools:

# Install gcloud CLI
curl https://sdk.cloud.google.com | bash
exec -l $SHELL

# Install kubectl
gcloud components install kubectl

# Install Terraform (for IaC)
brew install terraform  # macOS
# or: wget + install from terraform.io

# Install Helm (for Kubernetes packages)
brew install helm  # macOS

Authentication:

# Authenticate with GCP
gcloud auth login

# Set default project
gcloud config set project octollm-dev

# Configure kubectl
gcloud container clusters get-credentials octollm-dev --region us-central1

Cost Optimization Tips

1. Committed Use Discounts:

   • 1-year commitment: 25% discount
   • 3-year commitment: 52% discount
   • Apply to Compute Engine, GKE nodes
   • Savings: $6,000/year on production (25% discount)

2. Preemptible/Spot VMs (dev environment):

   • 60-91% discount vs on-demand
   • Suitable for dev workloads (can tolerate interruptions)
   • Savings: $80/month on dev

3. Sustained Use Discounts (automatic):

   • Up to 30% discount for sustained usage
   • No commitment required
   • Applied automatically

4. Rightsizing Recommendations:

   • Enable recommender API
   • Review monthly (downsize underutilized resources)

5. Storage Lifecycle Policies:

   • Move logs to Nearline after 30 days (50% cheaper)
   • Move logs to Coldline after 90 days (70% cheaper)
   • Delete logs after 1 year

Security Best Practices

1. Enable Binary Authorization (GKE):

   • Require signed container images
   • Prevent untrusted images from running

2. Enable GKE Sandbox (gVisor):

   • Additional container isolation
   • Recommended for executor-arm (untrusted code)

3. Configure Workload Identity:

   • Bind Kubernetes service accounts to GCP service accounts
   • Avoid service account keys (security risk)

4. Enable Private GKE Clusters:

   • No public IP addresses for nodes
   • Access via Cloud VPN or bastion host

5. Enable VPC Service Controls:

   • Protect against data exfiltration
   • Restrict access to GCP services

6. Configure Cloud Armor (production):

   • DDoS protection
   • WAF rules (SQL injection, XSS)

Compliance & Audit

Enable Audit Logging:

# Enable all audit logs (Admin Activity, Data Access, System Event)
gcloud logging read 'logName="projects/PROJECT_ID/logs/cloudaudit.googleapis.com"' \
  --limit 10 --format json

SOC 2 Requirements:

• Enable audit logging (all operations)
• Configure log retention (1 year minimum)
• Set up security monitoring alerts
• Regular access reviews (IAM)
• Encrypt data at rest (enabled by default)
• Encrypt data in transit (TLS 1.2+)

GDPR Requirements:

• Data residency (use europe-west1 for EU users)
• Data processing agreement with Google
• Right to erasure (document deletion procedures)
• Data portability (export procedures)

References

1. GCP Documentation:

   • GKE Overview: https://cloud.google.com/kubernetes-engine/docs
   • Cloud SQL PostgreSQL: https://cloud.google.com/sql/docs/postgres
   • Memorystore for Redis: https://cloud.google.com/memorystore/docs/redis
   • GCP Pricing Calculator: https://cloud.google.com/products/calculator

2. OctoLLM Documentation:

   • ADR-001: Technology Stack Selection
   • ADR-005: Deployment Platform
   • docs/operations/deployment-guide.md (2,863 lines)
   • to-dos/MASTER-TODO.md (Sprint 0.7 specification)

3. Competitor Comparisons:

   • AWS vs GCP vs Azure (Kubernetes): https://cloud.google.com/kubernetes-engine/docs/resources/kubernetes-on-aws-vs-gke
   • Database Comparison: https://db-engines.com/en/system/Amazon+RDS+for+PostgreSQL%3BGoogle+Cloud+SQL+for+PostgreSQL
   • Redis Comparison: ElastiCache vs Memorystore performance benchmarks

4. Community Resources:

   • r/googlecloud (Reddit community)
   • GCP Slack community
   • Stack Overflow (gcp tag)

Decision Date: 2025-11-12 Next Review: 2026-11-12 (annual review) Approved By: Architecture Team, DevOps Team, Finance Team Implementation Start: Sprint 0.7 (Infrastructure as Code - Week 1)

ADR-007: Unraid Local Deployment Strategy

Status: Proposed Date: 2025-11-12 Decision Makers: OctoLLM Architecture Team Consulted: DevOps, Infrastructure Team

Context

OctoLLM is a distributed AI architecture for offensive security and developer tooling that requires significant computational resources, particularly GPU acceleration for LLM inference. The project needs a local development deployment strategy that:

1. Leverages Available Hardware: Dell PowerEdge R730xd with dual Xeon E5-2683 v4 (64 threads), 504GB RAM, and NVIDIA Tesla P40 (24GB VRAM)
2. Minimizes Cloud Costs: Reduce dependency on expensive cloud LLM APIs (OpenAI/Anthropic)
3. Matches Production Architecture: Stay as close as possible to Kubernetes production deployment
4. Supports Rapid Iteration: Enable fast development cycles without complex orchestration overhead
5. Runs on Unraid 7.2.0: Integrate seamlessly with existing Unraid server infrastructure

Hardware Profile

Dell PowerEdge R730xd Specifications:

• CPU: Dual Intel Xeon E5-2683 v4 @ 2.10GHz (32 physical cores, 64 threads with HT)
• RAM: 503.8 GiB (492 GiB available)
• GPU: NVIDIA Tesla P40 (24GB VRAM, CUDA 13.0, Driver 580.105.08)
• Storage: 144TB array (51TB available), 1.8TB SSD cache
• Network: 4× Gigabit NICs bonded to 4Gbps aggregate (bond0)
• OS: Unraid 7.2.0 with Docker 27.5.1
• NUMA: 2 NUMA nodes (optimal for memory-intensive workloads)

Current Production Target

• Platform: Kubernetes (GKE/EKS) with multi-zone deployment
• LLM Strategy: Cloud APIs (OpenAI GPT-4, Anthropic Claude 3)
• Cost: $150-700/month for moderate development usage
• Complexity: High (requires K8s knowledge, Helm, kubectl, cloud account setup)

Decision

We will adopt a Hybrid Docker Compose + Local GPU Inference approach for Unraid local deployment:

Architecture Components

1. Docker Compose Stack:

   • All OctoLLM services (Orchestrator, Reflex, 6 Arms)
   • Infrastructure (PostgreSQL, Redis, Qdrant)
   • Monitoring (Prometheus, Grafana, Loki)
   • Exporters (node, cAdvisor, postgres, redis, nvidia-dcgm)

2. Local LLM Inference (Ollama):

   • GPU-accelerated inference on Tesla P40
   • Models: Llama 3.1 8B, Mixtral 8×7B, CodeLlama 13B, Nomic Embed Text
   • Replaces OpenAI/Anthropic APIs for 95% of requests
   • Cloud APIs available as fallback for edge cases

3. Unraid Integration:

   • App data in /mnt/user/appdata/octollm/ (standard Unraid location)
   • Permissions: nobody:users (99:100) per Unraid convention
   • Restart policy: unless-stopped (survives reboots)
   • Custom Docker network: octollm-net (172.20.0.0/16)

Resource Allocation

Utilization: 59% CPU, 9.5% RAM, 100% GPU during inference

Port Mapping

Core Services:
  3000  - Orchestrator API (main entry point)
  3001  - Reflex Layer API

Infrastructure:
  3010  - PostgreSQL
  3011  - Redis
  3012  - Qdrant HTTP API
  3013  - Qdrant gRPC API
  3014  - Ollama API

Arms:
  6001  - Planner Arm
  6002  - Executor Arm
  6003  - Retriever Arm
  6004  - Coder Arm
  6005  - Judge Arm
  6006  - Safety Guardian Arm

Monitoring:
  3030  - Grafana UI
  3100  - Loki (logs)
  8080  - cAdvisor
  9090  - Prometheus
  9100  - Node Exporter
  9121  - Redis Exporter
  9187  - PostgreSQL Exporter
  9400  - NVIDIA DCGM Exporter

Technology Stack

Alternatives Considered

Option 1: Pure Docker Compose (No GPU)

Approach: Docker Compose with all services, use cloud LLM APIs exclusively.

Pros:

• Simplest setup (no GPU drivers needed)
• Proven Docker Compose workflow
• Works on any hardware

Cons:

• Cost: $150-700/month in LLM API fees
• Wastes available Tesla P40 GPU
• Slower iteration (network latency to cloud APIs)
• API rate limits during development

Verdict: ❌ Rejected - Unnecessarily expensive, doesn't leverage available hardware

Option 2: K3s Virtual Machines (Lightweight Kubernetes)

Approach: Run k3s (lightweight K8s) in Unraid VMs, deploy with Helm charts.

Pros:

• Production parity: Near-identical to GKE/EKS deployment
• Kubernetes experience for team
• Could run multiple isolated environments
• GPU passthrough to VMs possible

Cons:

• Complexity overkill: Too heavy for single-developer local setup
• VM overhead (need 32GB+ RAM per VM for reasonable performance)
• Slower iteration (rebuild/deploy cycles)
• Requires Kubernetes expertise
• More failure points (VM networking, k3s networking, pod networking)
• Harder to debug (kubectl exec, logs aggregation)

Verdict: ⚠️ Deferred - Can add later for production testing, overkill for initial dev

Option 3: Hybrid Docker Compose + Local GPU (CHOSEN)

Approach: Docker Compose for services, Ollama for local GPU-accelerated LLM inference.

Pros:

• Cost savings: ~$0/month (electricity only vs. $150-700/month cloud APIs)
• Fast iteration: docker-compose up/down in seconds
• Leverages GPU: Tesla P40 runs Llama 3 70B, Mixtral 8×7B, CodeLlama 34B
• Unraid-native: Uses standard Unraid Docker patterns
• Production-similar: Services identical, only orchestration differs
• Debuggable: Direct docker logs, docker exec access
• Flexible: Can still use cloud APIs as fallback

Cons:

• Not 100% production-identical (Docker Compose vs. Kubernetes)
• Manual service management (no K8s auto-scaling, self-healing)
• Single-host limitations (no multi-node scheduling)

Mitigation:

• Services are containerized identically (Dockerfiles work in both)
• Can add k3s VMs later for Kubernetes testing
• Production deployment guide shows migration path

Verdict: ✅ CHOSEN - Best balance of cost, performance, and developer experience

Option 4: Docker Swarm

Approach: Docker Swarm for orchestration instead of Kubernetes.

Pros:

• Native Docker clustering
• Simpler than Kubernetes
• Built into Docker Engine

Cons:

• Production divergence: No one uses Swarm in production anymore
• Limited ecosystem compared to K8s
• Harder migration path to GKE/EKS
• Less learning value for team

Verdict: ❌ Rejected - Dead-end technology, no production alignment

Consequences

Positive

1. Dramatic Cost Reduction:

   • Before: $150-700/month in LLM API costs
   • After: ~$0/month (only electricity: ~$50/month for full server)
   • Annual Savings: $1,800-8,400

2. Faster Development Iteration:

   • Local inference: 2-10s latency (GPU-bound)
   • Cloud API: 5-30s latency (network + queue + inference)
   • No rate limits or quota concerns

3. Full Hardware Utilization:

   • Tesla P40 GPU: 100% utilized during inference
   • 64 CPU threads: 38 allocated (59%), 26 available for other services
   • 504GB RAM: 48GB allocated (9.5%), 450GB available
   • Efficient use of enterprise hardware

4. Production-Ready Learning Path:

   • Docker Compose → Docker images → Kubernetes deployment
   • Same service code, only orchestration changes
   • Team learns containerization first, orchestration second

5. Unraid Ecosystem Integration:

   • Appears in Unraid Docker tab
   • Uses standard appdata paths
   • Works with existing backup strategies
   • Compatible with Unraid Community Applications

6. Offline Development:

   • No internet required after initial setup
   • Works during cloud API outages
   • Data privacy (no external API calls)

Negative

1. Production Divergence:

   • Docker Compose vs. Kubernetes orchestration
   • Manual scaling vs. HorizontalPodAutoscaler
   • Docker networks vs. K8s Services/Ingress
   • Mitigation: Identical Docker images, migration guide provided

2. Single-Host Limitations:

   • No multi-node redundancy
   • No automatic failover
   • Mitigation: Acceptable for development, not for production

3. GPU Contention:

   • Only one GPU, shared by all arms
   • Ollama queues requests (max 4 parallel)
   • Mitigation: Still faster than cloud APIs, acceptable for dev

4. Model Management Overhead:

   • Need to pull/update models manually
   • 50-100GB model storage required
   • Mitigation: Setup script automates initial pull

5. Learning Curve for Ollama:

   • Team needs to understand local LLM deployment
   • Different prompt engineering vs. cloud APIs
   • Mitigation: Documentation provided, cloud APIs available as fallback

Migration Path to Production

When ready for cloud deployment:

1. Phase 1: Same Images, Different Orchestration

   • Use same Docker images from local development
   • Deploy to Kubernetes (GKE/EKS) with Helm charts
   • Switch from Ollama to OpenAI/Anthropic APIs

2. Phase 2: Cloud Infrastructure

   • Replace PostgreSQL with Cloud SQL
   • Replace Redis with Memorystore
   • Replace Qdrant self-hosted with Qdrant Cloud

3. Phase 3: Production Hardening

   • Add Ingress with TLS (cert-manager)
   • Configure HorizontalPodAutoscaler
   • Set up multi-region redundancy
   • Implement GitOps (ArgoCD/Flux)

Estimated Migration Time: 2-3 days for experienced team

Implementation Plan

Phase 1: Infrastructure Setup (Week 1)

• Create infrastructure/unraid/ directory structure
• Write docker-compose.unraid.yml (300-500 lines)
• Write .env.unraid.example (100 lines)
• Create setup-unraid.sh automated setup script (200-300 lines)
• Configure Prometheus with Unraid-specific metrics
• Create Grafana dashboard for Dell PowerEdge R730xd
• Write test suite (tests/*.sh)

Phase 2: Documentation (Week 1-2)

• Write ADR-007 (this document)
• Write comprehensive Unraid deployment guide (5,000 lines)
• Document Ollama model management
• Create troubleshooting playbook
• Write migration guide (Unraid → GKE)

Phase 3: Service Implementation (Week 2-4)

• Implement Orchestrator (Python FastAPI)
• Implement Reflex Layer (Rust Axum) - optional
• Implement 6 Arms (Planner, Executor, Retriever, Coder, Judge, Safety Guardian)
• Add Prometheus metrics to all services
• Integrate Ollama API calls

Phase 4: Testing & Validation (Week 4)

• Run full test suite
• Performance benchmarking (latency, throughput)
• Cost analysis (local vs. cloud)
• Load testing with multiple concurrent requests
• GPU utilization optimization

Metrics for Success

Risks and Mitigation

References

• OctoLLM Architecture
• Docker Compose Best Practices
• Ollama Documentation
• NVIDIA Tesla P40 Specifications
• Unraid Docker Documentation
• Prometheus Exporters

Approval

• Architecture Lead: ___________________ Date: __________
• DevOps Lead: ___________________ Date: __________
• Security Lead: ___________________ Date: __________

Changelog

• 2025-11-12: Initial proposal - Hybrid Docker Compose + Local GPU approach

Reflex Layer

Architecture

Pattern Matching

Performance

API Reference

Orchestrator

The central brain for strategic planning and coordination.

Status: Phase 1 Sprint 1.2 COMPLETE (v1.2.0)

Features

• Task submission and retrieval
• Reflex Layer integration with circuit breaker
• Async SQLAlchemy with PostgreSQL
• REST API with 6 endpoints

For implementation details, see services/orchestrator/.

Core Functionality

Database Layer

API Endpoints

Circuit Breaker

Implementation Details

Arms (Specialized Modules)

Arms are domain-specific execution modules with local autonomy and specialized expertise. Each arm handles a specific class of tasks and reports results back to the Orchestrator.

Arm Architecture

All arms share a common interface:

class ArmCapability:
    arm_id: str
    name: str
    description: str
    input_schema: JSONSchema
    output_schema: JSONSchema
    capabilities: List[str]  # Tags for routing
    cost_tier: int  # 1 (cheap) to 5 (expensive)
    endpoint: str  # Kubernetes service URL

Implemented Arms

1. Planner Arm (Sprint 1.3 - PLANNED)

Purpose: Task decomposition and workflow generation Technology: Python, GPT-3.5-turbo Status: 🚧 In Planning

Details: Planner Arm

2. Tool Executor Arm

Purpose: Execute external commands in sandboxed environments Technology: Rust for safety Status: ⏳ Not Started

Details: Tool Executor Arm

3. Retriever Arm

Purpose: Knowledge base search and information synthesis Technology: Python, Qdrant/Weaviate Status: ⏳ Not Started

Details: Retriever Arm

4. Coder Arm

Purpose: Code generation, debugging, and refactoring Technology: Python, specialized models Status: ⏳ Not Started

Details: Coder Arm

5. Judge Arm

Purpose: Output validation and quality assurance Technology: Python, validation frameworks Status: ⏳ Not Started

Details: Judge Arm

6. Safety Guardian Arm

Purpose: PII detection, content filtering, security checks Technology: Python/Rust, classifiers Status: ⏳ Not Started

Details: Safety Guardian Arm

Arm Capabilities

Communication Pattern

Orchestrator
    ↓ (TaskContract)
[Arm]
    ↓ (Execute with local autonomy)
[Arm] → Result
    ↓ (Response with confidence, provenance)
Orchestrator (integrate into global state)

See Also

• API Specifications
• Custom Arms Development
• Component Contracts

Planner Arm: Task Decomposition and Planning

Components > Arms > Planner Arm

Component: Planner Arm (Task Decomposition Specialist) Version: 1.0 Last Updated: 2025-11-10 Technology: Python 3.11+ / FastAPI Cost Tier: 2 (Medium) Average Latency: 1-2 seconds

Table of Contents

• Overview
• Core Functionality
• Architecture
• Implementation Details
• API Specification
• Data Structures
• Configuration
• Performance Characteristics
• Testing
• Error Handling
• Deployment
• See Also

Overview

The Planner Arm is a specialized component responsible for decomposing complex tasks into sequential subtasks with clear acceptance criteria, dependencies, and arm assignments. It serves as the strategic thinking component that bridges high-level goals with executable action plans.

Design Goals

• Intelligent Decomposition: Break complex goals into manageable, executable steps
• Dependency Awareness: Identify and track prerequisite relationships between steps
• Arm Selection: Match subtasks to the most appropriate specialized arms
• Quality Planning: Generate plans that maximize success probability
• Cost Awareness: Balance thoroughness with resource efficiency

Key Capabilities

1. Goal Parsing: Extract intent and requirements from natural language
2. Subtask Generation: Create 3-7 well-defined execution steps
3. Dependency Resolution: Establish correct execution order
4. Arm Selection: Match capabilities to subtasks
5. Acceptance Criteria: Define clear success conditions
6. Cost Estimation: Predict resource requirements

Core Functionality

Task Decomposition Algorithm

The Planner Arm uses an LLM-based approach with structured prompting to generate execution plans:

from typing import List, Dict, Any, Optional
from pydantic import BaseModel, Field
import openai
import json

class SubTask(BaseModel):
    """A single step in the execution plan."""
    step: int
    action: str = Field(..., description="What to do")
    required_arm: str = Field(..., description="Which arm executes this")
    acceptance_criteria: List[str] = Field(..., description="Success conditions")
    depends_on: List[int] = Field(default_factory=list, description="Prerequisite steps")
    estimated_cost_tier: int = Field(1, ge=1, le=5)
    estimated_duration_seconds: int = Field(30, ge=1)

class PlanResponse(BaseModel):
    """Complete execution plan."""
    plan: List[SubTask]
    rationale: str = Field(..., description="Why this approach")
    confidence: float = Field(..., ge=0.0, le=1.0)
    total_estimated_duration: int
    complexity_score: float = Field(..., ge=0.0, le=1.0)

class PlannerArm:
    """Task decomposition specialist."""

    def __init__(self, llm_model: str = "gpt-3.5-turbo"):
        self.model = llm_model
        self.system_prompt = self._build_system_prompt()

    def _build_system_prompt(self) -> str:
        return """You are an expert task planner for a distributed AI system.

Available arms and their capabilities:
- planner: Task decomposition, dependency resolution
- retriever: Search knowledge bases, documentation, web
- coder: Write/debug/refactor code, static analysis
- executor: Run shell commands, API calls, web scraping
- judge: Validate outputs, fact-check, quality assurance
- guardian: PII detection, safety checks, policy enforcement

Your task: Break down complex goals into 3-7 clear, executable steps.

For each step specify:
1. **action**: Clear, imperative description ("Search for...", "Generate...")
2. **required_arm**: Which arm should execute (match capabilities)
3. **acceptance_criteria**: 2-3 verifiable success conditions
4. **depends_on**: List of prerequisite step numbers (empty for first step)
5. **estimated_cost_tier**: 1=cheap, 5=expensive
6. **estimated_duration_seconds**: Realistic time estimate

Rules:
- Steps must be sequential and logically ordered
- Each step must have clear acceptance criteria
- Dependencies must reference earlier steps only
- Prefer specialized arms over generalists
- Include validation steps for critical outputs
- Always end with a verification/quality check step

Output valid JSON matching the PlanResponse schema."""

    async def generate_plan(
        self,
        goal: str,
        constraints: List[str],
        context: Dict[str, Any]
    ) -> PlanResponse:
        """Generate execution plan for goal."""

        user_prompt = f"""Goal: {goal}

Constraints:
{chr(10).join(f"- {c}" for c in constraints) if constraints else "None"}

Context:
{context if context else "None"}

Generate a detailed execution plan with 3-7 steps."""

        try:
            response = await openai.ChatCompletion.acreate(
                model=self.model,
                messages=[
                    {"role": "system", "content": self.system_prompt},
                    {"role": "user", "content": user_prompt}
                ],
                temperature=0.3,  # Lower for consistency
                max_tokens=2000,
                response_format={"type": "json_object"}
            )

            plan_data = json.loads(response.choices[0].message.content)

            # Calculate total duration
            total_duration = sum(
                step.get("estimated_duration_seconds", 30)
                for step in plan_data["plan"]
            )
            plan_data["total_estimated_duration"] = total_duration

            # Validate dependencies
            self._validate_dependencies(plan_data["plan"])

            return PlanResponse(**plan_data)

        except json.JSONDecodeError as e:
            raise ValueError(f"Failed to parse plan JSON: {e}")
        except Exception as e:
            raise RuntimeError(f"Planning failed: {e}")

    def _validate_dependencies(self, steps: List[Dict]) -> None:
        """Ensure dependencies reference valid steps."""
        step_numbers = {step["step"] for step in steps}

        for step in steps:
            for dep in step.get("depends_on", []):
                if dep not in step_numbers:
                    raise ValueError(
                        f"Step {step['step']} depends on non-existent step {dep}"
                    )
                if dep >= step["step"]:
                    raise ValueError(
                        f"Step {step['step']} cannot depend on later step {dep}"
                    )

Planning Flow

flowchart TD
    START([Receive Planning Request]) --> PARSE[Parse Goal & Constraints]
    PARSE --> LLM[Call LLM for Plan Generation]
    LLM --> VALIDATE{Valid JSON?}

    VALIDATE -->|No| RETRY{Retry Count < 3?}
    RETRY -->|Yes| LLM
    RETRY -->|No| ERROR([Return Error])

    VALIDATE -->|Yes| DEP_CHECK[Validate Dependencies]
    DEP_CHECK --> DEP_VALID{Dependencies Valid?}

    DEP_VALID -->|No| ERROR
    DEP_VALID -->|Yes| ESTIMATE[Calculate Estimates]

    ESTIMATE --> CONFIDENCE[Assess Confidence]
    CONFIDENCE --> RETURN([Return Plan])

    style START fill:#90EE90
    style RETURN fill:#90EE90
    style ERROR fill:#FFB6C1

Decision Tree for Arm Selection

graph TD
    ACTION[Action Description] --> KEYWORDS[Extract Keywords]

    KEYWORDS --> CODE{Contains code<br/>keywords?}
    CODE -->|Yes| CODER[Assign: Coder]

    CODE -->|No| SEARCH{Contains search<br/>keywords?}
    SEARCH -->|Yes| RETRIEVER[Assign: Retriever]

    SEARCH -->|No| EXEC{Contains execution<br/>keywords?}
    EXEC -->|Yes| EXECUTOR[Assign: Executor]

    EXEC -->|No| VALIDATE{Contains validation<br/>keywords?}
    VALIDATE -->|Yes| JUDGE[Assign: Judge]

    VALIDATE -->|No| SAFETY{Contains safety<br/>keywords?}
    SAFETY -->|Yes| GUARDIAN[Assign: Guardian]

    SAFETY -->|No| DEFAULT[Assign: Planner]

Architecture

Component Integration

graph TB
    subgraph "Planner Arm"
        PARSER[Intent Parser]
        GENERATOR[Plan Generator]
        VALIDATOR[Dependency Validator]
        ESTIMATOR[Cost Estimator]
    end

    subgraph "External Services"
        LLM[LLM API<br/>GPT-3.5/GPT-4]
        REGISTRY[Arm Registry<br/>Capability Database]
    end

    ORCHESTRATOR[Orchestrator] -->|Plan Request| PARSER
    PARSER --> GENERATOR
    GENERATOR --> LLM
    GENERATOR --> REGISTRY
    LLM --> VALIDATOR
    VALIDATOR --> ESTIMATOR
    ESTIMATOR -->|Plan Response| ORCHESTRATOR

Implementation Details

Complete FastAPI Implementation

from fastapi import FastAPI, HTTPException, BackgroundTasks
from fastapi.responses import JSONResponse
import structlog
from datetime import datetime
import uuid

logger = structlog.get_logger()

app = FastAPI(title="Planner Arm", version="1.0.0")

# Global planner instance
planner = PlannerArm(llm_model="gpt-3.5-turbo")

class PlanRequest(BaseModel):
    """Incoming planning request."""
    goal: str = Field(..., description="What to accomplish")
    constraints: List[str] = Field(default_factory=list)
    context: Dict[str, Any] = Field(default_factory=dict)
    request_id: Optional[str] = Field(default_factory=lambda: str(uuid.uuid4()))

@app.post("/plan", response_model=PlanResponse)
async def create_plan(request: PlanRequest):
    """Generate execution plan for given goal."""

    logger.info(
        "planner.plan.request",
        request_id=request.request_id,
        goal=request.goal[:100]
    )

    start_time = datetime.utcnow()

    try:
        plan = await planner.generate_plan(
            goal=request.goal,
            constraints=request.constraints,
            context=request.context
        )

        duration_ms = int((datetime.utcnow() - start_time).total_seconds() * 1000)

        logger.info(
            "planner.plan.success",
            request_id=request.request_id,
            steps=len(plan.plan),
            duration_ms=duration_ms,
            confidence=plan.confidence
        )

        return plan

    except ValueError as e:
        logger.error(
            "planner.plan.validation_error",
            request_id=request.request_id,
            error=str(e)
        )
        raise HTTPException(status_code=400, detail=str(e))

    except RuntimeError as e:
        logger.error(
            "planner.plan.runtime_error",
            request_id=request.request_id,
            error=str(e)
        )
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/health")
async def health_check():
    """Health check endpoint."""
    return {
        "status": "healthy",
        "version": "1.0.0",
        "model": planner.model,
        "timestamp": datetime.utcnow().isoformat()
    }

@app.get("/capabilities")
async def get_capabilities():
    """Return arm capabilities."""
    return {
        "arm_id": "planner",
        "capabilities": [
            "planning",
            "task_decomposition",
            "dependency_resolution",
            "arm_selection"
        ],
        "cost_tier": 2,
        "average_latency_ms": 1500,
        "success_rate": 0.92
    }

@app.get("/metrics")
async def get_metrics():
    """Prometheus metrics endpoint."""
    # Implement metrics collection
    return {"metrics": "not implemented"}

API Specification

POST /plan

Generate an execution plan for a given goal.

Request Body:

{
  "goal": "Fix authentication bug and add tests",
  "constraints": [
    "Don't modify database schema",
    "Complete in <5 minutes",
    "Maintain backward compatibility"
  ],
  "context": {
    "repository": "https://github.com/example/repo",
    "affected_files": ["auth/login.py"]
  }
}

Response (200 OK):

{
  "plan": [
    {
      "step": 1,
      "action": "Search codebase for authentication logic and recent bug reports",
      "required_arm": "retriever",
      "acceptance_criteria": [
        "Found auth/login.py implementation",
        "Identified related test files",
        "Located bug reports or issue references"
      ],
      "depends_on": [],
      "estimated_cost_tier": 1,
      "estimated_duration_seconds": 20
    },
    {
      "step": 2,
      "action": "Analyze authentication code to identify the bug",
      "required_arm": "coder",
      "acceptance_criteria": [
        "Root cause identified with line number",
        "Explanation of why bug occurs",
        "Proposed fix approach validated"
      ],
      "depends_on": [1],
      "estimated_cost_tier": 3,
      "estimated_duration_seconds": 60
    },
    {
      "step": 3,
      "action": "Generate code patch to fix authentication bug",
      "required_arm": "coder",
      "acceptance_criteria": [
        "Patch addresses root cause",
        "No breaking changes to API",
        "Code follows project style guide"
      ],
      "depends_on": [2],
      "estimated_cost_tier": 4,
      "estimated_duration_seconds": 45
    },
    {
      "step": 4,
      "action": "Generate test case that reproduces the bug scenario",
      "required_arm": "coder",
      "acceptance_criteria": [
        "Test fails on old code",
        "Test passes on patched code",
        "Test covers edge cases"
      ],
      "depends_on": [3],
      "estimated_cost_tier": 3,
      "estimated_duration_seconds": 40
    },
    {
      "step": 5,
      "action": "Run full test suite to verify no regressions",
      "required_arm": "executor",
      "acceptance_criteria": [
        "All existing tests pass",
        "New test passes",
        "No test timeouts or errors"
      ],
      "depends_on": [4],
      "estimated_cost_tier": 2,
      "estimated_duration_seconds": 90
    },
    {
      "step": 6,
      "action": "Validate fix meets acceptance criteria and constraints",
      "required_arm": "judge",
      "acceptance_criteria": [
        "All original acceptance criteria met",
        "No database schema changes",
        "Backward compatibility maintained"
      ],
      "depends_on": [5],
      "estimated_cost_tier": 2,
      "estimated_duration_seconds": 30
    }
  ],
  "rationale": "This plan follows a systematic debugging workflow: locate code, identify bug, fix it, test thoroughly, and validate. Each step has clear outputs that feed into the next, ensuring quality and meeting all constraints.",
  "confidence": 0.88,
  "total_estimated_duration": 285,
  "complexity_score": 0.65
}

Error Responses:

• 400 Bad Request: Invalid dependencies or malformed plan
• 500 Internal Server Error: LLM API failure or planning error
• 503 Service Unavailable: LLM service temporarily unavailable

Data Structures

All data structures use Pydantic models for validation and serialization:

class SubTask(BaseModel):
    """A single step in the execution plan."""
    step: int
    action: str = Field(..., description="What to do")
    required_arm: str = Field(..., description="Which arm executes this")
    acceptance_criteria: List[str] = Field(..., description="Success conditions")
    depends_on: List[int] = Field(default_factory=list, description="Prerequisite steps")
    estimated_cost_tier: int = Field(1, ge=1, le=5)
    estimated_duration_seconds: int = Field(30, ge=1)

class PlanResponse(BaseModel):
    """Complete execution plan."""
    plan: List[SubTask]
    rationale: str = Field(..., description="Why this approach")
    confidence: float = Field(..., ge=0.0, le=1.0)
    total_estimated_duration: int
    complexity_score: float = Field(..., ge=0.0, le=1.0)

class PlanRequest(BaseModel):
    """Incoming planning request."""
    goal: str = Field(..., description="What to accomplish")
    constraints: List[str] = Field(default_factory=list)
    context: Dict[str, Any] = Field(default_factory=dict)
    request_id: Optional[str] = Field(default_factory=lambda: str(uuid.uuid4()))

Configuration

Environment Variables

Configuration File

# planner-config.yaml
model:
  provider: "openai"
  name: "gpt-3.5-turbo"
  temperature: 0.3
  max_tokens: 2000

planning:
  min_steps: 3
  max_steps: 7
  require_validation_step: true
  require_dependency_check: true

arms:
  - id: "retriever"
    capabilities: ["search", "knowledge_retrieval"]
  - id: "coder"
    capabilities: ["code_generation", "debugging"]
  - id: "executor"
    capabilities: ["shell", "api_calls"]
  - id: "judge"
    capabilities: ["validation", "fact_checking"]
  - id: "guardian"
    capabilities: ["pii_detection", "safety_check"]

Performance Characteristics

Latency Breakdown

Resource Requirements

Per Instance:

• CPU: 200m (0.2 cores) baseline, 500m under load
• Memory: 256Mi baseline, 512Mi under load
• Disk: Negligible (<100Mi)

Success Rate Metrics

• Overall Success Rate: >92%
• Valid JSON Rate: >98%
• Dependency Validation Pass Rate: >95%
• Plan Execution Success Rate: >88% (downstream)

Cost Analysis

• Cost Tier: 2 (Medium)
• LLM Cost per Plan: $0.002-0.005 (GPT-3.5)
• Requests per Dollar: 200-500
• Monthly Cost (1000 plans): $2-5

Testing

Unit Tests

import pytest
from unittest.mock import AsyncMock, patch

@pytest.mark.asyncio
async def test_plan_generation():
    """Test basic plan generation."""
    planner = PlannerArm()

    plan = await planner.generate_plan(
        goal="Write a function to sort a list",
        constraints=["Use Python", "Include doctests"],
        context={}
    )

    assert len(plan.plan) >= 3
    assert len(plan.plan) <= 7
    assert all(step.step == idx + 1 for idx, step in enumerate(plan.plan))
    assert plan.confidence > 0.5

    # Validate dependencies
    for step in plan.plan:
        for dep in step.depends_on:
            assert dep < step.step

@pytest.mark.asyncio
async def test_complex_plan_with_dependencies():
    """Test complex plan with multiple dependencies."""
    planner = PlannerArm()

    plan = await planner.generate_plan(
        goal="Build and deploy a REST API",
        constraints=["Use FastAPI", "Include tests", "Deploy to Kubernetes"],
        context={"language": "Python"}
    )

    # Should have multiple dependent steps
    dependent_steps = [s for s in plan.plan if s.depends_on]
    assert len(dependent_steps) > 0

    # Should include different arms
    arms_used = {s.required_arm for s in plan.plan}
    assert "coder" in arms_used
    assert "executor" in arms_used or "judge" in arms_used

@pytest.mark.asyncio
async def test_dependency_validation():
    """Test dependency validation catches errors."""
    planner = PlannerArm()

    invalid_steps = [
        {"step": 1, "action": "Do A", "depends_on": []},
        {"step": 2, "action": "Do B", "depends_on": [3]},  # Invalid: depends on future
        {"step": 3, "action": "Do C", "depends_on": [1]}
    ]

    with pytest.raises(ValueError, match="cannot depend on later step"):
        planner._validate_dependencies(invalid_steps)

@pytest.mark.asyncio
async def test_invalid_json_handling():
    """Test handling of invalid JSON from LLM."""
    planner = PlannerArm()

    with patch.object(openai.ChatCompletion, 'acreate') as mock_create:
        mock_create.return_value = AsyncMock(
            choices=[AsyncMock(message=AsyncMock(content="Invalid JSON {"))]
        )

        with pytest.raises(ValueError, match="Failed to parse plan JSON"):
            await planner.generate_plan("Test goal", [], {})

Integration Tests

@pytest.mark.asyncio
@pytest.mark.integration
async def test_end_to_end_planning():
    """Test complete planning workflow with real LLM."""
    planner = PlannerArm(llm_model="gpt-3.5-turbo")

    plan = await planner.generate_plan(
        goal="Create a Python script to analyze CSV data",
        constraints=[
            "Use pandas library",
            "Include error handling",
            "Output results to JSON"
        ],
        context={
            "experience_level": "intermediate",
            "data_source": "sales_data.csv"
        }
    )

    # Verify plan structure
    assert isinstance(plan, PlanResponse)
    assert 3 <= len(plan.plan) <= 7
    assert plan.confidence > 0.6

    # Verify steps are properly ordered
    for idx, step in enumerate(plan.plan):
        assert step.step == idx + 1

    # Verify all dependencies are valid
    for step in plan.plan:
        for dep in step.depends_on:
            assert dep < step.step

    # Verify arms are assigned
    for step in plan.plan:
        assert step.required_arm in [
            "retriever", "coder", "executor", "judge", "guardian", "planner"
        ]

Error Handling

Error Types

class PlanningError(Exception):
    """Base exception for planning errors."""
    pass

class InvalidDependencyError(PlanningError):
    """Raised when dependencies are invalid."""
    pass

class PlanningTimeoutError(PlanningError):
    """Raised when planning exceeds timeout."""
    pass

class LLMError(PlanningError):
    """Raised when LLM API fails."""
    pass

Error Recovery Strategies

Deployment

Dockerfile

FROM python:3.11-slim

WORKDIR /app

# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy application
COPY . .

# Set environment
ENV PYTHONUNBUFFERED=1
ENV LOG_LEVEL=INFO

EXPOSE 8080

# Health check
HEALTHCHECK --interval=10s --timeout=3s --start-period=5s --retries=3 \
    CMD curl -f http://localhost:8080/health || exit 1

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8080"]

Kubernetes Manifest

apiVersion: apps/v1
kind: Deployment
metadata:
  name: planner-arm
  namespace: octollm
spec:
  replicas: 2
  selector:
    matchLabels:
      app: planner-arm
  template:
    metadata:
      labels:
        app: planner-arm
        component: arm
    spec:
      containers:
        - name: planner
          image: octollm/planner-arm:1.0.0
          ports:
            - containerPort: 8080
              name: http
          env:
            - name: OPENAI_API_KEY
              valueFrom:
                secretKeyRef:
                  name: llm-credentials
                  key: openai-api-key
            - name: LLM_MODEL
              value: "gpt-3.5-turbo"
            - name: LOG_LEVEL
              value: "INFO"
          resources:
            requests:
              cpu: 200m
              memory: 256Mi
            limits:
              cpu: 500m
              memory: 512Mi
          livenessProbe:
            httpGet:
              path: /health
              port: 8080
            initialDelaySeconds: 10
            periodSeconds: 5
          readinessProbe:
            httpGet:
              path: /health
              port: 8080
            initialDelaySeconds: 5
            periodSeconds: 3
---
apiVersion: v1
kind: Service
metadata:
  name: planner-arm
  namespace: octollm
spec:
  selector:
    app: planner-arm
  ports:
    - protocol: TCP
      port: 8080
      targetPort: 8080
  type: ClusterIP

See Also

• Orchestrator Specification - For task coordination
• Arm API Contracts - Standard message formats
• Memory Systems - Knowledge storage
• Testing Strategy - Testing approaches

Document Version: 1.0 Last Updated: 2025-11-10 Maintainer: OctoLLM Core Team

Tool Executor Arm: Sandboxed Command Execution

Components > Arms > Tool Executor Arm

Version: 1.0 Technology: Rust / actix-web Cost Tier: 3 (Medium-High) Average Latency: 0.5-5 seconds Status: Phase 1 Complete

Table of Contents

• Overview
• Architecture
• Security Model
  • Capability-Based Access Control
  • Capability Types
• Core Functionality
  • Command Allowlist
  • Sandboxed Execution
  • Resource Limits
• Implementation
  • Executor Structure
  • Command Validation
  • Execution Pipeline
• API Specification
  • Execute Command
  • Response Formats
• Data Models
• Configuration
• Performance Characteristics
• Testing
• Deployment
  • Docker Sandbox
  • Kubernetes Configuration
• Security Considerations
• See Also

Overview

The Tool Executor Arm is a security-first component that executes external commands, API calls, and scripts in isolated sandboxes with strict capability controls. It provides the system with the ability to interact with external tools while maintaining strong security boundaries.

Key Features

• Capability-Based Access Control: Fine-grained permissions for command execution
• Command Allowlist: Only pre-approved commands can be executed
• Sandbox Isolation: All executions run in isolated Docker containers
• Resource Limits: Timeouts, memory limits, and CPU restrictions
• Provenance Tracking: Complete audit trail of all executions
• Network Control: Host allowlisting for HTTP requests
• Non-Root Execution: All commands run as unprivileged users

Design Principles

1. Security by Default: Deny all, permit explicitly
2. Defense in Depth: Multiple layers of security controls
3. Least Privilege: Minimal capabilities granted for each operation
4. Auditability: Complete logging and provenance metadata
5. Fail-Safe: Errors default to blocking execution

Architecture

graph TB
    subgraph "Executor Arm"
        API[API Endpoint]
        VAL[Validator]
        EXEC[Executor]
        SAND[Sandbox Manager]
        PROV[Provenance Tracker]
    end

    subgraph "Security Layer"
        CAP[Capability Checker]
        ALLOW[Allowlist]
        HOST[Host Validator]
    end

    subgraph "Execution Environment"
        DOCKER[Docker Container]
        FS[Restricted Filesystem]
        NET[Network Namespace]
    end

    ORCH[Orchestrator] -->|Execute Request + Token| API
    API --> VAL
    VAL --> CAP
    VAL --> ALLOW
    VAL --> HOST

    CAP -->|Authorized| EXEC
    ALLOW -->|Permitted| EXEC
    HOST -->|Valid| EXEC

    EXEC --> SAND
    SAND --> DOCKER
    DOCKER --> FS
    DOCKER --> NET

    EXEC --> PROV
    PROV -->|Provenance Metadata| API
    API -->|Execution Result| ORCH

    CAP -->|Denied| API
    ALLOW -->|Blocked| API
    HOST -->|Invalid| API

    style DOCKER fill:#f9f,stroke:#333
    style CAP fill:#ff9,stroke:#333
    style PROV fill:#9ff,stroke:#333

Execution Flow

sequenceDiagram
    participant O as Orchestrator
    participant E as Executor API
    participant V as Validator
    participant S as Sandbox
    participant D as Docker

    O->>E: POST /execute (command + token)
    E->>V: Validate request

    alt Token Valid
        V->>V: Check capabilities
        alt Capability Granted
            V->>V: Check allowlist
            alt Command Allowed
                V->>S: Prepare sandbox
                S->>D: Create container
                D-->>S: Container ready
                S->>D: Execute command
                D-->>S: Output + exit code
                S->>E: Execution result
                E->>E: Generate provenance
                E-->>O: Success response
            else Command Blocked
                V-->>E: Allowlist violation
                E-->>O: Error: Command not allowed
            end
        else No Capability
            V-->>E: Capability violation
            E-->>O: Error: Insufficient privileges
        end
    else Token Invalid
        V-->>E: Auth failure
        E-->>O: Error: Invalid token
    end

Security Model

Capability-Based Access Control

The Executor Arm uses a capability-based security model where each operation requires specific permissions granted through time-limited tokens.

#[derive(Debug, Clone, Serialize, Deserialize)]
struct CapabilityToken {
    token_id: String,
    granted_capabilities: HashSet<Capability>,
    expires_at: DateTime<Utc>,
    issued_to: String,
}

#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
enum Capability {
    // Shell command execution
    ShellRead,        // Read-only commands (ls, cat, grep)
    ShellWrite,       // Write commands (echo >, mkdir)
    ShellExecute,     // Execute scripts

    // Network access
    HttpGet,          // HTTP GET requests
    HttpPost,         // HTTP POST requests
    HttpAllHosts,     // Access any host (vs allowlist)

    // File system
    FilesystemRead,   // Read files
    FilesystemWrite,  // Write files
    FilesystemDelete, // Delete files

    // Special
    PythonExec,       // Run Python scripts
    DockerAccess,     // Access Docker API
}

impl CapabilityToken {
    fn can_execute(&self, required: &Capability) -> bool {
        !self.is_expired() && self.granted_capabilities.contains(required)
    }

    fn is_expired(&self) -> bool {
        Utc::now() > self.expires_at
    }
}

Capability Types

Core Functionality

Command Allowlist

Only pre-approved commands can be executed, with required capabilities mapped to each command.

struct Executor {
    allowed_commands: HashMap<String, Vec<Capability>>,
    allowed_hosts: Vec<String>,
    timeout: Duration,
}

impl Executor {
    fn default_safe() -> Self {
        let mut allowed_commands = HashMap::new();

        // Read-only commands
        allowed_commands.insert("echo".to_string(), vec![Capability::ShellRead]);
        allowed_commands.insert("cat".to_string(), vec![Capability::ShellRead, Capability::FilesystemRead]);
        allowed_commands.insert("ls".to_string(), vec![Capability::ShellRead, Capability::FilesystemRead]);
        allowed_commands.insert("grep".to_string(), vec![Capability::ShellRead]);
        allowed_commands.insert("find".to_string(), vec![Capability::ShellRead, Capability::FilesystemRead]);
        allowed_commands.insert("head".to_string(), vec![Capability::ShellRead, Capability::FilesystemRead]);
        allowed_commands.insert("tail".to_string(), vec![Capability::ShellRead, Capability::FilesystemRead]);

        // Network commands
        allowed_commands.insert("curl".to_string(), vec![Capability::HttpGet]);
        allowed_commands.insert("wget".to_string(), vec![Capability::HttpGet]);

        // Version control (read-only)
        allowed_commands.insert("git".to_string(), vec![Capability::ShellRead, Capability::FilesystemRead]);

        Self {
            allowed_commands,
            allowed_hosts: vec![
                "api.github.com".to_string(),
                "registry.npmjs.org".to_string(),
                "pypi.org".to_string(),
            ],
            timeout: Duration::from_secs(30),
        }
    }
}

Sandboxed Execution

All commands execute in isolated environments with resource limits.

impl Executor {
    async fn execute(&self, req: ExecutionRequest, token: &CapabilityToken) -> Result<ExecutionResult> {
        // 1. Validate command is allowed
        self.validate_command(&req.command, token)?;

        // 2. For HTTP requests, validate host
        if req.action_type == "http" {
            self.validate_host(&req.command, token)?;
        }

        // 3. Execute with timeout and resource limits
        let result = self.execute_sandboxed(req).await?;

        // 4. Generate provenance metadata
        let provenance = self.generate_provenance(&req, &result);

        Ok(ExecutionResult {
            success: result.status.success(),
            stdout: String::from_utf8_lossy(&result.stdout).to_string(),
            stderr: String::from_utf8_lossy(&result.stderr).to_string(),
            exit_code: result.status.code(),
            duration_ms: result.duration.as_millis() as u64,
            provenance,
        })
    }

    async fn execute_sandboxed(&self, req: ExecutionRequest) -> Result<CommandOutput> {
        use tokio::process::Command;
        use tokio::time::timeout;

        let start = Instant::now();

        // Build command with resource limits
        let mut cmd = Command::new(&req.command);
        cmd.args(&req.args)
           .stdout(Stdio::piped())
           .stderr(Stdio::piped())
           .kill_on_drop(true);

        // Execute with timeout
        let output = timeout(self.timeout, cmd.output())
            .await
            .map_err(|_| Error::Timeout)?
            .map_err(|e| Error::Execution(e.to_string()))?;

        Ok(CommandOutput {
            status: output.status,
            stdout: output.stdout,
            stderr: output.stderr,
            duration: start.elapsed(),
        })
    }
}

Resource Limits

Implementation

Executor Structure

use actix_web::{web, App, HttpResponse, HttpServer};
use serde::{Deserialize, Serialize};
use std::collections::{HashMap, HashSet};
use std::time::{Duration, Instant};
use tokio::process::{Command, Stdio};
use chrono::{DateTime, Utc};

#[derive(Debug, Deserialize)]
struct ExecutionRequest {
    action_type: String,  // "shell", "http", "python"
    command: String,
    args: Vec<String>,
    timeout_seconds: Option<u64>,
    capability_token: String,
    metadata: HashMap<String, String>,
}

#[derive(Debug, Serialize)]
struct ExecutionResult {
    success: bool,
    stdout: String,
    stderr: String,
    exit_code: Option<i32>,
    duration_ms: u64,
    provenance: ProvenanceMetadata,
}

#[derive(Debug, Serialize)]
struct ProvenanceMetadata {
    arm_id: String,
    timestamp: DateTime<Utc>,
    action_type: String,
    command_hash: String,
    capabilities_used: Vec<String>,
}

struct CommandOutput {
    status: std::process::ExitStatus,
    stdout: Vec<u8>,
    stderr: Vec<u8>,
    duration: Duration,
}

Command Validation

impl Executor {
    fn validate_command(&self, command: &str, token: &CapabilityToken) -> Result<()> {
        // Check if command is in allowlist
        let required_caps = self.allowed_commands
            .get(command)
            .ok_or(Error::CommandNotAllowed(command.to_string()))?;

        // Check if token has all required capabilities
        for cap in required_caps {
            if !token.can_execute(cap) {
                return Err(Error::InsufficientCapability {
                    required: cap.clone(),
                    command: command.to_string(),
                });
            }
        }

        Ok(())
    }

    fn validate_host(&self, url: &str, token: &CapabilityToken) -> Result<()> {
        // If token has HttpAllHosts, allow any host
        if token.can_execute(&Capability::HttpAllHosts) {
            return Ok(());
        }

        // Otherwise, check allowlist
        let host = extract_host(url)?;
        if !self.allowed_hosts.contains(&host) {
            return Err(Error::HostNotAllowed(host));
        }

        Ok(())
    }

    fn generate_provenance(&self, req: &ExecutionRequest, result: &CommandOutput) -> ProvenanceMetadata {
        use sha2::{Sha256, Digest};

        let command_str = format!("{} {}", req.command, req.args.join(" "));
        let mut hasher = Sha256::new();
        hasher.update(command_str.as_bytes());
        let command_hash = format!("{:x}", hasher.finalize());

        ProvenanceMetadata {
            arm_id: "executor".to_string(),
            timestamp: Utc::now(),
            action_type: req.action_type.clone(),
            command_hash,
            capabilities_used: self.get_used_capabilities(&req.command),
        }
    }
}

Execution Pipeline

graph LR
    A[Request] --> B{Token Valid?}
    B -->|No| Z[Error: Auth]
    B -->|Yes| C{Capability?}
    C -->|No| Z
    C -->|Yes| D{Allowlist?}
    D -->|No| Z
    D -->|Yes| E{HTTP?}
    E -->|Yes| F{Host OK?}
    F -->|No| Z
    E -->|No| G[Execute]
    F -->|Yes| G
    G --> H[Result]
    H --> I[Provenance]
    I --> J[Response]

    style Z fill:#f99,stroke:#333
    style J fill:#9f9,stroke:#333

API Specification

Execute Command

Endpoint: POST /execute

Headers:

Content-Type: application/json
X-Request-ID: uuid (optional)

Request Body:

{
  "action_type": "shell",
  "command": "ls",
  "args": ["-la", "/tmp"],
  "timeout_seconds": 10,
  "capability_token": "tok_abc123xyz",
  "metadata": {
    "task_id": "task-123",
    "requested_by": "orchestrator"
  }
}

Field Descriptions:

Response Formats

Success Response (200 OK):

{
  "success": true,
  "stdout": "total 32\ndrwxrwxrwt 10 root root 4096 Nov 10 10:30 .\ndrwxr-xr-x 20 root root 4096 Oct 15 08:12 ..",
  "stderr": "",
  "exit_code": 0,
  "duration_ms": 45,
  "provenance": {
    "arm_id": "executor",
    "timestamp": "2025-11-10T10:30:00Z",
    "action_type": "shell",
    "command_hash": "5d41402abc4b2a76b9719d911017c592",
    "capabilities_used": ["ShellRead", "FilesystemRead"]
  }
}

Blocked Command (403 Forbidden):

{
  "success": false,
  "error": "Command 'rm' not in allowlist",
  "error_type": "CapabilityViolation",
  "allowed_commands": ["echo", "cat", "ls", "grep", "curl"]
}

Invalid Token (401 Unauthorized):

{
  "success": false,
  "error": "Capability token expired or invalid",
  "error_type": "AuthenticationFailure"
}

Execution Timeout (408 Request Timeout):

{
  "success": false,
  "error": "Command execution exceeded timeout of 30 seconds",
  "error_type": "ExecutionTimeout",
  "partial_output": "...",
  "duration_ms": 30000
}

Data Models

Capability Token

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CapabilityToken {
    pub token_id: String,
    pub granted_capabilities: HashSet<Capability>,
    pub expires_at: DateTime<Utc>,
    pub issued_to: String,
}

Error Types

#[derive(Debug, thiserror::Error)]
pub enum Error {
    #[error("Command '{0}' not in allowlist")]
    CommandNotAllowed(String),

    #[error("Host '{0}' not in allowlist")]
    HostNotAllowed(String),

    #[error("Insufficient capability: {command} requires {required:?}")]
    InsufficientCapability {
        required: Capability,
        command: String,
    },

    #[error("Token expired or invalid")]
    InvalidToken,

    #[error("Execution timeout")]
    Timeout,

    #[error("Execution failed: {0}")]
    Execution(String),
}

Configuration

Environment Variables

# Executor Configuration
EXECUTOR_PORT=8003
EXECUTOR_TIMEOUT_SECONDS=30
EXECUTOR_MAX_CONCURRENT=10

# Security
EXECUTOR_ALLOWLIST_PATH=/etc/executor/allowlist.yaml
EXECUTOR_HOST_ALLOWLIST_PATH=/etc/executor/hosts.yaml
CAPABILITY_TOKEN_VERIFIER_URL=http://orchestrator:8000/verify-token

# Sandbox
SANDBOX_TYPE=docker  # docker, kubernetes, firecracker
SANDBOX_IMAGE=executor-sandbox:latest
SANDBOX_MEMORY_LIMIT=512m
SANDBOX_CPU_LIMIT=1.0

# Logging
LOG_LEVEL=info
LOG_FORMAT=json
PROVENANCE_LOG_PATH=/var/log/executor/provenance.jsonl

Allowlist Configuration

allowlist.yaml:

commands:
  # Read-only commands
  - name: echo
    capabilities:
      - ShellRead
    description: "Print text"

  - name: cat
    capabilities:
      - ShellRead
      - FilesystemRead
    description: "Display file contents"

  - name: ls
    capabilities:
      - ShellRead
      - FilesystemRead
    description: "List directory contents"

  # Network commands
  - name: curl
    capabilities:
      - HttpGet
    description: "HTTP GET requests"

  - name: wget
    capabilities:
      - HttpGet
    description: "Download files"

# Host allowlist
hosts:
  - api.github.com
  - registry.npmjs.org
  - pypi.org
  - api.openai.com

# Sandbox configuration
sandbox:
  memory_limit: "512m"
  cpu_limit: 1.0
  timeout_seconds: 30
  max_processes: 10
  readonly_root: true
  writable_paths:
    - /tmp
    - /workspace

Performance Characteristics

Latency

Throughput

• Concurrent Executions: 10 (configurable)
• Queue Depth: 100 requests
• Requests/Second: ~40 (with 10 workers)

Resource Usage

• Memory: 50 MB base + 512 MB per sandbox
• CPU: Minimal (execution in sandbox)
• Disk: 10 MB logs per hour

Testing

Unit Tests

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_capability_validation() {
        let mut caps = HashSet::new();
        caps.insert(Capability::ShellRead);

        let token = CapabilityToken {
            token_id: "test".to_string(),
            granted_capabilities: caps,
            expires_at: Utc::now() + Duration::from_secs(3600),
            issued_to: "test".to_string(),
        };

        assert!(token.can_execute(&Capability::ShellRead));
        assert!(!token.can_execute(&Capability::ShellWrite));
    }

    #[test]
    fn test_token_expiration() {
        let token = CapabilityToken {
            token_id: "test".to_string(),
            granted_capabilities: HashSet::new(),
            expires_at: Utc::now() - Duration::from_secs(1),
            issued_to: "test".to_string(),
        };

        assert!(token.is_expired());
    }

    #[tokio::test]
    async fn test_command_allowlist() {
        let executor = Executor::default_safe();

        let mut caps = HashSet::new();
        caps.insert(Capability::ShellRead);
        caps.insert(Capability::FilesystemRead);

        let token = CapabilityToken {
            token_id: "test".to_string(),
            granted_capabilities: caps,
            expires_at: Utc::now() + Duration::from_secs(3600),
            issued_to: "test".to_string(),
        };

        // Should succeed
        assert!(executor.validate_command("ls", &token).is_ok());

        // Should fail (not in allowlist)
        assert!(executor.validate_command("rm", &token).is_err());
    }
}

Integration Tests

#[tokio::test]
async fn test_execute_safe_command() {
    let executor = Executor::default_safe();

    let mut caps = HashSet::new();
    caps.insert(Capability::ShellRead);

    let token = CapabilityToken {
        token_id: "test".to_string(),
        granted_capabilities: caps,
        expires_at: Utc::now() + Duration::from_secs(3600),
        issued_to: "test".to_string(),
    };

    let req = ExecutionRequest {
        action_type: "shell".to_string(),
        command: "echo".to_string(),
        args: vec!["Hello, World!".to_string()],
        timeout_seconds: Some(5),
        capability_token: token.token_id.clone(),
        metadata: HashMap::new(),
    };

    let result = executor.execute(req, &token).await.unwrap();

    assert!(result.success);
    assert_eq!(result.stdout.trim(), "Hello, World!");
    assert_eq!(result.exit_code, Some(0));
}

#[tokio::test]
async fn test_blocked_command() {
    let executor = Executor::default_safe();

    let mut caps = HashSet::new();
    caps.insert(Capability::ShellRead);

    let token = CapabilityToken {
        token_id: "test".to_string(),
        granted_capabilities: caps,
        expires_at: Utc::now() + Duration::from_secs(3600),
        issued_to: "test".to_string(),
    };

    let req = ExecutionRequest {
        action_type: "shell".to_string(),
        command: "rm".to_string(),  // Not in allowlist
        args: vec!["-rf".to_string(), "/".to_string()],
        timeout_seconds: Some(5),
        capability_token: token.token_id.clone(),
        metadata: HashMap::new(),
    };

    let result = executor.execute(req, &token).await;
    assert!(result.is_err());
}

Deployment

Docker Sandbox

Dockerfile:

FROM debian:bookworm-slim

# Install minimal toolset
RUN apt-get update && apt-get install -y \
    curl \
    git \
    && rm -rf /var/lib/apt/lists/*

# Create non-root user
RUN useradd -m -s /bin/bash executor
USER executor

# Set restrictive umask
RUN echo "umask 077" >> /home/executor/.bashrc

WORKDIR /workspace

# No CMD - controlled by executor service

Kubernetes Configuration

deployment.yaml:

apiVersion: apps/v1
kind: Deployment
metadata:
  name: executor-arm
  namespace: octollm
spec:
  replicas: 2
  selector:
    matchLabels:
      app: executor-arm
  template:
    metadata:
      labels:
        app: executor-arm
    spec:
      serviceAccountName: executor-arm

      # Security Context
      securityContext:
        runAsNonRoot: true
        runAsUser: 1000
        fsGroup: 1000
        seccompProfile:
          type: RuntimeDefault

      containers:
      - name: executor
        image: octollm/executor-arm:1.0

        # Container Security
        securityContext:
          allowPrivilegeEscalation: false
          readOnlyRootFilesystem: true
          capabilities:
            drop:
              - ALL

        # Resource Limits
        resources:
          requests:
            memory: "128Mi"
            cpu: "100m"
          limits:
            memory: "1Gi"
            cpu: "1000m"

        # Port
        ports:
        - containerPort: 8003
          name: http

        # Configuration
        env:
        - name: EXECUTOR_PORT
          value: "8003"
        - name: EXECUTOR_TIMEOUT_SECONDS
          value: "30"
        - name: SANDBOX_TYPE
          value: "docker"

        # Config Volume
        volumeMounts:
        - name: config
          mountPath: /etc/executor
          readOnly: true
        - name: tmp
          mountPath: /tmp

      volumes:
      - name: config
        configMap:
          name: executor-config
      - name: tmp
        emptyDir: {}

---
apiVersion: v1
kind: Service
metadata:
  name: executor-arm
  namespace: octollm
spec:
  selector:
    app: executor-arm
  ports:
  - port: 8003
    targetPort: 8003
    name: http
  type: ClusterIP

ConfigMap:

apiVersion: v1
kind: ConfigMap
metadata:
  name: executor-config
  namespace: octollm
data:
  allowlist.yaml: |
    commands:
      - name: echo
        capabilities: [ShellRead]
      - name: cat
        capabilities: [ShellRead, FilesystemRead]
      - name: ls
        capabilities: [ShellRead, FilesystemRead]
      - name: curl
        capabilities: [HttpGet]

    hosts:
      - api.github.com
      - pypi.org

    sandbox:
      memory_limit: "512m"
      timeout_seconds: 30

Security Considerations

Threat Model

Security Best Practices

1. Never Run as Root: All executions use unprivileged users
2. Minimal Capabilities: Grant only required capabilities
3. Short-Lived Tokens: Tokens expire after 1 hour by default
4. Audit Logging: Log all executions with provenance metadata
5. Network Isolation: Use network policies in Kubernetes
6. Regular Updates: Keep sandbox images and tools updated
7. Penetration Testing: Regular security assessments

See Also

• Orchestrator Component - Token issuance and coordination
• Planner Arm - Task decomposition that generates execution plans
• Safety Guardian Arm - Pre-execution validation
• Security Architecture - System-wide security model
• Capability Isolation - Detailed capability design
• API Reference - Complete API documentation

Document Status: Phase 1 Complete Last Updated: 2025-11-10 Maintainer: OctoLLM Core Team Next Review: 2025-12-10

Retriever Arm: Knowledge Search & Synthesis

Components > Arms > Retriever Arm

Version: 1.0 Technology: Python 3.11+ / FastAPI Cost Tier: 1 (Low) Average Latency: 100-500ms Status: Phase 1 Complete

Table of Contents

• Overview
• Architecture
• Core Functionality
  • Search Methods
  • Hybrid Search Strategy
  • Reranking
  • Synthesis
• Search Implementations
  • Vector Search
  • Keyword Search
  • Hybrid Fusion
• Implementation
  • RetrieverArm Class
  • Search Pipeline
  • Result Synthesis
• API Specification
  • Search Knowledge Base
  • Response Formats
• Data Models
• Configuration
• Performance Characteristics
• Testing
• Deployment
• See Also

Overview

The Retriever Arm performs hybrid search (vector + keyword) across knowledge bases, synthesizes information from multiple sources, and provides citations. It acts as the system's research specialist, combining dense and sparse retrieval methods for optimal recall and precision.

Key Features

• Hybrid Search: Combines vector (semantic) and keyword (lexical) search
• Dense Retrieval: Uses embeddings for semantic similarity
• Sparse Retrieval: Uses BM25 for keyword matching
• Reciprocal Rank Fusion: Intelligently merges search results
• Cross-Encoder Reranking: Improves result quality
• Information Synthesis: Generates coherent summaries with citations
• Multi-Source: Searches across multiple knowledge bases
• Configurable Filters: Filter by metadata, date, source, etc.

Design Principles

1. Best of Both Worlds: Combine semantic and lexical search
2. Rerank for Quality: Use cross-encoders for final ordering
3. Cite Everything: Provide source attribution
4. Fast by Default: <500ms for most queries
5. Scalable: Handle large corpora efficiently

Architecture

graph TB
    subgraph "Retriever Arm"
        API[API Endpoint]
        COORD[Search Coordinator]
        RERANK[Reranker]
        SYNTH[Synthesizer]
    end

    subgraph "Search Backends"
        QDRANT[Qdrant Vector DB]
        ES[Elasticsearch]
        ENCODER[Sentence Transformer]
    end

    subgraph "LLM Services"
        GPT[GPT-3.5 Turbo]
    end

    ORCH[Orchestrator] -->|Search Request| API
    API --> COORD

    COORD -->|Vector Search| ENCODER
    ENCODER -->|Query Embedding| QDRANT
    QDRANT -->|Vector Results| COORD

    COORD -->|Keyword Search| ES
    ES -->|Keyword Results| COORD

    COORD -->|Hybrid Fusion| COORD
    COORD -->|Fused Results| RERANK
    RERANK -->|Ranked Results| SYNTH

    SYNTH --> GPT
    GPT -->|Synthesis| SYNTH

    SYNTH -->|Search Response| API
    API -->|Results + Synthesis| ORCH

    style COORD fill:#ff9,stroke:#333
    style RERANK fill:#9ff,stroke:#333
    style GPT fill:#f9f,stroke:#333

Search Flow

sequenceDiagram
    participant O as Orchestrator
    participant R as Retriever
    participant V as Vector DB
    participant K as Keyword Engine
    participant RR as Reranker
    participant S as Synthesizer

    O->>R: Search request

    alt Vector Search
        R->>V: Search by embedding
        V-->>R: Vector results
    else Keyword Search
        R->>K: Search by keywords
        K-->>R: Keyword results
    else Hybrid Search
        par Vector + Keyword
            R->>V: Search by embedding
            V-->>R: Vector results
        and
            R->>K: Search by keywords
            K-->>R: Keyword results
        end
        R->>R: Fuse results (RRF)
    end

    R->>RR: Rerank results
    RR-->>R: Ranked results

    R->>R: Filter by min relevance
    R->>R: Limit results

    R->>S: Synthesize top results
    S-->>R: Synthesis + citations

    R-->>O: SearchResponse

Core Functionality

Search Methods

from enum import Enum

class SearchMethod(str, Enum):
    VECTOR = "vector"        # Dense retrieval (embeddings)
    KEYWORD = "keyword"      # Sparse retrieval (BM25)
    HYBRID = "hybrid"        # Fusion of both

Hybrid Search Strategy

Reciprocal Rank Fusion (RRF) combines results from multiple search methods:

RRF_score(d) = Σ (1 / (k + rank_i(d)))

Where:

• d is a document
• k is a constant (typically 60)
• rank_i(d) is the rank of document d in search method i

Reranking

After fusion, a cross-encoder reranks results based on query-document relevance:

class CrossEncoderReranker:
    """Rerank results using cross-encoder."""

    def __init__(self, model: str = "cross-encoder/ms-marco-MiniLM-L-6-v2"):
        self.model = CrossEncoder(model)

    async def rerank(self, query: str, results: List[SearchResult]) -> List[SearchResult]:
        """Rerank results by relevance."""

        if not results:
            return results

        # Prepare pairs for cross-encoder
        pairs = [(query, r.content) for r in results]

        # Score all pairs
        scores = self.model.predict(pairs)

        # Update relevance scores
        for result, score in zip(results, scores):
            result.relevance_score = float(score)

        # Sort by new scores
        results.sort(key=lambda x: x.relevance_score, reverse=True)

        # Update ranks
        for idx, result in enumerate(results):
            result.rank = idx + 1

        return results

Synthesis

Combines top results into a coherent summary with citations:

async def _synthesize_results(
    self,
    query: str,
    results: List[SearchResult]
) -> str:
    """Generate coherent synthesis from search results."""

    # Combine top results
    combined_content = "\n\n".join([
        f"Source {idx + 1} ({r.source}):\n{r.content}"
        for idx, r in enumerate(results[:5])
    ])

    synthesis_prompt = f"""Query: {query}

Retrieved information:
{combined_content}

Synthesize the above information into a coherent, accurate summary that directly answers the query. Include inline citations [1], [2], etc."""

    response = await openai.ChatCompletion.acreate(
        model="gpt-3.5-turbo",
        messages=[
            {"role": "system", "content": "You are a research assistant. Synthesize information accurately with citations."},
            {"role": "user", "content": synthesis_prompt}
        ],
        temperature=0.3,
        max_tokens=500
    )

    return response.choices[0].message.content

Search Implementations

Vector Search

Dense retrieval using semantic embeddings:

async def _vector_search(self, req: SearchRequest) -> List[SearchResult]:
    """Dense retrieval using vector embeddings."""

    # Encode query
    query_vector = self.encoder.encode(req.query).tolist()

    # Build filter
    search_filter = self._build_qdrant_filter(req.filters)

    # Search vector DB
    qdrant_results = self.vector_db.search(
        collection_name="knowledge_base",
        query_vector=query_vector,
        query_filter=search_filter,
        limit=req.limit * 2  # Get more for reranking
    )

    # Convert to SearchResult
    results = []
    for idx, hit in enumerate(qdrant_results):
        results.append(SearchResult(
            content=hit.payload["content"],
            source=hit.payload["source"],
            relevance_score=hit.score,
            rank=idx + 1,
            metadata=hit.payload.get("metadata", {})
        ))

    return results

Keyword Search

Sparse retrieval using BM25:

async def _keyword_search(self, req: SearchRequest) -> List[SearchResult]:
    """Sparse retrieval using BM25."""

    # Build Elasticsearch query
    es_query = {
        "query": {
            "bool": {
                "must": [
                    {"match": {"content": req.query}}
                ],
                "filter": self._build_es_filter(req.filters)
            }
        },
        "size": req.limit * 2
    }

    # Execute search
    es_results = await self.keyword_engine.search(
        index="knowledge_base",
        body=es_query
    )

    # Convert to SearchResult
    results = []
    for idx, hit in enumerate(es_results["hits"]["hits"]):
        results.append(SearchResult(
            content=hit["_source"]["content"],
            source=hit["_source"]["source"],
            relevance_score=hit["_score"] / 10.0,  # Normalize
            rank=idx + 1,
            metadata=hit["_source"].get("metadata", {})
        ))

    return results

Hybrid Fusion

Reciprocal Rank Fusion of vector and keyword results:

async def _hybrid_search(self, req: SearchRequest) -> List[SearchResult]:
    """Fusion of vector and keyword search."""

    # Perform both searches in parallel
    vector_results, keyword_results = await asyncio.gather(
        self._vector_search(req),
        self._keyword_search(req)
    )

    # Fusion: Reciprocal Rank Fusion (RRF)
    k = 60  # RRF constant
    fused_scores = {}

    # Add vector results
    for result in vector_results:
        key = result.source
        fused_scores[key] = fused_scores.get(key, 0) + 1 / (k + result.rank)

    # Add keyword results
    for result in keyword_results:
        key = result.source
        fused_scores[key] = fused_scores.get(key, 0) + 1 / (k + result.rank)

    # Combine and sort by fused score
    all_results = {r.source: r for r in vector_results + keyword_results}

    fused_results = []
    for source, score in sorted(fused_scores.items(), key=lambda x: x[1], reverse=True):
        result = all_results[source]
        result.relevance_score = score
        fused_results.append(result)

    # Update ranks
    for idx, result in enumerate(fused_results):
        result.rank = idx + 1

    return fused_results

Implementation

RetrieverArm Class

from typing import List, Dict, Any, Optional
from pydantic import BaseModel, Field
from qdrant_client import QdrantClient
from sentence_transformers import SentenceTransformer
from cross_encoder import CrossEncoder
import asyncio

class SearchRequest(BaseModel):
    query: str
    method: SearchMethod = SearchMethod.HYBRID
    limit: int = Field(10, ge=1, le=100)
    filters: Dict[str, Any] = Field(default_factory=dict)
    min_relevance_score: float = Field(0.5, ge=0.0, le=1.0)
    include_citations: bool = True

class SearchResult(BaseModel):
    content: str
    source: str
    relevance_score: float
    rank: int
    metadata: Dict[str, Any] = Field(default_factory=dict)

class SearchResponse(BaseModel):
    results: List[SearchResult]
    query: str
    method_used: SearchMethod
    total_results: int
    synthesis: Optional[str] = None
    citations: List[str] = Field(default_factory=list)

class RetrieverArm:
    """Knowledge search and synthesis specialist."""

    def __init__(
        self,
        vector_db_url: str = "http://qdrant:6333",
        elasticsearch_url: str = "http://elasticsearch:9200"
    ):
        self.vector_db = QdrantClient(url=vector_db_url)
        self.keyword_engine = ElasticsearchClient(url=elasticsearch_url)
        self.encoder = SentenceTransformer('all-MiniLM-L6-v2')
        self.reranker = CrossEncoderReranker()

    async def search(self, req: SearchRequest) -> SearchResponse:
        """Perform hybrid search across knowledge bases."""

        # Perform search based on method
        if req.method == SearchMethod.VECTOR:
            results = await self._vector_search(req)
        elif req.method == SearchMethod.KEYWORD:
            results = await self._keyword_search(req)
        else:  # HYBRID
            results = await self._hybrid_search(req)

        # Rerank results
        results = await self.reranker.rerank(req.query, results)

        # Filter by minimum relevance
        results = [r for r in results if r.relevance_score >= req.min_relevance_score]

        # Limit results
        results = results[:req.limit]

        # Generate synthesis
        synthesis = await self._synthesize_results(req.query, results) if results else None

        # Extract citations
        citations = [r.source for r in results] if req.include_citations else []

        return SearchResponse(
            results=results,
            query=req.query,
            method_used=req.method,
            total_results=len(results),
            synthesis=synthesis,
            citations=citations
        )

Search Pipeline

The complete search pipeline:

1. Query Analysis: Parse and understand the query
2. Parallel Search: Execute vector and/or keyword search
3. Result Fusion: Combine results using RRF (for hybrid)
4. Reranking: Apply cross-encoder for better ordering
5. Filtering: Remove low-relevance results
6. Limiting: Cap at requested limit
7. Synthesis: Generate summary with citations

Result Synthesis

FastAPI endpoint implementation:

from fastapi import FastAPI, HTTPException

app = FastAPI(title="Retriever Arm")
retriever = RetrieverArm()

@app.post("/search", response_model=SearchResponse)
async def search_knowledge_base(request: SearchRequest) -> SearchResponse:
    """Search knowledge base and synthesize results."""

    try:
        response = await retriever.search(request)
        return response
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/health")
async def health_check():
    """Health check endpoint."""
    return {
        "status": "healthy",
        "vector_db": await retriever.vector_db.get_collections(),
        "keyword_engine": "connected"
    }

API Specification

Search Knowledge Base

Endpoint: POST /search

Request Body:

{
  "query": "What are the benefits of hybrid search?",
  "method": "hybrid",
  "limit": 10,
  "filters": {
    "category": "search",
    "date_from": "2024-01-01"
  },
  "min_relevance_score": 0.5,
  "include_citations": true
}

Field Descriptions:

Response Formats

Successful Search (200 OK):

{
  "results": [
    {
      "content": "Hybrid search combines vector (semantic) and keyword (lexical) search methods. This approach leverages the strengths of both: semantic similarity from embeddings and exact matching from BM25. The result is higher recall and precision compared to using either method alone.",
      "source": "docs/search-methods.md",
      "relevance_score": 0.92,
      "rank": 1,
      "metadata": {
        "category": "search",
        "date": "2024-03-15",
        "author": "research-team"
      }
    },
    {
      "content": "Reciprocal Rank Fusion (RRF) is used to merge results from different search strategies. It assigns scores based on rank positions rather than raw relevance scores, which normalizes across different scoring functions.",
      "source": "docs/fusion-algorithms.md",
      "relevance_score": 0.87,
      "rank": 2,
      "metadata": {
        "category": "algorithms",
        "date": "2024-02-20"
      }
    }
  ],
  "query": "What are the benefits of hybrid search?",
  "method_used": "hybrid",
  "total_results": 2,
  "synthesis": "Hybrid search offers significant advantages by combining semantic and lexical search methods [1]. The key benefits include:\n\n1. **Higher Recall**: Captures both semantically similar and exact keyword matches\n2. **Better Precision**: Reciprocal Rank Fusion merges results effectively [2]\n3. **Robustness**: Works well across diverse query types\n4. **Complementary Strengths**: Semantic understanding + exact matching\n\nThis makes hybrid search ideal for general-purpose information retrieval systems.",
  "citations": [
    "docs/search-methods.md",
    "docs/fusion-algorithms.md"
  ]
}

No Results (200 OK):

{
  "results": [],
  "query": "nonexistent topic",
  "method_used": "hybrid",
  "total_results": 0,
  "synthesis": null,
  "citations": []
}

Data Models

Filter Building

def _build_qdrant_filter(self, filters: Dict[str, Any]):
    """Build Qdrant filter from dict."""
    from qdrant_client.models import Filter, FieldCondition, MatchValue

    conditions = []
    for key, value in filters.items():
        conditions.append(
            FieldCondition(
                key=key,
                match=MatchValue(value=value)
            )
        )

    return Filter(must=conditions) if conditions else None

def _build_es_filter(self, filters: Dict[str, Any]) -> List[Dict]:
    """Build Elasticsearch filter from dict."""
    return [
        {"term": {key: value}}
        for key, value in filters.items()
    ]

Configuration

Environment Variables

# Retriever Arm Configuration
RETRIEVER_PORT=8006
RETRIEVER_DEFAULT_METHOD=hybrid
RETRIEVER_DEFAULT_LIMIT=10
RETRIEVER_MIN_RELEVANCE=0.5

# Vector DB Configuration
QDRANT_URL=http://qdrant:6333
QDRANT_COLLECTION=knowledge_base
EMBEDDING_MODEL=all-MiniLM-L6-v2

# Keyword Engine Configuration
ELASTICSEARCH_URL=http://elasticsearch:9200
ELASTICSEARCH_INDEX=knowledge_base

# Reranker Configuration
RERANKER_MODEL=cross-encoder/ms-marco-MiniLM-L-6-v2
ENABLE_RERANKING=true

# Synthesis Configuration
ENABLE_SYNTHESIS=true
SYNTHESIS_MODEL=gpt-3.5-turbo
SYNTHESIS_MAX_TOKENS=500
SYNTHESIS_MAX_SOURCES=5

# Logging
LOG_LEVEL=info
LOG_QUERIES=true

Configuration File

config.yaml:

retriever_arm:
  port: 8006
  default_method: hybrid
  default_limit: 10
  min_relevance_score: 0.5

  vector_search:
    url: http://qdrant:6333
    collection: knowledge_base
    embedding_model: all-MiniLM-L6-v2
    embedding_dimension: 384

  keyword_search:
    url: http://elasticsearch:9200
    index: knowledge_base
    algorithm: bm25

  reranking:
    enabled: true
    model: cross-encoder/ms-marco-MiniLM-L-6-v2

  synthesis:
    enabled: true
    model: gpt-3.5-turbo
    max_tokens: 500
    max_sources: 5
    temperature: 0.3

  fusion:
    method: rrf
    k: 60

Performance Characteristics

Latency

Accuracy

Throughput

• Requests/Second: 100-200 (without synthesis)
• Requests/Second: 20-40 (with synthesis)
• Concurrent Searches: Up to 50
• Corpus Size: Scales to 10M+ documents

Testing

Unit Tests

import pytest
from retriever_arm import RetrieverArm, SearchRequest, SearchMethod

@pytest.fixture
async def retriever():
    return RetrieverArm()

@pytest.mark.asyncio
async def test_vector_search(retriever):
    request = SearchRequest(
        query="machine learning algorithms",
        method=SearchMethod.VECTOR,
        limit=5
    )

    response = await retriever.search(request)

    assert response.total_results > 0
    assert len(response.results) <= 5
    assert response.method_used == SearchMethod.VECTOR
    assert all(r.relevance_score > 0 for r in response.results)

@pytest.mark.asyncio
async def test_hybrid_search(retriever):
    request = SearchRequest(
        query="neural networks",
        method=SearchMethod.HYBRID,
        limit=10,
        min_relevance_score=0.6
    )

    response = await retriever.search(request)

    assert response.method_used == SearchMethod.HYBRID
    assert all(r.relevance_score >= 0.6 for r in response.results)
    # Results should be ranked
    scores = [r.relevance_score for r in response.results]
    assert scores == sorted(scores, reverse=True)

@pytest.mark.asyncio
async def test_synthesis(retriever):
    request = SearchRequest(
        query="benefits of vector databases",
        limit=5,
        include_citations=True
    )

    response = await retriever.search(request)

    if response.total_results > 0:
        assert response.synthesis is not None
        assert len(response.citations) > 0
        # Synthesis should include citations [1], [2], etc.
        assert any(f"[{i}]" in response.synthesis for i in range(1, len(response.citations) + 1))

Deployment

Dockerfile

FROM python:3.11-slim

WORKDIR /app

# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Download embedding model
RUN python -c "from sentence_transformers import SentenceTransformer; SentenceTransformer('all-MiniLM-L6-v2')"

# Copy application
COPY retriever_arm/ ./retriever_arm/

RUN useradd -m -u 1000 retriever && chown -R retriever:retriever /app
USER retriever

ENV PYTHONUNBUFFERED=1
EXPOSE 8006

CMD ["uvicorn", "retriever_arm.main:app", "--host", "0.0.0.0", "--port", "8006"]

Kubernetes Deployment

apiVersion: apps/v1
kind: Deployment
metadata:
  name: retriever-arm
  namespace: octollm
spec:
  replicas: 2
  selector:
    matchLabels:
      app: retriever-arm
  template:
    metadata:
      labels:
        app: retriever-arm
    spec:
      containers:
      - name: retriever
        image: octollm/retriever-arm:1.0
        ports:
        - containerPort: 8006
        env:
        - name: RETRIEVER_PORT
          value: "8006"
        - name: QDRANT_URL
          value: "http://qdrant:6333"
        - name: ELASTICSEARCH_URL
          value: "http://elasticsearch:9200"
        - name: OPENAI_API_KEY
          valueFrom:
            secretKeyRef:
              name: openai-credentials
              key: api-key
        resources:
          requests:
            memory: "512Mi"
            cpu: "250m"
          limits:
            memory: "2Gi"
            cpu: "1000m"
        livenessProbe:
          httpGet:
            path: /health
            port: 8006
          initialDelaySeconds: 30
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /health
            port: 8006
          initialDelaySeconds: 10
          periodSeconds: 5

---
apiVersion: v1
kind: Service
metadata:
  name: retriever-arm
  namespace: octollm
spec:
  selector:
    app: retriever-arm
  ports:
  - port: 8006
    targetPort: 8006
  type: ClusterIP

See Also

• Orchestrator Component - Coordinates searches
• Planner Arm - Plans multi-step research
• Coder Arm - Uses memory for code examples
• Memory Systems - Knowledge base architecture
• API Reference - Complete API documentation

Document Status: Phase 1 Complete Last Updated: 2025-11-10 Maintainer: OctoLLM Core Team Next Review: 2025-12-10

Coder Arm: Code Generation & Analysis

Components > Arms > Coder Arm

Version: 1.0 Technology: Python 3.11+ / FastAPI Cost Tier: 4 (High) Average Latency: 2-5 seconds Status: Phase 1 Complete

Table of Contents

• Overview
• Architecture
• Core Functionality
  • Code Request Types
  • Code Generation
  • Syntax Validation
  • Context-Aware Prompts
• Memory System
  • Local Episodic Memory
  • Solution Storage
  • Similarity Search
• Implementation
  • CoderArm Class
  • Request Processing
  • LLM Integration
• API Specification
  • Generate Code
  • Debug Code
  • Refactor Code
• Data Models
• Configuration
• Performance Characteristics
• Testing
• Deployment
• Supported Languages
• See Also

Overview

The Coder Arm is a specialized component that excels at code generation, debugging, refactoring, and static analysis across multiple programming languages. It leverages large language models (GPT-4) and maintains a local episodic memory of past solutions to improve future responses.

Key Features

• Multi-Language Support: Python, JavaScript, Go, Rust, Java, and more
• Multiple Operations: Generate, debug, refactor, analyze, test, explain, optimize
• Context-Aware: Uses past solutions and project context
• Syntax Validation: Automatic validation and error correction
• Episodic Memory: Stores and retrieves similar solutions
• High Confidence: Returns confidence scores and warnings
• Production-Ready: Follows language-specific best practices

Design Principles

1. Quality Over Speed: Prioritize correct, idiomatic code
2. Learn from Past: Use memory to improve over time
3. Validate Always: Check syntax before returning
4. Explain Clearly: Provide explanations and rationale
5. Handle Uncertainty: Return confidence scores and warnings

Architecture

graph TB
    subgraph "Coder Arm"
        API[API Endpoint]
        PROC[Request Processor]
        MEM[Memory Search]
        PROMPT[Prompt Builder]
        LLM[LLM Interface]
        VAL[Syntax Validator]
        STORE[Memory Storage]
    end

    subgraph "External Services"
        GPT[GPT-4 API]
        QDRANT[Qdrant Vector DB]
    end

    subgraph "Validation Tools"
        PY[Python AST]
        JS[ESLint]
        GO[gofmt]
        RUST[rustc]
    end

    ORCH[Orchestrator] -->|Code Request| API
    API --> PROC
    PROC --> MEM
    MEM --> QDRANT
    QDRANT -->|Similar Solutions| MEM

    MEM --> PROMPT
    PROC --> PROMPT
    PROMPT --> LLM
    LLM --> GPT
    GPT -->|Generated Code| LLM

    LLM --> VAL
    VAL --> PY
    VAL --> JS
    VAL --> GO
    VAL --> RUST

    VAL -->|Valid| STORE
    VAL -->|Invalid| LLM
    STORE --> QDRANT

    STORE -->|Code Response| API
    API -->|Result| ORCH

    style GPT fill:#f9f,stroke:#333
    style QDRANT fill:#9ff,stroke:#333
    style VAL fill:#ff9,stroke:#333

Code Generation Flow

sequenceDiagram
    participant O as Orchestrator
    participant C as Coder Arm
    participant M as Memory
    participant L as LLM (GPT-4)
    participant V as Validator

    O->>C: Code Request

    C->>M: Search similar solutions
    M-->>C: Past solutions (0-3)

    C->>C: Build context prompt
    C->>L: Generate code

    L-->>C: Generated code

    C->>V: Validate syntax

    alt Syntax Valid
        V-->>C: Valid
        C->>M: Store solution
        C-->>O: Code Response (success)
    else Syntax Invalid
        V-->>C: Errors
        C->>L: Fix syntax errors
        L-->>C: Fixed code
        C->>V: Re-validate
        alt Fixed
            V-->>C: Valid
            C->>M: Store solution
            C-->>O: Code Response (success)
        else Still Invalid
            V-->>C: Still invalid
            C-->>O: Code Response (error)
        end
    end

Core Functionality

Code Request Types

from enum import Enum

class CodeRequestType(str, Enum):
    GENERATE = "generate"      # Create new code from scratch
    DEBUG = "debug"            # Find and fix bugs
    REFACTOR = "refactor"      # Improve code structure
    ANALYZE = "analyze"        # Static analysis
    TEST = "test"              # Generate unit tests
    EXPLAIN = "explain"        # Explain code behavior
    OPTIMIZE = "optimize"      # Performance optimization

Code Generation

The Coder Arm generates code through a multi-step process:

1. Memory Search: Find similar past solutions
2. Prompt Building: Create context-aware prompt with constraints
3. LLM Generation: Generate code using GPT-4
4. Syntax Validation: Check for syntax errors
5. Error Correction: Attempt to fix invalid syntax
6. Memory Storage: Store successful solution

class CoderArm:
    """Code generation and analysis specialist."""

    def __init__(self, llm_model: str = "gpt-4"):
        self.model = llm_model
        self.memory = CoderMemory()  # Local episodic memory
        self.validators = CodeValidators()

    async def process_request(self, req: CodeRequest) -> CodeResponse:
        """Process code request based on type."""

        # Check memory for similar past solutions
        similar = await self.memory.search_similar(
            req.instruction,
            language=req.language,
            limit=3
        )

        # Build context-aware prompt
        prompt = self._build_prompt(req, similar)

        # Generate code using LLM
        code_result = await self._generate_code(prompt, req)

        # Validate syntax
        validation = await self.validators.validate_syntax(
            code_result["code"],
            req.language
        )

        if not validation.valid:
            # Attempt to fix syntax errors
            code_result = await self._fix_syntax(code_result, validation)

        # Store in memory for future reference
        await self.memory.store_solution(
            instruction=req.instruction,
            code=code_result["code"],
            language=req.language,
            metadata=code_result.get("metadata", {})
        )

        return CodeResponse(**code_result)

Syntax Validation

Language-specific validators check generated code:

class CodeValidators:
    """Syntax validators for multiple languages."""

    async def validate_syntax(self, code: str, language: str) -> ValidationResult:
        """Validate syntax for given language."""

        validators = {
            "python": self._validate_python,
            "javascript": self._validate_javascript,
            "typescript": self._validate_typescript,
            "go": self._validate_go,
            "rust": self._validate_rust,
            "java": self._validate_java,
        }

        validator = validators.get(language.lower())
        if not validator:
            return ValidationResult(valid=True, message="No validator for language")

        return await validator(code)

    async def _validate_python(self, code: str) -> ValidationResult:
        """Validate Python code using AST."""
        import ast
        try:
            ast.parse(code)
            return ValidationResult(valid=True, message="Valid Python")
        except SyntaxError as e:
            return ValidationResult(
                valid=False,
                message=f"Syntax error: {e}",
                line=e.lineno,
                column=e.offset
            )

Context-Aware Prompts

Prompts include constraints, existing code, and similar solutions:

def _build_prompt(self, req: CodeRequest, similar_solutions: List[Dict]) -> str:
    """Build context-aware prompt."""

    base_prompt = f"""You are an expert {req.language} programmer.

Task: {req.request_type.value}
Instruction: {req.instruction}

Language: {req.language}
Constraints:
{chr(10).join(f"- {c}" for c in req.constraints) if req.constraints else "None"}"""

    if req.existing_code:
        base_prompt += f"\n\nExisting code:\n```{req.language}\n{req.existing_code}\n```"

    if similar_solutions:
        base_prompt += "\n\nSimilar past solutions for reference:"
        for idx, sol in enumerate(similar_solutions, 1):
            base_prompt += f"\n{idx}. {sol['description']}\n```{sol['language']}\n{sol['code'][:200]}...\n```"

    base_prompt += """

Requirements:
1. Write clean, idiomatic code following best practices
2. Include helpful comments for complex logic
3. Handle edge cases and errors appropriately
4. Follow the language's style guide (PEP 8, Go fmt, etc.)
5. Ensure code is production-ready

Output format:
```json
{
  "code": "// Full code here",
  "explanation": "Brief explanation of approach and key decisions",
  "confidence": 0.85,
  "warnings": ["Any caveats or limitations"],
  "tests": "// Optional test code if requested"
}
```"""

    return base_prompt

Memory System

Local Episodic Memory

The Coder Arm maintains a local vector database of past code solutions using Qdrant.

from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct
from sentence_transformers import SentenceTransformer

class CoderMemory:
    """Local episodic memory for code solutions."""

    def __init__(self, qdrant_url: str = "http://qdrant:6333"):
        self.client = QdrantClient(url=qdrant_url)
        self.collection = "coder_memory"
        self.encoder = SentenceTransformer('all-MiniLM-L6-v2')
        self._init_collection()

    def _init_collection(self):
        """Initialize Qdrant collection."""
        try:
            self.client.create_collection(
                collection_name=self.collection,
                vectors_config=VectorParams(
                    size=384,  # all-MiniLM-L6-v2 dimension
                    distance=Distance.COSINE
                )
            )
        except Exception:
            pass  # Collection already exists

Solution Storage

Solutions are stored with embeddings for semantic search:

async def store_solution(
    self,
    instruction: str,
    code: str,
    language: str,
    metadata: Dict[str, Any]
) -> str:
    """Store code solution in memory."""

    # Create embedding from instruction + code snippet
    text_for_embedding = f"{instruction}\n{code[:500]}"
    embedding = self.encoder.encode(text_for_embedding).tolist()

    point_id = str(uuid.uuid4())

    self.client.upsert(
        collection_name=self.collection,
        points=[
            PointStruct(
                id=point_id,
                vector=embedding,
                payload={
                    "instruction": instruction,
                    "code": code,
                    "language": language,
                    "created_at": datetime.utcnow().isoformat(),
                    **metadata
                }
            )
        ]
    )

    return point_id

Similarity Search

Find similar solutions using vector similarity:

async def search_similar(
    self,
    query: str,
    language: Optional[str] = None,
    limit: int = 5
) -> List[Dict[str, Any]]:
    """Search for similar code solutions."""

    query_vector = self.encoder.encode(query).tolist()

    # Build filter
    search_filter = None
    if language:
        from qdrant_client.models import Filter, FieldCondition, MatchValue
        search_filter = Filter(
            must=[
                FieldCondition(
                    key="language",
                    match=MatchValue(value=language)
                )
            ]
        )

    results = self.client.search(
        collection_name=self.collection,
        query_vector=query_vector,
        query_filter=search_filter,
        limit=limit
    )

    return [
        {
            "description": r.payload["instruction"],
            "code": r.payload["code"],
            "language": r.payload["language"],
            "score": r.score,
            "created_at": r.payload["created_at"]
        }
        for r in results
    ]

Implementation

CoderArm Class

Full implementation with LLM integration:

from typing import List, Dict, Any, Optional
from pydantic import BaseModel, Field
import openai
import json
import uuid
from datetime import datetime

class CoderArm:
    """Code generation and analysis specialist."""

    def __init__(self, llm_model: str = "gpt-4"):
        self.model = llm_model
        self.memory = CoderMemory()
        self.validators = CodeValidators()

    async def _generate_code(self, prompt: str, req: CodeRequest) -> Dict[str, Any]:
        """Generate code using LLM."""

        response = await openai.ChatCompletion.acreate(
            model=self.model,
            messages=[
                {"role": "system", "content": f"You are an expert {req.language} programmer."},
                {"role": "user", "content": prompt}
            ],
            temperature=0.2 if req.request_type == "generate" else 0.1,
            max_tokens=4000
        )

        content = response.choices[0].message.content

        # Extract JSON from response
        if "```json" in content:
            json_str = content.split("```json")[1].split("```")[0]
        else:
            json_str = content

        result = json.loads(json_str)
        result["language"] = req.language
        result["success"] = True

        return result

    async def _fix_syntax(self, code_result: Dict, validation: ValidationResult) -> Dict:
        """Attempt to fix syntax errors."""

        fix_prompt = f"""The following code has syntax errors:

```{code_result['language']}
{code_result['code']}

Error: {validation.message} Line {validation.line}, Column {validation.column}

Please fix the syntax error and return the corrected code in the same JSON format."""

    response = await openai.ChatCompletion.acreate(
        model=self.model,
        messages=[
            {"role": "system", "content": f"You are an expert {code_result['language']} programmer."},
            {"role": "user", "content": fix_prompt}
        ],
        temperature=0.1,
        max_tokens=4000
    )

    content = response.choices[0].message.content

    if "```json" in content:
        json_str = content.split("```json")[1].split("```")[0]
    else:
        json_str = content

    fixed_result = json.loads(json_str)
    fixed_result["language"] = code_result["language"]
    fixed_result["success"] = True

    return fixed_result

### Request Processing

FastAPI endpoint implementation:

```python
from fastapi import FastAPI, HTTPException

app = FastAPI(title="Coder Arm")
coder = CoderArm()

@app.post("/code", response_model=CodeResponse)
async def generate_code(request: CodeRequest) -> CodeResponse:
    """Generate, debug, or refactor code."""

    try:
        response = await coder.process_request(request)
        return response
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/health")
async def health_check():
    """Health check endpoint."""
    return {"status": "healthy", "model": coder.model}

@app.get("/memory/stats")
async def memory_stats():
    """Get memory statistics."""
    collection_info = coder.memory.client.get_collection(coder.memory.collection)
    return {
        "total_solutions": collection_info.points_count,
        "vector_dimension": collection_info.config.params.vectors.size
    }

LLM Integration

OpenAI API integration with error handling:

async def call_llm_with_retry(
    messages: List[Dict],
    model: str,
    max_retries: int = 3
) -> str:
    """Call LLM with exponential backoff retry."""

    for attempt in range(max_retries):
        try:
            response = await openai.ChatCompletion.acreate(
                model=model,
                messages=messages,
                temperature=0.2,
                max_tokens=4000,
                timeout=30
            )
            return response.choices[0].message.content

        except openai.error.RateLimitError:
            wait_time = 2 ** attempt
            await asyncio.sleep(wait_time)

        except openai.error.APIError as e:
            if attempt == max_retries - 1:
                raise
            await asyncio.sleep(1)

    raise Exception("Max retries exceeded")

API Specification

Generate Code

Endpoint: POST /code

Request Body:

{
  "request_type": "generate",
  "language": "python",
  "instruction": "Create a function that validates email addresses using regex",
  "context": {
    "project_type": "web_api",
    "framework": "fastapi"
  },
  "constraints": [
    "Must support RFC 5322 standard",
    "Include docstring with examples",
    "Add type hints"
  ]
}

Response (200 OK):

{
  "success": true,
  "code": "import re\nfrom typing import Optional\n\ndef validate_email(email: str) -> bool:\n    \"\"\"Validate email address using RFC 5322 regex.\n    \n    Args:\n        email: Email address to validate\n    \n    Returns:\n        True if valid, False otherwise\n    \n    Examples:\n        >>> validate_email('user@example.com')\n        True\n        >>> validate_email('invalid.email')\n        False\n    \"\"\"\n    pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\\.[a-zA-Z]{2,}$'\n    return bool(re.match(pattern, email))",
  "explanation": "Created a simple email validator using regex. The pattern matches standard email formats per RFC 5322. Includes type hints and comprehensive docstring with examples.",
  "language": "python",
  "tests": "import pytest\n\ndef test_validate_email_valid():\n    assert validate_email('user@example.com') == True\n\ndef test_validate_email_invalid():\n    assert validate_email('invalid') == False",
  "confidence": 0.92,
  "warnings": [
    "Regex validation is not 100% RFC 5322 compliant - consider using email-validator library for production"
  ],
  "metadata": {
    "model": "gpt-4",
    "tokens_used": 450,
    "memory_hits": 2
  }
}

Debug Code

Request Body:

{
  "request_type": "debug",
  "language": "python",
  "instruction": "Fix the bug causing IndexError",
  "existing_code": "def get_item(items, index):\n    return items[index]\n\nresult = get_item([1, 2, 3], 5)",
  "constraints": [
    "Add proper error handling",
    "Return None for invalid indices"
  ]
}

Response:

{
  "success": true,
  "code": "def get_item(items, index):\n    \"\"\"Get item at index, returning None if invalid.\"\"\"\n    try:\n        return items[index]\n    except IndexError:\n        return None\n\nresult = get_item([1, 2, 3], 5)  # Returns None",
  "explanation": "Added try-except block to handle IndexError. Function now returns None for invalid indices instead of raising exception.",
  "language": "python",
  "confidence": 0.95,
  "warnings": [],
  "metadata": {
    "bug_type": "IndexError",
    "fix_applied": "exception_handling"
  }
}

Refactor Code

Request Body:

{
  "request_type": "refactor",
  "language": "javascript",
  "instruction": "Refactor to use async/await instead of callbacks",
  "existing_code": "function fetchData(url, callback) {\n  fetch(url)\n    .then(res => res.json())\n    .then(data => callback(null, data))\n    .catch(err => callback(err, null));\n}"
}

Response:

{
  "success": true,
  "code": "async function fetchData(url) {\n  try {\n    const response = await fetch(url);\n    const data = await response.json();\n    return data;\n  } catch (error) {\n    throw error;\n  }\n}",
  "explanation": "Converted callback-based function to async/await for cleaner error handling and better readability. Removed callback parameter and use direct return/throw.",
  "language": "javascript",
  "confidence": 0.94,
  "warnings": [
    "Callers must now use try-catch or .catch() when calling this function"
  ],
  "metadata": {
    "refactor_type": "callback_to_async"
  }
}

Data Models

Request Model

class CodeRequest(BaseModel):
    request_type: CodeRequestType
    language: str = Field(..., description="Programming language")
    instruction: str = Field(..., description="What to do")
    context: Dict[str, Any] = Field(default_factory=dict)
    existing_code: Optional[str] = None
    constraints: List[str] = Field(default_factory=list)

    class Config:
        schema_extra = {
            "example": {
                "request_type": "generate",
                "language": "python",
                "instruction": "Create a binary search function",
                "context": {"data_structure": "sorted_list"},
                "constraints": ["Use iterative approach", "Add type hints"]
            }
        }

Response Model

class CodeResponse(BaseModel):
    success: bool
    code: str = Field(..., description="Generated/modified code")
    explanation: str
    language: str
    tests: Optional[str] = None
    confidence: float = Field(..., ge=0.0, le=1.0)
    warnings: List[str] = Field(default_factory=list)
    metadata: Dict[str, Any] = Field(default_factory=dict)

Validation Result

class ValidationResult(BaseModel):
    valid: bool
    message: str
    line: Optional[int] = None
    column: Optional[int] = None
    suggestions: List[str] = Field(default_factory=list)

Configuration

Environment Variables

# Coder Arm Configuration
CODER_PORT=8004
CODER_MODEL=gpt-4  # or gpt-3.5-turbo for lower cost
CODER_TEMPERATURE=0.2
CODER_MAX_TOKENS=4000

# Memory Configuration
QDRANT_URL=http://qdrant:6333
CODER_MEMORY_COLLECTION=coder_memory
MEMORY_MAX_SOLUTIONS=10000

# OpenAI Configuration
OPENAI_API_KEY=sk-...
OPENAI_ORG_ID=org-...

# Validation
ENABLE_SYNTAX_VALIDATION=true
AUTO_FIX_SYNTAX=true
MAX_FIX_ATTEMPTS=2

# Logging
LOG_LEVEL=info
LOG_CODE_SAMPLES=true
LOG_PROMPTS=false  # Sensitive, disable in prod

Configuration File

config.yaml:

coder_arm:
  model: gpt-4
  temperature: 0.2
  max_tokens: 4000

  # Memory settings
  memory:
    backend: qdrant
    collection: coder_memory
    max_solutions: 10000
    embedding_model: all-MiniLM-L6-v2

  # Validation
  validation:
    enabled: true
    auto_fix: true
    max_attempts: 2

    validators:
      python:
        enabled: true
        linter: pylint
      javascript:
        enabled: true
        linter: eslint
      go:
        enabled: true
        formatter: gofmt

  # Supported languages
  languages:
    - python
    - javascript
    - typescript
    - go
    - rust
    - java
    - cpp
    - csharp

Performance Characteristics

Latency

Cost

• GPT-4 Usage: ~2,000 tokens per request (input + output)
• Monthly Cost: $0.06 per request (GPT-4 pricing)
• Memory Storage: ~1 KB per solution
• Total Cost: Tier 4 (High)

Accuracy

• Syntax Valid: 88% first attempt, 95% after fix
• Functionally Correct: 75-85% (varies by complexity)
• Best Practices: 80% compliance
• Memory Hits: 30-40% of requests find similar solutions

Testing

Unit Tests

import pytest
from coder_arm import CoderArm, CodeRequest, CodeRequestType

@pytest.fixture
async def coder():
    return CoderArm(llm_model="gpt-3.5-turbo")

@pytest.mark.asyncio
async def test_generate_python_function(coder):
    request = CodeRequest(
        request_type=CodeRequestType.GENERATE,
        language="python",
        instruction="Create a fibonacci function",
        constraints=["Use recursion", "Add docstring"]
    )

    response = await coder.process_request(request)

    assert response.success
    assert "def" in response.code
    assert response.language == "python"
    assert response.confidence > 0.7

@pytest.mark.asyncio
async def test_syntax_validation(coder):
    code = "def invalid_function(\n    print('missing closing paren')"

    validation = await coder.validators.validate_syntax(code, "python")

    assert not validation.valid
    assert "SyntaxError" in validation.message

@pytest.mark.asyncio
async def test_memory_storage(coder):
    solution_id = await coder.memory.store_solution(
        instruction="Test function",
        code="def test(): pass",
        language="python",
        metadata={}
    )

    assert solution_id is not None

    results = await coder.memory.search_similar("Test function", language="python")
    assert len(results) > 0
    assert results[0]["code"] == "def test(): pass"

Integration Tests

@pytest.mark.asyncio
async def test_end_to_end_generation(coder):
    """Test full generation pipeline."""

    request = CodeRequest(
        request_type=CodeRequestType.GENERATE,
        language="python",
        instruction="Binary search in sorted array",
        constraints=["Iterative", "Type hints", "Docstring"]
    )

    response = await coder.process_request(request)

    # Verify response
    assert response.success
    assert "def" in response.code
    assert "binary" in response.code.lower()

    # Verify syntax validity
    import ast
    ast.parse(response.code)  # Should not raise

    # Verify memory stored
    similar = await coder.memory.search_similar(
        "Binary search",
        language="python",
        limit=1
    )
    assert len(similar) > 0

Deployment

Dockerfile

FROM python:3.11-slim

WORKDIR /app

# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Install syntax validators
RUN pip install pylint eslint-py

# Copy application
COPY coder_arm/ ./coder_arm/
COPY config.yaml .

# Non-root user
RUN useradd -m -u 1000 coder && chown -R coder:coder /app
USER coder

# Environment
ENV PYTHONUNBUFFERED=1
ENV LOG_LEVEL=info

EXPOSE 8004

CMD ["uvicorn", "coder_arm.main:app", "--host", "0.0.0.0", "--port", "8004"]

Kubernetes Deployment

apiVersion: apps/v1
kind: Deployment
metadata:
  name: coder-arm
  namespace: octollm
spec:
  replicas: 2
  selector:
    matchLabels:
      app: coder-arm
  template:
    metadata:
      labels:
        app: coder-arm
    spec:
      containers:
      - name: coder
        image: octollm/coder-arm:1.0

        ports:
        - containerPort: 8004
          name: http

        env:
        - name: CODER_PORT
          value: "8004"
        - name: CODER_MODEL
          value: "gpt-4"
        - name: QDRANT_URL
          value: "http://qdrant:6333"
        - name: OPENAI_API_KEY
          valueFrom:
            secretKeyRef:
              name: openai-credentials
              key: api-key

        resources:
          requests:
            memory: "512Mi"
            cpu: "250m"
          limits:
            memory: "2Gi"
            cpu: "1000m"

        livenessProbe:
          httpGet:
            path: /health
            port: 8004
          initialDelaySeconds: 30
          periodSeconds: 10

        readinessProbe:
          httpGet:
            path: /health
            port: 8004
          initialDelaySeconds: 10
          periodSeconds: 5

---
apiVersion: v1
kind: Service
metadata:
  name: coder-arm
  namespace: octollm
spec:
  selector:
    app: coder-arm
  ports:
  - port: 8004
    targetPort: 8004
    name: http
  type: ClusterIP

Supported Languages

See Also

• Orchestrator Component - Task coordination
• Planner Arm - Task decomposition
• Executor Arm - Code execution
• Judge Arm - Code quality validation
• Memory Systems - Memory architecture
• API Reference - Complete API documentation

Document Status: Phase 1 Complete Last Updated: 2025-11-10 Maintainer: OctoLLM Core Team Next Review: 2025-12-10

Judge Arm: Validation & Quality Assurance

Components > Arms > Judge Arm

Version: 1.0 Technology: Python 3.11+ / FastAPI Cost Tier: 2 (Medium) Average Latency: 0.5-2 seconds Status: Phase 1 Complete

Table of Contents

• Overview
• Architecture
• Core Functionality
  • Validation Types
  • Multi-Layer Validation
  • Acceptance Criteria Checking
  • Hallucination Detection
• Validation Layers
  • Layer 1: Schema Validation
  • Layer 2: Fact-Checking
  • Layer 3: Criteria Validation
  • Layer 4: Hallucination Detection
  • Layer 5: Quality Assessment
• Implementation
  • JudgeArm Class
  • Schema Validator
  • Fact Checker
  • Quality Assessor
• API Specification
  • Validate Output
  • Response Formats
• Data Models
• Configuration
• Performance Characteristics
• Testing
• Deployment
• See Also