Struct TaskContext 

pub struct TaskContext { /* private fields */ }

Available on crate feature 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

Available on crate feature functional only.

Create a new TaskContext.

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

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

Available on crate feature functional only.

Get the current workflow state (read-only).

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

where T: DeserializeOwned,

Available on crate feature functional only.

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>)

Available on crate feature functional only.

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)

Available on crate feature functional only.

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,

Available on crate feature functional only.

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

Available on crate feature functional only.

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>

Available on crate feature functional only.

The values available to interrupt sites in this context.

pub fn thread_id(&self) -> &str

Available on crate feature functional only.

Get the thread identifier for this context.

pub fn cancel_token(&self) -> &CancellationToken

Available on crate feature functional only.

Get a reference to the cancellation token.

pub fn is_cancelled(&self) -> bool

Available on crate feature functional only.

Check if the workflow has been cancelled.

pub async fn current_step(&self) -> usize

Available on crate feature functional only.

Get the current step number from the execution log.

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

Available on crate feature functional only.

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>

Available on crate feature functional only.

Get the schema validator, if configured.

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

Available on crate feature functional only.

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>

Available on crate feature functional only.

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

Available on crate feature functional only.

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>

Available on crate feature functional only.

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)

Available on crate feature functional only.

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)

Available on crate feature functional only.

Reset all iteration counters.

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

Available on crate feature functional only.

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>>

Available on crate feature functional only.

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.