CODITECT License Server - Software Design Document
Version: 1.0 Status: Draft Last Updated: November 30, 2025 Authors: Architecture Team Stakeholders: Engineering, Product, Finance, Legal
Document Structure
This SDD is broken into multiple parts for manageability:
- Part 1 (this document): Executive Summary, Overview, Licensing Models 1-5
- Part 2: Licensing Models 6-10
- Part 3: CODITECT-CORE Enforcement Architecture
- Part 4: C4 Diagrams, Deployment, Technology Stack
- Supporting Documents: ADRs (20+), Sequence Diagrams (20+), Integration Guides
Executive Summary
Purpose
This document specifies the architecture for CODITECT's centralized license server, designed to achieve feature parity with modern licensing platforms (SLASCONE, Keygen, LicenseSpring) while addressing CODITECT-CORE's unique symlink-based distributed architecture.
Key Requirements
- 10 Licensing Models: Floating/Concurrent, Feature-Based, Device-Based, Named User, Subscription, Trial, Usage-Based, Version, Perpetual, Expiration
- CODITECT-CORE Integration: License enforcement for framework installed as git submodule with symlink chains
- Dual Usage Model: Builder licenses (per-developer) + Runtime licenses (embedded in applications)
- Offline Support: Tier-based offline grace periods (24-168 hours)
- Multi-Tenant SaaS: Row-level tenant isolation with django-multitenant
- High Availability: Redis atomic operations, PostgreSQL regional HA, GKE deployment
- Modern API-First: RESTful API, event-driven webhooks, CRM/billing integration
Business Context
Current State (60% Complete):
- Django 5.2.8 backend with DRF
- PostgreSQL 16 + Redis Memorystore
- Models, serializers, views for floating licenses
- Atomic seat counting via Lua scripts
Target State:
- Phase 1 (MVP - 2-3 days): Floating licenses, feature gating, expiration, offline grace
- Phase 2 (6-8 weeks): Device-based, named user, subscriptions, trials, webhooks
- Phase 3 (Q2 2026): Usage-based metering, analytics, perpetual licenses
Strategic Decision: Continue custom Django implementation vs. commercial platform ($0/year vs. $1,200-3,600/year + 5-7 days integration effort).
CODITECT-CORE Unique Architecture
CODITECT-CORE is installed as a git submodule with symlink chains:
PROJECT_ROOT/
├── .coditect/ # Physical directory (git submodule)
│ ├── scripts/init.sh # LICENSE ENFORCEMENT POINT
│ └── scripts/validate-license.py # LICENSE VALIDATION LOGIC
├── .claude -> .coditect # Symlink for Claude Code
├── submodule-1/
│ ├── .coditect -> ../.coditect # Symlink to parent
│ └── .claude -> .coditect
└── submodule-2/
├── .coditect -> ../.coditect # SAME RESOLVED PATH
└── .claude -> .coditect
License Enforcement Challenge:
- Multiple symlinks → same physical
.coditect/directory - Should this count as 1 session or multiple?
- Solution: Use
os.path.realpath('.coditect')to resolve symlinks - Fair pricing: 1 project = 1 session, regardless of symlink count
System Overview
Architecture Principles
- API-First Design: All operations via RESTful API (no UI-coupled logic)
- Atomic Operations: Redis Lua scripts for race condition-free seat management
- Multi-Tenant Isolation: django-multitenant with row-level security
- Event-Driven: Webhooks for lifecycle events (activation, expiration, upgrades)
- Offline-First Client: Cached validation with tier-based grace periods
- Feature Gating: Centralized entitlement matrix (free → pro → team → enterprise)
- Observability: Prometheus metrics, Cloud Logging, Cloud Trace
- Security: Cloud KMS RSA-4096 license signing, TLS 1.3, RBAC
System Context (C4 Level 1)
┌─────────────────┐
│ CODITECT User │
│ (Developer) │
└────────┬────────┘
│
│ validate license
▼
┌─────────────────────────────────────────────────────────┐
│ CODITECT License Server (SaaS) │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ Django API │ │ Redis Pool │ │ PostgreSQL │ │
│ │ (Licenses) │──│ (Atomic Ops) │──│ (Persistent) │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
└─────────────────────────────────────────────────────────┘
│ │
│ webhooks │ sync
▼ ▼
┌─────────────────┐ ┌─────────────────┐
│ CRM/Billing │ │ Analytics │
│ (Stripe, etc) │ │ (BigQuery) │
└─────────────────┘ └─────────────────┘
High-Level Data Flow
1. License Activation (CODITECT-CORE init.sh):
CODITECT Client (init.sh)
→ POST /api/v1/licenses/activate
{session_id, user_email, hardware_id, project_root, coditect_path, version}
← 200 OK {license_key, features, offline_cache, ttl}
→ Save .license-cache (encrypted JSON)
2. Heartbeat (6-minute TTL for floating licenses):
Background Thread (every 5 min)
→ POST /api/v1/licenses/heartbeat
{session_id, license_key}
← 200 OK {renewed: true, ttl: 360}
→ Update .license-cache timestamp
3. Graceful Release (session end):
CODITECT Exit Hook
→ POST /api/v1/licenses/release
{session_id, license_key}
← 200 OK {released: true, seats_available: 9}
→ Delete .license-cache
Part 1: Licensing Models (1-5)
1. Floating/Concurrent Licenses
Definition: A pool of N licenses shared among unlimited users. Users "check out" a license when starting CODITECT, return it when exiting. If pool exhausted, new users wait or denied.
Use Case: Team with 10 developers, 5 floating licenses. Up to 5 can use CODITECT simultaneously. License returns to pool when developer exits.
CODITECT-CORE Implementation:
Session Identification:
def generate_session_id() -> str:
"""Unique identifier for CODITECT-CORE session"""
# Resolved path handles symlinks
coditect_path = os.path.realpath('.coditect')
session_data = {
'user_email': get_user_email(), # Git config user.email
'hardware_id': get_hardware_id(), # MAC address + CPU ID
'project_root': get_project_root(), # Git repo root
'coditect_path': coditect_path, # Resolved .coditect path
'coditect_version': get_version(), # Framework version
'usage_type': 'builder' # 'builder' or 'runtime'
}
return hashlib.sha256(
json.dumps(session_data, sort_keys=True).encode()
).hexdigest()
Key Insight:
os.path.realpath('.coditect')resolves symlinks to physical directory- Parent project and symlinked submodules → same resolved path → same session_id
- Different projects with different
project_root→ different session_id - Fair pricing: 1 project = 1 seat, regardless of symlink architecture
API Endpoint:
POST /api/v1/licenses/floating/acquire
{
"tenant_id": "coditect-team-acme",
"license_key": "FL-XXXX-XXXX-XXXX",
"session_id": "sha256(...)",
"user_email": "dev@acme.com",
"hardware_id": "mac:xx:xx:xx:xx:xx:xx|cpu:intel-i9",
"project_root": "/Users/dev/projects/acme-app",
"coditect_path": "/Users/dev/projects/acme-app/.coditect",
"coditect_version": "1.2.0",
"usage_type": "builder"
}
Response 200 OK:
{
"session_id": "sha256(...)",
"license_key": "FL-XXXX-XXXX-XXXX",
"acquired": true,
"seat_number": 3,
"total_seats": 10,
"available_seats": 7,
"expires_at": "2025-12-01T12:00:00Z",
"heartbeat_ttl": 360, // seconds
"offline_cache": {
"encrypted": "base64(...)", // Encrypted license data
"expires_at": "2025-12-02T12:00:00Z", // 24-72h grace period
"features": ["all_agents", "all_commands", "unlimited_projects"]
}
}
Response 409 Conflict (pool exhausted):
{
"error": "no_seats_available",
"message": "All 10 floating licenses in use",
"total_seats": 10,
"available_seats": 0,
"active_sessions": [
{"user": "dev1@acme.com", "since": "2025-11-30T08:00:00Z"},
{"user": "dev2@acme.com", "since": "2025-11-30T09:15:00Z"},
...
],
"retry_after": 300 // seconds
}
Redis Atomic Operation (Lua Script):
-- acquire_floating_seat.lua
local license_key = KEYS[1]
local session_id = ARGV[1]
local max_seats = tonumber(ARGV[2])
local ttl = tonumber(ARGV[3])
-- Check if session already has a seat
local existing_seat = redis.call('HGET', license_key .. ':sessions', session_id)
if existing_seat then
-- Renew TTL
redis.call('EXPIRE', license_key .. ':sessions', ttl)
return {1, tonumber(existing_seat)} -- {acquired, seat_number}
end
-- Count active sessions
local active_count = redis.call('HLEN', license_key .. ':sessions')
if active_count >= max_seats then
return {0, 0} -- {not_acquired, no_seat}
end
-- Assign seat
local seat_number = active_count + 1
redis.call('HSET', license_key .. ':sessions', session_id, seat_number)
redis.call('EXPIRE', license_key .. ':sessions', ttl)
return {1, seat_number} -- {acquired, seat_number}
Django View:
# backend/licenses/views.py
@api_view(['POST'])
@permission_classes([IsAuthenticated])
def acquire_floating_license(request):
"""Acquire floating license seat"""
serializer = FloatingLicenseAcquireSerializer(data=request.data)
serializer.is_valid(raise_exception=True)
license_key = serializer.validated_data['license_key']
session_id = serializer.validated_data['session_id']
# Get license from PostgreSQL
license_obj = License.objects.get(
tenant=request.tenant,
license_key=license_key,
license_type='floating'
)
# Atomic acquire via Redis Lua
redis_client = get_redis_client()
script = redis_client.register_script(ACQUIRE_FLOATING_SEAT_LUA)
acquired, seat_number = script(
keys=[f"license:{license_key}"],
args=[session_id, license_obj.max_seats, 360] # 6-minute TTL
)
if not acquired:
# Pool exhausted
active_sessions = get_active_sessions(license_key)
return Response({
'error': 'no_seats_available',
'total_seats': license_obj.max_seats,
'active_sessions': active_sessions
}, status=409)
# Create session record
session = LicenseSession.objects.create(
tenant=request.tenant,
license=license_obj,
session_id=session_id,
user_email=serializer.validated_data['user_email'],
hardware_id=serializer.validated_data['hardware_id'],
project_root=serializer.validated_data['project_root'],
coditect_path=serializer.validated_data['coditect_path'],
seat_number=seat_number,
acquired_at=timezone.now(),
last_heartbeat=timezone.now()
)
# Generate offline cache
offline_cache = generate_offline_cache(
license_obj,
session,
grace_hours=license_obj.offline_grace_hours
)
return Response({
'session_id': session_id,
'license_key': license_key,
'acquired': True,
'seat_number': seat_number,
'total_seats': license_obj.max_seats,
'heartbeat_ttl': 360,
'offline_cache': offline_cache
}, status=200)
Data Model:
# backend/licenses/models.py
class License(TenantModel):
"""License entity"""
tenant = models.ForeignKey(Tenant, on_delete=models.CASCADE)
license_key = models.CharField(max_length=64, unique=True, db_index=True)
license_type = models.CharField(max_length=32, choices=[
('floating', 'Floating/Concurrent'),
('device', 'Device-Based'),
('named_user', 'Named User'),
('subscription', 'Subscription'),
('trial', 'Trial'),
('usage_based', 'Usage-Based'),
('version', 'Version-Based'),
('perpetual', 'Perpetual'),
('expiration', 'Expiration-Based'),
('feature', 'Feature-Based'),
])
# Floating license specific
max_seats = models.IntegerField(null=True, blank=True) # e.g., 10
# Expiration
expires_at = models.DateTimeField(null=True, blank=True)
# Offline grace period
offline_grace_hours = models.IntegerField(default=24) # Free: 24, Pro: 72, Team: 48
# Feature entitlements (JSON)
features = models.JSONField(default=dict) # {'agents': ['*'], 'commands': ['*']}
# Metadata
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
db_table = 'licenses'
indexes = [
models.Index(fields=['tenant', 'license_key']),
models.Index(fields=['license_type']),
]
class LicenseSession(TenantModel):
"""Active license session"""
tenant = models.ForeignKey(Tenant, on_delete=models.CASCADE)
license = models.ForeignKey(License, on_delete=models.CASCADE)
session_id = models.CharField(max_length=64, unique=True, db_index=True)
# Client identification
user_email = models.EmailField()
hardware_id = models.CharField(max_length=128)
project_root = models.CharField(max_length=512)
coditect_path = models.CharField(max_length=512) # Resolved path
coditect_version = models.CharField(max_length=32)
usage_type = models.CharField(max_length=16, choices=[
('builder', 'Builder License'),
('runtime', 'Runtime License'),
])
# Floating license specific
seat_number = models.IntegerField(null=True, blank=True)
# Lifecycle
acquired_at = models.DateTimeField(auto_now_add=True)
last_heartbeat = models.DateTimeField()
released_at = models.DateTimeField(null=True, blank=True)
class Meta:
db_table = 'license_sessions'
indexes = [
models.Index(fields=['tenant', 'license']),
models.Index(fields=['session_id']),
models.Index(fields=['last_heartbeat']), # Zombie session detection
]
CODITECT-CORE Client Integration:
# .coditect/scripts/validate-license.py
import os
import json
import hashlib
import requests
from pathlib import Path
def validate_license_online():
"""Validate license with server"""
# Generate session ID
session_id = generate_session_id()
# Load license key from config
license_key = load_license_key()
# Acquire floating seat
response = requests.post(
'https://license.coditect.ai/api/v1/licenses/floating/acquire',
json={
'session_id': session_id,
'license_key': license_key,
'user_email': get_user_email(),
'hardware_id': get_hardware_id(),
'project_root': get_project_root(),
'coditect_path': os.path.realpath('.coditect'),
'coditect_version': get_coditect_version(),
'usage_type': 'builder'
},
headers={'Authorization': f'Bearer {get_api_token()}'}
)
if response.status_code == 200:
data = response.json()
# Save offline cache
cache_path = Path('.coditect/.license-cache')
cache_path.write_text(json.dumps(data['offline_cache']))
print(f"✓ License activated (seat {data['seat_number']}/{data['total_seats']})")
# Start heartbeat background thread
start_heartbeat_thread(session_id, license_key, data['heartbeat_ttl'])
return True
elif response.status_code == 409:
data = response.json()
print(f"❌ No seats available ({data['total_seats']} in use)")
print(f" Active users: {', '.join([s['user'] for s in data['active_sessions'][:5]])}")
print(f" Retry in {data['retry_after']} seconds")
return False
else:
print(f"❌ License activation failed: {response.status_code}")
return False
def validate_license_cached():
"""Validate using cached license (offline mode)"""
cache_path = Path('.coditect/.license-cache')
if not cache_path.exists():
return False
cache_data = json.loads(cache_path.read_text())
# Check expiration
expires_at = datetime.fromisoformat(cache_data['expires_at'])
if datetime.now(timezone.utc) > expires_at:
print("❌ Offline cache expired")
return False
# Decrypt and validate
decrypted = decrypt_offline_cache(cache_data['encrypted'])
time_remaining = expires_at - datetime.now(timezone.utc)
hours_remaining = int(time_remaining.total_seconds() / 3600)
print(f"✓ License valid (offline mode, {hours_remaining}h remaining)")
return True
Heartbeat Background Thread:
# .coditect/scripts/heartbeat.py
import threading
import time
import requests
def start_heartbeat_thread(session_id, license_key, ttl):
"""Background thread to send heartbeat every 5 minutes"""
def heartbeat_loop():
while True:
time.sleep(300) # 5 minutes (TTL is 6 minutes)
try:
response = requests.post(
'https://license.coditect.ai/api/v1/licenses/heartbeat',
json={
'session_id': session_id,
'license_key': license_key
},
headers={'Authorization': f'Bearer {get_api_token()}'}
)
if response.status_code == 200:
print("✓ License renewed")
else:
print(f"⚠️ Heartbeat failed: {response.status_code}")
# Continue using offline cache
except Exception as e:
print(f"⚠️ Heartbeat error: {e}")
# Continue using offline cache
thread = threading.Thread(target=heartbeat_loop, daemon=True)
thread.start()
Zombie Session Cleanup (Server-Side):
# backend/licenses/tasks.py (Celery periodic task)
@celery_app.task
def cleanup_zombie_sessions():
"""Release sessions with expired heartbeat TTL"""
# Sessions with last_heartbeat > 6 minutes ago
cutoff = timezone.now() - timedelta(seconds=360)
zombie_sessions = LicenseSession.objects.filter(
released_at__isnull=True,
last_heartbeat__lt=cutoff
)
for session in zombie_sessions:
# Release seat in Redis
redis_client = get_redis_client()
redis_client.hdel(
f"license:{session.license.license_key}:sessions",
session.session_id
)
# Mark session as released
session.released_at = timezone.now()
session.save()
logger.info(f"Released zombie session {session.session_id}")
2. Feature-Based Licenses
Definition: License grants access to specific features/modules. Different tiers unlock different capabilities.
Use Case:
- Free Tier: 5 agents, 10 commands, 1 project
- Pro Tier: All 52 agents, all 81 commands, unlimited projects
- Team Tier: Pro features + team dashboard + floating seats
- Enterprise Tier: Team features + runtime embedding + SSO + private deployment
CODITECT-CORE Implementation:
Feature Gating Matrix:
# .coditect/config/feature_matrix.py
FEATURE_MATRIX = {
'free': {
'agents': [
'general-purpose',
'codebase-locator',
'codebase-analyzer',
'documentation-librarian',
'project-organizer'
], # 5 agents
'commands': [
'/help', '/analyze', '/search', '/document',
'/deliberation', '/implement', '/prototype',
'/optimize', '/strategy', '/action'
], # 10 commands
'skills': [
'search-strategies',
'git-workflow-automation',
'production-patterns',
'framework-patterns',
'documentation-librarian'
], # 5 skills
'max_projects': 1,
'max_seats': 1,
'offline_grace_hours': 24,
'team_dashboard': False,
'runtime_embedding': False,
'sso_integration': False,
'private_deployment': False,
'support_tier': 'community'
},
'pro': {
'agents': '*', # All 52 agents
'commands': '*', # All 81 commands
'skills': '*', # All 26 skills
'max_projects': -1, # Unlimited
'max_seats': 2, # Individual license (laptop + desktop)
'offline_grace_hours': 72, # 3 days
'team_dashboard': False,
'runtime_embedding': False,
'sso_integration': False,
'private_deployment': False,
'support_tier': 'email'
},
'team': {
'agents': '*',
'commands': '*',
'skills': '*',
'max_projects': -1,
'max_seats': 'floating', # 5-100 floating seats
'offline_grace_hours': 48, # 2 days
'team_dashboard': True, # Usage analytics, seat management
'runtime_embedding': False,
'sso_integration': True, # SAML/OIDC
'private_deployment': False,
'support_tier': 'priority'
},
'enterprise': {
'agents': '*',
'commands': '*',
'skills': '*',
'max_projects': -1,
'max_seats': 'floating', # Unlimited floating seats
'offline_grace_hours': 168, # 7 days
'team_dashboard': True,
'runtime_embedding': True, # Embed CODITECT in customer products
'sso_integration': True,
'private_deployment': True, # On-premises or private GKE
'support_tier': 'dedicated'
}
}
API Endpoint:
GET /api/v1/licenses/{license_key}/features
Response 200 OK:
{
"license_key": "PRO-XXXX-XXXX-XXXX",
"tier": "pro",
"features": {
"agents": "*", // or ["agent1", "agent2", ...]
"commands": "*",
"skills": "*",
"max_projects": -1,
"max_seats": 2,
"offline_grace_hours": 72,
"team_dashboard": false,
"runtime_embedding": false,
"sso_integration": false,
"private_deployment": false,
"support_tier": "email"
},
"entitlements": [
{
"feature": "agents",
"allowed": "*",
"description": "All 52 specialized agents"
},
{
"feature": "commands",
"allowed": "*",
"description": "All 81 slash commands"
},
{
"feature": "offline_grace",
"allowed": 72,
"description": "72-hour offline grace period"
}
]
}
CODITECT-CORE Client Integration:
# .coditect/scripts/feature_gate.py
def check_feature_access(feature_type, feature_name):
"""Check if feature is allowed by license"""
# Load license features from cache
license_data = load_license_cache()
features = license_data['features']
# Get allowed features for type
allowed = features.get(feature_type, [])
# Wildcard = all allowed
if allowed == '*':
return True
# Check if specific feature in allowed list
return feature_name in allowed
# Example: Agent invocation
def invoke_agent(agent_name):
"""Invoke agent if licensed"""
if not check_feature_access('agents', agent_name):
print(f"❌ Agent '{agent_name}' not available in your license tier")
print(f" Upgrade to Pro for access to all 52 agents")
print(f" Visit: https://coditect.ai/upgrade")
return False
# Proceed with agent invocation
return Task(subagent_type=agent_name, ...)
# Example: Command execution
def execute_command(command_name):
"""Execute command if licensed"""
if not check_feature_access('commands', command_name):
print(f"❌ Command '{command_name}' not available in your license tier")
print(f" Available commands in Free tier: /help, /analyze, /search, ...")
print(f" Upgrade to Pro for all 81 commands")
return False
# Proceed with command execution
return execute(command_name)
Django View:
# backend/licenses/views.py
@api_view(['GET'])
@permission_classes([IsAuthenticated])
def get_license_features(request, license_key):
"""Get feature entitlements for license"""
license_obj = License.objects.get(
tenant=request.tenant,
license_key=license_key
)
# Get tier from license metadata
tier = license_obj.metadata.get('tier', 'free')
# Load feature matrix
features = FEATURE_MATRIX.get(tier, FEATURE_MATRIX['free'])
# Build entitlements list
entitlements = []
for feature_type, allowed in features.items():
entitlements.append({
'feature': feature_type,
'allowed': allowed,
'description': get_feature_description(feature_type, allowed)
})
return Response({
'license_key': license_key,
'tier': tier,
'features': features,
'entitlements': entitlements
})
Data Model Extension:
# backend/licenses/models.py
class License(TenantModel):
# ... (existing fields)
# Feature entitlements (JSON)
features = models.JSONField(default=dict) # Feature matrix for this license
# Metadata
metadata = models.JSONField(default=dict) # {'tier': 'pro', 'plan_id': 'price_xxx'}
def get_tier(self):
"""Get license tier"""
return self.metadata.get('tier', 'free')
def is_feature_allowed(self, feature_type, feature_name):
"""Check if feature is allowed"""
allowed = self.features.get(feature_type, [])
if allowed == '*':
return True
return feature_name in allowed
3. Device-Based Licenses
Definition: License tied to specific hardware. User can activate on N devices (e.g., 2 for individual, 5 for team). Must deactivate to switch devices.
Use Case: Pro license allows activation on 2 devices (laptop + desktop). Deactivate laptop to activate on new laptop.
CODITECT-CORE Implementation:
Hardware Fingerprinting:
# .coditect/scripts/hardware_fingerprint.py
import platform
import hashlib
import uuid
import subprocess
def get_hardware_id():
"""Generate stable hardware fingerprint"""
components = []
# 1. MAC address (primary network interface)
try:
mac = ':'.join(['{:02x}'.format((uuid.getnode() >> elements) & 0xff)
for elements in range(0, 2*6, 2)][::-1])
components.append(f"mac:{mac}")
except:
pass
# 2. CPU info
try:
if platform.system() == 'Darwin': # macOS
cpu = subprocess.check_output(
['sysctl', '-n', 'machdep.cpu.brand_string']
).decode().strip()
elif platform.system() == 'Linux':
cpu = subprocess.check_output(
['cat', '/proc/cpuinfo']
).decode().split('model name')[1].split(':')[1].split('\n')[0].strip()
else: # Windows
cpu = platform.processor()
components.append(f"cpu:{cpu}")
except:
pass
# 3. Disk serial (if available)
try:
if platform.system() == 'Darwin':
disk_serial = subprocess.check_output(
['system_profiler', 'SPSerialATADataType']
).decode()
# Parse serial number
components.append(f"disk:{disk_serial}")
except:
pass
# 4. Machine UUID (macOS/Linux)
try:
if platform.system() == 'Darwin':
machine_uuid = subprocess.check_output(
['ioreg', '-rd1', '-c', 'IOPlatformExpertDevice']
).decode()
# Parse UUID
elif platform.system() == 'Linux':
machine_uuid = subprocess.check_output(
['cat', '/etc/machine-id']
).decode().strip()
components.append(f"uuid:{machine_uuid}")
except:
pass
# Combine and hash
fingerprint = '|'.join(components)
return hashlib.sha256(fingerprint.encode()).hexdigest()
API Endpoint:
POST /api/v1/licenses/device/activate
{
"license_key": "PRO-XXXX-XXXX-XXXX",
"hardware_id": "sha256(...)",
"device_name": "MacBook Pro (Hal's Laptop)",
"user_email": "hal@coditect.ai"
}
Response 200 OK:
{
"activated": true,
"device_id": "dev_abc123",
"device_name": "MacBook Pro (Hal's Laptop)",
"activated_at": "2025-11-30T12:00:00Z",
"max_devices": 2,
"active_devices": [
{
"device_id": "dev_abc123",
"device_name": "MacBook Pro (Hal's Laptop)",
"activated_at": "2025-11-30T12:00:00Z",
"last_seen": "2025-11-30T12:00:00Z"
},
{
"device_id": "dev_xyz789",
"device_name": "iMac (Office)",
"activated_at": "2025-11-25T08:00:00Z",
"last_seen": "2025-11-29T18:00:00Z"
}
]
}
Response 409 Conflict (max devices reached):
{
"error": "max_devices_reached",
"message": "Maximum 2 devices allowed for Pro license",
"max_devices": 2,
"active_devices": [
{"device_id": "dev_abc123", "device_name": "MacBook Pro (Old)", ...},
{"device_id": "dev_xyz789", "device_name": "iMac (Office)", ...}
],
"deactivate_url": "https://license.coditect.ai/devices/deactivate"
}
Deactivation Endpoint:
POST /api/v1/licenses/device/deactivate
{
"license_key": "PRO-XXXX-XXXX-XXXX",
"device_id": "dev_abc123"
}
Response 200 OK:
{
"deactivated": true,
"device_id": "dev_abc123",
"device_name": "MacBook Pro (Old)",
"deactivated_at": "2025-11-30T12:00:00Z",
"remaining_devices": 1
}
Data Model:
# backend/licenses/models.py
class DeviceActivation(TenantModel):
"""Device activation record"""
tenant = models.ForeignKey(Tenant, on_delete=models.CASCADE)
license = models.ForeignKey(License, on_delete=models.CASCADE)
device_id = models.CharField(max_length=64, unique=True, db_index=True)
hardware_id = models.CharField(max_length=64) # SHA256 fingerprint
device_name = models.CharField(max_length=256)
user_email = models.EmailField()
# Lifecycle
activated_at = models.DateTimeField(auto_now_add=True)
last_seen = models.DateTimeField(auto_now=True)
deactivated_at = models.DateTimeField(null=True, blank=True)
# Metadata
metadata = models.JSONField(default=dict) # OS, version, etc.
class Meta:
db_table = 'device_activations'
indexes = [
models.Index(fields=['tenant', 'license']),
models.Index(fields=['hardware_id']),
models.Index(fields=['deactivated_at']),
]
Django View:
# backend/licenses/views.py
@api_view(['POST'])
@permission_classes([IsAuthenticated])
def activate_device(request):
"""Activate device for license"""
serializer = DeviceActivationSerializer(data=request.data)
serializer.is_valid(raise_exception=True)
license_key = serializer.validated_data['license_key']
hardware_id = serializer.validated_data['hardware_id']
# Get license
license_obj = License.objects.get(
tenant=request.tenant,
license_key=license_key
)
# Check if device already activated
existing = DeviceActivation.objects.filter(
tenant=request.tenant,
license=license_obj,
hardware_id=hardware_id,
deactivated_at__isnull=True
).first()
if existing:
# Update last_seen
existing.save()
return Response({
'activated': True,
'device_id': existing.device_id,
'device_name': existing.device_name
})
# Check device limit
max_devices = license_obj.metadata.get('max_devices', 2)
active_devices = DeviceActivation.objects.filter(
tenant=request.tenant,
license=license_obj,
deactivated_at__isnull=True
)
if active_devices.count() >= max_devices:
return Response({
'error': 'max_devices_reached',
'message': f'Maximum {max_devices} devices allowed',
'max_devices': max_devices,
'active_devices': [serialize_device(d) for d in active_devices]
}, status=409)
# Create activation
device = DeviceActivation.objects.create(
tenant=request.tenant,
license=license_obj,
device_id=f"dev_{uuid.uuid4().hex[:12]}",
hardware_id=hardware_id,
device_name=serializer.validated_data['device_name'],
user_email=serializer.validated_data['user_email'],
metadata={
'os': platform.system(),
'os_version': platform.version(),
'python_version': platform.python_version()
}
)
return Response({
'activated': True,
'device_id': device.device_id,
'device_name': device.device_name,
'activated_at': device.activated_at,
'max_devices': max_devices,
'active_devices': [serialize_device(d) for d in active_devices]
})
4. Named User Licenses
Definition: License tied to specific user account (email). User can use CODITECT on any device, but only this user.
Use Case: Enterprise with 50 developers. Purchase 50 named user licenses. Each developer gets license tied to their work email.
CODITECT-CORE Implementation:
User Authentication:
# .coditect/scripts/user_auth.py
def authenticate_user():
"""Authenticate user for named user license"""
# Get user email from git config
user_email = subprocess.check_output(
['git', 'config', 'user.email']
).decode().strip()
# Authenticate with license server
response = requests.post(
'https://license.coditect.ai/api/v1/auth/named-user',
json={
'email': user_email,
'license_key': load_license_key()
}
)
if response.status_code == 200:
data = response.json()
# Save auth token
save_auth_token(data['access_token'])
print(f"✓ Authenticated as {user_email}")
return True
elif response.status_code == 403:
print(f"❌ User {user_email} not authorized for this license")
print(f" Contact admin to add your email to the license")
return False
else:
print(f"❌ Authentication failed: {response.status_code}")
return False
API Endpoint:
POST /api/v1/auth/named-user
{
"email": "dev@acme.com",
"license_key": "TEAM-XXXX-XXXX-XXXX"
}
Response 200 OK:
{
"authenticated": true,
"user_email": "dev@acme.com",
"access_token": "eyJ...",
"refresh_token": "eyJ...",
"expires_at": "2025-12-01T12:00:00Z",
"license": {
"license_key": "TEAM-XXXX-XXXX-XXXX",
"tier": "team",
"features": {...}
}
}
Response 403 Forbidden:
{
"error": "user_not_authorized",
"message": "User dev@acme.com not in allowed users list",
"license_key": "TEAM-XXXX-XXXX-XXXX",
"allowed_users": [
"admin@acme.com",
"engineer1@acme.com",
"engineer2@acme.com"
],
"contact_admin": "license-admin@acme.com"
}
Data Model:
# backend/licenses/models.py
class NamedUserLicense(TenantModel):
"""Named user license assignment"""
tenant = models.ForeignKey(Tenant, on_delete=models.CASCADE)
license = models.ForeignKey(License, on_delete=models.CASCADE)
user_email = models.EmailField()
assigned_at = models.DateTimeField(auto_now_add=True)
assigned_by = models.EmailField() # Admin who assigned
# Revocation
revoked_at = models.DateTimeField(null=True, blank=True)
revoked_by = models.EmailField(null=True, blank=True)
class Meta:
db_table = 'named_user_licenses'
unique_together = [['tenant', 'license', 'user_email']]
indexes = [
models.Index(fields=['user_email']),
models.Index(fields=['revoked_at']),
]
Django View:
# backend/licenses/views.py
@api_view(['POST'])
def authenticate_named_user(request):
"""Authenticate user for named user license"""
serializer = NamedUserAuthSerializer(data=request.data)
serializer.is_valid(raise_exception=True)
email = serializer.validated_data['email']
license_key = serializer.validated_data['license_key']
# Get license
license_obj = License.objects.get(
license_key=license_key,
license_type='named_user'
)
# Check if user is assigned
assignment = NamedUserLicense.objects.filter(
license=license_obj,
user_email=email,
revoked_at__isnull=True
).first()
if not assignment:
# Get allowed users for error message
allowed_users = NamedUserLicense.objects.filter(
license=license_obj,
revoked_at__isnull=True
).values_list('user_email', flat=True)
return Response({
'error': 'user_not_authorized',
'message': f'User {email} not in allowed users list',
'allowed_users': list(allowed_users)
}, status=403)
# Generate JWT token
access_token = generate_access_token(email, license_obj)
refresh_token = generate_refresh_token(email, license_obj)
return Response({
'authenticated': True,
'user_email': email,
'access_token': access_token,
'refresh_token': refresh_token,
'expires_at': timezone.now() + timedelta(hours=24),
'license': {
'license_key': license_key,
'tier': license_obj.get_tier(),
'features': license_obj.features
}
})
5. Subscription Licenses
Definition: Time-based recurring license (monthly/annual). Auto-renewal with billing integration. Graceful degradation on expiration.
Use Case: Monthly subscription at $29/month. Auto-renews on 1st of each month. If payment fails, 7-day grace period before downgrade to free tier.
CODITECT-CORE Implementation:
Subscription Status Check:
# .coditect/scripts/subscription_check.py
def check_subscription_status():
"""Check subscription status"""
license_key = load_license_key()
response = requests.get(
f'https://license.coditect.ai/api/v1/subscriptions/{license_key}/status',
headers={'Authorization': f'Bearer {get_auth_token()}'}
)
if response.status_code == 200:
data = response.json()
if data['status'] == 'active':
print(f"✓ Subscription active until {data['current_period_end']}")
return True
elif data['status'] == 'past_due':
days_remaining = data['grace_period_days_remaining']
print(f"⚠️ Payment failed - {days_remaining} days until downgrade")
print(f" Update payment: {data['billing_portal_url']}")
return True # Still active during grace period
elif data['status'] == 'canceled':
print(f"❌ Subscription canceled - expires {data['cancel_at']}")
if data['access_until'] > now():
print(f" {days_until(data['access_until'])} days remaining")
return True
else:
print(f" Downgraded to free tier")
return False
else:
print(f"❌ Subscription inactive: {data['status']}")
return False
else:
print(f"❌ Subscription check failed: {response.status_code}")
return False
API Endpoint:
GET /api/v1/subscriptions/{license_key}/status
Response 200 OK (active):
{
"license_key": "SUB-XXXX-XXXX-XXXX",
"status": "active",
"tier": "pro",
"current_period_start": "2025-11-01T00:00:00Z",
"current_period_end": "2025-12-01T00:00:00Z",
"next_billing_date": "2025-12-01T00:00:00Z",
"amount": 2900, // $29.00 in cents
"currency": "usd",
"billing_cycle": "monthly",
"cancel_at_period_end": false
}
Response 200 OK (past_due):
{
"license_key": "SUB-XXXX-XXXX-XXXX",
"status": "past_due",
"tier": "pro",
"payment_failed_at": "2025-12-01T00:00:00Z",
"grace_period_days": 7,
"grace_period_days_remaining": 5,
"downgrade_at": "2025-12-08T00:00:00Z",
"billing_portal_url": "https://billing.stripe.com/p/session_xxx"
}
Response 200 OK (canceled):
{
"license_key": "SUB-XXXX-XXXX-XXXX",
"status": "canceled",
"tier": "pro",
"canceled_at": "2025-11-20T00:00:00Z",
"cancel_at": "2025-12-01T00:00:00Z",
"access_until": "2025-12-01T00:00:00Z",
"days_remaining": 1
}
Stripe Integration (Webhook):
# backend/licenses/webhooks.py
@csrf_exempt
@require_POST
def stripe_webhook(request):
"""Handle Stripe webhooks for subscription lifecycle"""
payload = request.body
sig_header = request.META['HTTP_STRIPE_SIGNATURE']
try:
event = stripe.Webhook.construct_event(
payload, sig_header, settings.STRIPE_WEBHOOK_SECRET
)
except ValueError:
return HttpResponse(status=400)
except stripe.error.SignatureVerificationError:
return HttpResponse(status=400)
# Handle subscription events
if event['type'] == 'customer.subscription.created':
handle_subscription_created(event['data']['object'])
elif event['type'] == 'customer.subscription.updated':
handle_subscription_updated(event['data']['object'])
elif event['type'] == 'customer.subscription.deleted':
handle_subscription_deleted(event['data']['object'])
elif event['type'] == 'invoice.payment_succeeded':
handle_payment_succeeded(event['data']['object'])
elif event['type'] == 'invoice.payment_failed':
handle_payment_failed(event['data']['object'])
return HttpResponse(status=200)
def handle_payment_failed(invoice):
"""Handle failed payment - start grace period"""
subscription_id = invoice['subscription']
subscription = stripe.Subscription.retrieve(subscription_id)
# Get license from Stripe metadata
license_key = subscription.metadata.get('license_key')
license_obj = License.objects.get(license_key=license_key)
# Update subscription status
license_obj.metadata['subscription_status'] = 'past_due'
license_obj.metadata['payment_failed_at'] = timezone.now().isoformat()
license_obj.metadata['grace_period_days'] = 7
license_obj.save()
# Send email notification
send_payment_failed_email(license_obj)
def handle_subscription_deleted(subscription):
"""Handle subscription cancellation"""
license_key = subscription.metadata.get('license_key')
license_obj = License.objects.get(license_key=license_key)
# Update subscription status
license_obj.metadata['subscription_status'] = 'canceled'
license_obj.metadata['canceled_at'] = timezone.now().isoformat()
# Set expiration to end of current period
license_obj.expires_at = datetime.fromtimestamp(
subscription['current_period_end'],
tz=timezone.utc
)
license_obj.save()
# Send cancellation confirmation
send_subscription_canceled_email(license_obj)
Data Model:
# backend/licenses/models.py
class License(TenantModel):
# ... (existing fields)
# Subscription-specific
billing_cycle = models.CharField(
max_length=16,
choices=[('monthly', 'Monthly'), ('annual', 'Annual')],
null=True,
blank=True
)
current_period_start = models.DateTimeField(null=True, blank=True)
current_period_end = models.DateTimeField(null=True, blank=True)
# Metadata includes Stripe subscription details
# {
# 'subscription_id': 'sub_xxx',
# 'subscription_status': 'active',
# 'customer_id': 'cus_xxx',
# 'price_id': 'price_xxx'
# }
Part 2: Licensing Models (6-10)
6. Trial Licenses
Definition: Time-limited trial with full or limited features. Auto-converts to paid or expires. Graceful degradation on trial expiration.
Use Case: 14-day free trial with Pro features. On day 14, prompt user to upgrade. If no upgrade, downgrade to free tier.
CODITECT-CORE Implementation:
Trial Activation:
# .coditect/scripts/trial_activation.py
def activate_trial():
"""Activate trial license"""
user_email = get_user_email()
response = requests.post(
'https://license.coditect.ai/api/v1/trials/activate',
json={
'user_email': user_email,
'hardware_id': get_hardware_id(),
'trial_type': 'pro_14_day'
}
)
if response.status_code == 200:
data = response.json()
# Save trial license
save_license_key(data['license_key'])
print(f"✓ Trial activated - {data['trial_days']} days remaining")
print(f" Expires: {data['trial_end_date']}")
print(f" Features: {', '.join(data['features'])}")
return True
elif response.status_code == 409:
data = response.json()
print(f"❌ Trial already used on this hardware")
print(f" Previous trial: {data['previous_trial_date']}")
print(f" Upgrade to Pro: https://coditect.ai/upgrade")
return False
else:
print(f"❌ Trial activation failed: {response.status_code}")
return False
Trial Expiration Check:
# .coditect/scripts/trial_check.py
def check_trial_status():
"""Check trial status and prompt upgrade"""
license_key = load_license_key()
response = requests.get(
f'https://license.coditect.ai/api/v1/trials/{license_key}/status',
headers={'Authorization': f'Bearer {get_auth_token()}'}
)
if response.status_code == 200:
data = response.json()
if data['status'] == 'active':
days_remaining = data['days_remaining']
if days_remaining <= 3:
# Urgency messaging
print(f"⚠️ Trial expires in {days_remaining} days!")
print(f" Upgrade now to keep Pro features: https://coditect.ai/upgrade")
else:
print(f"✓ Trial active - {days_remaining} days remaining")
return True
elif data['status'] == 'expired':
print(f"❌ Trial expired on {data['trial_end_date']}")
print(f" Downgrading to Free tier")
print(f" Upgrade to continue: https://coditect.ai/upgrade")
# Automatic downgrade
downgrade_to_free_tier()
return False
else:
return False
else:
print(f"❌ Trial status check failed: {response.status_code}")
return False
API Endpoint:
POST /api/v1/trials/activate
{
"user_email": "dev@acme.com",
"hardware_id": "sha256(...)",
"trial_type": "pro_14_day"
}
Response 200 OK:
{
"license_key": "TRIAL-XXXX-XXXX-XXXX",
"trial_type": "pro_14_day",
"trial_start_date": "2025-11-30T00:00:00Z",
"trial_end_date": "2025-12-14T23:59:59Z",
"trial_days": 14,
"days_remaining": 14,
"features": {
"agents": "*",
"commands": "*",
"skills": "*",
"max_projects": -1,
"offline_grace_hours": 72
},
"auto_convert": false,
"upgrade_url": "https://coditect.ai/upgrade?trial=TRIAL-XXXX"
}
Response 409 Conflict (trial already used):
{
"error": "trial_already_used",
"message": "Trial already activated on this hardware",
"hardware_id": "sha256(...)",
"previous_trial_date": "2025-10-15T00:00:00Z",
"previous_trial_expired": "2025-10-29T23:59:59Z"
}
Trial Status Endpoint:
GET /api/v1/trials/{license_key}/status
Response 200 OK (active):
{
"license_key": "TRIAL-XXXX-XXXX-XXXX",
"status": "active",
"trial_start_date": "2025-11-30T00:00:00Z",
"trial_end_date": "2025-12-14T23:59:59Z",
"days_remaining": 10,
"hours_remaining": 240,
"percent_remaining": 71.4,
"upgrade_url": "https://coditect.ai/upgrade?trial=TRIAL-XXXX"
}
Response 200 OK (expired):
{
"license_key": "TRIAL-XXXX-XXXX-XXXX",
"status": "expired",
"trial_start_date": "2025-11-16T00:00:00Z",
"trial_end_date": "2025-11-30T23:59:59Z",
"days_expired": 5,
"auto_downgrade": true,
"current_tier": "free",
"upgrade_url": "https://coditect.ai/upgrade"
}
Data Model:
# backend/licenses/models.py
class TrialLicense(TenantModel):
"""Trial license record"""
tenant = models.ForeignKey(Tenant, on_delete=models.CASCADE)
license = models.ForeignKey(License, on_delete=models.CASCADE)
# Trial configuration
trial_type = models.CharField(max_length=32) # 'pro_14_day', 'team_7_day'
trial_start_date = models.DateTimeField()
trial_end_date = models.DateTimeField()
trial_days = models.IntegerField()
# Hardware binding (prevent multiple trials)
hardware_id = models.CharField(max_length=64, db_index=True)
user_email = models.EmailField()
# Conversion tracking
converted_to_paid = models.BooleanField(default=False)
converted_at = models.DateTimeField(null=True, blank=True)
converted_license_key = models.CharField(max_length=64, null=True, blank=True)
# Metadata
metadata = models.JSONField(default=dict)
class Meta:
db_table = 'trial_licenses'
indexes = [
models.Index(fields=['hardware_id']),
models.Index(fields=['trial_end_date']),
]
Django View:
# backend/licenses/views.py
@api_view(['POST'])
def activate_trial(request):
"""Activate trial license"""
serializer = TrialActivationSerializer(data=request.data)
serializer.is_valid(raise_exception=True)
hardware_id = serializer.validated_data['hardware_id']
trial_type = serializer.validated_data['trial_type']
# Check if hardware already used trial
previous_trial = TrialLicense.objects.filter(
hardware_id=hardware_id
).first()
if previous_trial:
return Response({
'error': 'trial_already_used',
'message': 'Trial already activated on this hardware',
'previous_trial_date': previous_trial.trial_start_date,
'previous_trial_expired': previous_trial.trial_end_date
}, status=409)
# Create trial license
trial_config = TRIAL_CONFIGS[trial_type] # {'days': 14, 'tier': 'pro'}
license_obj = License.objects.create(
tenant=request.tenant,
license_key=f"TRIAL-{uuid.uuid4().hex[:12]}",
license_type='trial',
expires_at=timezone.now() + timedelta(days=trial_config['days']),
features=FEATURE_MATRIX[trial_config['tier']],
metadata={'tier': trial_config['tier']}
)
trial = TrialLicense.objects.create(
tenant=request.tenant,
license=license_obj,
trial_type=trial_type,
trial_start_date=timezone.now(),
trial_end_date=timezone.now() + timedelta(days=trial_config['days']),
trial_days=trial_config['days'],
hardware_id=hardware_id,
user_email=serializer.validated_data['user_email']
)
return Response({
'license_key': license_obj.license_key,
'trial_type': trial_type,
'trial_start_date': trial.trial_start_date,
'trial_end_date': trial.trial_end_date,
'trial_days': trial.trial_days,
'days_remaining': trial.trial_days,
'features': license_obj.features,
'upgrade_url': f'https://coditect.ai/upgrade?trial={license_obj.license_key}'
})
Automatic Expiration (Celery Task):
# backend/licenses/tasks.py
@celery_app.task
def expire_trials():
"""Automatically expire trials and downgrade to free"""
# Find expired trials
expired_trials = TrialLicense.objects.filter(
trial_end_date__lt=timezone.now(),
converted_to_paid=False
)
for trial in expired_trials:
license_obj = trial.license
# Update license to free tier
license_obj.features = FEATURE_MATRIX['free']
license_obj.metadata['tier'] = 'free'
license_obj.metadata['previous_tier'] = license_obj.metadata.get('tier', 'unknown')
license_obj.metadata['downgraded_from_trial'] = True
license_obj.metadata['downgraded_at'] = timezone.now().isoformat()
license_obj.save()
# Send trial expiration email
send_trial_expired_email(trial)
logger.info(f"Expired trial {license_obj.license_key} - downgraded to free")
7. Usage-Based Licenses
Definition: License charged based on usage metrics (API calls, compute hours, data processed). Metering and billing integration.
Use Case: Runtime license for embedding CODITECT in customer products. Charge $0.10 per 1,000 API calls + $0.01 per agent invocation.
CODITECT-CORE Implementation:
Usage Tracking:
# .coditect/scripts/usage_meter.py
def track_usage_event(event_type, quantity=1, metadata=None):
"""Track usage event for metering"""
license_key = load_license_key()
response = requests.post(
'https://license.coditect.ai/api/v1/usage/track',
json={
'license_key': license_key,
'event_type': event_type, # 'api_call', 'agent_invocation', 'compute_seconds'
'quantity': quantity,
'timestamp': datetime.now(timezone.utc).isoformat(),
'metadata': metadata or {}
},
headers={'Authorization': f'Bearer {get_auth_token()}'}
)
if response.status_code == 200:
return True
else:
# Log locally for later sync
log_usage_event_offline(event_type, quantity, metadata)
return False
# Example: Track agent invocation
def invoke_agent(agent_name):
"""Invoke agent with usage tracking"""
# Track usage before invocation
track_usage_event('agent_invocation', quantity=1, metadata={
'agent_name': agent_name,
'project_root': get_project_root(),
'session_id': get_session_id()
})
# Proceed with agent invocation
return Task(subagent_type=agent_name, ...)
# Example: Track compute time
def track_compute_session():
"""Track compute time for billing"""
session_start = time.time()
# User's CODITECT session
# ...
session_end = time.time()
compute_seconds = int(session_end - session_start)
track_usage_event('compute_seconds', quantity=compute_seconds, metadata={
'session_id': get_session_id()
})
Usage Query:
# .coditect/scripts/usage_query.py
def get_usage_summary(period='current_month'):
"""Get usage summary for billing period"""
license_key = load_license_key()
response = requests.get(
f'https://license.coditect.ai/api/v1/usage/{license_key}/summary',
params={'period': period},
headers={'Authorization': f'Bearer {get_auth_token()}'}
)
if response.status_code == 200:
data = response.json()
print(f"Usage Summary ({period}):")
print(f" API Calls: {data['api_calls']:,}")
print(f" Agent Invocations: {data['agent_invocations']:,}")
print(f" Compute Hours: {data['compute_hours']:.2f}")
print(f" Estimated Cost: ${data['estimated_cost']:.2f}")
return data
else:
print(f"❌ Usage query failed: {response.status_code}")
return None
API Endpoint:
POST /api/v1/usage/track
{
"license_key": "RUNTIME-XXXX-XXXX-XXXX",
"event_type": "agent_invocation",
"quantity": 1,
"timestamp": "2025-11-30T12:34:56Z",
"metadata": {
"agent_name": "general-purpose",
"project_root": "/path/to/project",
"session_id": "sha256(...)"
}
}
Response 200 OK:
{
"tracked": true,
"event_id": "evt_abc123",
"license_key": "RUNTIME-XXXX-XXXX-XXXX",
"event_type": "agent_invocation",
"quantity": 1,
"timestamp": "2025-11-30T12:34:56Z",
"billing_period": "2025-11",
"period_total": 1234 // Total events this period
}
Usage Summary Endpoint:
GET /api/v1/usage/{license_key}/summary?period=current_month
Response 200 OK:
{
"license_key": "RUNTIME-XXXX-XXXX-XXXX",
"period": "2025-11",
"period_start": "2025-11-01T00:00:00Z",
"period_end": "2025-11-30T23:59:59Z",
"usage": {
"api_calls": 45678,
"agent_invocations": 1234,
"compute_hours": 123.45
},
"pricing": {
"api_calls_per_1000": 0.10,
"agent_invocation_each": 0.01,
"compute_hour": 0.50
},
"costs": {
"api_calls_cost": 4.57, // 45,678 / 1000 * $0.10
"agent_invocations_cost": 12.34, // 1,234 * $0.01
"compute_hours_cost": 61.73, // 123.45 * $0.50
"total": 78.64
},
"estimated_cost": 78.64,
"next_billing_date": "2025-12-01T00:00:00Z"
}
Data Model:
# backend/licenses/models.py
class UsageEvent(TenantModel):
"""Usage tracking event"""
tenant = models.ForeignKey(Tenant, on_delete=models.CASCADE)
license = models.ForeignKey(License, on_delete=models.CASCADE)
event_id = models.CharField(max_length=64, unique=True, db_index=True)
event_type = models.CharField(max_length=32, choices=[
('api_call', 'API Call'),
('agent_invocation', 'Agent Invocation'),
('compute_seconds', 'Compute Seconds'),
('data_processed_mb', 'Data Processed (MB)'),
])
quantity = models.IntegerField()
timestamp = models.DateTimeField(db_index=True)
# Billing period (year-month)
billing_period = models.CharField(max_length=7, db_index=True) # '2025-11'
# Metadata (JSON)
metadata = models.JSONField(default=dict)
class Meta:
db_table = 'usage_events'
indexes = [
models.Index(fields=['tenant', 'license', 'billing_period']),
models.Index(fields=['event_type', 'timestamp']),
]
class UsageSummary(TenantModel):
"""Pre-aggregated usage summary by period"""
tenant = models.ForeignKey(Tenant, on_delete=models.CASCADE)
license = models.ForeignKey(License, on_delete=models.CASCADE)
billing_period = models.CharField(max_length=7, db_index=True) # '2025-11'
# Aggregated counts
api_calls = models.BigIntegerField(default=0)
agent_invocations = models.BigIntegerField(default=0)
compute_seconds = models.BigIntegerField(default=0)
# Computed costs (in cents)
api_calls_cost_cents = models.IntegerField(default=0)
agent_invocations_cost_cents = models.IntegerField(default=0)
compute_cost_cents = models.IntegerField(default=0)
total_cost_cents = models.IntegerField(default=0)
# Synced to billing
synced_to_stripe = models.BooleanField(default=False)
synced_at = models.DateTimeField(null=True, blank=True)
class Meta:
db_table = 'usage_summaries'
unique_together = [['tenant', 'license', 'billing_period']]
Django View:
# backend/licenses/views.py
@api_view(['POST'])
@permission_classes([IsAuthenticated])
def track_usage(request):
"""Track usage event"""
serializer = UsageEventSerializer(data=request.data)
serializer.is_valid(raise_exception=True)
license_key = serializer.validated_data['license_key']
event_type = serializer.validated_data['event_type']
quantity = serializer.validated_data['quantity']
timestamp = serializer.validated_data['timestamp']
# Get license
license_obj = License.objects.get(
tenant=request.tenant,
license_key=license_key
)
# Create usage event
billing_period = timestamp.strftime('%Y-%m')
event = UsageEvent.objects.create(
tenant=request.tenant,
license=license_obj,
event_id=f"evt_{uuid.uuid4().hex[:12]}",
event_type=event_type,
quantity=quantity,
timestamp=timestamp,
billing_period=billing_period,
metadata=serializer.validated_data.get('metadata', {})
)
# Update usage summary (atomic)
summary, created = UsageSummary.objects.get_or_create(
tenant=request.tenant,
license=license_obj,
billing_period=billing_period
)
# Increment counters
if event_type == 'api_call':
summary.api_calls += quantity
elif event_type == 'agent_invocation':
summary.agent_invocations += quantity
elif event_type == 'compute_seconds':
summary.compute_seconds += quantity
# Recalculate costs
summary.api_calls_cost_cents = int(summary.api_calls / 1000 * 10) # $0.10 per 1K
summary.agent_invocations_cost_cents = summary.agent_invocations * 1 # $0.01 each
summary.compute_cost_cents = int(summary.compute_seconds / 3600 * 50) # $0.50 per hour
summary.total_cost_cents = (
summary.api_calls_cost_cents +
summary.agent_invocations_cost_cents +
summary.compute_cost_cents
)
summary.save()
return Response({
'tracked': True,
'event_id': event.event_id,
'billing_period': billing_period,
'period_total': get_period_total(license_obj, billing_period, event_type)
})
Stripe Usage-Based Billing Integration:
# backend/licenses/tasks.py
@celery_app.task
def sync_usage_to_stripe():
"""Sync usage summaries to Stripe for billing"""
# Find unsynced usage summaries for current period
current_period = timezone.now().strftime('%Y-%m')
summaries = UsageSummary.objects.filter(
billing_period=current_period,
synced_to_stripe=False
)
for summary in summaries:
license_obj = summary.license
# Get Stripe subscription from license metadata
subscription_id = license_obj.metadata.get('subscription_id')
if not subscription_id:
continue
# Report usage to Stripe
try:
# Create usage record
stripe.SubscriptionItem.create_usage_record(
subscription_item_id=license_obj.metadata['subscription_item_id'],
quantity=summary.total_cost_cents, # Report total cost in cents
timestamp=int(timezone.now().timestamp()),
action='set'
)
# Mark as synced
summary.synced_to_stripe = True
summary.synced_at = timezone.now()
summary.save()
logger.info(f"Synced usage for {license_obj.license_key}: ${summary.total_cost_cents / 100:.2f}")
except stripe.error.StripeError as e:
logger.error(f"Stripe sync failed for {license_obj.license_key}: {e}")
8. Version-Based Licenses
Definition: License tied to specific major version(s) of CODITECT. Upgrade to new major version requires new license or upgrade fee.
Use Case: License for CODITECT 1.x. When CODITECT 2.0 releases, user must upgrade license or stay on 1.x.
CODITECT-CORE Implementation:
Version Validation:
# .coditect/scripts/version_check.py
def validate_version():
"""Check if current version is covered by license"""
coditect_version = get_coditect_version() # e.g., '1.2.5'
major_version = int(coditect_version.split('.')[0]) # 1
license_key = load_license_key()
response = requests.get(
f'https://license.coditect.ai/api/v1/licenses/{license_key}/version-support',
headers={'Authorization': f'Bearer {get_auth_token()}'}
)
if response.status_code == 200:
data = response.json()
supported_versions = data['supported_major_versions'] # [1, 2]
if major_version in supported_versions:
print(f"✓ CODITECT {coditect_version} supported by license")
return True
else:
print(f"❌ CODITECT {coditect_version} not supported by license")
print(f" Supported versions: {', '.join([f'{v}.x' for v in supported_versions])}")
print(f" Upgrade license: https://coditect.ai/upgrade-version")
return False
else:
print(f"❌ Version check failed: {response.status_code}")
return False
Version Enforcement:
# .coditect/scripts/init.sh (embedded in license validation)
# After license validation
CODITECT_VERSION=$(get_coditect_version)
CODITECT_MAJOR=$(echo $CODITECT_VERSION | cut -d. -f1)
# Check version support
python3 .coditect/scripts/version_check.py --major $CODITECT_MAJOR
if [ $? -ne 0 ]; then
echo "❌ Current version not supported by license"
echo " Downgrade to supported version or upgrade license"
exit 1
fi
API Endpoint:
GET /api/v1/licenses/{license_key}/version-support
Response 200 OK:
{
"license_key": "PRO-XXXX-XXXX-XXXX",
"license_type": "version_based",
"supported_major_versions": [1, 2],
"latest_supported_version": "2.5.3",
"upgrade_available": false,
"upgrade_url": null
}
Response 200 OK (upgrade available):
{
"license_key": "PRO-XXXX-XXXX-XXXX",
"license_type": "version_based",
"supported_major_versions": [1],
"latest_supported_version": "1.9.7",
"upgrade_available": true,
"upgrade_url": "https://coditect.ai/upgrade-version?from=1&to=2",
"upgrade_cost": 29.00,
"new_major_version": 2,
"new_major_version_features": [
"Distributed multi-agent orchestration",
"Enhanced security with SSO",
"New runtime embedding capabilities"
]
}
Data Model:
# backend/licenses/models.py
class License(TenantModel):
# ... (existing fields)
# Version-based licensing
supported_major_versions = models.JSONField(default=list) # [1, 2]
version_upgrade_eligible = models.BooleanField(default=False)
def is_version_supported(self, version_string):
"""Check if version is supported by license"""
major_version = int(version_string.split('.')[0])
return major_version in self.supported_major_versions
Django View:
# backend/licenses/views.py
@api_view(['GET'])
@permission_classes([IsAuthenticated])
def get_version_support(request, license_key):
"""Get version support info for license"""
license_obj = License.objects.get(
tenant=request.tenant,
license_key=license_key
)
# Get latest version from metadata
latest_version = '2.5.3' # Could query from release API
latest_major = int(latest_version.split('.')[0])
# Check if upgrade available
upgrade_available = latest_major not in license_obj.supported_major_versions
response_data = {
'license_key': license_key,
'license_type': 'version_based',
'supported_major_versions': license_obj.supported_major_versions,
'latest_supported_version': latest_version if not upgrade_available else f"{max(license_obj.supported_major_versions)}.9.7",
'upgrade_available': upgrade_available
}
if upgrade_available:
response_data['upgrade_url'] = f'https://coditect.ai/upgrade-version?from={max(license_obj.supported_major_versions)}&to={latest_major}'
response_data['upgrade_cost'] = 29.00
response_data['new_major_version'] = latest_major
return Response(response_data)
9. Perpetual Licenses
Definition: One-time purchase, lifetime access to specific version(s). No recurring fees. Optional upgrade/support subscription.
Use Case: Purchase CODITECT 1.x perpetual license for $299. Use forever, but version 2.0 requires upgrade purchase.
CODITECT-CORE Implementation:
Perpetual License Validation:
# .coditect/scripts/perpetual_check.py
def validate_perpetual_license():
"""Validate perpetual license (offline-friendly)"""
license_key = load_license_key()
# Try online validation first
try:
response = requests.get(
f'https://license.coditect.ai/api/v1/licenses/{license_key}/perpetual-status',
headers={'Authorization': f'Bearer {get_auth_token()}'},
timeout=5
)
if response.status_code == 200:
data = response.json()
# Cache license data for offline use
cache_perpetual_license(data)
print(f"✓ Perpetual license validated")
print(f" Version: {data['licensed_versions']}")
print(f" Support until: {data['support_end_date'] or 'No support'}")
return True
except requests.exceptions.RequestException:
# Offline mode - validate from cache
pass
# Validate from cache
cached_license = load_cached_perpetual_license()
if cached_license:
# Verify signature (Cloud KMS public key)
if verify_license_signature(cached_license):
print(f"✓ Perpetual license valid (offline)")
return True
print(f"❌ Perpetual license invalid or missing")
return False
API Endpoint:
GET /api/v1/licenses/{license_key}/perpetual-status
Response 200 OK:
{
"license_key": "PERP-XXXX-XXXX-XXXX",
"license_type": "perpetual",
"purchased_date": "2025-11-30T00:00:00Z",
"licensed_versions": "1.x",
"supported_major_versions": [1],
"perpetual": true,
"never_expires": true,
"support_included": false,
"support_end_date": null,
"upgrade_available": true,
"upgrade_to_version": "2.x",
"upgrade_cost": 149.00,
"signature": "base64(RSA-4096-signature)"
}
Response 200 OK (with support subscription):
{
"license_key": "PERP-XXXX-XXXX-XXXX",
"license_type": "perpetual",
"purchased_date": "2025-11-30T00:00:00Z",
"licensed_versions": "1.x",
"supported_major_versions": [1],
"perpetual": true,
"never_expires": true,
"support_included": true,
"support_start_date": "2025-11-30T00:00:00Z",
"support_end_date": "2026-11-30T00:00:00Z",
"support_days_remaining": 365,
"support_renewal_cost": 49.00,
"signature": "base64(RSA-4096-signature)"
}
Data Model:
# backend/licenses/models.py
class License(TenantModel):
# ... (existing fields)
# Perpetual licensing
perpetual = models.BooleanField(default=False)
purchased_date = models.DateTimeField(null=True, blank=True)
# Support subscription (optional for perpetual)
support_included = models.BooleanField(default=False)
support_start_date = models.DateTimeField(null=True, blank=True)
support_end_date = models.DateTimeField(null=True, blank=True)
Django View:
# backend/licenses/views.py
@api_view(['GET'])
@permission_classes([IsAuthenticated])
def get_perpetual_status(request, license_key):
"""Get perpetual license status"""
license_obj = License.objects.get(
tenant=request.tenant,
license_key=license_key,
perpetual=True
)
# Generate signed license data
license_data = {
'license_key': license_key,
'license_type': 'perpetual',
'purchased_date': license_obj.purchased_date.isoformat(),
'licensed_versions': '1.x',
'supported_major_versions': license_obj.supported_major_versions,
'perpetual': True,
'never_expires': True,
'support_included': license_obj.support_included
}
if license_obj.support_included:
license_data['support_start_date'] = license_obj.support_start_date.isoformat()
license_data['support_end_date'] = license_obj.support_end_date.isoformat()
days_remaining = (license_obj.support_end_date - timezone.now()).days
license_data['support_days_remaining'] = days_remaining
# Sign with Cloud KMS
signature = sign_license_with_kms(license_data)
license_data['signature'] = signature
return Response(license_data)
10. Expiration-Based Licenses
Definition: Fixed-term license (not recurring). Expires on specific date. No auto-renewal.
Use Case: 6-month project license. Expires June 1, 2026. User must manually renew if needed.
CODITECT-CORE Implementation:
Expiration Check:
# .coditect/scripts/expiration_check.py
def check_license_expiration():
"""Check if license has expired"""
license_key = load_license_key()
response = requests.get(
f'https://license.coditect.ai/api/v1/licenses/{license_key}/expiration',
headers={'Authorization': f'Bearer {get_auth_token()}'}
)
if response.status_code == 200:
data = response.json()
if data['status'] == 'active':
days_remaining = data['days_until_expiration']
if days_remaining <= 30:
print(f"⚠️ License expires in {days_remaining} days")
print(f" Renew before {data['expiration_date']}")
print(f" Renew: {data['renewal_url']}")
else:
print(f"✓ License active - expires {data['expiration_date']}")
return True
elif data['status'] == 'expired':
print(f"❌ License expired on {data['expiration_date']}")
print(f" Days expired: {data['days_expired']}")
print(f" Renew: {data['renewal_url']}")
return False
else:
return False
else:
print(f"❌ Expiration check failed: {response.status_code}")
return False
API Endpoint:
GET /api/v1/licenses/{license_key}/expiration
Response 200 OK (active):
{
"license_key": "TERM-XXXX-XXXX-XXXX",
"status": "active",
"expiration_date": "2026-06-01T00:00:00Z",
"days_until_expiration": 183,
"percent_remaining": 50.4,
"renewal_url": "https://coditect.ai/renew?license=TERM-XXXX",
"renewal_cost": 299.00,
"renewal_term": "6 months"
}
Response 200 OK (expired):
{
"license_key": "TERM-XXXX-XXXX-XXXX",
"status": "expired",
"expiration_date": "2025-11-15T00:00:00Z",
"days_expired": 15,
"grace_period": false,
"renewal_url": "https://coditect.ai/renew?license=TERM-XXXX",
"renewal_cost": 299.00,
"renewal_term": "6 months"
}
Data Model:
# backend/licenses/models.py
class License(TenantModel):
# ... (existing fields)
# Expiration
expires_at = models.DateTimeField(null=True, blank=True)
expiration_notified = models.BooleanField(default=False)
expiration_notification_sent_at = models.DateTimeField(null=True, blank=True)
def is_expired(self):
"""Check if license is expired"""
if not self.expires_at:
return False
return timezone.now() > self.expires_at
def days_until_expiration(self):
"""Get days until expiration"""
if not self.expires_at:
return None
delta = self.expires_at - timezone.now()
return delta.days
Django View:
# backend/licenses/views.py
@api_view(['GET'])
@permission_classes([IsAuthenticated])
def get_expiration_status(request, license_key):
"""Get license expiration status"""
license_obj = License.objects.get(
tenant=request.tenant,
license_key=license_key
)
if not license_obj.expires_at:
return Response({
'error': 'license_does_not_expire',
'message': 'This license does not have an expiration date'
}, status=400)
now = timezone.now()
is_expired = license_obj.is_expired()
response_data = {
'license_key': license_key,
'expiration_date': license_obj.expires_at.isoformat(),
'renewal_url': f'https://coditect.ai/renew?license={license_key}'
}
if is_expired:
days_expired = (now - license_obj.expires_at).days
response_data['status'] = 'expired'
response_data['days_expired'] = days_expired
else:
days_remaining = license_obj.days_until_expiration()
response_data['status'] = 'active'
response_data['days_until_expiration'] = days_remaining
# Calculate percent remaining
total_days = (license_obj.expires_at - license_obj.created_at).days
percent_remaining = (days_remaining / total_days) * 100
response_data['percent_remaining'] = round(percent_remaining, 1)
return Response(response_data)
Expiration Notification (Celery Task):
# backend/licenses/tasks.py
@celery_app.task
def send_expiration_notifications():
"""Send expiration warnings 30, 14, 7, 1 days before expiration"""
notification_days = [30, 14, 7, 1]
for days in notification_days:
# Find licenses expiring in exactly `days` days
target_date = timezone.now() + timedelta(days=days)
licenses = License.objects.filter(
expires_at__date=target_date.date(),
expiration_notified=False
)
for license_obj in licenses:
# Send email notification
send_expiration_warning_email(
license_obj,
days_remaining=days
)
# Mark as notified (for this specific day)
license_obj.expiration_notification_sent_at = timezone.now()
license_obj.save()
logger.info(f"Sent {days}-day expiration warning for {license_obj.license_key}")
Part 3: CODITECT-CORE Enforcement Architecture
Overview: Symlink-Based Framework Licensing
The Challenge:
CODITECT-CORE is installed as a git submodule with a symlink chain architecture. This presents unique licensing challenges:
- Physical vs. Logical Paths: Multiple symlinks (
submodule-1/.coditect,submodule-2/.coditect) point to the same physical directory (.coditect/) - Session Identification: Should each symlink count as a separate session or one session?
- Fair Pricing: Don't charge per-symlink, charge per-project
- Deduplication: Avoid counting the same developer/project multiple times
- Dual Usage: Builder license (development tool) vs. Runtime license (embedded in products)
The Solution:
- Symlink Resolution: Use
os.path.realpath('.coditect')to resolve to physical directory - Session Deduplication: Hash project_root + coditect_path (resolved) + user + hardware
- Fair Pricing: 1 project = 1 session, regardless of symlink count
- Check-on-Init: Validate license when
.coditect/scripts/init.shruns - Offline-First: Signed licenses cached locally, work without network (24-168h grace period)
Session Identification Strategy
Problem:
PROJECT_ROOT/
├── .coditect/ # Physical directory (1x)
├── .claude -> .coditect # Symlink (resolved to .coditect/)
├── submodule-1/
│ ├── .coditect -> ../.coditect # Symlink (resolved to ../. coditect/)
│ └── .claude -> .coditect
└── submodule-2/
├── .coditect -> ../.coditect # Symlink (resolved to ../.coditect/)
└── .claude -> .coditect
Question: If developer works in submodule-1 and submodule-2, is this 1 session or 2?
Answer: 1 session - same project, same resolved .coditect/ path.
Implementation:
# .coditect/scripts/session_id.py
import os
import hashlib
import json
import subprocess
import platform
def generate_session_id():
"""Generate unique session identifier for CODITECT-CORE"""
# 1. Resolve symlinks to physical path
coditect_path = os.path.realpath('.coditect')
# 2. Get project root (git repository root)
try:
project_root = subprocess.check_output(
['git', 'rev-parse', '--show-toplevel'],
cwd=os.getcwd()
).decode().strip()
except:
# Not in git repo, use current directory
project_root = os.getcwd()
# 3. Get user identification
user_email = get_user_email() # git config user.email
# 4. Get hardware identification
hardware_id = get_hardware_id() # MAC address + CPU ID
# 5. Get CODITECT version
coditect_version = get_coditect_version() # From .coditect/VERSION
# 6. Get usage type (builder vs runtime)
usage_type = 'builder' # or 'runtime' for embedded CODITECT
# 7. Construct session data
session_data = {
'user_email': user_email,
'hardware_id': hardware_id,
'project_root': project_root,
'coditect_path': coditect_path, # RESOLVED PATH (deduplicates symlinks)
'coditect_version': coditect_version,
'usage_type': usage_type
}
# 8. Hash to generate stable session ID
session_id = hashlib.sha256(
json.dumps(session_data, sort_keys=True).encode()
).hexdigest()
return session_id, session_data
# Example outputs:
# Scenario 1: Developer in parent directory
# project_root: /Users/dev/projects/acme-app
# coditect_path: /Users/dev/projects/acme-app/.coditect
# session_id: sha256(user+hardware+/Users/dev/projects/acme-app+/Users/dev/projects/acme-app/.coditect+1.2.0+builder)
# Scenario 2: Developer in submodule-1
# project_root: /Users/dev/projects/acme-app (SAME - git root)
# coditect_path: /Users/dev/projects/acme-app/.coditect (SAME - resolved symlink)
# session_id: sha256(...) - SAME SESSION ID
# Scenario 3: Developer in different project
# project_root: /Users/dev/projects/other-project
# coditect_path: /Users/dev/projects/other-project/.coditect
# session_id: sha256(...) - DIFFERENT SESSION ID (different project_root)
Key Insights:
- Symlink Resolution:
os.path.realpath()resolves all symlinks to physical path - Git Root Detection: Project boundary is git repository root (not current working directory)
- Session Deduplication: Same project + same user + same hardware = same session_id
- Fair Pricing: 1 developer working on 1 project with 100 symlinks = 1 session = 1 seat
Check-on-Init Enforcement Pattern
Pattern: CODITECT validates license when framework initializes (.coditect/scripts/init.sh).
Flow:
# .coditect/scripts/init.sh
#!/bin/bash
# CODITECT Initialization Script
# This script runs when:
# - User opens Claude Code in project directory
# - User runs CODITECT command for first time in session
# - Framework auto-initialization on startup
echo "Initializing CODITECT..."
# Step 1: Check for cached license
if [ -f .coditect/.license-cache ]; then
echo "✓ Found cached license"
# Validate cached license
python3 .coditect/scripts/validate-license.py --cached
if [ $? -eq 0 ]; then
echo "✓ License valid (cached)"
start_heartbeat_background
exit 0
else
echo "⚠️ Cached license invalid or expired"
fi
fi
# Step 2: Online validation
echo "Validating license online..."
python3 .coditect/scripts/validate-license.py --online
if [ $? -eq 0 ]; then
echo "✓ License activated"
start_heartbeat_background
exit 0
fi
# Step 3: Offline grace period
echo "⚠️ Network unavailable - checking offline grace period"
python3 .coditect/scripts/validate-license.py --offline-grace
if [ $? -eq 0 ]; then
echo "⚠️ License valid (offline mode, grace period active)"
exit 0
fi
# Step 4: License invalid
echo "❌ License required to use CODITECT"
echo ""
echo "Options:"
echo " 1. Activate license: https://coditect.ai/activate"
echo " 2. Start free trial: coditect-cli trial start"
echo " 3. Use free tier: coditect-cli downgrade free"
echo ""
exit 1
Validation Logic:
# .coditect/scripts/validate-license.py
import sys
import argparse
from pathlib import Path
def main():
parser = argparse.ArgumentParser()
parser.add_argument('--cached', action='store_true')
parser.add_argument('--online', action='store_true')
parser.add_argument('--offline-grace', action='store_true')
args = parser.parse_args()
if args.cached:
return validate_cached_license()
elif args.online:
return validate_online_license()
elif args.offline_grace:
return validate_offline_grace()
else:
sys.exit(1)
def validate_cached_license():
"""Validate license from local cache"""
cache_path = Path('.coditect/.license-cache')
if not cache_path.exists():
return False
try:
cache_data = json.loads(cache_path.read_text())
# Check expiration
expires_at = datetime.fromisoformat(cache_data['expires_at'])
if datetime.now(timezone.utc) > expires_at:
print("⚠️ Cached license expired")
return False
# Verify signature (Cloud KMS public key)
if not verify_license_signature(cache_data):
print("❌ License signature invalid (tampered)")
return False
# Check version compatibility
if not is_version_supported(cache_data):
print("❌ CODITECT version not supported by license")
return False
# Success
return True
except Exception as e:
print(f"❌ Cache validation error: {e}")
return False
def validate_online_license():
"""Validate license with server"""
# Generate session ID
session_id, session_data = generate_session_id()
# Load license key from config
license_key = load_license_key()
if not license_key:
print("❌ No license key configured")
print(" Run: coditect-cli license set <LICENSE_KEY>")
return False
# Call license server
try:
response = requests.post(
'https://license.coditect.ai/api/v1/licenses/validate',
json={
'license_key': license_key,
'session_id': session_id,
'session_data': session_data
},
headers={'Authorization': f'Bearer {get_api_token()}'},
timeout=10
)
if response.status_code == 200:
data = response.json()
# Save offline cache
cache_license(data)
# Success
print(f"✓ License activated")
print(f" Tier: {data['tier']}")
print(f" Expires: {data.get('expires_at', 'Never')}")
return True
elif response.status_code == 409:
# Pool exhausted (floating licenses)
data = response.json()
print(f"❌ No seats available")
print(f" {data['total_seats']} seats in use")
print(f" Active users: {', '.join([s['user_email'] for s in data['active_sessions'][:5]])}")
return False
elif response.status_code == 403:
# License expired or invalid
data = response.json()
print(f"❌ License invalid: {data['message']}")
return False
else:
print(f"❌ Validation failed: HTTP {response.status_code}")
return False
except requests.exceptions.Timeout:
print("⚠️ License server timeout - trying offline grace period")
return False
except requests.exceptions.ConnectionError:
print("⚠️ Network unavailable - trying offline grace period")
return False
def validate_offline_grace():
"""Validate license using offline grace period"""
cache_path = Path('.coditect/.license-cache')
if not cache_path.exists():
print("❌ No cached license for offline mode")
return False
try:
cache_data = json.loads(cache_path.read_text())
# Check offline grace expiration
offline_expires_at = datetime.fromisoformat(cache_data['offline_expires_at'])
if datetime.now(timezone.utc) > offline_expires_at:
hours_expired = int((datetime.now(timezone.utc) - offline_expires_at).total_seconds() / 3600)
print(f"❌ Offline grace period expired {hours_expired} hours ago")
print(f" Connect to internet to renew license")
return False
# Calculate remaining time
time_remaining = offline_expires_at - datetime.now(timezone.utc)
hours_remaining = int(time_remaining.total_seconds() / 3600)
print(f"✓ License valid (offline mode)")
print(f" Grace period: {hours_remaining} hours remaining")
print(f" Tier: {cache_data['tier']}")
return True
except Exception as e:
print(f"❌ Offline validation error: {e}")
return False
# Helper functions
def cache_license(license_data):
"""Save license data to encrypted cache"""
cache_path = Path('.coditect/.license-cache')
# Encrypt sensitive data
encrypted_cache = {
'license_key': license_data['license_key'],
'tier': license_data['tier'],
'features': license_data['features'],
'expires_at': license_data.get('expires_at'),
'offline_expires_at': license_data['offline_expires_at'],
'signature': license_data['signature'],
'cached_at': datetime.now(timezone.utc).isoformat()
}
cache_path.write_text(json.dumps(encrypted_cache, indent=2))
cache_path.chmod(0o600) # Read/write for owner only
def verify_license_signature(license_data):
"""Verify license signature using Cloud KMS public key"""
# Get Cloud KMS public key (embedded in CODITECT)
public_key = load_kms_public_key()
# Extract signature
signature_b64 = license_data['signature']
signature = base64.b64decode(signature_b64)
# Prepare data for verification (without signature)
verify_data = {k: v for k, v in license_data.items() if k != 'signature'}
verify_bytes = json.dumps(verify_data, sort_keys=True).encode()
# Verify RSA-4096 signature
try:
public_key.verify(
signature,
verify_bytes,
padding.PSS(
mgf=padding.MGF1(hashes.SHA256()),
salt_length=padding.PSS.MAX_LENGTH
),
hashes.SHA256()
)
return True
except:
return False
if __name__ == '__main__':
success = main()
sys.exit(0 if success else 1)
Offline Grace Period Implementation
Tier-Based Grace Periods:
| Tier | Online Heartbeat | Offline Grace | Use Case |
|---|---|---|---|
| Free | Required every 6 min | 24 hours | Casual users, internet cafes |
| Pro | Required every 6 min | 72 hours | Individual developers, occasional offline |
| Team | Required every 6 min | 48 hours | Teams with VPN requirements |
| Enterprise | Required every 6 min | 168 hours (7 days) | Air-gapped environments, field work |
Implementation:
# Server-side: Generate offline cache with grace period
def generate_offline_cache(license_obj, session, grace_hours):
"""Generate signed offline cache for client"""
# Calculate offline expiration
offline_expires_at = timezone.now() + timedelta(hours=grace_hours)
# Prepare license data for client
license_data = {
'license_key': license_obj.license_key,
'tier': license_obj.get_tier(),
'features': license_obj.features,
'expires_at': license_obj.expires_at.isoformat() if license_obj.expires_at else None,
'offline_expires_at': offline_expires_at.isoformat(),
'session_id': session.session_id,
'created_at': timezone.now().isoformat()
}
# Sign with Cloud KMS
signature = sign_license_with_kms(license_data)
license_data['signature'] = signature
return {
'encrypted': encrypt_license_data(license_data),
'expires_at': offline_expires_at.isoformat(),
'features': license_obj.features
}
Client-side: Offline grace validation
# .coditect/scripts/offline_grace.py
def check_offline_grace():
"""Check if offline grace period is valid"""
cache_path = Path('.coditect/.license-cache')
if not cache_path.exists():
return False, "No cached license"
cache_data = json.loads(cache_path.read_text())
# Decrypt and verify
decrypted = decrypt_license_data(cache_data['encrypted'])
if not verify_license_signature(decrypted):
return False, "Invalid signature"
# Check offline expiration
offline_expires_at = datetime.fromisoformat(decrypted['offline_expires_at'])
now = datetime.now(timezone.utc)
if now > offline_expires_at:
hours_expired = int((now - offline_expires_at).total_seconds() / 3600)
return False, f"Grace period expired {hours_expired} hours ago"
# Valid
hours_remaining = int((offline_expires_at - now).total_seconds() / 3600)
return True, f"{hours_remaining} hours remaining"
Feature Gating Implementation
Centralized Feature Matrix:
# .coditect/config/feature_matrix.py
FEATURE_MATRIX = {
'free': {
# Limited agents (5 total)
'agents': [
'general-purpose',
'codebase-locator',
'codebase-analyzer',
'documentation-librarian',
'project-organizer'
],
# Limited commands (10 total)
'commands': [
'/help', '/analyze', '/search', '/document',
'/deliberation', '/implement', '/prototype',
'/optimize', '/strategy', '/action'
],
# Limited skills (5 total)
'skills': [
'search-strategies',
'git-workflow-automation',
'production-patterns',
'framework-patterns',
'documentation-librarian'
],
# Limits
'max_projects': 1,
'max_seats': 1,
'offline_grace_hours': 24,
'team_dashboard': False,
'runtime_embedding': False,
'sso_integration': False,
'private_deployment': False,
'support_tier': 'community'
},
'pro': {
# All agents
'agents': '*', # Expands to all 52 agents
# All commands
'commands': '*', # Expands to all 81 commands
# All skills
'skills': '*', # Expands to all 26 skills
# Limits
'max_projects': -1, # Unlimited
'max_seats': 2, # Individual (laptop + desktop)
'offline_grace_hours': 72, # 3 days
'team_dashboard': False,
'runtime_embedding': False,
'sso_integration': False,
'private_deployment': False,
'support_tier': 'email'
},
'team': {
'agents': '*',
'commands': '*',
'skills': '*',
'max_projects': -1,
'max_seats': 'floating', # 5-100 floating seats
'offline_grace_hours': 48, # 2 days
'team_dashboard': True,
'runtime_embedding': False,
'sso_integration': True,
'private_deployment': False,
'support_tier': 'priority'
},
'enterprise': {
'agents': '*',
'commands': '*',
'skills': '*',
'max_projects': -1,
'max_seats': 'floating', # Unlimited
'offline_grace_hours': 168, # 7 days
'team_dashboard': True,
'runtime_embedding': True,
'sso_integration': True,
'private_deployment': True,
'support_tier': 'dedicated'
}
}
Feature Gate Check (Client-Side):
# .coditect/scripts/feature_gate.py
def check_feature_access(feature_type, feature_name):
"""Check if feature is allowed by license tier"""
# Load license cache
cache_path = Path('.coditect/.license-cache')
if not cache_path.exists():
# No license - default to free tier
tier = 'free'
else:
cache_data = json.loads(cache_path.read_text())
tier = cache_data.get('tier', 'free')
# Get feature matrix for tier
features = FEATURE_MATRIX[tier]
# Check if feature is allowed
allowed = features.get(feature_type, [])
# Wildcard = all allowed
if allowed == '*':
return True
# Check specific feature
return feature_name in allowed
# Integration points:
# 1. Agent invocation
def invoke_agent(agent_name):
"""Invoke agent if licensed"""
if not check_feature_access('agents', agent_name):
print(f"❌ Agent '{agent_name}' not available in {get_license_tier()} tier")
print(f" Upgrade to Pro for access to all 52 agents")
print(f" https://coditect.ai/upgrade")
return False
# Proceed with invocation
return Task(subagent_type=agent_name, ...)
# 2. Command execution
def execute_command(command_name):
"""Execute command if licensed"""
if not check_feature_access('commands', command_name):
print(f"❌ Command '{command_name}' not available in {get_license_tier()} tier")
print(f" Available in Free tier: {', '.join(FEATURE_MATRIX['free']['commands'])}")
print(f" Upgrade to Pro for all 81 commands")
return False
# Proceed with execution
return execute(command_name)
# 3. Skill usage
def use_skill(skill_name):
"""Use skill if licensed"""
if not check_feature_access('skills', skill_name):
print(f"❌ Skill '{skill_name}' not available in {get_license_tier()} tier")
print(f" Upgrade to Pro for access to all 26 skills")
return False
# Proceed with skill usage
return use(skill_name)
Builder vs. Runtime Licensing Model
Dual Usage Model:
CODITECT serves two distinct use cases:
-
Builder License: Development tool (IDE-like)
- Developer uses CODITECT to build software
- Per-developer pricing
- Floating seats for teams
- Offline grace period for travel
-
Runtime License: Embedded service (database-like)
- CODITECT embedded in customer's product
- Per-application + usage-based pricing
- Usage metering (API calls, agent invocations)
- Higher availability requirements
Session Identification with Usage Type:
# Builder usage
session_data = {
'usage_type': 'builder',
'user_email': 'dev@acme.com',
'hardware_id': 'sha256(...)',
'project_root': '/Users/dev/projects/acme-app',
'coditect_path': '/Users/dev/projects/acme-app/.coditect',
'coditect_version': '1.2.0'
}
# Runtime usage (embedded in customer product)
session_data = {
'usage_type': 'runtime',
'application_id': 'customer-app-production',
'application_version': '2.3.1',
'tenant_id': 'acme-corp',
'deployment_environment': 'production',
'coditect_version': '1.2.0'
}
Pricing Models:
| Usage Type | Pricing Model | Typical Cost |
|---|---|---|
| Builder (Free) | Free | $0/month (1 project, 5 agents, 10 commands) |
| Builder (Pro) | Per-developer | $29/month (unlimited projects, all features) |
| Builder (Team) | Floating seats | $99/month (5 seats), $19/seat additional |
| Builder (Enterprise) | Custom | $499+/month (unlimited seats, SSO, private deploy) |
| Runtime | Per-application + usage | $299/month + $0.10/1K API calls + $0.01/agent invocation |
License Server Routing:
# backend/licenses/views.py
@api_view(['POST'])
@permission_classes([IsAuthenticated])
def validate_license(request):
"""Validate license (handles both builder and runtime)"""
serializer = LicenseValidationSerializer(data=request.data)
serializer.is_valid(raise_exception=True)
usage_type = serializer.validated_data['session_data']['usage_type']
if usage_type == 'builder':
# Builder license flow
return validate_builder_license(request, serializer.validated_data)
elif usage_type == 'runtime':
# Runtime license flow
return validate_runtime_license(request, serializer.validated_data)
else:
return Response({
'error': 'invalid_usage_type',
'message': f'Unknown usage type: {usage_type}'
}, status=400)
Heartbeat TTL and Zombie Session Cleanup
Problem: Developer closes laptop without gracefully releasing license seat.
Solution: 6-minute heartbeat TTL with automatic zombie session cleanup.
Client-Side Heartbeat:
# .coditect/scripts/heartbeat.py
import threading
import time
import requests
def start_heartbeat_background(session_id, license_key, ttl=360):
"""Start background heartbeat thread"""
def heartbeat_loop():
while True:
# Sleep for 5 minutes (TTL is 6 minutes)
time.sleep(300)
try:
# Send heartbeat
response = requests.post(
'https://license.coditect.ai/api/v1/licenses/heartbeat',
json={
'session_id': session_id,
'license_key': license_key
},
headers={'Authorization': f'Bearer {get_auth_token()}'},
timeout=10
)
if response.status_code == 200:
print("✓ License heartbeat renewed")
else:
print(f"⚠️ Heartbeat failed: {response.status_code}")
# Continue using offline cache
except Exception as e:
print(f"⚠️ Heartbeat error: {e}")
# Continue using offline cache
# Start daemon thread
thread = threading.Thread(target=heartbeat_loop, daemon=True)
thread.start()
# Graceful shutdown
import signal
import atexit
def release_license_on_exit():
"""Release license seat when CODITECT exits"""
session_id = get_session_id()
license_key = load_license_key()
try:
requests.post(
'https://license.coditect.ai/api/v1/licenses/release',
json={
'session_id': session_id,
'license_key': license_key
},
headers={'Authorization': f'Bearer {get_auth_token()}'},
timeout=5
)
print("✓ License seat released")
except:
# Best effort - TTL will clean up
pass
# Register cleanup handlers
atexit.register(release_license_on_exit)
signal.signal(signal.SIGTERM, lambda s, f: release_license_on_exit())
signal.signal(signal.SIGINT, lambda s, f: release_license_on_exit())
Server-Side Zombie Cleanup (Celery):
# backend/licenses/tasks.py
@celery_app.task
@celery_app.on_after_configure.connect
def setup_periodic_tasks(sender, **kwargs):
"""Setup periodic tasks"""
# Run zombie cleanup every minute
sender.add_periodic_task(60.0, cleanup_zombie_sessions.s(), name='cleanup zombies')
@celery_app.task
def cleanup_zombie_sessions():
"""Release sessions with expired heartbeat TTL"""
# Sessions with last_heartbeat > 6 minutes ago
cutoff = timezone.now() - timedelta(seconds=360)
zombie_sessions = LicenseSession.objects.filter(
released_at__isnull=True,
last_heartbeat__lt=cutoff
)
released_count = 0
for session in zombie_sessions:
# Release seat in Redis (atomic)
redis_client = get_redis_client()
redis_client.hdel(
f"license:{session.license.license_key}:sessions",
session.session_id
)
# Mark session as released in PostgreSQL
session.released_at = timezone.now()
session.save()
released_count += 1
logger.info(f"Released zombie session {session.session_id} (user: {session.user_email})")
if released_count > 0:
logger.info(f"Cleaned up {released_count} zombie sessions")
return released_count
Redis TTL Integration:
-- Redis Lua script: acquire_floating_seat_with_ttl.lua
local license_key = KEYS[1]
local session_id = ARGV[1]
local max_seats = tonumber(ARGV[2])
local ttl = tonumber(ARGV[3]) -- 360 seconds (6 minutes)
-- Check if session already has a seat
local existing_seat = redis.call('HGET', license_key .. ':sessions', session_id)
if existing_seat then
-- Renew TTL
redis.call('EXPIRE', license_key .. ':sessions', ttl)
redis.call('HSET', license_key .. ':last_heartbeat:' .. session_id, 'timestamp', redis.call('TIME')[1])
return {1, tonumber(existing_seat)} -- {acquired, seat_number}
end
-- Count active sessions
local active_count = redis.call('HLEN', license_key .. ':sessions')
if active_count >= max_seats then
return {0, 0} -- {not_acquired, no_seat}
end
-- Assign seat
local seat_number = active_count + 1
redis.call('HSET', license_key .. ':sessions', session_id, seat_number)
redis.call('EXPIRE', license_key .. ':sessions', ttl)
-- Track heartbeat timestamp
redis.call('HSET', license_key .. ':last_heartbeat:' .. session_id, 'timestamp', redis.call('TIME')[1])
redis.call('EXPIRE', license_key .. ':last_heartbeat:' .. session_id, ttl)
return {1, seat_number} -- {acquired, seat_number}
Security Considerations
1. License Signing (Cloud KMS RSA-4096):
# Server-side: Sign license with Cloud KMS
from google.cloud import kms
def sign_license_with_kms(license_data):
"""Sign license data using Cloud KMS"""
kms_client = kms.KeyManagementServiceClient()
# Prepare data for signing
sign_data = json.dumps(license_data, sort_keys=True).encode()
# Sign with Cloud KMS
key_name = 'projects/PROJECT_ID/locations/global/keyRings/license-signing/cryptoKeys/license-key/cryptoKeyVersions/1'
digest = {
'sha256': hashlib.sha256(sign_data).digest()
}
sign_response = kms_client.asymmetric_sign(
request={'name': key_name, 'digest': digest}
)
# Return base64-encoded signature
return base64.b64encode(sign_response.signature).decode()
2. Tamper Detection:
- All license data signed with RSA-4096 (Cloud KMS)
- Client verifies signature before using cached license
- Signature mismatch = tampered license = reject
3. Hardware Binding:
# Prevent license sharing by binding to hardware
def get_hardware_id():
"""Generate stable hardware fingerprint"""
components = []
# MAC address (primary network interface)
mac = ':'.join(['{:02x}'.format((uuid.getnode() >> elements) & 0xff)
for elements in range(0, 2*6, 2)][::-1])
components.append(f"mac:{mac}")
# CPU info (brand string)
if platform.system() == 'Darwin':
cpu = subprocess.check_output(
['sysctl', '-n', 'machdep.cpu.brand_string']
).decode().strip()
components.append(f"cpu:{cpu}")
# Machine UUID (if available)
if platform.system() == 'Darwin':
machine_uuid = subprocess.check_output(
['ioreg', '-rd1', '-c', 'IOPlatformExpertDevice']
).decode()
# Parse UUID from output
components.append(f"uuid:{machine_uuid}")
# Hash combined fingerprint
fingerprint = '|'.join(components)
return hashlib.sha256(fingerprint.encode()).hexdigest()
4. API Authentication:
- All API calls require Bearer token (JWT)
- Tokens expire after 24 hours
- Refresh token rotation
Part 4: Architecture Diagrams, Deployment, and Implementation
C4 Architecture Model
C4 Level 1: System Context
Key Interactions:
- Developer → License Server: Check-on-init validation, heartbeat renewal, seat acquisition
- Admin → License Server: License creation, user management, usage analytics
- Customer Application → License Server: Runtime usage metering, billing events
- License Server → Stripe: Payment processing, subscription lifecycle, usage-based billing
- License Server → SendGrid: Trial notifications, expiration warnings, renewal reminders
- License Server → GCP: License signing (Cloud KMS), audit logging, monitoring
C4 Level 2: Container Diagram
Container Responsibilities:
- FastAPI Application: REST API, license validation, session management, feature entitlements
- Celery Workers: Zombie session cleanup, trial expiration, usage sync to Stripe, email notifications
- Celery Beat: Periodic task scheduler (cleanup every 1 min, trial checks daily)
- PostgreSQL: Persistent storage for licenses, sessions, tenants, usage events
- Redis: Atomic seat counting via Lua scripts, heartbeat tracking, session cache
- Cloud KMS: RSA-4096 license signing for tamper-proof offline licenses
C4 Level 3: Component Diagram - API Application
Component Details:
- Floating License: Acquire/release seats, heartbeat renewal, zombie detection
- Device License: Hardware activation, deactivation, device limit enforcement
- Trial License: Trial activation, expiration checks, automatic downgrade
- Usage Tracking: Event ingestion, usage summaries, billing period aggregation
- Authentication: JWT token validation, refresh token rotation
- Multi-Tenant: Row-level tenant isolation, tenant context injection
- Serializers: Request/response validation, data transformation
- Django Models: ORM models for License, Session, Trial, Usage
- Redis Client: Lua script execution, atomic operations
- KMS Client: License signing with Cloud KMS asymmetric keys
C4 Level 4: Code Diagram - Floating License Acquisition
# API Endpoint
@api_view(['POST'])
@permission_classes([IsAuthenticated])
def acquire_floating_license(request):
# 1. Validate request
serializer = FloatingLicenseAcquireSerializer(data=request.data)
serializer.is_valid(raise_exception=True)
# 2. Extract session data
license_key = serializer.validated_data['license_key']
session_id = serializer.validated_data['session_id']
# 3. Get license from PostgreSQL
license_obj = License.objects.get(
tenant=request.tenant,
license_key=license_key,
license_type='floating'
)
# 4. Atomic seat acquisition (Redis Lua)
redis_client = get_redis_client()
script = redis_client.register_script(ACQUIRE_FLOATING_SEAT_LUA)
acquired, seat_number = script(
keys=[f"license:{license_key}"],
args=[session_id, license_obj.max_seats, 360]
)
# 5. Handle pool exhausted
if not acquired:
return Response({'error': 'no_seats_available'}, status=409)
# 6. Create session record (PostgreSQL)
session = LicenseSession.objects.create(
tenant=request.tenant,
license=license_obj,
session_id=session_id,
seat_number=seat_number,
# ...
)
# 7. Generate offline cache (signed)
offline_cache = generate_offline_cache(license_obj, session)
# 8. Return success
return Response({
'acquired': True,
'seat_number': seat_number,
'offline_cache': offline_cache
})
C4 Level 5: Deployment Diagram
Deployment Specifications:
GKE Cluster:
- Environment: Development (preemptible nodes), Production (standard nodes)
- Node Pool: 3-10 nodes (auto-scaling), n1-standard-2 (2 vCPU, 7.5GB RAM)
- Namespace:
license-server(isolated from other services)
FastAPI Pods:
- Replicas: 3 (HA), auto-scale 3-10 based on CPU utilization
- Resources: Request 1 core / 2GB RAM, Limit 2 cores / 4GB RAM
- Health Checks: Liveness
/health, Readiness/ready
Celery Workers:
- Replicas: 2 (background tasks), auto-scale 2-5 based on queue depth
- Resources: Request 0.5 core / 1GB RAM, Limit 1 core / 2GB RAM
- Concurrency: 4 concurrent tasks per worker
Cloud SQL PostgreSQL:
- Tier: db-custom-2-7680 (2 vCPU, 7.5GB RAM)
- High Availability: Regional (us-central1-a primary, us-central1-b replica)
- Backups: Automated daily, 7-day retention, point-in-time recovery
- Connections: Max 100 connections (connection pooling via PgBouncer)
Redis Memorystore:
- Tier: BASIC (6GB, no HA - acceptable for MVP)
- Persistence: RDB snapshots every 12 hours
- Upgrade Path: STANDARD tier with HA for production
Cloud KMS:
- Algorithm: RSA-4096 (asymmetric signing)
- Rotation: Manual (key versions managed)
- Access: Service account with
cloudkms.signerVerifierrole
Technology Stack
Backend Framework:
- Django 5.2.8: ORM, admin interface, migrations
- Django REST Framework 3.14: API serialization, authentication, permissions
- django-multitenant 3.2.3: Row-level tenant isolation
API Layer:
- FastAPI 0.104: High-performance async API (alternative: FastAPI for new services)
- Uvicorn: ASGI server
- Gunicorn: Process manager (4 workers)
Database:
- PostgreSQL 16: Primary data store (ACID compliance)
- Redis 7: Session caching, atomic operations (Lua scripts)
Background Tasks:
- Celery 5.3: Distributed task queue
- Redis: Message broker for Celery
- Celery Beat: Periodic task scheduler
Security:
- Cloud KMS: RSA-4096 asymmetric signing
- JWT: Authentication tokens (PyJWT)
- Cryptography: License encryption (Fernet)
Infrastructure:
- Google Kubernetes Engine: Container orchestration
- Cloud SQL: Managed PostgreSQL
- Redis Memorystore: Managed Redis
- Cloud KMS: Key management
- Cloud Logging: Centralized logging
- Cloud Monitoring: Metrics and alerting
Development Tools:
- OpenTofu 1.10.7: Infrastructure as Code
- Docker: Containerization
- pytest: Testing framework
- Black: Code formatting
- mypy: Type checking
Implementation Roadmap
Phase 1: MVP (2-3 days) - Core Floating Licenses
Goal: Production-ready floating license system with CODITECT-CORE integration.
Tasks:
-
Complete Django Backend (1 day)
- ✅ Models (License, LicenseSession) - DONE
- ✅ Serializers - DONE
- ✅ Views (acquire, heartbeat, release) - DONE
- ⏸️ Services layer (license signing, zombie cleanup)
- ⏸️ Admin interface configuration
-
Redis Lua Scripts (2 hours)
- ✅ acquire_floating_seat.lua - DONE
- ⏸️ heartbeat_renewal.lua
- ⏸️ release_seat.lua
- ⏸️ Integration tests for race conditions
-
Cloud KMS Integration (3 hours)
- ⏸️ Deploy Cloud KMS key (OpenTofu)
- ⏸️ Service account IAM permissions
- ⏸️ License signing function
- ⏸️ Signature verification (client-side)
-
CODITECT-CORE Client SDK (4 hours)
- ⏸️ Session ID generation (symlink resolution)
- ⏸️ validate-license.py (cached, online, offline-grace)
- ⏸️ init.sh integration
- ⏸️ Heartbeat background thread
- ⏸️ Feature gating (check_feature_access)
-
Testing (4 hours)
- ⏸️ Unit tests (80%+ coverage)
- ⏸️ Integration tests (API + Redis)
- ⏸️ Race condition tests (concurrent seat acquisition)
- ⏸️ End-to-end test (CODITECT client → API → cache → offline)
-
Deployment (3 hours)
- ⏸️ Dockerfile for FastAPI + Celery
- ⏸️ Kubernetes manifests (Deployment, Service, HPA)
- ⏸️ Deploy to GKE
- ⏸️ SSL certificate + custom domain
- ⏸️ Smoke tests in production
Deliverables:
- Floating license acquisition, heartbeat, release endpoints
- Redis atomic seat counting (race condition-free)
- Cloud KMS signed licenses (tamper-proof)
- CODITECT-CORE client SDK with symlink resolution
- Offline grace period (24-72 hours)
- 80%+ test coverage
- Deployed to GKE with monitoring
Success Criteria:
- Developer can activate CODITECT with floating license
- 10 developers share 5 floating seats successfully
- Zombie sessions cleaned up automatically (6-min TTL)
- Offline mode works for 3 days (Pro tier)
- No race conditions in concurrent seat acquisition tests
Phase 2: Full Feature Parity (6-8 weeks)
Goal: Complete implementation of all 10 licensing models with SLASCONE feature parity.
Week 1-2: Device-Based & Named User Licenses
- Hardware fingerprinting (MAC + CPU + UUID)
- Device activation/deactivation endpoints
- Named user authentication (JWT)
- User assignment management (admin UI)
Week 3-4: Subscription & Trial Licenses
- Stripe subscription integration
- Webhook handling (payment success/failure)
- Trial activation with hardware binding
- Automatic trial expiration and downgrade
- Grace period for failed payments (7 days)
Week 5-6: Usage-Based & Version Licenses
- Usage event tracking endpoint
- Usage summary aggregation (billing period)
- Stripe usage-based billing sync
- Version enforcement (major version checking)
- Upgrade path for new major versions
Week 7-8: Perpetual & Expiration Licenses
- Perpetual license generation
- Support subscription (optional for perpetual)
- Expiration warnings (30, 14, 7, 1 day)
- Renewal workflow
- Admin dashboard for license management
Deliverables:
- All 10 licensing models implemented
- Stripe integration (subscriptions + usage-based billing)
- Email notifications (trials, expirations, renewals)
- Admin dashboard (React SPA)
- Team dashboard (usage analytics, seat management)
- Complete API documentation (OpenAPI/Swagger)
Success Criteria:
- All licensing models tested in production
- Stripe webhooks handling 100% of subscription events
- Email notifications sent reliably (<1% failure rate)
- Admin dashboard allows full license lifecycle management
- API documentation complete and accurate
Phase 3: Enterprise Features (Q2 2026)
Goal: Enterprise-grade features for Team and Enterprise tiers.
Features:
-
SSO Integration
- SAML 2.0 support (Okta, Azure AD)
- OIDC support (Google, GitHub)
- Identity Platform configuration
- User provisioning/deprovisioning
-
Private Deployment
- On-premises installation guide
- Air-gapped environment support
- Customer-managed Cloud KMS keys
- Docker Compose deployment for small teams
-
Advanced Analytics
- BigQuery data export
- Custom usage reports
- Anomaly detection (unusual usage patterns)
- Cost forecasting
-
API Rate Limiting
- Redis-based rate limiting
- Tier-based limits (Free: 10/min, Pro: 100/min, Enterprise: unlimited)
- Burst allowance
- Rate limit headers
-
Audit Logging
- Cloud Logging integration
- License activation/deactivation events
- Admin actions (create license, assign user)
- Compliance reporting (SOC 2, GDPR)
Deliverables:
- SSO integration with major providers
- Private deployment documentation and scripts
- Analytics dashboard with BigQuery integration
- API rate limiting implemented
- Comprehensive audit logging
Success Criteria:
- Enterprise customers can deploy on-premises
- SSO authentication works with 5+ providers
- Analytics provide actionable insights
- Rate limiting prevents abuse
- Audit logs meet compliance requirements
Risk Assessment
Technical Risks:
-
Race Conditions in Seat Counting (HIGH)
- Mitigation: Redis Lua scripts (atomic operations)
- Validation: Concurrent acquisition load tests (1000+ requests/sec)
-
Zombie Sessions Accumulating (MEDIUM)
- Mitigation: 6-minute TTL + Celery periodic cleanup
- Validation: Load test with laptop sleep/wake cycles
-
License Tampering (HIGH)
- Mitigation: Cloud KMS RSA-4096 signing, signature verification
- Validation: Attempt to modify cached license, ensure rejection
-
Database Connection Exhaustion (MEDIUM)
- Mitigation: Connection pooling (PgBouncer), max 100 connections
- Validation: Load test with 1000+ concurrent users
-
Redis Single Point of Failure (MEDIUM)
- Mitigation: Upgrade to STANDARD tier (HA) for production
- Validation: Failover testing
Business Risks:
-
Customer Resistance to Online Validation (LOW)
- Mitigation: Offline grace period (24-168 hours)
- Alternative: Perpetual licenses with optional support
-
Pricing Model Misalignment (MEDIUM)
- Mitigation: Multiple tiers (Free, Pro, Team, Enterprise)
- Validation: Beta testing with 50+ early customers
-
Stripe Integration Complexity (LOW)
- Mitigation: Use Stripe Billing for subscription management
- Validation: Stripe webhook testing (all event types)
Monitoring and Observability
Prometheus Metrics:
license_acquisitions_total- Counter of successful acquisitionslicense_acquisition_failures_total- Counter by reason (pool_exhausted, invalid_license, etc.)license_seats_available- Gauge of available seats per licenselicense_heartbeat_latency_seconds- Histogram of heartbeat API latencyzombie_sessions_cleaned_total- Counter of zombie sessions cleanedredis_lua_script_duration_seconds- Histogram of Lua script execution time
Grafana Dashboards:
- License API Performance: Request rate, latency (p50/p95/p99), error rate
- Seat Utilization: Available vs. in-use seats per license, utilization percentage
- Heartbeat Health: Heartbeat success rate, missed heartbeats, TTL expiration events
- Celery Tasks: Task queue depth, task execution duration, failure rate
- Database Performance: Connection pool usage, query latency, slow queries
Alerting Rules:
- License API error rate >5% for 5 minutes
- Seat utilization >90% for 15 minutes (notify admin to add seats)
- Heartbeat failure rate >10% for 5 minutes
- Celery task queue depth >100 for 10 minutes
- Database connection pool exhausted
Cloud Logging:
- Structured JSON logging for all API requests
- Log levels: INFO (normal), WARNING (degraded), ERROR (failure), CRITICAL (outage)
- Log retention: 30 days (Cloud Logging), 1 year (BigQuery export)
Cost Analysis
Development Environment (Current):
- GKE: $100/month (3x n1-standard-2 preemptible)
- Cloud SQL: $150/month (db-custom-2-7680, Regional HA)
- Redis: $30/month (6GB BASIC)
- Cloud KMS: $10/month (1 key)
- Networking: $20/month (Cloud NAT, egress)
- Total: $310/month
Production Environment (Projected):
- GKE: $500/month (production-grade cluster, 5-10 nodes)
- Cloud SQL: $400/month (larger instance, more backups)
- Redis: $150/month (16GB STANDARD, HA)
- Cloud KMS: $10/month
- Identity Platform: $50/month (up to 50K MAU)
- Load Balancer + SSL: $50/month
- Monitoring: $40/month (Cloud Monitoring, Logging)
- Total: $1,200/month
Scaling Estimates:
| Metric | MVP | Year 1 | Year 2 |
|---|---|---|---|
| Active Licenses | 100 | 1,000 | 10,000 |
| Concurrent Sessions | 50 | 500 | 5,000 |
| API Requests/min | 100 | 1,000 | 10,000 |
| Database Size | 1GB | 10GB | 100GB |
| Monthly Cost | $310 | $1,200 | $2,500 |
| Cost per License | $3.10 | $1.20 | $0.25 |
Cost Optimization:
- Committed use discounts (37% for 1-year, 52% for 3-year)
- Preemptible nodes for dev/test (70% savings)
- Right-size resources based on metrics (reduce over-provisioning)
- Cold storage for old usage events (BigQuery)
End of Part 4
Completed:
- Part 1: Executive Summary, Overview, Licensing Models 1-5
- Part 2: Licensing Models 6-10
- Part 3: CODITECT-CORE Enforcement Architecture
- Part 4: C4 Diagrams (5 levels), Deployment Architecture, Technology Stack, Implementation Roadmap, Risk Assessment, Monitoring, Cost Analysis
Status: Software Design Document complete (4 parts, 3,600+ lines, 100+ code examples).
Next: ADRs and Sequence Diagrams in batches.