agentq 0.1.0

An embeddable job queue with idempotency keys, per-lane backpressure and bounded concurrency, for agent tool calls.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82

use std::collections::HashMap;
use std::sync::{Arc, Mutex};

use tokio::sync::oneshot;

use crate::handle::Outcome;
use crate::job::Key;

#[derive(Debug, Clone)]
pub enum State {
    Pending,
    Processing,
    Completed { output: String },
    Failed { reason: String },
}

impl From<&Outcome> for State {
    fn from(outcome: &Outcome) -> State {
        match outcome {
            Outcome::Completed { output } => State::Completed {
                output: output.clone(),
            },
            Outcome::Failed { reason } => State::Failed {
                reason: reason.clone(),
            },
        }
    }
}

pub(crate) type StateMap = Arc<Mutex<HashMap<Key, State>>>;

pub(crate) type WaiterMap = Arc<Mutex<HashMap<Key, Vec<oneshot::Sender<Outcome>>>>>;

pub(crate) struct JobGuard {
    pub(crate) statemap: StateMap,
    pub(crate) waiters: WaiterMap,
    pub(crate) key: Key,
    pub(crate) outcome: Option<Outcome>,
}

impl Drop for JobGuard {
    fn drop(&mut self) {
        let outcome = self.outcome.take().unwrap_or(Outcome::Failed {
            reason: "job panicked".to_string(),
        });

        {
            let mut statemap = self.statemap.lock().unwrap_or_else(|e| e.into_inner());
            statemap.insert(self.key.clone(), State::from(&outcome));
        }

        let senders = {
            let mut waiters = self.waiters.lock().unwrap_or_else(|e| e.into_inner());
            waiters.remove(&self.key).unwrap_or_default()
        };

        for sender in senders {
            let _ = sender.send(outcome.clone());
        }
    }
}

pub(crate) struct ClaimGuard {
    pub(crate) statemap: StateMap,
    pub(crate) waiters: WaiterMap,
    pub(crate) key: Key,
    pub(crate) armed: bool,
}

impl Drop for ClaimGuard {
    fn drop(&mut self) {
        if !self.armed {
            return;
        }

        let mut statemap = self.statemap.lock().unwrap_or_else(|e| e.into_inner());
        let mut waiters = self.waiters.lock().unwrap_or_else(|e| e.into_inner());

        statemap.remove(&self.key);
        waiters.remove(&self.key);
    }
}