Durable workflows. Durable agents.

# A real agent loop. Multi-turn conversation that survives restarts.
# Pattern from examples/ai/conversational_agent_anthropic.py

from flux import task, workflow, ExecutionContext
from flux.tasks import pause
from anthropic import AsyncAnthropic

@task.with_options(
    secret_requests=["ANTHROPIC_API_KEY"],
    retry_max_attempts=3,
    retry_backoff=2,
    timeout=60,
)
async def conversation_turn(messages, user_message, system_prompt, secrets={}):
    """One turn: append user message, call Claude, append response."""
    messages.append({"role": "user", "content": user_message})

    client = AsyncAnthropic(api_key=secrets["ANTHROPIC_API_KEY"])
    response = await client.messages.create(
        model="claude-sonnet-4-5-20250929",
        max_tokens=1024,
        system=system_prompt,
        messages=messages,
    )

    reply = response.content[0].text
    messages.append({"role": "assistant", "content": reply})
    return messages, reply

@workflow
async def conversational_agent(ctx: ExecutionContext[dict]):
    initial = ctx.input or {}
    system_prompt = initial.get("system_prompt", "You are a helpful assistant.")
    max_turns = initial.get("max_turns", 10)

    messages = []
    messages, _ = await conversation_turn(messages, initial["message"], system_prompt)

    # The loop: pause for the next user message, run the turn, repeat.
    # Each pause is a durable checkpoint — resume tomorrow if you want.
    for turn in range(1, max_turns):
        resume_input = await pause(f"awaiting_user_input_turn_{turn}")
        next_message = (resume_input or {}).get("message")
        if not next_message:
            break
        messages, _ = await conversation_turn(messages, next_message, system_prompt)

    return {"history": messages, "turns": len(messages) // 2}

# Any @task becomes a tool. The agent decides when to call it.
# Tool calls are checkpointed — debug-replay them deterministically.

from flux import task, workflow, ExecutionContext
from flux.tasks.ai import agent
from pydantic import BaseModel

@task
async def get_weather(city: str) -> dict:
    """Fetch current weather for a city."""
    return await open_meteo.fetch(city)

@task
async def get_forecast(city: str, days: int = 3) -> list:
    """Multi-day forecast for a city."""
    return await open_meteo.forecast(city, days)

class Forecast(BaseModel):
    summary: str
    high_c: float
    advice: str

@workflow
async def weather_briefing(ctx: ExecutionContext[str]):
    briefer = await agent(
        system_prompt="Brief the user on weather and travel conditions.",
        model="anthropic/claude-sonnet-4-5",
        tools=[get_weather, get_forecast],
        response_format=Forecast,    # typed output via Pydantic
    )
    return await briefer(ctx.input)

# Working memory + long-term memory. Plus dreaming —
# a background workflow that consolidates what mattered.

from flux import workflow, ExecutionContext
from flux.tasks.ai import agent, system_tools
from flux.tasks.ai.memory import working_memory, long_term_memory, sqlite
from flux.tasks.ai.dreaming import dream

@workflow
async def research_assistant(ctx: ExecutionContext[dict]):
    assistant = await agent(
        system_prompt="Research, store findings, recall when useful.",
        model="anthropic/claude-sonnet-4-5",
        tools=[system_tools(workspace=".")],
        working_memory=working_memory(),
        long_term_memory=long_term_memory(
            provider=sqlite("memory.db"),
            agent="researcher",
            scope="user:alice",
        ),
        on_complete=[dream],   # consolidate memory in the background
    )
    return await assistant(ctx.input["message"])

# Run tasks concurrently. Flux gathers, checkpoints, returns.

from flux import task, workflow, ExecutionContext
from flux.tasks import parallel

@task
async def say_hi(name): return f"Hi, {name}"

@task
async def say_hello(name): return f"Hello, {name}"

@task
async def say_hola(name): return f"Hola, {name}"

@workflow
async def greetings(ctx: ExecutionContext[str]):
    return await parallel(
        say_hi(ctx.input),
        say_hello(ctx.input),
        say_hola(ctx.input),
    )

# Chain tasks into a pipeline. Output flows in, transformed out.

from flux import task, workflow, ExecutionContext
from flux.tasks import pipeline

@task
async def multiply_by_two(x): return x * 2

@task
async def add_three(x): return x + 3

@task
async def square(x): return x * x

@workflow
async def transform(ctx: ExecutionContext[int]):
    return await pipeline(
        multiply_by_two,
        add_three,
        square,
        input=ctx.input,
    )

# Retries, fallbacks, rollbacks, timeouts — all declarative.

from flux import task, workflow, ExecutionContext

async def cleanup(): ...
async def use_cache(): return "stale-but-safe"

@task.with_options(
    retry_max_attempts=3,
    retry_delay=1,
    retry_backoff=2,
    timeout=30,
    fallback=use_cache,
    rollback=cleanup,
    secret_requests=['API_KEY'],
    cache=True,
)
async def charge_customer(amount, secrets: dict = {}):
    return await stripe.charge(amount, key=secrets['API_KEY'])

# Auto-schedule on registration. Cron, intervals, or one-time.

from flux import workflow, task, cron, ExecutionContext

@task
async def generate_report(date: str):
    return f"Report for {date}"

@workflow.with_options(
    name="daily_report",
    schedule=cron("0 9 * * MON-FRI", timezone="UTC"),
)
async def daily_report(ctx: ExecutionContext):
    report = await generate_report("today")
    return {"report": report}

# $ flux workflow register reports.py
# $ flux schedule list  →  daily_report_auto · 0 9 * * MON-FRI

# When linear isn't enough: define workflows as DAGs.

from flux import task, workflow, ExecutionContext
from flux.tasks import Graph

@task
async def fetch(id): ...

@task
async def parse(data): ...

@task
async def enrich(parsed): ...

@task
async def commit(enriched): ...

@workflow
async def ingest(ctx: ExecutionContext[str]):
    g = (Graph("ingest")
         .add_node("fetch", fetch)
         .add_node("parse", parse)
         .add_node("enrich", enrich)
         .add_node("commit", commit)
         .add_edge("fetch", "parse")
         .add_edge("parse", "enrich")
         .add_edge("enrich", "commit")
         .start_with("fetch")
         .end_with("commit"))
    return await g(ctx.input)