⚡ Kyronix Flow

A distributed task queue with AI-powered scheduling, failure analysis, and real-time observability — built with FastAPI, Redis, and React.

📋 Table of Contents

Overview
Architecture
Features
Installation
Quick Start
API Reference
CLI Usage
AI Features
Load Testing
Configuration
Roadmap
Testing
License

🔍 Overview

Kyronix Flow is a project demonstrating distributed systems engineering end-to-end. It implements a background task processing pipeline with priority queues, automatic retries, dead letter queues, stale task recovery, and an AI layer for natural language task creation and failure analysis — all served through a REST API, a React dashboard, and a CLI.

Note: This project demonstrates distributed systems concepts, reliability patterns, and AI integration depth. While it implements several production patterns, it is not hardened for production use in its current form. A path to production readiness is outlined in the Roadmap.

What it demonstrates

Concept	Implementation
Message queue	Redis lists as priority queues, `BRPOP` for blocking dequeue
State machine	`PENDING → RUNNING → COMPLETED / FAILED` with retry transitions
Exponential backoff	`2^n` second delays via Redis sorted set scheduling
Dead Letter Queue	Permanent failure storage with metadata and manual retry
Stale task recovery	Scheduler detects and requeues orphaned `RUNNING` tasks
Worker pool	`ThreadPoolExecutor` with heartbeat and TTL-based eviction
Optimistic locking	Redis `WATCH/MULTI` for concurrent state mutations
AI reliability layer	Prompt injection defense, hallucination guard, cost tracking, fallback heuristics

🏗️ Architecture

System Diagram

graph TB
    subgraph Clients["👥 Clients"]
        CLI[CLI]
        FE[React Dashboard]
    end

    subgraph Proxy["🔒 Proxy"]
        NGINX[Nginx<br/>Rate Limit + Security]
    end

    subgraph Services["🔧 Services"]
        FastAPI[FastAPI<br/>REST Endpoints]
        AI_SVC[AI Service<br/>OpenAI-compatible API]
    end

    subgraph Workers["⚙️ Workers"]
        CPU[CpuWorker]
        API_W[ApiWorker]
        FILE[FileWorker]
        ML[MlWorker]
    end

    subgraph Scheduler["⏱️ Scheduler"]
        RETRY[Retry Scheduler]
    end

    subgraph Data["💾 Data"]
        REDIS[(Redis 7)]
    end

    CLI -->|HTTP| FastAPI
    FE --> NGINX --> FastAPI
    FastAPI --> REDIS
    CPU --> REDIS
    API_W --> REDIS
    FILE --> REDIS
    ML --> REDIS
    RETRY --> REDIS
    FastAPI --> AI_SVC

Tech Stack

Layer	Technology	Purpose
API	FastAPI, Python 3.12	REST endpoints, async request handling
Data	Redis 7	Task state, priority queues, retry sorted set, metrics
Workers	Python, ThreadPoolExecutor	Parallel task execution with heartbeat monitoring
AI	OpenAI-compatible API (GPT-4o-mini, OpenRouter, etc.)	NL parsing, failure analysis, optimization suggestions
Frontend	React 19, Vite, TypeScript, TanStack Query, Recharts, TailwindCSS	Real-time dashboard
Proxy	Nginx 1.27	Rate limiting, gzip, security headers
CLI	Click, Rich	Formatted command-line interface
Infra	Docker, Docker Compose, GitHub Actions	Containerization, CI/CD

Redis Key Schema

taskflow:task:{id}               # Task state hash
taskflow:result:{id}             # Task result hash
taskflow:queue:{high|medium|low} # Priority queues (lists)
taskflow:retry_queue             # Retry sorted set (score = unix timestamp)
taskflow:dlq                     # Dead letter queue (list)
taskflow:dlq:{id}                # DLQ entry metadata (hash)
taskflow:metrics                 # Global metrics (hash)
taskflow:metrics:workers         # Worker registry (hash)
taskflow:metrics:workers:alive   # Worker heartbeat sorted set

Task State Machine

PENDING ──→ RUNNING ──→ COMPLETED
               │
               ├──→ PENDING (retry with backoff)
               │
               └──→ FAILED (max retries exceeded → DLQ)

✨ Features

Core Task Processing

Priority-based scheduling — Tasks routed across high / medium / low priority queues
Automatic retries — Exponential backoff (2^n seconds) with configurable max attempts
Dead Letter Queue — Permanently failed tasks stored with metadata; inspectable and manually retryable
Stale task recovery — Scheduler detects orphaned RUNNING tasks and requeues them
Worker pool — Horizontally scalable via docker compose up --scale worker=N
Optimistic locking — Redis WATCH/MULTI prevents race conditions on concurrent state mutations

AI Layer

Natural language task creation — Plain-English prompts parsed into structured tasks via OpenAI-compatible API
Failure analysis — Root cause categorization (timeout, network, validation, etc.) with confidence scores
Retry recommendations — AI-suggested retry strategy and optimal delay
Task optimization — Batching, chunking, and caching suggestions
Prompt injection defense — Rejects suspicious patterns before they reach the model
Hallucination guard — Validates AI output against allowed values before acting on it
Cost tracking — Token usage and estimated cost logged per request
Graceful fallback — Heuristic defaults when the OpenAI API is unavailable

Observability

Real-time dashboard — React frontend polling task counts, queue depths, worker health, and success rates
REST metrics endpoints — /metrics, /metrics/queues, /metrics/workers
Worker heartbeat — TTL-based eviction removes stale workers from the registry automatically

Infrastructure

Docker Compose — Dev and production configurations, single command startup
Nginx reverse proxy — Rate limiting, gzip compression, security headers
GitHub Actions CI — Lint, type check (mypy), tests, and Docker build verification on every push
CLI — Rich-formatted task management and monitoring from the terminal

🚀 Installation

Prerequisites

Python 3.12+
Redis 7+
Node.js 20+ (frontend only)
Docker & Docker Compose (recommended)

Option 1: Docker (Recommended)

git clone https://github.com/sumitchintanwar/kyronix-flow.git
cd kyronix-flow/taskflow
cp .env.example .env
# Edit .env and set OPENAI_API_KEY + OPENAI_BASE_URL for AI features (optional)

docker compose up -d

# Verify
curl http://localhost:8000/health

Option 2: Manual

git clone https://github.com/sumitchintanwar/kyronix-flow.git
cd kyronix-flow/taskflow
cp .env.example .env
# Edit .env and set OPENAI_API_KEY + OPENAI_BASE_URL for AI features (optional)

python -m venv venv
source venv/bin/activate       # Linux/macOS
# venv\Scripts\activate        # Windows

pip install -e .

# Start Redis
redis-server

# Start API (terminal 1)
uvicorn api.main:app --reload --port 8000

# Start a worker (terminal 2)
python -m workers.main

Option 3: Frontend Only

cd frontend
npm install
npm run dev
# http://localhost:5173

⚡ Quick Start

1. Submit a task

curl -X POST http://localhost:8000/tasks \
  -H "Content-Type: application/json" \
  -d '{
    "type": "cpu",
    "payload": {"action": "fibonacci", "n": 30},
    "priority": "medium",
    "max_retries": 3
  }'

{
  "task_id": "task_abc123",
  "status": "pending"
}

2. Check status

curl http://localhost:8000/tasks/task_abc123

3. Get result

curl http://localhost:8000/tasks/task_abc123/result

4. View metrics

curl http://localhost:8000/metrics

📡 API Reference

Tasks

# Create a CPU task
curl -X POST http://localhost:8000/tasks \
  -H "Content-Type: application/json" \
  -d '{"type": "cpu", "payload": {"action": "is_prime", "n": 97}, "priority": "high"}'

# Create a file processing task
curl -X POST http://localhost:8000/tasks \
  -H "Content-Type: application/json" \
  -d '{"type": "file", "payload": {"action": "csv_parse", "file_path": "/data/input.csv"}}'

# Create an API call task
curl -X POST http://localhost:8000/tasks \
  -H "Content-Type: application/json" \
  -d '{"type": "api", "payload": {"method": "GET", "url": "https://api.example.com/data"}}'

# List tasks
curl "http://localhost:8000/tasks?limit=10"

# Delete a task
curl -X DELETE http://localhost:8000/tasks/task_abc123

Metrics & Health

curl http://localhost:8000/metrics
curl http://localhost:8000/metrics/queues
curl http://localhost:8000/metrics/workers
curl http://localhost:8000/health

AI Endpoints

# Natural language task creation
curl -X POST http://localhost:8000/tasks/natural-language \
  -H "Content-Type: application/json" \
  -d '{"prompt": "Calculate the first 50 prime numbers and sort them"}'

# Failure analysis + retry recommendation for a task
curl http://localhost:8000/tasks/task_abc123/insights

# AI usage metrics
curl http://localhost:8000/metrics/ai

Dead Letter Queue

curl http://localhost:8000/tasks/dead                          # List DLQ entries
curl -X POST http://localhost:8000/dlq/task_abc123/retry       # Retry a failed task
curl -X DELETE http://localhost:8000/dlq                       # Clear the DLQ

🖥️ CLI Usage

pip install -e .

Task Management

taskflow task create --type cpu --payload '{"operation": "fibonacci", "n": 30}' --priority high
taskflow task list --limit 10
taskflow task get task_abc123
taskflow task result task_abc123
taskflow task delete task_abc123

Quick Submit

taskflow submit cpu fibonacci --n 30
taskflow submit ml sentiment --text "This is amazing!"
taskflow submit api webhook --url https://example.com
taskflow submit file csv-parse --rows 500 --columns 10

Monitoring

taskflow status task_abc123
taskflow metrics
taskflow metrics queues
taskflow metrics workers
taskflow workers
taskflow health

Dead Letter Queue

taskflow dlq
taskflow dlq show task_abc123
taskflow dlq retry task_abc123
taskflow dlq clear

Example Output

┌─────────────────────────────────────────┐
│         Task Status: task_abc123        │
├─────────────────────────────────────────┤
│ Status      │ completed                 │
│ Type        │ cpu                       │
│ Priority    │ high                      │
│ Retries     │ 0/3                       │
│ Created     │ 2026-01-15 10:30:00 UTC   │
│ Updated     │ 2026-01-15 10:30:02 UTC   │
└─────────────────────────────────────────┘

┌─────────────────────────────────────────┐
│              Task Result                │
├─────────────────────────────────────────┤
│ operation     │ fibonacci               │
│ n             │ 30                      │
│ result        │ 832040                  │
│ execution_ms  │ 1247                    │
└─────────────────────────────────────────┘

🤖 AI Features

Natural Language Task Creation

curl -X POST http://localhost:8000/tasks/natural-language \
  -H "Content-Type: application/json" \
  -d '{"prompt": "Urgently calculate the first 100 prime numbers"}'

{
  "task_id": "task_xyz789",
  "parsed_task": {
    "task_type": "cpu",
    "priority": "high",
    "payload": {"operation": "prime", "n": 100},
    "confidence": 0.95,
    "reasoning": "User requested prime calculation with urgency keyword"
  }
}

Failure Analysis

curl http://localhost:8000/tasks/task_abc123/insights

{
  "task_id": "task_abc123",
  "failure_analysis": {
    "root_cause": "Task exceeded 30s timeout due to large input size",
    "category": "timeout",
    "confidence": 0.92,
    "recommendation": "Increase timeout or split into smaller chunks",
    "prevention_strategy": "Add input size validation before processing"
  },
  "retry_recommendation": {
    "should_retry": true,
    "suggested_delay": 5.0,
    "reasoning": "Timeout failures are transient; retry with increased timeout"
  },
  "optimization": {
    "recommendations": ["Batch input into chunks of 1000"],
    "estimated_improvement": "50% faster execution",
    "risk_notes": "Chunking may increase memory usage"
  }
}

📊 Load Testing

cd taskflow/load_tests
pip install -r requirements.txt

# Run all scenarios
./run_load_tests.sh all

# Run a specific scenario
./run_load_tests.sh task_creation
./run_load_tests.sh task_polling
./run_load_tests.sh metrics_requests
./run_load_tests.sh ai_requests
./run_load_tests.sh combined

# Analyze results
python3 analyze_results.py ./results

Scenario	Description	Users
`task_creation`	POST /tasks write throughput	10-100
`task_polling`	Create → poll → completion latency	10-50
`metrics_requests`	Dashboard polling simulation	10-100
`ai_requests`	Natural language + insights endpoints	5-20
`combined`	Mixed realistic workload	20-100

🔧 Configuration

Environment Variables

Variable	Default	Description
`REDIS_HOST`	`localhost`	Redis host
`REDIS_PORT`	`6379`	Redis port
`OPENAI_API_KEY`	—	Required for AI features
`OPENAI_BASE_URL`	—	Custom OpenAI-compatible API base URL (e.g. `https://openrouter.ai/api/v1`)
`OPENAI_MODEL`	`gpt-4o-mini`	Model to use
`WORKER_CONCURRENCY`	`4`	Worker thread pool size

Scaling Workers

docker compose up -d --scale worker=5

🗺️ Roadmap

The following tracks what would be needed to move this toward production readiness:

Near-term

WebSocket support for real-time push (replacing polling)
Task dependencies and DAG-based execution ordering
Persistent task history backed by PostgreSQL

AI improvements

Multi-model support (Claude, Gemini, local models)
Predictive failure detection based on historical patterns
Batch task optimization suggestions

Production hardening

Role-based access control (RBAC)
Audit logging
SLA monitoring and alerting
Kubernetes deployment manifests
Prometheus metrics export
Distributed tracing via OpenTelemetry
Multi-tenant support

🧪 Testing

# Run all tests
pytest

# With coverage report
pytest --cov=api --cov=ai --cov=workers

# Specific modules
pytest tests/test_api_endpoints.py -v
pytest tests/test_integration.py -v

📄 License

MIT — see LICENSE.

🙏 Acknowledgments

FastAPI · Redis · OpenAI · Rich · Docker

Built to demonstrate distributed systems engineering end-to-end

Tested with: cohere/north-mini-code:free via OpenRouter

Report a Bug · Wiki

Name		Name	Last commit message	Last commit date
.github/workflows		.github/workflows
docs		docs
frontend		frontend
k8s		k8s
taskflow		taskflow
README.md		README.md
fix_main.py		fix_main.py
patch_main.py		patch_main.py

Latest commit

History

Folders and files

Repository files navigation

⚡ Kyronix Flow

📋 Table of Contents

🔍 Overview

What it demonstrates

🏗️ Architecture

System Diagram

Tech Stack

Redis Key Schema

Task State Machine

✨ Features

Core Task Processing

AI Layer

Observability

Infrastructure

🚀 Installation

Prerequisites

Option 1: Docker (Recommended)

Option 2: Manual

Option 3: Frontend Only

⚡ Quick Start

1. Submit a task

2. Check status

3. Get result

4. View metrics

📡 API Reference

Tasks

Metrics & Health

AI Endpoints

Dead Letter Queue

🖥️ CLI Usage

Task Management

Quick Submit

Monitoring

Dead Letter Queue

Example Output

🤖 AI Features

Natural Language Task Creation

Failure Analysis

📊 Load Testing

🔧 Configuration

Environment Variables

Scaling Workers

🗺️ Roadmap

🧪 Testing

📄 License

🙏 Acknowledgments

About

Resources

Stars

Watchers

Forks

Releases

Packages

Contributors

Languages