TTA.dev Primitive Composition

TTA.dev workflows are built by composing typed primitives — each a subclass of WorkflowPrimitive[Input, Output]. All primitives share a single execute interface and compose with >> (sequential) or | (parallel).

When to Use This Skill

• Composing a new TTA.dev workflow from scratch
• Wrapping an existing function in retry, timeout, or cache logic
• Adding a circuit breaker to an external service call
• Building a parallel fan-out or sequential pipeline
• Passing state between steps with WorkflowContext

Prerequisites

uv sync            # installs ttadev and all extras

All primitives live in ttadev.primitives:

from ttadev.primitives import (
    WorkflowPrimitive, WorkflowContext, LambdaPrimitive,
    SequentialPrimitive, ParallelPrimitive,
    RetryPrimitive, TimeoutPrimitive, CachePrimitive,
    FallbackPrimitive,
)
from ttadev.primitives.recovery import CircuitBreakerPrimitive

Core Primitive Types

Primitive	Import path	When to use
LambdaPrimitive	ttadev.primitives	Wrap any function/async fn
SequentialPrimitive	ttadev.primitives	Run steps in order; output feeds next
ParallelPrimitive	ttadev.primitives	Fan-out — same input, results list
RetryPrimitive	ttadev.primitives	Transient failures, exponential back-off
TimeoutPrimitive	ttadev.primitives	Cap wall-clock time
CachePrimitive	ttadev.primitives	Memoize expensive calls
FallbackPrimitive	ttadev.primitives	Try primary, then secondary
CircuitBreakerPrimitive	ttadev.primitives.recovery	Stop hammering a broken service

The >> Composition Operator

>> creates a SequentialPrimitive inline. The output of the left primitive becomes the input of the right primitive.

pipeline = step_a >> step_b >> step_c
result = await pipeline.execute(initial_input, ctx)

| creates a ParallelPrimitive — both sides receive the same input and run concurrently, returning list[Output].

fan_out = classifier | summariser
results = await fan_out.execute(text, ctx)   # [class_result, summary_result]

WorkflowContext — Passing State

WorkflowContext flows through every .execute() call. Use ctx.state to share values across steps without changing function signatures.

import asyncio
from ttadev.primitives import WorkflowContext, LambdaPrimitive

# Root context — starts a fresh workflow
ctx = WorkflowContext.root("my-pipeline")

# Write state in one step …
store = LambdaPrimitive(lambda data, c: (c.state.update({"key": data}), data)[1])

# … read it in another
read  = LambdaPrimitive(lambda _, c: c.state["key"].upper())

pipeline = store >> read
result = asyncio.run(pipeline.execute("hello", ctx))  # "HELLO"

Child contexts inherit state while keeping a new span ID:

child_ctx = WorkflowContext.child(ctx, step_name="validate-input")

Minimal Working Example

import asyncio
from ttadev.primitives import (
    LambdaPrimitive, RetryPrimitive, TimeoutPrimitive, WorkflowContext,
)

async def fetch(url: str, ctx: WorkflowContext) -> str:
    # … real HTTP call here
    return f"data from {url}"

fetcher = LambdaPrimitive(fetch)
resilient = TimeoutPrimitive(
    RetryPrimitive(fetcher, max_retries=3),
    timeout_seconds=10.0,
)

ctx = WorkflowContext.root("fetch-pipeline")
result = asyncio.run(resilient.execute("https://api.example.com/data", ctx))
print(result)

Recovery Primitives — Quick Reference

from ttadev.primitives import RetryPrimitive, TimeoutPrimitive, FallbackPrimitive
from ttadev.primitives.recovery import CircuitBreakerPrimitive, CircuitBreakerConfig

# Retry with exponential back-off
safe = RetryPrimitive(my_prim, max_retries=3, backoff_base=2.0)

# Hard wall-clock timeout
timed = TimeoutPrimitive(my_prim, timeout_seconds=5.0)

# Primary → fallback chain
resilient = FallbackPrimitive(primary_prim, fallback_prim)

# Circuit breaker — opens after 5 failures, retries after 60 s
cb = CircuitBreakerPrimitive(
    my_prim,
    config=CircuitBreakerConfig(failure_threshold=5, recovery_timeout=60.0),
)

Anti-Patterns

❌ Never do this	✅ Use this instead
for attempt in range(3): try/except	RetryPrimitive(prim, max_retries=3)
asyncio.wait_for(coro(), 10) manually	TimeoutPrimitive(prim, timeout_seconds=10)
Global dict for workflow state	ctx.state["key"] = value
from primitives.X import ...	from ttadev.primitives import ...
Hard-code retry counts in business logic	Wrap with RetryPrimitive

Running a Workflow End-to-End

import asyncio
from ttadev.primitives import (
    LambdaPrimitive, RetryPrimitive, CachePrimitive, WorkflowContext,
)

# 1. Define steps
validate = LambdaPrimitive(lambda x, _: x.strip() if x else ValueError("empty"))
enrich   = LambdaPrimitive(lambda x, _: {"text": x, "length": len(x)})

# 2. Compose
pipeline = (
    RetryPrimitive(validate, max_retries=2)
    >> enrich
)

# 3. Execute
ctx    = WorkflowContext.root("demo")
result = asyncio.run(pipeline.execute("  hello world  ", ctx))
print(result)  # {"text": "hello world", "length": 11}

Deep Reference

For parallel fan-out patterns, circuit breaker tuning, sequential data-passing, and FallbackPrimitive composition chains, see references/composition-patterns.md.