agentq

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

[dependencies]
agentq = "0.2"

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. The generated API reference is on docs.rs.

Durable workflows

For multi-step work that needs to survive a restart, enable the sqlite feature:

[dependencies]
agentq = { version = "0.2", features = ["sqlite"] }

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.

Example app

durable-agent is a full application built on agentq: an AI support-ticket workflow that classifies tickets with an LLM, waits for email approval, and survives worker crashes mid-workflow.

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