Package com.google.genkit.ai.agent.internal

Class DetachController

java.lang.Object

com.google.genkit.ai.agent.internal.DetachController

public final class DetachController
extends Object

Runs a detached agent turn: writes a PENDING snapshot, suppresses streaming, starts a heartbeat, runs the turn on a background daemon thread, and finalizes the pending snapshot to a terminal status when the turn completes.

Pragmatic approach. The bidi handler returns AgentOutput synchronously after the first detach input. True multi-turn continuation inside the live bidi stream would require keeping the stream open while running in the background, which is incompatible with the synchronous handler contract. Instead, detach(com.google.genkit.ai.agent.SessionRunner<S>, com.google.genkit.ai.agent.SessionStore<S>, com.google.genkit.ai.agent.SessionStoreOptions, com.google.genkit.ai.agent.internal.StreamEmitter<S>, java.util.concurrent.atomic.AtomicBoolean, com.google.genkit.ai.agent.internal.DetachController.DetachedTurn<S>) reserves the snapshot id, kicks off the single turn on a shared daemon executor (streaming suppressed), and returns immediately. The client polls getSnapshot until the snapshot reaches a terminal status. This matches the design spec §6.4 and the Go/JS reference behavior for the write side.

Abort race. Both the heartbeat and the finalize step go through AbortAwareMutator, and the finalize mutator additionally never overwrites an ABORTED row. If the abort companion flips the pending snapshot to ABORTED while the background turn is still running, the heartbeat becomes a no-op (status != PENDING) and the finalize declines the write, so the ABORTED terminal state is preserved.

Threading. All background work uses shared daemon-thread executors so the JVM can exit without blocking and no threads are leaked across invocations.

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
static interface	DetachController.DetachedTurn<S>	The body of a detached turn.

Method Summary

Modifier and Type	Method	Description
static <S> String	detach(SessionRunner<S> runner, SessionStore<S> store, SessionStoreOptions opts, StreamEmitter<S> emitter, AtomicBoolean abortSignal, DetachController.DetachedTurn<S> turnBody)	Handles a detached turn for a server-managed agent.
static long	setHeartbeatIntervalMillisForTest(long millis)	Overrides the heartbeat interval (test hook).

Methods inherited from class java.lang.Object

equals, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Method Details

setHeartbeatIntervalMillisForTest

public static long setHeartbeatIntervalMillisForTest(long millis)

Overrides the heartbeat interval (test hook). Returns the previous value so tests can restore it. Public so tests outside this package can make heartbeat timing deterministic; it is not part of the supported API surface.

Parameters:

millis - the new interval in milliseconds (must be positive)

Returns:

the previous interval

detach

public static <S> String detach(SessionRunner<S> runner,
 SessionStore<S> store,
 SessionStoreOptions opts,
 StreamEmitter<S> emitter,
 AtomicBoolean abortSignal,
 DetachController.DetachedTurn<S> turnBody)

Handles a detached turn for a server-managed agent.

1. Reserves a snapshot id and writes a PENDING snapshot (cumulative state baseline).
2. Suppresses the stream emitter so the detached run emits no chunks.
3. Schedules a heartbeat that refreshes heartbeatAt while the snapshot stays pending.
4. Runs turnBody on a shared daemon thread, then finalizes the snapshot (abort-aware) to COMPLETED/FAILED with the cumulative state and stops the heartbeat.

Abort signal registration. While the turn is PENDING, abortSignal (the same AtomicBoolean handed to the turn's AgentFnContext) is registered in PendingAbortRegistry under the reserved snapshot id, so an external Agent.abort(snapshotId) call can flip it while the background turn is still running. The registration is removed in a finally block when the turn finishes, regardless of outcome.

Type Parameters:

S - the type of custom session state

Parameters:

runner - the session runner (shared with the background thread; Session is synchronized)

store - the (non-null) server-side store

opts - store options

emitter - the stream emitter to suppress

abortSignal - the abort signal handed to this turn's AgentFnContext; registered under the reserved snapshot id for the lifetime of the background run

turnBody - the turn body to execute in the background

Returns:

the reserved snapshot id for the pending detached run