Metadata-Version: 2.4
Name: keble-task
Version: 2.22.0
Author-email: zhenhao-ma <bob0103779@gmail.com>
Requires-Python: <3.14,>=3.13
Requires-Dist: keble-db<2.0.0,>=1.7.1
Requires-Dist: keble-helpers<2.0.0,>=1.40.0
Requires-Dist: pydantic-ai-slim<2.0.0,>=1.41.0
Requires-Dist: tenacity<10.0.0,>=9
Provides-Extra: test
Requires-Dist: httpx[socks]<1.0.0,>=0.27; extra == 'test'
Requires-Dist: pydantic-ai-slim[openai]<2.0.0,>=1.41.0; extra == 'test'
Requires-Dist: pytest-asyncio<1.0.0,>=0; extra == 'test'
Requires-Dist: pytest<9.0.0,>=8; extra == 'test'
Description-Content-Type: text/markdown

# 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:

```bash
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:

```bash
uv run pytest -q
```

Run the live LLM canary:

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

Run static syntax/type checking:

```bash
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

```bash
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.

```python
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()
```

```python
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.

```python
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:

```python
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:

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

### TaskBase

The base schema for task data:

```python
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.

```python
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:

```python
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:

```python
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.

```python
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.

```python
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:

```python
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:

```python
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:

```python
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:

```python
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:

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

### TokenConsumptionPayload

A schema for token consumption operations:

```python
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:

```python
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:

```python
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:

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

## Examples

### Backend API: Creating a TaskClient

```python
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:

```python
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.
