Quota Management

Table of Contents

• Overview
• When to Use
• Core Concepts
• Quota Thresholds
• Quota Types
• Quick Start
• Check Quota Status
• Record Usage
• Estimate Before Execution
• Integration Pattern
• Detailed Resources
• Exit Criteria

Quota Management

Overview

Patterns for tracking and enforcing resource quotas across rate-limited services. This skill provides the infrastructure that other plugins use for consistent quota handling.

When To Use

• Building integrations with rate-limited APIs
• Need to track usage across sessions
• Want graceful degradation when limits approached
• Require cost estimation before operations

When NOT To Use

• Project doesn't use the leyline infrastructure patterns
• Simple scripts without service architecture needs

Core Concepts

Quota Thresholds

Three-tier threshold system for proactive management:

Quota Types

@dataclass
class QuotaConfig:
    requests_per_minute: int = 60
    requests_per_day: int = 1000
    tokens_per_minute: int = 100000
    tokens_per_day: int = 1000000

Quick Start

Check Quota Status

from leyline.quota_tracker import QuotaTracker

tracker = QuotaTracker(service="my-service")
status, warnings = tracker.get_quota_status()

if status == "CRITICAL":
    # Defer or use secondary service
    pass

Record Usage

tracker.record_request(
    tokens=estimated_tokens,
    success=True,
    duration=elapsed_seconds
)

Estimate Before Execution

can_proceed, issues = tracker.can_handle_task(estimated_tokens)
if not can_proceed:
    print(f"Quota issues: {issues}")

Integration Pattern

Other plugins reference this skill:

# In your skill's frontmatter
dependencies: [leyline:quota-management]

Then use the shared patterns: 1. Initialize tracker for your service 2. Check quota before operations 3. Record usage after operations 4. Handle threshold warnings gracefully

Detailed Resources

• Threshold Strategies: See modules/threshold-strategies.md for degradation patterns
• Estimation Patterns: See modules/estimation-patterns.md for token/cost estimation

Exit Criteria

• Quota status checked before operation
• Usage recorded after operation
• Threshold warnings handled appropriately

Estimation Patterns

Token Estimation

File-Based Estimation

# Tokens per character ratios by file type
TOKEN_RATIOS = {
    "code": 3.2,      # .py, .js, .ts, .go, .rs
    "json": 3.6,      # .json, .yaml, .toml
    "text": 4.2,      # .md, .txt, .rst
    "default": 4.0
}

def estimate_file_tokens(path: Path) -> int:
    """Estimate tokens for a file."""
    size = path.stat().st_size
    suffix = path.suffix.lower()

    if suffix in [".py", ".js", ".ts", ".go", ".rs"]:
        ratio = TOKEN_RATIOS["code"]
    elif suffix in [".json", ".yaml", ".yml", ".toml"]:
        ratio = TOKEN_RATIOS["json"]
    else:
        ratio = TOKEN_RATIOS["text"]

    return int(size / ratio)

Task-Based Estimation

Cost Estimation

Cost Calculation

def estimate_cost(
    input_tokens: int,
    output_tokens: int,
    model: str
) -> float:
    """Estimate cost in USD."""
    rates = {
        "gemini-pro": {"input": 0.50, "output": 1.50},
        "gemini-flash": {"input": 0.075, "output": 0.30},
        "qwen-max": {"input": 0.40, "output": 1.20},
    }

    rate = rates.get(model, rates["gemini-pro"])
    input_cost = (input_tokens / 1_000_000) * rate["input"]
    output_cost = (output_tokens / 1_000_000) * rate["output"]

    return input_cost + output_cost

Cost Thresholds

Pre-Flight Checks

Estimation Workflow

def preflight_check(files: list[Path], prompt: str) -> dict:
    """Estimate resources before operation."""
    input_tokens = sum(estimate_file_tokens(f) for f in files)
    input_tokens += len(prompt) // 4  # Prompt tokens

    output_tokens = estimate_output_tokens(task_type)
    cost = estimate_cost(input_tokens, output_tokens, model)

    return {
        "input_tokens": input_tokens,
        "output_tokens": output_tokens,
        "estimated_cost": cost,
        "within_quota": can_handle_task(input_tokens)
    }

Threshold Strategies

Degradation Patterns

Progressive Degradation

def get_degradation_strategy(usage_percent: float) -> str:
    if usage_percent < 80:
        return "full_operation"
    elif usage_percent < 90:
        return "reduce_batch_size"
    elif usage_percent < 95:
        return "essential_only"
    else:
        return "defer_or_secondary"

Batch Size Adjustment

Recovery Strategies

Wait for Reset

def wait_for_reset(quota_type: str) -> int:
    """Returns seconds until quota resets."""
    reset_times = {
        "rpm": 60,           # Per-minute resets
        "tpm": 60,           # Token per minute
        "daily": seconds_until_midnight()
    }
    return reset_times.get(quota_type, 3600)

Secondary Services

When primary service is at capacity: 1. Check alternative service quota 2. Use cached results if available 3. Return partial results with warning 4. Queue for later execution

Alerting Patterns

Threshold Notifications

def check_and_alert(tracker: QuotaTracker) -> list[str]:
    alerts = []
    usage = tracker.get_current_usage()

    if usage.rpm_percent > 80:
        alerts.append(f"RPM at {usage.rpm_percent}%")
    if usage.daily_percent > 90:
        alerts.append(f"Daily quota at {usage.daily_percent}%")

    return alerts

Proactive Warnings

• Alert at 70% for large planned operations
• Alert at 80% for any new operations
• Block at 95% except critical paths