Metadata-Version: 2.4
Name: tai42-backend-arq
Version: 5.0.7
Summary: arq execution backend for the TAI ecosystem — worker runtime and Redis-hash scheduling over arq.
Author-email: tai42 <oss@tai42.ai>
License-Expression: Apache-2.0
Project-URL: Homepage, https://tai42.ai
Project-URL: Repository, https://github.com/tai42ai/tai42/tree/main/plugins/backend-arq
Project-URL: Issues, https://github.com/tai42ai/tai42/issues
Project-URL: Changelog, https://github.com/tai42ai/tai42/blob/main/plugins/backend-arq/CHANGELOG.md
Keywords: mcp,backend,arq,redis,scheduler,tai
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.13
Classifier: Operating System :: OS Independent
Classifier: Typing :: Typed
Requires-Python: >=3.13
Description-Content-Type: text/markdown
License-File: LICENSE
License-File: NOTICE
Requires-Dist: tai42-contract<20,>=19.3
Requires-Dist: tai42-kit[jq,redis]<11,>=10.0
Requires-Dist: arq~=0.28.0
Requires-Dist: croniter~=6.2.2
Requires-Dist: orjson~=3.11.8
Requires-Dist: makefun~=1.16.0
Requires-Dist: fastmcp>=3.4.7
Requires-Dist: click>=8
Requires-Dist: pydantic>=2.12
Requires-Dist: pydantic-settings>=2.15.0
Requires-Dist: pydantic-core>=2
Dynamic: license-file

# tai42-backend-arq

[![License: Apache 2.0](https://img.shields.io/badge/License-Apache_2.0-blue.svg)](LICENSE)

An [arq](https://arq-docs.helpmanual.io/) execution backend for the TAI
ecosystem. It implements the `tai42_contract.backend.Backend` surface — the worker
runtime (`launch`) — and layers the platform's background-execution features
over arq: queued and awaited tool runs, recurring schedules (interval or
crontab) with export/import backup, and result-chaining callbacks. Fleet
propagation of config changes is not this backend's concern: a backend-runtime
process receives fleet ops through the app's own internal worker bus, exactly
like a serving HTTP worker.

## The TAI ecosystem

TAI is an open-source runtime for MCP tools, agents, and workflows. An
execution `Backend` is "how work runs beyond the request" — a pluggable
strategy the runtime uses to queue tool executions on a worker fleet. Fanning
control operations (manifest updates, config/tool/MCP reloads) out across every
worker is the skeleton's worker bus, not the backend — a backend-runtime process
just joins that bus like any other worker. This package is one such backend (arq
over Redis); any package
can back the same contract, so this repo is this provider's own full doc home,
and the documentation site covers the platform-level story:

- Backend concept: https://tai42.ai/concepts/backends
- Build a backend (author guide): https://tai42.ai/guides/authors/backend
- Ecosystem catalog: https://tai42.ai/reference/catalog

Its only tai-* dependencies are `tai42-contract` (the `Backend` ABC,
`CallbackSchema`, `Manifest`, and the `tai42_app` handle) and `tai42-kit[jq]`
(settings base, schedule normalization, signature helpers, jq). Beyond those it
depends on the broker stack — `arq`, `croniter`, `orjson`, `makefun`, `click` —
plus `fastmcp` (the platform's tool substrate) and `pydantic` /
`pydantic-settings` / `pydantic-core`.

## Install

Requires **Python 3.13+**. Install from PyPI into the environment that runs the
server:

```bash
uv add tai42-backend-arq
```

Or from source — clone this repo and add it as an editable dependency; the
`tai42-*` dependencies resolve in-tree from the workspace.

```bash
git clone https://github.com/tai42ai/tai42   # next to your app checkout
cd /path/to/your/app
uv add --editable ../tai42/plugins/backend-arq
```

## Discovery

The host discovers this backend by **importing its package** — importing
`tai42_backend_arq` registers everything through the global `tai42_app` handle as
a side-effect (there is no entry-point): the `ArqBackend`
(`@tai42_app.backends.register_backend`), the `backend_*` tool surface, the
`sync_task` / `schedule_task` / `async_task` BACKEND-kind tool extensions. The
enqueue side (enqueues, job status and results, schedule hashes) runs over the
platform's pooled Redis client, so its connections are pooled per configuration
generation and closed when that generation retires. Name the package in your
manifest's `backend_module` field:

```yaml
backend_module: tai42_backend_arq
```

Start a worker through the host's backend CLI; everything after `worker` is
this backend's own option surface (see `tai42_backend_arq.worker.main`):

```bash
tai backend worker --max-jobs 10
```

## Configuration

Settings are read from the `ARQ_` environment group (see `ArqSettings`):

| Env var | Default | Purpose |
| --- | --- | --- |
| `ARQ_REDIS_URL` | `redis://localhost:6379/0` | Redis connection for the queue and schedules |
| `ARQ_REDIS_MAX_CONNECTIONS` | — | Optional cap on the pool's connections |
| `ARQ_CALLBACK_TIMEOUT` | `5` | Seconds a callback job waits for its predecessor to complete |
| `ARQ_MANIFEST_KEY` | `MANIFEST_KEY` | Name of the env var the host exports the manifest under for worker processes |
| `ARQ_TASK_TIMEOUT` | `300` | The dispatch job budget in seconds: the longest a job may run on the worker (the `--job-timeout` default), and how long a synchronous branch tool waits for its queued job's result |
| `ARQ_TOOL_NAME_ARG` | `backend_tool_name` | kwargs key carrying the target tool name into a queued execution |

`ARQ_MANIFEST_KEY`, `ARQ_TASK_TIMEOUT`, and `ARQ_TOOL_NAME_ARG` mirror the
host's backend settings defaults — both sides
must agree on these values without sharing code, so override them only in
lockstep with the host.

Worker CLI options (after the `worker` subcommand): `--redis-url` (defaults to
`ARQ_REDIS_URL`), `--burst`, `--keep-result` (default `3600`; `0` disables
result retention), `--queue-name`, `--max-jobs` (default `10`),
`--job-timeout` (defaults to `ARQ_TASK_TIMEOUT`), `--poll-delay` (default `0.5`),
`--max-tries` (default `5`), `--health-check-interval` (default `60`). The worker
runs with arq's `allow_abort_jobs` enabled — task cancellation and schedule
replace/delete rely on abort processing. A dispatched tool runs once: the backend
never retries a tool body (a tool that wants retries declares a `ToolRetryPolicy`).
arq itself re-runs a job cut by a cold stop (`CancelledError`), bounded by
`--max-tries` — arq's own delivery semantics.

## Fleet ops

Config-change propagation across the worker fleet (manifest updates,
config/tool/MCP reloads) is the app's own internal worker bus, not a backend
concern — a backend-runtime process receives those ops through the app's bus
subscription exactly like a serving HTTP worker. See the skeleton's worker-bus
concept: https://tai42.ai/concepts/worker-bus

## Schedules

Recurring runs live in per-schedule Redis hashes (`arq:schedule:{name}`)
driven by a self-rescheduling `task_scheduler` job; a startup watchdog
restarts schedules whose pending job was lost. The schedule tools
(`backend_list_schedules` — the contract's `ScheduleListRow`: `name` / `tool` /
`cadence` / `enabled` / `next_run_at_ts` / `next_run_at_iso` / `error`, never the
stored arguments —, `backend_get_schedule`, `backend_delete_schedule`,
`backend_enable_schedule`, `backend_disable_schedule`,
`backend_run_schedule_now`, `backend_schedule_exists`,
`backend_update_schedule`) operate on those hashes, and
`backend_export_schedules` / `backend_import_schedules` round-trip the
contract's `ScheduleRecord` rows for backup (per-row import errors surfaced as
`{"index", "name", "error"}`, never swallowed). A schedule's cadence is the
contract's typed `ScheduleCadence` (`tai42_contract.backend.scheduling`); the hash
stores its JSON dump in the `schedule` field. arq cannot honour a relative
interval, so `relative: true` is refused on create, update and import.

### Upgrading: reset the schedule store

The `schedule` field's stored format is the typed cadence. A hash written in the
earlier format makes the schedule tools (list, export, get, update) raise
`ScheduleStoreFormatError` naming the schedule; arq keeps firing such a hash from
its `cron_or_interval` field. Reset the store and re-create the schedules:

```bash
redis-cli --scan --pattern 'arq:schedule:*' | xargs redis-cli del
```

The pattern is `ArqSettings.arq_schedule_pattern` (`arq:schedule:*`).

## Task execution

The `sync_task` / `async_task` extension branches dispatch the wrapped tool to
the queue via the `tool_execution` job; a `sync_task` wait re-raises a failed
job's stored failure as `TaskFailedError` (aborted jobs included). Task
options exposed on every branch:
`countdown` (defer by seconds), `eta` (ISO datetime; defer until then),
`expires` (seconds the queued job stays runnable), and `callback_kwargs` — a
callback schema chained via a `callback_job` that runs over the primary job's
result (jq condition gate, jq expression transform, optional follow-up tool).
`schedule_task` registers a recurring schedule (see below).

## Tools

Task/worker tools use arq's public `Job` status/result API:
`backend_task_status`, `backend_task_result` (timeout-aware; a task's stored
failure re-raises as `TaskFailedError` carrying the original exception's type,
`repr`, and traceback text — result payloads describe an unserializable value
in a tagged JSON shape the deserializer revives on read), `backend_cancel_task`
(a stored abort replays as arq's confirmed-abort verdict; stored failures are
reported with their detail), `backend_active_tasks` (a flat
job-id-keyed map — arq has no per-worker attribution), `backend_reserved_tasks`
(a flat list of due job ids), `backend_scheduled_tasks` (job id → due time in
epoch milliseconds), `backend_list_failed_tasks` (failed/aborted results
within the keep-result window, as `{"task_id", "error"}` rows carrying the
stored failure detail). Capabilities arq has no reliable data model
for raise `NotImplementedError` loudly: `backend_registered_tasks`,
`backend_worker_stats`, `backend_worker_queues`, `backend_ping_worker`,
`backend_list_active_workers`.

## Development

```bash
uv venv --python 3.13
uv pip install --no-sources --group dev --editable .
uv run --no-sync pytest --cov --cov-report=term-missing
uv run --no-sync ruff check .
uv run --no-sync ruff format --check .
uv run --no-sync pyright
```

## License

Apache-2.0. See `LICENSE` and `NOTICE`.
