diff --git a/packages/dev-primitives/DEPRECATED.md b/packages/dev-primitives/DEPRECATED.md deleted file mode 100644 index 9d25f6f2..00000000 --- a/packages/dev-primitives/DEPRECATED.md +++ /dev/null @@ -1,40 +0,0 @@ -# ⚠️ DEPRECATED - -This package has been deprecated in favor of `tta-dev-primitives` from the TTA.dev repository. - -## Migration - -**Old (deprecated):** -```python -from dev_primitives.recovery import CircuitBreaker, ErrorCategory -``` - -**New (recommended):** -```python -from tta_dev_primitives.recovery import CircuitBreaker, ErrorCategory -``` - -## Why? - -The `dev-primitives` package has been consolidated into `tta-dev-primitives` in the TTA.dev repository with: -- ✅ Better naming clarity (development tools, not game components) -- ✅ Professional packaging and versioning -- ✅ Comprehensive testing (35 tests, 100% passing) -- ✅ All features consolidated in one place -- ✅ Circuit breaker integration with error classification -- ✅ Maintained separately from TTA game code - -## Timeline - -- **Current**: This package still works but is no longer maintained -- **Next**: Code will be migrated to use `tta-dev-primitives` -- **Future**: This directory will be removed - -## Links - -- **New Package**: https://github.com/theinterneti/TTA.dev/tree/main/packages/tta-dev-primitives -- **Migration PR**: https://github.com/theinterneti/TTA/pull/101 - ---- - -*Last Updated: October 28, 2025* diff --git a/packages/dev-primitives/README.md b/packages/dev-primitives/README.md deleted file mode 100644 index 8cccc59e..00000000 --- a/packages/dev-primitives/README.md +++ /dev/null @@ -1,40 +0,0 @@ -# Development Primitives - -Meta-level development primitives for error recovery, retry logic, circuit breakers, and observability. - -## Features - -- **Error Recovery**: Automatic retry with exponential backoff -- **Circuit Breakers**: Prevent cascading failures -- **Error Classification**: Categorize errors (network, rate limit, transient, permanent) -- **Observability**: Structured logging and metrics - -## Installation - -```bash -uv pip install -e packages/dev-primitives -``` - -## Quick Start - -```python -from dev_primitives import with_retry, RetryConfig, CircuitBreaker - -# Simple retry -@with_retry(RetryConfig(max_retries=3)) -def flaky_operation(): - # Your code here - pass - -# Circuit breaker -cb = CircuitBreaker(failure_threshold=5, recovery_timeout=60.0) -result = cb.call(risky_function, arg1, arg2) -``` - -## Usage - -See the [scripts/primitives](../../scripts/primitives) directory for examples and the original implementation. - -## License - -Proprietary - TTA Storytelling Platform diff --git a/packages/dev-primitives/pyproject.toml b/packages/dev-primitives/pyproject.toml deleted file mode 100644 index e28a3afd..00000000 --- a/packages/dev-primitives/pyproject.toml +++ /dev/null @@ -1,40 +0,0 @@ -[project] -name = "dev-primitives" -version = "0.1.0" -description = "Development primitives for error recovery, retry logic, and observability" -authors = [ - {name = "TTA Development Team"} -] -readme = "README.md" -requires-python = ">=3.11" -dependencies = [ - "structlog>=24.1.0", - "tenacity>=8.2.3", -] - -[project.optional-dependencies] -dev = [ - "pytest>=8.0.0", - "pytest-asyncio>=0.23.0", - "pytest-cov>=4.1.0", - "ruff>=0.3.0", -] - -[build-system] -requires = ["hatchling"] -build-backend = "hatchling.build" - -[tool.hatch.build.targets.wheel] -packages = ["src/dev_primitives"] - -[tool.pytest.ini_options] -asyncio_mode = "auto" -testpaths = ["tests"] - -[tool.ruff] -line-length = 100 -target-version = "py311" - -[tool.ruff.lint] -select = ["E", "F", "I", "N", "W", "B", "UP"] -ignore = ["E501"] diff --git a/packages/dev-primitives/src/dev_primitives/__init__.py b/packages/dev-primitives/src/dev_primitives/__init__.py deleted file mode 100644 index 1d18573c..00000000 --- a/packages/dev-primitives/src/dev_primitives/__init__.py +++ /dev/null @@ -1,27 +0,0 @@ -"""Development primitives for error recovery, retry logic, and observability.""" - -from .recovery import ( - CircuitBreaker, - ErrorCategory, - ErrorSeverity, - RetryConfig, - calculate_delay, - classify_error, - should_retry, - with_retry, - with_retry_async, -) - -__all__ = [ - "CircuitBreaker", - "ErrorCategory", - "ErrorSeverity", - "RetryConfig", - "calculate_delay", - "classify_error", - "should_retry", - "with_retry", - "with_retry_async", -] - -__version__ = "0.1.0" diff --git a/packages/dev-primitives/src/dev_primitives/recovery.py b/packages/dev-primitives/src/dev_primitives/recovery.py deleted file mode 100644 index 6605824a..00000000 --- a/packages/dev-primitives/src/dev_primitives/recovery.py +++ /dev/null @@ -1,381 +0,0 @@ -#!/usr/bin/env python3 -""" -Error Recovery Framework for Development Scripts. - -This module provides error recovery patterns for development automation, -implementing the agentic primitive of error handling and recovery at the -meta-level (development process) before integrating into the product. - -Features: -- Error classification (network, rate limit, transient, permanent) -- Automatic retry with exponential backoff -- Fallback strategies -- Circuit breaker pattern -- Comprehensive error logging -""" - -import asyncio -import functools -import logging -import time -from collections.abc import Callable -from dataclasses import dataclass -from enum import Enum -from typing import ParamSpec, TypeVar - -logger = logging.getLogger(__name__) - -P = ParamSpec("P") -T = TypeVar("T") - - -class ErrorCategory(Enum): - """Categories of development errors.""" - - NETWORK = "network" # Network/API failures - RATE_LIMIT = "rate_limit" # Rate limiting - RESOURCE = "resource" # Resource exhaustion - TRANSIENT = "transient" # Temporary failures - PERMANENT = "permanent" # Permanent failures - - -class ErrorSeverity(Enum): - """Severity levels for errors.""" - - LOW = "low" # Minor issues, can continue - MEDIUM = "medium" # Significant but recoverable - HIGH = "high" # Critical, requires attention - CRITICAL = "critical" # System-breaking - - -@dataclass -class RetryConfig: - """Configuration for retry behavior.""" - - max_retries: int = 3 - base_delay: float = 1.0 # seconds - max_delay: float = 60.0 # seconds - exponential_base: float = 2.0 - jitter: bool = True - - def __post_init__(self): - """Validate configuration.""" - if self.max_retries < 0: - raise ValueError("max_retries must be non-negative") - if self.base_delay <= 0: - raise ValueError("base_delay must be positive") - if self.max_delay < self.base_delay: - raise ValueError("max_delay must be >= base_delay") - if self.exponential_base <= 1: - raise ValueError("exponential_base must be > 1") - - -def classify_error(error: Exception) -> tuple[ErrorCategory, ErrorSeverity]: - """ - Classify an error into category and severity. - - Args: - error: The exception to classify - - Returns: - Tuple of (category, severity) - """ - error_str = str(error).lower() - error_type = type(error).__name__.lower() - - # Network errors - if any( - x in error_str or x in error_type - for x in [ - "connection", - "timeout", - "network", - "unreachable", - "connectionerror", - "timeouterror", - ] - ): - return ErrorCategory.NETWORK, ErrorSeverity.MEDIUM - - # Rate limiting - if any(x in error_str for x in ["rate limit", "too many requests", "429", "quota"]): - return ErrorCategory.RATE_LIMIT, ErrorSeverity.MEDIUM - - # Resource errors - if any( - x in error_str or x in error_type - for x in ["memory", "disk", "resource", "out of memory", "no space"] - ): - return ErrorCategory.RESOURCE, ErrorSeverity.HIGH - - # Transient errors - if any(x in error_str for x in ["temporary", "unavailable", "503", "502", "504"]): - return ErrorCategory.TRANSIENT, ErrorSeverity.MEDIUM - - # Default to permanent - return ErrorCategory.PERMANENT, ErrorSeverity.HIGH - - -def should_retry(error: Exception, attempt: int, max_retries: int) -> bool: - """ - Determine if an error should be retried. - - Args: - error: The exception that occurred - attempt: Current attempt number (0-indexed) - max_retries: Maximum number of retries allowed - - Returns: - True if should retry, False otherwise - """ - if attempt >= max_retries: - return False - - category, severity = classify_error(error) - - # Don't retry critical permanent errors - if category == ErrorCategory.PERMANENT and severity == ErrorSeverity.CRITICAL: - return False - - # Retry network, rate limit, and transient errors - return category in [ - ErrorCategory.NETWORK, - ErrorCategory.RATE_LIMIT, - ErrorCategory.TRANSIENT, - ] - - -def calculate_delay(attempt: int, config: RetryConfig) -> float: - """ - Calculate delay before next retry using exponential backoff. - - Args: - attempt: Current attempt number (0-indexed) - config: Retry configuration - - Returns: - Delay in seconds - """ - import random - - # Exponential backoff - delay = min(config.base_delay * (config.exponential_base**attempt), config.max_delay) - - # Add jitter to prevent thundering herd - if config.jitter: - delay *= 0.5 + random.random() - - return delay - - -def with_retry( - config: RetryConfig | None = None, fallback: Callable[..., T] | None = None -) -> Callable[[Callable[P, T]], Callable[P, T]]: - """ - Decorator to add retry logic to a function. - - Args: - config: Retry configuration (uses defaults if None) - fallback: Optional fallback function to call if all retries fail - - Returns: - Decorated function with retry logic - - Example: - @with_retry(RetryConfig(max_retries=3)) - def flaky_function(): - # May fail transiently - pass - - @with_retry(fallback=lambda: "default_value") - def function_with_fallback(): - # Will return "default_value" if all retries fail - pass - """ - if config is None: - config = RetryConfig() - - def decorator(func: Callable[P, T]) -> Callable[P, T]: - @functools.wraps(func) - def wrapper(*args: P.args, **kwargs: P.kwargs) -> T: - last_error = None - - for attempt in range(config.max_retries + 1): - try: - return func(*args, **kwargs) - except Exception as e: - last_error = e - category, severity = classify_error(e) - - if not should_retry(e, attempt, config.max_retries): - logger.error( - f"{func.__name__} failed permanently: {e} " - f"(category={category.value}, severity={severity.value})" - ) - break - - delay = calculate_delay(attempt, config) - logger.warning( - f"{func.__name__} failed (attempt {attempt + 1}/{config.max_retries + 1}): {e}. " - f"Category: {category.value}, Severity: {severity.value}. " - f"Retrying in {delay:.1f}s..." - ) - - time.sleep(delay) - - # All retries exhausted - if fallback: - logger.info(f"{func.__name__} using fallback after {config.max_retries} retries") - return fallback(*args, **kwargs) - - # Re-raise the last error - raise last_error - - return wrapper - - return decorator - - -def with_retry_async( - config: RetryConfig | None = None, fallback: Callable[..., T] | None = None -) -> Callable[[Callable[P, T]], Callable[P, T]]: - """ - Async version of with_retry decorator. - - Args: - config: Retry configuration (uses defaults if None) - fallback: Optional async fallback function to call if all retries fail - - Returns: - Decorated async function with retry logic - - Example: - @with_retry_async(RetryConfig(max_retries=3)) - async def async_flaky_function(): - # May fail transiently - pass - """ - if config is None: - config = RetryConfig() - - def decorator(func: Callable[P, T]) -> Callable[P, T]: - @functools.wraps(func) - async def wrapper(*args: P.args, **kwargs: P.kwargs) -> T: - last_error = None - - for attempt in range(config.max_retries + 1): - try: - return await func(*args, **kwargs) - except Exception as e: - last_error = e - category, severity = classify_error(e) - - if not should_retry(e, attempt, config.max_retries): - logger.error( - f"{func.__name__} failed permanently: {e} " - f"(category={category.value}, severity={severity.value})" - ) - break - - delay = calculate_delay(attempt, config) - logger.warning( - f"{func.__name__} failed (attempt {attempt + 1}/{config.max_retries + 1}): {e}. " - f"Category: {category.value}, Severity: {severity.value}. " - f"Retrying in {delay:.1f}s..." - ) - - await asyncio.sleep(delay) - - # All retries exhausted - if fallback: - logger.info(f"{func.__name__} using fallback after {config.max_retries} retries") - return await fallback(*args, **kwargs) - - # Re-raise the last error - raise last_error - - return wrapper - - return decorator - - -class CircuitBreaker: - """ - Circuit breaker pattern for preventing cascading failures. - - States: - - CLOSED: Normal operation, requests pass through - - OPEN: Too many failures, requests fail immediately - - HALF_OPEN: Testing if service recovered - """ - - def __init__( - self, - failure_threshold: int = 5, - recovery_timeout: float = 60.0, - expected_exception: type[Exception] = Exception, - ): - """ - Initialize circuit breaker. - - Args: - failure_threshold: Number of failures before opening circuit - recovery_timeout: Seconds to wait before attempting recovery - expected_exception: Exception type to catch - """ - self.failure_threshold = failure_threshold - self.recovery_timeout = recovery_timeout - self.expected_exception = expected_exception - - self.failure_count = 0 - self.last_failure_time: float | None = None - self.state = "CLOSED" - - def call(self, func: Callable[P, T], *args: P.args, **kwargs: P.kwargs) -> T: - """ - Call function through circuit breaker. - - Args: - func: Function to call - *args: Positional arguments - **kwargs: Keyword arguments - - Returns: - Function result - - Raises: - Exception: If circuit is open or function fails - """ - if self.state == "OPEN": - if self._should_attempt_reset(): - self.state = "HALF_OPEN" - else: - raise Exception(f"Circuit breaker is OPEN (failures: {self.failure_count})") - - try: - result = func(*args, **kwargs) - self._on_success() - return result - except self.expected_exception as e: - self._on_failure() - raise e - - def _should_attempt_reset(self) -> bool: - """Check if enough time has passed to attempt reset.""" - if self.last_failure_time is None: - return True - return time.time() - self.last_failure_time >= self.recovery_timeout - - def _on_success(self) -> None: - """Handle successful call.""" - self.failure_count = 0 - self.state = "CLOSED" - - def _on_failure(self) -> None: - """Handle failed call.""" - self.failure_count += 1 - self.last_failure_time = time.time() - - if self.failure_count >= self.failure_threshold: - self.state = "OPEN" - logger.warning(f"Circuit breaker opened after {self.failure_count} failures") diff --git a/packages/tta-workflow-primitives/DEPRECATED.md b/packages/tta-workflow-primitives/DEPRECATED.md deleted file mode 100644 index 07a865b8..00000000 --- a/packages/tta-workflow-primitives/DEPRECATED.md +++ /dev/null @@ -1,44 +0,0 @@ -# ⚠️ DEPRECATED - -This package has been deprecated in favor of `tta-dev-primitives` from the TTA.dev repository. - -## Migration - -**Old (deprecated):** - -```python -from tta_workflow_primitives.core import SequentialPrimitive -from tta_workflow_primitives.recovery import RetryPrimitive -``` - -**New (recommended):** - -```python -from tta_dev_primitives.core import SequentialPrimitive -from tta_dev_primitives.recovery import RetryPrimitive -``` - -## Why? - -The `tta-workflow-primitives` package has been renamed to `tta-dev-primitives` and moved to the TTA.dev repository with: - -- ✅ Better naming clarity (development tools, not game components) -- ✅ Professional packaging and versioning -- ✅ Comprehensive testing (35 tests, 100% passing) -- ✅ Circuit breaker consolidation from dev-primitives -- ✅ Maintained separately from TTA game code - -## Timeline - -- **Current**: This package still works but is no longer maintained -- **Next**: Code will be migrated to use `tta-dev-primitives` -- **Future**: This directory will be removed - -## Links - -- [New Package](https://github.com/theinterneti/TTA.dev/tree/main/packages/tta-dev-primitives) -- [Migration PR](https://github.com/theinterneti/TTA/pull/101) - ---- - -Last Updated: October 28, 2025 diff --git a/packages/tta-workflow-primitives/IMPROVEMENTS_QUICK_START.md b/packages/tta-workflow-primitives/IMPROVEMENTS_QUICK_START.md deleted file mode 100644 index dfde395f..00000000 --- a/packages/tta-workflow-primitives/IMPROVEMENTS_QUICK_START.md +++ /dev/null @@ -1,584 +0,0 @@ -# Quick Start: Priority Improvements - -**Target:** Implement 3 high-impact primitives in Week 1 - ---- - -## 1. Router Primitive (Day 1-2) - -### File: `src/tta_workflow_primitives/core/routing.py` - -```python -"""Routing primitive for intelligent workflow branching.""" - -from __future__ import annotations - -from typing import Any, Callable - -from .base import WorkflowContext, WorkflowPrimitive -from ..observability.logging import get_logger - -logger = get_logger(__name__) - - -class RouterPrimitive(WorkflowPrimitive[Any, Any]): - """ - Route input to appropriate primitive based on routing function. - - Example: - ```python - router = RouterPrimitive( - routes={ - "openai": openai_primitive, - "anthropic": anthropic_primitive, - "local": local_llm_primitive - }, - router_fn=lambda data, ctx: ctx.metadata.get("provider", "openai"), - default="openai" - ) - ``` - """ - - def __init__( - self, - routes: dict[str, WorkflowPrimitive], - router_fn: Callable[[Any, WorkflowContext], str], - default: str | None = None - ): - """ - Initialize router. - - Args: - routes: Map of route keys to primitives - router_fn: Function to determine route from input/context - default: Default route if router_fn returns unknown key - """ - self.routes = routes - self.router_fn = router_fn - self.default = default - - async def execute(self, input_data: Any, context: WorkflowContext) -> Any: - """Execute routing logic and invoke selected primitive.""" - # Determine route - route_key = self.router_fn(input_data, context) - - # Get primitive - primitive = self.routes.get(route_key) - - # Fallback to default - if not primitive and self.default: - route_key = self.default - primitive = self.routes.get(route_key) - - if not primitive: - available = ", ".join(self.routes.keys()) - raise ValueError( - f"No route found for key '{route_key}'. " - f"Available routes: {available}" - ) - - # Log routing decision - logger.info( - "routing_decision", - route=route_key, - available_routes=list(self.routes.keys()) - ) - - # Execute selected primitive - return await primitive.execute(input_data, context) -``` - -### Tests: `tests/test_routing.py` - -```python -"""Tests for routing primitive.""" - -import pytest - -from tta_workflow_primitives.core.routing import RouterPrimitive -from tta_workflow_primitives.core.base import WorkflowContext -from tta_workflow_primitives.testing.mocks import MockPrimitive - - -@pytest.mark.asyncio -async def test_router_basic(): - """Test basic routing.""" - route_a = MockPrimitive("a", return_value={"result": "A"}) - route_b = MockPrimitive("b", return_value={"result": "B"}) - - router = RouterPrimitive( - routes={"a": route_a, "b": route_b}, - router_fn=lambda data, ctx: data["route"] - ) - - context = WorkflowContext() - result = await router.execute({"route": "a"}, context) - - assert result == {"result": "A"} - assert route_a.call_count == 1 - assert route_b.call_count == 0 - - -@pytest.mark.asyncio -async def test_router_default(): - """Test default route fallback.""" - default = MockPrimitive("default", return_value={"result": "DEFAULT"}) - - router = RouterPrimitive( - routes={"a": default}, - router_fn=lambda data, ctx: data.get("route", "unknown"), - default="a" - ) - - context = WorkflowContext() - result = await router.execute({"route": "unknown"}, context) - - assert result == {"result": "DEFAULT"} - - -@pytest.mark.asyncio -async def test_router_no_route_error(): - """Test error when no route found.""" - router = RouterPrimitive( - routes={"a": MockPrimitive("a", return_value={})}, - router_fn=lambda data, ctx: "nonexistent" - ) - - with pytest.raises(ValueError, match="No route found"): - await router.execute({}, WorkflowContext()) -``` - ---- - -## 2. Timeout Primitive (Day 2-3) - -### File: `src/tta_workflow_primitives/recovery/timeout.py` - -```python -"""Timeout enforcement for primitives.""" - -from __future__ import annotations - -import asyncio -from typing import Any - -from ..core.base import WorkflowContext, WorkflowPrimitive -from ..observability.logging import get_logger - -logger = get_logger(__name__) - - -class TimeoutError(Exception): - """Timeout exceeded during execution.""" - pass - - -class TimeoutPrimitive(WorkflowPrimitive[Any, Any]): - """ - Enforce execution timeout with optional fallback. - - Example: - ```python - workflow = TimeoutPrimitive( - primitive=slow_operation, - timeout_seconds=30.0, - fallback=fast_cached_operation - ) - ``` - """ - - def __init__( - self, - primitive: WorkflowPrimitive, - timeout_seconds: float, - fallback: WorkflowPrimitive | None = None - ): - """ - Initialize timeout primitive. - - Args: - primitive: Primitive to execute with timeout - timeout_seconds: Maximum execution time - fallback: Optional fallback primitive on timeout - """ - self.primitive = primitive - self.timeout_seconds = timeout_seconds - self.fallback = fallback - - async def execute(self, input_data: Any, context: WorkflowContext) -> Any: - """Execute with timeout enforcement.""" - try: - result = await asyncio.wait_for( - self.primitive.execute(input_data, context), - timeout=self.timeout_seconds - ) - - logger.info( - "timeout_success", - primitive=self.primitive.__class__.__name__, - timeout=self.timeout_seconds - ) - - return result - - except asyncio.TimeoutError: - logger.warning( - "timeout_exceeded", - primitive=self.primitive.__class__.__name__, - timeout=self.timeout_seconds, - has_fallback=self.fallback is not None - ) - - if self.fallback: - logger.info("executing_fallback") - return await self.fallback.execute(input_data, context) - - raise TimeoutError( - f"Execution exceeded {self.timeout_seconds}s timeout" - ) -``` - -### Tests: `tests/test_timeout.py` - -```python -"""Tests for timeout primitive.""" - -import asyncio -import pytest - -from tta_workflow_primitives.recovery.timeout import TimeoutPrimitive, TimeoutError -from tta_workflow_primitives.core.base import WorkflowContext, LambdaPrimitive -from tta_workflow_primitives.testing.mocks import MockPrimitive - - -@pytest.mark.asyncio -async def test_timeout_success(): - """Test successful execution within timeout.""" - fast = LambdaPrimitive(lambda data, ctx: {"result": "fast"}) - - timeout_prim = TimeoutPrimitive( - primitive=fast, - timeout_seconds=1.0 - ) - - result = await timeout_prim.execute({}, WorkflowContext()) - assert result == {"result": "fast"} - - -@pytest.mark.asyncio -async def test_timeout_exceeded(): - """Test timeout exceeded without fallback.""" - async def slow(data, ctx): - await asyncio.sleep(2.0) - return {"result": "slow"} - - slow_prim = LambdaPrimitive(slow) - timeout_prim = TimeoutPrimitive( - primitive=slow_prim, - timeout_seconds=0.1 - ) - - with pytest.raises(TimeoutError): - await timeout_prim.execute({}, WorkflowContext()) - - -@pytest.mark.asyncio -async def test_timeout_with_fallback(): - """Test fallback on timeout.""" - async def slow(data, ctx): - await asyncio.sleep(2.0) - return {"result": "slow"} - - slow_prim = LambdaPrimitive(slow) - fallback = MockPrimitive("fallback", return_value={"result": "fallback"}) - - timeout_prim = TimeoutPrimitive( - primitive=slow_prim, - timeout_seconds=0.1, - fallback=fallback - ) - - result = await timeout_prim.execute({}, WorkflowContext()) - assert result == {"result": "fallback"} - assert fallback.call_count == 1 -``` - ---- - -## 3. Cache Primitive (Day 3-4) - -### File: `src/tta_workflow_primitives/performance/cache.py` - -```python -"""Caching primitive for workflow results.""" - -from __future__ import annotations - -import time -from typing import Any, Callable - -from ..core.base import WorkflowContext, WorkflowPrimitive -from ..observability.logging import get_logger - -logger = get_logger(__name__) - - -class CachePrimitive(WorkflowPrimitive[Any, Any]): - """ - Cache primitive execution results. - - Example: - ```python - cached = CachePrimitive( - primitive=expensive_llm_call, - cache_key_fn=lambda data, ctx: f"{data['prompt']}:{ctx.player_id}", - ttl_seconds=3600.0 - ) - ``` - """ - - def __init__( - self, - primitive: WorkflowPrimitive, - cache_key_fn: Callable[[Any, WorkflowContext], str], - ttl_seconds: float = 3600.0 - ): - """ - Initialize cache primitive. - - Args: - primitive: Primitive to cache - cache_key_fn: Function to generate cache key - ttl_seconds: Time-to-live for cached values - """ - self.primitive = primitive - self.cache_key_fn = cache_key_fn - self.ttl_seconds = ttl_seconds - self._cache: dict[str, tuple[Any, float]] = {} - - async def execute(self, input_data: Any, context: WorkflowContext) -> Any: - """Execute with caching.""" - # Generate cache key - cache_key = self.cache_key_fn(input_data, context) - - # Check cache - if cache_key in self._cache: - result, timestamp = self._cache[cache_key] - age = time.time() - timestamp - - if age < self.ttl_seconds: - logger.info( - "cache_hit", - key=cache_key, - age_seconds=age, - ttl=self.ttl_seconds - ) - return result - else: - logger.debug("cache_expired", key=cache_key, age=age) - del self._cache[cache_key] - - # Cache miss - execute and store - logger.info("cache_miss", key=cache_key) - result = await self.primitive.execute(input_data, context) - - self._cache[cache_key] = (result, time.time()) - logger.debug("cache_store", key=cache_key, cache_size=len(self._cache)) - - return result - - def clear_cache(self) -> None: - """Clear all cached values.""" - self._cache.clear() - logger.info("cache_cleared") - - def get_stats(self) -> dict: - """Get cache statistics.""" - return { - "size": len(self._cache), - "keys": list(self._cache.keys()) - } -``` - -### Tests: `tests/test_cache.py` - -```python -"""Tests for cache primitive.""" - -import time -import pytest - -from tta_workflow_primitives.performance.cache import CachePrimitive -from tta_workflow_primitives.core.base import WorkflowContext -from tta_workflow_primitives.testing.mocks import MockPrimitive - - -@pytest.mark.asyncio -async def test_cache_hit(): - """Test cache hit on second call.""" - mock = MockPrimitive("test", return_value={"result": "cached"}) - - cached = CachePrimitive( - primitive=mock, - cache_key_fn=lambda data, ctx: data["key"], - ttl_seconds=60.0 - ) - - # First call - cache miss - result1 = await cached.execute({"key": "test"}, WorkflowContext()) - assert result1 == {"result": "cached"} - assert mock.call_count == 1 - - # Second call - cache hit - result2 = await cached.execute({"key": "test"}, WorkflowContext()) - assert result2 == {"result": "cached"} - assert mock.call_count == 1 # Not called again - - -@pytest.mark.asyncio -async def test_cache_miss_different_keys(): - """Test cache miss with different keys.""" - mock = MockPrimitive("test", return_value={"result": "value"}) - - cached = CachePrimitive( - primitive=mock, - cache_key_fn=lambda data, ctx: data["key"], - ttl_seconds=60.0 - ) - - await cached.execute({"key": "a"}, WorkflowContext()) - await cached.execute({"key": "b"}, WorkflowContext()) - - assert mock.call_count == 2 - - -@pytest.mark.asyncio -async def test_cache_expiration(): - """Test cache expiration after TTL.""" - mock = MockPrimitive("test", return_value={"result": "value"}) - - cached = CachePrimitive( - primitive=mock, - cache_key_fn=lambda data, ctx: "key", - ttl_seconds=0.1 # Very short TTL - ) - - # First call - await cached.execute({}, WorkflowContext()) - assert mock.call_count == 1 - - # Wait for expiration - time.sleep(0.2) - - # Second call after expiration - await cached.execute({}, WorkflowContext()) - assert mock.call_count == 2 - - -@pytest.mark.asyncio -async def test_cache_clear(): - """Test cache clearing.""" - mock = MockPrimitive("test", return_value={"result": "value"}) - - cached = CachePrimitive( - primitive=mock, - cache_key_fn=lambda data, ctx: "key", - ttl_seconds=60.0 - ) - - await cached.execute({}, WorkflowContext()) - assert cached.get_stats()["size"] == 1 - - cached.clear_cache() - assert cached.get_stats()["size"] == 0 -``` - ---- - -## Usage Example: Combining All Three - -```python -"""Example workflow using routing, timeout, and caching.""" - -from tta_workflow_primitives.core.routing import RouterPrimitive -from tta_workflow_primitives.recovery.timeout import TimeoutPrimitive -from tta_workflow_primitives.performance.cache import CachePrimitive -from tta_workflow_primitives.core.base import LambdaPrimitive - -# Define provider-specific primitives -openai_primitive = LambdaPrimitive(lambda data, ctx: call_openai(data)) -anthropic_primitive = LambdaPrimitive(lambda data, ctx: call_anthropic(data)) -local_primitive = LambdaPrimitive(lambda data, ctx: call_local_llm(data)) - -# Build workflow with all improvements -workflow = ( - # Route based on cost/speed tradeoff - RouterPrimitive( - routes={ - "fast": CachePrimitive( - TimeoutPrimitive(local_primitive, timeout_seconds=5.0), - cache_key_fn=lambda d, c: f"local:{d['prompt'][:50]}", - ttl_seconds=1800.0 - ), - "balanced": CachePrimitive( - TimeoutPrimitive(anthropic_primitive, timeout_seconds=30.0), - cache_key_fn=lambda d, c: f"anthropic:{d['prompt'][:50]}", - ttl_seconds=3600.0 - ), - "premium": CachePrimitive( - TimeoutPrimitive(openai_primitive, timeout_seconds=30.0), - cache_key_fn=lambda d, c: f"openai:{d['prompt'][:50]}", - ttl_seconds=3600.0 - ) - }, - router_fn=lambda data, ctx: ctx.metadata.get("tier", "balanced"), - default="balanced" - ) -) - -# Execute -context = WorkflowContext(metadata={"tier": "fast"}) -result = await workflow.execute({"prompt": "Tell me a story"}, context) -``` - ---- - -## Integration Checklist - -- [ ] Add to `__init__.py` exports -- [ ] Update package README -- [ ] Run tests: `pytest tests/test_routing.py tests/test_timeout.py tests/test_cache.py` -- [ ] Update CHANGELOG.md -- [ ] Create migration guide for existing workflows -- [ ] Benchmark performance impact -- [ ] Update documentation site - ---- - -## Performance Targets - -| Primitive | Target | Measurement | -|-----------|--------|-------------| -| Router | <5ms overhead | Routing decision time | -| Timeout | <1% false positives | Unnecessary timeouts | -| Cache | >60% hit rate | Production workload | -| Cache | <1ms hit latency | Cache lookup time | - ---- - -## Next Steps (Week 2) - -After implementing these 3 primitives: - -1. **Context Management** (Day 5-7) - - ContextFilter - - ContextManager with pruning - -2. **Rate Limiting** (Day 8-10) - - RateLimitPrimitive - - Token bucket algorithm - -3. **Integration Testing** (Day 11-12) - - End-to-end workflow tests - - Performance benchmarks - - Production rollout plan diff --git a/packages/tta-workflow-primitives/README.md b/packages/tta-workflow-primitives/README.md deleted file mode 100644 index 808543f5..00000000 --- a/packages/tta-workflow-primitives/README.md +++ /dev/null @@ -1,180 +0,0 @@ -# TTA Workflow Primitives - -Production-ready composable workflow primitives for building reliable, observable, and maintainable agent workflows. - -## Features - -### Core Primitives -- **Composable Workflows**: Build complex workflows from simple primitives -- **Type-Safe Composition**: Generics-based type safety -- **Operator Overloading**: Ergonomic `>>` and `|` operators for chaining - -### Observability -- **Distributed Tracing**: OpenTelemetry integration -- **Structured Logging**: Correlation IDs and context -- **Execution Traces**: Complete workflow execution history -- **Metrics Collection**: Performance and success rate tracking - -### Error Recovery -- **Retry Strategies**: Exponential backoff with jitter -- **Fallback Mechanisms**: Graceful degradation -- **Compensation Patterns**: Saga pattern support -- **Circuit Breakers**: Prevent cascading failures - -### Testing -- **Mock Primitives**: Easy workflow testing -- **Test Fixtures**: Pre-built test utilities -- **Assertion Framework**: Workflow-specific assertions - -## Installation - -```bash -uv pip install -e packages/tta-workflow-primitives -``` - -For tracing support: -```bash -uv pip install -e "packages/tta-workflow-primitives[tracing]" -``` - -## Quick Start - -### Basic Composition - -```python -from tta_workflow_primitives import WorkflowPrimitive, SequentialPrimitive - -# Define primitives -safety_check = SafetyValidationPrimitive() -input_proc = InputProcessingPrimitive() -narrative_gen = NarrativeGenerationPrimitive() - -# Compose with >> operator -workflow = safety_check >> input_proc >> narrative_gen - -# Execute -result = await workflow.execute(user_input, context) -``` - -### With Error Recovery - -```python -from tta_workflow_primitives.recovery import RetryPrimitive, FallbackStrategy - -# Retry with fallback -workflow = ( - safety_check >> - input_proc >> - RetryPrimitive( - narrative_gen, - max_retries=3, - strategies=[FallbackStrategy(safe_narrative_gen)] - ) -) -``` - -### With Observability - -```python -from tta_workflow_primitives.observability import ObservablePrimitive - -# Wrap primitives for tracing -workflow = ( - ObservablePrimitive(safety_check, "safety") >> - ObservablePrimitive(input_proc, "input") >> - ObservablePrimitive(narrative_gen, "narrative") -) - -# Automatic tracing, logging, and metrics -result = await workflow.execute(user_input, context) -``` - -### Parallel Execution - -```python -from tta_workflow_primitives import ParallelPrimitive - -# Execute in parallel with | operator -parallel = world_build | character_analysis | theme_analysis - -# Or explicit -parallel = ParallelPrimitive([world_build, character_analysis, theme_analysis]) - -workflow = input_proc >> parallel >> narrative_gen -``` - -### Conditional Branching - -```python -from tta_workflow_primitives import ConditionalPrimitive - -# Branch based on safety level -workflow = ( - safety_check >> - ConditionalPrimitive( - condition=lambda result, ctx: result.safety_level != "blocked", - then_primitive=standard_narrative, - else_primitive=safe_narrative - ) -) -``` - -## Architecture - -``` -tta_workflow_primitives/ -├── core/ # Core primitive abstractions -│ ├── base.py # WorkflowPrimitive base class -│ ├── sequential.py # Sequential composition -│ ├── parallel.py # Parallel composition -│ └── conditional.py # Conditional branching -├── observability/ # Observability features -│ ├── tracing.py # OpenTelemetry integration -│ ├── logging.py # Structured logging -│ └── metrics.py # Metrics collection -├── recovery/ # Error recovery patterns -│ ├── retry.py # Retry strategies -│ ├── fallback.py # Fallback mechanisms -│ └── compensation.py # Saga pattern -└── testing/ # Testing utilities - ├── mocks.py # Mock primitives - └── assertions.py # Test assertions -``` - -## Testing - -```python -from tta_workflow_primitives.testing import MockPrimitive, WorkflowTestCase - -async def test_workflow(): - # Create mocks - mock_safety = MockPrimitive("safety", return_value={"level": "safe"}) - mock_input = MockPrimitive("input", return_value={"intent": "explore"}) - - # Build test case - test = WorkflowTestCase(workflow) - test.with_mock("safety", mock_safety) - test.with_mock("input", mock_input) - - # Execute and assert - result = await test.execute({"user_input": "test"}) - test.assert_primitive_called("safety", times=1) - test.assert_primitive_called("input", times=1) -``` - -## Examples - -See the [examples](./examples) directory for complete workflow examples: - -- `basic_composition.py` - Simple workflow composition -- `error_recovery.py` - Error handling and recovery -- `observability.py` - Tracing and monitoring -- `therapeutic_workflow.py` - Complete therapeutic narrative workflow - -## Migration Guide - -See [MIGRATION.md](./MIGRATION.md) for migrating existing TTA workflows to use primitives. - -## License - -Proprietary - TTA Storytelling Platform diff --git a/packages/tta-workflow-primitives/examples/apm_example.py b/packages/tta-workflow-primitives/examples/apm_example.py deleted file mode 100644 index 5962d183..00000000 --- a/packages/tta-workflow-primitives/examples/apm_example.py +++ /dev/null @@ -1,149 +0,0 @@ -"""Example: Using APM with workflow primitives. - -This example demonstrates how to use OpenTelemetry APM with workflow primitives -to track performance, collect metrics, and export to Prometheus. -""" - -import asyncio -import logging -from typing import Any - -# Setup logging -logging.basicConfig(level=logging.INFO) -logger = logging.getLogger(__name__) - -# Import workflow primitives -from tta_workflow_primitives.apm import setup_apm -from tta_workflow_primitives.apm.decorators import trace_workflow, track_metric -from tta_workflow_primitives.apm.instrumented import APMWorkflowPrimitive -from tta_workflow_primitives.core.base import WorkflowContext - - -# Example 1: Using APMWorkflowPrimitive base class -class DataProcessor(APMWorkflowPrimitive): - """Example primitive that processes data with APM tracking.""" - - async def _execute_impl( - self, input_data: dict[str, Any], context: WorkflowContext - ) -> dict[str, Any]: - """Process the data.""" - logger.info(f"Processing data: {input_data}") - - # Simulate processing - await asyncio.sleep(0.1) - - result = { - "processed": True, - "input_count": len(input_data), - "output": f"Processed {input_data.get('value', 'unknown')}", - } - - return result - - -class DataValidator(APMWorkflowPrimitive): - """Example primitive that validates data with APM tracking.""" - - async def _execute_impl( - self, input_data: dict[str, Any], context: WorkflowContext - ) -> dict[str, Any]: - """Validate the data.""" - logger.info(f"Validating data: {input_data}") - - # Simulate validation - await asyncio.sleep(0.05) - - is_valid = input_data.get("processed", False) - - if not is_valid: - raise ValueError("Data validation failed") - - return {**input_data, "validated": True} - - -# Example 2: Using decorators for custom functions -@trace_workflow("custom_transform") -@track_metric("transform_operations", "counter", "Number of transformations") -async def custom_transform(data: dict[str, Any]) -> dict[str, Any]: - """Custom transformation with decorators.""" - await asyncio.sleep(0.1) - return {**data, "transformed": True, "timestamp": "2025-10-26"} - - -async def main() -> None: - """Run the APM example.""" - - # Step 1: Setup APM - logger.info("Setting up APM with Prometheus export...") - setup_apm( - service_name="apm-example", - enable_prometheus=True, - enable_console=True, # Enable console output for demo - ) - - # Step 2: Create workflow context - context = WorkflowContext( - workflow_id="example-workflow-001", - session_id="session-123", - metadata={"environment": "development"}, - ) - - # Step 3: Create and compose primitives - processor = DataProcessor(name="processor") - validator = DataValidator(name="validator") - - # Compose workflow using >> operator - workflow = processor >> validator - - # Step 4: Execute workflow - logger.info("Executing workflow...") - - input_data = {"value": "test_data", "priority": "high"} - - try: - result = await workflow.execute(input_data, context) - logger.info(f"Workflow result: {result}") - except Exception as e: - logger.error(f"Workflow failed: {e}") - - # Step 5: Try with custom function - logger.info("Running custom transform...") - transformed = await custom_transform(result) - logger.info(f"Transformed result: {transformed}") - - # Step 6: Simulate multiple executions for metrics - logger.info("Running multiple executions for metrics...") - for i in range(5): - try: - test_data = {"value": f"test_{i}", "priority": "normal"} - await workflow.execute(test_data, context) - await asyncio.sleep(0.2) - except Exception as e: - logger.error(f"Execution {i} failed: {e}") - - logger.info("✓ APM example complete!") - logger.info("Metrics are being exported to Prometheus on port 9464") - logger.info("Access metrics at: http://localhost:9464/metrics") - - -if __name__ == "__main__": - # Run the example - asyncio.run(main()) - - print("\n" + "=" * 70) - print("APM Example Summary") - print("=" * 70) - print("\n✓ Executed workflow with APM instrumentation") - print("✓ Collected metrics:") - print(" - primitive.processor.executions (counter)") - print(" - primitive.processor.duration (histogram)") - print(" - primitive.validator.executions (counter)") - print(" - primitive.validator.duration (histogram)") - print(" - transform_operations (counter)") - print("\n✓ Traces captured with OpenTelemetry") - print("✓ Metrics exported to Prometheus") - print("\nNext steps:") - print("1. View metrics: http://localhost:9464/metrics") - print("2. Import into Prometheus") - print("3. Create Grafana dashboards") - print("4. Add to your own workflows!") diff --git a/packages/tta-workflow-primitives/examples/quick_wins_demo.py b/packages/tta-workflow-primitives/examples/quick_wins_demo.py deleted file mode 100644 index 234af6a4..00000000 --- a/packages/tta-workflow-primitives/examples/quick_wins_demo.py +++ /dev/null @@ -1,57 +0,0 @@ -"""Quick Wins Implementation Example""" - -import asyncio - -from tta_workflow_primitives import ( - CachePrimitive, - LambdaPrimitive, - RouterPrimitive, - TimeoutPrimitive, - WorkflowContext, -) - - -# Simulate LLM providers -async def openai_call(data, ctx): - await asyncio.sleep(0.3) - return {"provider": "openai", "response": "High quality", "cost": 0.10} - - -async def local_llm_call(data, ctx): - await asyncio.sleep(0.05) - return {"provider": "local", "response": "Quick", "cost": 0.01} - - -# Build workflow -workflow = CachePrimitive( - TimeoutPrimitive( - RouterPrimitive( - routes={ - "openai": LambdaPrimitive(openai_call), - "local": LambdaPrimitive(local_llm_call), - }, - router_fn=lambda d, c: c.metadata.get("tier", "local"), - default="local", - ), - timeout_seconds=5.0, - ), - cache_key_fn=lambda d, c: f"{d.get('prompt', '')}:{c.metadata.get('tier')}", - ttl_seconds=3600.0, -) - - -async def main() -> None: - print("✓ Quick Wins Captured - All 23 tests passing!") - print(" Router, Timeout, Cache primitives ready to use") - - # Demo - ctx = WorkflowContext(metadata={"tier": "local"}) - result = await workflow.execute({"prompt": "test"}, ctx) - print(f" Demo: Provider={result['provider']}, Cost=${result['cost']}") - - stats = workflow.get_stats() - print(f" Cache: {stats['hits']} hits, {stats['misses']} misses") - - -if __name__ == "__main__": - asyncio.run(main()) diff --git a/packages/tta-workflow-primitives/pyproject.toml b/packages/tta-workflow-primitives/pyproject.toml deleted file mode 100644 index e3cce532..00000000 --- a/packages/tta-workflow-primitives/pyproject.toml +++ /dev/null @@ -1,59 +0,0 @@ -[project] -name = "tta-workflow-primitives" -version = "0.1.0" -description = "Production-ready composable workflow primitives for TTA agent orchestration" -authors = [{ name = "TTA Development Team" }] -readme = "README.md" -requires-python = ">=3.11" -dependencies = [ - "pydantic>=2.6.0", - "structlog>=24.1.0", - "opentelemetry-api>=1.24.0", - "opentelemetry-sdk>=1.24.0", - "tenacity>=8.2.3", -] - -[project.optional-dependencies] -dev = [ - "pytest>=8.0.0", - "pytest-asyncio>=0.23.0", - "pytest-cov>=4.1.0", - "pytest-mock>=3.12.0", - "ruff>=0.3.0", - "mypy>=1.8.0", -] -tracing = [ - "opentelemetry-instrumentation>=0.45b0", - "opentelemetry-exporter-jaeger>=1.24.0", -] -apm = [ - "opentelemetry-api>=1.20.0", - "opentelemetry-sdk>=1.20.0", - "opentelemetry-exporter-prometheus>=0.41b0", - "opentelemetry-instrumentation>=0.41b0", -] - -[build-system] -requires = ["hatchling"] -build-backend = "hatchling.build" - -[tool.hatch.build.targets.wheel] -packages = ["src/tta_workflow_primitives"] - -[tool.pytest.ini_options] -asyncio_mode = "auto" -testpaths = ["tests"] - -[tool.ruff] -line-length = 100 -target-version = "py311" - -[tool.ruff.lint] -select = ["E", "F", "I", "N", "W", "B", "UP", "ANN"] -ignore = ["E501", "ANN101", "ANN102"] - -[tool.mypy] -python_version = "3.11" -strict = true -warn_return_any = true -warn_unused_configs = true diff --git a/packages/tta-workflow-primitives/src/tta_workflow_primitives/__init__.py b/packages/tta-workflow-primitives/src/tta_workflow_primitives/__init__.py deleted file mode 100644 index e263a96c..00000000 --- a/packages/tta-workflow-primitives/src/tta_workflow_primitives/__init__.py +++ /dev/null @@ -1,43 +0,0 @@ -"""TTA Workflow Primitives - Composable workflow building blocks.""" - -from .core.base import LambdaPrimitive, WorkflowContext, WorkflowPrimitive -from .core.conditional import ConditionalPrimitive -from .core.parallel import ParallelPrimitive -from .core.routing import RouterPrimitive -from .core.sequential import SequentialPrimitive -from .performance.cache import CachePrimitive -from .recovery.timeout import TimeoutError, TimeoutPrimitive - -# APM support (optional) -try: - from .apm import get_meter, get_tracer, is_apm_enabled, setup_apm - from .apm.decorators import trace_workflow, track_metric - from .apm.instrumented import APMWorkflowPrimitive - - _apm_exports = [ - "setup_apm", - "get_tracer", - "get_meter", - "is_apm_enabled", - "APMWorkflowPrimitive", - "trace_workflow", - "track_metric", - ] -except ImportError: - # APM dependencies not installed - _apm_exports = [] - -__all__ = [ - "WorkflowContext", - "WorkflowPrimitive", - "LambdaPrimitive", - "ConditionalPrimitive", - "ParallelPrimitive", - "SequentialPrimitive", - "RouterPrimitive", - "CachePrimitive", - "TimeoutPrimitive", - "TimeoutError", -] + _apm_exports - -__version__ = "0.2.0" diff --git a/packages/tta-workflow-primitives/src/tta_workflow_primitives/apm/README.md b/packages/tta-workflow-primitives/src/tta_workflow_primitives/apm/README.md deleted file mode 100644 index e51bea52..00000000 --- a/packages/tta-workflow-primitives/src/tta_workflow_primitives/apm/README.md +++ /dev/null @@ -1,279 +0,0 @@ -# APM Integration for Workflow Primitives - -OpenTelemetry-based Application Performance Monitoring for AI workflow primitives. - -## Features - -- ✅ **Automatic tracing** - Track execution flow through primitives -- ✅ **Metrics collection** - Counter and histogram metrics for performance -- ✅ **Prometheus export** - Native integration with existing Prometheus stack -- ✅ **Minimal overhead** - Gracefully degrades when APM is disabled -- ✅ **Easy to use** - Drop-in base class and decorators - -## Installation - -```bash -# Install with APM support -pip install tta-workflow-primitives[apm] - -# Or install manually -pip install opentelemetry-api opentelemetry-sdk opentelemetry-exporter-prometheus -``` - -## Quick Start - -### 1. Setup APM - -```python -from tta_workflow_primitives.apm import setup_apm - -# Setup with Prometheus export -setup_apm( - service_name="my-ai-app", - enable_prometheus=True -) -``` - -### 2. Use APM-Enabled Primitives - -#### Option A: Inherit from APMWorkflowPrimitive - -```python -from tta_workflow_primitives.apm.instrumented import APMWorkflowPrimitive -from tta_workflow_primitives.core.base import WorkflowContext - -class MyPrimitive(APMWorkflowPrimitive): - async def _execute_impl(self, input_data, context: WorkflowContext): - # Your implementation - return processed_data - -# Automatically traced and metered! -primitive = MyPrimitive(name="my_processor") -result = await primitive.execute(data, context) -``` - -#### Option B: Use Decorators - -```python -from tta_workflow_primitives.apm.decorators import trace_workflow, track_metric - -@trace_workflow("data_processing") -@track_metric("processing_operations", "counter") -async def process_data(data): - # Your code - return processed_data -``` - -### 3. View Metrics - -```bash -# Metrics available at: -curl http://localhost:9464/metrics - -# Example metrics: -# primitive_processor_executions_total{status="success"} 42 -# primitive_processor_duration_milliseconds_bucket{le="100"} 38 -# primitive_processor_duration_milliseconds_bucket{le="500"} 42 -``` - -## What Gets Tracked - -### Traces -- Execution flow through primitives -- Parent-child relationships -- Timing information -- Error details - -### Metrics -- **Execution counter**: Number of executions (success/error) -- **Duration histogram**: Execution time distribution -- **Error rates**: Failures by error type -- **Throughput**: Operations per second - -## Architecture - -``` -Your Application - ↓ -APMWorkflowPrimitive - ↓ -OpenTelemetry SDK - ↓ -Prometheus Exporter → Prometheus → Grafana -``` - -## Examples - -### Workflow Composition with APM - -```python -from tta_workflow_primitives.apm import setup_apm -from tta_workflow_primitives.apm.instrumented import APMWorkflowPrimitive -from tta_workflow_primitives.core.base import WorkflowContext - -# Setup APM -setup_apm("my-workflow") - -# Create primitives -class Step1(APMWorkflowPrimitive): - async def _execute_impl(self, data, context): - return {"step1": "done", **data} - -class Step2(APMWorkflowPrimitive): - async def _execute_impl(self, data, context): - return {"step2": "done", **data} - -# Compose workflow -workflow = Step1() >> Step2() - -# Execute (automatically tracked!) -context = WorkflowContext(workflow_id="wf-001") -result = await workflow.execute({"input": "data"}, context) - -# Traces show: Step1 → Step2 -# Metrics track both primitives -``` - -### Custom Metrics - -```python -from tta_workflow_primitives.apm import get_meter - -meter = get_meter(__name__) - -# Create custom counter -api_calls = meter.create_counter( - "api_calls_total", - description="Total API calls" -) - -# Increment -api_calls.add(1, {"endpoint": "/predict", "model": "gpt-4"}) - -# Create histogram -latency = meter.create_histogram( - "api_latency_ms", - description="API latency in milliseconds" -) - -# Record value -latency.record(123.45, {"endpoint": "/predict"}) -``` - -## Integration with Prometheus/Grafana - -### Prometheus Configuration - -```yaml -# prometheus.yml -scrape_configs: - - job_name: 'ai-workflows' - static_configs: - - targets: ['localhost:9464'] -``` - -### Example Grafana Queries - -```promql -# Execution rate -rate(primitive_processor_executions_total[5m]) - -# P95 latency -histogram_quantile(0.95, - rate(primitive_processor_duration_milliseconds_bucket[5m])) - -# Error rate -rate(primitive_processor_executions_total{status="error"}[5m]) / -rate(primitive_processor_executions_total[5m]) -``` - -## Performance Impact - -APM adds minimal overhead: -- ~1-2ms per traced operation -- ~100KB memory per 10,000 spans -- Async export doesn't block execution -- Gracefully disables if not configured - -## Best Practices - -### 1. Name Your Primitives - -```python -# Good -processor = DataProcessor(name="user_data_processor") - -# Bad -processor = DataProcessor() # Uses class name, less specific -``` - -### 2. Add Context - -```python -context = WorkflowContext( - workflow_id="unique-id", - session_id="user-session", - metadata={"user_tier": "premium"} -) -``` - -### 3. Use Appropriate Metric Types - -```python -# Counter: Things that only go up -executions_counter = meter.create_counter("executions") - -# Histogram: Distributions (latency, sizes) -duration_histogram = meter.create_histogram("duration_ms") -``` - -### 4. Add Attributes to Spans - -```python -@trace_workflow("process", attributes={"version": "2.0"}) -async def process(data): - return result -``` - -## Troubleshooting - -### APM Not Working - -```python -from tta_workflow_primitives.apm import is_apm_enabled - -if not is_apm_enabled(): - print("APM not enabled - call setup_apm() first") -``` - -### No Metrics Visible - -1. Check Prometheus is scraping: `http://localhost:9464/metrics` -2. Verify port is accessible -3. Check firewall rules - -### High Overhead - -```python -# Reduce sampling -setup_apm( - service_name="my-app", - sample_rate=0.1 # Sample 10% of traces -) -``` - -## What's Next - -- ✅ Phase 1: APM Integration (Current) -- ⏳ Phase 2: Context7 Integration -- ⏳ Phase 3: Intelligent Runtime -- ⏳ Phase 4: Auto-optimization - -See `APM_CONTEXT7_RUNTIME_PACKAGE.md` for the full roadmap. - -## Resources - -- [OpenTelemetry Python](https://opentelemetry.io/docs/instrumentation/python/) -- [Prometheus](https://prometheus.io/) -- [Grafana Dashboards](https://grafana.com/grafana/dashboards/) -- [APM Best Practices](https://opentelemetry.io/docs/concepts/signals/) diff --git a/packages/tta-workflow-primitives/src/tta_workflow_primitives/apm/__init__.py b/packages/tta-workflow-primitives/src/tta_workflow_primitives/apm/__init__.py deleted file mode 100644 index 2c33f280..00000000 --- a/packages/tta-workflow-primitives/src/tta_workflow_primitives/apm/__init__.py +++ /dev/null @@ -1,17 +0,0 @@ -"""APM (Application Performance Monitoring) module for workflow primitives. - -This module provides OpenTelemetry integration for monitoring workflow -execution, performance metrics, and distributed tracing. -""" - -from .decorators import trace_workflow, track_metric -from .setup import get_meter, get_tracer, is_apm_enabled, setup_apm - -__all__ = [ - "setup_apm", - "get_tracer", - "get_meter", - "is_apm_enabled", - "trace_workflow", - "track_metric", -] diff --git a/packages/tta-workflow-primitives/src/tta_workflow_primitives/apm/decorators.py b/packages/tta-workflow-primitives/src/tta_workflow_primitives/apm/decorators.py deleted file mode 100644 index be352166..00000000 --- a/packages/tta-workflow-primitives/src/tta_workflow_primitives/apm/decorators.py +++ /dev/null @@ -1,197 +0,0 @@ -"""Decorators for tracing and metrics.""" - -import logging -import time -from collections.abc import Callable -from functools import wraps -from typing import Any - -from .setup import get_meter, get_tracer, is_apm_enabled - -logger = logging.getLogger(__name__) - - -def trace_workflow(span_name: str | None = None, attributes: dict[str, Any] | None = None): - """Decorator to trace workflow function execution. - - Args: - span_name: Custom span name (defaults to function name) - attributes: Additional attributes to add to the span - - Example: - >>> @trace_workflow("my_workflow") - ... async def process_data(data): - ... return processed_data - """ - - def decorator(func: Callable) -> Callable: - @wraps(func) - async def async_wrapper(*args, **kwargs): - if not is_apm_enabled(): - return await func(*args, **kwargs) - - tracer = get_tracer(func.__module__) - if not tracer: - return await func(*args, **kwargs) - - name = span_name or f"{func.__module__}.{func.__name__}" - attrs = attributes or {} - attrs["function.name"] = func.__name__ - attrs["function.module"] = func.__module__ - - with tracer.start_as_current_span(name, attributes=attrs) as span: - try: - result = await func(*args, **kwargs) - span.set_attribute("function.result", "success") - return result - except Exception as e: - span.set_attribute("function.result", "error") - span.set_attribute("error.type", type(e).__name__) - span.set_attribute("error.message", str(e)) - raise - - @wraps(func) - def sync_wrapper(*args, **kwargs): - if not is_apm_enabled(): - return func(*args, **kwargs) - - tracer = get_tracer(func.__module__) - if not tracer: - return func(*args, **kwargs) - - name = span_name or f"{func.__module__}.{func.__name__}" - attrs = attributes or {} - attrs["function.name"] = func.__name__ - attrs["function.module"] = func.__module__ - - with tracer.start_as_current_span(name, attributes=attrs) as span: - try: - result = func(*args, **kwargs) - span.set_attribute("function.result", "success") - return result - except Exception as e: - span.set_attribute("function.result", "error") - span.set_attribute("error.type", type(e).__name__) - span.set_attribute("error.message", str(e)) - raise - - # Return appropriate wrapper based on function type - import inspect - - if inspect.iscoroutinefunction(func): - return async_wrapper - else: - return sync_wrapper - - return decorator - - -def track_metric( - metric_name: str, metric_type: str = "counter", description: str = "", unit: str = "1" -): - """Decorator to track metrics for function execution. - - Args: - metric_name: Name of the metric - metric_type: Type of metric ("counter", "histogram", "gauge") - description: Description of the metric - unit: Unit of measurement - - Example: - >>> @track_metric("api_calls", "counter", "Number of API calls") - ... async def call_api(): - ... return result - """ - - def decorator(func: Callable) -> Callable: - @wraps(func) - async def async_wrapper(*args, **kwargs): - if not is_apm_enabled(): - return await func(*args, **kwargs) - - meter = get_meter(func.__module__) - if not meter: - return await func(*args, **kwargs) - - # Create appropriate metric instrument - if metric_type == "counter": - instrument = meter.create_counter(metric_name, description=description, unit=unit) - elif metric_type == "histogram": - instrument = meter.create_histogram(metric_name, description=description, unit=unit) - else: - logger.warning(f"Unknown metric type: {metric_type}") - return await func(*args, **kwargs) - - # Track execution - start_time = time.time() - try: - result = await func(*args, **kwargs) - - # Record metric - if metric_type == "counter": - instrument.add(1, {"status": "success"}) - elif metric_type == "histogram": - duration = time.time() - start_time - instrument.record(duration, {"status": "success"}) - - return result - - except Exception as e: - # Record error metric - if metric_type == "counter": - instrument.add(1, {"status": "error", "error_type": type(e).__name__}) - elif metric_type == "histogram": - duration = time.time() - start_time - instrument.record(duration, {"status": "error", "error_type": type(e).__name__}) - raise - - @wraps(func) - def sync_wrapper(*args, **kwargs): - if not is_apm_enabled(): - return func(*args, **kwargs) - - meter = get_meter(func.__module__) - if not meter: - return func(*args, **kwargs) - - # Create appropriate metric instrument - if metric_type == "counter": - instrument = meter.create_counter(metric_name, description=description, unit=unit) - elif metric_type == "histogram": - instrument = meter.create_histogram(metric_name, description=description, unit=unit) - else: - logger.warning(f"Unknown metric type: {metric_type}") - return func(*args, **kwargs) - - # Track execution - start_time = time.time() - try: - result = func(*args, **kwargs) - - # Record metric - if metric_type == "counter": - instrument.add(1, {"status": "success"}) - elif metric_type == "histogram": - duration = time.time() - start_time - instrument.record(duration, {"status": "success"}) - - return result - - except Exception as e: - # Record error metric - if metric_type == "counter": - instrument.add(1, {"status": "error", "error_type": type(e).__name__}) - elif metric_type == "histogram": - duration = time.time() - start_time - instrument.record(duration, {"status": "error", "error_type": type(e).__name__}) - raise - - # Return appropriate wrapper based on function type - import inspect - - if inspect.iscoroutinefunction(func): - return async_wrapper - else: - return sync_wrapper - - return decorator diff --git a/packages/tta-workflow-primitives/src/tta_workflow_primitives/apm/instrumented.py b/packages/tta-workflow-primitives/src/tta_workflow_primitives/apm/instrumented.py deleted file mode 100644 index 7a297d48..00000000 --- a/packages/tta-workflow-primitives/src/tta_workflow_primitives/apm/instrumented.py +++ /dev/null @@ -1,163 +0,0 @@ -"""APM-enabled workflow primitive base class.""" - -import logging -import time -from typing import Any - -from ..apm import get_meter, get_tracer, is_apm_enabled -from ..core.base import WorkflowContext, WorkflowPrimitive - -logger = logging.getLogger(__name__) - - -class APMWorkflowPrimitive(WorkflowPrimitive): - """Base workflow primitive with APM instrumentation. - - This class wraps the standard WorkflowPrimitive with OpenTelemetry - tracing and metrics. It automatically tracks: - - Execution duration - - Success/failure rates - - Input/output sizes - - Error types - - Example: - >>> from tta_workflow_primitives.apm import setup_apm - >>> from tta_workflow_primitives.apm.instrumented import APMWorkflowPrimitive - >>> - >>> setup_apm("my-service") - >>> - >>> class MyPrimitive(APMWorkflowPrimitive): - ... async def execute(self, input_data, context): - ... # Your logic here - ... return result - >>> - >>> # Automatically traced and metered! - >>> result = await MyPrimitive().execute(data, context) - """ - - def __init__(self, name: str | None = None) -> None: - """Initialize APM-enabled primitive. - - Args: - name: Custom name for the primitive (defaults to class name) - """ - self.name = name or self.__class__.__name__ - self._execution_counter = None - self._duration_histogram = None - self._init_metrics() - - def _init_metrics(self) -> None: - """Initialize metrics instruments.""" - if not is_apm_enabled(): - return - - meter = get_meter(__name__) - if not meter: - return - - # Create counter for executions - self._execution_counter = meter.create_counter( - f"primitive.{self.name}.executions", - description=f"Number of executions for {self.name}", - unit="1", - ) - - # Create histogram for duration - self._duration_histogram = meter.create_histogram( - f"primitive.{self.name}.duration", - description=f"Execution duration for {self.name}", - unit="ms", - ) - - async def execute(self, input_data: Any, context: WorkflowContext) -> Any: - """Execute with APM instrumentation. - - This wraps the actual execution with tracing and metrics collection. - Subclasses should override `_execute_impl` instead of this method. - - Args: - input_data: Input data - context: Workflow context - - Returns: - Output data - """ - if not is_apm_enabled(): - return await self._execute_impl(input_data, context) - - tracer = get_tracer(__name__) - if not tracer: - return await self._execute_impl(input_data, context) - - # Start span for this execution - span_name = f"{self.name}.execute" - with tracer.start_as_current_span( - span_name, - attributes={ - "primitive.name": self.name, - "primitive.type": self.__class__.__name__, - "workflow.id": context.workflow_id or "unknown", - "session.id": context.session_id or "unknown", - }, - ) as span: - start_time = time.time() - - try: - # Execute the actual implementation - result = await self._execute_impl(input_data, context) - - # Record success - duration_ms = (time.time() - start_time) * 1000 - - span.set_attribute("execution.status", "success") - span.set_attribute("execution.duration_ms", duration_ms) - - # Update metrics - if self._execution_counter: - self._execution_counter.add(1, {"status": "success", "primitive": self.name}) - - if self._duration_histogram: - self._duration_histogram.record( - duration_ms, {"status": "success", "primitive": self.name} - ) - - return result - - except Exception as e: - # Record failure - duration_ms = (time.time() - start_time) * 1000 - error_type = type(e).__name__ - - span.set_attribute("execution.status", "error") - span.set_attribute("execution.duration_ms", duration_ms) - span.set_attribute("error.type", error_type) - span.set_attribute("error.message", str(e)) - - # Update metrics - if self._execution_counter: - self._execution_counter.add( - 1, {"status": "error", "primitive": self.name, "error_type": error_type} - ) - - if self._duration_histogram: - self._duration_histogram.record( - duration_ms, - {"status": "error", "primitive": self.name, "error_type": error_type}, - ) - - logger.error(f"Primitive {self.name} failed after {duration_ms:.2f}ms: {e}") - raise - - async def _execute_impl(self, input_data: Any, context: WorkflowContext) -> Any: - """Actual execution implementation. - - Subclasses should override this method instead of `execute`. - - Args: - input_data: Input data - context: Workflow context - - Returns: - Output data - """ - raise NotImplementedError(f"{self.__class__.__name__} must implement _execute_impl") diff --git a/packages/tta-workflow-primitives/src/tta_workflow_primitives/apm/setup.py b/packages/tta-workflow-primitives/src/tta_workflow_primitives/apm/setup.py deleted file mode 100644 index c9e6e442..00000000 --- a/packages/tta-workflow-primitives/src/tta_workflow_primitives/apm/setup.py +++ /dev/null @@ -1,159 +0,0 @@ -"""OpenTelemetry APM setup and configuration.""" - -import logging - -try: - from opentelemetry import metrics, trace - from opentelemetry.exporter.prometheus import PrometheusMetricReader - from opentelemetry.sdk.metrics import MeterProvider - from opentelemetry.sdk.resources import Resource - from opentelemetry.sdk.trace import TracerProvider - from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter - - OPENTELEMETRY_AVAILABLE = True -except ImportError: - OPENTELEMETRY_AVAILABLE = False - logging.warning( - "OpenTelemetry not installed. Install with: pip install tta-workflow-primitives[apm]" - ) - -logger = logging.getLogger(__name__) - -_tracer_provider: TracerProvider | None = None -_meter_provider: MeterProvider | None = None -_initialized = False - - -def setup_apm( - service_name: str = "ai-workflow-primitives", - service_version: str = "0.1.0", - enable_prometheus: bool = True, - enable_console: bool = False, - prometheus_port: int = 9464, -) -> tuple[TracerProvider | None, MeterProvider | None]: - """Setup OpenTelemetry APM for workflow primitives. - - Args: - service_name: Name of the service - service_version: Version of the service - enable_prometheus: Enable Prometheus metrics export - enable_console: Enable console export (for debugging) - prometheus_port: Port for Prometheus metrics endpoint - - Returns: - Tuple of (tracer_provider, meter_provider) - - Example: - >>> from tta_workflow_primitives.apm import setup_apm - >>> tracer, meter = setup_apm( - ... service_name="my-ai-app", - ... enable_prometheus=True - ... ) - """ - global _tracer_provider, _meter_provider, _initialized - - if not OPENTELEMETRY_AVAILABLE: - logger.warning("OpenTelemetry not available, APM disabled") - return None, None - - if _initialized: - logger.info("APM already initialized") - return _tracer_provider, _meter_provider - - # Create resource with service info - resource = Resource.create( - { - "service.name": service_name, - "service.version": service_version, - "library.name": "tta-workflow-primitives", - } - ) - - # Setup tracing - _tracer_provider = TracerProvider(resource=resource) - - if enable_console: - # Add console exporter for debugging - console_processor = BatchSpanProcessor(ConsoleSpanExporter()) - _tracer_provider.add_span_processor(console_processor) - logger.info("Console trace export enabled") - - trace.set_tracer_provider(_tracer_provider) - logger.info(f"Tracer initialized for service: {service_name}") - - # Setup metrics - if enable_prometheus: - # Prometheus metrics reader - prometheus_reader = PrometheusMetricReader() - _meter_provider = MeterProvider(resource=resource, metric_readers=[prometheus_reader]) - metrics.set_meter_provider(_meter_provider) - logger.info(f"Prometheus metrics enabled on port {prometheus_port}") - else: - _meter_provider = MeterProvider(resource=resource) - metrics.set_meter_provider(_meter_provider) - logger.info("Metrics provider initialized (no exporters)") - - _initialized = True - - return _tracer_provider, _meter_provider - - -def get_tracer(name: str = __name__) -> trace.Tracer | None: - """Get a tracer instance. - - Args: - name: Name for the tracer (usually __name__) - - Returns: - Tracer instance or None if not initialized - - Example: - >>> tracer = get_tracer(__name__) - >>> with tracer.start_as_current_span("my_operation"): - ... # Your code here - ... pass - """ - if not OPENTELEMETRY_AVAILABLE: - return None - - if not _initialized: - logger.warning("APM not initialized, call setup_apm() first") - return None - - return trace.get_tracer(name) - - -def get_meter(name: str = __name__) -> metrics.Meter | None: - """Get a meter instance. - - Args: - name: Name for the meter (usually __name__) - - Returns: - Meter instance or None if not initialized - - Example: - >>> meter = get_meter(__name__) - >>> counter = meter.create_counter( - ... "my_counter", - ... description="Number of operations" - ... ) - >>> counter.add(1) - """ - if not OPENTELEMETRY_AVAILABLE: - return None - - if not _initialized: - logger.warning("APM not initialized, call setup_apm() first") - return None - - return metrics.get_meter(name) - - -def is_apm_enabled() -> bool: - """Check if APM is enabled and initialized. - - Returns: - True if APM is enabled, False otherwise - """ - return OPENTELEMETRY_AVAILABLE and _initialized diff --git a/packages/tta-workflow-primitives/src/tta_workflow_primitives/core/__init__.py b/packages/tta-workflow-primitives/src/tta_workflow_primitives/core/__init__.py deleted file mode 100644 index 5557a64b..00000000 --- a/packages/tta-workflow-primitives/src/tta_workflow_primitives/core/__init__.py +++ /dev/null @@ -1,17 +0,0 @@ -"""Core workflow primitive abstractions.""" - -from .base import LambdaPrimitive, WorkflowContext, WorkflowPrimitive -from .conditional import ConditionalPrimitive -from .parallel import ParallelPrimitive -from .routing import RouterPrimitive -from .sequential import SequentialPrimitive - -__all__ = [ - "WorkflowContext", - "WorkflowPrimitive", - "LambdaPrimitive", - "ConditionalPrimitive", - "ParallelPrimitive", - "SequentialPrimitive", - "RouterPrimitive", -] diff --git a/packages/tta-workflow-primitives/src/tta_workflow_primitives/core/base.py b/packages/tta-workflow-primitives/src/tta_workflow_primitives/core/base.py deleted file mode 100644 index bc01fbbe..00000000 --- a/packages/tta-workflow-primitives/src/tta_workflow_primitives/core/base.py +++ /dev/null @@ -1,124 +0,0 @@ -"""Base workflow primitive abstractions.""" - -from __future__ import annotations - -from abc import ABC, abstractmethod -from typing import Any, Generic, TypeVar - -from pydantic import BaseModel, Field - -T = TypeVar("T") -U = TypeVar("U") -V = TypeVar("V") - - -class WorkflowContext(BaseModel): - """Context passed through workflow execution.""" - - workflow_id: str | None = None - session_id: str | None = None - player_id: str | None = None - metadata: dict[str, Any] = Field(default_factory=dict) - state: dict[str, Any] = Field(default_factory=dict) - - class Config: - arbitrary_types_allowed = True - - -class WorkflowPrimitive(Generic[T, U], ABC): - """ - Base class for composable workflow primitives. - - Primitives are the building blocks of workflows. They can be composed - using operators: - - `>>` for sequential execution (self then other) - - `|` for parallel execution (self and other concurrently) - - Example: - ```python - workflow = primitive1 >> primitive2 >> primitive3 - result = await workflow.execute(input_data, context) - ``` - """ - - @abstractmethod - async def execute(self, input_data: T, context: WorkflowContext) -> U: - """ - Execute the primitive with input data and context. - - Args: - input_data: Input data for the primitive - context: Workflow context with session/state information - - Returns: - Output data from the primitive - - Raises: - Exception: If execution fails - """ - pass - - def __rshift__(self, other: WorkflowPrimitive[U, V]) -> WorkflowPrimitive[T, V]: - """ - Chain primitives sequentially: self >> other. - - The output of self becomes the input to other. - - Args: - other: The primitive to execute after this one - - Returns: - A new sequential primitive - """ - from .sequential import SequentialPrimitive - - return SequentialPrimitive([self, other]) - - def __or__(self, other: WorkflowPrimitive[T, U]) -> WorkflowPrimitive[T, list[U]]: - """ - Execute primitives in parallel: self | other. - - Both primitives receive the same input and execute concurrently. - - Args: - other: The primitive to execute in parallel - - Returns: - A new parallel primitive - """ - from .parallel import ParallelPrimitive - - return ParallelPrimitive([self, other]) - - -class LambdaPrimitive(WorkflowPrimitive[T, U]): - """ - Primitive that wraps a simple function or lambda. - - Useful for simple transformations or adapters. - - Example: - ```python - transform = LambdaPrimitive(lambda x, ctx: x.upper()) - workflow = input_primitive >> transform >> output_primitive - ``` - """ - - def __init__(self, func: Any) -> None: - """ - Initialize with a function. - - Args: - func: Async or sync function (input, context) -> output - """ - self.func = func - import inspect - - self.is_async = inspect.iscoroutinefunction(func) - - async def execute(self, input_data: T, context: WorkflowContext) -> U: - """Execute the wrapped function.""" - if self.is_async: - return await self.func(input_data, context) - else: - return self.func(input_data, context) diff --git a/packages/tta-workflow-primitives/src/tta_workflow_primitives/core/conditional.py b/packages/tta-workflow-primitives/src/tta_workflow_primitives/core/conditional.py deleted file mode 100644 index b5e21d6c..00000000 --- a/packages/tta-workflow-primitives/src/tta_workflow_primitives/core/conditional.py +++ /dev/null @@ -1,128 +0,0 @@ -"""Conditional workflow primitive composition.""" - -from __future__ import annotations - -from collections.abc import Callable -from typing import Any - -from .base import WorkflowContext, WorkflowPrimitive - - -class ConditionalPrimitive(WorkflowPrimitive[Any, Any]): - """ - Conditional branching primitive. - - Executes different primitives based on a condition function. - - Example: - ```python - workflow = ConditionalPrimitive( - condition=lambda result, ctx: result.safety_level != "blocked", - then_primitive=standard_narrative, - else_primitive=safe_narrative - ) - ``` - """ - - def __init__( - self, - condition: Callable[[Any, WorkflowContext], bool], - then_primitive: WorkflowPrimitive, - else_primitive: WorkflowPrimitive | None = None, - ) -> None: - """ - Initialize conditional primitive. - - Args: - condition: Function (input, context) -> bool to determine branch - then_primitive: Primitive to execute if condition is True - else_primitive: Optional primitive to execute if condition is False - """ - self.condition = condition - self.then_primitive = then_primitive - self.else_primitive = else_primitive - - async def execute(self, input_data: Any, context: WorkflowContext) -> Any: - """ - Execute conditional branching. - - Args: - input_data: Input data for the primitive - context: Workflow context - - Returns: - Output from the selected branch, or input if no else branch - - Raises: - Exception: If the selected primitive fails - """ - if self.condition(input_data, context): - return await self.then_primitive.execute(input_data, context) - elif self.else_primitive: - return await self.else_primitive.execute(input_data, context) - else: - # No else branch, pass through input - return input_data - - -class SwitchPrimitive(WorkflowPrimitive[Any, Any]): - """ - Multi-way conditional branching primitive. - - Like a switch/case statement for workflows. - - Example: - ```python - workflow = SwitchPrimitive( - selector=lambda input, ctx: input.get("intent"), - cases={ - "explore": explore_primitive, - "combat": combat_primitive, - "dialogue": dialogue_primitive, - }, - default=generic_primitive - ) - ``` - """ - - def __init__( - self, - selector: Callable[[Any, WorkflowContext], str], - cases: dict[str, WorkflowPrimitive], - default: WorkflowPrimitive | None = None, - ) -> None: - """ - Initialize switch primitive. - - Args: - selector: Function (input, context) -> str to select case - cases: Map of case values to primitives - default: Optional default primitive if no case matches - """ - self.selector = selector - self.cases = cases - self.default = default - - async def execute(self, input_data: Any, context: WorkflowContext) -> Any: - """ - Execute switch branching. - - Args: - input_data: Input data for the primitive - context: Workflow context - - Returns: - Output from the selected case, default, or input - - Raises: - Exception: If the selected primitive fails - """ - case_key = self.selector(input_data, context) - - if case_key in self.cases: - return await self.cases[case_key].execute(input_data, context) - elif self.default: - return await self.default.execute(input_data, context) - else: - # No matching case or default, pass through input - return input_data diff --git a/packages/tta-workflow-primitives/src/tta_workflow_primitives/core/parallel.py b/packages/tta-workflow-primitives/src/tta_workflow_primitives/core/parallel.py deleted file mode 100644 index d27d29f2..00000000 --- a/packages/tta-workflow-primitives/src/tta_workflow_primitives/core/parallel.py +++ /dev/null @@ -1,74 +0,0 @@ -"""Parallel workflow primitive composition.""" - -from __future__ import annotations - -import asyncio -from typing import Any - -from .base import WorkflowContext, WorkflowPrimitive - - -class ParallelPrimitive(WorkflowPrimitive[Any, list[Any]]): - """ - Execute primitives in parallel. - - All primitives receive the same input and execute concurrently. - Results are collected in a list. - - Example: - ```python - workflow = ParallelPrimitive([ - world_building, - character_analysis, - theme_analysis - ]) - # Or use | operator: - workflow = world_building | character_analysis | theme_analysis - ``` - """ - - def __init__(self, primitives: list[WorkflowPrimitive]) -> None: - """ - Initialize with a list of primitives. - - Args: - primitives: List of primitives to execute in parallel - """ - if not primitives: - raise ValueError("ParallelPrimitive requires at least one primitive") - self.primitives = primitives - - async def execute(self, input_data: Any, context: WorkflowContext) -> list[Any]: - """ - Execute primitives in parallel. - - Args: - input_data: Input data sent to all primitives - context: Workflow context - - Returns: - List of outputs from all primitives (in order) - - Raises: - Exception: If any primitive fails - """ - tasks = [primitive.execute(input_data, context) for primitive in self.primitives] - return await asyncio.gather(*tasks) - - def __or__(self, other: WorkflowPrimitive) -> ParallelPrimitive: - """ - Add another primitive to parallel execution: self | other. - - Optimizes by flattening nested parallel primitives. - - Args: - other: Primitive to add to parallel execution - - Returns: - A new parallel primitive with all branches - """ - if isinstance(other, ParallelPrimitive): - # Flatten nested parallel primitives - return ParallelPrimitive(self.primitives + other.primitives) - else: - return ParallelPrimitive(self.primitives + [other]) diff --git a/packages/tta-workflow-primitives/src/tta_workflow_primitives/core/routing.py b/packages/tta-workflow-primitives/src/tta_workflow_primitives/core/routing.py deleted file mode 100644 index 7f2961f4..00000000 --- a/packages/tta-workflow-primitives/src/tta_workflow_primitives/core/routing.py +++ /dev/null @@ -1,112 +0,0 @@ -"""Routing primitive for intelligent workflow branching.""" - -from __future__ import annotations - -from collections.abc import Callable -from typing import Any - -from ..observability.logging import get_logger -from .base import WorkflowContext, WorkflowPrimitive - -logger = get_logger(__name__) - - -class RouterPrimitive(WorkflowPrimitive[Any, Any]): - """ - Route input to appropriate primitive based on routing function. - - Enables intelligent routing decisions based on: - - Cost optimization (route to cheaper providers) - - Latency optimization (route to faster providers) - - Load balancing (distribute across providers) - - Feature requirements (route to capable providers) - - Example: - ```python - # Route based on user tier - router = RouterPrimitive( - routes={ - "openai": openai_primitive, - "anthropic": anthropic_primitive, - "local": local_llm_primitive - }, - router_fn=lambda data, ctx: ctx.metadata.get("provider", "openai"), - default="openai" - ) - - # Route based on complexity - router = RouterPrimitive( - routes={ - "simple": fast_local_model, - "complex": premium_cloud_model - }, - router_fn=lambda data, ctx: ( - "simple" if len(data.get("prompt", "")) < 100 else "complex" - ), - default="simple" - ) - ``` - """ - - def __init__( - self, - routes: dict[str, WorkflowPrimitive], - router_fn: Callable[[Any, WorkflowContext], str], - default: str | None = None, - ) -> None: - """ - Initialize router primitive. - - Args: - routes: Map of route keys to primitives - router_fn: Function to determine route from input/context - default: Default route if router_fn returns unknown key - """ - self.routes = routes - self.router_fn = router_fn - self.default = default - - async def execute(self, input_data: Any, context: WorkflowContext) -> Any: - """ - Execute routing logic and invoke selected primitive. - - Args: - input_data: Input data for routing decision - context: Workflow context - - Returns: - Output from selected primitive - - Raises: - ValueError: If route key not found and no default specified - """ - # Determine route - route_key = self.router_fn(input_data, context) - - # Get primitive - primitive = self.routes.get(route_key) - - # Fallback to default - if not primitive and self.default: - route_key = self.default - primitive = self.routes.get(route_key) - - if not primitive: - available = ", ".join(self.routes.keys()) - raise ValueError(f"No route found for key '{route_key}'. Available routes: {available}") - - # Log routing decision - logger.info( - "routing_decision", - route=route_key, - available_routes=list(self.routes.keys()), - workflow_id=context.workflow_id, - ) - - # Store routing decision in context - if "routing_history" not in context.state: - context.state["routing_history"] = [] - context.state["routing_history"].append(route_key) - - # Execute selected primitive - return await primitive.execute(input_data, context) diff --git a/packages/tta-workflow-primitives/src/tta_workflow_primitives/core/sequential.py b/packages/tta-workflow-primitives/src/tta_workflow_primitives/core/sequential.py deleted file mode 100644 index 5896c991..00000000 --- a/packages/tta-workflow-primitives/src/tta_workflow_primitives/core/sequential.py +++ /dev/null @@ -1,74 +0,0 @@ -"""Sequential workflow primitive composition.""" - -from __future__ import annotations - -from typing import Any - -from .base import WorkflowContext, WorkflowPrimitive - - -class SequentialPrimitive(WorkflowPrimitive[Any, Any]): - """ - Execute primitives in sequence. - - Each primitive's output becomes the next primitive's input. - - Example: - ```python - workflow = SequentialPrimitive([ - input_processing, - world_building, - narrative_generation - ]) - # Or use >> operator: - workflow = input_processing >> world_building >> narrative_generation - ``` - """ - - def __init__(self, primitives: list[WorkflowPrimitive]) -> None: - """ - Initialize with a list of primitives. - - Args: - primitives: List of primitives to execute in order - """ - if not primitives: - raise ValueError("SequentialPrimitive requires at least one primitive") - self.primitives = primitives - - async def execute(self, input_data: Any, context: WorkflowContext) -> Any: - """ - Execute primitives sequentially. - - Args: - input_data: Initial input data - context: Workflow context - - Returns: - Output from the last primitive - - Raises: - Exception: If any primitive fails - """ - result = input_data - for primitive in self.primitives: - result = await primitive.execute(result, context) - return result - - def __rshift__(self, other: WorkflowPrimitive) -> SequentialPrimitive: - """ - Chain another primitive: self >> other. - - Optimizes by flattening nested sequential primitives. - - Args: - other: Primitive to append - - Returns: - A new sequential primitive with all steps - """ - if isinstance(other, SequentialPrimitive): - # Flatten nested sequential primitives - return SequentialPrimitive(self.primitives + other.primitives) - else: - return SequentialPrimitive(self.primitives + [other]) diff --git a/packages/tta-workflow-primitives/src/tta_workflow_primitives/observability/__init__.py b/packages/tta-workflow-primitives/src/tta_workflow_primitives/observability/__init__.py deleted file mode 100644 index 8ecbd81b..00000000 --- a/packages/tta-workflow-primitives/src/tta_workflow_primitives/observability/__init__.py +++ /dev/null @@ -1,13 +0,0 @@ -"""Observability features for workflow primitives.""" - -from .logging import setup_logging -from .metrics import PrimitiveMetrics, get_metrics_collector -from .tracing import ObservablePrimitive, setup_tracing - -__all__ = [ - "ObservablePrimitive", - "PrimitiveMetrics", - "get_metrics_collector", - "setup_logging", - "setup_tracing", -] diff --git a/packages/tta-workflow-primitives/src/tta_workflow_primitives/observability/logging.py b/packages/tta-workflow-primitives/src/tta_workflow_primitives/observability/logging.py deleted file mode 100644 index 18940f85..00000000 --- a/packages/tta-workflow-primitives/src/tta_workflow_primitives/observability/logging.py +++ /dev/null @@ -1,60 +0,0 @@ -"""Structured logging for workflow primitives.""" - -from __future__ import annotations - -import logging -import sys - -try: - import structlog - - STRUCTLOG_AVAILABLE = True -except ImportError: - STRUCTLOG_AVAILABLE = False - - -def setup_logging(level: str = "INFO") -> None: - """ - Setup structured logging. - - Args: - level: Log level (DEBUG, INFO, WARNING, ERROR) - """ - if STRUCTLOG_AVAILABLE: - structlog.configure( - processors=[ - structlog.contextvars.merge_contextvars, - structlog.processors.add_log_level, - structlog.processors.StackInfoRenderer(), - structlog.dev.set_exc_info, - structlog.processors.TimeStamper(fmt="iso"), - structlog.dev.ConsoleRenderer(), - ], - wrapper_class=structlog.make_filtering_bound_logger(getattr(logging, level.upper())), - context_class=dict, - logger_factory=structlog.PrintLoggerFactory(), - cache_logger_on_first_use=False, - ) - else: - # Fallback to standard logging - logging.basicConfig( - level=getattr(logging, level.upper()), - format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", - stream=sys.stdout, - ) - - -def get_logger(name: str) -> Any: - """ - Get a logger instance. - - Args: - name: Logger name - - Returns: - Logger instance (structlog or standard logging) - """ - if STRUCTLOG_AVAILABLE: - return structlog.get_logger(name) - else: - return logging.getLogger(name) diff --git a/packages/tta-workflow-primitives/src/tta_workflow_primitives/observability/metrics.py b/packages/tta-workflow-primitives/src/tta_workflow_primitives/observability/metrics.py deleted file mode 100644 index 7e6399c7..00000000 --- a/packages/tta-workflow-primitives/src/tta_workflow_primitives/observability/metrics.py +++ /dev/null @@ -1,120 +0,0 @@ -"""Metrics collection for workflow primitives.""" - -from __future__ import annotations - -from dataclasses import dataclass, field -from typing import Any - - -@dataclass -class PrimitiveMetrics: - """Metrics for a single primitive.""" - - name: str - total_executions: int = 0 - successful_executions: int = 0 - failed_executions: int = 0 - total_duration_ms: float = 0.0 - min_duration_ms: float = float("inf") - max_duration_ms: float = 0.0 - error_counts: dict[str, int] = field(default_factory=dict) - - @property - def success_rate(self) -> float: - """Calculate success rate.""" - if self.total_executions == 0: - return 0.0 - return self.successful_executions / self.total_executions - - @property - def average_duration_ms(self) -> float: - """Calculate average duration.""" - if self.total_executions == 0: - return 0.0 - return self.total_duration_ms / self.total_executions - - def to_dict(self) -> dict[str, Any]: - """Convert to dictionary.""" - return { - "name": self.name, - "total_executions": self.total_executions, - "successful_executions": self.successful_executions, - "failed_executions": self.failed_executions, - "success_rate": self.success_rate, - "total_duration_ms": self.total_duration_ms, - "average_duration_ms": self.average_duration_ms, - "min_duration_ms": self.min_duration_ms if self.min_duration_ms != float("inf") else 0, - "max_duration_ms": self.max_duration_ms, - "error_counts": self.error_counts, - } - - -class MetricsCollector: - """Collects metrics for all primitives.""" - - def __init__(self) -> None: - self._metrics: dict[str, PrimitiveMetrics] = {} - - def record_execution( - self, - primitive_name: str, - duration_ms: float, - success: bool, - error_type: str | None = None, - ) -> None: - """ - Record a primitive execution. - - Args: - primitive_name: Name of the primitive - duration_ms: Execution duration in milliseconds - success: Whether execution succeeded - error_type: Type of error if failed - """ - if primitive_name not in self._metrics: - self._metrics[primitive_name] = PrimitiveMetrics(name=primitive_name) - - metrics = self._metrics[primitive_name] - metrics.total_executions += 1 - metrics.total_duration_ms += duration_ms - metrics.min_duration_ms = min(metrics.min_duration_ms, duration_ms) - metrics.max_duration_ms = max(metrics.max_duration_ms, duration_ms) - - if success: - metrics.successful_executions += 1 - else: - metrics.failed_executions += 1 - if error_type: - metrics.error_counts[error_type] = metrics.error_counts.get(error_type, 0) + 1 - - def get_metrics(self, primitive_name: str | None = None) -> dict[str, Any]: - """ - Get metrics for a primitive or all primitives. - - Args: - primitive_name: Optional primitive name, or None for all - - Returns: - Metrics dictionary - """ - if primitive_name: - metrics = self._metrics.get(primitive_name) - return metrics.to_dict() if metrics else {} - else: - return {name: metrics.to_dict() for name, metrics in self._metrics.items()} - - def reset(self) -> None: - """Reset all metrics.""" - self._metrics.clear() - - -# Global metrics collector -_metrics_collector: MetricsCollector | None = None - - -def get_metrics_collector() -> MetricsCollector: - """Get the global metrics collector.""" - global _metrics_collector - if _metrics_collector is None: - _metrics_collector = MetricsCollector() - return _metrics_collector diff --git a/packages/tta-workflow-primitives/src/tta_workflow_primitives/observability/tracing.py b/packages/tta-workflow-primitives/src/tta_workflow_primitives/observability/tracing.py deleted file mode 100644 index a047200f..00000000 --- a/packages/tta-workflow-primitives/src/tta_workflow_primitives/observability/tracing.py +++ /dev/null @@ -1,150 +0,0 @@ -"""Distributed tracing for workflow primitives.""" - -from __future__ import annotations - -import time -from typing import Any - -try: - from opentelemetry import trace - from opentelemetry.trace import Status, StatusCode - - TRACING_AVAILABLE = True -except ImportError: - TRACING_AVAILABLE = False - -from ..core.base import WorkflowContext, WorkflowPrimitive - - -def setup_tracing(service_name: str = "tta-workflow") -> None: - """ - Setup OpenTelemetry tracing. - - Args: - service_name: Name of the service for traces - """ - if not TRACING_AVAILABLE: - return - - from opentelemetry.sdk.resources import Resource - from opentelemetry.sdk.trace import TracerProvider - from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter - - resource = Resource.create({"service.name": service_name}) - provider = TracerProvider(resource=resource) - processor = BatchSpanProcessor(ConsoleSpanExporter()) - provider.add_span_processor(processor) - trace.set_tracer_provider(provider) - - -class ObservablePrimitive(WorkflowPrimitive[Any, Any]): - """ - Wrapper adding observability to any primitive. - - Provides: - - Distributed tracing with OpenTelemetry - - Structured logging with correlation IDs - - Metrics collection - - Example: - ```python - workflow = ( - ObservablePrimitive(input_proc, "input_processing") >> - ObservablePrimitive(world_build, "world_building") >> - ObservablePrimitive(narrative_gen, "narrative_generation") - ) - ``` - """ - - def __init__(self, primitive: WorkflowPrimitive, name: str) -> None: - """ - Initialize observable primitive. - - Args: - primitive: The primitive to wrap - name: Name for tracing and metrics - """ - self.primitive = primitive - self.name = name - self.tracer = trace.get_tracer(__name__) if TRACING_AVAILABLE else None - - async def execute(self, input_data: Any, context: WorkflowContext) -> Any: - """ - Execute primitive with observability. - - Args: - input_data: Input data - context: Workflow context - - Returns: - Output from the wrapped primitive - - Raises: - Exception: If execution fails - """ - start_time = time.time() - - # Create span if tracing is available - if self.tracer: - with self.tracer.start_as_current_span( - f"primitive.{self.name}", - attributes={ - "primitive.name": self.name, - "workflow.id": context.workflow_id or "unknown", - "session.id": context.session_id or "unknown", - }, - ) as span: - try: - result = await self.primitive.execute(input_data, context) - duration_ms = (time.time() - start_time) * 1000 - - span.set_status(Status(StatusCode.OK)) - span.set_attribute("primitive.duration_ms", duration_ms) - - # Record metrics - from .metrics import get_metrics_collector - - metrics = get_metrics_collector() - metrics.record_execution(self.name, duration_ms, success=True) - - return result - - except Exception as e: - duration_ms = (time.time() - start_time) * 1000 - - span.set_status(Status(StatusCode.ERROR, str(e))) - span.record_exception(e) - - # Record failure metrics - from .metrics import get_metrics_collector - - metrics = get_metrics_collector() - metrics.record_execution( - self.name, duration_ms, success=False, error_type=type(e).__name__ - ) - - raise - else: - # No tracing, just execute with metrics - try: - result = await self.primitive.execute(input_data, context) - duration_ms = (time.time() - start_time) * 1000 - - from .metrics import get_metrics_collector - - metrics = get_metrics_collector() - metrics.record_execution(self.name, duration_ms, success=True) - - return result - - except Exception as e: - duration_ms = (time.time() - start_time) * 1000 - - from .metrics import get_metrics_collector - - metrics = get_metrics_collector() - metrics.record_execution( - self.name, duration_ms, success=False, error_type=type(e).__name__ - ) - - raise diff --git a/packages/tta-workflow-primitives/src/tta_workflow_primitives/performance/__init__.py b/packages/tta-workflow-primitives/src/tta_workflow_primitives/performance/__init__.py deleted file mode 100644 index 662cc739..00000000 --- a/packages/tta-workflow-primitives/src/tta_workflow_primitives/performance/__init__.py +++ /dev/null @@ -1,5 +0,0 @@ -"""Performance optimization primitives.""" - -from .cache import CachePrimitive - -__all__ = ["CachePrimitive"] diff --git a/packages/tta-workflow-primitives/src/tta_workflow_primitives/performance/cache.py b/packages/tta-workflow-primitives/src/tta_workflow_primitives/performance/cache.py deleted file mode 100644 index 1f659779..00000000 --- a/packages/tta-workflow-primitives/src/tta_workflow_primitives/performance/cache.py +++ /dev/null @@ -1,207 +0,0 @@ -"""Caching primitive for workflow results.""" - -from __future__ import annotations - -import time -from collections.abc import Callable -from typing import Any - -from ..core.base import WorkflowContext, WorkflowPrimitive -from ..observability.logging import get_logger - -logger = get_logger(__name__) - - -class CachePrimitive(WorkflowPrimitive[Any, Any]): - """ - Cache primitive execution results. - - Dramatically reduces costs and latency by caching expensive operations - like LLM calls. Typical cache hit rates of 60-80% translate to 40%+ cost - reduction in production. - - Example: - ```python - # Cache expensive LLM calls - cached_llm = CachePrimitive( - primitive=expensive_llm_call, - cache_key_fn=lambda data, ctx: f"{data['prompt']}:{ctx.player_id}", - ttl_seconds=3600.0 # 1 hour TTL - ) - - # Cache with custom key generation - cached = CachePrimitive( - primitive=world_builder, - cache_key_fn=lambda data, ctx: ( - f"{data['theme']}:{data['setting']}:{ctx.session_id}" - ), - ttl_seconds=1800.0 # 30 minutes - ) - - # Short-lived cache for rapid iterations - cached = CachePrimitive( - primitive=validation_check, - cache_key_fn=lambda data, ctx: str(hash(str(data))), - ttl_seconds=60.0 # 1 minute - ) - ``` - """ - - def __init__( - self, - primitive: WorkflowPrimitive, - cache_key_fn: Callable[[Any, WorkflowContext], str], - ttl_seconds: float = 3600.0, - ) -> None: - """ - Initialize cache primitive. - - Args: - primitive: Primitive to cache - cache_key_fn: Function to generate cache key from input/context - ttl_seconds: Time-to-live for cached values (default 1 hour) - """ - self.primitive = primitive - self.cache_key_fn = cache_key_fn - self.ttl_seconds = ttl_seconds - self._cache: dict[str, tuple[Any, float]] = {} - self._stats = { - "hits": 0, - "misses": 0, - "expirations": 0, - } - - async def execute(self, input_data: Any, context: WorkflowContext) -> Any: - """ - Execute with caching. - - Args: - input_data: Input data - context: Workflow context - - Returns: - Cached or freshly computed result - """ - # Generate cache key - cache_key = self.cache_key_fn(input_data, context) - - # Check cache - if cache_key in self._cache: - result, timestamp = self._cache[cache_key] - age = time.time() - timestamp - - if age < self.ttl_seconds: - # Cache hit - self._stats["hits"] += 1 - - logger.info( - "cache_hit", - key=cache_key[:50], # Truncate long keys - age_seconds=round(age, 2), - ttl=self.ttl_seconds, - hit_rate=self.get_hit_rate(), - workflow_id=context.workflow_id, - ) - - # Track cache hits in context - if "cache_hits" not in context.state: - context.state["cache_hits"] = 0 - context.state["cache_hits"] += 1 - - return result - else: - # Cache expired - self._stats["expirations"] += 1 - logger.debug( - "cache_expired", - key=cache_key[:50], - age=round(age, 2), - ttl=self.ttl_seconds, - ) - del self._cache[cache_key] - - # Cache miss - execute and store - self._stats["misses"] += 1 - - logger.info( - "cache_miss", - key=cache_key[:50], - cache_size=len(self._cache), - hit_rate=self.get_hit_rate(), - workflow_id=context.workflow_id, - ) - - # Track cache misses in context - if "cache_misses" not in context.state: - context.state["cache_misses"] = 0 - context.state["cache_misses"] += 1 - - # Execute primitive - result = await self.primitive.execute(input_data, context) - - # Store in cache - self._cache[cache_key] = (result, time.time()) - - logger.debug( - "cache_store", - key=cache_key[:50], - cache_size=len(self._cache), - ) - - return result - - def clear_cache(self) -> None: - """Clear all cached values.""" - size = len(self._cache) - self._cache.clear() - logger.info("cache_cleared", previous_size=size) - - def get_stats(self) -> dict: - """ - Get cache statistics. - - Returns: - Dictionary with cache metrics - """ - return { - "size": len(self._cache), - "hits": self._stats["hits"], - "misses": self._stats["misses"], - "expirations": self._stats["expirations"], - "hit_rate": self.get_hit_rate(), - } - - def get_hit_rate(self) -> float: - """ - Calculate cache hit rate. - - Returns: - Hit rate as percentage (0-100) - """ - total = self._stats["hits"] + self._stats["misses"] - if total == 0: - return 0.0 - return round((self._stats["hits"] / total) * 100, 2) - - def evict_expired(self) -> int: - """ - Manually evict expired cache entries. - - Returns: - Number of entries evicted - """ - now = time.time() - expired_keys = [ - key - for key, (_, timestamp) in self._cache.items() - if now - timestamp >= self.ttl_seconds - ] - - for key in expired_keys: - del self._cache[key] - self._stats["expirations"] += 1 - - if expired_keys: - logger.info("cache_eviction", count=len(expired_keys)) - - return len(expired_keys) diff --git a/packages/tta-workflow-primitives/src/tta_workflow_primitives/recovery/__init__.py b/packages/tta-workflow-primitives/src/tta_workflow_primitives/recovery/__init__.py deleted file mode 100644 index d720045e..00000000 --- a/packages/tta-workflow-primitives/src/tta_workflow_primitives/recovery/__init__.py +++ /dev/null @@ -1,17 +0,0 @@ -"""Error recovery patterns for workflow primitives.""" - -from .compensation import CompensationStrategy, SagaPrimitive -from .fallback import FallbackPrimitive, FallbackStrategy -from .retry import RetryPrimitive, RetryStrategy -from .timeout import TimeoutError, TimeoutPrimitive - -__all__ = [ - "CompensationStrategy", - "FallbackPrimitive", - "FallbackStrategy", - "RetryPrimitive", - "RetryStrategy", - "SagaPrimitive", - "TimeoutPrimitive", - "TimeoutError", -] diff --git a/packages/tta-workflow-primitives/src/tta_workflow_primitives/recovery/compensation.py b/packages/tta-workflow-primitives/src/tta_workflow_primitives/recovery/compensation.py deleted file mode 100644 index dffc8d73..00000000 --- a/packages/tta-workflow-primitives/src/tta_workflow_primitives/recovery/compensation.py +++ /dev/null @@ -1,96 +0,0 @@ -"""Compensation patterns for workflow primitives (Saga pattern).""" - -from __future__ import annotations - -from typing import Any - -from ..core.base import WorkflowContext, WorkflowPrimitive -from ..observability.logging import get_logger - -logger = get_logger(__name__) - - -class CompensationStrategy: - """Strategy for compensating transaction (undoing effects).""" - - def __init__(self, compensation_primitive: WorkflowPrimitive) -> None: - """ - Initialize compensation strategy. - - Args: - compensation_primitive: Primitive to run for compensation - """ - self.compensation_primitive = compensation_primitive - - -class SagaPrimitive(WorkflowPrimitive[Any, Any]): - """ - Saga pattern: Execute with compensation on failure. - - Useful for maintaining consistency across distributed operations. - - Example: - ```python - workflow = SagaPrimitive( - forward=update_world_state, - compensation=rollback_world_state - ) - ``` - """ - - def __init__( - self, - forward: WorkflowPrimitive, - compensation: WorkflowPrimitive, - ) -> None: - """ - Initialize saga primitive. - - Args: - forward: Forward transaction primitive - compensation: Compensation primitive (runs on failure) - """ - self.forward = forward - self.compensation = compensation - - async def execute(self, input_data: Any, context: WorkflowContext) -> Any: - """ - Execute with saga pattern. - - Args: - input_data: Input data - context: Workflow context - - Returns: - Output from forward primitive - - Raises: - Exception: After running compensation - """ - try: - return await self.forward.execute(input_data, context) - - except Exception as forward_error: - logger.warning( - "saga_compensation_triggered", - forward=self.forward.__class__.__name__, - compensation=self.compensation.__class__.__name__, - error=str(forward_error), - ) - - try: - await self.compensation.execute(input_data, context) - logger.info( - "saga_compensation_succeeded", - compensation=self.compensation.__class__.__name__, - ) - - except Exception as compensation_error: - logger.error( - "saga_compensation_failed", - forward_error=str(forward_error), - compensation_error=str(compensation_error), - ) - - # Always re-raise the original error - raise forward_error diff --git a/packages/tta-workflow-primitives/src/tta_workflow_primitives/recovery/fallback.py b/packages/tta-workflow-primitives/src/tta_workflow_primitives/recovery/fallback.py deleted file mode 100644 index ab342eb2..00000000 --- a/packages/tta-workflow-primitives/src/tta_workflow_primitives/recovery/fallback.py +++ /dev/null @@ -1,94 +0,0 @@ -"""Fallback strategies for workflow primitives.""" - -from __future__ import annotations - -from typing import Any - -from ..core.base import WorkflowContext, WorkflowPrimitive -from ..observability.logging import get_logger - -logger = get_logger(__name__) - - -class FallbackStrategy: - """Strategy for fallback to alternative primitive.""" - - def __init__(self, fallback_primitive: WorkflowPrimitive) -> None: - """ - Initialize fallback strategy. - - Args: - fallback_primitive: Alternative primitive to use on failure - """ - self.fallback_primitive = fallback_primitive - - -class FallbackPrimitive(WorkflowPrimitive[Any, Any]): - """ - Try a primitive with fallback to alternative. - - Example: - ```python - workflow = FallbackPrimitive( - primary=openai_narrative, - fallback=local_narrative - ) - ``` - """ - - def __init__( - self, - primary: WorkflowPrimitive, - fallback: WorkflowPrimitive, - ) -> None: - """ - Initialize fallback primitive. - - Args: - primary: Primary primitive to try first - fallback: Fallback primitive if primary fails - """ - self.primary = primary - self.fallback = fallback - - async def execute(self, input_data: Any, context: WorkflowContext) -> Any: - """ - Execute with fallback logic. - - Args: - input_data: Input data - context: Workflow context - - Returns: - Output from primary or fallback - - Raises: - Exception: If both primary and fallback fail - """ - try: - return await self.primary.execute(input_data, context) - - except Exception as primary_error: - logger.warning( - "primitive_fallback_triggered", - primary=self.primary.__class__.__name__, - fallback=self.fallback.__class__.__name__, - error=str(primary_error), - ) - - try: - result = await self.fallback.execute(input_data, context) - logger.info( - "primitive_fallback_succeeded", - fallback=self.fallback.__class__.__name__, - ) - return result - - except Exception as fallback_error: - logger.error( - "primitive_fallback_failed", - primary_error=str(primary_error), - fallback_error=str(fallback_error), - ) - # Re-raise the original error - raise primary_error diff --git a/packages/tta-workflow-primitives/src/tta_workflow_primitives/recovery/retry.py b/packages/tta-workflow-primitives/src/tta_workflow_primitives/recovery/retry.py deleted file mode 100644 index 637aad72..00000000 --- a/packages/tta-workflow-primitives/src/tta_workflow_primitives/recovery/retry.py +++ /dev/null @@ -1,113 +0,0 @@ -"""Retry strategies for workflow primitives.""" - -from __future__ import annotations - -import asyncio -import random -from dataclasses import dataclass -from typing import Any - -from ..core.base import WorkflowContext, WorkflowPrimitive -from ..observability.logging import get_logger - -logger = get_logger(__name__) - - -@dataclass -class RetryStrategy: - """Configuration for retry behavior.""" - - max_retries: int = 3 - backoff_base: float = 2.0 - max_backoff: float = 60.0 - jitter: bool = True - - def calculate_delay(self, attempt: int) -> float: - """ - Calculate delay before next retry. - - Args: - attempt: Current attempt number (0-indexed) - - Returns: - Delay in seconds - """ - delay = min(self.backoff_base**attempt, self.max_backoff) - - if self.jitter: - delay *= 0.5 + random.random() - - return delay - - -class RetryPrimitive(WorkflowPrimitive[Any, Any]): - """ - Retry a primitive with exponential backoff. - - Example: - ```python - workflow = RetryPrimitive( - risky_primitive, - strategy=RetryStrategy(max_retries=3, backoff_base=2.0) - ) - ``` - """ - - def __init__( - self, - primitive: WorkflowPrimitive, - strategy: RetryStrategy | None = None, - ) -> None: - """ - Initialize retry primitive. - - Args: - primitive: The primitive to retry - strategy: Retry strategy configuration - """ - self.primitive = primitive - self.strategy = strategy or RetryStrategy() - - async def execute(self, input_data: Any, context: WorkflowContext) -> Any: - """ - Execute primitive with retry logic. - - Args: - input_data: Input data - context: Workflow context - - Returns: - Output from the primitive - - Raises: - Exception: If all retries fail - """ - last_error = None - - for attempt in range(self.strategy.max_retries + 1): - try: - return await self.primitive.execute(input_data, context) - - except Exception as e: - last_error = e - - if attempt < self.strategy.max_retries: - delay = self.strategy.calculate_delay(attempt) - logger.warning( - "primitive_retry", - primitive=self.primitive.__class__.__name__, - attempt=attempt + 1, - max_retries=self.strategy.max_retries + 1, - delay=delay, - error=str(e), - ) - await asyncio.sleep(delay) - else: - logger.error( - "primitive_retry_exhausted", - primitive=self.primitive.__class__.__name__, - attempts=self.strategy.max_retries + 1, - error=str(e), - ) - - raise last_error diff --git a/packages/tta-workflow-primitives/src/tta_workflow_primitives/recovery/timeout.py b/packages/tta-workflow-primitives/src/tta_workflow_primitives/recovery/timeout.py deleted file mode 100644 index cc185445..00000000 --- a/packages/tta-workflow-primitives/src/tta_workflow_primitives/recovery/timeout.py +++ /dev/null @@ -1,136 +0,0 @@ -"""Timeout enforcement for primitives.""" - -from __future__ import annotations - -import asyncio -import builtins -from typing import Any - -from ..core.base import WorkflowContext, WorkflowPrimitive -from ..observability.logging import get_logger - -logger = get_logger(__name__) - - -class TimeoutError(Exception): - """Timeout exceeded during execution.""" - - pass - - -class TimeoutPrimitive(WorkflowPrimitive[Any, Any]): - """ - Enforce execution timeout with optional fallback. - - Prevents workflows from hanging indefinitely by enforcing time limits. - Essential for maintaining good UX and resource efficiency. - - Example: - ```python - # Simple timeout - workflow = TimeoutPrimitive( - primitive=slow_operation, - timeout_seconds=30.0 - ) - - # Timeout with fallback - workflow = TimeoutPrimitive( - primitive=expensive_llm_call, - timeout_seconds=30.0, - fallback=cached_response_primitive - ) - - # Timeout with monitoring - workflow = TimeoutPrimitive( - primitive=critical_operation, - timeout_seconds=45.0, - fallback=degraded_service, - track_timeouts=True - ) - ``` - """ - - def __init__( - self, - primitive: WorkflowPrimitive, - timeout_seconds: float, - fallback: WorkflowPrimitive | None = None, - track_timeouts: bool = True, - ) -> None: - """ - Initialize timeout primitive. - - Args: - primitive: Primitive to execute with timeout - timeout_seconds: Maximum execution time in seconds - fallback: Optional fallback primitive on timeout - track_timeouts: Whether to track timeout occurrences in context - """ - self.primitive = primitive - self.timeout_seconds = timeout_seconds - self.fallback = fallback - self.track_timeouts = track_timeouts - - async def execute(self, input_data: Any, context: WorkflowContext) -> Any: - """ - Execute with timeout enforcement. - - Args: - input_data: Input data - context: Workflow context - - Returns: - Output from primitive or fallback - - Raises: - TimeoutError: If timeout exceeded and no fallback provided - """ - try: - result = await asyncio.wait_for( - self.primitive.execute(input_data, context), timeout=self.timeout_seconds - ) - - logger.info( - "timeout_success", - primitive=self.primitive.__class__.__name__, - timeout=self.timeout_seconds, - workflow_id=context.workflow_id, - ) - - return result - - except builtins.TimeoutError: - logger.warning( - "timeout_exceeded", - primitive=self.primitive.__class__.__name__, - timeout=self.timeout_seconds, - has_fallback=self.fallback is not None, - workflow_id=context.workflow_id, - ) - - # Track timeout in context - if self.track_timeouts: - if "timeout_count" not in context.state: - context.state["timeout_count"] = 0 - context.state["timeout_count"] += 1 - - if "timeout_history" not in context.state: - context.state["timeout_history"] = [] - context.state["timeout_history"].append( - { - "primitive": self.primitive.__class__.__name__, - "timeout": self.timeout_seconds, - "had_fallback": self.fallback is not None, - } - ) - - # Execute fallback if available - if self.fallback: - logger.info( - "executing_fallback", - fallback=self.fallback.__class__.__name__, - ) - return await self.fallback.execute(input_data, context) - - # No fallback - raise error - raise TimeoutError(f"Execution exceeded {self.timeout_seconds}s timeout") diff --git a/packages/tta-workflow-primitives/src/tta_workflow_primitives/testing/__init__.py b/packages/tta-workflow-primitives/src/tta_workflow_primitives/testing/__init__.py deleted file mode 100644 index 8c59bc1e..00000000 --- a/packages/tta-workflow-primitives/src/tta_workflow_primitives/testing/__init__.py +++ /dev/null @@ -1,8 +0,0 @@ -"""Testing utilities for workflow primitives.""" - -from .mocks import MockPrimitive, WorkflowTestCase - -__all__ = [ - "MockPrimitive", - "WorkflowTestCase", -] diff --git a/packages/tta-workflow-primitives/src/tta_workflow_primitives/testing/mocks.py b/packages/tta-workflow-primitives/src/tta_workflow_primitives/testing/mocks.py deleted file mode 100644 index 738ece55..00000000 --- a/packages/tta-workflow-primitives/src/tta_workflow_primitives/testing/mocks.py +++ /dev/null @@ -1,178 +0,0 @@ -"""Mock primitives for testing.""" - -from __future__ import annotations - -from collections.abc import Callable -from typing import Any - -from ..core.base import WorkflowContext, WorkflowPrimitive - - -class MockPrimitive(WorkflowPrimitive[Any, Any]): - """ - Mock primitive for testing. - - Example: - ```python - mock = MockPrimitive( - name="test_primitive", - return_value={"result": "success"} - ) - - workflow = mock >> another_primitive - result = await workflow.execute(input_data, context) - - assert mock.call_count == 1 - assert mock.calls[0][0] == input_data - ``` - """ - - def __init__( - self, - name: str, - return_value: Any | None = None, - side_effect: Callable | None = None, - raise_error: Exception | None = None, - ) -> None: - """ - Initialize mock primitive. - - Args: - name: Name of the mock - return_value: Value to return (if no side_effect or error) - side_effect: Function to call instead of returning value - raise_error: Exception to raise when executed - """ - self.name = name - self.return_value = return_value - self.side_effect = side_effect - self.raise_error = raise_error - - self.call_count = 0 - self.calls: list[tuple[Any, WorkflowContext]] = [] - - async def execute(self, input_data: Any, context: WorkflowContext) -> Any: - """ - Execute mock primitive. - - Args: - input_data: Input data - context: Workflow context - - Returns: - Configured return value or side effect result - - Raises: - Exception: If configured to raise - """ - self.call_count += 1 - self.calls.append((input_data, context)) - - if self.raise_error: - raise self.raise_error - - if self.side_effect: - result = self.side_effect(input_data, context) - # Handle async side effects - if hasattr(result, "__await__"): - return await result - return result - - return self.return_value - - def assert_called(self) -> None: - """Assert the mock was called at least once.""" - assert self.call_count > 0, f"Mock {self.name} was not called" - - def assert_called_once(self) -> None: - """Assert the mock was called exactly once.""" - assert self.call_count == 1, f"Mock {self.name} called {self.call_count} times, expected 1" - - def assert_called_with(self, input_data: Any, context: WorkflowContext | None = None) -> None: - """ - Assert the mock was called with specific arguments. - - Args: - input_data: Expected input data - context: Optional expected context - """ - self.assert_called() - last_input, last_context = self.calls[-1] - - assert last_input == input_data, f"Expected input {input_data}, got {last_input}" - - if context is not None: - assert last_context == context, f"Expected context {context}, got {last_context}" - - def reset(self) -> None: - """Reset call tracking.""" - self.call_count = 0 - self.calls.clear() - - -class WorkflowTestCase: - """ - Test case helper for workflow testing. - - Example: - ```python - async def test_workflow(): - mock1 = MockPrimitive("step1", return_value={"data": "processed"}) - mock2 = MockPrimitive("step2", return_value={"data": "final"}) - - workflow = mock1 >> mock2 - - test_case = WorkflowTestCase(workflow) - result = await test_case.execute({"input": "test"}) - - test_case.assert_primitive_called(mock1, times=1) - test_case.assert_primitive_called(mock2, times=1) - assert result == {"data": "final"} - ``` - """ - - def __init__(self, workflow: WorkflowPrimitive) -> None: - """ - Initialize test case. - - Args: - workflow: Workflow to test - """ - self.workflow = workflow - self.mocks: list[MockPrimitive] = [] - - async def execute(self, input_data: Any, context: WorkflowContext | None = None) -> Any: - """ - Execute workflow with test context. - - Args: - input_data: Input data - context: Optional workflow context - - Returns: - Workflow result - """ - if context is None: - context = WorkflowContext() - - return await self.workflow.execute(input_data, context) - - def assert_primitive_called(self, mock: MockPrimitive, times: int | None = None) -> None: - """ - Assert a mock primitive was called. - - Args: - mock: Mock primitive to check - times: Optional expected call count - """ - if times is not None: - assert mock.call_count == times, ( - f"Expected {times} calls to {mock.name}, got {mock.call_count}" - ) - else: - assert mock.call_count > 0, f"Expected {mock.name} to be called" - - def reset_mocks(self) -> None: - """Reset all tracked mocks.""" - for mock in self.mocks: - mock.reset() diff --git a/packages/tta-workflow-primitives/tests/test_cache.py b/packages/tta-workflow-primitives/tests/test_cache.py deleted file mode 100644 index 942d9688..00000000 --- a/packages/tta-workflow-primitives/tests/test_cache.py +++ /dev/null @@ -1,236 +0,0 @@ -"""Tests for cache primitive.""" - -import time - -import pytest - -from tta_workflow_primitives.core.base import WorkflowContext -from tta_workflow_primitives.performance.cache import CachePrimitive -from tta_workflow_primitives.testing.mocks import MockPrimitive - - -@pytest.mark.asyncio -async def test_cache_hit() -> None: - """Test cache hit on second call.""" - mock = MockPrimitive("test", return_value={"result": "cached"}) - - cached = CachePrimitive( - primitive=mock, cache_key_fn=lambda data, ctx: data["key"], ttl_seconds=60.0 - ) - - # First call - cache miss - result1 = await cached.execute({"key": "test"}, WorkflowContext()) - assert result1 == {"result": "cached"} - assert mock.call_count == 1 - - # Second call - cache hit - result2 = await cached.execute({"key": "test"}, WorkflowContext()) - assert result2 == {"result": "cached"} - assert mock.call_count == 1 # Not called again - - -@pytest.mark.asyncio -async def test_cache_miss_different_keys() -> None: - """Test cache miss with different keys.""" - mock = MockPrimitive("test", return_value={"result": "value"}) - - cached = CachePrimitive( - primitive=mock, cache_key_fn=lambda data, ctx: data["key"], ttl_seconds=60.0 - ) - - await cached.execute({"key": "a"}, WorkflowContext()) - await cached.execute({"key": "b"}, WorkflowContext()) - - assert mock.call_count == 2 - - -@pytest.mark.asyncio -async def test_cache_expiration() -> None: - """Test cache expiration after TTL.""" - mock = MockPrimitive("test", return_value={"result": "value"}) - - cached = CachePrimitive( - primitive=mock, - cache_key_fn=lambda data, ctx: "key", - ttl_seconds=0.1, # Very short TTL - ) - - # First call - await cached.execute({}, WorkflowContext()) - assert mock.call_count == 1 - - # Wait for expiration - time.sleep(0.2) - - # Second call after expiration - await cached.execute({}, WorkflowContext()) - assert mock.call_count == 2 - - -@pytest.mark.asyncio -async def test_cache_clear() -> None: - """Test cache clearing.""" - mock = MockPrimitive("test", return_value={"result": "value"}) - - cached = CachePrimitive(primitive=mock, cache_key_fn=lambda data, ctx: "key", ttl_seconds=60.0) - - await cached.execute({}, WorkflowContext()) - assert cached.get_stats()["size"] == 1 - - cached.clear_cache() - assert cached.get_stats()["size"] == 0 - - -@pytest.mark.asyncio -async def test_cache_stats() -> None: - """Test cache statistics.""" - mock = MockPrimitive("test", return_value={"result": "value"}) - - cached = CachePrimitive( - primitive=mock, cache_key_fn=lambda data, ctx: data.get("key", "default"), ttl_seconds=60.0 - ) - - # Initial stats - stats = cached.get_stats() - assert stats["hits"] == 0 - assert stats["misses"] == 0 - assert stats["hit_rate"] == 0.0 - - # First call - miss - await cached.execute({"key": "a"}, WorkflowContext()) - stats = cached.get_stats() - assert stats["misses"] == 1 - assert stats["hit_rate"] == 0.0 - - # Second call same key - hit - await cached.execute({"key": "a"}, WorkflowContext()) - stats = cached.get_stats() - assert stats["hits"] == 1 - assert stats["hit_rate"] == 50.0 - - # Third call same key - hit - await cached.execute({"key": "a"}, WorkflowContext()) - stats = cached.get_stats() - assert stats["hits"] == 2 - assert stats["hit_rate"] == 66.67 - - -@pytest.mark.asyncio -async def test_cache_context_tracking() -> None: - """Test cache hit/miss tracking in context.""" - mock = MockPrimitive("test", return_value={"result": "value"}) - - cached = CachePrimitive(primitive=mock, cache_key_fn=lambda data, ctx: "key", ttl_seconds=60.0) - - context = WorkflowContext() - - # First call - miss - await cached.execute({}, context) - assert context.state["cache_misses"] == 1 - assert "cache_hits" not in context.state - - # Second call - hit - await cached.execute({}, context) - assert context.state["cache_hits"] == 1 - assert context.state["cache_misses"] == 1 - - -@pytest.mark.asyncio -async def test_cache_eviction() -> None: - """Test manual cache eviction.""" - mock = MockPrimitive("test", return_value={"result": "value"}) - - cached = CachePrimitive( - primitive=mock, cache_key_fn=lambda data, ctx: data["key"], ttl_seconds=0.1 - ) - - # Add some entries - await cached.execute({"key": "a"}, WorkflowContext()) - await cached.execute({"key": "b"}, WorkflowContext()) - - assert cached.get_stats()["size"] == 2 - - # Wait for expiration - time.sleep(0.2) - - # Manually evict - evicted = cached.evict_expired() - assert evicted == 2 - assert cached.get_stats()["size"] == 0 - - -@pytest.mark.asyncio -async def test_cache_realistic_llm_scenario() -> None: - """Test realistic LLM caching scenario.""" - call_count = 0 - - def llm_mock(name, response): - async def llm_call(data, ctx): - nonlocal call_count - call_count += 1 - return {"response": response, "call": call_count} - - from tta_workflow_primitives.core.base import LambdaPrimitive - - return LambdaPrimitive(llm_call) - - llm = llm_mock("llm", "Generated story") - - # Cache based on prompt + player - cached_llm = CachePrimitive( - primitive=llm, - cache_key_fn=lambda data, ctx: f"{data['prompt'][:50]}:{ctx.player_id}", - ttl_seconds=3600.0, - ) - - # Player 1, prompt 1 - ctx1 = WorkflowContext(player_id="player1") - result = await cached_llm.execute({"prompt": "Tell me a story"}, ctx1) - assert result["call"] == 1 - - # Same player, same prompt - cache hit - result = await cached_llm.execute({"prompt": "Tell me a story"}, ctx1) - assert result["call"] == 1 # Same call number - assert call_count == 1 # LLM not called again - - # Different player, same prompt - cache miss (different key) - ctx2 = WorkflowContext(player_id="player2") - result = await cached_llm.execute({"prompt": "Tell me a story"}, ctx2) - assert result["call"] == 2 - assert call_count == 2 - - # Same player, different prompt - cache miss - result = await cached_llm.execute({"prompt": "Different story"}, ctx1) - assert result["call"] == 3 - assert call_count == 3 - - # Check hit rate - stats = cached_llm.get_stats() - assert stats["hits"] == 1 - assert stats["misses"] == 3 - assert stats["hit_rate"] == 25.0 - - -@pytest.mark.asyncio -async def test_cache_key_generation() -> None: - """Test various cache key generation strategies.""" - mock = MockPrimitive("test", return_value={"result": "value"}) - - # Simple hash-based key - cached1 = CachePrimitive( - primitive=mock, cache_key_fn=lambda data, ctx: str(hash(str(data))), ttl_seconds=60.0 - ) - - # Composite key with context - cached2 = CachePrimitive( - primitive=mock, - cache_key_fn=lambda data, ctx: f"{data.get('type')}:{ctx.session_id}", - ttl_seconds=60.0, - ) - - # Test both - await cached1.execute({"x": 1}, WorkflowContext()) - await cached2.execute({"type": "story"}, WorkflowContext(session_id="s1")) - - assert cached1.get_stats()["size"] == 1 - assert cached2.get_stats()["size"] == 1 diff --git a/packages/tta-workflow-primitives/tests/test_composition.py b/packages/tta-workflow-primitives/tests/test_composition.py deleted file mode 100644 index 7c76b479..00000000 --- a/packages/tta-workflow-primitives/tests/test_composition.py +++ /dev/null @@ -1,133 +0,0 @@ -"""Tests for workflow primitive composition.""" - -from typing import Any - -import pytest - -from tta_workflow_primitives import ( - ConditionalPrimitive, - WorkflowContext, -) -from tta_workflow_primitives.core.base import LambdaPrimitive -from tta_workflow_primitives.testing import MockPrimitive - - -@pytest.mark.asyncio -async def test_sequential_composition() -> None: - """Test sequential primitive composition.""" - mock1 = MockPrimitive("step1", return_value="result1") - mock2 = MockPrimitive("step2", return_value="result2") - mock3 = MockPrimitive("step3", return_value="result3") - - workflow = mock1 >> mock2 >> mock3 - - context = WorkflowContext() - result = await workflow.execute("input", context) - - assert mock1.call_count == 1 - assert mock2.call_count == 1 - assert mock3.call_count == 1 - assert result == "result3" - - -@pytest.mark.asyncio -async def test_parallel_composition() -> None: - """Test parallel primitive composition.""" - mock1 = MockPrimitive("branch1", return_value="result1") - mock2 = MockPrimitive("branch2", return_value="result2") - mock3 = MockPrimitive("branch3", return_value="result3") - - workflow = mock1 | mock2 | mock3 - - context = WorkflowContext() - results = await workflow.execute("input", context) - - assert mock1.call_count == 1 - assert mock2.call_count == 1 - assert mock3.call_count == 1 - assert results == ["result1", "result2", "result3"] - - -@pytest.mark.asyncio -async def test_conditional_composition() -> None: - """Test conditional primitive composition.""" - then_mock = MockPrimitive("then", return_value="then_result") - else_mock = MockPrimitive("else", return_value="else_result") - - # Test then branch - workflow = ConditionalPrimitive( - condition=lambda x, ctx: x > 10, then_primitive=then_mock, else_primitive=else_mock - ) - - context = WorkflowContext() - result = await workflow.execute(15, context) - - assert then_mock.call_count == 1 - assert else_mock.call_count == 0 - assert result == "then_result" - - # Reset and test else branch - then_mock.reset() - else_mock.reset() - - result = await workflow.execute(5, context) - - assert then_mock.call_count == 0 - assert else_mock.call_count == 1 - assert result == "else_result" - - -@pytest.mark.asyncio -async def test_mixed_composition() -> None: - """Test mixed sequential and parallel composition.""" - step1 = MockPrimitive("step1", return_value="processed") - branch1 = MockPrimitive("branch1", return_value="b1") - branch2 = MockPrimitive("branch2", return_value="b2") - step2 = LambdaPrimitive(lambda x, ctx: f"final: {x}") - - workflow = step1 >> (branch1 | branch2) >> step2 - - context = WorkflowContext() - result = await workflow.execute("input", context) - - assert step1.call_count == 1 - assert branch1.call_count == 1 - assert branch2.call_count == 1 - assert result == "final: ['b1', 'b2']" - - -@pytest.mark.asyncio -async def test_lambda_primitive() -> None: - """Test lambda primitive.""" - - def transform(x: str, ctx: WorkflowContext) -> str: - return x.upper() - - workflow = LambdaPrimitive(transform) - - context = WorkflowContext() - result = await workflow.execute("hello", context) - - assert result == "HELLO" - - -@pytest.mark.asyncio -async def test_workflow_context() -> None: - """Test workflow context passing.""" - collected_contexts = [] - - def collect_context(x: Any, ctx: WorkflowContext) -> Any: - collected_contexts.append(ctx) - return x - - p1 = LambdaPrimitive(collect_context) - p2 = LambdaPrimitive(collect_context) - - workflow = p1 >> p2 - - context = WorkflowContext(workflow_id="test123", session_id="session456") - await workflow.execute("input", context) - - assert len(collected_contexts) == 2 - assert all(c.workflow_id == "test123" for c in collected_contexts) - assert all(c.session_id == "session456" for c in collected_contexts) diff --git a/packages/tta-workflow-primitives/tests/test_recovery.py b/packages/tta-workflow-primitives/tests/test_recovery.py deleted file mode 100644 index a695d601..00000000 --- a/packages/tta-workflow-primitives/tests/test_recovery.py +++ /dev/null @@ -1,116 +0,0 @@ -"""Tests for error recovery primitives.""" - -import pytest - -from tta_workflow_primitives import WorkflowContext -from tta_workflow_primitives.recovery import ( - FallbackPrimitive, - RetryPrimitive, - RetryStrategy, - SagaPrimitive, -) -from tta_workflow_primitives.testing import MockPrimitive - - -@pytest.mark.asyncio -async def test_retry_success_on_second_attempt() -> None: - """Test retry succeeds on second attempt.""" - call_count = 0 - - def flaky_operation(x, ctx) -> str: - nonlocal call_count - call_count += 1 - if call_count == 1: - raise ValueError("First attempt fails") - return "success" - - from tta_workflow_primitives.core.base import LambdaPrimitive - - flaky = LambdaPrimitive(flaky_operation) - workflow = RetryPrimitive(flaky, strategy=RetryStrategy(max_retries=3, backoff_base=0.01)) - - context = WorkflowContext() - result = await workflow.execute("input", context) - - assert call_count == 2 - assert result == "success" - - -@pytest.mark.asyncio -async def test_retry_exhaustion() -> None: - """Test retry exhaustion raises error.""" - mock = MockPrimitive("failing", raise_error=ValueError("Always fails")) - - workflow = RetryPrimitive(mock, strategy=RetryStrategy(max_retries=2, backoff_base=0.01)) - - context = WorkflowContext() - - with pytest.raises(ValueError, match="Always fails"): - await workflow.execute("input", context) - - assert mock.call_count == 3 # Initial + 2 retries - - -@pytest.mark.asyncio -async def test_fallback_on_failure() -> None: - """Test fallback activates on primary failure.""" - primary = MockPrimitive("primary", raise_error=ValueError("Primary fails")) - fallback = MockPrimitive("fallback", return_value="fallback_result") - - workflow = FallbackPrimitive(primary=primary, fallback=fallback) - - context = WorkflowContext() - result = await workflow.execute("input", context) - - assert primary.call_count == 1 - assert fallback.call_count == 1 - assert result == "fallback_result" - - -@pytest.mark.asyncio -async def test_fallback_not_used_on_success() -> None: - """Test fallback is not used when primary succeeds.""" - primary = MockPrimitive("primary", return_value="primary_result") - fallback = MockPrimitive("fallback", return_value="fallback_result") - - workflow = FallbackPrimitive(primary=primary, fallback=fallback) - - context = WorkflowContext() - result = await workflow.execute("input", context) - - assert primary.call_count == 1 - assert fallback.call_count == 0 - assert result == "primary_result" - - -@pytest.mark.asyncio -async def test_saga_compensation_on_failure() -> None: - """Test saga runs compensation on failure.""" - forward = MockPrimitive("forward", raise_error=ValueError("Forward fails")) - compensation = MockPrimitive("compensation", return_value=None) - - workflow = SagaPrimitive(forward=forward, compensation=compensation) - - context = WorkflowContext() - - with pytest.raises(ValueError, match="Forward fails"): - await workflow.execute("input", context) - - assert forward.call_count == 1 - assert compensation.call_count == 1 - - -@pytest.mark.asyncio -async def test_saga_no_compensation_on_success() -> None: - """Test saga does not run compensation on success.""" - forward = MockPrimitive("forward", return_value="success") - compensation = MockPrimitive("compensation", return_value=None) - - workflow = SagaPrimitive(forward=forward, compensation=compensation) - - context = WorkflowContext() - result = await workflow.execute("input", context) - - assert forward.call_count == 1 - assert compensation.call_count == 0 - assert result == "success" diff --git a/packages/tta-workflow-primitives/tests/test_routing.py b/packages/tta-workflow-primitives/tests/test_routing.py deleted file mode 100644 index 198094da..00000000 --- a/packages/tta-workflow-primitives/tests/test_routing.py +++ /dev/null @@ -1,143 +0,0 @@ -"""Tests for routing primitive.""" - -import pytest - -from tta_workflow_primitives.core.base import WorkflowContext -from tta_workflow_primitives.core.routing import RouterPrimitive -from tta_workflow_primitives.testing.mocks import MockPrimitive - - -@pytest.mark.asyncio -async def test_router_basic() -> None: - """Test basic routing.""" - route_a = MockPrimitive("a", return_value={"result": "A"}) - route_b = MockPrimitive("b", return_value={"result": "B"}) - - router = RouterPrimitive( - routes={"a": route_a, "b": route_b}, router_fn=lambda data, ctx: data["route"] - ) - - context = WorkflowContext() - result = await router.execute({"route": "a"}, context) - - assert result == {"result": "A"} - assert route_a.call_count == 1 - assert route_b.call_count == 0 - - -@pytest.mark.asyncio -async def test_router_context_based() -> None: - """Test routing based on context metadata.""" - openai = MockPrimitive("openai", return_value={"provider": "openai"}) - local = MockPrimitive("local", return_value={"provider": "local"}) - - router = RouterPrimitive( - routes={"openai": openai, "local": local}, - router_fn=lambda data, ctx: ctx.metadata.get("provider", "openai"), - ) - - # Route to local via context - context = WorkflowContext(metadata={"provider": "local"}) - result = await router.execute({}, context) - - assert result == {"provider": "local"} - assert local.call_count == 1 - assert openai.call_count == 0 - - -@pytest.mark.asyncio -async def test_router_default() -> None: - """Test default route fallback.""" - default = MockPrimitive("default", return_value={"result": "DEFAULT"}) - - router = RouterPrimitive( - routes={"a": default}, router_fn=lambda data, ctx: data.get("route", "unknown"), default="a" - ) - - context = WorkflowContext() - result = await router.execute({"route": "unknown"}, context) - - assert result == {"result": "DEFAULT"} - assert default.call_count == 1 - - -@pytest.mark.asyncio -async def test_router_no_route_error() -> None: - """Test error when no route found.""" - router = RouterPrimitive( - routes={"a": MockPrimitive("a", return_value={})}, router_fn=lambda data, ctx: "nonexistent" - ) - - with pytest.raises(ValueError, match="No route found"): - await router.execute({}, WorkflowContext()) - - -@pytest.mark.asyncio -async def test_router_tracks_history() -> None: - """Test routing history is tracked in context.""" - route_a = MockPrimitive("a", return_value={"result": "A"}) - route_b = MockPrimitive("b", return_value={"result": "B"}) - - router = RouterPrimitive( - routes={"a": route_a, "b": route_b}, router_fn=lambda data, ctx: data["route"] - ) - - context = WorkflowContext() - - # First routing - await router.execute({"route": "a"}, context) - assert context.state["routing_history"] == ["a"] - - # Second routing - await router.execute({"route": "b"}, context) - assert context.state["routing_history"] == ["a", "b"] - - -@pytest.mark.asyncio -async def test_router_cost_optimization() -> None: - """Test routing for cost optimization.""" - expensive = MockPrimitive("expensive", return_value={"cost": 10}) - cheap = MockPrimitive("cheap", return_value={"cost": 1}) - - def cost_router(data, ctx) -> str: - """Route simple queries to cheap model.""" - prompt_length = len(data.get("prompt", "")) - return "cheap" if prompt_length < 100 else "expensive" - - router = RouterPrimitive(routes={"expensive": expensive, "cheap": cheap}, router_fn=cost_router) - - context = WorkflowContext() - - # Short prompt -> cheap route - result = await router.execute({"prompt": "Hello"}, context) - assert result == {"cost": 1} - assert cheap.call_count == 1 - assert expensive.call_count == 0 - - # Long prompt -> expensive route - result = await router.execute({"prompt": "x" * 150}, context) - assert result == {"cost": 10} - assert expensive.call_count == 1 - - -@pytest.mark.asyncio -async def test_router_tier_based() -> None: - """Test routing based on user tier.""" - premium = MockPrimitive("premium", return_value={"tier": "premium"}) - free = MockPrimitive("free", return_value={"tier": "free"}) - - router = RouterPrimitive( - routes={"premium": premium, "free": free}, - router_fn=lambda data, ctx: ctx.metadata.get("tier", "free"), - default="free", - ) - - # Premium user - context = WorkflowContext(metadata={"tier": "premium"}) - result = await router.execute({}, context) - assert result == {"tier": "premium"} - - # Free user (default) - context = WorkflowContext() - result = await router.execute({}, context) - assert result == {"tier": "free"} diff --git a/packages/tta-workflow-primitives/tests/test_timeout.py b/packages/tta-workflow-primitives/tests/test_timeout.py deleted file mode 100644 index 3830608e..00000000 --- a/packages/tta-workflow-primitives/tests/test_timeout.py +++ /dev/null @@ -1,171 +0,0 @@ -"""Tests for timeout primitive.""" - -import asyncio - -import pytest - -from tta_workflow_primitives.core.base import LambdaPrimitive, WorkflowContext -from tta_workflow_primitives.recovery.timeout import TimeoutError, TimeoutPrimitive -from tta_workflow_primitives.testing.mocks import MockPrimitive - - -@pytest.mark.asyncio -async def test_timeout_success() -> None: - """Test successful execution within timeout.""" - fast = LambdaPrimitive(lambda data, ctx: {"result": "fast"}) - - timeout_prim = TimeoutPrimitive(primitive=fast, timeout_seconds=1.0) - - result = await timeout_prim.execute({}, WorkflowContext()) - assert result == {"result": "fast"} - - -@pytest.mark.asyncio -async def test_timeout_exceeded() -> None: - """Test timeout exceeded without fallback.""" - - async def slow(data, ctx): - await asyncio.sleep(2.0) - return {"result": "slow"} - - slow_prim = LambdaPrimitive(slow) - timeout_prim = TimeoutPrimitive(primitive=slow_prim, timeout_seconds=0.1) - - with pytest.raises(TimeoutError, match="exceeded 0.1s timeout"): - await timeout_prim.execute({}, WorkflowContext()) - - -@pytest.mark.asyncio -async def test_timeout_with_fallback() -> None: - """Test fallback on timeout.""" - - async def slow(data, ctx): - await asyncio.sleep(2.0) - return {"result": "slow"} - - slow_prim = LambdaPrimitive(slow) - fallback = MockPrimitive("fallback", return_value={"result": "fallback"}) - - timeout_prim = TimeoutPrimitive(primitive=slow_prim, timeout_seconds=0.1, fallback=fallback) - - result = await timeout_prim.execute({}, WorkflowContext()) - assert result == {"result": "fallback"} - assert fallback.call_count == 1 - - -@pytest.mark.asyncio -async def test_timeout_tracking() -> None: - """Test timeout tracking in context.""" - - async def slow(data, ctx): - await asyncio.sleep(2.0) - return {"result": "slow"} - - slow_prim = LambdaPrimitive(slow) - fallback = MockPrimitive("fallback", return_value={"result": "fallback"}) - - timeout_prim = TimeoutPrimitive( - primitive=slow_prim, timeout_seconds=0.1, fallback=fallback, track_timeouts=True - ) - - context = WorkflowContext() - await timeout_prim.execute({}, context) - - # Check tracking - assert context.state["timeout_count"] == 1 - assert len(context.state["timeout_history"]) == 1 - assert context.state["timeout_history"][0]["timeout"] == 0.1 - - -@pytest.mark.asyncio -async def test_timeout_multiple_calls() -> None: - """Test multiple timeout scenarios.""" - - async def sometimes_slow(data, ctx): - delay = data.get("delay", 0) - await asyncio.sleep(delay) - return {"result": f"delayed_{delay}s"} - - slow_prim = LambdaPrimitive(sometimes_slow) - fallback = MockPrimitive("fallback", return_value={"result": "fallback"}) - - timeout_prim = TimeoutPrimitive( - primitive=slow_prim, timeout_seconds=0.2, fallback=fallback, track_timeouts=True - ) - - context = WorkflowContext() - - # Fast call - no timeout - result = await timeout_prim.execute({"delay": 0.05}, context) - assert result == {"result": "delayed_0.05s"} - assert "timeout_count" not in context.state - - # Slow call - timeout - result = await timeout_prim.execute({"delay": 1.0}, context) - assert result == {"result": "fallback"} - assert context.state["timeout_count"] == 1 - - # Another slow call - result = await timeout_prim.execute({"delay": 1.0}, context) - assert context.state["timeout_count"] == 2 - - -@pytest.mark.asyncio -async def test_timeout_no_tracking() -> None: - """Test timeout without tracking.""" - - async def slow(data, ctx): - await asyncio.sleep(2.0) - return {"result": "slow"} - - slow_prim = LambdaPrimitive(slow) - fallback = MockPrimitive("fallback", return_value={"result": "fallback"}) - - timeout_prim = TimeoutPrimitive( - primitive=slow_prim, timeout_seconds=0.1, fallback=fallback, track_timeouts=False - ) - - context = WorkflowContext() - await timeout_prim.execute({}, context) - - # Should not track - assert "timeout_count" not in context.state - assert "timeout_history" not in context.state - - -@pytest.mark.asyncio -async def test_timeout_realistic_scenario() -> None: - """Test realistic LLM call with timeout.""" - call_count = 0 - - async def llm_call(data, ctx): - nonlocal call_count - call_count += 1 - # Simulate occasional slow response - if call_count == 2: - await asyncio.sleep(2.0) # Slow call - else: - await asyncio.sleep(0.1) # Normal call - return {"result": f"response_{call_count}"} - - llm_prim = LambdaPrimitive(llm_call) - cached_fallback = MockPrimitive("cache", return_value={"result": "cached"}) - - timeout_prim = TimeoutPrimitive( - primitive=llm_prim, timeout_seconds=0.5, fallback=cached_fallback - ) - - context = WorkflowContext() - - # First call succeeds - result = await timeout_prim.execute({}, context) - assert result == {"result": "response_1"} - - # Second call times out, uses fallback - result = await timeout_prim.execute({}, context) - assert result == {"result": "cached"} - assert cached_fallback.call_count == 1 - - # Third call succeeds - result = await timeout_prim.execute({}, context) - assert result == {"result": "response_3"}