Cyclium.EpisodeRunner.Strategy behaviour (Cyclium v0.1.15)

Behaviour for episode strategies. A strategy defines the investigation logic for a specific expectation — how to gather data, classify, and produce outputs.

Summary

Callbacks

converge(state, episode_ctx)

handle_budget_exhausted(state, episode_ctx)

Optional hook invoked when the turn/token budget is exhausted between steps.

handle_result(state, step, result)

init(episode, trigger)

next_step(state, episode_ctx)

resume_from_block(state, episode)

Optional hook to resume a :blocked episode from the journal instead of a state checkpoint.

workflow_result(state, converge_result)

Callbacks

converge(state, episode_ctx)

@callback converge(state :: map(), episode_ctx :: map()) ::
  {:ok, Cyclium.ConvergeResult.t()}
  | {:partial, Cyclium.ConvergeResult.t(), failures :: [term()]}

handle_budget_exhausted(state, episode_ctx)

(optional)

@callback handle_budget_exhausted(state :: map(), episode_ctx :: map()) ::
  {:converge, new_state :: map()} | :fail

Optional hook invoked when the turn/token budget is exhausted between steps.

Lets a strategy convert a hard budget failure into a graceful convergence — e.g. an interactive actor can emit a "ran out of budget for this turn" summary instead of leaving the episode :failed with no user-facing message.

Return {:converge, new_state} to run the normal converge path with the given state, or :fail to keep the default behaviour (episode fails with error_class: "budget_exceeded"). Strategies that don't implement this callback always fail. This is not called for wall-time exhaustion (max_wall_ms), which can fire mid-step and is always a hard failure.

handle_result(state, step, result)

@callback handle_result(
  state :: map(),
  step :: %Cyclium.Schemas.EpisodeStep{
    __meta__: term(),
    args_hash: term(),
    args_redacted: term(),
    cost_ms: term(),
    cost_tokens: term(),
    created_at: term(),
    episode_id: term(),
    error_class: term(),
    error_detail: term(),
    id: term(),
    kind: term(),
    metadata: term(),
    result_ref: term(),
    side_effect_key: term(),
    step_no: term(),
    tool_name: term()
  },
  result :: term()
) ::
  {:ok, new_state :: map()}
  | {:retry, new_state :: map()}
  | {:abort, reason :: term()}

init(episode, trigger)

@callback init(
  episode :: %Cyclium.Schemas.Episode{
    __meta__: term(),
    actor_id: term(),
    archived_at: term(),
    attempts: term(),
    budget: term(),
    checkpoints: term(),
    classification: term(),
    confidence: term(),
    conversation_id: term(),
    dedupe_key: term(),
    dry_run_opts: term(),
    error_class: term(),
    error_detail: term(),
    expectation_id: term(),
    finished_at: term(),
    id: term(),
    log_strategy: term(),
    max_attempts: term(),
    metadata: term(),
    mode: term(),
    outputs: term(),
    parent_episode_id: term(),
    phase: term(),
    queued_at: term(),
    source_env: term(),
    source_stack: term(),
    spec_rev: term(),
    started_at: term(),
    status: term(),
    steps: term(),
    summary: term(),
    tokens_used: term(),
    trigger_ref: term(),
    trigger_type: term(),
    turns_used: term(),
    workflow_instance_id: term(),
    workflow_step_id: term(),
    workflow_step_no: term()
  },
  trigger :: Cyclium.Trigger.t()
) :: {:ok, state :: map()}

next_step(state, episode_ctx)

@callback next_step(state :: map(), episode_ctx :: map()) ::
  {:tool_call, atom(), atom(), map()}
  | {:synthesize, map()}
  | {:observe, map()}
  | {:output, atom(), map()}
  | {:checkpoint, binary()}
  | {:approval, map()}
  | {:wait, map()}
  | :converge
  | :done

resume_from_block(state, episode)

(optional)

@callback resume_from_block(
  state :: map(),
  episode :: %Cyclium.Schemas.Episode{
    __meta__: term(),
    actor_id: term(),
    archived_at: term(),
    attempts: term(),
    budget: term(),
    checkpoints: term(),
    classification: term(),
    confidence: term(),
    conversation_id: term(),
    dedupe_key: term(),
    dry_run_opts: term(),
    error_class: term(),
    error_detail: term(),
    expectation_id: term(),
    finished_at: term(),
    id: term(),
    log_strategy: term(),
    max_attempts: term(),
    metadata: term(),
    mode: term(),
    outputs: term(),
    parent_episode_id: term(),
    phase: term(),
    queued_at: term(),
    source_env: term(),
    source_stack: term(),
    spec_rev: term(),
    started_at: term(),
    status: term(),
    steps: term(),
    summary: term(),
    tokens_used: term(),
    trigger_ref: term(),
    trigger_type: term(),
    turns_used: term(),
    workflow_instance_id: term(),
    workflow_step_id: term(),
    workflow_step_no: term()
  }
) :: {:ok, new_state :: map()}

Optional hook to resume a :blocked episode from the journal instead of a state checkpoint.

When a strategy implements this, the runner does not write a "__blocked__" state checkpoint at an approval/wait block — avoiding an extra per-block DB row (and the need to JSON-serialize an arbitrary strategy state). On resume, EpisodeTask rebuilds a fresh state via init/2 and then calls this hook so the strategy can reconstruct where it left off from already- journaled steps (e.g. the approval_requested plan + approval_resolved decision) and return a state positioned to continue (e.g. phase: :execute).

Return {:ok, state}; return the given state unchanged if there's nothing to resume. Strategies that don't implement this keep the checkpoint-based resume.

workflow_result(state, converge_result)

(optional)

@callback workflow_result(state :: map(), converge_result :: Cyclium.ConvergeResult.t()) ::
  map()