task_manager

Fire-and-forget task manager for tool execution.

Wraps tool handler coroutines with a configurable timeout. If a tool completes within the timeout its result is returned inline. Otherwise the coroutine continues as a background asyncio.Task and a JSON envelope containing a task ID is returned so the LLM can poll for results later via the check_task tool.

Background tasks may still hold the ToolContext injected into the tool and mutate it after the enclosing tool-calling loop has moved on. Prefer no_background on tools that perform persistent ctx side effects the model must observe in-order.

Output redirect

Any backgrounded task can have its result automatically delivered to a channel on any platform when it finishes. Call TaskManager.set_output_redirect() (or use the redirect_task tool) to configure this.

class task_manager.TaskStatus(*values)[source]

Bases: str, Enum

Lifecycle states for a tracked background tool task.

A string-valued enum (so members serialize directly into JSON envelopes and Redis fields as their plain values) covering the three terminal-or-not states a task can be in: RUNNING while the coroutine is in flight, COMPLETED on success, and FAILED on cancellation or exception. Used throughout TaskManager and persisted into the sg:task:{task_id} Redis hash and the legacy JSON envelopes that the check_task tool reads back.

RUNNING = 'running'
COMPLETED = 'completed'
FAILED = 'failed'
class task_manager.TaskRecord(task_id, tool_name, status, created_at=<factory>, result=None, error=None, user_id='', channel_id='', platform='', asyncio_task=None, redirect_channel_id='', redirect_platform='', redirect_adapter=None, redirect_max_chars=0)[source]

Bases: object

In-memory bookkeeping for one tracked background tool task.

Captures everything TaskManager needs to track, persist, and deliver a backgrounded tool invocation: identity and origin (task_id, tool_name, user_id, channel_id, platform), the live asyncio.Task handle, the current TaskStatus, and the captured result or error once it finishes. The trailing output-redirect fields hold an optional delivery target — channel, platform, resolved PlatformAdapter, and a character cap — so a finished task’s result can be pushed to a chat channel. Created by TaskManager.execute() and mutated in place by TaskManager._on_task_done().

Parameters:

  • task_id (str)

  • tool_name (str)

  • status (TaskStatus)

  • created_at (float)

  • result (str | None)

  • error (str | None)

  • user_id (str)

  • channel_id (str)

  • platform (str)

  • asyncio_task (Task | None)

  • redirect_channel_id (str)

  • redirect_platform (str)

  • redirect_adapter (Any)

  • redirect_max_chars (int)

task_id: str
tool_name: str
status: TaskStatus
created_at: float
result: str | None = None
error: str | None = None
user_id: str = ''
channel_id: str = ''
platform: str = ''
asyncio_task: Task | None = None
redirect_channel_id: str = ''

Channel to deliver the result to when the task finishes.

redirect_platform: str = ''

Platform name for the redirect target.

redirect_adapter: Any = None

Resolved PlatformAdapter for delivery.

redirect_max_chars: int = 0

Max characters of output body to deliver (0 = use default).

class task_manager.TaskManager(timeout=10.0, redis=None)[source]

Bases: object

Manage fire-and-forget tool execution with timeout.

Parameters:

  • timeout (float) – Seconds to wait for a tool to complete before backgrounding it. Defaults to 10.0.

  • redis (Any) – Optional async Redis client for persisting completed results.

__init__(timeout=10.0, redis=None)[source]

Configure the timeout budget and Redis handle for task tracking.

Stores the inline-vs-background timeout threshold and the optional async Redis client used to persist task state, and initializes the empty self._tasks registry that maps each task ID to its TaskRecord. Performs no I/O. Constructed once per worker by the inference and web services (inference_main and web_main, both with timeout=10.0 and the shared Redis client) and exposed to tools through the injected ToolContext.

Parameters:

  • timeout (float) – Seconds to wait for a tool coroutine to finish before backgrounding it; defaults to 10.0.

  • redis (Any) – Optional async Redis client for persisting running and completed results; when None, tasks are tracked in memory only.

Return type:

None

async execute(coro, tool_name='', user_id='', channel_id='', platform='')[source]

Run coro with a timeout; background it if it takes too long.

Returns the tool result string directly when the coroutine finishes within timeout, or a JSON envelope with a task_id when it does not.

Return type:

str

Parameters:
async get_result(task_id, user_id=None)[source]

Return the result for task_id, or a status update.

If user_id is set, only tasks owned by that user are returned.

Return type:

str

Parameters:

  • task_id (str)

  • user_id (str | None)

async await_result(task_id, timeout=300.0, user_id=None)[source]

Block until task_id completes and return its result.

Unlike get_result() which returns immediately with a status update, this method awaits the underlying asyncio.Task so the caller’s coroutine is suspended until the work finishes.

If user_id is set, only tasks owned by that user are awaited or returned (same semantics as get_result()).

Parameters:

  • timeout (float) – Maximum seconds to wait. Defaults to 300 (5 minutes). If exceeded, a timeout error JSON envelope is returned.

  • task_id (str)

  • user_id (str | None)

Return type:

str

async list_tasks(user_id=None)[source]

Return a JSON summary of tracked tasks.

If user_id is provided, only tasks belonging to that user are returned. Pass None to list all tasks.

Return type:

str

Parameters:

user_id (str | None)

set_output_redirect(task_id, channel_id, platform, adapter, max_chars=0)[source]

Configure a task to deliver its result to channel_id on finish.

Returns an error string if the task is not found or already finished, otherwise None.

Return type:

str | None

Parameters: