Make your own event-sourced agent harness using stream processors

An agent already keeps a history of messages and actions. Why should debugging it require finding a side effect that appears only in an OpenTelemetry trace? The starting problem for Jonas Templestein and Misha Kaletsky’s workshop is that an agent’s existing log often stops short of describing everything it does. Make every occurrence an event, and that log becomes the place to reconstruct what happened and decide what happens next. The Iterate team is exploring that idea with an experimental SDK pushed immediately before the session.

The next requirement is extensibility. Pi provides an example of how useful extensions can be when both people and agents can write them. Jonas wants independently developed extensions to compose without all running in one process on one computer. An agent should also receive a URL as soon as it exists: a Slack webhook or a human’s web-form submission can then become ordinary HTTP input. The workshop deployment has no authentication, so participants can inspect one another’s streams and agents. Event payloads must not contain secrets; exposed credentials need rotation.

Distribution follows from that interface. One processor might run on a server, another on a laptop; one can be written in Rust and another in TypeScript. They coordinate through events rather than shared process memory. That freedom also admits race conditions and feedback loops: two plugins can keep responding to one another forever. Those failure modes must become part of the harness design.

The primitive is a hosted event store at events.iterate.com. Its web interface displays raw events as YAML beneath hierarchical paths, much like files in a filesystem. Paths begin with /, and the service tolerates URL-encoded slashes. The workshop repository and its workshop instructions supply the cURL exercises. They target the live proof of concept; participants do not deploy a server of their own. A project slug, also passable through an HTTP header, selects a separate database namespace when participants need isolation from one another’s experiments.

Jonas appends Hello World to his stream. The returned envelope establishes the data model:

The server is the serialization point: it assigns increasing offsets as events enter the log. A processor can therefore refer to a definite position in the stream rather than infer order from message contents.

A second terminal observes /jonas-templestein/hello-world through server-sent events. With curl -N, cURL does not buffer the arriving output; with ?live=true, the connection remains open for subsequent appends. Another Hello World appears in that same subscription. The API draws inspiration from Durable Streams, but Iterate does not implement that specification exactly. Extracting SSE data elements with sed and formatting them with jq . makes the raw stream easier to inspect. If Misha were watching the input and appending replies, the transport would already support the response loop that an agent will later automate.

A permissive ingestion interface makes third-party webhooks easier to accept. Jonas’s premise is that an LLM can often interpret a webhook’s raw contents without a bespoke connector. When the submitted object lacks the required type, however, the server records an invalid-event error in the stream. Its own event types use documentation URLs as identifiers; callers can still use ordinary strings. An idempotency key handles a different problem: submitting the same key twice does not append the event twice.

Feedback loops require a stronger control. Jonas describes the workshop circuit breaker as pausing a stream at roughly more than 100 events per second. A pause can also be requested explicitly by appending a pause event. His first attempt needs a reason; adding reason: demo makes it succeed. A subsequent Hello World is rejected. That rejection does not append an error event, because doing so would let a runaway writer continue growing the supposedly paused log through errors. Processing can resume after an explicit resume event.

The same configuration-by-event pattern supports several other behaviors:

• Transformation: a JSONata transformer configuration tells a plugin which events to observe and how to rewrite them into new events.
• Scheduling: after resuming the paused stream, Jonas configures a heartbeat every five seconds. Scheduling can also target a future delay or a specific time, and a later event can cancel it.
• Subscriptions: an appended configuration can request delivery to a server or a Slack API endpoint. A pull subscriber opens an SSE connection; a push subscriber receives outbound notifications for all events or a filtered subset.

The stream now supplies inputs, timers, transformations, and notifications through one interface.

Existing harnesses are already event-driven in many respects. The proposed difference is the API boundary. OpenCode, Pi, and Claude expose extension interfaces in which streaming chunks or other operations may have separate treatment. Here, even a few letters of a partial completion are ordinary events. A processor reduces those events into state and occasionally performs a side effect. Iterate has no launched product at this point; this is an experiment in making that interface sufficient for a harness.

The hosted service is storage—closer to a database than to an LLM response pipe. Jonas compares the programmable experience to Convex: a stream could receive an event containing processor source code and run that processor on later events. To support this architecture, the infrastructure must also wake external programs and APIs when they have work. The wire interface is language-independent, with API documentation and an OpenAPI specification; TypeScript is simply the workshop’s chosen client language.

The quick start uses the repository’s small example file and published npm client. Misha pushes a client-side type fix while participants follow along, and additional processor examples are copied into the repository during the session. The repository screenshot shows the starting files, including 00-example.ts, README.md, package.json, and workshop.md. The linked repository has since evolved, so its current runtime names and commands should not be read as an exact snapshot of the code on screen.

The local response loop is small:

1. Install the dependencies with pnpm install.
2. Call createEventsClient with the server base URL and a path prefix.
3. Subscribe to a stream with live: true.
4. Ignore events whose type is not ping.
5. Use client.append to emit a pong for each relevant input.

The presenter’s initial Node attempt reports invalid TypeScript syntax, amid editor and terminal problems. Participants confirm ping/pong works, and after switching laptops the hello-world example runs successfully. That establishes the working baseline before the session attempts an agent processor using the OpenAI SDK.

The hello-world processor begins with a question: what state does this feature need? In this case, it needs the number of Hello World events encountered. A synchronous reduce function takes the current state and an event, then returns the next state. The processor abstraction can also treat no returned value as an ignored event. Side effects belong in afterAppend, outside the reducer.

That separation matters during recovery. Close a laptop, allow 100 events to accumulate, then restart the processor. Issuing an LLM request while reducing each historical event could trigger a burst of obsolete work. Instead, the runtime first catches up to the end of the available log and then lets the effect hook decide what the reconstructed state requires. Replay rebuilds knowledge; the effect hook acts on that knowledge.

The counting part can be expressed independently of the subscription machinery:

type StreamEvent = {
  type: string;
  payload?: unknown;
};

type CountState = {
  count: number;
};

function reduce(state: CountState, event: StreamEvent): CountState {
  if (event.type !== "hello-world") return state;
  return { count: state.count + 1 };
}

const events: StreamEvent[] = [
  { type: "hello-world" },
  { type: "ping" },
  { type: "hello-world" },
];

const state = events.reduce(reduce, { count: 0 });
// state.count is 2; reduction performs no external actions.

Appends performed by the effect hook need not target the same stream. Filesystem-like addressing allows a parent processor to append work to ./Boris, then subscribe to a result event from that child path. Boris becomes a sub-agent through a routing and response convention; its result can wake the parent.

An exported processor is inert until a runtime drives it. The pull-subscription runtime owns the stream-consumption boilerplate, invokes the reducer, and runs afterAppend. Live wiring runs into a mismatch between older workshop-one examples and the intended workshop-two files, some of which are not yet committed. The test at jonas/example fails; a machine-derived mmkal path is suggested as one possible cause. With dependencies still installing, Jonas switches to a walkthrough instead of establishing a successful run of the more complex processor.

The next feature starts from the desired action: when someone adds agent input, make an LLM request. The UI shortcut creates an agent-input-added event with a string payload. An OpenAI client’s openai.responses.create call needs a model, instructions, and input history, so those become the processor’s initial state fields.

A proposed model-change event would update the model field through a small reducer branch. Additional capabilities could introduce state such as compaction status or a tool call in progress. The walkthrough initializes an empty history, a system prompt, and GPT-5.4.

Next come the facts needed to derive that state. A Zod schema describes the string input event. Everything returned by OpenAI—including a small streaming fragment—is stored as another event. The log therefore retains more detail than the finished conversation view needs.

A last-minute rewrite by a coding agent has changed the displayed examples substantially, but the input transformation remains straightforward. The walkthrough uses a Zod matching helper credited to Misha for payload type safety. An agent input string becomes a Responses API history item with role: "user" and the string in content. This small reducer expresses that transformation for the input hi:

type HistoryItem = {
  role: "user";
  content: string;
};

type AgentState = {
  model: string;
  systemPrompt: string;
  history: HistoryItem[];
};

type InputAdded = {
  type: "agent-input-added";
  payload: string;
};

function addInput(state: AgentState, event: InputAdded): AgentState {
  return {
    ...state,
    history: [
      ...state.history,
      { role: "user", content: event.payload },
    ],
  };
}

const initial: AgentState = {
  model: "gpt-5.4",
  systemPrompt: "Respond helpfully.",
  history: [],
};

const next = addInput(initial, {
  type: "agent-input-added",
  payload: "hi",
});

The model and prompt stay unchanged; the history gains one user item. Making the request remains a separate side effect.

Response handling is less settled in the live example. Streaming produces many events, followed by an end event, but the walkthrough does not finish the completion-handling branch or establish its exact event identifier. The screenshot retains the visible TODO. Workshop two is pushed largely as-is; Misha subsequently encounters insufficient quota, and the full live agent build is abandoned. The working baseline remains ping/pong, while the architecture still permits processors to run on separate servers or in dynamic workers.

An audience question separates two concepts that are easy to conflate: are streaming chunks persisted, or does a client materialize them first? Every chunk enters the event stream. Reducers can then run in many places and build the state each consumer needs. Persisting the raw facts does not require every derived view to become another persisted event.

The service’s circuit breaker provides a concrete example. Jonas shows a processor he describes as running in the live service, initialized with paused: false and a null pause reason. Its reducer retains the last 100 event timestamps; its effect hook trips the breaker when that window fits within approximately one second. The reducer accumulates evidence, and afterAppend acts on it.

The UI follows the same pattern. It projects selected events into more readable feed items. A feed item is derived state, not the underlying event itself. Because a reducer is a synchronous function, another processor can import and reuse it, then build additional behavior on the same event types. Jonas extends this idea to Pi, Claude, and OpenCode: their events could enter the shared log, with other processors interpreting or extending them. He also describes source-code events as a route to automatically hosted, horizontally scalable processing and promises a fuller demonstration later.

The dynamic-worker demonstration makes the deployment mechanism visible. Posting to an unused path implicitly creates a stream. In the UI, Jonas chooses a Dynamic Worker Configured preset containing a script string; that string holds a reducer and an after-append hook. Appending the configuration installs the ping-pong worker in the stream.

He then submits ping, and the stream produces pong. The stream began with no response behavior; an event containing JavaScript supplied it. This is the successful live code-as-event demonstration. Processor files can also be bundled into an event rather than entered directly into the UI. Jonas reports having run an AI agent this way previously, but the full AI version is not demonstrated here.

Credentials introduce a boundary the event log should not cross. Jonas describes keeping the OpenAI key outside the stream in environment variables, then substituting the secret into the header of an outgoing fetch request. He characterizes a basic agent as roughly 40 lines that can be appended to turn a stream into an agent. Directly entered scripts cannot depend on separately installed npm packages: those dependencies must be bundled into the script string.

A URL-safe slug identifies the worker configuration, allowing a later configuration event to replace it. That also gives an agent a path to self-extension: append different JavaScript under the configuration identifier. Jonas proposes a further layer in which an unbundled-worker event carries a package.json field and a script field. A separate bundling processor would turn that input into the complete executable configuration event. Such a workflow could resemble an IDE built from cooperating processors, though that bundling extension remains a proposal.

What happens when the appended JavaScript is wrong? The current evaluator runs JavaScript without TypeScript checking; Jonas says execution failures should produce error events. A proposed compilation stage could emit TypeScript diagnostics in the same way, after which an LLM could consume the error and append corrected code. The settled interface shown during this discussion contains the configured ping-pong worker, ping and pong entries, and an error card reading Failed to fetch remote state. That card illustrates an error appearing in the stream, without establishing a completed compiler-and-repair loop.

For hosted execution, this implementation uses Cloudflare Dynamic Workers. The processor can move through three execution arrangements:

A developer can experiment locally, then host the processor on Cloudflare Workers, Vercel, or another service, or package it into the dynamic-worker configuration. The event interface stays central while the execution location changes.

A remote prompt-injection checker illustrates why push notifications matter. Rather than require the model to explicitly call every plugin, the stream could notify an independently operated service that a new request is being prepared. Jonas proposes waiting up to 200 milliseconds before an LLM request for a checker’s contribution. His MCP comparison concerns the proactive agent-loop integration he wanted at the time, rather than all forms of server-initiated interaction. The checker could run elsewhere and charge for its service, initially billing the human.

The Claude Code plugin example makes the integration burden concrete: a service may otherwise need instructions prompting the model to call an MCP tool, or an authenticated CLI that the model must invoke. Event-stream participation gives the service another route. It observes that the agent is about to act and appends information for the loop to consume.

There is a deliberate limit: third-party processors have no before-append hook. Built-in processors can prevent an event from entering the stream, which is why a third party cannot enforce the same pause behavior. Jonas prefers this boundary to broadly available before hooks, citing OpenClaw caching, performance, and cost problems he attributes to them.

The alternative is an eventually consistent loop with a bounded contribution window. Once the wait expires, the LLM request proceeds even if a plugin has not responded. A processor holding an indexed Notion knowledge base could opportunistically add relevant retrieval context; arriving late would not hold the agent indefinitely. This is a useful resilience policy for optional context. Applied to the earlier safety-checker example, however, it does not guarantee that a safety check finishes before the request proceeds.

The architecture does not settle whether everyone should run their own event service or share a central one. Jonas expects it could use other infrastructure shaped like a durable stream. The desired primitive combines an append-only database, a queue, pub/sub delivery, and code execution. Electric’s Durable Streams and NATS JetStream provide reference points, with push subscriptions especially useful for waking processors.

Offset ownership determines how a consumer resumes:

Jonas uses Durable Streams and Kafka to illustrate client-controlled consumption, then contrasts Google Cloud and AWS pub/sub services that manage delivery position for subscribers. These are different responsibilities around an ordered log, not interchangeable delivery guarantees.

The final audience example exposes the consequence of making agents publicly reachable. Suppose an images child path forwards requests to a local image-generating model and returns results. Anyone able to append could consume that machine’s resources. Asked how to prevent flooding, Jonas says authentication is necessary and agrees that this should not yet be exposed that way. A disposable public namespace cleared periodically is floated as an experimental convenience, not a protection for valuable computation.

A more complete design would attach client provenance to each event: who submitted it, what capabilities they had, and the basis for believing they were authorized. Access-policy changes could themselves be events. Jonas proposes creator-only writes by default, followed by an explicit event making the stream public. Those controls are future work, but they identify the boundary the public demo leaves open: a stream can make useful computation easy to reach only if the service also decides who is entitled to trigger it.