agentmux_srv/backend/storage/
jekt_held.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Jekts held for an absent agent — `SPEC_DURABLE_JEKT_DELIVERY_2026_09_24.md`
5//! Phase 1. A message to a known agent of this channel that no delivery tier
6//! could take is kept here for up to 24 h and replayed when the agent
7//! registers. The trust verdicts it was accepted with are explicit columns:
8//! on `InjectionRequest` they are `skip_deserializing`, so they would not
9//! survive a JSON round-trip.
10
11use rusqlite::{params, OptionalExtension, Row};
12
13use super::error::StoreError;
14use super::store::Store;
15
16/// Held messages per target UID, and per channel (§2.2).
17pub const HELD_PER_TARGET: i64 = 64;
18pub const HELD_PER_CHANNEL: i64 = 1000;
19/// How long a message is held (§2.1).
20pub const HELD_TTL_MS: i64 = 24 * 60 * 60 * 1000;
21
22#[derive(Debug, Clone, PartialEq)]
23pub struct HeldJekt {
24    pub request_id: String,
25    pub target_uid: String,
26    pub target_agent: String,
27    pub source_agent: String,
28    pub audit_source_uid: String,
29    pub message: String,
30    pub priority: String,
31    pub jekt_tier: String,
32    pub delivery_tier: String,
33    pub sig_verified: Option<bool>,
34    pub reagent_verified: Option<bool>,
35    pub lan_verified: Option<bool>,
36    pub channel_verified: Option<bool>,
37    /// The accept-time transcript-request fields, restored as they were: a
38    /// replay addresses the target by UID, and recomputing them by slug
39    /// would lose a forced escalation (review of #3632).
40    pub is_transcript_request: bool,
41    pub transcript_request_escalate_forced: bool,
42    pub sent_at_ms: i64,
43    pub expires_at_ms: i64,
44    pub attempts: i64,
45    pub last_error: String,
46}
47
48/// What [`Store::jekt_held_insert`] did.
49#[derive(Debug, Clone, Copy, PartialEq, Eq)]
50pub enum HoldOutcome {
51    /// Stored now.
52    Held,
53    /// This `request_id` was already held — nothing stored twice.
54    AlreadyHeld,
55    /// A cap was reached — nothing stored.
56    Full,
57}
58
59fn verdict_to_sql(v: Option<bool>) -> Option<i64> {
60    v.map(i64::from)
61}
62
63fn verdict_from_sql(v: Option<i64>) -> Option<bool> {
64    v.map(|n| n != 0)
65}
66
67fn row_to_held(row: &Row) -> rusqlite::Result<HeldJekt> {
68    Ok(HeldJekt {
69        request_id: row.get("request_id")?,
70        target_uid: row.get("target_uid")?,
71        target_agent: row.get("target_agent")?,
72        source_agent: row.get("source_agent")?,
73        audit_source_uid: row.get("audit_source_uid")?,
74        message: row.get("message")?,
75        priority: row.get("priority")?,
76        jekt_tier: row.get("jekt_tier")?,
77        delivery_tier: row.get("delivery_tier")?,
78        sig_verified: verdict_from_sql(row.get("sig_verified")?),
79        reagent_verified: verdict_from_sql(row.get("reagent_verified")?),
80        lan_verified: verdict_from_sql(row.get("lan_verified")?),
81        channel_verified: verdict_from_sql(row.get("channel_verified")?),
82        is_transcript_request: row.get::<_, i64>("is_transcript_request")? != 0,
83        transcript_request_escalate_forced: row.get::<_, i64>("transcript_request_escalate_forced")? != 0,
84        sent_at_ms: row.get("sent_at_ms")?,
85        expires_at_ms: row.get("expires_at_ms")?,
86        attempts: row.get("attempts")?,
87        last_error: row.get("last_error")?,
88    })
89}
90
91impl Store {
92    /// Hold `held`, unless its `request_id` is already held or a cap is
93    /// reached — decided under the one connection lock, so two concurrent
94    /// holds cannot both slip under a cap.
95    pub fn jekt_held_insert(&self, held: &HeldJekt) -> Result<HoldOutcome, StoreError> {
96        let conn = self.conn().lock().unwrap();
97        let exists: Option<i64> = conn
98            .query_row(
99                "SELECT 1 FROM db_jekt_held WHERE request_id = ?1",
100                params![held.request_id],
101                |r| r.get(0),
102            )
103            .optional()?;
104        if exists.is_some() {
105            return Ok(HoldOutcome::AlreadyHeld);
106        }
107        let n = conn.execute(
108            "INSERT INTO db_jekt_held
109                (request_id, target_uid, target_agent, source_agent, audit_source_uid,
110                 message, priority, jekt_tier, delivery_tier,
111                 sig_verified, reagent_verified, lan_verified, channel_verified,
112                 is_transcript_request, transcript_request_escalate_forced,
113                 sent_at_ms, expires_at_ms, attempts, last_error)
114             SELECT ?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?18, ?19, ?14, ?15, 0, ''
115              WHERE (SELECT COUNT(*) FROM db_jekt_held WHERE target_uid = ?2) < ?16
116                AND (SELECT COUNT(*) FROM db_jekt_held) < ?17",
117            params![
118                held.request_id,
119                held.target_uid,
120                held.target_agent,
121                held.source_agent,
122                held.audit_source_uid,
123                held.message,
124                held.priority,
125                held.jekt_tier,
126                held.delivery_tier,
127                verdict_to_sql(held.sig_verified),
128                verdict_to_sql(held.reagent_verified),
129                verdict_to_sql(held.lan_verified),
130                verdict_to_sql(held.channel_verified),
131                held.sent_at_ms,
132                held.expires_at_ms,
133                HELD_PER_TARGET,
134                HELD_PER_CHANNEL,
135                i64::from(held.is_transcript_request),
136                i64::from(held.transcript_request_escalate_forced),
137            ],
138        )?;
139        Ok(if n > 0 { HoldOutcome::Held } else { HoldOutcome::Full })
140    }
141
142    /// Delete held messages past their expiry; returns how many.
143    pub fn jekt_held_delete_expired(&self, now_ms: i64) -> Result<usize, StoreError> {
144        let conn = self.conn().lock().unwrap();
145        Ok(conn.execute(
146            "DELETE FROM db_jekt_held WHERE expires_at_ms <= ?1",
147            params![now_ms],
148        )?)
149    }
150
151    /// Every target UID with a held message, the one waiting longest first.
152    pub fn jekt_held_targets(&self) -> Result<Vec<String>, StoreError> {
153        let conn = self.conn().lock().unwrap();
154        let mut stmt = conn.prepare(
155            "SELECT target_uid FROM db_jekt_held GROUP BY target_uid ORDER BY MIN(sent_at_ms)",
156        )?;
157        let rows = stmt.query_map([], |r| r.get(0))?;
158        Ok(rows.collect::<Result<Vec<String>, _>>()?)
159    }
160
161    /// One target's held messages, oldest first, at most `limit`.
162    pub fn jekt_held_for_target(&self, target_uid: &str, limit: usize) -> Result<Vec<HeldJekt>, StoreError> {
163        let conn = self.conn().lock().unwrap();
164        let mut stmt = conn.prepare(
165            "SELECT * FROM db_jekt_held WHERE target_uid = ?1
166              ORDER BY sent_at_ms ASC, request_id ASC LIMIT ?2",
167        )?;
168        let rows = stmt.query_map(params![target_uid, limit as i64], row_to_held)?;
169        Ok(rows.collect::<Result<Vec<_>, _>>()?)
170    }
171
172    /// Remove one held message (delivered, or dropped).
173    pub fn jekt_held_delete(&self, request_id: &str) -> Result<(), StoreError> {
174        let conn = self.conn().lock().unwrap();
175        conn.execute("DELETE FROM db_jekt_held WHERE request_id = ?1", params![request_id])?;
176        Ok(())
177    }
178
179    /// Count a failed replay; returns the new attempt count.
180    pub fn jekt_held_record_failure(&self, request_id: &str, error: &str) -> Result<i64, StoreError> {
181        let conn = self.conn().lock().unwrap();
182        conn.execute(
183            "UPDATE db_jekt_held SET attempts = attempts + 1, last_error = ?2 WHERE request_id = ?1",
184            params![request_id, error],
185        )?;
186        Ok(conn
187            .query_row(
188                "SELECT attempts FROM db_jekt_held WHERE request_id = ?1",
189                params![request_id],
190                |r| r.get(0),
191            )
192            .optional()?
193            .unwrap_or(0))
194    }
195}
196
197#[cfg(test)]
198mod tests {
199    use super::*;
200
201    pub(crate) fn held(id: &str, target: &str, sent_at_ms: i64) -> HeldJekt {
202        HeldJekt {
203            request_id: id.into(),
204            target_uid: target.into(),
205            target_agent: "agenty".into(),
206            source_agent: "sender".into(),
207            audit_source_uid: "uid-sender".into(),
208            message: "hello".into(),
209            priority: "normal".into(),
210            jekt_tier: String::new(),
211            delivery_tier: "host".into(),
212            sig_verified: Some(false),
213            reagent_verified: None,
214            lan_verified: None,
215            channel_verified: Some(true),
216            is_transcript_request: true,
217            transcript_request_escalate_forced: true,
218            sent_at_ms,
219            expires_at_ms: sent_at_ms + HELD_TTL_MS,
220            attempts: 0,
221            last_error: String::new(),
222        }
223    }
224
225    #[test]
226    fn holds_once_round_trips_verdicts_and_orders_oldest_first() {
227        let s = Store::open_in_memory().unwrap();
228        assert_eq!(s.jekt_held_insert(&held("r2", "uid-y", 200)).unwrap(), HoldOutcome::Held);
229        assert_eq!(s.jekt_held_insert(&held("r1", "uid-y", 100)).unwrap(), HoldOutcome::Held);
230        assert_eq!(s.jekt_held_insert(&held("r1", "uid-y", 100)).unwrap(), HoldOutcome::AlreadyHeld);
231        let rows = s.jekt_held_for_target("uid-y", 10).unwrap();
232        assert_eq!(rows.iter().map(|r| r.request_id.as_str()).collect::<Vec<_>>(), ["r1", "r2"]);
233        assert_eq!(rows[0].sig_verified, Some(false), "a failed verdict survives");
234        assert_eq!(rows[0].channel_verified, Some(true));
235        assert_eq!(rows[0].reagent_verified, None);
236        assert!(rows[0].is_transcript_request && rows[0].transcript_request_escalate_forced);
237        assert_eq!(s.jekt_held_targets().unwrap(), ["uid-y"]);
238    }
239
240    /// Deleting the target agent purges what is held for it, through both
241    /// deletion paths (Codex P1 on #3632); another agent's held rows stay.
242    #[test]
243    fn deleting_the_target_purges_its_held_messages() {
244        use crate::backend::storage::agents::test_agent_def;
245        for via_instance_delete in [false, true] {
246            let s = Store::open_in_memory().unwrap();
247            for id in ["uid-y", "uid-z"] {
248                let mut def = test_agent_def(id, id, "claude", "agent", 1, "");
249                s.agent_def_insert(&mut def).unwrap();
250            }
251            s.jekt_held_insert(&held("for-y", "uid-y", 1)).unwrap();
252            s.jekt_held_insert(&held("for-z", "uid-z", 1)).unwrap();
253            let deleted = if via_instance_delete {
254                s.instance_delete("uid-y").unwrap()
255            } else {
256                s.agent_def_delete("uid-y").unwrap()
257            };
258            assert!(deleted);
259            assert!(s.jekt_held_for_target("uid-y", 10).unwrap().is_empty(), "instance_delete={via_instance_delete}");
260            assert_eq!(s.jekt_held_for_target("uid-z", 10).unwrap().len(), 1);
261        }
262    }
263
264    #[test]
265    fn caps_refuse_rather_than_overflow() {
266        let s = Store::open_in_memory().unwrap();
267        for i in 0..HELD_PER_TARGET {
268            assert_eq!(s.jekt_held_insert(&held(&format!("r{i}"), "uid-y", i)).unwrap(), HoldOutcome::Held);
269        }
270        assert_eq!(s.jekt_held_insert(&held("one-more", "uid-y", 999)).unwrap(), HoldOutcome::Full);
271        assert_eq!(s.jekt_held_insert(&held("other-target", "uid-z", 999)).unwrap(), HoldOutcome::Held);
272    }
273
274    #[test]
275    fn expiry_failure_count_and_delete() {
276        let s = Store::open_in_memory().unwrap();
277        s.jekt_held_insert(&held("old", "uid-y", 0)).unwrap();
278        s.jekt_held_insert(&held("new", "uid-y", HELD_TTL_MS)).unwrap();
279        assert_eq!(s.jekt_held_delete_expired(HELD_TTL_MS).unwrap(), 1);
280        assert_eq!(s.jekt_held_record_failure("new", "boom").unwrap(), 1);
281        assert_eq!(s.jekt_held_record_failure("new", "boom").unwrap(), 2);
282        assert_eq!(s.jekt_held_for_target("uid-y", 10).unwrap()[0].last_error, "boom");
283        s.jekt_held_delete("new").unwrap();
284        assert!(s.jekt_held_targets().unwrap().is_empty());
285    }
286}
agentmux_srv/backend/storage/jekt_held.rs

agentmux_srv/backend/storage/
jekt_held.rs
