Event-Sourced Agent Harnesses via Dynamic Stream Processors

Event Streams as Immutable Agent Logs

Event streams at events.iterate.com serve as the foundational primitive for agents, structured like a file system with hierarchical paths (e.g., /jonas-templestein/agent). Append events via POST requests with curl or SDK, each containing a type (optionally namespaced like https://events.iterate.com/...), payload, auto-incrementing offset, streamPath, and createdAt. Invalid events trigger error events prefixed with the service URL for documentation. Streams support idempotency keys to deduplicate, pausing/resuming via streamPaused/streamResumed events (with reason), scheduled events (e.g., heartbeat every 5s via scheduleRecurring), cancellations, and push subscriptions to external endpoints. Live streaming with ?live=true delivers SSE events indefinitely. Circuit breakers auto-pause after excessive appends (e.g., >100/s) to prevent infinite loops, requiring manual resume.

To interact: curl -N -X POST https://events.iterate.com/v1/events -d '{"type":"userMessage","payload":{"text":"hello"}}' -H 'X-Stream-Path: /your/path'. Pipe output through jq for formatting: | jq -r '.data[] | fromjson'. This log captures everything—user inputs, LLM chunks, tool calls, errors—ensuring full replayability without hidden state.

Pure State Derivation via Synchronous Reducers

Agents derive state purely from events using a reducer function: processor(state: any, event: Event): any. On startup or replay, fold over all events from offset 0 to compute current state without side effects, avoiding LLM re-calls on restart. Side effects (e.g., API calls, LLM invocations) occur only in an afterAppend hook: sideEffects(state: any, event: Event): Promise<void> | void, triggered post-reduce.

Example reducer initializes empty state {} and handles events like userMessage by accumulating messages:

function processor(state, event) {
  switch (event.type) {
    case 'userMessage':
      state.messages ??= [];
      state.messages.push({ role: 'user', content: event.payload.text });
      return state;
  }
}

Side effects check state: if last message unanswered and stream unpaused, invoke LLM (e.g., Anthropic), append assistantMessageChunk for tokens, toolCall, or error. Key principle: separation ensures idempotency—replays derive state fast, side effects idempotent or skipped.

Common mistake: mixing state and effects leads to non-replayable harnesses. Quality criteria: state must fully reconstruct from log; effects must not mutate external state unexpectedly. Prerequisites: familiarity with event sourcing (e.g., Redux reducers) and async JS.

Dynamic Processor Configuration for Zero-Deployment Agents

Transform any stream into an agent by appending a dynamicWorkerConfigured event with payload.processor as a JavaScript string. The service evaluates it safely (no eval, sandboxed) on every append, running the reducer and hook. No servers needed—processors run serverlessly on the edge.

Steps to create:

1. Append initial userMessage.
2. Define processor JS: export processor and sideEffects functions.
3. POST {"type":"dynamicWorkerConfigured","payload":{"processor":"<JS string>"}}.
4. Stream responds with agent behavior.

SDK simplifies: npm i ai-engineer-workshop, const client = createEventsClient({ baseUrl: 'https://events.iterate.com/v1' }); client.stream('/path', { live: true });. Full example in repo clones to local TS files for iteration.

Composability and Distributed Extensions

Multiple processors compose on one stream: append competing dynamicWorkerConfigured events; all run in sequence per append. Authors deploy independently (TS, Rust via custom clients), enabling plugins like JSON transformers (observe/rewrite events) or safety checkers injecting context pre-LLM (200ms window, non-blocking).

Example safety: processor appends safetyContext before llmCall. Infinite loops preempted by pausing. Broader workflow: edge-deployed (HTTP-only), public URLs for webhooks/Slack/human forms. Polyglot via simple HTTP API (OpenAPI spec available). Tensions: races/loops from distribution, mitigated by pausing/idempotency but not eliminated.

LLM Integration and Tooling Patterns

In sideEffects, query state for pending tasks: extract system/user prompts from messages, call LLM (e.g., anthropic.messages.create({ model: 'claude-3-5-sonnet-20240620', stream: true })). Stream tokens as assistantMessageChunk events. Handle tool_calls: append toolCallRequested, execute (idempotent), append toolCallResult.

Before/after: raw stream (user events only) → processor-configured (autonomous responses). Exercises: Add tools (e.g., calculator), heartbeat for polling, subscribe to external APIs. Fits in workflow post-prompt engineering: event log as debug UI, extensible beyond single-threaded frameworks like Pi/Claude.

"The split matters: when your program restarts after 100 events, you want to catch up state without replaying LLM requests."

"Processors from different authors on different servers can compose against the same stream."

"Everything that happens (streaming chunks, tool calls, errors, circuit breaker triggers) is an event in the log."

"I would like to do it purely event sourced aka debugable... everything that could possibly happen is in there."

"The moment an agent exists it should have a URL... otherwise you end up inventing a connector concept."

Key Takeaways

• Append events to public streams at events.iterate.com to log agent interactions immutably.
• Implement reducer for pure state from events; confine effects to post-reduce hooks for replayability.
• Activate agents via dynamicWorkerConfigured with JS processor string—no deploys needed.
• Compose multi-author extensions; use pausing/idempotency for loop prevention.
• Integrate LLMs by streaming chunks as events, handling tools idempotently.
• Debug via curl/SSE replays; extend with schedules, subscriptions, transformers.
• Avoid secrets in payloads; rotate post-workshop—no auth yet.
• Experiment locally via github.com/iterate/ai-engineer-workshop; polyglot via HTTP.
• Prioritize edge HTTP for internet-facing entities; public URLs enable webhooks/forms.