Skuld.Effects.EffectLogger (skuld v0.1.26)
View SourceEffect logging for replay, resume, and rerun capabilities.
EffectLogger captures effect invocations in a flat log, enabling:
- Replay: Short-circuit completed effects with logged values
- Resume: Continue suspended computations from where they left off
- Rerun: Re-execute failed computations, replaying successful effects
Architecture
EffectLogger works by wrapping effect handlers to intercept their invocations. Each wrapped handler:
- Creates a log entry in
:startedstate - Installs a
leave_scopeto mark the entry as:discardedif the continuation is abandoned (e.g., by a Throw effect) - Wraps the continuation
kto mark the entry as:executedwhen it completes - Calls the original handler with the wrapped continuation
Flat Log Structure
Unlike a tree-structured log, entries are stored flat in execution order.
The leave_scope mechanism marks entries as :discarded when their
continuations are abandoned, preserving the information needed for replay.
Example - Basic Logging
alias Skuld.Comp
alias Skuld.Effects.{EffectLogger, State}
{{result, log}, _env} =
my_comp
|> EffectLogger.with_logging()
|> State.with_handler(0)
|> Comp.run()Entry States
:started- Handler invoked, continuation not yet completed (includes suspended):executed- Handler called wrapped_k, continuation completed normally:discarded- Handler never called wrapped_k (e.g., Throw effect)
Hot vs Cold Resume
When a computation suspends (e.g., via Yield), there are two ways to resume:
Hot Resume (In-Memory)
If you have the live %Comp.ExternalSuspend{} struct, call the resume function directly:
alias Skuld.Effects.{EffectLogger, State, Yield}
# Run until suspension
{%Comp.ExternalSuspend{value: yielded, resume: resume}, env} =
my_comp
|> EffectLogger.with_logging()
|> Yield.with_handler()
|> State.with_handler(0)
|> Comp.run()
# Hot resume - call continuation directly with input
{result, final_env} = resume.(my_input)The resume function captures the live continuation and environment.
Cold Resume (Deserialized)
If you've serialized the log and later deserialize it, use with_resume/3,4:
alias Skuld.Effects.EffectLogger.Log
# Serialize log to JSON
json = Jason.encode!(log)
# Later... deserialize and cold resume
cold_log = json |> Jason.decode!() |> Log.from_json()
{{result, new_log}, _env} =
my_comp
|> EffectLogger.with_resume(cold_log, my_input)
|> Yield.with_handler()
|> State.with_handler(0)
|> Comp.run()Cold resume re-runs the computation, short-circuits :executed entries with
their logged values, and injects the resume value at the :started Yield entry.
Use Cases
- Logging/Tracing: Capture effect invocations for debugging or audit
- Replay: Re-run computation, short-circuiting with logged values
- Retry: Re-run after failure -
:discardedentries are re-executed - Resume: Continue after Yield suspension (hot or cold)
Loop Pruning with mark_loop
For long-running loop-based computations (like LLM conversation loops), the log
can grow unboundedly. The mark_loop/1 operation enables efficient pruning of
completed loop iterations while preserving state checkpoints for cold resume.
Basic Usage
alias Skuld.Effects.{EffectLogger, Yield}
defcomp conversation_loop(state) do
# Mark iteration start - env.state is captured automatically
_ <- EffectLogger.mark_loop(ConversationLoop)
input <- Yield.yield(:await_input)
state = handle_input(state, input)
conversation_loop(state)
end
# Run with pruning enabled - no extra handler needed
{{result, log}, _env} =
conversation_loop(initial_state)
|> EffectLogger.with_logging(prune_loops: true)
|> Yield.with_handler()
|> Comp.run()How It Works
- A root mark (
:__root__) is auto-inserted on the first intercepted effect, capturing the initialenv.state - Each
EffectLogger.mark_loop(loop_id)captures currentenv.state - When
prune_loops: true, completed iterations are removed on finalization - Only the last iteration's effects are retained, plus checkpoints
- Cold resume restores
env.statefrom the most recent checkpoint
Nested Loops
For nested loops with different loop-ids, pruning respects the hierarchy:
defcomp outer_loop(state) do
_ <- EffectLogger.mark_loop(OuterLoop)
{result, state} <- inner_loop(state)
outer_loop(state)
end
defcomp inner_loop(state) do
_ <- EffectLogger.mark_loop(InnerLoop)
# ... inner loop logic ...
inner_loop(updated_state)
endThe hierarchy :__root__ <- OuterLoop <- InnerLoop is inferred from nesting
order. Pruning InnerLoop only removes entries between inner loop marks,
preserving outer loop structure. The root mark is never pruned.
Benefits
- Bounded log size: O(current_iteration) instead of O(total_iterations)
- Fast cold resume: Replay only current iteration, not entire history
- State restoration:
env.stateis restored from checkpoint on cold resume - Preserved semantics: Full logging within each iteration
See EffectLogger.Log and EffectLogger.EffectLogEntry for details.
Summary
Functions
Install EffectLogger via catch clause syntax.
Get the current log from the environment
Handle EffectLogger operations.
Mark the start of a loop iteration with a loop identifier.
Put a log into the environment
Returns the root loop ID used for the implicit root mark.
Update the log in the environment
Wrap a computation with effect logging.
Resume a suspended computation from a cold (deserialized) log.
Wrap an effect handler with logging.
Wrap a handler for replay mode.
Wrap a handler for resume mode.
Functions
Install EffectLogger via catch clause syntax.
Config is opts, a Log for replay, or {log, opts}:
catch
EffectLogger -> nil # fresh logging
EffectLogger -> [effects: [State]] # log specific effects
EffectLogger -> log # replay from log
EffectLogger -> {log, allow_divergence: true} # replay with opts
@spec get_log(Skuld.Comp.Types.env()) :: Skuld.Effects.EffectLogger.Log.t() | nil
Get the current log from the environment
@spec handle(term(), Skuld.Comp.Types.env(), Skuld.Comp.Types.k()) :: {Skuld.Comp.Types.result(), Skuld.Comp.Types.env()}
Handle EffectLogger operations.
Currently handles:
MarkLoop- Records loop iteration boundary, returns:ok
@spec mark_loop(atom()) :: Skuld.Comp.Types.computation()
Mark the start of a loop iteration with a loop identifier.
This creates a boundary in the effect log that enables pruning of completed
loop iterations. The current env.state is automatically captured for
cold resume. Only meaningful when used with with_logging(prune_loops: true).
Parameters
loop_id- Atom identifying which loop this mark belongs to (module atoms work well)
Example
defcomp conversation_loop(state) do
# Mark iteration start - env.state captured automatically
_ <- EffectLogger.mark_loop(ConversationLoop)
input <- Yield.yield(:await_input)
state = handle_input(state, input)
conversation_loop(state)
end
# Run with pruning - only last iteration kept in log
{{result, log}, _env} =
conversation_loop(initial_state)
|> EffectLogger.with_logging(prune_loops: true)
|> Yield.with_handler()
|> Comp.run()
@spec put_log(Skuld.Comp.Types.env(), Skuld.Effects.EffectLogger.Log.t()) :: Skuld.Comp.Types.env()
Put a log into the environment
@spec root_loop_id() :: atom()
Returns the root loop ID used for the implicit root mark.
@spec update_log(Skuld.Comp.Types.env(), (Skuld.Effects.EffectLogger.Log.t() -> Skuld.Effects.EffectLogger.Log.t())) :: Skuld.Comp.Types.env()
Update the log in the environment
@spec with_logging( Skuld.Comp.Types.computation(), keyword() ) :: Skuld.Comp.Types.computation()
@spec with_logging(Skuld.Comp.Types.computation(), Skuld.Effects.EffectLogger.Log.t()) :: Skuld.Comp.Types.computation()
Wrap a computation with effect logging.
Automatically wraps handlers already installed in the env with logging.
Initializes the log and extracts it when the computation completes.
The result is transformed to {original_result, log}.
This should be the INNERMOST wrapper (immediately after the computation in the pipe), so it can see handlers installed by outer wrappers.
Variants
with_logging(comp)- Fresh logging, capture all effectswith_logging(comp, opts)- Fresh logging with optionswith_logging(comp, log)- Replay from existing logwith_logging(comp, log, opts)- Replay with options
Options
:effects- List of effect signatures to log. Default is all handlers in env.:allow_divergence- (replay only) If true, allow effects that don't match the log.:prune_loops- If true (default), prune completed loop segments eagerly after eachmark_loop/1call. This keeps memory bounded during long-running computations. Set tofalseto preserve all entries (e.g., for debugging).:state_keys- List of env.state keys to include inEnvStateSnapshotcaptures. Default:allincludes everything. Use this to filter out constant Reader state:state_keys: [{Skuld.Effects.State, MyApp.Counter}]This only captures the specified State keys, excluding Reader/other constant state.
:decorate_suspend- If true (default), attach the current finalized log toExternalSuspend.data[EffectLogger]when yielding. This allows AsyncComputation callers to access the log without needing the full env. Set tofalseto disable.
Example - Fresh Logging
{{result, log}, _env} =
my_comp
|> EffectLogger.with_logging()
|> State.with_handler(0)
|> Comp.run()
# Or specify which effects to log:
{{result, log}, _env} =
my_comp
|> EffectLogger.with_logging(effects: [State])
|> State.with_handler(0)
|> Reader.with_handler(:config)
|> Comp.run()Example - Replay
# First run - capture log
{{result1, log}, _} =
my_comp
|> EffectLogger.with_logging()
|> State.with_handler(0)
|> Comp.run()
# Replay - short-circuit with logged values
{{result2, _}, _} =
my_comp
|> EffectLogger.with_logging(log)
|> State.with_handler(0)
|> Comp.run()
assert result1 == result2
@spec with_logging( Skuld.Comp.Types.computation(), Skuld.Effects.EffectLogger.Log.t(), keyword() ) :: Skuld.Comp.Types.computation()
@spec with_resume( Skuld.Comp.Types.computation(), Skuld.Effects.EffectLogger.Log.t(), term() ) :: Skuld.Comp.Types.computation()
Resume a suspended computation from a cold (deserialized) log.
Re-runs the computation from scratch, short-circuiting completed effects with
their logged values. When reaching the suspension point (the :started Yield
entry), injects resume_value instead of actually suspending, then continues
with fresh execution.
Parameters
comp- The computation to runlog- The log ending with a:startedYield entry (from suspension)resume_value- The value to inject at the suspension pointopts- Options (same aswith_logging/3)
Example
alias Skuld.Effects.EffectLogger.Log
# Original run - suspended
{{%Comp.ExternalSuspend{value: yielded}, log}, _} =
my_comp
|> EffectLogger.with_logging()
|> Yield.with_handler()
|> State.with_handler(0)
|> Comp.run()
# Serialize and later deserialize
json = Log.to_json(log)
{:ok, cold_log} = Log.from_json(json)
# Cold resume with injected value
{{result, new_log}, _} =
my_comp
|> EffectLogger.with_resume(cold_log, :my_input)
|> Yield.with_handler()
|> State.with_handler(0)
|> Comp.run()
@spec with_resume( Skuld.Comp.Types.computation(), Skuld.Effects.EffectLogger.Log.t(), term(), keyword() ) :: Skuld.Comp.Types.computation()
@spec wrap_handler(module(), Skuld.Comp.Types.handler()) :: Skuld.Comp.Types.handler()
Wrap an effect handler with logging.
The wrapped handler will log each effect invocation, tracking:
- When the effect starts (creates entry in
:startedstate) - When the effect completes (marks entry as
:executedwith value) - When the continuation is abandoned (marks entry as
:discarded)
Parameters
sig- The effect signature (module) for logginghandler- The original handler function(args, env, k) -> {result, env}
Returns
A wrapped handler function with the same signature.
Example
logged_handler = EffectLogger.wrap_handler(State, &State.handle/3)
comp
|> Comp.with_handler(State, logged_handler)
|> EffectLogger.with_logging()
|> Comp.run()
@spec wrap_replay_handler(module(), Skuld.Comp.Types.handler()) :: Skuld.Comp.Types.handler()
Wrap a handler for replay mode.
During replay, if the next entry in the log queue matches this effect and can be short-circuited, returns the logged value directly. Otherwise, executes the handler normally.
Parameters
sig- The effect signature (module)handler- The original handler function
Example
replay_handler = EffectLogger.wrap_replay_handler(State, &State.handle/3)
{{result, _log}, _env} =
my_comp
|> Comp.with_handler(State, replay_handler)
|> EffectLogger.with_logging(previous_log)
|> Comp.run()
@spec wrap_resume_handler(module(), Skuld.Comp.Types.handler()) :: Skuld.Comp.Types.handler()
Wrap a handler for resume mode.
Like wrap_replay_handler/2, but when encountering a :started Yield entry
(the suspension point), injects the stored resume value instead of suspending.
The resume value is stored in env by with_resume/4 and cleared after use.
Parameters
sig- The effect signature (module)handler- The original handler function
Returns
A wrapped handler function.