Restate + OpenAI: durably stream structured JSON (JSONL) agent events. Each line is parsed into a typed AgentEvent and journaled via durableSource, so replay never re-calls the model.

app.ts

import {
  Operation,
  object,
  run,
} from "@restatedev/restate-sdk-gen";
import { serve } from "@restatedev/restate-sdk";
import OpenAI from "openai";
import * as utils from "./utils";

const openai = new OpenAI(); // reads OPENAI_API_KEY from the environment

// The structured events the agent streams. Each is a complete JSON object the
// model emits as it works, parsed and yielded one at a time as the completion
// streams in.
type AgentEvent =
  | { type: "reasoning"; text: string }
  | { type: "tool_call"; tool: string; args: Record<string, unknown> }
  | { type: "message"; text: string };

// Ask for a stream of JSON objects rather than a single JSON document, so the
// stream stays incremental: a complete object is ready as soon as its closing
// brace arrives, instead of only once the whole response is done.
const SYSTEM = [
  "You are an agent that streams its work as a sequence of JSON objects.",
  "Emit one compact JSON object per step. No prose, no markdown fences.",
  "Each object must be one of:",
  '{"type":"reasoning","text":"..."}\n',
  '{"type":"tool_call","tool":"...","args":{...}}\n',
  '{"type":"message","text":"..."}',
].join(" ");

// Pull every *complete* top-level JSON object out of `buf`, returning the parsed
// events and the not-yet-complete remainder. We track brace depth (respecting
// strings/escapes) instead of trusting a delimiter — the model is not reliable
// about putting exactly one object per line, so objects can arrive separated by
// newlines, by spaces, by nothing at all, or wrapped in an array/fence. Anything
// outside a top-level {...} (commas, brackets, ```json fences, whitespace) is
// simply ignored.
function drainObjects(buf: string): { events: AgentEvent[]; rest: string } {
  const events: AgentEvent[] = [];
  let depth = 0;
  let inString = false;
  let escape = false;
  let start = -1;

  for (let i = 0; i < buf.length; i++) {
    const ch = buf[i];
    if (inString) {
      if (escape) escape = false;
      else if (ch === "\\") escape = true;
      else if (ch === '"') inString = false;
      continue;
    }
    if (ch === '"') inString = true;
    else if (ch === "{") {
      if (depth === 0) start = i;
      depth++;
    } else if (ch === "}" && depth > 0) {
      depth--;
      if (depth === 0) {
        events.push(JSON.parse(buf.slice(start, i + 1)) as AgentEvent);
        start = -1;
      }
    }
  }
  // Keep from the start of an in-progress object; otherwise we've consumed it all.
  return { events, rest: start === -1 ? "" : buf.slice(start) };
}

// A real streaming completion. OpenAI hands us an async iterable of small token
// deltas when stream:true; we accumulate them and emit each JSON object the
// moment it's complete, as a typed AgentEvent.
//
// This is a non-replayable source: a re-run would call the model again and get
// different output. That's exactly why it's consumed through durableSource —
// the events we've already pulled are journaled and replayed verbatim.
async function* prompt(input: string): AsyncGenerator<AgentEvent> {
  const stream = await openai.chat.completions.create({
    model: "gpt-5",
    stream: true,
    messages: [
      { role: "system", content: SYSTEM },
      { role: "user", content: input },
    ],
  });

  let buffer = "";
  for await (const part of stream) {
    buffer += part.choices[0]?.delta?.content ?? "";
    const { events, rest } = drainObjects(buffer);
    buffer = rest;
    for (const event of events) yield event;
  }
  // Drain any final complete object left in the buffer.
  for (const event of drainObjects(buffer).events) yield event;
}

// A virtual object so each agent instance has its own durable identity. The
// handler streams an LLM completion and processes each structured event durably.
const agent = object({
  name: "agent",
  handlers: {
    *run(): Operation<void> {
      // Wrap the live, non-replayable OpenAI stream so it becomes replay-safe.
      // durableSource is generic, so it yields typed AgentEvents here.
      const chunks = yield* utils.durableSource(() =>
        prompt("What is the weather in NYC? Reason briefly, then answer."),
      );

      while (true) {
        const res = yield* chunks.next();
        if (res.type === 'done' || res.type === 'aborted') {
          break;
        }
        const event = res.value;
        // Act on each structured event inside `run` so the side effect is
        // journaled and not repeated on replay.
        yield* run(
          async () => {
            switch (event.type) {
              case "reasoning":
                console.log("🧠", event.text);
                break;
              case "tool_call":
                console.log("🔧", event.tool, event.args);
                break;
              case "message":
                console.log("💬", event.text);
                break;
            }
          },
          { name: "processEvent" },
        );
      }
    },
  },
});

// Expose the service over an HTTP/2 Restate endpoint.
serve({
  services: [agent],
});

package.json

{
  "name": "restate-ts-template",
  "version": "0.0.1",
  "description": "Template for JavaScript/TypeScript services running with Restate (https://github.com/restatedev/) ",
  "main": "app.js",
  "type": "commonjs",
  "scripts": {
    "build": "tsc --noEmitOnError",
    "start": "node ./dist/app.js",
    "dev": "tsx watch ./src/app.ts",
    "prebundle": "rm -rf dist",
    "bundle": "esbuild src/app.ts --bundle --minify --sourcemap --platform=node --target=es2020 --outfile=dist/app.js",
    "postbundle": "cd dist && zip -r index.zip app.js*",
    "app": "npm run start",
    "app-dev": "npm run dev"
  },
  "dependencies": {
    "@restatedev/restate-sdk": "1.15.0",
    "@restatedev/restate-sdk-gen": "1.15.0",
    "openai": "^4.77.0",
    "zod": "^4.3"
  },
  "devDependencies": {
    "@types/node": "^22",
    "esbuild": "^0.25.4",
    "tsx": "^4.19.2",
    "typescript": "^6.0.3"
  }
}

utils.ts

import { run, Operation, Future } from "@restatedev/restate-sdk-gen";

// you don't really need these, it just makes
// the main example code simpler.

// The result of pulling once from a durable source:
//   next    -> here's the next value
//   done    -> the underlying stream ended normally
//   aborted -> we're replaying past the point the live stream existed; the
//              source can no longer produce fresh values (see below).
export type Next<T> =
  | { type: "next"; value: T }
  | { type: "done" }
  | { type: "aborted" };

export type DurableSource<T> = {
  next(): Future<Next<T>>;
};

//
// create a durable source from a non-resumable source.
// an example to a non-resumable source might be any non-deterministic stream, such as random number generator
// or an LLM inference call that generally can not be re-done and to be expected to produces the same output sequence.
// This is useful for situation where you would like to consume from that source and already start applying effects durably.
// For example: applying tool calls from a streaming LLM. If the invocation gets aborted, then this durableSource will replay
// The captured prefix, and result with an 'aborted'.
//
// note: it doesn't really have to be an async generator, it can be rework to stream from a 'fetch' or
// some completion api for example.
//
export function* durableSource<T>(
  nonResumableSource: () => AsyncGenerator<T>,
): Operation<DurableSource<T>> {
  // Lives only in this process's memory — it is NOT restored on replay. After a
  // crash/replay we re-enter the function with `stream` undefined, which is how
  // we detect that the live source is gone.
  let stream: AsyncGenerator<T> | undefined;

  async function create() {
    stream = nonResumableSource();
  }

  async function next(): Promise<Next<T>> {
    // No live stream (e.g. we're replaying after a restart): we can't safely
    // re-run the non-deterministic source, so report the stream as aborted.
    if (!stream) {
      return { type: "aborted" };
    }
    const next = await stream.next();
    if (next.done) {
      return { type: "done" };
    }
    return { type: "next", value: next.value };
  }

  // Run the constructor through `run` so the act of starting the stream is
  // journaled — on replay this is a no-op and `stream` stays undefined.
  yield* run(create);

  // Each pull is wrapped in `run`, so every value we yielded is recorded in the
  // journal. On replay the recorded values are returned without touching the
  // live generator at all.
  return {
    next: () => run(next),
  };
}