Subtasks

Subtasks connect a parent job to child jobs. Parent code receives a SubTaskHandler; child code completes that handler with a result or error.

Symbol	Import path	Purpose	Stability
`SubTaskHandler`	`statek.task` or `statek`	Parent-visible child job handle	Core
`SubTaskState`	`statek.task` or `statek`	Handler lifecycle enum	Core
`TaskError`	`statek.task` or `statek`	Stored subtask error text	Core
`create_sub_task`	`statek.task` or `statek`	Wrap an existing child job	Core
`create_new_job`	`statek.task` or `statek`	Create a ready child or standalone job	Core
`complete_sub_task`	`statek.task` or `statek`	Complete the current child job's handler	Core

`SubTaskState`

SubTaskState.WAITING
SubTaskState.STARTED
SubTaskState.COMPLETED
SubTaskState.ERROR

SubTaskState describes the parent-side handler state. It is separate from the child JobStatus.

`SubTaskHandler`

@db0.memo
class SubTaskHandler:
    def __init__(self, job: Job, id: Optional[Any] = None)

Represents a delegated child job from the parent job.

Attributes and properties

Member	Returns	Description
`job`	`Job`	Child job.
`id`	`Any	None`
`is_completed`	`bool`	Whether an outcome has been recorded.
`error`	`TaskError	None`
`result`	`Any	None`
`state`	`SubTaskState`	Derived handler state.
`complete(result=None, error=None)`	`None`	Records the outcome and notifies the parent job.
`get_log_message()`	`str`	LLM-facing notification text.

Raises

complete(...) raises RuntimeError if called twice and ValueError if both result and error are provided. str(handler) raises RuntimeError for unfinished or errored handlers.

from statek.task import create_new_job, create_sub_task
 
child = create_new_job(agent=researcher, parent_job=parent_job, topic="pricing")
handler = create_sub_task(child, id="pricing-research")

Job creation and completion helpers

`create_new_job`

def create_new_job(
    agent: Agent,
    shared_vars: dict[str, Any] | None = None,
    parent_job: Job | None = None,
    warmup_code: str | Sequence[str] | None = None,
    locale=None,
    caller_frame=None,
    **kwargs,
) -> Job

Creates a ready job with copied locals for referenced warmup variables, inherited locale when applicable, and inherited error handlers from the parent job.

`create_sub_task`

def create_sub_task(
    job: Job,
    handler_type: type[SubTaskHandler] = SubTaskHandler,
    **kwargs,
) -> SubTaskHandler

Creates a handler for an existing job and injects it into child job locals as sub_task_handler.

`complete_sub_task`

def complete_sub_task(result: Any | None = None, error: str | None = None) -> None

Completes the sub_task_handler in the current child job context.

Raises

complete_sub_task(...) raises RuntimeError outside a current job or when no sub_task_handler is registered, and TypeError when that local is not a SubTaskHandler.

from statek import complete_sub_task
 
complete_sub_task(result={"status": "ready"})
exit("done")

Related APIs: Tools, Jobs, Futures.

Tools Futures