Enum Task 

pub enum Task {
    Sync(SyncFn),
    Async(AsyncFn),
    SyncIter(SyncIterFn),
    AsyncStream(AsyncStreamFn),
    SyncBatch(SyncBatchFn),
    AsyncBatch(AsyncBatchFn),
    SyncIterBatch(SyncIterBatchFn),
    AsyncStreamBatch(AsyncStreamBatchFn),
}

A single reusable unit of work in a cognee pipeline.

Variant	Execution	Input	Output
Task::Sync	blocking	single value	single value
Task::Async	non-blocking	single value	single value
Task::SyncIter	blocking	single value	lazy iterator
Task::AsyncStream	non-blocking	single value	async stream
Task::SyncBatch	blocking	slice of values	single value
Task::AsyncBatch	non-blocking	slice of values	single value
Task::SyncIterBatch	blocking	slice of values	lazy iterator
Task::AsyncStreamBatch	non-blocking	slice of values	async stream

Single-value variants are called once per item. Batch variants receive a &[Box<dyn Value>] slice of items accumulated up to the task’s batch_size. The pipeline executor detects which kind the next task is and routes accordingly.

Variants

Sync(SyncFn)

Async(AsyncFn)

SyncIter(SyncIterFn)

AsyncStream(AsyncStreamFn)

SyncBatch(SyncBatchFn)

AsyncBatch(AsyncBatchFn)

SyncIterBatch(SyncIterBatchFn)

AsyncStreamBatch(AsyncStreamBatchFn)

Implementations

impl Task

pub fn is_batch(&self) -> bool

Returns true if this task accepts a batch slice rather than a single value.

pub fn python_task_type(&self) -> &'static str

Python-compat label used in the ${task_type} Task Started/ Completed/Errored analytics event names.

Mirrors Python’s tasks/task.py:194-207 inspect.isasyncgenfunction / iscoroutinefunction branch:

Rust variant	Python label
Task::Sync, Task::SyncBatch	"Function"
Task::Async, Task::AsyncBatch	"Coroutine"
Task::SyncIter, Task::SyncIterBatch	"Generator"
Task::AsyncStream, Task::AsyncStreamBatch	"Async Generator"

The match is intentionally exhaustive (no wildcard arm) so that adding a new Task::* variant fails the build until the analytics mapping is decided.

impl Task

pub fn sync<F>(f: F) -> Self

where F: Fn(Arc<dyn Value>, Arc<TaskContext>) -> Result<Arc<dyn Value>, TaskError> + Send + Sync + 'static,

Create a Task::Sync from a raw closure.

pub fn async_fn<F>(f: F) -> Self

where F: Fn(Arc<dyn Value>, Arc<TaskContext>) -> BoxFuture<'static, Result<Arc<dyn Value>, TaskError>> + Send + Sync + 'static,

Create a Task::Async from a raw closure returning a BoxFuture.

pub fn sync_iter<F>(f: F) -> Self

where F: Fn(Arc<dyn Value>, Arc<TaskContext>) -> Result<ValueIter, TaskError> + Send + Sync + 'static,

Create a Task::SyncIter from a raw closure returning a ValueIter.

pub fn async_stream<F>(f: F) -> Self

where F: Fn(Arc<dyn Value>, Arc<TaskContext>) -> Result<ValueStream, TaskError> + Send + Sync + 'static,

Create a Task::AsyncStream from a raw closure returning a ValueStream.

pub fn sync_batch<F>(f: F) -> Self

where F: for<'a> Fn(&'a [Box<dyn Value>], Arc<TaskContext>) -> Result<Arc<dyn Value>, TaskError> + Send + Sync + 'static,

Create a Task::SyncBatch from a raw closure.

pub fn async_batch<F>(f: F) -> Self

where F: for<'a> Fn(&'a [Box<dyn Value>], Arc<TaskContext>) -> BoxFuture<'static, Result<Arc<dyn Value>, TaskError>> + Send + Sync + 'static,

Create a Task::AsyncBatch from a raw closure returning a BoxFuture.

pub fn sync_iter_batch<F>(f: F) -> Self

where F: for<'a> Fn(&'a [Box<dyn Value>], Arc<TaskContext>) -> Result<ValueIter, TaskError> + Send + Sync + 'static,

Create a Task::SyncIterBatch from a raw closure returning a ValueIter.

pub fn async_stream_batch<F>(f: F) -> Self

where F: for<'a> Fn(&'a [Box<dyn Value>], Arc<TaskContext>) -> Result<ValueStream, TaskError> + Send + Sync + 'static,

Create a Task::AsyncStreamBatch from a raw closure returning a ValueStream.

pub fn sync_typed<I, O, F>(f: F) -> Self

where I: Value, O: Value, F: Fn(&I, Arc<TaskContext>) -> Result<Box<O>, TaskError> + Send + Sync + 'static,

Create a Task::Sync from a typed closure.

Task::sync_typed(|input: &MyInput, ctx| {
    Ok(Box::new(process(input)))
})

pub fn async_fn_typed<I, O, F>(f: F) -> Self

where I: Value, O: Value, F: Fn(&I, Arc<TaskContext>) -> BoxFuture<'static, Result<Box<O>, TaskError>> + Send + Sync + 'static,

Create a Task::Async from a typed closure returning a 'static future.

Data needed inside the async block must be copied/cloned before it:

Task::async_fn_typed(|input: &MyInput, ctx| {
    let id = input.id;  // copy before async block
    Box::pin(async move {
        Ok(Box::new(fetch(id).await?))
    })
})

pub fn sync_iter_typed<I, O, F, Iter>(f: F) -> Self

where I: Value, O: Value, F: Fn(&I, Arc<TaskContext>) -> Result<Iter, TaskError> + Send + Sync + 'static, Iter: Iterator<Item = Box<O>> + Send + 'static,

Create a Task::SyncIter from a typed closure returning a concrete iterator. The iterator must be 'static (may not borrow the input).

Task::sync_iter_typed(|input: &Document, ctx| {
    let chunks = split(input.text.clone());
    Ok(chunks.into_iter().map(Box::new))
})

pub fn async_stream_typed<I, O, F, S>(f: F) -> Self

where I: Value, O: Value, F: Fn(&I, Arc<TaskContext>) -> Result<S, TaskError> + Send + Sync + 'static, S: Stream<Item = Box<O>> + Send + 'static,

Create a Task::AsyncStream from a typed closure returning a concrete stream. The stream must be 'static.

Task::async_stream_typed(|input: &DatasetId, ctx| {
    let id = *input;
    Ok(stream_chunks(id))
})

pub fn sync_batch_typed<I, O, F>(f: F) -> Self

where I: Value, O: Value, F: for<'a> Fn(&'a [&'a I], Arc<TaskContext>) -> Result<Box<O>, TaskError> + Send + Sync + 'static,

Create a Task::SyncBatch from a typed closure receiving &[&I].

Task::sync_batch_typed(|chunks: &[&DocumentChunk], ctx| {
    Ok(Box::new(embed_all(chunks)))
})

pub fn async_batch_typed<I, O, F>(f: F) -> Self

where I: Value, O: Value, F: for<'a> Fn(&'a [&'a I], Arc<TaskContext>) -> BoxFuture<'static, Result<Box<O>, TaskError>> + Send + Sync + 'static,

Create a Task::AsyncBatch from a typed closure returning a 'static future.

Data needed inside the async block must be copied/cloned before it:

Task::async_batch_typed(|chunks: &[&DocumentChunk], ctx| {
    let texts: Vec<String> = chunks.iter().map(|c| c.text.clone()).collect();
    Box::pin(async move {
        Ok(Box::new(embed_batch(texts).await?))
    })
})

pub fn sync_iter_batch_typed<I, O, F, Iter>(f: F) -> Self

where I: Value, O: Value, F: for<'a> Fn(&'a [&'a I], Arc<TaskContext>) -> Result<Iter, TaskError> + Send + Sync + 'static, Iter: Iterator<Item = Box<O>> + Send + 'static,

Create a Task::SyncIterBatch from a typed closure returning a concrete iterator.

pub fn async_stream_batch_typed<I, O, F, S>(f: F) -> Self

where I: Value, O: Value, F: for<'a> Fn(&'a [&'a I], Arc<TaskContext>) -> Result<S, TaskError> + Send + Sync + 'static, S: Stream<Item = Box<O>> + Send + 'static,

Create a Task::AsyncStreamBatch from a typed closure returning a concrete stream.

pub fn call(&self, input: Arc<dyn Value>, ctx: Arc<TaskContext>) -> TaskCall

Call this task with a single input value.

Panics if called on a batch variant — use Task::call_batch for those. The task is Fn, so &self suffices — the same task object handles every input in a fan-out scenario and every retry attempt.

pub fn parallel(tasks: Vec<Task>) -> Self

Build a task that runs multiple sub-tasks concurrently on the same input.

Semantics (matching Python run_tasks_parallel):

• Each sub-task receives Arc::clone(&input) and the shared context.
• All sub-tasks run concurrently via futures::future::join_all.
• If any sub-task fails, the whole parallel task fails with that error.
• On success, returns the result of the last sub-task (by position).

Only single-value (Sync / Async) sub-tasks are supported. Iter/stream sub-tasks inside a parallel group don’t have well-defined “last result” semantics and will panic at call time.

pub fn call_batch( &self, items: &[Box<dyn Value>], ctx: Arc<TaskContext>, ) -> TaskCall

Call this batch task with a slice of accumulated values.

Panics if called on a single-value variant — use Task::call for those.

Trait Implementations

impl From<Task> for TaskInfo

fn from(task: Task) -> Self

Converts to this type from the input type.

impl<I: Value, O: Value> From<TypedTask<I, O>> for Task

fn from(typed: TypedTask<I, O>) -> Self

Erase I and O, producing the type-erased Task.

Delegates to the corresponding Task::sync_typed / Task::async_fn_typed / … constructor, reusing their downcast logic.