agentmux_srv/server/
work_facts_handlers.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! HTTP side of agents' work facts
5//! (docs/specs/SPEC_AGENT_OVERLAP_AWARENESS_2026_10_10.md §3.1, §3.2):
6//!
7//! - `GET /api/v1/work-facts`: this srv's own agents' facts. Other channels
8//!   on this computer call it; it never fans out itself.
9//! - `GET /api/v1/work-facts/who`: the `WhoIsWorkingOn` MCP tool. This srv's
10//!   agents plus every other channel's (through the host-global registry,
11//!   the same reach `ListConversations` has), matched against the question.
12//!   LAN and WAN agents are listed by name only: their facts stay on their
13//!   own computer.
14//!
15//! Both answers include live work claims (`ClaimWork`, §3.4,
16//! `work_claims_handlers`). Also adds the facts to `ListConversations`
17//! entries ([`annotate_conversations`]).
18
19use std::collections::HashMap;
20
21use axum::extract::{Query, State};
22use axum::http::StatusCode;
23use axum::response::{IntoResponse, Json, Response};
24use serde::Deserialize;
25use serde_json::json;
26
27use crate::backend::reactive::types::AgentRegistration;
28use crate::backend::work_facts::cross_channel::{self, own_channel};
29use crate::backend::work_facts::{self, claims, matching, WorkFacts};
30
31use super::caller::Caller;
32use super::AppState;
33
34/// Shown with every answer, so an agent reads it as information.
35const NOTE: &str = "Informational only: nothing is locked or blocked. If someone else is changing the same \
36                    thing, message them (SendMessage) before you change it.";
37
38/// This srv's own agents' facts. Store reads, so off the async workers.
39pub(super) async fn local_facts(state: &AppState) -> Vec<WorkFacts> {
40    let regs = state.reactive_handler.list_agents();
41    let store = state.mstore.clone();
42    let channel = own_channel();
43    tokio::task::spawn_blocking(move || work_facts::collect_local(&store, &regs, &channel))
44        .await
45        .unwrap_or_default()
46}
47
48/// Every other channel's agents' facts, plus the channels that didn't answer.
49pub(super) async fn cross_channel_facts(state: &AppState) -> (Vec<WorkFacts>, Vec<String>) {
50    cross_channel::fetch(&state.http_client, &state.local_web_url).await
51}
52
53/// `GET /api/v1/work-facts`: this srv's own agents' work facts.
54pub async fn handle_work_facts(State(state): State<AppState>) -> impl IntoResponse {
55    Json(json!({ "channel": own_channel(), "agents": local_facts(&state).await }))
56}
57
58/// `WhoIsWorkingOn`'s arguments, plus who is asking (`agent`, `block`: the
59/// MCP's own `AGENTMUX_AGENT_ID` and block id; the request's agent token wins
60/// over both).
61#[derive(Debug, Default, Deserialize)]
62pub struct WhoParams {
63    pub path: Option<String>,
64    pub repo: Option<String>,
65    pub branch: Option<String>,
66    pub query: Option<String>,
67    pub agent: Option<String>,
68    pub block: Option<String>,
69}
70
71/// The caller's own block: by its token's UID, else the block id it sent,
72/// else its name when exactly one registered agent has it. Only used to
73/// leave the caller out and to read its own repository, so a name is enough.
74pub(super) fn caller_block(regs: &[AgentRegistration], uid: Option<&str>, block: Option<&str>, agent: Option<&str>) -> Option<String> {
75    if let Some(uid) = uid {
76        if let Some(r) = regs.iter().find(|r| r.uid.as_deref() == Some(uid)) {
77            return Some(r.block_id.clone());
78        }
79    }
80    let block = block.map(str::trim).filter(|b| !b.is_empty());
81    if let Some(r) = block.and_then(|b| regs.iter().find(|r| r.block_id == b)) {
82        return Some(r.block_id.clone());
83    }
84    let agent = agent.map(str::trim).filter(|a| !a.is_empty())?;
85    let mut named = regs.iter().filter(|r| r.agent_id.eq_ignore_ascii_case(agent));
86    match (named.next(), named.next()) {
87        (Some(r), None) => Some(r.block_id.clone()),
88        _ => None,
89    }
90}
91
92/// Take the caller's own facts out of `facts`.
93pub(super) fn split_caller(facts: Vec<WorkFacts>, caller_block: Option<&str>) -> (Option<WorkFacts>, Vec<WorkFacts>) {
94    let (mine, others): (Vec<_>, Vec<_>) =
95        facts.into_iter().partition(|f| caller_block.is_some_and(|b| f.block_id == b));
96    (mine.into_iter().next(), others)
97}
98
99/// LAN and WAN agents: name and liveness only.
100fn remote_agents(state: &AppState) -> Vec<serde_json::Value> {
101    let mut out = Vec::new();
102    for lan in state.lan_discovery.get_instances() {
103        for name in &lan.agents {
104            out.push(json!({ "agent": name, "channel": "lan", "status": "online" }));
105        }
106    }
107    if let Some(sub) = crate::muxbus::cloud_subscriber::get_global_subscriber() {
108        for name in sub.subscribed_agents() {
109            out.push(json!({ "agent": name, "channel": "wan", "status": "online" }));
110        }
111    }
112    out
113}
114
115/// `GET /api/v1/work-facts/who`: `WhoIsWorkingOn`.
116pub(crate) async fn handle_who_is_working_on(
117    State(state): State<AppState>,
118    caller: Option<axum::Extension<Caller>>,
119    Query(p): Query<WhoParams>,
120) -> Response {
121    let regs = state.reactive_handler.list_agents();
122    let uid = caller.as_ref().and_then(|c| c.0.uid().map(str::to_string));
123    let me_block = caller_block(&regs, uid.as_deref(), p.block.as_deref(), p.agent.as_deref());
124    let store = state.identity_store.clone();
125    let now = agentmux_common::time::now_ms();
126    let live_claims = tokio::task::spawn_blocking(move || store.work_claims_live(now).unwrap_or_default());
127    let (local, (cross, unreachable), live_claims) =
128        tokio::join!(local_facts(&state), cross_channel_facts(&state), live_claims);
129    let (me, mut others) = split_caller(local, me_block.as_deref());
130    others.extend(cross);
131
132    let q = matching::WhoQuery { path: p.path, repo: p.repo, branch: p.branch, query: p.query };
133    let target = match matching::resolve(&q, me.as_ref(), &others) {
134        Ok(t) => t,
135        Err(e) => return (StatusCode::BAD_REQUEST, Json(json!({ "error": e }))).into_response(),
136    };
137    let mut agents = matching::who_is_working_on(&target, &others);
138    // Everyone's claims but the caller's own (by UID, else by name).
139    let me_name = me.as_ref().map(|m| m.agent.clone()).or(p.agent).unwrap_or_default();
140    let theirs: Vec<_> = live_claims
141        .unwrap_or_default()
142        .into_iter()
143        .filter(|c| !claims::is_own(c, &me_name, uid.as_deref().unwrap_or("")))
144        .collect();
145    claims::add_claim_matches(&mut agents, &target, &theirs, &others, now);
146    Json(json!({
147        "asked": target,
148        "caller": me.as_ref().map(|m| json!({ "agent": m.agent, "repo": m.repo, "branch": m.branch })),
149        "agents": agents,
150        "remote_agents": remote_agents(&state),
151        "unreachable_channels": unreachable,
152        "note": NOTE,
153    }))
154    .into_response()
155}
156
157/// Start gathering every agent's facts for `ListConversations`, this srv's
158/// and the other channels' at once, so the handler can run its own preview
159/// fan-out meanwhile rather than paying a second round after it.
160pub(super) fn start_conversation_facts(state: &AppState) -> tokio::task::JoinHandle<Vec<WorkFacts>> {
161    let state = state.clone();
162    tokio::spawn(async move {
163        let (mut local, (cross, _)) = tokio::join!(local_facts(&state), cross_channel_facts(&state));
164        local.extend(cross);
165        local
166    })
167}
168
169/// Add a `work` field (see `work_facts::conversation_summary`) to every
170/// host and cross-channel `ListConversations` entry that names its block,
171/// from the facts [`start_conversation_facts`] gathered. If gathering failed
172/// the entries are left as they were.
173pub(super) async fn annotate_conversations(
174    facts: tokio::task::JoinHandle<Vec<WorkFacts>>,
175    entries: &mut [serde_json::Value],
176) {
177    let facts = facts.await.unwrap_or_default();
178    let by_block: HashMap<&str, &WorkFacts> = facts.iter().map(|f| (f.block_id.as_str(), f)).collect();
179    for e in entries.iter_mut() {
180        let Some(f) = e.get("block_id").and_then(|b| b.as_str()).and_then(|b| by_block.get(b)) else {
181            continue;
182        };
183        let summary = work_facts::conversation_summary(f);
184        if let Some(obj) = e.as_object_mut() {
185            obj.insert("work".to_string(), summary);
186        }
187    }
188}
189
190#[cfg(test)]
191mod tests {
192    use super::*;
193
194    fn reg(agent: &str, block: &str, uid: Option<&str>) -> AgentRegistration {
195        AgentRegistration {
196            agent_id: agent.into(),
197            block_id: block.into(),
198            tab_id: None,
199            uid: uid.map(Into::into),
200            registered_at: 0,
201            last_seen: 0,
202            registration_nonce: 0,
203        }
204    }
205
206    #[test]
207    fn the_caller_is_found_by_token_then_block_then_unique_name() {
208        let regs = vec![reg("AgentY", "b1", Some("u1")), reg("Agent2", "b2", None), reg("Agent2", "b3", None)];
209        assert_eq!(caller_block(&regs, Some("u1"), Some("b2"), None).as_deref(), Some("b1"));
210        assert_eq!(caller_block(&regs, None, Some("b2"), Some("AgentY")).as_deref(), Some("b2"));
211        assert_eq!(caller_block(&regs, None, None, Some("agenty")).as_deref(), Some("b1"));
212        assert_eq!(caller_block(&regs, None, None, Some("Agent2")), None, "an ambiguous name names no one");
213        assert_eq!(caller_block(&regs, None, Some("gone"), None), None);
214    }
215
216    /// The caller is left out of the answer, and its own facts are what
217    /// a relative path resolves against.
218    #[test]
219    fn the_caller_is_excluded_and_its_repository_is_assumed() {
220        let fact = |agent: &str, block: &str, root: &str| WorkFacts {
221            agent: agent.into(),
222            block_id: block.into(),
223            repo: Some("o/r".into()),
224            repo_root: Some(root.into()),
225            dirty_files: vec!["a.rs".into()],
226            ..Default::default()
227        };
228        let (me, others) = split_caller(vec![fact("Me", "b1", "/me"), fact("Other", "b2", "/o")], Some("b1"));
229        assert_eq!(me.as_ref().map(|m| m.agent.as_str()), Some("Me"));
230        let q = matching::WhoQuery { path: Some("a.rs".into()), ..Default::default() };
231        let t = matching::resolve(&q, me.as_ref(), &others).unwrap();
232        let names: Vec<String> = matching::who_is_working_on(&t, &others).into_iter().map(|r| r.agent).collect();
233        assert_eq!(names, vec!["Other".to_string()]);
234
235        let (none, all) = split_caller(vec![fact("A", "b1", "/a")], None);
236        assert!(none.is_none());
237        assert_eq!(all.len(), 1, "an unknown caller excludes no one");
238    }
239
240    #[tokio::test]
241    async fn the_facts_endpoint_lists_this_srvs_agents() {
242        let state = crate::server::tests::test_state();
243        let unique = uuid::Uuid::new_v4();
244        let (agent, block) = (format!("facts-agent-{unique}"), format!("facts-block-{unique}"));
245        state.reactive_handler.register_agent(&agent, &block, None).unwrap();
246        let resp = handle_work_facts(State(state)).await.into_response();
247        assert_eq!(resp.status(), StatusCode::OK);
248        let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX).await.unwrap();
249        let body: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
250        let agents: Vec<WorkFacts> = serde_json::from_value(body["agents"].clone()).unwrap();
251        let mine = agents.iter().find(|f| f.agent == agent).expect("registered agent listed");
252        assert_eq!(mine.block_id, block);
253        assert_eq!(mine.status, "stopped", "no controller in a test");
254    }
255
256    #[tokio::test]
257    async fn who_explains_a_question_it_cannot_answer() {
258        let state = crate::server::tests::test_state();
259        let resp = handle_who_is_working_on(
260            State(state),
261            None,
262            Query(WhoParams { path: Some("relative/x.rs".into()), ..Default::default() }),
263        )
264        .await;
265        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
266    }
267
268    #[tokio::test]
269    async fn conversations_entries_gain_a_work_field_by_block() {
270        let state = crate::server::tests::test_state();
271        let unique = uuid::Uuid::new_v4();
272        let (agent, block) = (format!("conv-agent-{unique}"), format!("conv-block-{unique}"));
273        state.reactive_handler.register_agent(&agent, &block, None).unwrap();
274        let mut entries = vec![
275            json!({ "name": agent, "tier": "host", "block_id": block }),
276            json!({ "name": "remote", "tier": "lan" }),
277        ];
278        annotate_conversations(start_conversation_facts(&state), &mut entries).await;
279        assert!(entries[0]["work"].is_object(), "{}", entries[0]);
280        assert_eq!(entries[0]["work"]["dirty_count"], 0);
281        assert!(entries[1].get("work").is_none(), "LAN entries carry no facts");
282    }
283}
agentmux_srv/server/work_facts_handlers.rs

agentmux_srv/server/
work_facts_handlers.rs
