Cyclium

Autonomous agent framework for Elixir — actors monitor domains, run multi-turn episodes with budget enforcement, and produce persistent findings and typed outputs.

Cyclium is an Elixir library for building agentic systems that monitor domains, run multi-turn episodes, classify situations, and produce typed outputs. Actors declare expectations about how things should be; when triggers fire, episodes execute strategies that can gather data, call tools, synthesize with LLMs, and converge into findings and outputs. Think of it as an OTP-native agent framework where the episode — not the request — is the unit of work.

Key features

• Declarative Actor DSL — Define actors, expectations, triggers, and budgets in a compact macro-based syntax
• Strategy Pattern — Pluggable investigation logic with a clear init → observe → converge lifecycle
• Episode Runner — Budget-enforced execution loop with step journaling, checkpointing, and crash recovery
• Findings Lifecycle — Persistent observations with raise/update/clear semantics, upsert-by-key, causality chains, TTL expiration, severity escalation, and post-raise enrichment hooks
• Output Router — Deduplicated, adapter-based delivery (email, Slack, webhooks) with approval gates and adapter registry
• Event Bus — Phoenix.PubSub-backed event system connecting actors without coupling
• Workflow Engine — Multi-actor coordination with dependency graphs, failure policies, retry with backoff, cross-workflow episode dedup, and cancellation cascade
• Circuit Breaker — Per-expectation circuit breaker with configurable thresholds, half-open recovery, and optional in-flight episode cancellation
• Episode Sampling — Probabilistic firing control via sample_rate on expectations
• Service Level Tracking — Declarative performance objectives with breach detection and telemetry
• Adaptive Budgets — Advisory budget recommendations based on historical episode usage (p95 with headroom)
• Backpressure Controls — Per-actor concurrency limits with queue, drop, or shed-oldest overflow policies
• Debounce and Cooldown — Temporal controls to coalesce rapid-fire events and enforce minimum gaps
• Log Projection — Materialized human-readable logs at configurable verbosity (none → full_debug)
• Telemetry — 36 structured telemetry events for observability
• OTP-Native — No Oban or external job queue required; episodes run as Tasks under DynamicSupervisor
• Test Kit — Assertion macros and fakes for validating actors, strategies, synthesizers, output adapters, workflows, and checkpoint migrations in host apps
• SQL Server 2017 Compatible — Transaction-based upserts, denormalized query columns, no JSON operators in DDL

Who is this for?

Cyclium is designed for Elixir teams building autonomous agent systems where:

• Business rules define what should be true (SLAs, health thresholds, compliance checks)
• Episodes involve multiple steps: data gathering, LLM synthesis, tool calls, human approval
• Findings need to persist and evolve over time (raised → updated → cleared)
• Actions need deduplication, audit trails, and typed delivery through adapters
• Multiple actors need to coordinate through workflows with dependency ordering
• Real-time visibility into agent state is essential (Phoenix LiveView integration via Bus)

If you need a simple cron job or a one-shot script, Cyclium is overkill. Cyclium shines when you have ongoing, stateful processes that produce findings and outputs — and need the lifecycle, audit trail, and coordination to go with them.

How Cyclium differs

vs. Oban — Oban is a job queue: enqueue work, run it, done. Cyclium is an agent framework that manages stateful, multi-turn episodes with budgets, findings, outputs, and workflows. Episodes happen to run as OTP Tasks, so you don't need Oban — but the two solve different problems. You could use both: Oban for fire-and-forget jobs, Cyclium for ongoing autonomous processes.

vs. Sagents — Sagents is built for interactive AI conversations where users chat with LLM-powered agents in real time. Cyclium is built for autonomous operational agents that monitor domains, classify situations, and act — with or without an LLM in the loop. Cyclium's strategies can call LLMs via :synthesize, but they can also run purely deterministic logic. The execution model (expectations → episodes → findings → outputs) is designed for operational workflows, not chat.

vs. GenServers / custom OTP — You could build all of this with raw GenServers, but Cyclium gives you the episode lifecycle (budgets, journaling, checkpoints, crash recovery), the findings system (upsert-by-key, severity, evidence), the output router (deduplication, adapters, approval gates), the workflow engine (dependency graphs, failure policies), and the event bus — all wired together with telemetry and audit trails.

Strategy-driven vs. LLM-routed

Most agent frameworks put the LLM in the driver's seat — it decides which tool to call, when to stop, and how to recover from errors. Cyclium inverts this. The developer is the router: your next_step/2 function is a deterministic state machine that decides what happens next. The LLM is a powerful tool you call at specific points via :synthesize, but it never controls the flow.

This means you can mix deterministic and AI-powered steps in a single episode — gather data with a tool call, classify it with an LLM, then act on the result with another tool call — all under explicit developer control with budget enforcement at every turn:

def next_step(%{phase: :gather} = state, _ctx) do
  {:tool_call, :erp_read, :search_pos, %{status: "STALLED"}}
end
def next_step(%{phase: :classify, po_data: data} = state, _ctx) do
  {:synthesize, %{task: :classify_po, data: data}}
end
def next_step(%{phase: :act, classification: "vendor_delay"} = state, _ctx) do
  {:tool_call, :email_write, :send_followup, build_email(state)}
end

The LLM is powerful, but it's not the control plane. You get repeatability, testability, and full visibility into exactly which steps ran and why — without sacrificing the ability to use AI where it adds value.

Architecture

The examples throughout this README use a client health monitoring system: a ClientHealthActor that evaluates client metrics (MRR, active users, support tickets) on each change and classifies health status, plus a ClientAdvisorActor that synthesizes an LLM-powered summary. See the demo application for the full working implementation.

Supervision tree

YourApp.Supervisor
├── YourApp.Repo
├── Phoenix.PubSub
├── YourApp.Actors.ClientHealthActor (GenServer)
├── YourApp.Actors.ClientAdvisorActor (GenServer)
├── Cyclium.Supervisor
│   ├── Cyclium.ActorSupervisor (DynamicSupervisor)
│   ├── Cyclium.EpisodeSupervisor (DynamicSupervisor)
│   │   └── Cyclium.EpisodeTask (one per running episode)
│   ├── Cyclium.TaskSupervisor (Task.Supervisor)
│   ├── Cyclium.Reconciler (optional — spec change detection)
│   ├── Cyclium.WorkflowEngine (optional — multi-actor workflows)
│   └── Cyclium.Findings.FindingSweep (optional — expiration + escalation)
└── YourAppWeb.Endpoint

Actors are GenServers that subscribe to Bus events and manage episode lifecycle. They're started by the consuming app's supervision tree, above Cyclium's supervisors. When a trigger fires, the actor creates an Episode row and starts an EpisodeTask under the EpisodeSupervisor. Each task resolves a strategy from the registry and runs the episode loop.

Key design principles:

• Actors own concurrency limits — they track active/queued episodes in-process
• Episodes are durable — the cyclium_episodes table is itself a work queue
• The Bus connects everything — actors, LiveViews, and workflows all subscribe to the same event stream
• Strategies are stateless modules — all state lives in the episode's strategy state map

Execution model

Bus event arrives
  → Actor.handle_info matches expectation trigger
  → Check debounce/cooldown → circuit breaker → sample_rate
  → Check concurrency (active < max?)
    → yes: create Episode row, start EpisodeTask under DynamicSupervisor
    → no:  apply overflow policy (queue / drop / shed_oldest)

EpisodeTask starts
  → Resolve strategy (persistent_term registration from actor boot → registry override)
  → strategy.init(episode, trigger)
  → EpisodeRunner.execute_loop:

    ┌─────────────────────────────────────────────┐
    │  check_budget → check_loop → increment_turn │
    │  strategy.next_step(state, ctx)             │
    │    :done         → journal, set done        │
    │    :converge     → run converge pipeline    │
    │    {:tool_call}  → exec tool, handle_result │
    │    {:observe}    → journal, handle_result   │
    │    {:synthesize} → journal, handle_result   │
    │    {:checkpoint} → save state, loop         │
    │    {:approval}   → block, wait for human    │
    │    {:wait}       → block, wait for external │
    │    ...           → loop                     │
    └─────────────────────────────────────────────┘

  Converge pipeline (post_converge):
    1. Persist findings (raise/update/clear) → enrich → Bus events per finding
    2. Route outputs through adapters → dedup by dedupe_key, deliver
    3. Compute final episode status from delivery outcomes
    4. Journal completion/failure step
    5. Project log via LogProjector
    6. Record service levels + check for breach
    7. Record adaptive budget sample (if enabled)
    8. Broadcast episode.completed/failed on Bus
    9. Emit telemetry

Core concepts

Actors

An actor is a GenServer that owns one or more expectations. Each actor watches a domain (e.g., :procurement, :client_health) and fires episodes when triggers match.

defmodule MyApp.Actors.ClientHealthActor do
  use Cyclium.Actor

  actor do
    domain(:client_health)
    spec_rev("v0.1.0")
    max_concurrent_episodes(5)
    episode_overflow(:queue)

    expectation(:client_should_be_healthy,
      strategy: MyApp.Strategies.ClientHealth,
      trigger: {:event, "client.health_check_requested"},
      subject_key: :client_id,
      debounce_ms: :timer.seconds(3),
      budget: %{max_turns: 3, max_tokens: 1_000, max_wall_ms: 10_000}
    )

    expectation(:contract_review,
      strategy: MyApp.Strategies.ContractReview,
      trigger: {:schedule, :timer.hours(24)},
      recovery_policy: :restart,
      budget: %{max_turns: 12, max_tokens: 25_000, max_wall_ms: 120_000}
    )
  end
end

Trigger types:

• {:event, "event.name"} — fires when a matching Bus event arrives
• {:schedule, interval_ms} — fires on a recurring timer
• :manual — fires on explicit request
• :workflow — fires as part of a multi-actor workflow
• List — combine multiple triggers: [{:event, "client.health_check_requested"}, :workflow]

List triggers allow an expectation to fire from multiple sources. Event subscriptions and schedule timers are extracted from the list automatically. This is the recommended pattern when an expectation participates in a workflow but should also be independently triggerable:

expectation(:client_should_be_healthy,
  strategy: MyApp.Strategies.ClientHealth,
  trigger: [{:event, "client.health_check_requested"}, :workflow],
  subject_key: :client_id,
  debounce_ms: :timer.seconds(3),
  budget: %{max_turns: 3, max_tokens: 1_000, max_wall_ms: 10_000}
)

The actor subscribes to "client.health_check_requested" for standalone use (with debounce), while the :workflow marker documents that workflows can also invoke this expectation. Workflow-triggered episodes bypass the actor GenServer entirely — the workflow engine creates episodes directly — so actor-level debounce/cooldown only applies to event triggers.

Backpressure options (episode_overflow):

• :queue — buffer excess episodes (default)
• :drop — discard when at capacity
• :shed_oldest — cancel the oldest queued episode to make room

Expectation options:

Actor ID convention: Actor IDs are atoms in-process and strings in the database. Use identifier/1 in the DSL to set a stable actor ID that survives module renames:

actor do
  identifier(:client_health)   # explicit, rename-proof
  domain(:health)
  # ...
end

If omitted, the ID is derived from the module name (MyApp.Actors.ClientHealthActor → :client_health_actor). The boundary is at episode creation — Cyclium.Actor.Server calls to_string(state.actor_id) when building the episode params. Everything upstream is atoms, everything downstream (DB, strategies, findings) is strings. Actors with persisted episodes should always declare an explicit identifier.

Strategies

A strategy implements the investigation logic for an expectation. It's the brain of an episode — a stateless module that receives state and returns actions.

defmodule MyApp.Strategies.ClientHealth do
  @behaviour Cyclium.EpisodeRunner.Strategy

  @impl true
  def init(_episode, trigger) do
    client_id = trigger.payload["client_id"]
    {:ok, %{client_id: client_id}}
  end

  @impl true
  def next_step(state, _episode_ctx) do
    :converge  # go straight to classification
  end

  @impl true
  def handle_result(state, _step, _result) do
    {:ok, state}
  end

  @impl true
  def converge(state, _episode_ctx) do
    client = MyApp.Clients.get!(state.client_id)
    {class, severity, summary} = classify(client)

    {:ok, %Cyclium.ConvergeResult{
      classification: %{"primary" => class, "severity" => to_string(severity)},
      confidence: 1.0,
      summary: summary,
      findings: [
        {:raise, %{
          actor_id: "client_health_actor",
          finding_key: "client:health:#{client.id}",
          class: class,
          severity: severity,
          confidence: 1.0,
          subject: %{kind: "client", id: client.id},
          subject_kind: "client",
          subject_id: client.id,
          summary: summary,
          evidence_refs: %{"active_users" => client.active_users}
        }}
      ],
      outputs: []
    }}
  end
end

Strategy callbacks:

next_step return values:

handle_result step kinds:

The step argument passed to handle_result/3 is an %EpisodeStep{} struct. Pattern-match on kind to distinguish which action produced the result:

The step struct also carries tool_name (for :tool_call steps) which is useful for per-tool retry tracking:

def handle_result(state, %{kind: :synthesis}, {:ok, result}), do: ...
def handle_result(state, %{kind: :tool_call, tool_name: "erp_read"}, {:ok, data}), do: ...
def handle_result(state, %{kind: :observation}, {:ok, data}), do: ...

Multi-turn strategies

Strategies can run multiple turns before converging. next_step decides actions, handle_result absorbs outcomes — expensive work like LLM calls should be delegated to actions (:synthesize, :tool_call), not done inside handle_result.

Use a :phase field in state to drive progression. This is the recommended pattern — it makes flow explicit, prevents ambiguous matching in handle_result, and avoids accidental loops:

defmodule MyApp.Strategies.ClientAdvisor do
  @behaviour Cyclium.EpisodeRunner.Strategy

  @system_prompt "You are a customer success analyst. Assess the client's health."

  @impl true
  def init(_episode, trigger) do
    # Load data in init — next_step should be pure routing, not queries
    client_id = trigger.payload["client_id"]
    client = MyApp.Clients.get!(client_id)
    client_data = Map.take(client, [:name, :status, :mrr])
    {:ok, %{phase: :synthesize, client_id: client_id, client_data: client_data, ai_summary: nil}}
  end

  # --- next_step: pure routing based on phase ---

  @impl true
  def next_step(%{phase: :synthesize} = state, _episode_ctx) do
    {:synthesize, %{
      system_prompt: @system_prompt,
      user_message: "Client: #{state.client_data.name}, MRR: $#{state.client_data.mrr}"
    }}
  end

  def next_step(%{phase: :done}, _episode_ctx), do: :converge

  # --- handle_result: ALWAYS guard on phase ---

  @impl true
  def handle_result(%{phase: :synthesize} = state, _step, {:ok, result}) do
    summary = if is_map(result), do: result[:text] || result["text"] || inspect(result), else: inspect(result)
    {:ok, %{state | phase: :done, ai_summary: summary}}
  end

  def handle_result(%{phase: :synthesize} = state, %{kind: :synthesis}, {:error, {_class, _detail}}) do
    # Transient failure — retry the same step (next_step will re-emit :synthesize)
    {:retry, state}
  end

  def handle_result(_state, _step, {:error, reason}) do
    {:abort, reason}
  end

  @impl true
  def converge(state, _episode_ctx) do
    {:ok, %Cyclium.ConvergeResult{
      classification: %{"primary" => "ai_summary", "severity" => "low"},
      confidence: 0.9,
      summary: state.ai_summary,
      findings: [
        {:raise, %{
          actor_id: "client_advisor_actor",
          finding_key: "client:advisor:#{state.client_id}",
          class: "ai_summary",
          severity: :low,
          confidence: 0.9,
          subject_kind: "client",
          subject_id: state.client_id,
          summary: state.ai_summary
        }}
      ],
      outputs: []
    }}
  end
end

Key points:

• next_step is pure decision-making — it returns what action to take, never does expensive work itself
• handle_result absorbs outcomes — it pattern-matches on step kind and result, then updates state
• {:retry, state} re-enters the loop with the same state, letting next_step retry the action
• {:abort, reason} immediately fails the episode with the given reason
• The :synthesize action delegates LLM calls to the app-provided Cyclium.Synthesizer, keeping the strategy free of HTTP concerns

Multi-step patterns and pitfalls

Always guard handle_result on :phase. The most common multi-step bug is an unguarded handle_result clause that matches too broadly, causing the strategy to cycle between phases instead of progressing:

# BAD — matches ANY success result during any phase
def handle_result(state, _step, {:ok, result}) do
  {:ok, %{state | phase: :done, ai_summary: inspect(result)}}
end

# GOOD — each clause is scoped to its phase
def handle_result(%{phase: :synthesize} = state, _step, {:ok, result}) do
  {:ok, %{state | phase: :done, ai_summary: extract_text(result)}}
end

def handle_result(%{phase: :tool_result} = state, _step, {:ok, data}) do
  {:ok, %{state | phase: :synthesize, gathered: data}}
end

Without phase guards, handle_result matches the wrong clause → phase doesn't advance → next_step re-emits the same action → loop. The budget will eventually kill it, but you'll burn turns for no reason.

Handle the no-synthesizer passthrough. When no Cyclium.Synthesizer is configured (or the registry returns nil for your actor), the runner passes prompt_ctx through as-is to handle_result with {:ok, prompt_ctx}. Your :synthesize phase handler must handle both the LLM response shape AND the raw passthrough:

def handle_result(%{phase: :synthesize} = state, _step, {:ok, result}) do
  summary =
    cond do
      is_binary(result) -> result
      is_map(result) && Map.has_key?(result, :text) -> result.text
      is_map(result) && Map.has_key?(result, "text") -> result["text"]
      true -> inspect(result)  # passthrough — no synthesizer configured
    end

  {:ok, %{state | phase: :done, ai_summary: summary}}
end

Handle both trigger types if your actor can be triggered by workflows. Workflow-created episodes use %Cyclium.Trigger.Workflow{input: ...}, not %Cyclium.Trigger.Event{payload: ...}:

def init(_episode, trigger) do
  client_id =
    case trigger do
      %Cyclium.Trigger.Event{payload: %{client_id: id}} -> id
      %Cyclium.Trigger.Event{payload: payload} -> payload["client_id"]
      %Cyclium.Trigger.Workflow{input: %{client_id: id}} -> id
      %Cyclium.Trigger.Workflow{input: input} when is_map(input) -> input["client_id"]
      _ -> nil
    end

  {:ok, %{phase: :gather, client_id: client_id}}
end

Loop detection

The episode runner automatically detects repeating step cycles. It fingerprints each next_step return value and watches for repeated patterns — a 1-step cycle (A, A, A), a 2-step cycle (A, B, A, B), or longer. When a cycle is detected, the episode fails immediately with error_class: "loop_detected".

This is a safety net, not a substitute for correct phase guards. If loop detection fires, it means your strategy has a bug — fix the root cause (usually a missing phase guard on handle_result).

Note that consecutive steps of the same kind but different data are fine — the fingerprint includes the full action payload via :erlang.phash2/1. A dispatch strategy that emits multiple :observe steps with different entity data won't trigger loop detection. Only identical actions repeated in a cycle will.

Episodes

An episode is one execution of a strategy. It tracks:

• Budget usage (turns, tokens, wall time)
• Step journal (every action recorded as an EpisodeStep)
• Classification and summary (set during converge)
• Status lifecycle: :running → :done | :failed | :blocked | :canceled | :partially_failed

Episodes run as Tasks under a DynamicSupervisor — no Oban required. The cyclium_episodes table serves as a durable work queue.

Querying episodes:

Cyclium.Episodes.get!(episode_id)
Cyclium.Episodes.list_by_status([:running, :done, :failed])
Cyclium.Episodes.list_steps(episode_id)   # step journal
Cyclium.Episodes.get_log(episode_id)      # materialized log
Cyclium.Episodes.cancel(episode_id)       # cancellation sequence

# List episodes by actor(s)
Cyclium.Episodes.list_by_actors(["my_actor"], limit: 20, order: :desc)

# Filter by subject — DB-level JSON filtering across trigger types
# Checks trigger_ref.payload.<key> (event-triggered) and
# trigger_ref.input.<key> (workflow-triggered) in a single query.
Cyclium.Episodes.list_by_actors_and_subject(
  ["client_health_actor", "client_advisor_actor"],
  :client_id,
  client.id,
  limit: 20, order: :desc
)

list_by_actors_and_subject/4 detects the repo adapter at runtime and uses the appropriate JSON text extraction — Postgres #>> or SQL Server JSON_VALUE. This is the recommended way to fetch episodes for a specific entity — avoids pulling all episodes and filtering in memory.

Findings

A finding is a persistent observation about an entity. Findings have a lifecycle:

• Raise — create or update an active finding (upsert by finding_key)
• Update — modify mutable fields on an active finding
• Clear — mark a finding as resolved (idempotent)

# In your converge/2 callback:
findings: [
  {:raise, %{
    actor_id: "client_health_actor",
    finding_key: "client:health:123",
    class: "churned",
    severity: :high,          # :low | :medium | :high | :critical
    confidence: 1.0,
    subject: %{kind: "client", id: "123"},
    subject_kind: "client",   # denormalized for SQL Server compat
    subject_id: "123",
    summary: "Client has churned",
    evidence_refs: %{"status" => "churned"}
  }},
  {:update, "client:health:123", %{confidence: 0.8}},
  {:clear, "client:health:123"},
  {:clear, "client:health:123", "customer reactivated"}
]

Finding key scoping — deduplicated vs. distinct:

The finding_key controls deduplication. An active finding with the same key is updated in place (last-writer-wins on mutable fields). Choose your key strategy based on intent:

• Deduplicated (default pattern): Use a stable key like "client:health:123". Repeated episodes update the same active finding — ideal for ongoing status tracking where you want one finding per subject.
• Distinct per episode: Include the episode ID in the key, e.g. "po_review:PO-1955:#{episode.id}". Each episode creates a separate finding — useful for audit trails or point-in-time snapshots where every run should produce its own record.

The episode_ctx map passed to converge/2 contains episode_id, so you can reference it directly:

# Deduplicated: one active finding per client, updated each run
def converge(state, _episode_ctx) do
  {:ok, %Cyclium.ConvergeResult{
    findings: [{:raise, %{finding_key: "client:health:#{state.client_id}", ...}}]
  }}
end

# Distinct: one finding per episode run
def converge(state, episode_ctx) do
  {:ok, %Cyclium.ConvergeResult{
    findings: [{:raise, %{finding_key: "po_review:#{state.po_id}:#{episode_ctx.episode_id}", ...}}]
  }}
end

Findings are queried via Cyclium.Findings.active_for/1:

Cyclium.Findings.active_for(actor: "client_health_actor")
Cyclium.Findings.active_for(subject: %{kind: "client", id: "123"})
Cyclium.Findings.active_for(finding_key: "client:health:123")
Cyclium.Findings.active_for(class: "churned")
Cyclium.Findings.active_for(caused_by: "parent:finding:key")

Causality chains: Findings can reference a parent finding via caused_by_key. This enables tracing root causes through chains of related findings:

# Raise a child finding linked to a parent
{:raise, %{finding_key: "vendor:delay:PO-123", caused_by_key: "vendor:health:acme", ...}}

# Query helpers
Cyclium.Findings.caused_by("parent:key")        # direct children
Cyclium.Findings.causal_chain("child:key", 10)   # walk chain upward (max depth)
Cyclium.Findings.root_cause("child:key")         # find root (no caused_by_key)

TTL / Expiration: Findings can auto-expire after a duration. Declare a default TTL on the expectation, or pass ttl_seconds / expires_at per finding:

# Default TTL for all findings raised by this expectation
expectation(:check_temp,
  strategy: MyApp.Strategies.TempCheck,
  trigger: {:event, "sensor.updated"},
  finding_ttl_seconds: 3600
)

# Or override per finding in converge:
{:raise, %{finding_key: "temp:alert:123", ttl_seconds: 7200, ...}}

Expired findings are cleared and active findings are escalated by Cyclium.Findings.FindingSweep, an optional GenServer that runs on a configurable interval:

# config/config.exs
config :cyclium, :finding_sweep, true
config :cyclium, :finding_sweep_interval_ms, 300_000   # 5 minutes (default)
config :cyclium, :finding_sweep_batch_size, 100         # per sweep (default)

Severity escalation: Time-based rules automatically escalate finding severity based on how long a finding has been active. Declare rules on the expectation:

expectation(:check_vendor,
  strategy: MyApp.Strategies.VendorHealth,
  trigger: {:event, "vendor.updated"},
  escalation_rules: %{
    "vendor_delay" => [
      %{after_minutes: 60, escalate_to: :high},
      %{after_minutes: 1440, escalate_to: :critical}
    ]
  }
)

Escalation runs as part of the finding sweep cycle. Each sweep interval, every active finding matching a registered escalation pair (actor + expectation + classes) is loaded from the database and checked against the time-based rules. This means the sweep interval (finding_sweep_interval_ms) controls how often escalation is evaluated — rules with after_minutes granularity finer than the sweep interval won't fire any faster. For expectations with many active findings, keep this in mind when tuning the sweep interval and batch size. Application config (config :cyclium, :escalation_rules) is supported as a fallback.

Post-raise enrichment: An optional callback enriches findings immediately after they're raised. Declare it on the expectation:

expectation(:check_health,
  strategy: MyApp.Strategies.ClientHealth,
  trigger: {:event, "client.updated"},
  finding_enrichment: fn finding, _episode ->
    {:ok, %{summary: "Enriched: #{finding.summary}", confidence: 0.95}}
  end
)

# Or use a module/function tuple:
expectation(:check_health,
  strategy: MyApp.Strategies.ClientHealth,
  trigger: {:event, "client.updated"},
  finding_enrichment: {MyApp.FindingEnricher, :enrich}
)

The callback receives (finding, episode) and returns {:ok, %{...}} or :skip. Only safe fields are applied: evidence_refs, summary, confidence. Errors in the callback are logged — the finding persists unchanged. Application config (config :cyclium, :finding_enrichment) is supported as a fallback.

Outputs

Outputs are typed proposals produced during converge. They flow through the Output Router, which handles deduplication (via dedupe_key) and delivery through app-provided adapters.

# In converge result:
outputs: [
  %Cyclium.OutputProposal{
    type: :email,
    dedupe_key: "alert:client:123:#{Cyclium.Window.bucket(:h4, DateTime.utc_now())}",
    payload: %{to: "team@co.com", subject: "Client 123 churned"},
    requires_approval: false
  }
]

Register adapters in config:

config :cyclium, :output_adapters, %{
  email: MyApp.Adapters.Email,
  slack: MyApp.Adapters.Slack
}

Adapters implement Cyclium.Output.Adapter:

defmodule MyApp.Adapters.Email do
  @behaviour Cyclium.Output.Adapter

  @impl true
  def deliver(:email, payload, _ctx) do
    case MyApp.Mailer.send(payload) do
      :ok -> {:ok, %{message_id: "abc123"}}
      {:error, reason} -> {:error, reason}
    end
  end
end

The adapter registry provides programmatic access:

Cyclium.Output.Adapter.resolve(:email)    # => MyApp.Adapters.Email
Cyclium.Output.Adapter.resolve("slack")   # => MyApp.Adapters.Slack
Cyclium.Output.Adapter.all()              # => [:email, :slack]

Bus

The event bus connects actors, LiveViews, and workflows without coupling. It wraps Phoenix.PubSub.

# Publish a domain event (from your app code):
Cyclium.Bus.broadcast("client.updated", %{client_id: "123"})

# Subscribe to all events (actors do this automatically):
Cyclium.Bus.subscribe()

# Subscribe to a specific event:
Cyclium.Bus.subscribe("episode.completed")

# In a LiveView or GenServer:
def handle_info({:bus, "episode.completed", payload}, socket) do
  # payload contains: episode_id, actor_id, status, workflow_instance_id
end

Runtime events emitted by Cyclium:

Setup

1. Add dependency

# mix.exs
def deps do
  [{:cyclium, path: "../cyclium_ex"}]
end

Dependencies pulled in: ecto, ecto_sql, jason, phoenix_pubsub.

2. Run migrations

# In a migration file:
def up do
  Cyclium.Migrations.V1.up()   # episodes, steps, checkpoints, findings, outputs
  Cyclium.Migrations.V2.up()   # episode_logs
  Cyclium.Migrations.V3.up()   # workflow_instances
  Cyclium.Migrations.V4.up()   # archived_at on episodes and findings
  Cyclium.Migrations.V5.up()   # unique index on episode dedupe_key
  Cyclium.Migrations.V6.up()   # work_claims table for lease-based coordination
  # ...V7 through V13...
  Cyclium.Migrations.V14.up()  # trigger_requests table for deferred execution
  # ...V15 through V18...
  Cyclium.Migrations.V19.up()  # SQL Server: convert legacy TEXT columns to nvarchar(max)
end

def down do
  Cyclium.Migrations.V19.down()
  Cyclium.Migrations.V14.down()
  # ...V13 through V7...
  Cyclium.Migrations.V6.down()
  Cyclium.Migrations.V5.down()
  Cyclium.Migrations.V4.down()
  Cyclium.Migrations.V3.down()
  Cyclium.Migrations.V2.down()
  Cyclium.Migrations.V1.down()
end

Authoring migrations: do not use bare :text. On Ecto.Adapters.Tds it emits SQL Server's legacy non-Unicode TEXT type, which silently replaces emoji and other non-CP1252 characters with ?. Use {:string, size: :max} (which becomes nvarchar(max) on Tds and TEXT on Postgres/SQLite), or branch on repo().__adapter__() for finer control. V19 is the one-shot repair migration for columns that were already declared :text.

3. Configure

# config.exs
config :cyclium, :repo, MyApp.Repo

# Optional: registry for strategy/synthesizer overrides (see "Strategy registry" section)
# config :cyclium, :strategy_registry, MyApp.StrategyRegistry

# Optional: episode runner (default: Cyclium.Runner.OTP)
# Use Cyclium.Runner.Deferred for trigger-only mode (see "Trigger-Only Mode" section)
config :cyclium, :runner, Cyclium.Runner.OTP

# Optional: node identity override for shared-name environments (see "Node Identity")
# config :cyclium, :node_identity, "my-unique-node-name"

# Optional: tool capabilities
config :cyclium, :capability_registry, %{
  erp_read: MyApp.Tools.ERP,
  vendor_api: MyApp.Tools.VendorAPI
}

# Optional: output adapters
config :cyclium, :output_adapters, %{
  email: MyApp.Adapters.Email,
  slack: MyApp.Adapters.Slack
}

# Optional: checkpoint schemas for versioned state migration
config :cyclium, :checkpoint_schemas, %{
  {"client_health_actor", "client_should_be_healthy"} => MyApp.Checkpoints.HealthCheck
}

# Optional: enable reconciler for hot spec changes
config :cyclium, :reconciler, true

# Optional: register workflows
config :cyclium, :workflows, [MyApp.Workflows.ClientReview]

4. Declare strategies on expectations

The preferred approach is declaring the strategy module directly on each expectation in the actor DSL. Cyclium registers the mapping automatically when the actor GenServer boots — no separate registry needed:

defmodule MyApp.Actors.ClientHealthActor do
  use Cyclium.Actor

  actor do
    domain(:client_health)
    spec_rev("v0.1.0")
    synthesizer(MyApp.Synthesizers.ClientHealth)  # actor-level default
    max_concurrent_episodes(5)
    episode_overflow(:queue)

    expectation(:client_should_be_healthy,
      strategy: MyApp.Strategies.ClientHealth,
      trigger: {:event, "client.health_check_requested"},
      subject_key: :client_id,
      debounce_ms: :timer.seconds(3),
      budget: %{max_turns: 3, max_tokens: 1_000, max_wall_ms: 10_000}
    )

    expectation(:client_ai_summary,
      strategy: MyApp.Strategies.ClientAdvisor,
      synthesizer: MyApp.Synthesizers.FastSummary,  # override for this expectation
      trigger: {:event, "client.summary_requested"},
      budget: %{max_turns: 5, max_tokens: 5_000, max_wall_ms: 60_000}
    )
  end
end

Optional: strategy registry for overrides

If you need to override a strategy or synthesizer without changing the actor code — for example, in a staging environment or during an A/B test — you can configure a registry:

# config.exs (optional)
config :cyclium, :strategy_registry, MyApp.StrategyRegistry

defmodule MyApp.StrategyRegistry do
  # Only add clauses for explicit overrides — anything not matched here
  # falls through to the actor's declared strategy.
  def strategy_for("client_health_actor", _exp), do: MyApp.Strategies.ClientHealthV2
end

5. Supervision tree

# application.ex
children = [
  MyApp.Repo,
  {Phoenix.PubSub, name: MyApp.PubSub},
  {Cyclium.Supervisor, pubsub: MyApp.PubSub},
  MyApp.Actors.ClientHealthActor,
  MyApp.Actors.ClientAdvisorActor,
  MyAppWeb.Endpoint
]

Cyclium.Supervisor starts the DynamicSupervisors, TaskSupervisor, and optionally the Reconciler and WorkflowEngine.

Budgets

Every expectation declares a budget. The runner enforces all three dimensions:

budget: %{
  max_turns: 12,        # loop iterations (incremented every next_step call)
  max_tokens: 25_000,   # LLM token cost (incremented by tool_call results)
  max_wall_ms: 120_000  # wall-clock deadline (enforced via Process.send_after)
}

When any limit is hit, the episode fails with error_class: "budget_exceeded". Wall time is enforced asynchronously — a :budget_wall_exceeded message interrupts the loop even if the strategy is blocked on a tool call.

Deduplication: actor-level vs. strategy-level

Cyclium provides three layers of temporal dedup:

Actor-level global (cooldown_ms) — enforced by the actor GenServer before an episode starts. Simple and zero-cost, but applies globally to the expectation — all subjects are blocked during the window.

expectation(:client_ai_summary,
  strategy: MyApp.Strategies.ClientAdvisor,
  trigger: {:event, "client.summary_requested"},
  cooldown_ms: :timer.minutes(5)  # no advisor episodes for ANY client for 5 min
)

Actor-level per-subject (subject_key + debounce_ms/cooldown_ms) — when subject_key is set, debounce and cooldown are scoped per subject value. Each unique subject gets its own independent trailing-edge timer and cooldown window. Client A and client B are debounced independently.

expectation(:client_ai_summary,
  strategy: MyApp.Strategies.ClientAdvisor,
  trigger: {:event, "client.summary_requested"},
  subject_key: :client_id,
  debounce_ms: :timer.seconds(10),   # trailing-edge, per-client
  cooldown_ms: :timer.minutes(5)     # minimum gap, per-client
)

When subject_key is set but the payload doesn't contain that key, the subject value is nil and the key becomes {expectation_id, nil} — still isolated from real subjects, never crashing.

Strategy-level (Findings.recent?/2) — checked in init/2 using the existing finding for a specific subject. This is DB-backed, so it survives actor restarts unlike the in-memory actor-level dedup.

@summary_cooldown_ms :timer.minutes(5)

def init(_episode, trigger) do
  client_id = trigger.payload["client_id"]
  skip = Cyclium.Findings.recent?("client:advisor:#{client_id}", @summary_cooldown_ms)
  {:ok, %{client_id: client_id, skip: skip}}
end

def next_step(%{skip: true}, _ctx), do: :done

When to use which:

Workflows

Workflows coordinate multiple actors in a dependency graph. Data flows between steps via the workflow_result/2 strategy callback and input: functions on downstream steps.

Defining a workflow

defmodule MyApp.Workflows.ClientReview do
  use Cyclium.Workflow

  workflow do
    trigger {:event, "client.updated"}
    debounce_ms :timer.seconds(3)
    subject_key :client_id

    step :health_check,
      actor: :client_health_actor,
      expectation: :client_should_be_healthy

    step :ai_summary,
      actor: :client_advisor_actor,
      expectation: :client_ai_summary,
      depends_on: [:health_check],
      input: fn _trigger, prior ->
        # prior[:health_check] contains the map returned by
        # the health strategy's workflow_result/2 callback
        %{client_id: prior[:health_check].client_id}
      end

    on_failure :health_check, policy: :retry, max_step_attempts: 3, backoff_ms: 5_000
    on_failure :ai_summary, policy: :abort
  end
end

Workflow debounce:debounce_ms and subject_key coalesce rapid events before starting the workflow. When subject_key is set, each unique subject value gets its own debounce window — client A and client B debounce independently. Without subject_key, all events for the workflow share a single timer. Each new event resets the debounce window (trailing-edge). This is useful for workflows that trigger on high-frequency events like "entity.updated".

Episode reuse (cross-workflow dedup): By default, workflows reuse recent completed episodes when two workflows trigger the same actor + expectation + input within a 5-minute window. To disable this:

workflow do
  disable_episode_reuse

  trigger {:event, "client.review_requested"}
  step :health_check, actor: :client_health_actor, expectation: :client_should_be_healthy
end

Cancellation cascade: When a workflow fails, pending and retrying steps are automatically canceled. To also clear active findings raised by the workflow's episodes, set clear_findings_on_cancel in the workflow instance metadata.

Passing data between steps

When a workflow step completes, the engine calls the strategy's optional workflow_result/2 callback to extract the data that downstream steps receive via prior. If workflow_result/2 is not implemented, downstream steps receive nil for that step's prior.

defmodule MyApp.Strategies.ClientHealth do
  @behaviour Cyclium.EpisodeRunner.Strategy

  # ... init, next_step, handle_result, converge as usual ...

  # Optional: extract data for downstream workflow steps
  @impl true
  def workflow_result(state, _converge_result) do
    # This map becomes prior[:health_check] in downstream input functions
    %{client_id: state.client_id, classification: state.classification}
  end
end

Configuration and usage

Register workflows in config:

config :cyclium, :workflows, [MyApp.Workflows.ClientReview]

The WorkflowEngine GenServer:

• Listens for trigger events on the Bus
• Creates a WorkflowInstance record to track execution
• Fires steps in dependency order (DAG validated at compile time)
• Passes data between steps via workflow_result/2 → input functions
• Applies failure policies per-step: :abort (cancel all), :retry (with backoff), :pause (wait for manual intervention)

Workflows can also be started manually:

Cyclium.WorkflowEngine.start_workflow(
  MyApp.Workflows.ClientReview,
  %{client_id: "123"},
  []
)

Dynamic workflows

Dynamic workflows can be defined in the database and registered at runtime — no compiled modules required. They follow the same step-dependency model as compiled workflows but use declarative input mappings instead of Elixir functions.

Defining a workflow in the database

Insert a row into cyclium_workflow_definitions:

%Cyclium.Schemas.WorkflowDefinition{
  workflow_id: "vendor_onboarding",
  trigger_type: "event",
  trigger_event: "vendor.registration_submitted",
  steps: Jason.encode!([
    %{
      "id" => "compliance_check",
      "actor_id" => "compliance_monitor",
      "expectation" => "vendor_risk",
      "depends_on" => [],
      "input_map" => %{"vendor_id" => "trigger.vendor_id"},
      "failure_policy" => "abort"
    },
    %{
      "id" => "connector_setup",
      "actor_id" => "integration_actor",
      "expectation" => "setup_connector",
      "depends_on" => ["compliance_check"],
      "input_map" => %{
        "vendor_id" => "trigger.vendor_id",
        "risk_level" => "prior.compliance_check.classification.primary"
      },
      "failure_policy" => "retry",
      "max_step_attempts" => 2,
      "backoff_ms" => 30000
    }
  ]),
  enabled: true
}

Input mapping syntax

Dynamic workflows use dot-notation paths instead of Elixir functions:

Loading dynamic workflows

# At startup — loads all enabled definitions
Cyclium.DynamicWorkflow.Loader.load_all()

# Load a single workflow
Cyclium.DynamicWorkflow.Loader.load("vendor_onboarding")

# Reload after updating the definition in DB
Cyclium.DynamicWorkflow.Loader.reload("vendor_onboarding")

# Unregister a workflow
Cyclium.DynamicWorkflow.Loader.unload("vendor_onboarding")

Starting dynamic workflows manually

Cyclium.WorkflowEngine.start_dynamic_workflow(
  "vendor_onboarding",
  %{"vendor_id" => "v-123"}
)

Dynamic workflows are event-triggered (via Bus) like compiled workflows. The Watcher also listens for workflow_definition.created/updated/disabled events for automatic refresh.

Circuit breaker

Per-expectation circuit breaker prevents cascading failures when a tool or external service is down. When consecutive episode failures exceed a threshold, the circuit opens and rejects new episodes. After a cooldown period, one probe episode is allowed through (half-open state) — if it succeeds, the circuit closes.

expectation(:check_vendor_api,
  strategy: MyApp.Strategies.VendorCheck,
  trigger: {:event, "vendor.updated"},
  circuit_breaker: %{
    threshold: 5,               # consecutive failures to trip
    half_open_after_ms: 60_000, # cooldown before probe
    cancel_in_flight: false     # cancel running episodes when circuit trips
  }
)

States::closed (normal) → :open (rejecting) → :half_open (probe) → :closed

In-flight cancellation: When cancel_in_flight: true, tripping the circuit also cancels any running or blocked episodes for that actor + expectation, preventing wasted work against a known-broken dependency.

Scope: Circuit breakers are node-local (ETS-backed). In a cluster, each node tracks failures independently. This is intentional — a service might be unreachable from one node but fine from another. For cluster-wide coordination, combine with the Bus (circuit breaker events are broadcast).

Force-fired episodes bypass the circuit breaker check.

Query state: Cyclium.CircuitBreaker.get_state(actor_id, expectation_id)

Episode sampling

Probabilistic episode firing for high-frequency triggers. Set sample_rate on an expectation to control what fraction of triggers actually fire episodes:

expectation(:health_check,
  strategy: MyApp.Strategies.MetricsCheck,
  trigger: {:event, "metrics.updated"},
  sample_rate: 0.1  # fire ~10% of triggers
)

• nil or 1.0 = always fire (default)
• 0.0 = never fire
• Sampled-out episodes emit [:cyclium, :episode, :sampled_out] telemetry
• Force-fired episodes bypass sampling

Service level tracking

Declarative performance objectives with automatic breach detection. Define success rate and duration thresholds per expectation:

expectation(:process_order,
  strategy: MyApp.Strategies.OrderProcessor,
  trigger: {:event, "order.created"},
  service_levels: %{
    max_duration_ms: 30_000,  # p95 target
    success_rate: 0.95,       # 95% success target
    window_episodes: 20       # rolling window size
  }
)

Breaches emit [:cyclium, :service_levels, :breach] telemetry and a "service_levels.breach" Bus event with details:

%{type: :success_rate, current: 0.85, threshold: 0.95}
%{type: :duration, current: 45_000, threshold: 30_000}

Query metrics: Cyclium.ServiceLevels.metrics(actor_id, expectation_id) returns %{success_rate: f, p95_duration_ms: n, sample_count: n}.

Adaptive budgets

Advisory budget tracking based on historical episode resource usage. When enabled, Cyclium records turns, tokens, and wall time for each completed episode and recommends budgets based on p95 values with 25% headroom.

expectation(:classify_ticket,
  strategy: MyApp.Strategies.TicketClassifier,
  trigger: {:event, "ticket.created"},
  adaptive_budget: true
)

Query recommendations:

# After enough samples (minimum 5):
Cyclium.AdaptiveBudget.recommend(actor_id, expectation_id)
# => %{max_turns: 8, max_tokens: 15_000, max_wall_ms: 25_000}

# Detailed stats:
Cyclium.AdaptiveBudget.stats(actor_id, expectation_id)
# => %{samples: 47, p50: %{...}, p95: %{...}, max: %{...}}

Adaptive budgets are advisory only — they do not automatically adjust episode budgets. Use the recommendations to tune your expectation configs over time.

Logging and observability

Log strategies

Set per-expectation via log_strategy. Controls both what gets stored in step journal columns (args_redacted, result_ref) and what the materialized log renders:

Use :full_debug for audit-sensitive workflows where you need to reconstruct exactly what context an LLM had (EOX predictions, SKU classifications). Use :timeline for high-frequency episodes where you want the flow visible without storing full payloads.

Important:log_strategy is the last layer in a pipeline. By the time it runs, earlier layers have already trimmed the data. "Full" in :full_debug means "everything that survived the earlier layers", not necessarily everything the strategy originally passed. See the pipeline below.

The materialized log is built by LogProjector reading back from the already-stored step rows — it never re-processes the original payload. Whatever log_strategy stored is exactly what appears in the log.

Storage pipeline for synthesis steps

Every synthesis step passes through three filtering layers in order before anything reaches the database:

synthesis payload from next_step
        │
        ▼
1. __transient__ stripping   — keys listed in :__transient__ are removed from storage
        │                       but the synthesizer receives the full payload
        ▼
2. tool redact callbacks     — redact/1 and redact_result/1 on tool steps
        │                       (not applicable to synthesis, but runs for tool_call steps)
        ▼
3. log_strategy filtering    — controls final shape of args_redacted / result_ref
        │
        ▼
   cyclium_episode_steps (args_redacted, result_ref)
        │
        ▼
   cyclium_episode_logs (rendered by LogProjector from stored step rows)

Each layer has a different scope:

Transient synthesis data

Sometimes a strategy needs to pass large data to the synthesizer (full record lists, raw API payloads) that the LLM needs for context but that you don't want persisted in the step journal or materialized log. Mark those keys under :__transient__ in the synthesis payload:

def next_step(state, _ctx) do
  orders = load_orders(state.project_id)

  {:synthesize, %{
    project_id: state.project_id,
    project_name: state.project_name,
    order_count: length(orders),           # small scalar — kept in storage
    orders: serialize_orders(orders),      # full list — synthesizer needs it, storage does not
    evidence: build_evidence(orders),      # small structured summary — kept in storage
    __transient__: [:orders]               # strip :orders before writing args_redacted
  }}
end

The runner passes the full map (minus :__transient__ itself) to synthesizer.synthesize/2, then drops the listed keys before handing off to log_strategy filtering. The synthesizer receives orders; the stored step and rendered log do not, regardless of log_strategy.

This is the right tool when:

• You need full detail for synthesis quality (long order lists, raw API responses, document text)
• You don't want that data in the audit trail or materialized log
• You still want other context fields (counts, summaries, IDs) persisted for debugging

It is not a substitute for log_strategy — if you want no storage at all for an episode type, use :none or :summary_only. __transient__ is surgical; log_strategy is wholesale.

Materialized logs are stored in cyclium_episode_logs by Cyclium.LogProjector and can be queried via Cyclium.Episodes.get_log(episode_id).

Telemetry

Cyclium emits 36 structured telemetry events under the [:cyclium, ...] prefix. Attach a handler for development:

Cyclium.Telemetry.attach_default_logger()

Key events:

Full list: Cyclium.Telemetry.events/0

Step journal

Every episode action is recorded as an EpisodeStep with one of 16 kinds:

tool_call, synthesis, observation, checkpoint, output_proposed, output_delivered, output_failed, approval_requested, approval_resolved, wait_started, wait_resolved, finding_raised, finding_updated, finding_cleared, episode_completed, episode_failed

Each step records: step_no, kind, tool_name, args_redacted, result_ref, error_class, error_detail, cost_tokens, cost_ms, created_at.

Query steps: Cyclium.Episodes.list_steps(episode_id)

Checkpointing

Strategies can save state mid-episode for crash recovery. Return {:checkpoint, phase_name} from next_step/2 to persist the current state:

def next_step(state, _ctx) do
  if state.phase == :data_collected do
    {:checkpoint, "data_collected"}
  else
    {:tool_call, :erp_read, :read_po, %{"po_id" => state.po_id}}
  end
end

The checkpoint saves the full strategy state map to the cyclium_episode_checkpoints table. On resume, EpisodeTask loads the latest checkpoint by checkpoint_no and passes it to the strategy — execution continues from where it left off.

Checkpoint schema versioning

If your strategy's state shape changes between deploys, register a checkpoint schema to migrate old checkpoints forward:

defmodule MyApp.Checkpoints.HealthCheck do
  use Cyclium.CheckpointSchema, version: 2

  # Migrate from version 1 -> 2
  def migrate(1, state), do: {:ok, Map.put(state, :new_field, nil)}
  def migrate(2, state), do: {:ok, state}
end

Register in config:

config :cyclium, :checkpoint_schemas, %{
  {"client_health_actor", "client_should_be_healthy"} => MyApp.Checkpoints.HealthCheck
}

If migration fails, EpisodeTask falls back to a fresh strategy.init/2 — the episode restarts from scratch.

When to checkpoint vs. restart

Use checkpoints when:

• The strategy accumulates state across many turns that would be expensive to recompute (multi-turn LLM conversations, progressive data aggregation)
• Steps have non-idempotent side effects that shouldn't be repeated

Use restart (no checkpoint) when:

• The strategy re-queries all data from the DB each turn (most monitoring/health strategies)
• Side effects are idempotent (findings upsert by key, outputs are deduplicated)
• Episodes are short (< a few minutes)

Most strategies don't need checkpoints — recovery_policy: :restart on the expectation handles recovery by re-running the episode from scratch. See the Recovery section below.

Recovery

Cyclium provides built-in recovery for orphaned episodes after server restarts or deploys.

The problem

When a node shuts down (deploy, crash, scaling event), in-flight episodes are killed and left as :running in the database. Without recovery, these episodes stay orphaned forever.

Distributed episode claiming

In a multi-node cluster, all nodes run the same actors and receive the same Bus events via PG2. Without coordination, every node would independently create and run an episode for every trigger — tripling (or more) the work. Cyclium uses DB-based coordination with no Redis or leader election required.

Episode creation: dedupe_key

When a trigger fires, every node's actor calls maybe_fire_episode. Before inserting, the actor generates a deterministic dedupe_key:

• Schedule triggers:"schedule:{actor_id}:{expectation_id}:{date}" — one episode per schedule window
• Event triggers:"event:{actor_id}:{expectation_id}:{payload_hash}" — one episode per distinct event payload

A filtered unique index on dedupe_key (WHERE dedupe_key IS NOT NULL AND archived_at IS NULL) ensures only one node's insert succeeds. The other nodes receive a constraint violation and silently skip — no episode created, no work duplicated. Archived episodes are excluded from the constraint so that re-triggered work isn't blocked by old runs.

A random jitter (0-200ms) before insert spreads winners evenly across nodes, preventing one faster node from consistently claiming all episodes:

# In Actor.Server.maybe_fire_episode/3
Process.sleep(:rand.uniform(200))
enqueue_episode(state, episode_params)

The constraint violation is caught in enqueue_episode using the same pattern as the Output router:

{:error, %Ecto.Changeset{} = cs} ->
  if has_dedupe_violation?(cs) do
    Logger.debug("[#{state.actor_id}] Dedupe skip: #{params.dedupe_key}")
  end
  state

Recovery claims: optimistic update

After a restart, all surviving nodes run Cyclium.Recovery.sweep/1. Each node sees the same list of stale episodes, but only one node can claim each episode:

# Episodes.claim_for_recovery/1
from(e in Episode,
  where: e.id == ^episode_id and e.status == :running and is_nil(e.archived_at)
)
|> repo().update_all(set: [phase: "recovering"])

This is an atomic UPDATE ... WHERE — the first node to execute it sets phase: "recovering" and gets {1, _} back. All other nodes get {0, _} (no rows affected) and skip. No locks, no races, no distributed coordination protocol needed.

Summary of coordination guarantees

Recovery sweep

Cyclium.Recovery.sweep/1 finds stale :running episodes and recovers them:

1. Query episodes where the most recent step journal entry is older than stale_after_ms (default: 2 minutes)
2. Attempt optimistic claim on each stale episode
3. If claimed, apply the expectation's recovery_policy:
   • :restart — enqueue fresh (re-runs strategy.init/2)
   • :fail (default) — mark :failed with error_class: "orphaned"
4. Emit [:cyclium, :recovery, :sweep] telemetry with counts

Policy resolution checks the compiled :actor_registry map first, then falls back to the cyclium_agent_definitions table for dynamic actors. Unknown actors default to :fail.

Workflow reconciliation

Cyclium.Recovery.reconcile_workflows/0 handles a related problem: workflow instances stuck in :running because the WorkflowEngine missed a Bus event during a restart.

The WorkflowEngine is purely event-driven — it advances workflows when it receives episode.completed or episode.failed Bus events. If the engine wasn't running when those events fired (e.g. during a deploy), the workflow step_states become stale and the workflow hangs.

Reconciliation fixes this by:

1. Finding all :running / :blocked workflow instances
2. For each step marked "running" in step_states, loading the actual episode
3. If the episode has already reached a terminal state, re-broadcasting the appropriate Bus event
4. The WorkflowEngine handles the replayed event through its normal path — no special logic needed

Call reconcile_workflows/0aftersweep/1 and after workflow configs are registered (compiled modules booted, dynamic workflows loaded).

Setting recovery policy

Add recovery_policy to the expectation DSL:

actor do
  expectation(:evaluate_project,
    strategy: MyApp.Strategies.ProjectHealth,
    trigger: {:event, "project_health.check_requested"},
    recovery_policy: :restart,
    budget: %{max_turns: 5, max_tokens: 25_000, max_wall_ms: 120_000}
  )
end

Use :restart for idempotent strategies that re-query data from the DB. Use :fail (default) for strategies with non-idempotent side effects where automatic recovery could cause harm.

Wiring up recovery in your app

Add a delayed recovery task to your Cyclium supervisor with an :actor_registry map that maps actor identifier() strings to their modules. Cyclium looks up the recovery_policy from each actor's compiled expectations automatically. Dynamic actors not in the registry are resolved from the DB.

# Maps identifier (as DB string) → actor module for recovery sweep.
# Must match the identifier() declared in each actor's DSL block.
@actor_registry %{
  "project_health_actor" => MyApp.Actors.ProjectHealthActor,
  "client_health_actor" => MyApp.Actors.ClientHealthActor
}

children = [
  {Cyclium.Supervisor, pubsub: MyApp.PubSub},
  MyApp.Actors.ProjectHealthActor,
  MyApp.Actors.ClientHealthActor,
  {Task, fn ->
    # Wait for cluster to settle after deploy
    Process.sleep(:timer.minutes(2))
    Cyclium.Recovery.sweep(actor_registry: @actor_registry)
    Cyclium.Recovery.reconcile_workflows()
  end}
]

For custom policy logic, pass :resolve_policy instead:

Cyclium.Recovery.sweep(
  resolve_policy: fn episode ->
    if episode.actor_id == "critical_actor", do: :restart, else: :fail
  end
)

Deploy sequence

1. Node gets SIGTERM -> 30s graceful shutdown timeout
2. Episodes trap exits, try to finish current step within remaining time
3. Episodes that don't finish stay :running in DB
4. After boot, each node waits 2 minutes before running recovery sweep
5. Sweep finds stale episodes via step journal recency
6. First node to claim each orphan via optimistic update handles recovery
7. Workflow reconciliation replays missed Bus events for stale workflow steps

Step Retry Helper

Cyclium.Strategy.Retry provides a lightweight helper for retrying failed steps within an episode. This is distinct from workflow-level retry (on_failure :step, :retry) which retries entire episodes — the step retry helper retries individual steps (e.g. a synthesis call) within a single episode run.

The problem

When a strategy calls :synthesize and the LLM provider returns a transient error (timeout, rate limit, 503), the strategy needs to retry. Without a helper, you'd manually track attempt counts in the state map:

# Without helper — manual retry tracking
def handle_result(state, %{kind: :synthesis}, {:error, _}) do
  if state[:synthesis_retries] < 3 do
    {:retry, Map.update(state, :synthesis_retries, 1, &(&1 + 1))}
  else
    {:abort, "synthesis_failed"}
  end
end

This is error-prone (forgetting to reset counters, tracking multiple step types, off-by-one errors).

Using Cyclium.Strategy.Retry

alias Cyclium.Strategy.Retry

# On success — reset the counter so future failures get fresh attempts
def handle_result(state, %{kind: :synthesis}, {:ok, result}) do
  {:ok, state |> Retry.reset(:synthesis) |> Map.put(:assessment, result)}
end

# On failure — retry up to 3 times with 2-second backoff
def handle_result(state, %{kind: :synthesis} = step, {:error, _}) do
  case Retry.check(state, step, max_attempts: 3, backoff_ms: 2_000) do
    {:retry, new_state}              -> {:retry, new_state}
    {:give_up, _attempts, new_state} -> {:abort, "synthesis_failed_after_retries"}
  end
end

When handle_result returns {:retry, state}, the runner calls do_loop → next_step again. The strategy should naturally re-emit the same step type (e.g. :synthesize) since its phase/state hasn't changed — only the internal __retries counter was updated.

Options

Custom step keys

Track retries separately for different tool calls within the same episode:

def handle_result(state, %{kind: :tool_call, tool_name: tool} = step, {:error, _}) do
  case Retry.check(state, step, step_key: {:tool, tool}, max_attempts: 2) do
    {:retry, new_state}              -> {:retry, new_state}
    {:give_up, _attempts, new_state} -> {:ok, %{new_state | phase: :skip_tool}}
  end
end

API reference

• Retry.check(state, step, opts) — returns {:retry, state} or {:give_up, count, state}
• Retry.reset(state, key) — clears the counter for one key (call on success)
• Retry.reset_all(state) — clears all retry tracking

Retry layers summary

Work Claims (Distributed Lease Coordination)

For clusters where multiple applications share the same database and actor definitions, work claims provide lease-based coordination to ensure at-most-once execution.

How it works

1. Before executing an episode, EpisodeTask calls WorkClaims.gate_acquire/3 with the episode's dedupe_key
2. If claimed successfully, a Heartbeat GenServer renews the lease periodically (every lease/3 seconds)
3. On completion, the claim is marked :done; on crash, :failed
4. If a node dies, the lease expires and another node can steal it

Work claims also coordinate trigger request dispatch — TriggerRequests.Poller acquires a claim per request before dispatching to prevent multiple full-mode nodes from processing the same deferred episode.

Configuration

# Use the built-in Ecto-based implementation:
config :cyclium, work_claims: Cyclium.WorkClaims.EctoClaims

# Or a SQL Server-optimized adapter in consuming apps:
config :cyclium, work_claims: MyApp.WorkClaims.SqlServer

# Lease duration (default: 120 seconds)
config :cyclium, work_claims_lease_seconds: 180

# Or omit work_claims entirely — no claiming, fully backwards compatible

When unconfigured, all gate_* functions return passthrough values with zero overhead.

Writing a custom adapter

Implement the Cyclium.WorkClaims behaviour with 5 callbacks:

defmodule MyApp.WorkClaims.SqlServer do
  @behaviour Cyclium.WorkClaims

  @impl true
  def acquire(dedupe_key, owner_node, opts) do
    # Use hints: ["UPDLOCK"] for SQL Server lock acquisition
    # Transaction-based: read with lock, then insert or update
  end

  @impl true
  def renew(dedupe_key, owner_node, lease_seconds), do: # ...
  def complete(dedupe_key, owner_node), do: # ...
  def fail(dedupe_key, owner_node, error_detail), do: # ...
  def reclaim_expired(limit), do: # ...
end

The default EctoClaims implementation uses plain transactions (no lock hints) and works with any Ecto adapter. For SQL Server, a custom adapter can use hints: ["UPDLOCK"] on the read query inside the transaction for stronger concurrency guarantees.

Database table

The cyclium_work_claims table is created by V6 migration:

Integration with recovery

When work claims are configured, Recovery.sweep/1 uses gate_acquire to coordinate across nodes before claiming orphaned episodes. This provides two layers of coordination: the work claim lease prevents concurrent execution, and the optimistic claim_for_recovery update prevents duplicate recovery actions.

Lease tuning

The lease duration (work_claims_lease_seconds, default: 120s) controls the trade-off between availability and the "zombie window" — the time between a node crash and another node stealing the work.

Guidelines:

• Heartbeat fires at lease / 3 — set the lease to at least 3x your worst-case DB round-trip time
• If your episodes typically run for minutes, 120s is fine — the heartbeat keeps the lease alive indefinitely
• If network partitions last >30s regularly, increase the lease to avoid false steals
• After a steal, the new node restarts the episode fresh (strategies should be idempotent)

Heartbeat failure modes

The heartbeat GenServer is linked to the EpisodeTask process:

• Heartbeat crashes — The EpisodeTask traps the EXIT and can restart the heartbeat. The lease has margin (only 1/3 expired per interval), so a brief restart is safe.
• EpisodeTask crashes — The heartbeat dies with it. The rescue block marks the claim as :failed. If it doesn't (hard kill), the lease expires naturally and another node can steal it.
• DB becomes unreachable — Heartbeat renewal fails, lease expires. When DB comes back, another node may steal. This is by design — if you can't reach the DB, you can't guarantee exclusive access.
• Lost ownership — If gate_renew returns {:error, :not_owner} (another node stole the claim), the heartbeat stops itself.

Idempotency guidance: Since lease expiry can cause a second node to start the same work, strategies that perform side effects should use idempotency keys. For DB writes, use unique constraints keyed by the episode's dedupe_key or step number. For external API calls, include an idempotency header derived from the episode ID + step number.

Telemetry events

All events are prefixed with [:cyclium, :work_claims, ...]:

Key metrics to alert on:

• steal rate > 0 during normal operation → nodes are dying or leases are too short
• busy rate proportional to node count → expected (N-1 nodes get busy per dedupe key)
• renew_failed > 0 → possible clock drift or DB contention

Testing work claims

Unit tests: Use Cyclium.WorkClaims.FakeClaims — an Agent-backed in-memory implementation:

setup do
  {:ok, _} = Cyclium.WorkClaims.FakeClaims.start_link()
  Application.put_env(:cyclium, :work_claims, Cyclium.WorkClaims.FakeClaims)
  on_exit(fn -> Application.delete_env(:cyclium, :work_claims) end)
end

test "second acquire is busy" do
  assert {:ok, _} = Cyclium.WorkClaims.gate_acquire("key:1", "node-a")
  Cyclium.WorkClaims.FakeClaims.set_busy("key:2")
  assert {:error, :busy} = Cyclium.WorkClaims.gate_acquire("key:2", "node-b")
end

Integration tests (single node): Configure EctoClaims with your test repo. Verify:

1. Two concurrent acquire calls on the same key — one succeeds, one gets :busy
2. After complete, a new acquire on the same key succeeds (reclaim)
3. After lease expiry (set a short lease), acquire steals from the previous owner

Multi-node tests: Deploy to a staging cluster with 2+ nodes. Use a test actor with a short schedule:

1. Verify only one node's episode runs (check owner_node in cyclium_work_claims)
2. Kill a node mid-episode, wait for lease expiry, verify another node steals and completes
3. Monitor telemetry — steal events should only appear after the kill, never during normal operation

Node Identity

By default, Cyclium uses node() to identify the current BEAM instance for work claims, trigger requests, and recovery coordination. In environments where multiple instances share the same Erlang node name (e.g., dev containers all starting as app@app), this breaks lease semantics — every node looks like the same owner.

Cyclium.NodeIdentity provides a pluggable identity layer:

# Static override — set per instance via config or env var
config :cyclium, :node_identity, "dev-jane"

# MFA callback for dynamic resolution (hostname, env var, etc.)
config :cyclium, :node_identity, {MyApp.NodeIdentity, :resolve, []}

When unconfigured and running in non-distributed mode (:nonode@nohost), a random stable identity is generated per BEAM instance and stored in :persistent_term — unique for the process lifetime but not across restarts.

All work claim operations (EpisodeTask, Heartbeat, Runner.Deferred, TriggerRequests.Poller) use Cyclium.NodeIdentity.name() instead of raw node().

Multi-Stack Deployments

A stack is one logical cyclium cluster that shares a database with other clusters but runs its own Elixir nodes (its own Phoenix.PubSub, its own in-memory persistent_term / ETS registries). Typical reasons to run multiple stacks against one schema: independent release cadences, blast-radius isolation, or partitioning actors across regions.

The library contract is cluster-level

Cyclium itself exposes no per-actor DSL option for stacks. It reads :stack_slug once per cluster, stamps every row its actors produce with that value, and scopes Recovery to the matching slug. Which actors actually run on a given cluster is the consumer's decision, implemented in the host app's supervisor.

Episodes, workflow instances, and deferred trigger requests are stamped with the current source_stack at insert time. Cyclium.Recovery.sweep/1 and Cyclium.Recovery.reconcile_workflows/1 read :stack_slug by default and only scan rows from their own stack — this prevents a crashed cluster's work from being re-driven on a cluster whose persistent_term / PubSub state doesn't know about it.

Partitioning actors across stacks

The simplest approach is one actor list per deployment:

# On the stack_a cluster's host app:
config :cyclium, :stack_slug, System.get_env("CYCLIUM_STACK_SLUG")   # "stack_a"
config :my_app, :cyclium_actors, [StackAOnlyActor, SharedActor]

# On the stack_b cluster's host app:
config :cyclium, :stack_slug, System.get_env("CYCLIUM_STACK_SLUG")   # "stack_b"
config :my_app, :cyclium_actors, [StackBOnlyActor, SharedActor]

A richer approach declares the allowed stacks on each actor's child-spec and has the supervisor filter the list at init time — useful when the same release is deployed to every cluster and you want a single source of truth for which actor runs where:

config :my_app, :cyclium_actors, [
  {SharedActor, []},
  {StackAOnlyActor, stacks: [:stack_a]},
  {CrossStackActor, stacks: [:stack_a, :stack_b]}
]

# In the supervisor's init/1:
stack = Application.get_env(:cyclium, :stack_slug)
children = Enum.filter(configured_actors, &actor_runs_on_stack?(&1, stack))

Either pattern is fine — cyclium doesn't care how the list was built, only what actually gets supervised. Because each cluster has its own node processes, PubSub, and persistent_term cache, a stack-local actor's strategy / budget / log-strategy lookups only exist on the cluster that supervises it — which is exactly why Recovery must be stack-scoped.

Runtime configuration

For a single release that can be deployed into multiple stacks, drive the slug from an env var in runtime.exs:

# runtime.exs
config :cyclium, :stack_slug, System.get_env("CYCLIUM_STACK_SLUG")

Leaving :stack_slug unset (or nil) is the single-stack default: rows are stamped NULL and Recovery scans without a stack filter. Pre-migration rows with NULLsource_stack are swept by any stack for one release so legacy episodes aren't orphaned — once all rows have been stamped, you can tighten Recovery by stamping a real slug on every cluster.

Related config

• :stack_slug — identity of this cluster (stamps rows and scopes Recovery)
• :trigger_poll_source_stack — separate filter on TriggerRequests.Poller: which stacks' deferred requests this full-mode node will pick up (often the same value as :stack_slug, but can be broader)
• Cyclium.StackSlug.current/0 — read the slug; returns nil when unset

Trigger-Only Mode (Deferred Execution)

In shared environments — dev machines on a common test DB, QC/sandbox instances, CI — running cyclium actors on every node leads to competing work claims and unpredictable execution. Conversely, disabling cyclium entirely on non-processing nodes leaves the UI hobbled: episodes never fire, statuses never update.

Trigger-only mode solves both problems by decoupling event processing from episode execution.

Three operating modes

How it works

In :trigger_only mode, the actor supervision tree starts normally — Bus subscriptions, schedule timers, debounce, circuit breakers all work. But the runner is swapped to Cyclium.Runner.Deferred, which writes a row to cyclium_trigger_requests instead of spawning a Task. The episode record is still created in the DB so the UI can display it.

On :full mode nodes, a Cyclium.TriggerRequests.Poller watches the trigger requests table and dispatches deferred episodes to Runner.OTP for local execution. The poller uses WorkClaims.gate_acquire/3 (with dedupe key "trigger_request:<id>") to coordinate dispatch across nodes — only one full-mode node will pick up a given request. If work claims are not configured, the poller falls through to passthrough mode. The poller can be scoped by source_stack to only pick up requests from specific stacks.

Configuration

# Host app config — set per environment
config :my_app, :cyclium_mode, :full          # :full | :trigger_only | :disabled

# On full-mode nodes: enable the poller
config :cyclium, :trigger_poller, true
config :cyclium, :trigger_poll_interval_ms, 5_000      # default
config :cyclium, :trigger_poll_source_stack, "stack_a"   # nil = pick up all

# On trigger-only nodes: runner is set automatically
# config :cyclium, :runner, Cyclium.Runner.Deferred
# config :cyclium, :stack_slug, :stack_a

Deployment scenarios

Dev machines + shared test DB:

• QC/test node runs :full with the poller enabled
• Dev machines run :trigger_only — events flow, UI works, episodes defer to the processing node
• A dev who wants to process locally overrides to :full and narrows their actor list

Sandbox / feature-branch testing:

• Sandbox runs :trigger_only — UI flows complete, episode records exist, Bus events fire
• Designated processing node runs :full and picks up deferred triggers

Production (unchanged):

• All nodes run :full as before; work claims handle multi-node coordination
• The trigger requests table stays empty

Database table

The cyclium_trigger_requests table (V14 migration):

Indexed on (status, inserted_at) for efficient polling.

Runtime mode switching

Cyclium.Mode supports live mode changes without restart — both node-wide and per-actor:

# Switch the whole node (via remote console, admin endpoint, etc.)
Cyclium.Mode.set(:trigger_only)   # stop local execution, defer to DB
Cyclium.Mode.set(:full)           # resume local execution + polling

# Per-actor override — yield one actor to another node while keeping the rest
Cyclium.Mode.set_actor_override(:client_health, :trigger_only)
Cyclium.Mode.clear_actor_override(:client_health)
Cyclium.Mode.clear_all_overrides()

# Inspect current state
Cyclium.Mode.status()
# %{node_mode: :full, overrides: %{client_health: :trigger_only}, node_identity: "..."}

Mode reads are ETS-backed (read_concurrency: true) for zero overhead in hot paths. The trigger request poller self-gates on each cycle — it only polls when the node-wide mode is :full.

Dynamic Actors

Dynamic actors allow agent definitions to be stored in the database and hydrated into running supervised processes at runtime — without requiring compiled Elixir modules.

When to use dynamic actors

• Users create custom monitoring agents through a UI
• Agent definitions are stored per-tenant
• Agent configurations change frequently without requiring code deploys

How it works

A single Cyclium.DynamicActor GenServer module serves all DB-defined actors. Each instance is started with different config/expectations args under Cyclium.ActorSupervisor. This avoids runtime module pollution — no Module.create/3 needed.

Cyclium.ActorSupervisor (DynamicSupervisor)
├── MyApp.Agents.CompiledActor    (compiled, use Cyclium.Actor)
├── Cyclium.DynamicActor          (from DB: "user_monitor_1")
└── Cyclium.DynamicActor          (from DB: "user_monitor_2")

Defining an agent in the database

Insert a row into cyclium_agent_definitions:

%Cyclium.Schemas.AgentDefinition{
  actor_id: "custom_health_check",
  domain: "monitoring",
  strategy_template: "observe_classify_converge",   # built-in template
  config: Jason.encode!(%{max_concurrent_episodes: 3, episode_overflow: "queue"}),
  expectations: Jason.encode!([
    %{
      id: "check_target",
      trigger: %{type: "schedule", interval_ms: 300_000},
      budget: %{max_turns: 5, max_tokens: 10_000, max_wall_ms: 60_000},
      log_strategy: "timeline"
    }
  ]),
  enabled: true
}

Loading dynamic actors

# At application startup — loads all enabled definitions
Cyclium.DynamicActor.Loader.load_all()

# Load a single actor
Cyclium.DynamicActor.Loader.load("custom_health_check")

# Reload after updating the definition in DB
Cyclium.DynamicActor.Loader.reload("custom_health_check")

# Stop a dynamic actor
Cyclium.DynamicActor.Loader.stop("custom_health_check")

Database table

cyclium_agent_definitions (V7 migration):

Strategy templates (data-driven strategies)

Dynamic actors use strategy templates — built-in parameterized strategy modules that are configured via strategy_config JSON in the DB. The compiled app defines what data sources (gatherers) and outputs are available; the DB definition composes them.

Gatherers

Gatherers are compiled modules that know how to collect domain-specific data. The compiled app implements and registers them:

defmodule MyApp.Gatherers.ProjectData do
  @behaviour Cyclium.Gatherer

  @impl true
  def gather(trigger_payload, _opts) do
    project_id = trigger_payload["project_id"]
    project = Repo.get!(Project, project_id)
    orders = load_orders(project_id)
    {:ok, %{project: project, orders: orders, order_count: length(orders)}}
  end
end

Register in app config:

config :cyclium, :gatherer_registry, %{
  "project_data" => MyApp.Gatherers.ProjectData,
  "client_metrics" => MyApp.Gatherers.ClientMetrics
}

Observe → Synthesize → Converge

The main workhorse. Gathers data, sends to LLM, maps result to findings:

%AgentDefinition{
  actor_id: "project_health_dynamic",
  strategy_template: "observe_synthesize_converge",
  strategy_config: Jason.encode!(%{
    "gatherer" => "project_data",
    "system_prompt" => "You are a project health analyst. Evaluate the project data and classify its health status.",
    "finding_config" => %{
      "actor_id_field" => "project_health_dynamic",
      "finding_key_template" => "project:health:${subject_id}",
      "class_field" => "class",
      "severity_field" => "severity",
      "summary_field" => "summary",
      "subject_kind" => "project",
      "subject_id_key" => "project_id"
    },
    "outputs" => ["email"]
  }),
  expectations: Jason.encode!([
    %{id: "evaluate", trigger: %{type: "event", event_type: "project.check_requested"}}
  ])
}

Observe → Classify → Converge

Rule-based classification without LLM. Rules are evaluated in order, first match wins:

%AgentDefinition{
  actor_id: "client_risk_monitor",
  strategy_template: "observe_classify_converge",
  strategy_config: Jason.encode!(%{
    "gatherer" => "client_metrics",
    "classify_rules" => [
      %{"field" => "mrr", "op" => "lt", "value" => 500, "class" => "at_risk", "severity" => "high"},
      %{"field" => "last_login_days_ago", "op" => "gt", "value" => 30, "class" => "inactive", "severity" => "medium"}
    ],
    "default_class" => "healthy",
    "default_severity" => "low",
    "finding_config" => %{
      "finding_key_template" => "client:risk:${subject_id}",
      "subject_kind" => "client",
      "subject_id_key" => "client_id"
    }
  })
}