Restate signal-stream example: an agent VO consuming an LLM service's output chunk-by-chunk via durable signals

app.ts

import {
  Operation,
  object,
  service,
  invocation,
  sendClient,
  handlerRequest,
  run,
} from "@restatedev/restate-sdk-gen";
import { serve, InvocationId } from "@restatedev/restate-sdk";
import * as utils from "./utils";

//
// A two-actor streaming example built on Restate:
//
//   agent.run  ──(one-way call)──▶  gpt5.completionStreamer
//      ▲                                   │
//      └────────(signals, one per chunk)───┘
//
// The `agent` kicks off a long-running `gpt5` invocation and then sits back
// and consumes the LLM output chunk-by-chunk. Instead of returning a single
// result, `gpt5` pushes each chunk back to the `agent` as a durable *signal*.
// Both sides are durable: if either invocation crashes and is retried, Restate
// replays the journal so we resume exactly where we left off — no chunk is lost
// or re-applied.
//

// The payload `gpt5` receives: what to prompt, plus the id of the invocation
// that should receive the streamed chunks back (so it knows whom to signal).
type CompletionRequest = {
  message: string;
  caller: InvocationId;
};

// A single streamed chunk. `done: true` is the end-of-stream sentinel (no text);
// `done: false` carries the next piece of text.
type Completion =
  | { done: false; text: string }
  | { done: true; text: undefined };

// imagine this is a completion api that yields some tool calls, and reasoning etc'
async function* prompt(input: string) {
  yield "hi";

  for (let i = 0; i < 20; i++) {
    yield `chunk-${i}`;
    await new Promise((resolve) => setTimeout(resolve, 1000));
  }

  yield "bye";
}

// The consumer side. A virtual object so each agent instance has its own
// durable identity that `gpt5` can address its signals to.
const agent = object({
  name: "agent",
  handlers: {
    *run(): Operation<void> {
      //
      // Fire-and-forget call to the LLM service (sendClient = one-way, we don't
      // await a return value). We hand it our own invocation id so it knows
      // where to deliver the chunks. (see llm below)
      //
      sendClient(llm).completionStreamer({
        message: "What is the weather in SF?",
        caller: handlerRequest().id, // <--- this is us
      });

      //
      // Turn the incoming "chunk" signals into a stream we can pull from.
      // Each `.next()` durably awaits the next signal `gpt5` resolves.
      //
      const chunks = utils.signalStream<Completion>("chunk");

      while (true) {
        // Block until the next chunk arrives (or the done sentinel).
        const { done, text } = yield* chunks.next();
        if (done) {
          break;
        }

        // Process the chunk inside `run` so the side effect is journaled and
        // not repeated on replay.
        yield* run(
          async () => {
            // imagine doing something with this chunk
            console.log(text);
          },
          { name: "processChunk" },
        );
      }
    },
  },
});

// The producer side. A plain service (no durable identity of its own needed);
// it simply drains the LLM stream and signals each chunk back to its caller.
const llm = service({
  name: "gpt5",
  handlers: {
    *completionStreamer(req: CompletionRequest): Operation<void> {
      // A handle to the caller's invocation, so we can push signals back to it.
      const caller = invocation(req.caller);

      // Wrap the non-replayable LLM stream in a durable source: on replay it
      // yields the already-consumed prefix from the journal instead of
      // re-running the (non-deterministic) generator. See utils.durableSource.
      const source = yield* utils.durableSource(() => prompt(req.message));

      // Pull chunks one at a time and forward each to the caller as a "chunk"
      // signal. The signal name must match what the agent's signalStream awaits.
      while (true) {
        const chunk = yield* source.next();
        switch (chunk.type) {
          // Stream finished, or the source was interrupted/aborted on replay:
          // tell the caller we're done so its loop can exit.
          case "done":
          case "aborted": {
            caller
              .signal<Completion>("chunk")
              .resolve({ done: true, text: undefined });
            return;
          }
          // Another chunk of text — hand it to the caller and keep going.
          case "next": {
            caller
              .signal<Completion>("chunk")
              .resolve({ done: false, text: chunk.value });
            break;
          }
        }
      }
    },
  },
});

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

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",
    "zod": "^4.3"
  },
  "devDependencies": {
    "@types/node": "^22",
    "esbuild": "^0.25.4",
    "tsx": "^4.19.2",
    "typescript": "^6.0.3"
  }
}

utils.ts

import { run, signal, 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),
  };
}

// Just a tiny quality of life utility, that creates a stream from signals.
// A signal is a single-shot durable future; this wraps the "await one, then
// arm the next" boilerplate so callers can pull repeatedly from the same name.
export function signalStream<T>(name: string): { next(): Operation<T> } {
  // The future for the signal we're currently waiting on.
  let current = signal<T>(name);

  return {
    *next(): Operation<T> {
      // Durably block until this signal is resolved...
      const value = yield* current;
      current = signal(name); // ...then arm a fresh future for the next value.
      return value;
    },
  };
}