Struct TaskContext 

pub struct TaskContext { /* private fields */ }

Available on crate features functional and graph only.

Runtime context passed to #[entrypoint] and #[task] functions.

Provides access to workflow state, checkpointing, interrupt/resume, and progress streaming. Each task function receives a mutable reference to TaskContext enabling state reads, writes, event emission, and interrupt requests.

Example

use adk_graph::functional::TaskContext;

#[task]
async fn my_step(ctx: &mut TaskContext) -> Result<Value> {
    // Read state
    let count: i64 = ctx.get("counter").unwrap_or(0);

    // Write state
    ctx.set("counter", serde_json::json!(count + 1));

    // Emit progress
    ctx.emit(StreamEvent::custom("my_step", "progress", serde_json::json!({"count": count + 1})));

    Ok(serde_json::json!({"new_count": count + 1}))
}

Implementations

impl TaskContext

pub fn new( thread_id: String, state: HashMap<String, Value>, checkpointer: Arc<dyn Checkpointer>, event_tx: Sender<StreamEvent>, execution_log: Arc<RwLock<ExecutionLog>>, cancel_token: CancellationToken, schema: Option<StateSchema>, ) -> TaskContext

Create a new TaskContext.

Typically constructed by the macro-generated entrypoint, not by user code.

pub fn state(&self) -> &HashMap<String, Value>

Get the current workflow state (read-only).

pub fn get<T>(&self, key: &str) -> Option<T>

where T: DeserializeOwned,

Get a typed value from state.

Attempts to deserialize the value stored at key into type T. Returns None if the key does not exist or deserialization fails.

Example

let count: Option<i64> = ctx.get("counter");

pub fn set(&mut self, key: &str, value: impl Into<Value>)

Set a value in state.

If a StateSchema is configured, the update is applied using the appropriate reducer for the key. Otherwise the value is set directly (overwrite semantics).

Example

ctx.set("counter", serde_json::json!(42));

pub fn emit(&self, event: StreamEvent)

Emit a progress event to stream listeners.

Events are broadcast to all registered receivers. If no listeners are active the event is silently dropped.

Example

ctx.emit(StreamEvent::custom("my_task", "progress", json!({"pct": 50})));

pub async fn interrupt<T>(&self, message: &str) -> Result<T, GraphError>

where T: DeserializeOwned,

Interrupt execution and wait for external input.

Persists the current state as an interrupt checkpoint, emits an interrupted event, and suspends execution. When the workflow is resumed with an interrupt value, the value is deserialized into T and returned.

Errors

Returns FunctionalError::InterruptTypeMismatch if the resume value cannot be deserialized into T.

Returns FunctionalError::CheckpointFailed if persisting the interrupt checkpoint fails.

Example

let approval: bool = ctx.interrupt("Please approve this action").await?;

pub fn with_resume_values( self, resume_values: HashMap<String, Value>, ) -> TaskContext

Supplies values for interrupt sites, keyed by continuation key.

Re-invoke the entrypoint with these set to resume: an interrupt whose key is present returns the deserialized value instead of suspending.

Example

// First run suspends and reports its continuation key.
let Err(e) = run(ctx).await else { unreachable!() };

// Second run supplies the value under that key.
let ctx = ctx.with_resume_values(HashMap::from([
    ("interrupt-1".to_string(), serde_json::json!({ "approved": true })),
]));
let output = run(ctx).await?;

pub fn resume_values(&self) -> &HashMap<String, Value>

The values available to interrupt sites in this context.

pub fn thread_id(&self) -> &str

Get the thread identifier for this context.

pub fn cancel_token(&self) -> &CancellationToken

Get a reference to the cancellation token.

pub fn is_cancelled(&self) -> bool

Check if the workflow has been cancelled.

pub async fn current_step(&self) -> usize

Get the current step number from the execution log.

pub fn with_schema_validator( self, validator: StateSchemaValidator, ) -> TaskContext

Set a StateSchemaValidator for this context.

When set, the validator is used to validate initial state at workflow start and task output before applying reducers.

pub fn schema_validator(&self) -> Option<&StateSchemaValidator>

Get the schema validator, if configured.

pub fn validate_state(&self) -> Result<(), FunctionalError>

Validate the current state against the schema validator.

Called at workflow start to validate initial state.

Errors

Returns FunctionalError::SchemaValidation if validation fails.

pub fn validate_task_output( &self, output: &HashMap<String, Value>, ) -> Result<(), FunctionalError>

Validate task output against the schema validator.

Called after a task produces output, before applying reducers.

Errors

Returns FunctionalError::SchemaValidation if validation fails.

pub fn iteration_key(&mut self, task_name: &str) -> String

Generate a unique checkpoint key for a task inside a loop.

Each call to this method for the same task_name increments the iteration counter, producing keys like "step_a::iter_0", "step_a::iter_1", etc. Keys are deterministic from task name and iteration index.

Example

for item in items {
    let key = ctx.iteration_key("process_item");
    // key = "process_item::iter_0", "process_item::iter_1", ...
}

pub fn current_iteration(&self, task_name: &str) -> Option<usize>

Get the current iteration index for a task without incrementing.

Returns None if the task has not been called in a loop yet.

pub fn reset_iteration(&mut self, task_name: &str)

Reset the iteration counter for a task.

Useful when re-entering a loop (e.g., nested loops or retry).

pub fn reset_all_iterations(&mut self)

Reset all iteration counters.

pub fn route_to(&mut self, targets: &[&str])

Records the task names this task chose, for its caller to read.

Nothing in this crate acts on the value. The functional API runs your own control flow: #[entrypoint] calls your function once, and the awaits inside it are the order of execution. So a task states its choice here and the surrounding code reads it with Self::take_pending_route and calls what it names.

For routing the framework performs, build the workflow with StateGraph::add_conditional_edges, which resolves a route key against declared targets and dispatches to them.

Example

use std::collections::HashMap;
use std::sync::Arc;

use adk_graph::checkpoint::{Checkpointer, MemoryCheckpointer};
use adk_graph::functional::{ExecutionLog, TaskContext};
use tokio::sync::{RwLock, broadcast};
use tokio_util::sync::CancellationToken;

let (event_tx, _rx) = broadcast::channel(8);
let mut ctx = TaskContext::new(
    "thread-1".to_string(),
    HashMap::new(),
    Arc::new(MemoryCheckpointer::new()) as Arc<dyn Checkpointer>,
    event_tx,
    Arc::new(RwLock::new(ExecutionLog::new())),
    CancellationToken::new(),
    None,
);

// A task states which branches it chose.
ctx.route_to(&["process_a", "process_b"]);

// The surrounding code reads the choice and calls those tasks itself.
let chosen = ctx.take_pending_route().expect("a route was set");
assert_eq!(chosen, vec!["process_a".to_string(), "process_b".to_string()]);

// Reading it clears it, so the next task starts with no choice pending.
assert_eq!(ctx.take_pending_route(), None);

pub fn take_pending_route(&mut self) -> Option<Vec<String>>

Takes the task names recorded by Self::route_to, clearing them.

Returns None when no task set a route. Call this from the code that sequences your tasks; nothing in this crate calls it for you.