Agent Sessions & Persistence (Experimental)

Experimental / preview API. Session stores come from ai/exp/localstore. Read agents.md for the basics first.

When an agent has a SessionStore (via aix.WithSessionStore), the server owns the session history. Each turn produces an immutable snapshot; the snapshot chain carries conversation state forward. A store also enables branching and background execution. (Interrupts work with or without a store — see human-in-the-loop.)

Pick a store

ai/exp/localstore ships two single-process stores. Both are generic over the custom-state type State (use any when you have no typed custom state).

import "github.com/genkit-ai/genkit/go/ai/exp/localstore"

// In-memory: great for tests/dev; lost on restart.
memStore := localstore.NewInMemorySessionStore[any]()

// File-backed: snapshots persisted as <dir>/<prefix>/<snapshotId>.json
// (the prefix defaults to "global"). Returns an error if dir can't be created.
fileStore, err := localstore.NewFileSessionStore[any]("./.snapshots")
if err != nil {
	log.Fatal(err)
}

localstore is for local development, tests, and single-instance apps. For multi-instance production, implement aix.SessionStore against a real database (see below).

NewFileSessionStore options:

• localstore.WithMaxPersistedChainLength(n): keep only the newest n snapshots in a chain (each save prunes older ancestors). n >= 1.
• localstore.WithSnapshotPathPrefix(fn): derive a per-call subdirectory from context to isolate snapshots per tenant (e.g. "org-42/user-7").
• localstore.WithPollInterval(d): how often the store re-reads subscribed snapshot files to observe cross-process status changes (default 2s). This is what lets one process abort a detached turn another process is running.

pruning, err := localstore.NewFileSessionStore[any]("./.snapshots",
	localstore.WithMaxPersistedChainLength(3),
)

Attach a store to the agent with aix.WithSessionStore:

logbookAgent := genkitx.DefineAgent(g, "logbookAgent",
	aix.InlinePrompt{
		ai.WithModelName("googleai/gemini-flash-latest"),
		ai.WithSystem("You are a personal logbook assistant."),
	},
	aix.WithSessionStore(fileStore),
)

Turns persist automatically

Each turn writes a snapshot chained off the previous one. The turn's snapshot ID is on out.SnapshotID; the conversation's stable ID is on out.SessionID.

out1, _ := logbookAgent.RunText(ctx, "Log this: I started studying Genkit today.")
out2, _ := logbookAgent.RunText(ctx, "What did I study today?",
	aix.WithSessionID(out1.SessionID)) // remembers turn 1
fmt.Println(out1.SnapshotID, out2.SnapshotID)

Resume a conversation

Two ways to resume a server-managed conversation:

• aix.WithSessionID(id) — resume the session's latest snapshot. Use it when you track the conversation, not individual snapshots.
• aix.WithSnapshotID(id) — resume a specific snapshot (see branching).

// Continue from the session's latest snapshot:
out, _ := logbookAgent.RunText(ctx, "Add another note.",
	aix.WithSessionID(sessionID))

// Or continue from an exact checkpoint:
out, _ = logbookAgent.RunText(ctx, "Add another note.",
	aix.WithSnapshotID(snapshotID))

Resume is rejected if the latest snapshot is a failed, aborted, or pending dead end (a pending tip means a detached invocation is still running — wait for it or abort it). aix.WithState is for client-managed agents only and is mutually exclusive with WithSessionID / WithSnapshotID.

Read a snapshot without running a turn

Agent.GetSnapshot / Agent.GetLatestSnapshot fetch a stored snapshot (applying any configured aix.WithStateTransform). Handy for restoring a UI after a reload. They return FAILED_PRECONDITION on a client-managed agent (no store).

snap, err := logbookAgent.GetSnapshot(ctx, snapshotID)
if err != nil {
	log.Fatal(err)
}
fmt.Println(snap.Status) // aix.SnapshotStatusCompleted, ...
for _, m := range snap.State.Messages {
	fmt.Println(m.Role, m.Text())
}

latest, _ := logbookAgent.GetLatestSnapshot(ctx, sessionID)

Typed session state

Parameterize the store (and agent) with a struct to attach typed custom state. State is validated (via JSON) when a snapshot is loaded. See working with state for reading and mutating it inside tools.

type Profile struct {
	Name string `json:"name"`
	Tier string `json:"tier"` // "free" | "pro"
}

profileAgent := genkitx.DefineAgent(g, "profileAgent",
	aix.InlinePrompt{
		ai.WithModelName("googleai/gemini-flash-latest"),
		ai.WithSystem("Greet the user by name and tailor answers to their tier."),
	},
	aix.WithSessionStore(localstore.NewInMemorySessionStore[Profile]()),
)

// Seed custom state on the first turn (client-managed here for illustration;
// with a store you would seed via WithState on a fresh session).
out, _ := profileAgent.Run(ctx,
	&aix.AgentInput{Message: ai.NewUserTextMessage("Hello")},
	aix.WithState(&aix.SessionState[Profile]{
		Custom: Profile{Name: "Ada", Tier: "pro"},
	}),
)

Interrupts (human-in-the-loop)

Interrupts pause a turn so a human can approve or provide input, then resume from the exact pause point. They work with a store or with client-managed state (persistence is orthogonal). See Human-in-the-loop / interrupts.

Implementing a custom SessionStore

For production, implement aix.SessionStore (which is aix.SnapshotReader + aix.SnapshotWriter) against your database. Add aix.SnapshotSubscriber to support background/detach aborts.

type SnapshotReader[State any] interface {
	GetSnapshot(ctx context.Context, snapshotID string) (*SessionSnapshot[State], error)
	GetLatestSnapshot(ctx context.Context, sessionID string) (*SessionSnapshot[State], error)
}

type SnapshotWriter[State any] interface {
	// Atomically read the row at id (nil if absent), apply fn, and persist the
	// result. fn must be pure (it may be retried). Returning (nil, nil) skips
	// the write. The store owns identity (SnapshotID); the caller owns
	// timestamps and Status. An empty id means "mint a fresh one".
	SaveSnapshot(ctx context.Context, snapshotID string,
		fn func(existing *SessionSnapshot[State]) (*SessionSnapshot[State], error),
	) (*SessionSnapshot[State], error)
}

// Optional: enables detach/abort by observing status changes without polling.
type SnapshotSubscriber interface {
	OnSnapshotStatusChange(ctx context.Context, snapshotID string) <-chan SnapshotStatus
}

Contract notes: preserve SessionID across rewrites (a row's session never changes); default an empty Status to SnapshotStatusCompleted; keep CreatedAt/UpdatedAt caller-managed (persist them verbatim, never stamp them); GetLatestSnapshot returns the greatest-CreatedAt row, ties broken by SnapshotID. See the localstore implementations for a reference.