core.worker¶
Worker modules: conversational orchestrator, background task execution, web research protocol, and priority task queue.
Orchestrator¶
core.worker.orchestrator
¶
Primary conversation orchestrator with helper-worker delegation.
Collapses the default runtime from six visible agents speaking in turn into one primary conversation orchestrator that can delegate to helper workers for planning, retrieval, memory summarization, coding, or background tasks.
The orchestrator wraps a primary Agent instance and a WorkerPool.
On each conversation turn it:
- Detects user intent (memory lookup, planning, summarization).
- Spawns helper workers as needed (non-speaking internal workers).
- Calls the primary agent with enriched context.
- Returns an
OrchestratorResultwith response text and worker traces.
The six-agent multi-agent mode is preserved as a legacy_mode flag
in ModeRegistry. When legacy_mode is True, the old behaviour
(all active agents respond in random order) is restored.
Components¶
OrchestratorResult— frozen dataclass with response summaryOrchestrator— primary orchestrator class
Orchestrator
¶
Primary conversation orchestrator with helper-worker delegation.
Wraps a primary Agent and a WorkerPool. On each turn it
optionally spawns helper workers (planner, retriever, summarizer),
enriches the agent prompt with worker results, and streams the
primary agent's response.
Parameters¶
primary_agent :
The agent instance that produces user-facing responses.
Must have agent_name and generate_text_stream attributes.
worker_pool :
A WorkerPool (or compatible) for spawning helper workers.
event_bus :
Optional event bus for publishing orchestrator lifecycle events.
system_prompt :
Override system prompt for the orchestrator. If not provided
the primary agent's own system prompt is used.
primary_agent: Any
property
¶
Return the wrapped primary agent.
delegate(worker_type: WorkerType, inputs: dict[str, Any] | None = None) -> WorkerResult
async
¶
orchestrate(user_text: str, vector_text: str = '', *, correlation_id: str = '', memory_mode: bool = False, extra_context: str = '') -> OrchestratorResult
async
¶
Run one full orchestration turn.
Steps: 1. Generate a correlation ID if not provided. 2. Detect user intent and spawn helper workers. 3. Wait for worker results. 4. Call the primary agent with enriched context. 5. Collect response text, publish events, record telemetry.
Parameters¶
user_text :
The user's spoken or typed message.
vector_text :
Optional vector context (screenshot caption, audio transcript).
correlation_id :
Optional — links all events for this turn.
memory_mode :
If True, spawn a RETRIEVER worker for memory lookup.
extra_context :
Additional context to prepend to the agent prompt.
Returns¶
OrchestratorResult Response text, worker traces, timing info.
OrchestratorResult
dataclass
¶
Immutable result from a single orchestrator turn.
Parameters¶
response_text : The primary agent's response text. worker_results : Results from any helper workers that were spawned. correlation_id : Links all events for this turn. duration_ms : End-to-end orchestration time in milliseconds. primary_agent_name : Name of the primary agent that produced the response. worker_count : Number of helper workers spawned. first_token_ms : Time from orchestration start to first agent token (0 if unknown).
to_dict() -> dict[str, Any]
¶
Serialize to a plain dictionary (JSON-friendly).
Worker System¶
core.worker.worker_system
¶
Helper-worker architecture for non-speaking internal workers.
Workers communicate through typed internal events (not free-form chat loops). Each worker has budget limits, cancellation support, trace events, and explicit ownership of inputs/outputs.
Worker types: - planner: Pre-compute action plans before agent generation - retriever: Memory / vector store lookups - summarizer: Condense long text into summaries - indexer: Background content indexing - tool_preparer: Pre-load tool schemas / prepare tool results - reasoner: Optional specialist deep-reasoning worker
WorkerConfig
dataclass
¶
Configuration for a helper worker.
WorkerError
dataclass
¶
Bases: Exception
Error raised when a worker fails.
WorkerPool
¶
Bounded pool that manages concurrent helper workers.
Workers execute asynchronously with: - Concurrency limits (max concurrent workers, per-type limits) - Per-worker timeouts and budget limits - Cancellation support via asyncio.CancelledError - Trace events for observability - Optional EventBus integration for broadcast - Bounded result history
Security boundaries: - All exceptions caught and returned as WorkerResult(status=FAILED) - Token budget enforced per-worker - Timeout enforced per-worker via asyncio.wait_for()
history_count: int
property
¶
Number of results in history.
running_count: int
property
¶
Number of currently running workers.
__init__(*, max_concurrent: int = 8, max_per_type: int | None = None, max_history: int = 200, default_timeout: float | None = 30.0, event_bus: Any | None = None) -> None
¶
Initialize the worker pool.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
max_concurrent
|
int
|
Maximum concurrent workers across all types. |
8
|
max_per_type
|
int | None
|
Maximum concurrent workers per WorkerType (None = no limit). |
None
|
max_history
|
int
|
Maximum completed results to retain. |
200
|
default_timeout
|
float | None
|
Default timeout in seconds (overridden by WorkerConfig). |
30.0
|
event_bus
|
Any | None
|
Optional EventBus for broadcasting worker events. |
None
|
cancel(worker_id: str) -> bool
async
¶
Request cancellation of a running worker.
clear_history() -> int
async
¶
Clear completed worker history. Return count cleared.
get_history(*, worker_type: WorkerType | None = None, status: WorkerStatus | None = None, limit: int | None = None) -> list[WorkerResult]
async
¶
Query worker result history.
Returns results most recent first.
get_result(worker_id: str) -> WorkerResult | None
async
¶
Look up a completed worker result by ID.
get_running_by_type(worker_type: WorkerType) -> int
¶
Return the number of running workers of a given type.
get_running_count() -> int
¶
Return the number of currently running workers.
list_tasks() -> list[dict[str, Any]]
async
¶
Return a summary of all currently running workers.
register_executor(worker_type: WorkerType, executor: WorkerExecutor) -> None
async
¶
Register an async executor function for a WorkerType.
spawn(config: WorkerConfig, inputs: dict[str, Any], *, correlation_id: str = '', cancel_event: asyncio.Event | None = None) -> str
async
¶
Spawn a new helper worker.
Returns the worker_id. The worker runs asynchronously; use
get_result() or wait_for() to obtain the output.
Raises:
| Type | Description |
|---|---|
WorkerError
|
If no executor is registered for the worker type. |
wait_for(worker_id: str, *, timeout: float | None = None) -> WorkerResult | None
async
¶
Wait for a worker to complete and return the result.
WorkerRequest
dataclass
¶
A request to spawn a helper worker.
WorkerResult
dataclass
¶
The output of a worker execution.
WorkerStatus
¶
Bases: Enum
Worker execution status.
WorkerTraceEvent
dataclass
¶
A trace event emitted during worker execution.
WorkerType
¶
Bases: Enum
Types of helper workers.
Background Task Runner¶
core.worker.background_task_runner
¶
Background task execution subsystem.
Runs complex multi-step tasks against Ollama in a background asyncio task, independent of the main conversation flow. The task model uses the same tool capabilities as the main Agent, ensuring all operations pass through existing safety boundaries.
Designed so the Agent can delegate a task and immediately continue chatting with the user while work proceeds in parallel.
BackgroundTask
dataclass
¶
Represents a single delegated background task.
BackgroundTaskRunner
¶
Manages background task execution using the existing Ollama + tool infrastructure.
Each task runs in its own asyncio coroutine with an isolated message history.
Tool calls are routed through the parent Agent's toolCall() dispatch,
enforcing the same safety boundaries as synchronous tool use.
cancel_task(task_id: str) -> bool
async
¶
Cancel a running background task.
get_or_create(agent: Agent) -> BackgroundTaskRunner
classmethod
¶
Get or create the task runner attached to agent.
get_task_result(task_id: str) -> str
async
¶
Return the full result text, or an error message.
Checks local _active_tasks first, then class-level _all_tasks.
get_task_status(task_id: str) -> dict[str, Any]
async
¶
Return a summary dict for task_id.
Checks local _active_tasks first, then falls back to the class-level _all_tasks for cross-agent visibility.
list_tasks() -> list[dict[str, Any]]
¶
Return status summaries for all known tasks.
Merges local _active_tasks with class-level _all_tasks for cross-agent visibility, preferring local entries when both exist.
set_max_parallel(value: int) -> None
¶
Limit concurrent background tasks.
spawn_task(description: str, context: str = '', allowed_tools: list[str] | None = None, timeout_seconds: int = 300, max_tool_calls: int = 10, max_output_chars: int = 50000, model_override: str | None = None, on_complete: Any | None = None) -> str
async
¶
Spawn a background task. Returns immediately with task_id.
The task runs asynchronously in a separate asyncio coroutine.
wait_for_task(task_id: str, timeout: float = 120.0) -> str
async
¶
Await a task's completion with an optional timeout.
Checks local _active_tasks first, then class-level _all_tasks.
TaskConstraints
dataclass
¶
Safety and resource constraints for a background task.
Task Queue¶
core.worker.task_queue
¶
Background task queue with backpressure, status tracking, and cancellation.
Provides a bounded async task queue that prevents unbounded spawning of background coroutines. Each task gets a unique ID, lifecycle status, optional timeout, retry budget, priority ordering, and progress updates.
Key features: - Bounded concurrency — a semaphore limits parallel execution; excess tasks queue up (backpressure) until a worker slot frees. - Task IDs — caller-provided or auto-generated (12-char hex). - Status tracking — pending, running, completed, failed, cancelled. - Cancellation — co-operative via asyncio.CancelledError + notification. - Retries — configurable retry count with exponential back-off. - Progress — tasks can emit progress updates via their task handle. - Timeouts — per-task execution deadline via asyncio.wait_for(). - Priority — lower numbers run first (0 is default, negative = urgent). - Event publishing — integrates with EventBus for lifecycle events. - Persistent records — task history retained after completion (bounded).
BackgroundTaskQueue
¶
Bounded async task queue with backpressure.
Parameters¶
max_concurrency : int
Maximum number of tasks executing in parallel.
max_queue_size : int
Maximum pending tasks in the queue. enqueue() raises
QueueFullError when the limit is reached.
max_history : int
Maximum completed task records to retain (bounded retention).
event_bus : EventBus | None
Optional event bus for publishing lifecycle events.
default_timeout : float | None
Default per-task timeout (overridden per-task if specified).
policy : WorkerPolicy | None
WorkerPolicy enforcement for submitted tasks. None means
no enforcement (default None). When set, enqueue()
validates against denied_tools and allowed_tools.
active_count: int
property
¶
Number of tasks currently executing.
history_count: int
property
¶
Number of completed/failed/cancelled task records retained.
is_running: bool
property
¶
Return whether the queue is actively processing tasks.
pending_count: int
property
¶
Number of tasks waiting in the queue.
policy: WorkerPolicy | None
property
¶
Current worker policy (may be None).
cancel(task_id: str) -> bool
async
¶
Cancel a running or pending task.
Returns True if the task was found and cancelled, False
if the task was not found or already completed.
enqueue(*, task_id: str, description: str, coroutine: Any, priority: int = 0, timeout: float | None = None, max_retries: int = 0) -> None
async
¶
Add a coroutine to the bounded queue.
Parameters¶
task_id : str Unique task identifier. description : str Human-readable description. coroutine : coroutine or Callable The async callable to execute. priority : int Lower numbers run first (0 is default, negative = urgent). timeout : float | None Per-task execution timeout in seconds. max_retries : int Number of retry attempts on failure (0 = no retry).
Raises¶
QueueFullError
If the queue has reached max_queue_size pending tasks.
ValueError
If a task with the same task_id is already active.
WorkerPolicyValidationError
If the policy denies tools mentioned in the description.
get_history(*, limit: int = 50, status: TaskStatus | None = None) -> list[dict[str, Any]]
async
¶
Return completed task history, most recent first.
get_policy() -> dict[str, Any]
¶
Return the current policy as a dict.
list_tasks() -> list[dict[str, Any]]
async
¶
Return all active (pending + running) task status dicts.
result(task_id: str, *, timeout: float | None = None) -> Any
async
¶
Wait for a task to complete and return its result.
Parameters¶
task_id : str
The task to wait for.
timeout : float | None
How long to wait (seconds). None = wait forever.
Returns¶
The task's return value.
Raises¶
TaskNotFoundError If the task_id is unknown. asyncio.TimeoutError If the timeout expires before the task completes. TaskFailedError If the task completed with an error.
set_policy(updates: dict[str, Any]) -> WorkerPolicy
¶
Update the worker policy with provided field overrides.
Returns the new (frozen) policy instance.
Raises ValueError if an unknown field is provided.
start() -> None
async
¶
Start the queue worker loop.
status(task_id: str) -> dict[str, Any] | None
async
¶
Return task status dict, or None if unknown.
stop(*, cancel_running: bool = False) -> None
async
¶
Stop accepting new tasks and optionally cancel running ones.
QueueFullError
¶
Bases: Exception
Raised when the task queue has reached max_queue_size.
TaskFailedError
¶
Bases: Exception
Raised when result() is called on a failed task.
TaskHandle
dataclass
¶
Lightweight handle returned by enqueue() for progress updates.
update_progress(progress: float, message: str = '') -> None
¶
Emit a progress update (0.0 - 1.0).
TaskNotFoundError
¶
Bases: Exception
Raised when a task_id is not found in the queue.
TaskRecord
dataclass
¶
Immutable snapshot of a task's current state.
Updated in-place by the queue as the task progresses through its lifecycle. External callers should treat this as read-only.
to_dict() -> dict[str, Any]
¶
Serialize to a plain dictionary (JSON-friendly).
TaskStatus
¶
Bases: Enum
Lifecycle states for a background task.
CANCELLED = auto()
class-attribute
instance-attribute
¶
Cancelled by the caller or due to timeout.
COMPLETED = auto()
class-attribute
instance-attribute
¶
Finished successfully.
FAILED = auto()
class-attribute
instance-attribute
¶
Finished with an error.
PENDING = auto()
class-attribute
instance-attribute
¶
Queued but not yet executing.
RETRYING = auto()
class-attribute
instance-attribute
¶
Scheduling a retry after a failure.
RUNNING = auto()
class-attribute
instance-attribute
¶
Currently executing.
WorkerPolicyValidationError
¶
Bases: Exception
Raised when a task violates WorkerPolicy constraints.