# agentq
[](https://crates.io/crates/agentq)
[](https://docs.rs/agentq)
[](https://agentq.hiesenbug.dev)
[](https://github.com/Arshdeep54/agentq/actions/workflows/rust.yml)
[](LICENSE)
> An embeddable job queue for agent tool calls. Idempotency keys so a retry
> doesn't re-execute work that already succeeded, bounded lanes for
> backpressure, and a concurrency limit per lane.
>
> No Redis. No separate service. A crate you drop into your own process.
---
## Why
Agent frameworks retry failed tool calls. If a call actually succeeded but the
acknowledgment was lost, a naive retry runs it again: duplicate writes,
duplicate charges, corrupted state. And when a batch of calls fails at once,
the retries all fire at once too, against whatever rate-limited or metered API
you were talking to.
Durable execution platforms solve this, but they want you to run a separate
service and adopt their workflow model. agentq is the small version: the
primitives that stop the bleeding, embedded directly in your binary.
## Features
- **Idempotency keys.** Push a key that already completed and you get that
job's output back from cache instead of running it again. Push one that's
still running and you join it, receiving the same outcome when it lands.
- **Per-lane concurrency.** Each priority lane has its own capacity and its
own concurrency limit, so cheap work and expensive work get different
budgets and a saturated lane never starves another.
- **Backpressure.** Lanes are bounded. When one fills, `push` waits rather
than letting the queue grow without limit.
- **Failure isolation.** A job that returns an error, or panics outright,
records a terminal state and leaves every other job untouched.
- **Cancel safe.** Dropping a `push` future part-way through leaves no
claimed key behind.
- **Durable workflows.** An ordered list of steps with per-step retry and
backoff, driven through the same queue, with status and event history
persisted by a pluggable `DurableStore`. A `Waiting` step can block on
something outside the process (an approval, a webhook) and be unblocked
later with `resume`. Ships with a SQLite-backed store behind the optional
`sqlite` feature — off by default, so consumers who only want the raw
queue never pull in `rusqlite`.
## Quick start
```toml
[dependencies]
agentq = "0.2"
```
```rust
use agentq::{Job, LaneConfig, Priority, Queue};
let queue = Queue::builder()
.lane(Priority::High, LaneConfig { capacity: 32, permits: 1 })
.lane(Priority::Low, LaneConfig { capacity: 128, permits: 4 })
.start();
let job = Job::new(
"charge-order-4821".to_string(),
Priority::High,
Box::new(|| Box::pin(async { Ok("charged".to_string()) })),
);
let output = queue.push_and_wait(job).await?;
```
Guides covering idempotency, priority lanes, error handling and the
architecture are at **[agentq.hiesenbug.dev](https://agentq.hiesenbug.dev)**.
The generated API reference is on [docs.rs](https://docs.rs/agentq).
## Durable workflows
For multi-step work that needs to survive a restart, enable the `sqlite`
feature:
```toml
[dependencies]
agentq = { version = "0.2", features = ["sqlite"] }
```
```rust
use agentq::{Backoff, Queue, RetryPolicy, SqliteStore, StepDef, Workflow, WorkflowEngine};
use std::time::Duration;
let engine = WorkflowEngine::new(
Queue::builder().start(),
SqliteStore::new("workflows.db")?,
"worker-1".to_string(),
Duration::from_secs(30),
agentq::Priority::High,
);
let workflow = Workflow {
id: "onboard-user-42".to_string(),
steps: vec![StepDef {
name: "send-welcome-email".to_string(),
retry_policy: RetryPolicy {
max_attempts: 3,
backoff: Backoff::Fixed(Duration::from_secs(5)),
},
timeout: None,
}],
};
engine.run(workflow, vec![Box::new(|| Box::pin(async { Ok("sent".to_string()) }))]).await?;
```
The engine retries a failed step per its `RetryPolicy` and records every
transition as an `Event`, both through the same `Queue` and `Job` V1 uses —
`Queue` itself stays exactly as it was, retry-agnostic and in-memory. A step
that returns `WaitForInput` parks as `Waiting` until you call
`engine.resume(workflow_id, step_index, input)`; a step whose worker died
mid-lease gets picked back up by `agentq::recover(&engine)`.
`DurableStore` is a trait — `SqliteStore` is the one implementation this
crate ships, but a consumer who wants Postgres or an in-memory store for
tests can implement it directly.
## Caveats
A job that pushes to its own queue and waits on the result can deadlock, if
every permit in that lane is held by jobs doing the same thing.
Keys and their cached outputs are retained for the life of the process, so a
producer with unbounded distinct keys grows memory. Dropping the queue
abandons in-flight work.
`Queue` itself deliberately does not persist anything and does not coordinate
across machines. For a step's state to survive a restart, use a `Workflow`
with a `DurableStore` instead of the raw `Queue` — but agentq still does not
coordinate across machines; if you need that, you want a different tool.
## Upcoming
- Capped retries with exponential backoff for the raw `Queue` — a plain job
is still recorded as failed and not re-attempted; workflows already have
this via `RetryPolicy`
- Time-windowed keys, so cached outputs expire
- Graceful shutdown, draining in-flight work before exit
- Arbitration between lanes, so `High` can preempt `Low`
## License
MIT