agentmux_srv/server/
http_fleet.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! `GET /agentmux/fleet` and `GET /agentmux/fleet/events`: this instance's
5//! agent names for a LAN peer holding the `lan_key`, as a snapshot and as a
6//! Server-Sent Events stream (`backend::fleet_feed`,
7//! `docs/specs/SPEC_LAN_FLEET_FEED_2026_10_03.md`). Names plus instance
8//! metadata a LAN peer already sees in the mDNS record; nothing else about an
9//! agent.
10
11use super::*;
12
13/// Does `If-None-Match` name `etag`? Accepts a list, weak validators, `*`,
14/// and a tag a client sent without its quotes.
15fn if_none_match_matches(header_value: Option<&axum::http::HeaderValue>, etag: &str) -> bool {
16    let Some(value) = header_value.and_then(|v| v.to_str().ok()) else {
17        return false;
18    };
19    let bare = etag.trim_matches('"');
20    value.split(',').map(str::trim).any(|tag| {
21        let tag = tag.strip_prefix("W/").unwrap_or(tag);
22        tag == "*" || tag == etag || tag == bare
23    })
24}
25
26/// The current snapshot, `304 Not Modified` (no body) when the client's
27/// `If-None-Match` already names it.
28pub(super) async fn handle_fleet(
29    State(state): State<AppState>,
30    headers: axum::http::HeaderMap,
31) -> Response {
32    let feed = &state.fleet_feed;
33    let snapshot = feed.snapshot();
34    let etag = format!("\"{}\"", feed.event_id(&snapshot));
35    if if_none_match_matches(headers.get(header::IF_NONE_MATCH), &etag) {
36        return (
37            StatusCode::NOT_MODIFIED,
38            [
39                (header::ETAG, etag),
40                (header::CACHE_CONTROL, "no-cache".to_string()),
41            ],
42        )
43            .into_response();
44    }
45    (
46        [
47            (header::CONTENT_TYPE, "application/json".to_string()),
48            (header::ETAG, etag),
49            (header::CACHE_CONTROL, "no-cache".to_string()),
50        ],
51        feed.body_json(&snapshot),
52    )
53        .into_response()
54}
55
56/// The event stream: `retry`, the snapshot unless `Last-Event-ID` is already
57/// current, then one `fleet` event per change and a heartbeat comment every
58/// 15 s. 503 once `fleet_feed::MAX_STREAMS` streams are open.
59pub(super) async fn handle_fleet_events(
60    State(state): State<AppState>,
61    headers: axum::http::HeaderMap,
62) -> Response {
63    let last_event_id = headers
64        .get("last-event-id")
65        .and_then(|v| v.to_str().ok())
66        .map(str::trim);
67    let Some(stream) = state.fleet_feed.open_stream(last_event_id) else {
68        return (
69            StatusCode::SERVICE_UNAVAILABLE,
70            Json(json!({"error": "too many fleet event streams"})),
71        )
72            .into_response();
73    };
74    (
75        [
76            (header::CONTENT_TYPE, "text/event-stream"),
77            (header::CACHE_CONTROL, "no-cache"),
78        ],
79        Body::from_stream(stream),
80    )
81        .into_response()
82}
83
84#[cfg(test)]
85mod tests {
86    use super::if_none_match_matches;
87    use axum::http::HeaderValue;
88
89    #[test]
90    fn if_none_match_handles_lists_weak_tags_and_wildcards() {
91        let etag = "\"abc:2\"";
92        let check = |v: &str| if_none_match_matches(Some(&HeaderValue::from_str(v).unwrap()), etag);
93        assert!(check("\"abc:2\""));
94        assert!(check("W/\"abc:2\""));
95        assert!(check("\"abc:1\", \"abc:2\""));
96        assert!(check("*"));
97        assert!(check("abc:2"), "a client that dropped the quotes");
98        assert!(!check("\"abc:1\""));
99        assert!(!check(""));
100        assert!(!if_none_match_matches(None, etag));
101    }
102}
agentmux_srv/server/http_fleet.rs

agentmux_srv/server/
http_fleet.rs
