keble-task 2.22.0

No project description provided

Keble Task

Keble Task is a package for managing asynchronous task execution with MongoDB storage and Redis locking. It provides a client for creating, tracking, and managing tasks with support for retries, timeouts, and token consumption.

Version 2.22.0 Update

• Added the cross-repo Keble pytest marker vocabulary and strict marker config.
• Default tests no longer fail at collection when tests/.env is absent: Mongo/Redis-backed tests skip at fixture setup when MONGO_DB_URI or REDIS_URI is not configured.
• The live LLM agent-tool test now requires both RUN_LLM_LIVE=1 and LIVE_LLM_* credentials, so full local pytest does not spend model tokens by accident.
• Added AGENTS.md, CLAUDE.md, and pyrightconfig.json for future test workers.
• Removed the legacy custom classifier rejected by package upload validation.

Testing

Default fast test command:

uv run pytest -m "not live and not slow and not eval and not local_stack"

Run all non-live tests with local Mongo/Redis configured:

uv run pytest -q

Run the live LLM canary:

RUN_LLM_LIVE=1 uv run pytest -m "live and llm"

Run static syntax/type checking:

npx --yes pyright .

Version 2.21.0 Update

Task agent tool metadata now names keble_helpers.AgentToolRegistrationConfig directly. Package code and tests use AgentToolApprovalMode for approval semantics; do not reintroduce bool approval aliases in task provider contracts.

Build packaging explicitly excludes local agent scratch directories (.claude/, .worktrees/, .wt-discovery/) so private symlinks never enter release artifacts.

Version 2.18.0 Update

Code-breaking agent-schema naming + placement refactor (no backward-compat aliases), mirroring the shipped keble-positioning convention. Re-rebased onto the settled origin/main 2.17.0 (the AgenticActionEventSource SCREAMING_SNAKE realignment + the no-emit-without-drain static guard) and bumped above main to 2.18.0:

• schemas.py is now the keble_task.schemas PACKAGE (schemas/__init__.py keeps all prior exports identical) so agent tool I/O schemas can live in the new schemas/for_agent.py.
• Renames: TaskSummaryView -> TaskSummaryForAgent, TaskQueryToolsConfig -> TaskAgentQueryToolsConfig, TaskMutationToolsConfig -> TaskAgentMutationToolsConfig (all three now DEFINED in schemas/for_agent.py).
• Registrars moved to a new keble_task.agent.tools subpackage (behavior only, zero BaseModel): tools/query.py::register_query_tools and tools/mutation.py::register_mutation_tools. The old bare register / register_action_tools names and the agent/registry.py / agent/query_registry.py modules are REMOVED.
• New convention guard tests/schemas/test_for_agent.py.

Agentic schema convention (ForAgent + agent/tools/)

Every schema lands in exactly one bucket; the bucket dictates its NAME and its LOCATION. This is enforced by tests/schemas/test_for_agent.py.

Bucket	What it is	Name	Lives in
Agent tool I/O	typed input to / return projection of a pydantic-ai @agent.tool	*ForAgent suffix	schemas/for_agent.py only — never in agent/
Agent tool config/enum	tool-registration config or approval/mode enum used only by agent tools	*Agent…Config / *Agent…Policy infix	schemas/for_agent.py
Mutation action payload	the typed @agent.tool mutation input	unchanged	keble_task/actions.py (external action module)
Persisted / CRUD / event	*Base/*Update/*MongoObject/*Event	unchanged	schemas/

The registrars under agent/tools/ (mutation.py, query.py) are BEHAVIOR ONLY — they import contracts from schemas/ (and actions.py) and define zero BaseModel classes. Do NOT reintroduce *View/*Display names for agent tool payloads.

Version 2.12.0 Update

Added TaskClient.apublic_list_indexable(task_types, stages, limit) -> list[TaskPublicRef]: a lean, projected ({_id, updated}) public read backing consumer-side SEO sitemaps — filters sharing_scope=PUBLIC + task_type $in + stage $in, newest-updated first, capped by limit, no childs/redis. Backed by the new sharing_scope_task_type_stage_updated_desc_idx (the package owns its indexes via aensure_task_indexes). Also completed the latent apublic_get_multi overload contract (stages/title_contains are now honored, mirroring aowner_get_multi).

Package Line

1. Package version: 2.22.0
2. Python baseline: >=3.13,<3.14
3. Runtime ownership:
   • TaskBase.language is the canonical, task-owned output/report language; report-generating handlers and child-task creation read task.language instead of parsing domain-specific metadata blobs (optional for back-compat)
   • flat task lists return root tasks unless parent_task is explicitly filtered
   • TaskRelationType.MAPPED records viewport-map edges without changing parentage
   • TaskClient.aapply_actions(...) creates related tasks through a pure generic action layer; feature packages own their own typed tools
   • TaskClient.astart(...) is the single task-start entrypoint
   • uncaught handler exceptions are finalized into terminal FAILURE
   • start locks are always cleared before the run returns
   • package events use direct keble_helpers.AgenticActionEvent JSON; backend transports them without room-specific wrapper schemas
   • TaskClient can emit canonical task lifecycle events after persisted PROCESSING, SUCCESS, FAILURE, and timeout stage transitions
   • downstream TypeScript room consumers should use keble-core 0.1.32+ direct-event builders rather than task-owned workspace snapshots

Version 2.5.1 Update

1. Hardened the agentic mutate_task_workspace tool against ID hallucination: CreateRelatedTaskAction.parent_task_id and from_task_ids now carry model-facing Field(description=...) guidance to leave them unset (attach under the current task) unless real ids are provided — never invent ids.
2. The tool boundary (agent/registry.py) now maps model-fixable action-validation errors (ClientSideInvalidParams / ClientSideMissingParams / ServerSideInvalidParams from aapply_actions, e.g. a hallucinated parent_task_id) to pydantic_ai.ModelRetry, so the model self-corrects instead of aborting the run. Wiring/infra faults still propagate.
3. Added a deterministic FunctionModel self-correction test (bad parentTaskId -> ModelRetry -> recover) and made the live LLM prompt explicit about omitting the ids.
4. Package metadata synced to 2.5.1 across pyproject.toml, pyproject.poetry.toml, uv.lock.

Version 2.5.0 Update

1. Added TaskBase.language: Optional[Language] — a typed, task-owned output/report language so the generic task framework is the single source of truth for the language a handler should produce output in, instead of consumers digging it out of domain-specific metadata dicts.
2. TaskClient.acreate(..., language=...) threads the value onto the created task.
3. Optional with a None default, so legacy callers and legacy persisted documents still validate and load.
4. Package metadata synced to 2.5.0 across pyproject.toml, pyproject.poetry.toml, uv.lock.

Version 2.4.23 Update

1. Added owner/public root-list indexes — CRUDTask.aensure_task_indexes now also creates owner_parent_task_created_desc_idx and sharing_scope_parent_task_created_desc_idx so aowner_get_multi / apublic_get_multi root lists filter owner / sharing_scope on an index instead of residually after a parent_task scan (created at ainit/startup).
2. CREATE_RELATED_TASK is now self-healing: standalone Mongo has no multi-document transactions, so if a relation write fails after the child task is created, a saga compensation (_arollback_created_related_task) deletes the orphaned child and any partial relation rows, then re-raises — the workspace never keeps a child without its relations.
3. All package tests are pyright-clean (pyright keble_task and pyright tests both 0 errors); removed stale tracked dist/keble_task-0.0.0-* build artifacts (dist/ stays gitignored).
4. Package metadata synced to 2.4.23 across pyproject.toml, pyproject.poetry.toml, uv.lock.

Version 2.4.22 Update

1. build_task_tree(...) now fails hard on corrupted persisted references instead of silently shrinking trees: a child whose parent_task is absent from the loaded root tree or a missing requested root raises keble_exceptions.ObjectNotFound, and a parent cycle raises keble_exceptions.DataIntegrityCompromised. The sibling sort was simplified to use the non-Optional persisted created datetime (no try/except). Behavior change: list/tree endpoints surface corruption as typed errors rather than partial results.
2. The gated live agent-tool test guards its OpenAI-model imports with pytest.importorskip so collection never fails where the openai extra is absent, and the test dependency group now includes pydantic-ai-slim[openai] so uv run pytest can run it when LIVE_LLM_* creds are present.
3. Package metadata synced to 2.4.22 across pyproject.toml, pyproject.poetry.toml, and uv.lock.

Version 2.4.18 Update

1. TaskCostAggregateResponse.build(...) now takes costs: Sequence[TaskCostBase] (read-only/covariant) so CRUD's list[TaskCostMongoObject] is accepted without a list-invariance type error, and folds rows through new typed TaskCostAggregateBucket.empty() / accumulate() methods instead of an untyped dict[str, Any] accumulator — bucket math lives on the type that owns it.
2. The task collection now has tree/stage read indexes via CRUDTask.aensure_task_indexes(...) (root_task+created, parent_task+created, stage+created), wired into TaskClient.aensure_indexes(...)/ainit(...) so root-tree, root-list, and retry/timeout sweeps are index-backed (idempotent).
3. TaskMetadata = dict[str, JsonValue] is the single strongly-typed contract for free-form task metadata; TaskBase.metadata and the TaskClient metadata params use it instead of a bare dict.
4. The agent registrar uses AgentToolRegistrationConfig directly, removing the old local config shadowing that caused ToolFuncContext assignment errors.
5. Added tests for timeout lifecycle, the processing guard, zero-usage cost rollup, parent+child cost rollup by root, and a gated live-LLM mutate_task_workspace agent-tool run (LIVE_LLM_* creds in tests/.env).

Version 2.4.16 Update

1. TaskCostCreate.from_task_usage(...) is the canonical cost-row constructor; it denormalizes owner and task type from the persisted task row.
2. TaskCostAggregateResponse.build(...) owns in-memory response aggregation from already-filtered rows.
3. Runtime task-cost create/list/aggregate paths no longer create Mongo indexes; callers must use TaskClient.ainit(...) or startup wiring.
4. TaskHandlerRequest inherits AgentDbDeps, so usage_recorder travels with the same DB-rooted request object without making the task package own pricing or persistence policy outside cost rows.

Version 2.4.15 Update

1. TaskCostFilterBase is now the single source of truth for task-cost read filters shared by list and aggregate requests.
2. TaskCostListRequest owns only pagination fields on top of the shared filter contract.
3. TaskCostAggregateRequest owns only aggregation controls on top of the shared filter contract.
4. Task-cost tag filters remain all-tags filters: every requested tag must be present on the cost row.
5. Package metadata is synchronized across pyproject.toml, pyproject.poetry.toml, and uv.lock.

Version 2.4.14 Update

1. TaskClient.ainit(amongo=...) exposes the package-owned Mongo index setup as a backend startup hook over existing aensure_indexes(...).

Version 2.4.13 Update

1. Added separate task-cost storage through TaskCostBase, TaskCostCreate, TaskCostMongoObject, TaskCostListRequest, and TaskCostAggregateRequest.
2. TaskClient.acreate_task_cost(...) now denormalizes root task, owner, and task type from the persisted task row before writing the cost row.
3. TaskClient.alist_task_costs(...) and TaskClient.aaggregate_task_costs(...) provide indexed admin reporting with total/hour/day/week/month buckets and optional tag fan-out.
4. Aggregation uses RunUsage, per-million Money rates, and caller-provided ExchangeRateInUsd values while preserving sub-cent Decimal token costs.
5. TaskClient.aensure_indexes(...) creates public-id, relation, and task-cost indexes from one startup-friendly entrypoint.

Version 2.4.12 Update

1. keble-task now declares tenacity>=9,<10.0.0 directly because the task runtime imports it for handler retry.
2. TaskClient.acreate_task_relations(...) rejects duplicate relation edge payloads before inserting any row.
3. TaskClient.aapply_actions(...) rejects duplicate from_task_ids before creating the child task, preventing partial child-task persistence when an action would generate duplicate relation edges.
4. Previously skipped start/retry/timeout tests now exercise the real async handler path with ExtendedAsyncRedis.

Version 2.4.11 Update

1. Added TaskEventType.TASK_STAGE_CHANGED as the package-owned lifecycle event for task stage transitions.
2. Added TaskLifecycleEventPayload(task=TaskMongoObject) and TaskLifecycleEvent, reusing the persisted task schema instead of adding UI-specific state payloads.
3. TaskClient.astart(...), aon_task_processing(...), aon_task_success(...), aon_task_failure(...), and aon_task_timeout(...) can emit lifecycle events after DB updates reload the final task row.
4. Consumers should use lifecycle events for room task state, while CREATE_RELATED_TASK events remain the canonical child/relation creation events.

Version 2.4.8 Update

1. CreateRelatedTaskAction relation rows now persist action.metadata on TaskRelationCreate.metadata instead of writing {}.
2. TaskActionCreatedRelation remains a slim public DTO but now includes relation metadata so downstream event consumers can render edge context.
3. No new relation schema was added; TaskRelationBase.metadata is the canonical extension point for task-action relation annotations.

Version 2.4.9 Update

1. Relation metadata now uses the package-owned TaskRelationMetadata JSON-safe contract before persistence and event emission.
2. CreateRelatedTaskAction.metadata, TaskRelationBase.metadata, and TaskActionCreatedRelation.metadata normalize BSON ObjectIds, datetimes, and Pydantic models into JSON-compatible values.
3. Unsupported runtime objects are rejected at the metadata boundary instead of leaking into SSE or TypeScript JsonObject consumers.

Version 2.4.10 Update

1. Unsupported relation metadata now raises ValueError inside the JSON fallback so Pydantic wraps model construction failures as ValidationError.
2. Regression coverage protects both CreateRelatedTaskAction.metadata and TaskRelationCreate.metadata from surfacing raw TypeError.

Installation

pip install keble-task

Core Concepts

TaskClient

The TaskClient is the main entry point for backend APIs that need to create and manage tasks. All operations are asynchronous with 'a' prefix (e.g., acreate, astart, aget).

astart(...) now owns the terminal failure boundary for task execution:

1. it loads the task and checks the Redis start lock
2. it moves the task to PROCESSING
3. it runs the configured handler with retry support
4. if the handler still raises, it writes a terminal FAILURE row unless the handler already finalized the task
5. it always clears the Redis start lock before returning

Task Tree And Relations

Task trees and task relations are intentionally separate:

1. root_task scopes the whole workspace.
2. parent_task stores the one canonical creator/trigger edge.
3. TaskRelationMongoObject stores extra lineage edges such as multi-source REDUCED children and viewport-map MAPPED children.

Flat task list APIs now return only root tasks by default. Use include_childs=True for task trees, parent_task=<id> for direct child listing, and relation APIs for non-tree edges.

The package does not expose a workspace snapshot schema. Backend routes should fetch the existing task tree and relation rows separately, then decide how to compose them for frontend views.

Task Room Resolution

Task-room identity is intentionally generic and task-tree-only:

1. TaskClient.aresolve_task_room(...) accepts any owner-visible task id.
2. It returns:
   • root_task_id
   • requested_task_id
3. The package does not know feature focus shape, selection metadata, or chat state.
4. Backend/frontend layers should use the returned root task id as room identity and keep feature-specific metadata outside this package.

Task Room Graph Context

TaskRoomGraphContext is the compact agent-facing room graph. It is generic: task ids, parent edges, task type/title/subtitle, and sidecar MAPPED / REDUCED relation edges. It does not include positioning ids, grid ids, UX selection, or frontend state.

context = await task_client.aget_task_room_graph_context(
    amongo=amongo,
    owner=owner,
    root_task_id=root_task_id,
    focused_task_id=focused_task_id,
)
prompt_text = context.to_prompt_text()

relations = await task_client.acreate_task_relations(
    amongo=amongo,
    objs_in=[
        TaskRelationCreate(
            root_task=root.id,
            from_task_id=source_a.id,
            to_task_id=child.id,
            relation_type=TaskRelationType.REDUCED,
        ),
        TaskRelationCreate(
            root_task=root.id,
            from_task_id=source_b.id,
            to_task_id=child.id,
            relation_type=TaskRelationType.REDUCED,
        ),
        TaskRelationCreate(
            root_task=root.id,
            from_task_id=source_a.id,
            to_task_id=mapped_view_task.id,
            relation_type=TaskRelationType.MAPPED,
        ),
    ],
)

root_relations = await task_client.alist_task_relations_by_root(
    amongo=amongo,
    root_task=root.id,
)

Task Cost Tracking

Task costs are stored in a dedicated collection and are denormalized from the task row at creation time. The task collection stays focused on execution state, while cost reports can filter by task, root task, owner, task type, tags, and occurred_at windows.

result = await agent.run(prompt, deps=deps)

await task_client.acreate_task_cost(
    amongo=amongo,
    task_id=task.id,
    tags=["positioning", "classify_cells"],
    run_usage=result.usage,
    token_rates_per_million=TaskCostTokenRates(
        input_tokens=Money(float_money=1.25, currency=Currency.USD),
        output_tokens=Money(float_money=10.00, currency=Currency.USD),
    ),
    additional_cost=Money(float_money=0.01, currency=Currency.USD),
    seconds=12,
    retry=max(task.attempts - 1, 0),
    metadata={"model": "gateway/openai:gpt-5.2"},
)

Task Handler

For packages that want to support keble-task, implementing a task handler function is required rather than creating a TaskClient instance.

Task Lifecycle Events

ProgressTask is not the current task-handler transport. Handlers receive a TaskHandlerRequest, read DB clients and cross-cutting context directly from that request, and return TaskHandlerResponse | None.

Task lifecycle and domain progress are emitted through the shared AgenticEventEmitter on request.event_emitter. The emitter is intentionally detached: aemit() schedules callbacks and the task runtime drains before the execution boundary returns.

Schemas

The package provides several schemas for working with tasks:

TaskStage

An enum representing the different stages of a task:

class TaskStage(str, Enum):
    PENDING = "PENDING"     # Task is created but not yet started
    PROCESSING = "PROCESSING"  # Task is currently being processed
    SUCCESS = "SUCCESS"     # Task completed successfully
    FAILURE = "FAILURE"     # Task failed to complete

Tasks in PENDING or PROCESSING stages can be started or restarted.

SharingScope

An enum defining the visibility scope of a task:

class SharingScope(str, Enum):
    PRIVATE = "PRIVATE"  # Only accessible by the owner
    PUBLIC = "PUBLIC"    # Accessible by anyone

TaskBase

The base schema for task data:

class TaskBase(SchemaBase):
    # Error information
    error: Optional[str] = None
    exception_type: Optional[TaskExceptionType] = None

    # Current task stage
    stage: TaskStage

    # Task type identifier
    task_type: str  

    # Optional progress tracking key
    progress_key: Optional[str]

    # Display information
    title: Optional[str]
    subtitle: Optional[str]
    image: Optional[str]
    
    # Custom metadata for the task
    metadata: Optional[dict]

    # Access control
    sharing_scope: SharingScope = SharingScope.PRIVATE
    status: Status = Status.ACTIVE
    owner: str

    # Timestamp tracking
    started_ts: Optional[Timestamp] = None
    success_ts: Optional[Timestamp] = None
    failure_ts: Optional[Timestamp] = None

    # Token management
    expected_token: int  # Expected tokens to consume
    consumed_token: int = 0  # Actual tokens consumed

    # Retry and timeout handling
    attempts: int = 0  # Number of attempts so far
    timeout_mins: int = 120  # Timeout in minutes

The TaskBase provides two helpful properties:

• unfinshed_timeout: Returns True if the task has timed out but is still in PENDING or PROCESSING stage
• allow_to_retry: Returns True if the task has fewer than 3 attempts and is in a stage that allows starting

TaskUpdate

A schema for the fields an owner may patch post-creation. Unset fields are not written (aowner_update dumps with exclude_unset), so a partial patch never clobbers other columns. image reuses the exact same URL validation as creation (TaskBase.validate_image_url), so an edit can never store an image creation would have rejected.

class TaskUpdate(BaseModel):
    sharing_scope: Optional[SharingScope] = None
    title: Optional[str] = None
    image: Optional[str] = None  # owner thumbnail; full http(s) URL only

TaskMongoObject

The MongoDB document representation of a task:

class TaskMongoObject(MongoObjectBase, TaskBase):
    pass

This combines TaskBase with MongoDB-specific functionality from MongoObjectBase.

TaskRelationMongoObject

The MongoDB document representation of extra lineage between tasks in one workspace:

class TaskRelationType(str, Enum):
    MAPPED = "MAPPED"
    REDUCED = "REDUCED"


class TaskRelationBase(SchemaBase):
    root_task: ObjectId
    from_task_id: ObjectId
    to_task_id: ObjectId
    relation_type: TaskRelationType
    metadata: dict[str, Any] = Field(default_factory=dict)


class TaskRelationMongoObject(MongoObjectBase, TaskRelationBase):
    pass

Relation rows do not modify root_task, parent_task, or task status. They are sidecar lineage edges for backend/frontend composition.

Generic Task Actions

Use TaskActions when an agent or backend API needs task-owned mutations without feature-specific side effects.

out = await task_client.aapply_actions(
    amongo=amongo,
    current_task=source_task,
    payload=TaskActions(
        message="Create a related child task.",
        actions=[
            CreateRelatedTaskAction(
                relation_type=TaskRelationType.MAPPED,
                task_type="POSITIONING",
                title="Related workspace",
                metadata={"source": "agent"},
            )
        ],
    ),
    extended_aredis=extended_aredis,
)

The task package creates the child task, validates root ownership, writes relation rows, and emits TaskActionEvent using keble_helpers.AgenticActionEvent. Positioning-specific mapped/reduced-child behavior belongs in backend-owned typed tools, not in keble-task hook registries.

Action results now expose slim DTOs instead of full Mongo rows:

1. TaskActionCreatedTask carries task_id, root/parent ids, task type, stage, title, and progress key.
2. TaskActionCreatedRelation carries relation id, root/source/target ids, and relation type.
3. Internal TaskMongoObject and TaskRelationMongoObject remain the persistence models, but tool/action payloads should not return them directly.

Agent Tool Registration

Task owns the pydantic-ai registrar for generic task workspace mutations. Backend should register this package tool instead of defining a manual duplicate in backend chat code.

from keble_task.agent import TaskAgentDeps, register_mutation_tools

agent = Agent[TaskAgentDeps, Any](...)

register_mutation_tools(
    agent,
    task_client=task_client,
)

Deps shape:

• TaskAgentDeps inherits keble_db.AgentDbDeps; Mongo/Redis are not separate tool args.
• Task runtime state is under ctx.deps.task.
• The tool delegates to TaskClient.aapply_actions(...), so task parentage, generic relations, and task action events stay package-owned.

Native Chat Tool Provider (query tools)

The package also owns the READ/QUERY registrar (register_query_tools: owner-scoped list_tasks / get_task) and, since 2.7.0, ships it as a NATIVE keble_helpers.ChatToolProviderProtocol provider so hosts no longer need a generic adapter around the registrar:

from keble_task.agent import TaskQueryChatToolProvider

provider = TaskQueryChatToolProvider(
    task_client=task_client,
    tools_config=None,  # optional TaskAgentQueryToolsConfig | dict overrides
)

# A chat host composes providers declaratively; register attaches the tools.
provider.register(agent=agent, context=None)

Contract notes:

• provider.provider_id is EXACTLY "task_query" — it is recorded in room diagnostics and mapped to a user-readable label by the frontend; do not change it.
• All deps are captured at construction; register(*, agent, context) ignores context, which the cross-repo protocol explicitly permits.

Task Handler Dependencies

The retired TaskResources bag is no longer part of the active handler API. TaskHandlerRequest inherits keble_db.AgentDbDeps, so handlers read clients and cross-cutting context directly from the request:

request.amongo
request.extended_aredis
request.aneo4j
request.qdrant_client
request.event_emitter
request.usage_recorder

The event_emitter field is the canonical, single-source channel for task lifecycle and package-owned domain events. Package handlers emit typed AgenticActionEvent values through this emitter; the task runtime drains it before returning from lifecycle boundaries so owner-list and room listeners see deterministic terminal state.

TaskHandlerRequest

A schema for task handling requests passed to task handlers:

class TaskHandlerRequest(AgentDbDeps):
    model_config = PydanticModelConfig.default(arbitrary_types_allowed=True)
    task: TaskMongoObject  # The task to be processed
    metadata: TaskMetadata | None = None

Handlers return TaskHandlerResponse when the current process completed the task. Handlers return None only when completion is intentionally delegated to a separate process; in that case the runtime leaves the task in PROCESSING until the delegate finalizes it.

TaskHandlerResponse

A schema for responses returned from task handlers:

class TaskHandlerResponse(BaseModel):
    model_config = PydanticModelConfig.default()
    task: TaskMongoObject  # The processed task
    success: bool  # Whether the task was successful
    consuming_token: int  # Actual tokens consumed
    exception_type: Optional[TaskExceptionType] = None  # Type of exception if failed
    error: Optional[str] = None  # Error message if failed

TokenConsumptionType

An enum representing token consumption actions:

class TokenConsumptionType(str, Enum):
    CONSUME = "CONSUME"  # Consume tokens
    RECOVER = "RECOVER"  # Recover (return) tokens

TokenConsumptionPayload

A schema for token consumption operations:

class TokenConsumptionPayload(BaseModel):
    model_config = PydanticModelConfig.default(arbitrary_types_allowed=True)
    consumption_type: TokenConsumptionType  # Whether to consume or recover tokens
    owner: str  # The owner of the tokens
    token: int  # The amount of tokens to consume or recover
    task_id: ObjectId | None = None
    amongo: AsyncIOMotorClient
    extended_aredis: ExtendedAsyncRedis
    aneo4j: Neo4jAsyncDriver | None = None
    metadata: TaskMetadata | None = None

TaskExceptionType

An enum representing different types of task exceptions:

class TaskExceptionType(str, Enum):
    UNKNOWN = "UNKNOWN"  # Unknown error
    FAILED_TO_START = "FAILED_TO_START"  # Task failed to start
    NO_SUFFICIENT_DATA = "NO_SUFFICIENT_DATA"  # Insufficient data to process task
    TIMEOUT = "TIMEOUT"  # Task timed out

TaskException

Base exception class for task-related errors:

class TaskException(KebleException):
    def __init__(
        self,
        *,
        exception_type: TaskExceptionType = TaskExceptionType.UNKNOWN,
        error: Optional[str] = None,
    ):
        self.exception_type = exception_type
        self.error = error
        # Inherits from KebleException

Subclasses of TaskException:

• TaskNoSufficientDataException: Raised when there is insufficient data to process a task
• TaskFailedToStartException: Raised when a task fails to start
• TaskTimeoutException: Raised when a task times out
• TaskUnknownException: Raised for unknown errors

Difficulty

A utility enum for representing task difficulty levels:

class Difficulty(str, Enum):
    EASY = "EASY"
    MEDIUM = "MEDIUM"
    HARD = "HARD"

Examples

Backend API: Creating a TaskClient

import asyncio
from keble_task import (
    TaskClient,
    TaskHandlerRequest,
    TaskHandlerResponse,
    TokenConsumptionPayload,
    TokenConsumptionType,
    TaskExceptionType,
    TaskStage,
)
from keble_helpers import AgenticEventEmitter, SharingScope
from motor.motor_asyncio import AsyncIOMotorClient
from keble_db import ExtendedAsyncRedis

# Define a token consumption handler (synchronous or asynchronous)
def token_consumption_handler(payload: TokenConsumptionPayload) -> bool:
    # Access DB clients directly from the payload; there is no resources bag.
    amongo = payload.amongo
    extended_aredis = payload.extended_aredis

    # Implement logic to handle token consumption or recovery
    if payload.consumption_type == TokenConsumptionType.CONSUME:
        print(f"Consuming {payload.token} tokens for {payload.owner}")
    else:  # TokenConsumptionType.RECOVER
        print(f"Recovering {payload.token} tokens for {payload.owner}")

    return True

# Alternatively, you can define an asynchronous token consumption handler
async def async_token_consumption_handler(payload: TokenConsumptionPayload) -> bool:
    amongo = payload.amongo
    extended_aredis = payload.extended_aredis

    if payload.consumption_type == TokenConsumptionType.CONSUME:
        print(f"Async consuming {payload.token} tokens for {payload.owner}")
    else:  # TokenConsumptionType.RECOVER
        print(f"Async recovering {payload.token} tokens for {payload.owner}")

    return True

# Define an async task handler
async def task_handler(request: TaskHandlerRequest) -> TaskHandlerResponse:
    task = request.task
    amongo = request.amongo
    extended_aredis = request.extended_aredis

    # Process the task with direct request clients and return the final response.
    result = await process_task(task, amongo=amongo, extended_aredis=extended_aredis)

    return TaskHandlerResponse(
        task=task,
        success=result.success,
        consuming_token=result.tokens_used,
        exception_type=None if result.success else TaskExceptionType.UNKNOWN,
        error=None if result.success else result.error,
    )

# Initialize MongoDB and Redis connections
async def setup():
    # Initialize MongoDB and Redis connections
    amongo_client = AsyncIOMotorClient("mongodb://localhost:27017")
    extended_aredis = ExtendedAsyncRedis("redis://localhost:6379")

    # Create TaskClient instance
    task_client = TaskClient(
        token_consumption_handler=token_consumption_handler,  # Or async_token_consumption_handler
        task_handler=task_handler,
        mongo_database="my_database",  # Optional, defaults to "__keble_task__"
        task_collection="tasks"        # Optional, defaults to "__keble_task__task__"
    )
    
    return task_client, amongo_client, extended_aredis

# Create a new task
async def create_task(task_client, amongo):
    task = await task_client.acreate(
        amongo=amongo,
        expected_token=5,               # Expected tokens to consume
        owner="user123",                # Owner ID
        task_type="image_processing",   # Type of task
        title="Process Image",          # Optional title
        metadata={"image_url": "https://example.com/image.jpg"},  # Optional metadata
        progress_key="process_image_123",  # Optional progress key
        image=None,                     # Optional image URL
        subtitle=None,                  # Optional subtitle
        timeout_mins=120,               # Optional timeout in minutes (default 120)
        sharing_scope=SharingScope.PRIVATE,  # Optional sharing scope
        stage=TaskStage.PENDING,        # Optional initial stage
        attempts=0,                     # Optional initial attempts
        consumed_token=0                # Optional initial consumed tokens
    )
    return task

# Start a task and route lifecycle events through the shared emitter
async def start_task(task_client, amongo, extended_aredis, task_id):
    event_emitter = AgenticEventEmitter()

    await task_client.astart(
        amongo=amongo,
        extended_aredis=extended_aredis,
        task_id=task_id,
        task_lifecycle_event_emitter=event_emitter,
    )

    return event_emitter

# Get a task by ID
async def get_task(task_client, amongo, task_id):
    task = await task_client.aget(
        amongo=amongo,
        task_id=task_id,
        task_type=None,  # Optional filter by task type
        include_childs=True,  # Optional: return TaskMongoObjectExtended with childs
    )
    return task

# Get a task by owner and ID
async def get_owner_task(task_client, amongo, owner, task_id):
    task = await task_client.aowner_get(
        amongo=amongo,
        owner=owner,
        task_id=task_id,
        task_type=None,  # Optional filter by task type
        sharing_scope=None,  # Optional filter by sharing scope
        include_childs=True,  # Optional: return TaskMongoObjectExtended with childs
    )
    return task

# Get multiple tasks
async def get_multiple_tasks(task_client, amongo, extended_aredis):
    tasks = await task_client.aget_multi(
        amongo=amongo,
        extended_aredis=extended_aredis,
        skip=0,
        limit=10,
        task_types=["image_processing", "text_processing"],  # Optional filter by task types
        include_childs=True,  # Optional: return a tree (roots + childs)
        root_task=None,  # Optional: filter to a specific root tree
        parent_task=None,  # Optional: list direct childs of a parent task
    )
    return tasks

# Get multiple tasks for an owner
async def get_owner_multiple_tasks(task_client, amongo, extended_aredis, owner):
    tasks = await task_client.aowner_get_multi(
        amongo=amongo,
        extended_aredis=extended_aredis,
        owner=owner,
        skip=0,
        limit=10,
        task_types=["image_processing"],  # Optional filter by task types
        sharing_scopes=[SharingScope.PRIVATE],  # Optional filter by sharing scopes
        include_childs=True,  # Optional: return a tree (roots + childs)
        root_task=None,  # Optional: filter to a specific root tree
        parent_task=None,  # Optional: list direct childs of a parent task
    )
    return tasks

# Run example
async def main():
    task_client, amongo, extended_aredis = await setup()

    # Create a task
    task = await create_task(task_client, amongo)
    print(f"Created task with ID: {task.id}")

    # Start task with lifecycle events
    await start_task(task_client, amongo, extended_aredis, task.id)
    print("Started task with lifecycle events")

    # Get task by ID
    retrieved_task = await get_task(task_client, amongo, task.id)
    print(f"Retrieved task: {retrieved_task.stage}")
    
    # Get task by owner
    owner_task = await get_owner_task(task_client, amongo, "user123", task.id)
    print(f"Retrieved owner task: {owner_task.stage}")
    
    # Get multiple tasks
    tasks = await get_multiple_tasks(task_client, amongo, extended_aredis)
    print(f"Retrieved {len(tasks)} tasks")
    
    # Get multiple owner tasks
    owner_tasks = await get_owner_multiple_tasks(task_client, amongo, extended_aredis, "user123")
    print(f"Retrieved {len(owner_tasks)} owner tasks")

if __name__ == "__main__":
    asyncio.run(main())

Package Supporting Keble Task: Implementing a Task Handler

If you're developing a package that wants to support keble-task, you only need to provide a task handler function:

from keble_task import TaskHandlerRequest, TaskHandlerResponse, TaskExceptionType

async def my_package_task_handler(
    request: TaskHandlerRequest,
) -> TaskHandlerResponse | None:
    task = request.task

    # Return None only when this handler intentionally delegates completion to
    # another process. The task runtime leaves the task PROCESSING in that case.
    if task.task_type != "my_package_task_type":
        return None

    result = await process_task(
        task,
        amongo=request.amongo,
        extended_aredis=request.extended_aredis,
        event_emitter=request.event_emitter,
    )

    return TaskHandlerResponse(
        task=task,
        success=result.success,
        consuming_token=result.tokens_used,
        exception_type=None if result.success else TaskExceptionType.UNKNOWN,
        error=None if result.success else result.error,
    )

async def process_task(task, *, amongo, extended_aredis, event_emitter):
    # Implement your task processing logic here
    # ...
    return result

Error Handling

The package uses keble_exceptions.KebleException which will be thrown in various error scenarios. Specific task exceptions include:

• TaskFailedToStartException: When a task fails to start
• TaskNoSufficientDataException: When there is insufficient data to process a task
• TaskTimeoutException: When a task exceeds its timeout duration
• TaskUnknownException: For general unexpected errors

Always handle these exceptions appropriately in your implementation.

Task Lifecycle

1. Creation: Tasks are created with acreate() method
2. Starting: Tasks are started with astart() method
3. Processing: Tasks are processed by the task handler
4. Completion: The handler returns TaskHandlerResponse and the runtime finalizes success or failure.

The package automatically handles retries, timeouts, and token consumption based on the configuration provided.

Lifecycle And Progress Events

The active progress path is event-based:

1. Lifecycle: TaskClient emits KEBLE_TASK / TASK_STAGE_CHANGED after persisted stage transitions.
2. Domain progress: Package handlers emit their own typed AgenticActionEvent payloads through request.event_emitter.
3. Drain boundary: aemit() schedules callbacks; adrain() is the deterministic side-effect boundary.
4. Frontend delivery: Backend bridges the same event envelope to owner-list and room websocket listeners.

ProgressTask may still appear in historical compatibility notes, but it is not the current task-handler progress transport.

Breaking API Changes in Version 0.0.5

This version introduces significant breaking changes from the previous version:

1. All Functions are Now Async: All functions have been converted to asynchronous with an 'a' prefix (e.g., acreate, astart, aget)

2. Direct Resource Parameters: Functions no longer accept a TaskResources object. Handler requests and token payloads carry amongo, extended_aredis, and optional graph/vector clients directly.

3. Async Handlers: Task handlers should now be implemented as async functions. Token consumption handlers can be either synchronous or asynchronous.

4. MongoDB and Redis Requirements: The package now requires the async versions of MongoDB (AsyncIOMotorClient) and Redis clients.

5. Progress Tracking: Current progress uses AgenticEventEmitter and typed package events; older ProgressTask examples are historical only.

6. Token Consumption Handler: The parameter token_consumption_handler now supports both synchronous and asynchronous implementations.

If you're upgrading from a previous version, you'll need to update all your code to follow these new conventions.

Dependency Documentation

This repo no longer stores copied dependency README.md snapshots under a local readme/ directory. Read dependency documentation from the maintained source repos so local docs do not drift from released packages.

Mongo Startup Indexes

Backend startup should call TaskClient.ainit(amongo=...). The method delegates to the existing package-owned task, task-relation, and task-cost index setup so APIs and workers do not create indexes inside hot read/write paths.

2.4.17 Task-Cost Source Aggregation

TaskCostCreate.from_task_usage(...) is the canonical constructor for durable cost rows. It copies owner, task_type, and root_task from the stored task context and accepts first-class usage source without duplicating it as a tag.

TaskCostAggregateResponse.build(...) now groups by TOTAL, TAG, SOURCE, or TASK_TYPE through TaskCostAggregateGroupBy. List and aggregate reads continue to share TaskCostFilterBase.to_mongo_filters(), and runtime create/list/aggregate paths do not create indexes.