WardrikEvents (wardrik_events v0.1.0-rc.1)
Copy MarkdownKeeps each agent's events as CloudEvents, in a log you can read back and follow.
children = [
{WardrikStore, store},
{WardrikEvents, name: MyApp.Events, store: store, deployment: "acme-prod"}
]
WardrikCore.start_agent(
agent_id: "acme/triage",
model: "anthropic:claude-haiku-4-5",
events: {WardrikEvents, MyApp.Events}
)
{:ok, stream} = WardrikEvents.stream(MyApp.Events, ["acme/*"])
Enum.each(stream, fn {event, cursor} -> save(event, WardrikEvents.Cursor.encode(cursor)) end)Each agent writes one log, named by its id. Functions take logs as patterns, where * matches
any run of characters and ? matches one. See WardrikEvents.Pattern.
Functions that take the events' name raise ArgumentError if they aren't started.
Summary
Functions
Returns what the agent writing log is doing now, as the protocol's activity.
Writes an event into the log of the agent agent_id.
Returns a page of the audit, oldest first.
Returns a specification to start this module under a supervisor.
Returns the cursor at the end of the logs resources match. Pass it as :after to read only newer events.
Returns a page of kept events from the logs resources match, oldest first.
Records a change someone made in the audit.
Starts the events.
Streams the events of the logs resources match, as {event, cursor}.
Subscribes the calling process to the logs resources match.
Drops the events and audit changes written before cutoff. Retention calls this hourly.
Unsubscribes the calling process from resources.
Folds every event log keeps into view. Returns the view's state and the cursor after the last
event.
Functions
Returns what the agent writing log is doing now, as the protocol's activity.
WardrikEvents.activity(MyApp.Events, "acme/triage")
#=> %{"doing" => [...], "since" => "2026-10-22T09:00:03.000000Z", "pending" => 1}Writes an event into the log of the agent agent_id.
data = %{"trigger" => %{"name" => "daily"}, "status" => "asked"}
:ok = WardrikEvents.append(MyApp.Events, "acme/triage", {"trigger.fired", "triggers/daily", data})The event is {type, subject, data}, with a type from WardrikProtocol. It's handled like the
agent's own events: kept, sent to subscribers and folded into its activity.
Options
:span- theWardrikCore.Spanthe event happened in.
@spec audit(atom(), keyword()) :: %{events: [map()], next: non_neg_integer()}
Returns a page of the audit, oldest first.
%{events: changes, next: next} = WardrikEvents.audit(MyApp.Events, limit: 100)
%{events: more} = WardrikEvents.audit(MyApp.Events, after: next)next is the sequence number of the last change returned.
Options
:after- the sequence number to read after.0unless set.:limit- the most changes to return.100unless set, and never more than500.
Returns a specification to start this module under a supervisor.
See Supervisor.
Returns the cursor at the end of the logs resources match. Pass it as :after to read only newer events.
@spec page(atom(), [String.t()], keyword()) :: {:ok, %{events: [map()], next: String.t()}} | {:error, :invalid_cursor}
Returns a page of kept events from the logs resources match, oldest first.
{:ok, %{events: events, next: next}} = WardrikEvents.page(MyApp.Events, ["acme/*"], limit: 100)
{:ok, %{events: more}} = WardrikEvents.page(MyApp.Events, ["acme/*"], after: next)next is the cursor after the page. Returns {:error, :invalid_cursor} if :after can't be
decoded.
If retention dropped events past the cursor, the page starts with a wardrik.events.dropped.
Options
:after- the cursor to read after. The start of the logs unless set.:limit- the most events to return.100unless set, and never more than500.
Records a change someone made in the audit.
change = {"agent.forgotten", "agents/acme/triage", %{"agent" => %{"id" => "acme/triage"}}}
:ok = WardrikEvents.record(MyApp.Events, change, actor: principal)change is {type, subject, data}. The audit is one log for the deployment, kept apart from the
agents' logs.
Options
:actor- theWardrikCore.Principalwho made the change.:assumed_by- the principals who acted for the actor, first to last.[]unless set.:span- theWardrikCore.Spanthe change happened in.
@spec start_link(keyword()) :: Supervisor.on_start()
Starts the events.
WardrikEvents.start_link(name: MyApp.Events, store: store, deployment: "acme-prod")Options
:name- required. What the events are reached by.:store- required. TheWardrikStoreto keep the logs in.:deployment- the deployment's name in each event'ssource."local"unless set.:retention- days events are kept, or:infinity.30unless set. Older events are dropped hourly.
@spec stream(atom(), [String.t()], keyword()) :: {:ok, Enumerable.t({map(), WardrikEvents.Cursor.t()})} | {:error, :invalid_cursor}
Streams the events of the logs resources match, as {event, cursor}.
{:ok, stream} = WardrikEvents.stream(MyApp.Events, ["acme/*"], after: saved)
Enum.each(stream, fn {event, cursor} ->
show(event)
save(WardrikEvents.Cursor.encode(cursor))
end)The stream reads back kept events, then follows new ones, live events too. It never skips an event or repeats one between the two. It runs until the caller stops taking from it.
Returns {:error, :invalid_cursor} if :after can't be decoded.
Options
:after- the cursor to read after, asWardrikEvents.Cursor.encode/1returns it. Only events written from now on unless set.:types- the event types to take, as["wardrik.run.completed"]. Every type unless set. Awardrik.events.droppedalways comes through.
Subscribes the calling process to the logs resources match.
The process is sent {WardrikEvents, log, event} for each event, live or kept.
@spec trim(atom(), DateTime.t()) :: :ok
Drops the events and audit changes written before cutoff. Retention calls this hourly.
Unsubscribes the calling process from resources.
Folds every event log keeps into view. Returns the view's state and the cursor after the last
event.
{timeline, cursor} = WardrikEvents.view(MyApp.Events, "acme/triage", WardrikEvents.Timeline)view is a WardrikEvents.View. opts are passed to its WardrikEvents.View.init/1.