agentmux_srv/server/
work_claims_handlers.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! HTTP side of work claims
5//! (docs/specs/SPEC_AGENT_OVERLAP_AWARENESS_2026_10_10.md §3.4):
6//!
7//! - `POST /api/v1/work-claims`: `ClaimWork`. Records (or renews) the
8//!   caller's claim on a path, branch or topic, and answers with who else is
9//!   already working on it.
10//! - `POST /api/v1/work-claims/release`: `ReleaseWork`. Releases one of the
11//!   caller's claims, or all of them.
12//!
13//! Claims are stored in the identity store (`storage::work_claims`), so
14//! every channel's `WhoIsWorkingOn` and overlap notes see them. Nothing is
15//! ever locked: a claim only informs.
16
17use axum::extract::State;
18use axum::http::StatusCode;
19use axum::response::{IntoResponse, Json, Response};
20use serde::Deserialize;
21use serde_json::json;
22
23use crate::backend::storage::work_claims::WorkClaim;
24use crate::backend::work_facts::cross_channel::own_channel;
25use crate::backend::work_facts::{claims, matching};
26
27use super::caller::Caller;
28use super::work_facts_handlers::{caller_block, cross_channel_facts, local_facts, split_caller};
29use super::AppState;
30
31/// `ClaimWork`'s arguments, plus who is asking (`agent`, `block`: the MCP's
32/// own `AGENTMUX_AGENT_ID` and block id; the request's agent token wins).
33#[derive(Debug, Default, Deserialize)]
34pub struct ClaimParams {
35    pub path: Option<String>,
36    pub repo: Option<String>,
37    pub branch: Option<String>,
38    pub topic: Option<String>,
39    pub note: Option<String>,
40    pub ttl_minutes: Option<u64>,
41    pub agent: Option<String>,
42    pub block: Option<String>,
43}
44
45/// `ReleaseWork`'s arguments: a claim id, or none for all of the caller's.
46#[derive(Debug, Default, Deserialize)]
47pub struct ReleaseParams {
48    pub id: Option<String>,
49    pub agent: Option<String>,
50    pub block: Option<String>,
51}
52
53fn bad_request(msg: impl Into<String>) -> Response {
54    (StatusCode::BAD_REQUEST, Json(json!({ "error": msg.into() }))).into_response()
55}
56
57/// The longest note, topic, path, branch and repository a claim keeps, in
58/// characters. Another agent reads them in its notes and answers.
59const MAX_NOTE: usize = 200;
60const MAX_TOPIC: usize = 120;
61const MAX_PATH: usize = 400;
62const MAX_NAME: usize = 200;
63
64/// A claim's free text as it is kept: one line (control characters and line
65/// breaks become spaces, whitespace runs collapse) of at most `max`
66/// characters. Marker quoting happens where the text is shown inside a
67/// system note (`work_facts::overlap`).
68fn one_line(s: &Option<String>, max: usize) -> Option<String> {
69    let flat: String = s.as_deref()?.chars().map(|c| if c.is_control() { ' ' } else { c }).collect();
70    let flat = flat.split_whitespace().collect::<Vec<_>>().join(" ");
71    if flat.is_empty() {
72        return None;
73    }
74    Some(match flat.char_indices().nth(max) {
75        Some((cut, _)) => format!("{}…", &flat[..cut]),
76        None => flat,
77    })
78}
79
80fn nonempty(s: &Option<String>) -> Option<&str> {
81    s.as_deref().map(str::trim).filter(|s| !s.is_empty())
82}
83
84/// Who is asking: the name and UID of the caller's registration (found by
85/// token, block, then unique name, as `WhoIsWorkingOn` does), else the name
86/// it sent. `None` when there is no name at all.
87fn claimer(
88    state: &AppState,
89    caller: Option<&Caller>,
90    agent: Option<&str>,
91    block: Option<&str>,
92) -> Option<(String, String, Option<String>)> {
93    let regs = state.reactive_handler.list_agents();
94    let token_uid = caller.and_then(|c| c.uid().map(str::to_string));
95    let block_id = caller_block(&regs, token_uid.as_deref(), block, agent);
96    let reg = block_id.as_deref().and_then(|b| regs.iter().find(|r| r.block_id == b));
97    let name = reg
98        .map(|r| r.agent_id.clone())
99        .or_else(|| agent.map(str::trim).filter(|a| !a.is_empty()).map(str::to_string))?;
100    let uid = token_uid.or_else(|| reg.and_then(|r| r.uid.clone())).unwrap_or_default();
101    Some((name, uid, block_id))
102}
103
104/// `POST /api/v1/work-claims`: `ClaimWork`.
105pub(crate) async fn handle_claim_work(
106    State(state): State<AppState>,
107    caller: Option<axum::Extension<Caller>>,
108    Json(p): Json<ClaimParams>,
109) -> Response {
110    let p = ClaimParams {
111        path: one_line(&p.path, MAX_PATH),
112        repo: one_line(&p.repo, MAX_NAME),
113        branch: one_line(&p.branch, MAX_NAME),
114        topic: one_line(&p.topic, MAX_TOPIC),
115        note: one_line(&p.note, MAX_NOTE),
116        ..p
117    };
118    if [&p.path, &p.repo, &p.branch, &p.topic].iter().all(|v| nonempty(v).is_none()) {
119        return bad_request("nothing to claim: pass path, branch or topic (and repo for a whole repository)");
120    }
121    let Some((agent, uid, me_block)) =
122        claimer(&state, caller.as_ref().map(|c| &c.0), p.agent.as_deref(), p.block.as_deref())
123    else {
124        return bad_request("who is claiming is unknown: this needs an agent started by AgentMux");
125    };
126    let (local, (cross, _)) = tokio::join!(local_facts(&state), cross_channel_facts(&state));
127    let (me, mut others) = split_caller(local, me_block.as_deref());
128    others.extend(cross);
129
130    let q = matching::WhoQuery {
131        path: p.path.clone(),
132        repo: p.repo.clone(),
133        branch: p.branch.clone(),
134        query: p.topic.clone(),
135    };
136    let mut target = match matching::resolve(&q, me.as_ref(), &others) {
137        Ok(t) => t,
138        Err(e) => return bad_request(e),
139    };
140    // Only a repository named: the whole of it.
141    if [&p.path, &p.branch, &p.topic].iter().all(|v| nonempty(v).is_none()) {
142        target.path = Some(String::new());
143    }
144
145    let now = agentmux_common::time::now_ms();
146    let claim = WorkClaim {
147        id: uuid::Uuid::new_v4().to_string(),
148        agent: agent.clone(),
149        agent_uid: uid.clone(),
150        channel: own_channel(),
151        repo: target.repo.clone(),
152        path: target.path.clone(),
153        absolute_path: target.absolute_path.clone(),
154        branch: target.branch.clone(),
155        topic: target.query.clone(),
156        note: p.note.clone().unwrap_or_default(),
157        created_at: now,
158        expires_at: now + claims::ttl_ms(p.ttl_minutes),
159    };
160    let store = state.identity_store.clone();
161    let stored = tokio::task::spawn_blocking(move || {
162        let stored = store.work_claim_put(&claim, now)?;
163        let live = store.work_claims_live(now)?;
164        Ok::<_, crate::backend::storage::StoreError>((stored, live))
165    })
166    .await;
167    let (stored, live) = match stored {
168        Ok(Ok(v)) => v,
169        Ok(Err(e)) => {
170            return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({ "error": format!("claim not saved: {e}") })))
171                .into_response()
172        }
173        Err(e) => {
174            return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({ "error": format!("claim not saved: {e}") })))
175                .into_response()
176        }
177    };
178
179    // Who else is already on it, their claims included.
180    let theirs: Vec<WorkClaim> = live.into_iter().filter(|c| !claims::is_own(c, &agent, &uid)).collect();
181    let others: Vec<_> = others.into_iter().filter(|f| !f.agent.eq_ignore_ascii_case(&agent)).collect();
182    let mut overlapping = matching::who_is_working_on(&target, &others);
183    claims::add_claim_matches(&mut overlapping, &target, &theirs, &others, now);
184    Json(json!({
185        "claim": stored,
186        "claimed": claims::describe(&stored),
187        "others_working_on_it": overlapping,
188        "note": "Claimed. Other agents see this in WhoIsWorkingOn and are told when they edit a file in it. It \
189                 blocks no one, and lapses at expires_at unless you claim it again. Release it with ReleaseWork \
190                 when you are done. If others_working_on_it names anyone, message them (SendMessage).",
191    }))
192    .into_response()
193}
194
195/// `POST /api/v1/work-claims/release`: `ReleaseWork`.
196pub(crate) async fn handle_release_work(
197    State(state): State<AppState>,
198    caller: Option<axum::Extension<Caller>>,
199    Json(p): Json<ReleaseParams>,
200) -> Response {
201    let Some((agent, uid, _)) = claimer(&state, caller.as_ref().map(|c| &c.0), p.agent.as_deref(), p.block.as_deref())
202    else {
203        return bad_request("who is releasing is unknown: this needs an agent started by AgentMux");
204    };
205    let id = nonempty(&p.id).map(str::to_string);
206    let store = state.identity_store.clone();
207    let now = agentmux_common::time::now_ms();
208    let released = tokio::task::spawn_blocking(move || store.work_claim_release(&agent, &uid, id.as_deref(), now)).await;
209    match released {
210        Ok(Ok(released)) => {
211            let note = if released.is_empty() {
212                "Nothing released: you hold no live claim with that id."
213            } else {
214                "Released."
215            };
216            Json(json!({ "released": released, "note": note })).into_response()
217        }
218        Ok(Err(e)) => (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({ "error": format!("release failed: {e}") })))
219            .into_response(),
220        Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({ "error": format!("release failed: {e}") })))
221            .into_response(),
222    }
223}
224
225#[cfg(test)]
226mod tests {
227    use super::*;
228
229    async fn body(resp: Response) -> (StatusCode, serde_json::Value) {
230        let status = resp.status();
231        let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX).await.unwrap();
232        (status, serde_json::from_slice(&bytes).unwrap())
233    }
234
235    #[tokio::test]
236    async fn a_claim_needs_something_to_claim_and_someone_claiming() {
237        let state = crate::server::tests::test_state();
238        let (status, b) = body(handle_claim_work(State(state.clone()), None, Json(ClaimParams::default())).await).await;
239        assert_eq!(status, StatusCode::BAD_REQUEST);
240        assert!(b["error"].as_str().unwrap().contains("nothing to claim"));
241        let anon = ClaimParams { topic: Some("x".into()), ..Default::default() };
242        let (status, b) = body(handle_claim_work(State(state), None, Json(anon)).await).await;
243        assert_eq!(status, StatusCode::BAD_REQUEST);
244        assert!(b["error"].as_str().unwrap().contains("who is claiming"));
245    }
246
247    #[test]
248    fn claim_text_is_kept_as_one_short_line() {
249        assert_eq!(one_line(&Some("  a\n[AgentMux] b\t\u{7}c  ".into()), 50).as_deref(), Some("a [AgentMux] b c"));
250        assert_eq!(one_line(&Some("abcdef".into()), 3).as_deref(), Some("abc…"));
251        assert_eq!(one_line(&Some("ééé".into()), 3).as_deref(), Some("ééé"), "counted in characters");
252        assert_eq!(one_line(&Some(" \n ".into()), 3), None);
253        assert_eq!(one_line(&None, 3), None);
254    }
255
256    #[tokio::test]
257    async fn claim_who_and_release_round_trip() {
258        let state = crate::server::tests::test_state();
259        let unique = uuid::Uuid::new_v4().simple().to_string();
260        let (me, other) = (format!("claimer-{unique}"), format!("asker-{unique}"));
261        let topic = format!("topic{unique}");
262
263        let claim = ClaimParams {
264            topic: Some(topic.clone()),
265            note: Some("presence work".into()),
266            ttl_minutes: Some(30),
267            agent: Some(me.clone()),
268            ..Default::default()
269        };
270        let (status, b) = body(handle_claim_work(State(state.clone()), None, Json(claim)).await).await;
271        assert_eq!(status, StatusCode::OK, "{b}");
272        assert_eq!(b["claim"]["agent"], me.as_str());
273        assert_eq!(b["claim"]["topic"], topic.as_str());
274        let id = b["claim"]["id"].as_str().unwrap().to_string();
275
276        // Another agent asking about the topic sees the claim.
277        let who = super::super::work_facts_handlers::WhoParams {
278            query: Some(topic.clone()),
279            agent: Some(other.clone()),
280            ..Default::default()
281        };
282        let (status, b) = body(
283            super::super::work_facts_handlers::handle_who_is_working_on(
284                State(state.clone()),
285                None,
286                axum::extract::Query(who),
287            )
288            .await,
289        )
290        .await;
291        assert_eq!(status, StatusCode::OK, "{b}");
292        let agents = b["agents"].as_array().unwrap();
293        let found = agents.iter().find(|a| a["agent"] == me.as_str()).expect("the claimer is listed");
294        assert_eq!(found["matches"][0]["kind"], "claim");
295        assert_eq!(found["status"], "not running", "no registration in this test");
296
297        // Only the claimer releases it.
298        let wrong = ReleaseParams { id: Some(id.clone()), agent: Some(other), ..Default::default() };
299        let (_, b) = body(handle_release_work(State(state.clone()), None, Json(wrong)).await).await;
300        assert!(b["released"].as_array().unwrap().is_empty());
301        let mine = ReleaseParams { id: Some(id), agent: Some(me), ..Default::default() };
302        let (_, b) = body(handle_release_work(State(state), None, Json(mine)).await).await;
303        assert_eq!(b["released"].as_array().unwrap().len(), 1);
304    }
305}
agentmux_srv/server/work_claims_handlers.rs

agentmux_srv/server/
work_claims_handlers.rs
