Sub-Tasks¶

Sub-tasking lets a task spawn other tasks and pause until they finish. Each child is a normal TaskState with a parent_id, dispatched through the same Docket queue infrastructure as root tasks, so it inherits durability, retry, streaming, and human-in-the-loop for free.

Sub-tasking is opt-in. When disabled, ctx.spawn* raises NotImplementedError and no extra middleware runs. A single-shot agent stays flat.

Enabling sub-tasks¶

from redis_agent_kit import AgentKit, SubTaskConfig

kit = AgentKit(
    agent_callable=my_agent,
    queue_name="my_agent",
    subtask_config=SubTaskConfig(enabled=True),
)

Or via environment: RAK_SUBTASK__ENABLED=true.

SubTaskConfig knobs:

Field	Default	Purpose
`enabled`	`False`	Master switch — gates all `ctx.spawn*` methods.
`max_depth`	`5`	Hard cap on nesting (root is depth 0).
`max_children_per_task`	`32`	Per-task fan-out limit.
`default_on_failure`	`"propagate"`	How `spawn_and_wait` reacts to a failed child. `"collect"` returns an error dict; `"ignore"` returns `None`.
`cascade_cancel`	`True`	Whether `kit.cancel_task(...)` cascades to descendants by default.
`child_queue_name`	`None`	Docket queue for children. Defaults to `<queue_name>_children`.
`isolate_memory`	`False`	When `True`, each child gets its own `session_id` instead of sharing the parent's.

Running workers¶

Children dispatch to a separate queue (<queue_name>_children by default), so a runaway parent can't starve root tasks. Run two worker pools:

rak worker --name my_agent --tasks my_module:my_agent_tasks            # root queue
rak worker --name my_agent_children --tasks my_module:my_agent_tasks   # child queue

Both pools use the same handler module — only the queue differs. Scale child workers independently based on observed fan-out.

Spawning children¶

Inside an agent handler, use the TaskContext API:

async def my_agent(ctx):
    # Fire and forget — caller doesn't wait.
    child = await ctx.spawn(
        message="Summarize arxiv:2401.0123",
        spawn_key="paper_summary",
    )
    return {"queued": child.task_id}

ctx.spawn writes the child to Redis, indexes it under the parent, and dispatches it on the children queue. It returns the freshly-written TaskState.

Waiting on a child — `spawn_and_wait`¶

The load-bearing primitive. spawn_and_wait parks the parent until the child finishes and returns the child's result on resume:

async def my_agent(ctx):
    summary = await ctx.spawn_and_wait(
        message="Summarize arxiv:2401.0123",
        spawn_key="paper_summary",
        context={"paper_id": "2401.0123"},
    )
    return {"answer": f"Paper says: {summary['result']}"}

What happens on the first invocation:

The framework spawns the child (status QUEUED).
The parent transitions to AWAITING_CHILDREN and its handler is aborted cleanly.
The children worker picks up the child and runs it.
When the child terminates, the fan-in hook decrements the parent's pending counter and re-enqueues the parent on the root queue.
The parent handler runs again. spawn_and_wait sees the existing child via spawn_key, reads its result from Redis, and returns it.

The handler runs from the top on resume. Use @side_effect to cache expensive non-spawn work (LLM calls, API writes) so it doesn't repeat. See the Input Handling guide for the pattern.

Fan-out — `ctx.gather`¶

Concurrent fan-out with one re-entry:

from redis_agent_kit import SpawnSpec

async def research_agent(ctx):
    specs = [
        SpawnSpec(
            message=f"Read paper {paper_id}",
            spawn_key=f"paper_{paper_id}",
            context={"paper_id": paper_id},
        )
        for paper_id in ctx.context["paper_ids"]
    ]
    results = await ctx.gather(specs)
    # results[i] is the dict returned by spec[i]'s handler
    return {"summaries": results}

gather spawns all N children, parks the parent in AWAITING_CHILDREN, and re-enqueues the parent exactly once after the last child completes — atomic DECR on the pending counter guarantees this even under concurrent terminations.

If a spec omits spawn_key, gather assigns one based on position (_gather:0, _gather:1, …) so resumption is still idempotent.

`spawn_key` — when to set it¶

spawn_key is the de-duplication anchor on re-entry. The rule:

Set it whenever the parent will ever pause. Without it, the parent will spawn a fresh child on every resume.
Skip it for fire-and-forget work where the parent never re-enters.
Omit it inside gather — gather keys positionally.

The framework warns once with a UserWarning if spawn_and_wait is called without spawn_key inside a handler that has previously paused. Pass spawn_key=None explicitly to silence the warning if you really mean "spawn a fresh child every time."

Failure handling¶

spawn_and_wait (and gather) accept on_failure:

Mode	Behavior
`"propagate"` (default)	Raises `ChildTaskFailed` — by default the parent fails too.
`"collect"`	Returns `{"error": ..., "task_id": ..., "spawn_key": ..., "status": ...}` instead of raising.
`"ignore"`	Returns `None` for the failed child.

from redis_agent_kit import ChildTaskFailed

async def my_agent(ctx):
    try:
        result = await ctx.spawn_and_wait("step", spawn_key="step")
    except ChildTaskFailed as e:
        await ctx.emitter.emit(f"Child {e.spawn_key} failed: {e.error}")
        return {"failed": True}
    return {"result": result}

For per-call overrides:

results = await ctx.gather(specs, on_failure="collect")
# results contains a mix of success dicts and error dicts; check each

Human-in-the-loop in a child¶

Children get the same input-pause mechanism as root tasks. The parent stays in AWAITING_CHILDREN while the child sits in AWAITING_INPUT:

async def confirmation_child(ctx):
    task = await ctx.kit.task_manager.get_task(ctx.task_id)
    if task and task.input_response:
        return {"confirmed": task.input_response["confirm"]}

    await ctx.kit.task_manager.request_input(
        ctx.task_id,
        prompt="Proceed?",
        json_schema={
            "type": "object",
            "properties": {"confirm": {"type": "boolean"}},
            "required": ["confirm"],
        },
    )
    return {}  # handler returns; framework leaves task in AWAITING_INPUT

A subtask_input_required event fires on the parent's channel so dashboards watching only the root task can route the HITL prompt to a human. When the user submits input, the child runs, terminates, and fan-in re-enqueues the parent.

Memory isolation¶

By default, children share the parent's session_id, so the user-facing conversation thread reflects what every subagent discovered. Set isolate_memory=True for research/tool agents whose intermediate chatter would pollute the conversation:

# Globally
kit = AgentKit(
    agent_callable=my_agent,
    subtask_config=SubTaskConfig(enabled=True, isolate_memory=True),
)

# Per spawn (always wins)
child = await ctx.spawn(
    message="Search arxiv",
    isolate_memory=True,
)

Isolated children stay linked to the parent through parent_id, not session membership.

Cancellation¶

Cancelling a parent cancels its descendants by default:

await kit.cancel_task(parent_id)              # uses SubTaskConfig.cascade_cancel
await kit.cancel_task(parent_id, cascade=True)  # explicit
await kit.cancel_task(parent_id, cascade=False) # cancel parent only

Cancellation flips terminal status (CANCELLED) on each task; in-flight workers observe the change and stop dispatching new children.

Streaming events¶

When streaming is enabled, four sub-task event types fire on the parent's task channel (and the session/global channels if configured), so a single SSE subscription to the root task sees the whole tree progress:

Event	When
`subtask_spawned`	A child has been dispatched.
`subtask_done`	A child reached `DONE`.
`subtask_failed`	A child reached `FAILED` or `CANCELLED`.
`subtask_input_required`	A child requested user input.

Payload (under metadata):

{
  "parent_task_id": "01HXYZ...",
  "child_task_id": "01HABC...",
  "spawn_key": "paper_summary",
  "status": "done"
}

REST API¶

When sub-tasking is enabled and you're using create_app, three task endpoints expose the tree:

Method	Path	Description
`GET`	`/tasks/{id}/children`	Direct children with status and spawn_key.
`GET`	`/tasks/{id}/tree?max_depth=10`	Recursive descendant tree.
`POST`	`/tasks/{id}/cancel?cascade=true`	Cancel and cascade.

Cancellation uses the kit's SubTaskConfig.cascade_cancel as the default when the cascade query parameter is omitted.

Inspecting from the handler¶

The handler can introspect the tree without going through Redis directly:

async def my_agent(ctx):
    # Read a completed child's result by spawn_key (None if still running).
    prior = await ctx.child_result("paper_summary")
    if prior is not None:
        return {"reused": prior}

    # Walk children:
    for child in await ctx.kit.task_manager.list_children(ctx.task_id):
        ...

ctx.parent_id, ctx.child_ids, and ctx.depth are also populated.

Data model additions¶

TaskState gains five fields when sub-tasking is enabled (all default to neutral values so disabled kits are unaffected):

Field	Type	Description
`parent_id`	`str \\| None`	Parent task's ID; `None` for root tasks.
`spawn_key`	`str \\| None`	Stable label set by the parent at spawn time.
`child_ids`	`list[str]`	Ordered, append-only list of children.
`pending_children`	`int`	Non-terminal child count (used by fan-in).
`depth`	`int`	0 for root, +1 per nesting level.

TaskStatus adds AWAITING_CHILDREN — the equivalent of AWAITING_INPUT but waiting on child completion rather than user input. A child that itself paused on input keeps the parent in AWAITING_CHILDREN (its pending counter doesn't decrement until the child reaches a terminal status).

Out of scope (deliberately)¶

DAG scheduling between siblings. Sequential spawn_and_wait calls, or passing one child's result as another's input, cover the use cases without a dedicated graph engine.
Cross-session sub-tasks. Children inherit the parent's session_id (or get a fresh one with isolate_memory=True); arbitrary session re-attachment is not exposed.
Persistent tree visualization. The /tasks/{id}/tree endpoint exposes the data; rendering is a downstream concern.

Example¶

See examples/subtasking/ for a runnable agent that spawns research sub-tasks, gathers their results, and serves the tree over the REST API.