Durable streaming

Preview

Durable streaming for Go is in preview, hosted in the core/x/streaming and plugins/firebase/exp packages. It can take breaking changes in a minor release, or be promoted to stable and moved to the root package. See API stability channels.

Report issues and feedback on Github

Genkit supports durable streaming, which allows flow state to be persisted. This enables clients to disconnect and reconnect to a stream and replay the full result. This is particularly useful for long-running operations or unreliable network connections.

How it works

When durable streaming is enabled, Genkit uses a StreamManager to store the chunks of a stream as they are generated. The client receives a streamId which can be used to reconnect to the stream and replay the full transcript.

Configuration

To enable durable streaming, you need to configure a StreamManager and pass it to genkit.Handler().

Development

For development and testing, or simple single-instance server, you can use the InMemoryStreamManager.

import "github.com/firebase/genkit/go/core/x/streaming"

// Create an in-memory stream manager with optional TTL for completed streams
sm := streaming.NewInMemoryStreamManager(
    streaming.WithTTL(10 * time.Minute),  // Optional: how long to retain completed streams
)

Note that InMemoryStreamManager stores streams in memory, so they will be lost if the server restarts. For production use cases where persistence across restarts is required, use FirestoreStreamManager.

Production

For production, you should use a durable storage solution.

The firebase plugin provides FirestoreStreamManager for durable stream storage.

import (
    "github.com/firebase/genkit/go/genkit"
    "github.com/firebase/genkit/go/plugins/firebase"
    firebasex "github.com/firebase/genkit/go/plugins/firebase/exp"
)

// Initialize Genkit with the Firebase plugin
g := genkit.Init(ctx, genkit.WithPlugins(&firebase.Firebase{}))

// Create a Firestore stream manager
sm, err := firebasex.NewFirestoreStreamManager(ctx, g,
    firebasex.WithCollection("genkit-streams"),  // Required: Firestore collection for stream documents
    firebasex.WithTimeout(2 * time.Minute),      // Optional: how long subscribers wait for new events
    firebasex.WithTTL(10 * time.Minute),         // Optional: how long completed streams are retained
)
if err != nil {
    log.Fatalf("Failed to create Firestore stream manager: %v", err)
}

FirestoreStreamManager provides:

• Persistence across restarts: Clients can reconnect to streams after server restarts
• Multi-instance support: Multiple server instances can serve the same stream
• Automatic cleanup: Completed streams are automatically deleted via Firestore TTL policies

Firestore TTL setup

For automatic cleanup of old streams, configure a TTL policy on your Firestore collection:

gcloud firestore fields ttls update expiresAt \
  --collection-group=genkit-streams \
  --enable-ttl \
  --project=YOUR_PROJECT_ID

See Firestore TTL documentation for more details.

Framework integration

net/http server

To enable durable streaming with Go’s standard net/http server, pass the StreamManager to genkit.Handler() using the WithStreamManager option:

package main

import (
    "context"
    "fmt"
    "log"
    "net/http"
    "time"

    "github.com/firebase/genkit/go/core/x/streaming"
    "github.com/firebase/genkit/go/genkit"
    "github.com/firebase/genkit/go/plugins/server"
)

func main() {
    ctx := context.Background()
    g := genkit.Init(ctx)

    // Define a streaming flow
    myFlow := genkit.DefineStreamingFlow(g, "myFlow",
        func(ctx context.Context, input string, sendChunk func(context.Context, string) error) (string, error) {
            // Your streaming logic here
            for i := 0; i < 5; i++ {
                if err := sendChunk(ctx, fmt.Sprintf("Chunk %d", i)); err != nil {
                    return "", err
                }
                time.Sleep(1 * time.Second)
            }
            return "Done!", nil
        })

    // Set up HTTP server with durable streaming
    mux := http.NewServeMux()
    mux.HandleFunc("POST /myFlow", genkit.Handler(myFlow,
        genkit.WithStreamManager(streaming.NewInMemoryStreamManager(
            streaming.WithTTL(10 * time.Minute),
        )),
    ))
    log.Fatal(server.Start(ctx, "127.0.0.1:8080", mux))
}

The basic-durable-streaming-exp sample is this program in runnable form, with a countdown slow enough to reconnect to while it is still running.

Firestore-backed durable streaming

For production deployments with persistence across restarts:

package main

import (
    "context"
    "log"
    "net/http"
    "time"

    "github.com/firebase/genkit/go/genkit"
    "github.com/firebase/genkit/go/plugins/firebase"
    firebasex "github.com/firebase/genkit/go/plugins/firebase/exp"
    "github.com/firebase/genkit/go/plugins/server"
)

func main() {
    ctx := context.Background()

    g := genkit.Init(ctx, genkit.WithPlugins(&firebase.Firebase{}))

    myFlow := genkit.DefineStreamingFlow(g, "myFlow",
        func(ctx context.Context, input string, sendChunk func(context.Context, string) error) (string, error) {
            // Your streaming logic here
            return "Done!", nil
        })

    sm, err := firebasex.NewFirestoreStreamManager(ctx, g,
        firebasex.WithCollection("genkit-streams"),
        firebasex.WithTimeout(2 * time.Minute),
        firebasex.WithTTL(10 * time.Minute),
    )
    if err != nil {
        log.Fatalf("Failed to create Firestore stream manager: %v", err)
    }

    mux := http.NewServeMux()
    mux.HandleFunc("POST /myFlow", genkit.Handler(myFlow, genkit.WithStreamManager(sm)))
    log.Fatal(server.Start(ctx, "127.0.0.1:8080", mux))
}

Client usage

Clients can initiate a stream and receive a streamId. This ID can be used to reconnect.

When durable streaming is enabled, the server returns a X-Genkit-Stream-Id header with the stream ID. Clients can use this ID to reconnect to the stream.

Starting a new stream

curl -N -i -H "Accept: text/event-stream" \
     -d '{"data": "your input"}' \
     http://localhost:8080/myFlow

The response headers will include X-Genkit-Stream-Id: <stream-id>. Save this ID to reconnect later.

Reconnecting to an existing stream

To reconnect to an in-progress or completed stream, pass the stream ID in the X-Genkit-Stream-Id header:

curl -N -H "Accept: text/event-stream" \
     -H "X-Genkit-Stream-Id: <stream-id-from-above>" \
     -d '{"data": "your input"}' \
     http://localhost:8080/myFlow

The subscription will:

• Replay any buffered chunks that were already sent
• Continue with live updates if the stream is still in progress
• Return all chunks plus the final result if the stream has already completed

The request body is ignored on resume. When X-Genkit-Stream-Id is present and a StreamManager is configured, the handler subscribes to the existing record and returns. The flow is never re-executed and never re-billed. The body still has to be valid JSON, so send -d '{}' if you have nothing to send.

The flow also keeps running after the original client disconnects: it executes on a detached context, so the remaining chunks and the final result still reach durable storage. Several clients may subscribe to the same ID at once, and each receives the buffered chunks followed by the live ones.

Resume outcomes

Stream ID	Response
Valid, run in progress or completed	200 with Content-Type: text/event-stream, replaying buffered chunks and then the final data: {"result": ...}
Unknown or TTL-expired	204 No Content, empty body. This is not an error, so check the status code rather than waiting for events.
Valid, but the run failed	200 with the chunks emitted before the failure, then a terminal error event

Failed runs

If the flow fails, the terminal error is written to the stream record, not just to the first client. Later subscribers get the chunks emitted before the failure, then the same data: {"error": {...}} frame described in Errors on a streaming request. The redaction rules are identical: only a message built with status.PublicErrorf reaches the wire, and the real error is logged server-side as streaming flow failed. The failed record is retained for the manager’s TTL like any other.

Security

Stream IDs are bearer capabilities

genkit.Handler performs no authorization on resume. Any caller presenting a valid stream ID receives the full replay: every chunk, the final result, and any error frame. The IDs are random UUIDv4 values generated server-side, so they are not guessable, but nothing binds a stream to the caller who started it.

To scope a stream to its owner, authenticate the request with genkit.WithContextProviders and record the owner for each ID you hand out, then reject an X-Genkit-Stream-Id the caller does not own in your own HTTP middleware before the Genkit handler runs.

Stream errors

If you call a StreamManager yourself rather than through genkit.Handler, match its failures with errors.Is against the sentinels in core/x/streaming: ErrStreamNotFound, ErrStreamAlreadyExists, ErrStreamWriterClosed, ErrStreamCompleted, and ErrStreamTimeout. Every implementation returns errors that satisfy them, so the same check works for the in-memory manager, the Firestore one, and any manager you write:

events, unsubscribe, err := sm.Subscribe(ctx, streamID)
if errors.Is(err, streaming.ErrStreamNotFound) {
    // Nothing to resume: the stream expired or never existed.
    return startFreshRun(ctx)
}
if err != nil {
    return err
}
defer unsubscribe()

Limitations

• Firestore: The entire stream history (chunks and final result) is stored in a single document. Firestore has a strict 1MB limitation on document size. If your stream output exceeds this limit, the flow will fail.
• InMemoryStreamManager: Streams are stored in memory and will be lost if the server restarts. Not suitable for production use cases where persistence is required.

Configuration options

InMemoryStreamManager options

Option	Default	Description
streaming.WithTTL(duration)	5 minutes	How long completed streams are retained in memory before cleanup

FirestoreStreamManager options

Option	Default	Description
firebasex.WithCollection(name)	(required)	Firestore collection for stream documents
firebasex.WithTimeout(duration)	60 seconds	How long subscribers wait for new events before timeout
firebasex.WithTTL(duration)	5 minutes	How long completed streams are retained before auto-deletion