ATOM Documentation

┌─────────────────────────────────────────────────────────────┐
│                     Frontend (Next.js)                       │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐     │
│  │   Proposal   │  │ Supervision  │  │   Training   │     │
│  │ Management   │  │  Dashboard   │  │   Sessions   │     │
│  └──────┬───────┘  └──────┬───────┘  └──────┬───────┘     │
└─────────┼─────────────────┼─────────────────┼──────────────┘
          │                 │                 │
          │ WebSocket       │ REST API        │ REST API
          │                 │                 │
┌─────────┼─────────────────┼─────────────────┼──────────────┐
│         │         Backend (FastAPI)          │              │
│  ┌──────▼───────┐  ┌──────┴───────┐  ┌─────┴────────┐   │
│  │     API      │  │   Training   │  │   Metrics    │   │
│  │   Routes     │  │   Services   │  │  Collector   │   │
│  └──────┬───────┘  └──────┬───────┘  └─────┬────────┘   │
│         │                 │                 │              │
│  ┌──────▼─────────────────▼─────────────────▼────────┐   │
│  │              Database (PostgreSQL)                │   │
│  │  - AgentProposal                                 │   │
│  │  - SupervisionSession                            │   │
│  │  - TrainingSession                               │   │
│  │  - TrainingMetrics                               │   │
│  └──────────────────────────────────────────────────┘   │
│                                                           │
│  ┌──────────────────────────────────────────────────┐   │
│  │              Cache (Redis)                       │   │
│  │  - Dashboard data (5-min TTL)                   │   │
│  │  - Metrics cache (1-hour TTL)                   │   │
│  │  - Alert deduplication (10-min TTL)             │   │
│  └──────────────────────────────────────────────────┘   │
└───────────────────────────────────────────────────────────┘

# Training System Features
TRAINING_SYSTEM_ENABLED=true
TRAINING_ANALYTICS_ENABLED=true
TRAINING_ALERTS_ENABLED=true
TRAINING_DASHBOARD_ENABLED=true

# WebSocket Configuration
SUPERVISION_WEBSOCKET_TIMEOUT=3600000  # 1 hour
WEBSOCKET_HEARTBEAT_INTERVAL=30000     # 30 seconds

# Metrics Collection
METRICS_BATCH_SIZE=100
METRICS_FLUSH_INTERVAL=5000  # 5 seconds

# Alert Configuration
ALERT_STUCK_IN_TRAINING_DAYS=14
ALERT_DECLINING_CONFIDENCE_RATE=0.05
ALERT_LOW_COMPLETION_RATE=0.80
ALERT_HIGH_INTERVENTION_RATE=0.50

# Cache Configuration
DASHBOARD_CACHE_TTL=300      # 5 minutes
METRICS_CACHE_TTL=3600       # 1 hour
ALERT_DEDUP_TTL=600          # 10 minutes

# Background Jobs
ALERT_CHECK_INTERVAL_HOURLY=0 * * * *    # Every hour
ALERT_CHECK_INTERVAL_DAILY=0 9 * * *     # Daily at 9 AM
METRICS_AGGREGATION_SCHEDULE=*/5 * * * * # Every 5 minutes

# backend-saas/core/config.py
TRAINING_SYSTEM_ENABLED = os.getenv("TRAINING_SYSTEM_ENABLED", "true").lower() == "true"
TRAINING_ANALYTICS_ENABLED = os.getenv("TRAINING_ANALYTICS_ENABLED", "true").lower() == "true"
TRAINING_ALERTS_ENABLED = os.getenv("TRAINING_ALERTS_ENABLED", "true").lower() == "true"

// src/lib/features.ts
export const TRAINING_UI_ENABLED = process.env.NEXT_PUBLIC_TRAINING_UI_ENABLED === 'true';
export const TRAINING_ANALYTICS_ENABLED = process.env.NEXT_PUBLIC_TRAINING_ANALYTICS_ENABLED === 'true';

# Example: Configure custom alert threshold for tenant
await alerting_service.configure_alert_thresholds(
    tenant_id="tenant-123",
    alert_type="stuck_in_training",
    thresholds={"max_days_in_training": 21}  # Override default 14 days
)

# Configure notification channels per alert type
ALERT_NOTIFICATION_CHANNELS = {
    "stuck_in_training": ["email", "in_app"],
    "declining_confidence": ["email", "in_app"],
    "low_completion_rate": ["email", "in_app", "slack"],
    "high_intervention_rate": ["email", "in_app"],
    "blocked_trigger_spike": ["in_app"],
}

-- Find slow queries (>500ms)
SELECT query, mean_exec_time, calls
FROM pg_stat_statements
WHERE query LIKE '%training_metrics%'
  AND mean_exec_time > 500
ORDER BY mean_exec_time DESC;

-- Check materialized view refresh performance
SELECT schemaname, matviewname, last_refresh
FROM pg_matviews
WHERE matviewname LIKE '%training%';

# Monitor Redis cache performance
import redis

redis_client = redis.Redis.from_url(settings.REDIS_URL)
info = redis_client.info('stats')

cache_hits = info['keyspace_hits']
cache_misses = info['keyspace_misses']
hit_rate = cache_hits / (cache_hits + cache_misses) if (cache_hits + cache_misses) > 0 else 0

print(f"Cache hit rate: {hit_rate:.2%}")  # Target: ≥80%

# Check API logs
fly logs -a atom-saas --tail 100 | grep training

# Verify database connectivity
psql $DATABASE_URL -c "SELECT 1"

# Check Redis status
redis-cli ping

# Review slow queries
psql $DATABASE_URL -c "SELECT * FROM pg_stat_statements WHERE query LIKE '%training%' ORDER BY mean_exec_time DESC LIMIT 10"

# Test WebSocket connection
wscat -c "wss://atom-saas.fly.dev/api/maturity/supervision/session-id/ws" \
  -H "Authorization: Bearer $TOKEN"

# Check server logs
fly logs -a atom-saas --tail 100 | grep websocket

# Verify WebSocket configuration
grep SUPERVISION_WEBSOCKET_TIMEOUT .env

# Manually trigger alert check
await alerting_service.check_alerts(tenant_id="tenant-123")

# Check background job logs
grep "alert_check" /var/log/supervisor/background-worker.log

# Verify alert thresholds
SELECT * FROM alert_configurations WHERE tenant_id = 'tenant-123';

# Enable pagination for large datasets
@router.get("/analytics/training/dashboard")
async def get_dashboard(
    limit: int = Query(100, le=1000),  # Max 1000 records
    offset: int = Query(0)
):
    # ... pagination logic

# Use materialized views instead of raw queries
# Implement cursor-based pagination for large datasets

# Monitor memory usage
import psutil
process = psutil.Process()
print(f"Memory usage: {process.memory_info().rss / 1024 / 1024:.2f} MB")

# backend-saas/core/logging_config.py
LOG_LEVEL = os.getenv("LOG_LEVEL", "DEBUG")

# Enable query logging
import logging
logging.getLogger('sqlalchemy.engine').setLevel(logging.DEBUG)

# Track query execution time
import time

start_time = time.time()
result = await db.execute(query)
execution_time = time.time() - start_time

if execution_time > 0.5:  # Log slow queries
    logger.warning(f"Slow query ({execution_time:.2f}s): {query}")

-- Add composite index for dashboard queries
CREATE INDEX CONCURRENTLY idx_training_metrics_dashboard
ON training_metrics(tenant_id, metric_name, recorded_at DESC)
WHERE recorded_at > NOW() - INTERVAL '90 days';

-- Analyze query performance
EXPLAIN ANALYZE
SELECT * FROM get_training_dashboard_data('tenant-123', '30 days');

-- Increase refresh frequency for real-time data
CREATE MATERIALIZED VIEW training_metrics_aggregated_5min AS
SELECT
    tenant_id,
    metric_name,
    date_trunc('5 minutes', recorded_at) AS time_bucket,
    AVG(metric_value) AS avg_value,
    MIN(metric_value) AS min_value,
    MAX(metric_value) AS max_value,
    COUNT(*) AS count
FROM training_metrics
GROUP BY tenant_id, metric_name, date_trunc('5 minutes', recorded_at);

-- Refresh every 5 minutes via cron or pg_cron
REFRESH MATERIALIZED VIEW CONCURRENTLY training_metrics_aggregated_5min;

# Dashboard data caching
async def get_dashboard_data_cached(tenant_id: str, time_range: str):
    cache_key = f"dashboard:{tenant_id}:{time_range}"
    cached_data = await redis_client.get(cache_key)

    if cached_data:
        return json.loads(cached_data)

    # Cache miss - fetch from database
    data = await fetch_dashboard_data(tenant_id, time_range)
    await redis_client.setex(
        cache_key,
        300,  # 5 minutes TTL
        json.dumps(data)
    )
    return data

-- Row-Level Security (RLS) policies
ALTER TABLE training_metrics ENABLE ROW LEVEL SECURITY;

CREATE POLICY tenant_isolation_policy ON training_metrics
FOR ALL
USING (tenant_id = current_tenant_id());

# Verify tenant_id on all queries
async def get_dashboard_data(tenant_id: str, db: Session):
    # Always filter by tenant_id
    query = db.query(TrainingMetric).filter(
        TrainingMetric.tenant_id == tenant_id
    )
    return query.all()

# backend-saas/core/audit_service.py
async def log_training_action(
    tenant_id: str,
    action: str,
    user_id: str,
    details: dict
):
    await audit_log.create(
        tenant_id=tenant_id,
        action=f"training_{action}",
        actor_id=user_id,
        details=details,
        timestamp=datetime.utcnow()
    )

-- Generate audit report for tenant
SELECT
    action,
    actor_id,
    details,
    timestamp
FROM audit_logs
WHERE tenant_id = 'tenant-123'
  AND action LIKE 'training_%'
  AND timestamp >= NOW() - INTERVAL '30 days'
ORDER BY timestamp DESC;

# Raw metrics: 90 days
# Aggregated metrics: 1 year
# Alert history: 90 days

# Background job to purge old data
@background_job(schedule="0 2 * * *")  # Daily at 2 AM
async def purge_old_metrics():
    cutoff_date = datetime.utcnow() - timedelta(days=90)
    await db.query(TrainingMetric).filter(
        TrainingMetric.recorded_at < cutoff_date
    ).delete()
    await db.commit()

# Automated backup via Fly.io
fly postgres create --name atom-saas-backup
fly backups create --app atom-saas-db

# Manual backup
pg_dump $DATABASE_URL > training-backup-$(date +%Y%m%d).sql

# List available backups
fly backups list --app atom-saas-db

# Restore from backup
fly backups restore --app atom-saas-db <backup-id>

-- Verify agent proposals count
SELECT tenant_id, COUNT(*) as proposal_count
FROM agent_proposals
GROUP BY tenant_id;

-- Verify metrics continuity
SELECT
    DATE(recorded_at) as date,
    COUNT(*) as metric_count
FROM training_metrics
WHERE recorded_at >= NOW() - INTERVAL '7 days'
GROUP BY DATE(recorded_at)
ORDER BY date DESC;

@dataclass
class CustomAlertType:
    name: str = "custom_declining_performance"
    description: str = "Agent performance declining over 30 days"
    severity: str = "warning"
    check_frequency: str = "1week"
    threshold_config: dict[str, Any] = field(default_factory=lambda: {
        "min_decline_rate": 0.10,
        "min_performance_score": 0.70
    })

# Register custom alert
ALERT_TYPES.append(CustomAlertType())

async def check_custom_alert(
    tenant_id: str,
    agent_id: str
) -> Alert | None:
    # Calculate performance decline over 30 days
    current_score = await get_current_performance_score(agent_id)
    historical_score = await get_historical_performance_score(
        agent_id,
        days_ago=30
    )

    decline_rate = (historical_score - current_score) / historical_score

    if decline_rate >= 0.10 and current_score < 0.70:
        return Alert(
            tenant_id=tenant_id,
            alert_type="custom_declining_performance",
            severity="warning",
            agent_id=agent_id,
            details={
                "current_score": current_score,
                "historical_score": historical_score,
                "decline_rate": decline_rate
            }
        )
    return None

# Primary region (FRA)
DATABASE_URL_PRIMARY="postgresql://...@atom-saas-db.fra.aws.neon.tech"

# Replica region (EWR)
DATABASE_URL_REPLICA="postgresql://...@atom-saas-db.ewr.aws.neon.tech"

# Route reads to replica, writes to primary
if request.method == "GET":
    db = get_db_session(DATABASE_URL_REPLICA)
else:
    db = get_db_session(DATABASE_URL_PRIMARY)

// Connect to nearest region
const region = detectNearestRegion();
const wsUrl = `wss://${region}.atom-saas.fly.dev/api/maturity/supervision/${sessionId}/ws`;

# backend-saas/core/monitoring.py
from prometheus_client import Counter, Histogram, Gauge

# Training metrics
training_proposals_total = Counter(
    'training_proposals_total',
    'Total training proposals',
    ['tenant_id', 'status']
)

training_sessions_duration = Histogram(
    'training_session_duration_seconds',
    'Training session duration',
    ['tenant_id']
)

supervision_sessions_active = Gauge(
    'supervision_sessions_active',
    'Active supervision sessions',
    ['tenant_id']
)

alert_triggered_total = Counter(
    'alert_triggered_total',
    'Total alerts triggered',
    ['tenant_id', 'alert_type', 'severity']
)

{
  "dashboard": {
    "title": "Training System Overview",
    "panels": [
      {
        "title": "Training Proposals by Status",
        "targets": [
          {
            "expr": "sum by (status) (training_proposals_total)"
          }
        ]
      },
      {
        "title": "Active Supervision Sessions",
        "targets": [
          {
            "expr": "supervision_sessions_active"
          }
        ]
      },
      {
        "title": "Alerts Triggered (Last 24h)",
        "targets": [
          {
            "expr": "sum by (alert_type) (increase(alert_triggered_total[24h]))"
          }
        ]
      }
    ]
  }
}

# Send critical alerts to PagerDuty
from pdpyras import EventsAPISession

pagerduty = EventsAPISession(settings.PAGERDUTY_ROUTING_KEY)

async def send_critical_alert(alert: Alert):
    if alert.severity == "critical":
        pagerduty.post(
            dedup_key=f"{alert.tenant_id}-{alert.alert_type}",
            event_type="trigger",
            severity="critical",
            summary=alert.title,
            custom_details=alert.details
        )