Background execution

Background execution lets a client submit work, disconnect, and return later. It requires server-managed state because the server needs a snapshot to track progress, liveness, completion, failure, and cancellation.

Use background execution for work that may outlive the request or the browser tab, such as report generation, long research tasks, multi-step planning, or tool-heavy workflows. Keep normal foreground streaming for short turns where the user is actively waiting and foreground cancellation is enough.

Store support matters for background work. See Session stores for which stores support snapshot status changes and aborting detached work.

Server requirements

Configure a store and expose companion endpoints when using a remote client:

const reportAgent = ai.defineAgent({
  name: 'reportAgent',
  system: 'Create detailed research reports.',
  store,
});

app.post('/api/reportAgent', expressHandler(reportAgent));
app.post(
  '/api/reportAgent/getSnapshot',
  expressHandler(reportAgent.getSnapshotDataAction),
);
app.post(
  '/api/reportAgent/abort',
  expressHandler(reportAgent.abortAgentAction),
);

The runtime writes a pending snapshot and refreshes its heartbeat while work runs. If the heartbeat becomes stale, reads surface the snapshot as expired.

Detach from a turn

detach() submits a turn with detach: true. It returns after the server accepts the background work.

const chat = reportAgent.chat({ sessionId: 'report-123' });
const task = await chat.detach('Write the quarterly market report.');

savePendingSnapshot(task.snapshotId);

The chat updates its snapshotId to the pending snapshot ID. Store that ID so another process or browser session can inspect or abort the task.

Poll or wait

poll() yields snapshots until the task reaches a terminal status.

for await (const snapshot of task.poll({ intervalMs: 1000 })) {
  renderStatus(snapshot.status);

  if (snapshot.status === 'completed') {
    renderMessages(snapshot.state.messages);
  }
}

Use wait() when the caller can block:

const finalSnapshot = await task.wait({ intervalMs: 1000 });

if (finalSnapshot.status === 'failed') {
  showError(finalSnapshot.error);
}

Terminal statuses are completed, failed, aborted, and expired.

Use poll() for UI progress because it lets you render every status change. Use wait() for server code, tests, or short-lived command-line tools where blocking is acceptable. Store the pending snapshot ID before navigating away from the page so another client session can reconnect.

Reconnect by snapshot ID

If the process that started the task no longer has the DetachedTask, read the stored snapshot ID and resume from it:

const snapshot = await reportAgent.getSnapshot({ snapshotId });

if (snapshot?.status === 'completed') {
  const chat = await reportAgent.loadChat({ snapshotId });
  await chat.send('Summarize the report in three bullets.');
}

Only completed snapshots can be resumed.

Abort work

await task.abort();

Or abort directly from the agent:

await reportAgent.abort(snapshotId);

Abort flips a pending snapshot to aborted. The background worker observes the status change and cancels the work.

Requirements

Background execution requires a server-managed agent and a store that implements status subscriptions. localstore.FileSessionStore supports this.

Choose background execution when the caller should receive a pending snapshot immediately and let the agent continue on the server. For local command-line tools or services that can keep the connection open, a normal streaming Connect call is often simpler.

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, err := localstore.NewFileSessionStore[ReportState]("./.genkit/snapshots/reports")
if err != nil {
    // Fails if the snapshot directory cannot be created or is not writable.
    log.Fatalf("open report store: %v", err)
}

agent := genkitx.DefineAgent(g, "report",
    aix.InlinePrompt{
        ai.WithModelName("googleai/gemini-flash-latest"),
        ai.WithSystem("Create detailed research reports."),
    },
    aix.WithSessionStore(store),
)

The default HTTP abort route is /agents/{name}/abort. The basic-agents-server sample walks the whole background lifecycle over plain HTTP with curl: detach, poll, and abort.

Detach locally

Go has no handle type for a background task. A detached run is identified only by the pending snapshot ID on the output, which you poll with Agent.GetSnapshot or Agent.GetLatestSnapshot and cancel with Agent.Abort. Store that ID before the process that started the work goes away.

With a local AgentConnection, send an input that has Detach: true.

conn, err := agent.Connect(ctx)
if err != nil {
    // Connect fails if the init payload is rejected, such as a sessionId on a
    // client-managed agent.
    return fmt.Errorf("connect to agent: %w", err)
}

if err := conn.Send(&aix.AgentInput{
    Message: ai.NewUserTextMessage("Write the quarterly report."),
    Detach:  true,
}); err != nil {
    return fmt.Errorf("send detached input: %w", err)
}

out, err := conn.Output()
if err != nil {
    // The detached turn could not be started; a started one resolves in-band.
    return fmt.Errorf("read detached output: %w", err)
}

fmt.Println(out.FinishReason)
fmt.Println(out.SnapshotID)

A detached output has FinishReason set to aix.AgentFinishReasonDetached and SnapshotID set to the pending snapshot.

conn.Detach() sends the same directive with no message, for leaving a turn that is already under way:

if err := conn.Detach(); err != nil {
    return fmt.Errorf("detach: %w", err)
}

A bare detach does not start an extra turn. To ride a final input on the detach, set Message alongside Detach as in the first example.

The client stream is suppressed the moment the detach directive is read, so nothing the background turn produces afterwards reaches conn.Receive(). Session-level side effects still apply: an artifact sent through Responder.SendArtifact still lands in the final snapshot’s state, so agent code does not have to branch on detach.

Wait for completion

Read a snapshot by ID with agent.GetSnapshot, or a session’s most recent one with agent.GetLatestSnapshot(ctx, sessionID).

snap, err := agent.GetSnapshot(ctx, snapshotID)
if err != nil {
    return fmt.Errorf("read snapshot: %w", err)
}

fmt.Println(snap.Status)

When the store implements aix.SnapshotSubscriber, subscribe to status changes instead of polling, then read the final snapshot.

subscriber := agent.Store().(aix.SnapshotSubscriber)
statusCh := subscriber.OnSnapshotStatusChange(ctx, snapshotID)

for status := range statusCh {
    if status != aix.SnapshotStatusPending {
        break
    }
}

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

If the store does not support subscriptions, poll agent.GetSnapshot(ctx, snapshotID) until Status leaves pending. Terminal statuses are completed, failed, aborted, and expired. Only a completed snapshot is resumable.

The basic-agents sample drives this from its CLI: /detach leaves a turn running, and returning to the agent later waits on the pending snapshot.

Abort work

status, err := agent.Abort(ctx, snapshotID)
if err != nil {
    // A non-nil error is an abort the store could not attempt, not a snapshot
    // that was already terminal (that returns the terminal status, nil).
    return fmt.Errorf("abort snapshot: %w", err)
}

fmt.Println(status)

Abort is a no-op for missing snapshots and for snapshots that already reached a terminal status. Client-managed agents return FAILED_PRECONDITION because there is no server snapshot to cancel.

Server requirements

Configure a store and expose companion endpoints when serving a remote agent:

final reportAgent = ai.defineAgent(
  name: 'reportAgent',
  system: 'Create detailed research reports.',
  store: FileSessionStore('.sessions'),
);

void main() {
  final router = Router();
  router.post('/api/reportAgent', shelfHandler(reportAgent.action));
  router.post('/api/reportAgent/getSnapshot', shelfHandler(reportAgent.getSnapshotDataAction));
  router.post('/api/reportAgent/abort', shelfHandler(reportAgent.abortAgentAction));
}

The runtime writes a pending snapshot and refreshes its heartbeat while the background thread processes the turn.

Detach from a turn

Submit a background task from the client using detach() on the AgentChat:

final chat = reportAgent.chat(sessionId: 'report-123');
final task = await chat.detach(text: 'Write the quarterly market report.');

// Save snapshotId so you can poll or abort it later
final snapshotId = task.snapshotId;

Poll or wait

Use poll() to yield status snapshots over a Stream until the task reaches a terminal status:

await for (final snapshot in task.poll(interval: Duration(milliseconds: 1500))) {
  print('Current Status: ${snapshot.status?.value}');

  if (snapshot.status?.value == 'completed') {
    final report = snapshot.messages.last.content.first.text;
    print(report);
  }
}

Use wait() to block execution as a Future until completion:

final finalSnapshot = await task.wait(interval: Duration(milliseconds: 1500));

if (finalSnapshot.status?.value == 'failed') {
  print('Task failed: ${finalSnapshot.error?.message}');
}

Terminal statuses are completed, failed, aborted, and expired.

Reconnect by snapshot ID

To reconnect and inspect or resume a detached task from a different client process, read the stored snapshot ID and load it:

final snapshot = await reportAgent.getSnapshot(snapshotId: snapshotId);

if (snapshot?.status?.value == 'completed') {
  final chat = await reportAgent.loadChat(snapshotId: snapshotId);
  final res = await chat.send(text: 'Summarize this report.');
  print(res.text);
}

Only completed snapshots can be resumed.

Abort work

Cancel a pending task from the client using task.abort():

await task.abort();

Or abort directly by snapshot ID from the AgentApi handle:

await reportAgent.abort(snapshotId);

Aborting shifts the pending snapshot status to aborted. The background worker observes this change and safely terminates the turn loop.

Server requirements

Configure a store and expose companion endpoints when serving a remote agent:

from fastapi import FastAPI
from genkit.agent import FileSessionStore
from genkit_fastapi import serve_agent

report_agent = ai.define_agent(
    name='reportAgent',
    model='googleai/gemini-flash-latest',
    system='Create detailed research reports.',
    store=FileSessionStore('.sessions'),
)

app = FastAPI()
app.include_router(serve_agent(report_agent), prefix='/api')

The runtime writes a pending snapshot and refreshes its heartbeat while the background work processes the turn.

Detach from a turn

Submit a background task from the client using detach() on the AgentChat:

chat = report_agent.chat(session_id='report-123')
task = await chat.detach('Write the quarterly market report.')

# Save snapshot_id so you can poll or abort it later
snapshot_id = task.snapshot_id

Poll or wait

Use poll() to yield status snapshots until the task reaches a terminal status:

from genkit.agent import SnapshotStatus

async for snapshot in task.poll(interval=1.5):
    print('Current status:', snapshot.status)
    if snapshot.status == SnapshotStatus.COMPLETED:
        print(snapshot.state)

Use wait() to block until completion:

final_snapshot = await task.wait(interval=1.5)
if final_snapshot.status == SnapshotStatus.FAILED:
    print('Task failed:', final_snapshot.error.message if final_snapshot.error else None)

Terminal statuses are completed, failed, aborted, and expired.

Reconnect by snapshot ID

To reconnect and inspect or resume a detached task from a different client process, read the stored snapshot ID and load it:

snapshot = await report_agent.get_snapshot(snapshot_id=snapshot_id)

if snapshot and snapshot.status == SnapshotStatus.COMPLETED:
    chat = await report_agent.load_chat(snapshot_id=snapshot_id)
    res = await chat.send('Summarize this report.')
    print(res.text)

Only completed snapshots can be resumed.

Abort work

Cancel a pending task from the client using task.abort():

await task.abort()

Or abort directly by snapshot ID from the agent handle:

await report_agent.abort(snapshot_id)

Aborting sets the pending snapshot status to aborted. The background worker stops the turn when it observes that change. Long-running tools should check ctx.abort_signal.is_set() on ToolRunContext so they can exit cleanly.