# Psyche Orchestrator

Session coordination and pipeline management for the Psyche AI platform.

Part of the Psyche AI Virtual Assistant Platform.

## Overview

The Orchestrator is the central coordination service for Psyche, managing
session lifecycles, resource allocation, real-time pipelines, and multi-party
communication.

## Features

- **Session Lifecycle**: Create, pause, resume, terminate sessions
- **State Management**: Conversation, emotion, tool, screen share state
- **Resource Allocation**: GPU/CPU/memory with pooling and scaling
- **Pipeline Routing**: Real-time audio/video/data streams
- **Multi-Party Support**: Participants, turns, synchronization
- **Semantic Caching**: LLM response caching with vector similarity
- **Batch Processing**: Intelligent LLM request batching

## Architecture

```
orchestrator/
├── session/              # Session management
│   ├── types.py          # Session, SessionStatus, SessionConfig
│   ├── lifecycle.py      # SessionFactory, TimeoutManager, Handoff
│   └── state.py          # State persistence, checkpoints
├── resources/            # Resource allocation
│   ├── types.py          # ResourceType, AllocationStrategy
│   ├── allocator.py      # GPU/CPU/memory allocation
│   ├── pool.py           # Resource pooling and scaling
│   └── monitor.py        # Resource health monitoring
├── pipeline/             # Real-time pipelines
│   ├── types.py          # Stream, Pipeline, Message types
│   ├── router.py         # PipelineRouter, StreamRouter
│   └── communication.py  # gRPC, WebSocket, events
├── multiparty/           # Multi-party sessions
│   ├── types.py          # Participant, Turn, Sync types
│   ├── participants.py   # ParticipantManager, permissions
│   ├── turns.py          # TurnManager, interruptions
│   └── sync.py           # StateSynchronizer, conflict resolution
├── caching/              # LLM response caching
│   ├── semantic_cache.py # Vector-based similarity caching
│   └── embedding_service.py
└── batching/             # Batch processing
    └── batch_processor.py
```

## Session Lifecycle

```
                    ┌───────────────┐
                    │   CREATED     │
                    └───────┬───────┘
                            │ start()
                    ┌───────▼───────┐
              ┌─────│    ACTIVE     │─────┐
              │     └───────┬───────┘     │
         pause()            │         terminate()
              │             │ idle timeout  │
       ┌──────▼──────┐      │         ┌────▼────┐
       │   PAUSED    │      │         │ ENDED   │
       └──────┬──────┘      │         └─────────┘
              │             │
         resume()      handoff()
              │             │
              └─────►──┬────▼────┐
                       │ HANDED_OFF │
                       └──────────┘
```

## Session States

| State      | Description                             |
| ---------- | --------------------------------------- |
| CREATED    | Session initialized, not yet active     |
| ACTIVE     | Session running, processing requests    |
| PAUSED     | Session paused, state preserved         |
| HANDED_OFF | Session transferred to another instance |
| ENDED      | Session terminated, resources released  |

## Resource Management

### Allocation Strategies

| Strategy       | Description                |
| -------------- | -------------------------- |
| FIRST_FIT      | First available resource   |
| BEST_FIT       | Smallest adequate resource |
| WORST_FIT      | Largest available resource |
| ROUND_ROBIN    | Cycle through resources    |
| LEAST_LOADED   | Least utilized resource    |
| PRIORITY_BASED | Based on session priority  |

### Scaling Modes

| Mode       | Description                 |
| ---------- | --------------------------- |
| MANUAL     | Operator-controlled scaling |
| AUTO       | Utilization-based scaling   |
| SCHEDULED  | Time-based scaling          |
| PREDICTIVE | ML-based demand prediction  |

## Pipeline Routing

### Stream Types

| Type    | Description             |
| ------- | ----------------------- |
| AUDIO   | Audio data streams      |
| VIDEO   | Video frame streams     |
| DATA    | Arbitrary data messages |
| CONTROL | Control signals         |
| EVENTS  | Event notifications     |

### Pipeline Stages

```
INPUT → PREPROCESSING → PROCESSING → POSTPROCESSING → OUTPUT
```

## Multi-Party Support

### Participant Roles

| Role         | Description                   |
| ------------ | ----------------------------- |
| HOST         | Full control, can end session |
| MODERATOR    | Manage participants           |
| SPEAKER      | Can speak, share screen       |
| PARTICIPANT  | Limited speaking              |
| VIEWER       | View only                     |
| AI_ASSISTANT | Psyche AI avatar              |

### Turn Management

- **Speaking Queue**: FIFO with priority override
- **Interruption Handling**: Configurable policies
- **Silence Detection**: Auto-yield after silence

## Quick Start

### Installation

```bash
# Using Nx
nx install psyche-orchestrator

# With all optional dependencies
nx install-all psyche-orchestrator

# Or directly with Poetry
cd apps/psyche/orchestrator
poetry install
poetry install --extras all
```

### Environment Variables

```bash
# Database
DATABASE_URL=postgresql://user:pass@localhost:5432/psyche_orchestrator

# Redis (state management)
REDIS_URL=redis://localhost:6379

# Service Config
ORCHESTRATOR_PORT=8007
ORCHESTRATOR_HOST=0.0.0.0
LOG_LEVEL=INFO

# Session Defaults
SESSION_IDLE_TIMEOUT=300
SESSION_MAX_DURATION=3600
SESSION_CONNECTION_TIMEOUT=30

# gRPC
GRPC_PORT=50051
```

### Basic Usage

```python
from orchestrator import (
    get_lifecycle_manager,
    get_allocator,
    get_pipeline_router,
    CreateSessionRequest,
    SessionConfig,
    SessionType,
    ResourceRequest,
    ResourceType,
)

# Create session
lifecycle = get_lifecycle_manager()
request = CreateSessionRequest(
    user_id="user-123",
    config=SessionConfig(session_type=SessionType.VIDEO_CALL),
)
session = await lifecycle.create_session(request)
session = await lifecycle.start_session(session)

# Allocate GPU
allocator = get_allocator()
allocation = await allocator.allocate(ResourceRequest(
    session_id=session.session_id,
    resource_type=ResourceType.GPU,
    gpu_count=1,
))

# Start pipeline
pipeline = get_pipeline_router()
await pipeline.start()

# Process messages
async for message in pipeline.receive():
    await process_message(message)
```

## API Endpoints

### Sessions

- `POST /sessions` - Create session
- `GET /sessions/{id}` - Get session
- `POST /sessions/{id}/start` - Start session
- `POST /sessions/{id}/pause` - Pause session
- `POST /sessions/{id}/resume` - Resume session
- `DELETE /sessions/{id}` - Terminate session

### Resources

- `POST /resources/allocate` - Allocate resources
- `DELETE /resources/{id}` - Release allocation
- `GET /resources/pools` - List resource pools
- `GET /resources/utilization` - Get utilization metrics

### Pipeline

- `WS /pipeline/ws` - WebSocket pipeline connection
- `POST /pipeline/message` - Send message
- `GET /pipeline/status` - Get pipeline status

### Multi-Party

- `POST /participants/join` - Join session
- `POST /participants/leave` - Leave session
- `POST /turns/request` - Request speaking turn
- `POST /turns/yield` - Yield speaking turn

### Health

- `GET /health` - Health check
- `GET /health/live` - Liveness probe
- `GET /health/ready` - Readiness probe
- `GET /metrics` - Prometheus metrics

## Development

### Running the Server

```bash
# Development mode
nx serve psyche-orchestrator

# Production mode
nx serve-prod psyche-orchestrator
```

### Database Migrations

```bash
# Run migrations
nx migrate psyche-orchestrator

# Create new migration
nx migrate-create psyche-orchestrator "add_session_metadata"
```

### Running Tests

```bash
nx test psyche-orchestrator
nx test-unit psyche-orchestrator
nx test-integration psyche-orchestrator
nx test-cov psyche-orchestrator
```

### Docker

```bash
# Build image
nx docker-build psyche-orchestrator

# Run container
nx docker-run psyche-orchestrator
```

## Nx Integration

```bash
# Available targets
nx serve psyche-orchestrator          # Development server
nx serve-prod psyche-orchestrator     # Production server (4 workers)
nx build psyche-orchestrator          # Build package
nx install psyche-orchestrator        # Install dependencies
nx install-all psyche-orchestrator    # Install with extras
nx lint psyche-orchestrator           # Run linters
nx format psyche-orchestrator         # Format code
nx test psyche-orchestrator           # Run all tests
nx test-unit psyche-orchestrator      # Unit tests only
nx test-integration psyche-orchestrator  # Integration tests
nx test-cov psyche-orchestrator       # Tests with coverage
nx migrate psyche-orchestrator        # Run DB migrations
nx docker-build psyche-orchestrator   # Build Docker image
nx docker-run psyche-orchestrator     # Run container
```

## Performance

### Timeouts

| Timeout      | Default | Description            |
| ------------ | ------- | ---------------------- |
| Idle         | 5 min   | Session inactivity     |
| Max Duration | 1 hour  | Maximum session length |
| Connection   | 30s     | Client connection      |

### Resource Defaults

- **CPU**: 500m request, 2000m limit
- **Memory**: 1Gi request, 4Gi limit
- **Workers**: 4 (configurable)

## Security

- **Non-root execution**: psyche user (UID 1000)
- **Read-only root filesystem**: Immutable container
- **No privilege escalation**: Capabilities dropped
- **State encryption**: Redis encryption at rest
- **Audit logging**: All session actions logged

## License

Proprietary - Oshun Platform