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.

Detach locally

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 detached and SnapshotID set to the pending snapshot.

Wait for completion

When the store implements aix.SnapshotSubscriber, subscribe to status changes and 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.

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.