Session stores

Beta

The Agents API is in Beta. It can introduce breaking changes in minor version releases.

Session stores persist snapshots for server-managed agents. They are the storage layer behind sessionId and snapshotId resumption, snapshot reads, branching, background execution, and aborting detached work.

Use the Sessions and state guide first when you are deciding between server-managed and client-managed state. Use this page when you know the server should own state and need to choose or implement the persistence layer.

Choose a store

• In-memory store for tests, demos, local examples, and single-process experiments.
• File store for local development, prototypes, CLIs, and single-host apps that need snapshots to survive process restarts.
• Firestore store for production apps on Google Cloud or Firebase that want a managed, multi-instance database without writing a store.
• Custom store for production apps that need a different database, centralized authorization, or retention policies that the built-in stores do not cover.

Most applications only configure a store on the agent. The agent runtime calls the store when it creates a snapshot, resumes with sessionId or snapshotId, serves loadChat(), reads a snapshot, starts detached work, or aborts an in-flight turn.

Use an in-memory store

InMemorySessionStore keeps snapshots in process memory. It is fast, requires no setup, and supports status-change callbacks inside the same process, which makes it useful for testing background execution and abort behavior locally.

import { InMemorySessionStore } from 'genkit/beta';

const store = new InMemorySessionStore<SupportState>();

export const supportAgent = ai.defineAgent({
  name: 'supportAgent',
  system: 'Help customers understand their order status.',
  stateSchema: SupportStateSchema,
  store,
});

Do not use the in-memory store when conversations must survive a process restart or when multiple server instances need to share sessions. Each process gets its own isolated map, so a sessionId created by one process is invisible to another.

The in-memory store accepts one option:

const store = new InMemorySessionStore<SupportState>({
  rejectBranchingSessions: true,
});

rejectBranchingSessions makes sessionId lookup fail when a session has more than one leaf snapshot. This is useful during development when you want accidental branching to be obvious. Clients can always resume by exact snapshotId.

Use a file-backed store

FileSessionStore stores each snapshot as a JSON file. It is a good fit for local development and single-host deployments where you want persistent state without operating a database.

import { FileSessionStore } from 'genkit/beta';

const store = new FileSessionStore<SupportState>(
  './.genkit/snapshots/support',
  {
    maxPersistedChainLength: 20,
    snapshotPathPrefix: ({ context }) => context?.auth?.uid ?? 'anonymous',
    rejectBranchingSessions: true,
    snapshotWatchPollIntervalMs: 1000,
  },
);

Snapshots are written under the directory passed to the constructor. By default, they go under a global subdirectory. When you provide snapshotPathPrefix, the prefix determines the subdirectory for every read and write.

Use maxPersistedChainLength to bound how many snapshots are retained in one parent chain. This helps local stores avoid growing forever. Because snapshots are full checkpoints, pruning an old ancestor removes it as a resume point, but surviving snapshots remain loadable.

Use snapshotPathPrefix to scope reads and writes to a subdirectory. Return a stable, app-controlled tenant or user segment from options.context, and do not return raw user input. The prefix becomes part of a filesystem path, so normalize or encode identifiers before returning them.

rejectBranchingSessions makes sessionId lookup fail when a session has more than one leaf snapshot. This is useful during development when you want accidental branching to be obvious. Clients can always resume by exact snapshotId.

Use snapshotWatchPollIntervalMs to control the polling fallback for file watching. The file store uses directory watching plus polling to observe status changes, which helps one process notice an abort or completion written by another process sharing the same snapshot directory. The default is 2000 milliseconds.

The file store serializes writes per snapshot file and writes by renaming a temporary file into place. That helps avoid torn JSON files when concurrent reads happen during a write. It does not turn the filesystem into a multi-instance production database, so use a custom store when many app instances need to coordinate durable sessions.

Use a Firestore store

FirestoreSessionStore persists snapshots in Cloud Firestore. It is a managed, multi-instance store, so it is the built-in option for production apps where several server instances share sessions, and it supports snapshot watching for background execution and abort.

It ships in two packages with the same API. Use @genkit-ai/google-cloud on Google Cloud, or @genkit-ai/firebase when you already have a Firebase Admin app. Both export from the /beta entry point.

import { FirestoreSessionStore } from '@genkit-ai/google-cloud/beta';

const store = new FirestoreSessionStore<SupportState>({
  collection: 'genkit-sessions',
  snapshotPathPrefix: (options) => options?.context?.auth?.uid ?? 'global',
});

export const supportAgent = ai.defineAgent({
  name: 'supportAgent',
  system: 'Help customers understand their order status.',
  stateSchema: SupportStateSchema,
  store,
});

All options are optional:

• db is an explicit Firestore instance. It defaults to a new client that picks up Application Default Credentials and the FIRESTORE_EMULATOR_HOST environment variable.
• collection is the collection that holds snapshot documents. It defaults to genkit-sessions. Two companion collections, <collection>-pointers and <collection>-shards, are derived from it for per-session pointers and sharded state.
• snapshotPathPrefix returns a per-tenant prefix from the call’s SessionStoreOptions, such as the authenticated user ID from options.context. When set, snapshots, pointers, and shards are nested under a tenant-scoped subcollection, so one tenant can never read another tenant’s snapshots even with a snapshotId. It defaults to global.
• checkpointInterval is the number of turns between full-state checkpoints. Between checkpoints the store writes diffs. A larger value writes fewer full snapshots but reconstructs over more diffs. It defaults to 25.
• shardSize is the maximum size in bytes of a single shard or diff document. State is split into chunks of this size so no document approaches Firestore’s 1 MiB limit. It defaults to 512 KiB.

On Firebase, import from @genkit-ai/firebase/beta instead and pass firebaseApp to derive the Firestore instance from an existing Admin app:

import { FirestoreSessionStore } from '@genkit-ai/firebase/beta';

const store = new FirestoreSessionStore<SupportState>({
  firebaseApp: app,
  collection: 'genkit-sessions',
});

The Firestore store watches snapshot documents to observe status changes, so background execution and aborts work across instances without polling. It does not prune snapshots, so plan retention for long-running or artifact-heavy conversations as described in Production guidance.

Implement a production store

When the built-in Firestore store does not fit, implement a custom SessionStore so snapshots live in your own production data layer. This is the right approach for Cloud SQL, Spanner, Postgres, Redis-backed systems with durability, or application-specific storage that already handles user and tenant authorization.

A custom store implements three capabilities:

• getSnapshot() loads either one exact snapshot or the latest snapshot for a session.
• saveSnapshot() applies an atomic read-modify-write.
• onSnapshotStateChange() lets the runtime observe status changes for abort and background work.

import type {
  SessionSnapshot,
  SessionStore,
  SessionStoreOptions,
  SnapshotMutator,
} from 'genkit/beta';

type SnapshotLookup = {
  snapshotId?: string;
  sessionId?: string;
  metadataOnly?: boolean;
  context?: SessionStoreOptions['context'];
};

class DatabaseSessionStore<S> implements SessionStore<S> {
  async getSnapshot(
    opts: SnapshotLookup,
  ): Promise<SessionSnapshot<S> | undefined> {
    // Load by opts.snapshotId, or load the latest leaf for opts.sessionId.
    throw new Error('Not implemented');
  }

  async saveSnapshot(
    snapshotId: string | undefined,
    mutator: SnapshotMutator<S>,
    options?: SessionStoreOptions,
  ): Promise<string | null> {
    // Atomically read the current snapshot, call mutator, and persist the result.
    throw new Error('Not implemented');
  }

  onSnapshotStateChange(
    snapshotId: string,
    callback: (snapshot: SessionSnapshot<S>) => void,
    options?: SessionStoreOptions,
  ): void | (() => void) {
    // Optional, but needed for responsive abort and background status updates.
  }
}

getSnapshot() must support exactly one lookup mode at a time:

• snapshotId loads that exact snapshot.
• sessionId loads the latest leaf snapshot for that session.

Here, “latest” means the latest leaf snapshot in the session history. A leaf is a snapshot that no other snapshot references as its parent. If branching exists, the built-in stores either select the most recently created leaf or throw when rejectBranchingSessions is enabled.

A lookup can also carry metadataOnly: true when the caller needs only the snapshot’s status and metadata, such as a getSnapshot request sent with metadataOnly. The store may then skip loading state, which spares reading a large conversation. Returning state anyway is still correct, because the runtime drops it from a metadata-only read.

saveSnapshot() must be atomic. In a database, wrap the read, mutator call, and write in a transaction or use optimistic concurrency with retries. The mutator may return a snapshot to save, return null to skip the write, or throw to fail the operation. If your retry logic can call the mutator more than once, keep the mutator invocation free of external side effects.

When snapshotId is undefined, the store should assign a new snapshot ID. When a snapshot ID is provided, the store should write that ID even if the mutator returns a different one.

An abort of detached work is two guarded writes through saveSnapshot(). The abort request flips a pending snapshot to aborting, and the background worker’s finalize then settles it as aborted with the state the run committed. Heartbeats refresh heartbeatAt on pending and aborting snapshots and change nothing else. Each mutator checks the status it starts from, so a store that runs the mutator inside a transaction handles these transitions correctly.

Implement onSnapshotStateChange() when detached background work should respond quickly to aborts or when clients should observe status changes without polling. Call back whenever a save changes a snapshot’s status: the runtime stops detached work when it sees aborting, and waitForSnapshot returns as soon as the snapshot settles. Return an unsubscribe function when the store opens a listener, subscription, or timer. If a custom store omits this method, normal server-managed state still works, but background abort behavior is limited.

Production guidance

Store snapshots as sensitive user data. They can contain message history, custom state, artifacts, tool inputs, and generated outputs.

Scope every read and write by authenticated user, organization, or tenant. Do not rely on snapshot IDs alone as authorization. Use SessionStoreOptions.context to pass request context into store operations.

Index by snapshot ID, session ID, parent ID, and creation time. sessionId lookup should be efficient because it is the common path for continuing a conversation. Parent relationships matter for leaf selection.

Plan retention before launch. Snapshots are full conversation checkpoints, so long-running conversations and artifact-heavy agents can grow quickly.

Choose a store

• In-memory store (InMemorySessionStore) for tests, local examples, command-line interfaces, and single-process experiments.
• File store (FileSessionStore) for local development, prototypes, and single-host applications where snapshots must persist across process restarts.
• Custom store (implementing SessionStore) for production apps utilizing centralized databases like Cloud SQL, Spanner, Postgres, or Redis.

Configure your session store directly on the agent’s constructor. The agent runtime manages reads, writes, and updates behind the scenes.

Use an in-memory store

InMemorySessionStore keeps snapshots in local process memory. It is fast and requires no setup.

import 'package:genkit/genkit.dart';
import 'package:genkit/experimental.dart';

final store = InMemorySessionStore();

final supportAgent = ai.defineAgent(
  name: 'supportAgent',
  system: 'Help customers with their orders.',
  store: store,
);

Do not use the in-memory store if session history must survive process restarts or when scaling horizontally across multiple server instances.

Use a file-backed store

FileSessionStore stores snapshots as JSON files inside a local directory. This is the standard choice for local development or single-host deployments.

Import FileSessionStore from the IO-safe entry point: package:genkit/experimental_io.dart.

import 'package:genkit/genkit.dart';
import 'package:genkit/experimental.dart';
import 'package:genkit/experimental_io.dart';

final store = FileSessionStore('.sessions');

final weatherAgent = ai.defineAgent(
  name: 'weatherAgent',
  system: 'You help with weather questions.',
  store: store,
);

Snapshots will be saved to files inside the .sessions folder.

Implement a production store

To store snapshots in a shared production database (such as PostgreSQL, Spanner, or Redis), implement a custom SessionStore:

import 'package:genkit/genkit.dart';
import 'package:genkit/experimental.dart';

class MyDatabaseSessionStore implements SessionStore {
  @override
  Future<SessionSnapshot?> getSnapshot({
    String? snapshotId,
    String? sessionId,
    Map<String, dynamic>? context,
  }) async {
    // Load snapshot from your database by ID or resolve latest for session ID
  }

  @override
  Future<String?> saveSnapshot(
    String? snapshotId,
    SnapshotMutator mutator, {
    Map<String, dynamic>? context,
  }) async {
    // Perform an atomic read-modify-write on the snapshot.
    // Call the mutator: final updated = mutator(existingSnapshot);
    // Write and commit the updated snapshot to your database.
  }
}

• getSnapshot fetches a snapshot by exact snapshotId or resolves the latest snapshot in the sequence for sessionId. The optional context carries ambient request/action context (such as the authenticated user) for multi-tenant routing.
• saveSnapshot must be atomic. Run the read, mutator execution, and write inside a single database transaction. This prevents concurrent writes from clobbering each other.
• SnapshotMetadataReader (optional): Implement this interface (getSnapshotMetadata and getLatestSnapshotMetadata) if your database supports projection queries, allowing metadataOnly: true status checks to skip loading the full conversation state payload.

Production guidance

• Security: Treat snapshots as sensitive user data. They can contain raw message history, tool results, and personal information. Apply authorization checks in your API or store layer before returning snapshot data.
• Payload size: Because snapshots contain full conversational checkpoints, their size grows over long sessions. Plan database indexes and cleanup/archival routines before launching production systems.

Choose a store

• In-memory store for tests, demos, local examples, and single-process experiments.
• File store for local development, prototypes, CLIs, and single-host apps that need snapshots to survive process restarts.
• Firestore store for production apps on Google Cloud or Firebase that want a managed, multi-instance database without writing a store.
• PostgreSQL store for production apps that already run PostgreSQL, such as Cloud SQL, and want several server instances to share sessions.
• Custom store for production web apps that need a different database, cloud storage, centralized authorization, retention policies, or tenant-aware persistence that the built-in stores do not cover.

Most applications only pass a store to aix.WithSessionStore(store). The agent runtime calls the store when it creates a snapshot, resumes with aix.WithSessionID or aix.WithSnapshotID, serves Agent.GetSnapshot, Agent.GetLatestSnapshot, and Agent.WaitForSnapshot, starts detached work, or aborts an in-flight turn with Agent.Abort.

The snapshot record

Every store method reads or writes an aix.SessionSnapshot[State]. Both the built-in stores and a custom store move this value verbatim:

type SessionSnapshot[State any] struct {
    SnapshotID   string               `json:"snapshotId"`
    SessionID    string               `json:"sessionId,omitempty"`
    ParentID     string               `json:"parentId,omitempty"`
    State        *SessionState[State] `json:"state,omitempty"`
    Status       SnapshotStatus       `json:"status,omitempty"`
    FinishReason AgentFinishReason    `json:"finishReason,omitempty"`
    Error        *status.Error        `json:"error,omitempty"`
    CreatedAt    time.Time            `json:"createdAt"`
    UpdatedAt    time.Time            `json:"updatedAt,omitempty"`
    HeartbeatAt  *time.Time           `json:"heartbeatAt,omitempty"`
}

The runtime owns the meaning of SnapshotID, SessionID, ParentID, State, Status, FinishReason, Error, and HeartbeatAt. The store owns durable persistence: it must preserve SessionID across rewrites, and it must not advance UpdatedAt on a heartbeat-only refresh, so liveness stays distinct from state changes. SessionState[State] is described on Sessions and state.

Status is an aix.SnapshotStatus, a string with six constants: pending, aborting, completed, failed, aborted, expired. Terminal() reports whether a status is settled, which is every one except pending and aborting.

Important implementation details:

• The identifier field is SnapshotID.
• State is a pointer and is nil on a pending or aborting snapshot, because a detached invocation commits its state only when it settles. It is also nil on a metadata-only read. Check for nil before reading snap.State.Messages, snap.State.Custom, or snap.State.Artifacts.
• An empty Status should be treated as completed for backwards compatibility. SnapshotStatusExpired is not directly persisted; it is computed on read from a stale HeartbeatAt, while the underlying record remains pending or aborting.

Full field documentation is on pkg.go.dev.

Use an in-memory store

localstore.NewInMemorySessionStore keeps snapshots in process memory. It is fast, requires no setup, and implements aix.SnapshotSubscriber and aix.SnapshotMetadataReader, which makes it useful for testing background execution and abort behavior locally.

import (
    aix "github.com/firebase/genkit/go/ai/exp"
    "github.com/firebase/genkit/go/ai/exp/localstore"
    genkitx "github.com/firebase/genkit/go/genkit/exp"
)

store := localstore.NewInMemorySessionStore[SupportState]()

supportAgent := genkitx.DefineAgent(g, "supportAgent",
    aix.InlinePrompt{
        ai.WithSystem("Help customers understand their order status."),
    },
    aix.WithSessionStore(store),
)

Do not use the in-memory store when conversations must survive a process restart or when multiple server instances need to share sessions. Each process gets its own isolated map, so a session created by one process is invisible to another.

The in-memory store does not have configuration options. If you need retention or tenant-scoped local files, use the file store.

Use a file-backed store

localstore.NewFileSessionStore stores each snapshot as a JSON file. It is a good fit for local development and single-host deployments where you want persistent state without operating a database.

store, err := localstore.NewFileSessionStore[SupportState](
    "./.genkit/snapshots/support",
    localstore.WithMaxPersistedChainLength(20),
    localstore.WithSnapshotPathPrefix(func(ctx context.Context) string {
        return tenantIDFromContext(ctx)
    }),
    localstore.WithPollInterval(time.Second),
)
if err != nil {
    // Fails if the snapshot directory cannot be created or an option is
    // invalid, such as a path prefix that escapes the store directory.
    log.Fatalf("open support store: %v", err)
}

NewFileSessionStore[State] returns (*localstore.FileSessionStore[State], error). The in-memory, Firestore, and PostgreSQL constructors likewise return their concrete types, *localstore.InMemorySessionStore[State], (*firebasex.FirestoreSessionStore[State], error), and (*pgexp.PostgresSessionStore[State], error). All of them satisfy aix.SessionStore[State], so hold one in a struct field or pass it around typed as the interface:

type Service struct {
    Store aix.SessionStore[SupportState]
}

Type the value concretely only when you need methods that a specific store adds.

Snapshots are written under the directory passed to NewFileSessionStore, inside a subdirectory named by the path prefix. Without the option, every session shares one global subdirectory. Each agent in go/samples/basic-agents keeps its own file store under ./.genkit/snapshots/<agent>/, so conversations survive a restart of the command-line client.

Use WithMaxPersistedChainLength to bound how many snapshots are retained in one parent chain. A value of 1 keeps only the latest snapshot in a chain. Omitting the option leaves pruning disabled. Pruning follows parent links, so sibling branches are pruned independently when they are extended.

Use WithSnapshotPathPrefix to scope reads and writes to a subdirectory. Return a stable tenant or user identity from context.Context so one caller cannot read another caller’s snapshots. Prefixes may contain / for nested directories, but values that escape the store directory are rejected. See Scope a store by tenant for where the identity in ctx comes from.

Use WithPollInterval to control how often the file store checks for status changes written by another process or store instance sharing the same directory. The default is one second. A value less than or equal to zero disables cross-process polling, so subscriptions observe only changes written through the same store instance.

The file store is safe for concurrent use inside one process, writes snapshots atomically with temporary files and rename, and implements aix.SnapshotSubscriber and aix.SnapshotMetadataReader. It is still a local filesystem store, so use the Firestore or PostgreSQL store when many app instances need to coordinate durable sessions.

Scope a store by tenant

The WithSnapshotPathPrefix option of the file, Firestore, and PostgreSQL stores takes a func(ctx context.Context) string. The ctx it receives is the request context the handler built, so the tenant identity travels from the HTTP request into the store through the action context.

Populate the action context with genkit.WithContextProviders when you mount the routes:

import (
    "context"
    "net/http"

    "github.com/firebase/genkit/go/core"
    "github.com/firebase/genkit/go/genkit"
    genkitx "github.com/firebase/genkit/go/genkit/exp"
)

for _, route := range genkitx.AgentRoutes(supportAgent) {
    mux.HandleFunc(route.Pattern(), route.Handler(
        genkit.WithContextProviders(func(ctx context.Context, req core.RequestData) (core.ActionContext, error) {
            // Authenticate the request first, then return the identity it proved.
            return core.ActionContext{"tenantId": req.Headers["x-tenant-id"]}, nil
        }),
    ))
}

Read it back out inside the prefix function with core.FromContext:

func tenantIDFromContext(ctx context.Context) string {
    tenantID, _ := core.FromContext(ctx)["tenantId"].(string)
    if tenantID == "" {
        return "global"
    }
    return tenantID
}

The prefix is recomputed from context.Context on every store call, so there is no operator bypass. A privileged reader such as a support console or a back-office approval queue must authorize the operator first, then build a context whose prefix function resolves to the target tenant, and only then read:

ctx = core.WithActionContext(ctx, core.ActionContext{"tenantId": targetTenant})

snap, err := supportAgent.GetSnapshot(ctx, snapshotID)

Do the authorization check outside the prefix function. The prefix is a scoping mechanism, not an access-control check.

Concurrent turns on one session

The runtime does not serialize invocations that share a sessionId. Two concurrent turns both resume from the same latest snapshot and both commit, creating sibling snapshots with the same ParentID. The next GetLatestSnapshot returns whichever was created last, so the other turn’s changes stop being reachable without its snapshotId.

Serialize turns per session in your application: a per-session lock, a queue, or disabling the send button client-side. To detect a fork after the fact, compare the snapshotId you resumed from with AgentOutput.SnapshotID.

An *aix.Agent is read-only after definition and safe to share across goroutines. An *aix.AgentConnection is a single-owner value: do not call Output() from one goroutine while another iterates Receive().

Use a Firestore store

firebasex.NewFirestoreSessionStore persists snapshots in Cloud Firestore. It is a managed, multi-instance store, so it suits production apps where several server instances share sessions, and it implements aix.SnapshotSubscriber for background execution and abort.

The store resolves its Firestore client from the Firebase plugin registered with the Genkit instance, so pass the Firebase plugin to genkit.Init before constructing the store.

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

g := genkit.Init(ctx,
    genkit.WithExperimental(),
    genkit.WithPlugins(&firebase.Firebase{ProjectId: "my-project"}),
)

store, err := firebasex.NewFirestoreSessionStore[SupportState](ctx, g,
    firebasex.WithCollection("genkit-sessions"),
    firebasex.WithSnapshotPathPrefix(func(ctx context.Context) string {
        return tenantIDFromContext(ctx)
    }),
)
if err != nil {
    // Fails if the Firebase plugin is not registered on g or the Firestore
    // client cannot be resolved.
    log.Fatalf("open Firestore store: %v", err)
}

supportAgent := genkitx.DefineAgent(g, "supportAgent",
    aix.InlinePrompt{
        ai.WithSystem("Help customers understand their order status."),
    },
    aix.WithSessionStore(store),
)

The State type parameter is the user-defined custom-state type carried in the session state; it must be JSON-serializable.

All options are optional:

• WithCollection sets the root collection that holds snapshot documents. It defaults to genkit-sessions. Two companion collections, <collection>-shards and <collection>-pointers, are derived from it for sharded checkpoint state and per-session pointers.
• WithSnapshotPathPrefix derives a per-tenant prefix from context.Context, such as an authenticated user or organization ID. When set, snapshots, shards, and pointers are nested under a tenant-scoped subcollection, so one tenant can never read another tenant’s snapshots even with a snapshot ID. The value must be a single valid Firestore document ID (no / separators) and stable for a snapshot’s lifetime, since every read recomputes it. An empty result falls back to global, which is also the default when the option is omitted.
• WithCheckpointInterval sets the number of turns between full-state checkpoints. Between checkpoints the store writes JSON Patch diffs; a larger value writes fewer full checkpoints but reconstructs over more diffs. The number of diff documents read or written per turn is bounded by this value rather than by total session length. Must be at least 1; defaults to 25.
• WithShardSize sets the maximum size in bytes of a single shard or diff document. Checkpoint state is split into chunks of this size, and any diff exceeding it is promoted to a sharded checkpoint, so no document approaches Firestore’s 1 MiB limit. Must be positive; defaults to 512 KiB.

The store picks up Application Default Credentials and the FIRESTORE_EMULATOR_HOST environment variable through the Firebase plugin, so the same code runs against the Firestore emulator for local testing.

Because it implements aix.SnapshotSubscriber over Firestore’s native real-time listeners, a status change such as an abort committed by one process is observed by the process running the detached turn even across instances, without polling. It also implements aix.SnapshotMetadataReader, so a read with aix.WithMetadataOnly() costs one document read where a full read reconstructs the state from its checkpoint shards and diff chain. The store does not prune snapshots, so plan retention for long-running or artifact-heavy conversations as described in Production guidance.

Use a PostgreSQL store

pgexp.NewPostgresSessionStore persists snapshots in one PostgreSQL table. Like the Firestore store, it is a multi-instance store: servers on one database share every session, and one server can abort a background turn that another runs.

Preview

This store is in the experimental plugins/postgresql/exp package and can change in a minor release.

The store takes its connection pool from the PostgreSQL plugin registered with the Genkit instance, so pass the plugin to genkit.Init before constructing the store. See Cloud SQL for PostgreSQL for the engine’s connection options.

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

engine, err := postgresql.NewPostgresEngine(ctx,
    postgresql.WithCloudSQLInstance("my-project", "us-central1", "my-instance"),
    postgresql.WithDatabase("my-database"),
)
if err != nil {
    log.Fatalf("connect to PostgreSQL: %v", err)
}
defer engine.Close()

g := genkit.Init(ctx,
    genkit.WithExperimental(),
    genkit.WithPlugins(&postgresql.Postgres{Engine: engine}),
)

store, err := pgexp.NewPostgresSessionStore[SupportState](ctx, g,
    pgexp.WithSnapshotPathPrefix(tenantIDFromContext),
)
if err != nil {
    // Fails if the PostgreSQL plugin is not registered on g or the table
    // cannot be created.
    log.Fatalf("open PostgreSQL store: %v", err)
}

supportAgent := genkitx.DefineAgent(g, "supportAgent",
    aix.InlinePrompt{
        ai.WithSystem("Help customers understand their order status."),
    },
    aix.WithSessionStore(store),
)

To connect to a database outside Cloud SQL, build a pgxpool.Pool yourself and pass it with postgresql.WithPool(pool). The engine leaves a pool you passed in open when it closes, so close that pool yourself.

All options are optional, and a later option of the same kind replaces an earlier one:

• WithTableName sets the snapshot table. It defaults to genkit_session_snapshots, and is at most 51 bytes so the index names derived from it fit PostgreSQL’s identifier limit.
• WithSchemaName sets the schema the table lives in. The schema must exist. Without it, the table name resolves through the connection’s search_path.
• WithCheckpointInterval sets the number of turns between full-state checkpoints. Between checkpoints each row stores only the JSON Patch from its parent’s state, so a read applies at most this many patches however long the session grows. Must be between 1 and 1000; defaults to 25. With 1, every row is a checkpoint.
• WithSnapshotPathPrefix derives a per-tenant key from context.Context. Every row is keyed by it, so one tenant can never read another tenant’s snapshots, even with a snapshot ID. The value must be non-empty and stable for a snapshot’s lifetime. Omit the option for the default global key.
• WithPollInterval sets how often a status subscription re-reads the rows it watches. Status changes arrive at once over LISTEN/NOTIFY; the poll covers a change written without a notification, such as a manual UPDATE, and a connection that cannot receive one, such as one through a transaction-mode pooler. Defaults to 5 seconds; a value less than or equal to zero turns the poll off.

The store creates its table and indexes when the table does not exist, which needs the CREATE privilege on the schema. Where the schema is managed elsewhere, construct the store once with a role that has it; from then on the store needs only SELECT, INSERT, and UPDATE on the table.

The store implements aix.SnapshotSubscriber over LISTEN/NOTIFY, so an abort committed by one process reaches the process running the detached turn at once. It holds one connection for this while it has subscribers. It also implements aix.SnapshotMetadataReader. State and patches are stored as json, not jsonb, which keeps the exact text and accepts the NUL character that tool output can carry. A save that changes the state of a row another row stores a diff against fails with FAILED_PRECONDITION instead of corrupting it. Like the Firestore store, it does not prune snapshots.

The basic-agents-server sample runs on this store with -store=postgres, reading the connection string from DATABASE_URL.

Implement a production store

When neither the Firestore store nor the PostgreSQL store fits, implement a custom aix.SessionStore so snapshots live in your own production data layer. This is the right approach for Spanner, Redis-backed systems with durability, or application-specific storage that already handles user and tenant authorization.

aix.SessionStore is the pair of smaller interfaces aix.SnapshotReader and aix.SnapshotWriter, so a custom store implements snapshot reading and writing:

type SessionStore[State any] interface {
    GetSnapshot(ctx context.Context, snapshotID string) (*aix.SessionSnapshot[State], error)
    GetLatestSnapshot(ctx context.Context, sessionID string) (*aix.SessionSnapshot[State], error)
    SaveSnapshot(
        ctx context.Context,
        snapshotID string,
        fn func(existing *aix.SessionSnapshot[State]) (*aix.SessionSnapshot[State], error),
    ) (*aix.SessionSnapshot[State], error)
}

• GetSnapshot loads a snapshot by exact ID.
• GetLatestSnapshot loads the most recently created snapshot for a session, whatever its status. Use CreatedAt for recency, not UpdatedAt, because heartbeat writes update liveness without changing the conversation state. If two snapshots have the same CreatedAt, break ties deterministically. Parent IDs are lineage metadata for this lookup, not how the latest session snapshot is resolved.
• SaveSnapshot must be atomic. In a database, run the read, callback, and write in one transaction or use optimistic concurrency with retries. The callback may return a snapshot to save, return nil, nil to skip the write, or return an error to fail the operation. Stores that retry on contention may call the callback more than once, so keep the callback free of external side effects.

The store owns identity. If snapshotID is empty, generate a fresh ID. If snapshotID is provided, persist that ID even if the callback returns a different one. Preserve a row’s existing session ID on update.

Three more rules keep the runtime’s reads correct:

• Keep timestamps to at least millisecond precision. CreatedAt orders a session’s rows, and the runtime creates each row at least a millisecond after its parent.
• A save replaces the row. A field the callback leaves empty is empty afterward, which is how a finalize clears the heartbeat.
• Never change the State of a row that another row names as its parent. The runtime never does, and a store that keeps children as diffs against their parent would corrupt them.

To support detached background work and abort, also implement aix.SnapshotSubscriber.

type SnapshotSubscriber interface {
    OnSnapshotStatusChange(ctx context.Context, snapshotID string) <-chan aix.SnapshotStatus
}

The channel should yield the current status when a subscription starts, then yield later status changes until the context is canceled. The runtime uses this to notice when an abort flips a pending snapshot to aborting, and WaitForSnapshot uses it to return as soon as a row settles. Yield a status whenever a save changes it, including a save that edits the row in place.

Stores that do not implement SnapshotSubscriber can still support server-managed state, snapshots, and GetLatestSnapshot. The runtime rejects detach attempts because it cannot signal background work to stop.

An abort is two guarded writes through SaveSnapshot: the flip from pending to aborting, and the finalize from aborting to the status the run settled on, with its state. That status is aborted for a turn the stop cut short, and completed or failed for a turn that ended on its own. Heartbeats refresh HeartbeatAt on both statuses and change nothing else. Because each mutator checks the status it starts from, a store that runs the callback inside a transaction gets these transitions right for free.

A store may also implement aix.SnapshotMetadataReader, both methods or neither:

type SnapshotMetadataReader[State any] interface {
    GetSnapshotMetadata(ctx context.Context, snapshotID string) (*aix.SessionSnapshot[State], error)
    GetLatestSnapshotMetadata(ctx context.Context, sessionID string) (*aix.SessionSnapshot[State], error)
}

They serve reads made with aix.WithMetadataOnly(), which the runtime uses wherever it only needs to know where a row stands: waiting on a snapshot, deciding whether a background task can be continued. Return the row with State nil. Without the interface the runtime reads in full and drops the state, so the capability is a cost saving rather than a requirement.

Test a store with the conformance suite

sessionstoretest.Run, in github.com/firebase/genkit/go/ai/exp/sessionstoretest, checks a store against the contract the runtime relies on. The in-memory, file, Firestore, and PostgreSQL stores all run it. Call it from a test in your store’s package, with a function that returns a fresh, empty store:

import (
    "context"
    "testing"

    aix "github.com/firebase/genkit/go/ai/exp"
    "github.com/firebase/genkit/go/ai/exp/sessionstoretest"
)

func TestConformance(t *testing.T) {
    sessionstoretest.Run(t, func(t *testing.T) aix.SessionStore[MyState] {
        return newEmptyStore(t)
    }, &sessionstoretest.Options[MyState]{
        // A second store over the same storage, as another process opens it.
        Reopen: func(t *testing.T, s aix.SessionStore[MyState]) aix.SessionStore[MyState] {
            return reopenStore(t, s)
        },
        // A store that scopes rows by tenant, and two contexts for two tenants.
        Scoped: func(t *testing.T) (aix.SessionStore[MyState], context.Context, context.Context) {
            return newTenantStore(t), tenantContext("a"), tenantContext("b")
        },
    })
}

The suite runs grouped subtests: Save, Lifecycle, Chains, Latest, Metadata, Subscriber, SecondInstance, and Scopes. It checks aix.SnapshotSubscriber and aix.SnapshotMetadataReader when the store implements them, and each Options hook turns on its group. A group whose capability or hook is missing is skipped with the reason, and a nil *Options runs the single-store checks only. Spell out the State type argument on the Options literal, since Go does not infer type parameters from struct fields. The test data lives in messages, artifacts, and usage, and the custom state stays at its zero value, so any State type works.

Production guidance

Store snapshots as sensitive user data. They can contain message history, custom state, artifacts, tool inputs, and generated outputs.

Scope every read and write by authenticated user, organization, or tenant. Do not rely on snapshot IDs alone as authorization. Derive tenancy from context.Context and apply it consistently in every store method.

Index by snapshot ID and by session ID plus creation time. GetLatestSnapshot should be efficient because it is the common path for continuing a conversation by sessionId.

Plan retention before launch. Snapshots are full conversation checkpoints, so long-running conversations and artifact-heavy agents can grow quickly.

Choose a store

• In-memory store (InMemorySessionStore from genkit.exp.agent) for tests, local examples, command-line interfaces, and single-process experiments.
• File store (FileSessionStore from genkit.exp.agent) for local development, prototypes, and single-host applications where snapshots must persist across process restarts.
• Firestore store (FirestoreSessionStore from genkit_google_cloud.exp) for production apps where several server instances share sessions.
• Custom store (implementing SessionStore from genkit.exp.agent) when you need a different shared database such as Cloud SQL, Spanner, Postgres, or Redis.

Configure the store on the agent. The runtime handles reads and writes. With HTTP serving, /getSnapshot and /abort are mounted only when a store is configured.

Use an in-memory store

InMemorySessionStore keeps snapshots in local process memory. It is fast and requires no setup.

from genkit.exp.agent import InMemorySessionStore

store = InMemorySessionStore()

support_agent = ai.define_agent(
    name='supportAgent',
    model=GoogleAI.gemini('gemini-flash-latest'),
    system='Help customers with their orders.',
    store=store,
)

Do not use the in-memory store if session history must survive process restarts or when scaling horizontally across multiple server instances.

Both built-in local stores accept reject_ambiguous_session=True. Branching from an earlier snapshot gives a session a second leaf; with this flag set, a session_id lookup on a forked session raises instead of picking the newest leaf, so callers must resume by snapshot_id.

Use a file-backed store

FileSessionStore writes one <snapshot_id>.json file per snapshot inside a local directory. This is the standard choice for local development or single-host deployments.

from genkit.exp.agent import FileSessionStore

store = FileSessionStore('.sessions', max_persisted_chain_length=50)

weather_agent = ai.define_agent(
    name='weatherAgent',
    model=GoogleAI.gemini('gemini-flash-latest'),
    system='You help with weather questions.',
    store=store,
)

max_persisted_chain_length caps how many snapshots of one conversation stay on disk. Once a chain grows past it, each save deletes the oldest turns. Resuming still works, but you cannot rewind or branch past the retained window. Leave it unset to keep the full history.

Use a Firestore store

FirestoreSessionStore from genkit_google_cloud.exp (genkit-google-cloud package) persists snapshots in Cloud Firestore. It is the built-in option for multi-instance production deployments, such as Cloud Run, that share sessions across instances.

Storage model

• Diffs and sharded checkpoints: Saves each turn as an RFC 6902 JSON Patch diff from its parent and writes a full-state checkpoint every checkpoint_interval turns. Checkpoints split across shard documents, so long sessions stay under Firestore’s 1 MiB document limit.
• Direct document-ID reads: Resolves sessions through a pointer document and direct document-ID fetches inside a read transaction. No composite indexes are required.
• Transactional writes: Commits snapshot documents and session pointers atomically, retrying contended pointer updates up to transaction_max_attempts times before raising GenkitError with status ABORTED.
• Cross-instance status watches: Streams status changes through Firestore document watches, so task.wait(), task.poll(), and task.abort() work across instances.

from typing import Any

from genkit_google_cloud.exp import FirestoreSessionStore


def by_user(context: dict[str, Any] | None) -> str:
    uid = (context or {}).get('uid')
    return uid if isinstance(uid, str) else 'global'


store = FirestoreSessionStore(
    collection='genkit-sessions',
    snapshot_path_prefix=by_user,
)

support_agent = ai.define_agent(
    name='supportAgent',
    model=GoogleAI.gemini('gemini-flash-latest'),
    system='Help customers check order status.',
    store=store,
)

Constructor options

All options are keyword-only and optional:

• client: Explicit google.cloud.firestore.AsyncClient. Defaults to a client using Application Default Credentials, or FIRESTORE_EMULATOR_HOST when set.
• sync_client: Explicit synchronous google.cloud.firestore.Client for document watches. When omitted, the store creates one per event loop and closes it in await store.close().
• collection: Base collection name (defaults to 'genkit-sessions'). The store derives <collection>-pointers and <collection>-shards. Document formats differ by SDK, so give each SDK its own collection.
• snapshot_path_prefix: Function that maps the call context to a tenant path segment, such as the authenticated user ID (defaults to 'global').
• checkpoint_interval: Turns between full-state checkpoints (defaults to 25).
• shard_size: Maximum bytes per shard document (defaults to 512 KiB).
• transaction_max_attempts: Retry budget for contended pointer transactions (defaults to 5).

The Firestore store does not prune snapshots, so plan retention for long-running or artifact-heavy conversations as described in Production guidance.

Implement a production store

When the built-in stores do not fit, implement the SessionStore protocol from genkit.exp.agent. Add SnapshotSubscriber when you need detach and abort. The exported helpers keep the runtime’s invariants:

import asyncio
from typing import Any

from genkit.exp.agent import (
    TERMINAL_STATUSES,
    SaveFn,
    SessionSnapshot,
    SessionStore,
    SnapshotStatus,
    SnapshotStatusStream,
    SnapshotSubscriber,
    apply_save,
    iterate_statuses,
    require_one_selector,
)


class DatabaseSessionStore(SessionStore[Any], SnapshotSubscriber):
    def __init__(self) -> None:
        self._lock = asyncio.Lock()
        self._subs: dict[str, list[asyncio.Queue[SnapshotStatus | None]]] = {}

    async def get_snapshot(
        self,
        *,
        snapshot_id: str | None = None,
        session_id: str | None = None,
        context: dict[str, Any] | None = None,
    ) -> SessionSnapshot | None:
        require_one_selector(snapshot_id=snapshot_id, session_id=session_id)
        # 1. Load by snapshot_id, or resolve the latest leaf for session_id.
        ...

    async def save_snapshot(
        self,
        snapshot_id: str,
        fn: SaveFn,
        *,
        context: dict[str, Any] | None = None,
    ) -> SessionSnapshot | None:
        # 2. Read, mutate, and write in one transaction (or under a lock).
        async with self._lock:
            existing = await self._read(snapshot_id)
            updated = apply_save(existing=existing, snapshot_id=snapshot_id, fn=fn)
            if updated is not None:
                await self._write(snapshot_id, updated)
                self._notify(snapshot_id, updated.status)
            return updated

    async def on_snapshot_status_change(
        self,
        snapshot_id: str,
        *,
        context: dict[str, Any] | None = None,
    ) -> SnapshotStatusStream:
        # 3. Seed the current status, then stream changes until the snapshot settles.
        queue: asyncio.Queue[SnapshotStatus | None] = asyncio.Queue()
        async with self._lock:
            current = await self._read(snapshot_id)
            if current is not None:
                _push(queue, current.status)
            self._subs.setdefault(snapshot_id, []).append(queue)

        async def unsubscribe() -> None:
            self._subs[snapshot_id].remove(queue)

        return iterate_statuses(queue, on_close=unsubscribe)

    def _notify(self, snapshot_id: str, status: SnapshotStatus | None) -> None:
        for queue in self._subs.get(snapshot_id, []):
            _push(queue, status)

    async def _read(self, snapshot_id: str) -> SessionSnapshot | None: ...

    async def _write(self, snapshot_id: str, snapshot: SessionSnapshot) -> None: ...


def _push(queue: asyncio.Queue[SnapshotStatus | None], status: SnapshotStatus | None) -> None:
    queue.put_nowait(status)
    if status in TERMINAL_STATUSES:
        queue.put_nowait(None)  # End the stream.

• get_snapshot fetches a snapshot by exact snapshot_id or resolves the latest snapshot for session_id. require_one_selector rejects calls that pass both or neither.
• save_snapshot must be atomic. Run the read, mutator, and write inside one database transaction, or synchronize with a lock. The mutator (SaveFn) is synchronous and side-effect free, since stores may call it more than once under contention. apply_save runs it, stamps the store fields, and rejects status changes once a snapshot reaches a terminal status (completed, failed, or aborted).
• on_snapshot_status_change is an async def that returns the stream. The runtime subscribes when a turn detaches and cancels the background work when the snapshot flips to aborted. Seed the current status so an abort written before the subscription is not missed, and put None after a terminal status so iterate_statuses ends the stream. Without SnapshotSubscriber, chat.detach() and task.wait() still work, but task.abort() only marks the snapshot aborted while the background turn runs to completion.
• Multiple instances: The in-process queues above only see writes made through this store object. To abort a turn running on another instance, feed _notify from your database’s change stream, such as Postgres LISTEN/NOTIFY.

Production guidance

• Security: Treat snapshots as sensitive user data. They can contain raw message history, tool results, and personal information. Scope reads, writes, and status subscriptions by the caller identity in context (for example with snapshot_path_prefix on FirestoreSessionStore) so a leaked snapshot_id cannot read or abort another user’s session.
• Payload size: Because snapshots contain full conversational checkpoints, their size grows over long sessions. Plan database indexes and cleanup or archival routines before launching production systems.