feat: add Groove feedback-o-tron and inter-service health mesh

Add three new endpoints to PanLL's groove server:
- POST /.well-known/groove/feedback — accepts and routes feedback events
  through the Groove mesh to any connected service
- GET /.well-known/groove/mesh — returns cached health view of all groove
  peers discovered via periodic probing
- GET /.well-known/groove/status — health status endpoint for mesh probing

Mesh monitor runs on a background thread, probing known groove ports
(6473, 8080, 8081, 8091-8093) every 30 seconds. Feedback routing looks
up the target service in the mesh and forwards via raw HTTP.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: hyperpolymath

src-gossamer/src/groove.rs

@@ -35,6 +35,30 @@
 
 use std::io::{Read, Write};
 use std::net::{SocketAddr, TcpListener, TcpStream};
+use std::sync::Mutex;
+use std::time::{SystemTime, UNIX_EPOCH};
+
+use once_cell::sync::Lazy;
+
+/// Known groove peers for health mesh probing.
+/// PanLL probes these ports to discover neighbouring services.
+const MESH_PROBE_PORTS: &[u16] = &[6473, 8080, 8081, 8091, 8092, 8093];
+
+/// Cached mesh health state: peer service_id -> status.
+static MESH_STATE: Lazy<Mutex<Vec<PeerHealth>>> = Lazy::new(|| Mutex::new(Vec::new()));
+
+/// Health status of a single peer in the mesh.
+#[derive(Clone)]
+struct PeerHealth {
+    /// Peer's self-reported service ID (or "unknown" if probe failed).
+    service_id: String,
+    /// Port the peer was discovered on.
+    port: u16,
+    /// "up", "degraded", or "down".
+    status: String,
+    /// Unix timestamp (milliseconds) of last successful probe.
+    last_seen_ms: u64,
+}
 
 /// The groove capability manifest for PanLL.
 /// Matches Groove.idr panllManifest: offers [PanelUI (websocket, /ws)].
@@ -50,6 +74,22 @@ const MANIFEST: &str = r#"{
       "endpoint": "/ws",
       "requires_auth": false,
       "panel_compatible": true
+    },
+    "feedback": {
+      "type": "feedback",
+      "description": "Groove-routed feedback collector — accepts and routes feedback to any mesh service",
+      "protocol": "http",
+      "endpoint": "/.well-known/groove/feedback",
+      "requires_auth": false,
+      "panel_compatible": true
+    },
+    "health_mesh": {
+      "type": "health-mesh",
+      "description": "Inter-service health mesh — monitors peer status via groove probing",
+      "protocol": "http",
+      "endpoint": "/.well-known/groove/mesh",
+      "requires_auth": false,
+      "panel_compatible": true
     }
   },
   "consumes": ["voice", "text", "presence", "integrity", "octad-storage", "scanning"],
@@ -82,6 +122,9 @@ pub fn spawn() {
             }
         })
         .expect("Failed to spawn groove server thread");
+
+    // Start the mesh health monitor (probes peers every 30s).
+    spawn_mesh_monitor();
 }
 
 /// Run the groove discovery HTTP server (blocking).
@@ -128,6 +171,34 @@ fn handle_request(stream: &mut TcpStream) -> Result<(), Box<dyn std::error::Erro
             send_response(stream, 200, "application/json", MANIFEST)?;
         }
 
+        // GET /.well-known/groove/status — Health status for mesh probing.
+        ("GET", "/.well-known/groove/status") => {
+            send_response(
+                stream,
+                200,
+                "application/json",
+                r#"{"status":"ok","service":"panll","groove_version":"1"}"#,
+            )?;
+        }
+
+        // GET /.well-known/groove/mesh — Return the cached mesh health view.
+        ("GET", "/.well-known/groove/mesh") => {
+            let mesh_json = build_mesh_json();
+            send_response(stream, 200, "application/json", &mesh_json)?;
+        }
+
+        // POST /.well-known/groove/feedback — Receive feedback routed via Groove.
+        //
+        // Schema: { "type": "feedback", "target_service": string,
+        //           "category": string, "message": string, "metadata": object }
+        //
+        // If target_service == "panll", saves locally via feedback::commands.
+        // Otherwise, routes to the target service through the mesh.
+        ("POST", "/.well-known/groove/feedback") => {
+            let body = extract_body(request);
+            handle_groove_feedback(stream, &body)?;
+        }
+
         // GET /health — Simple health check.
         ("GET", "/health") => {
             send_response(
@@ -147,6 +218,269 @@ fn handle_request(stream: &mut TcpStream) -> Result<(), Box<dyn std::error::Erro
     Ok(())
 }
 
+/// Extract the HTTP body from a raw request string.
+///
+/// Finds the blank line separating headers from body and returns everything
+/// after it. Returns an empty string if no body is present.
+fn extract_body(request: &str) -> String {
+    if let Some(idx) = request.find("\r\n\r\n") {
+        request[idx + 4..].to_string()
+    } else if let Some(idx) = request.find("\n\n") {
+        request[idx + 2..].to_string()
+    } else {
+        String::new()
+    }
+}
+
+/// Handle a POST /.well-known/groove/feedback request.
+///
+/// Validates the feedback JSON, saves it locally if targeted at PanLL,
+/// or attempts to route it to the target service via the mesh.
+fn handle_groove_feedback(
+    stream: &mut TcpStream,
+    body: &str,
+) -> Result<(), Box<dyn std::error::Error>> {
+    let parsed: Result<serde_json::Value, _> = serde_json::from_str(body);
+    let feedback = match parsed {
+        Ok(v) => v,
+        Err(_) => {
+            send_response(stream, 400, "application/json", r#"{"ok":false,"error":"invalid JSON"}"#)?;
+            return Ok(());
+        }
+    };
+
+    let target = feedback
+        .get("target_service")
+        .and_then(|v| v.as_str())
+        .unwrap_or("panll");
+
+    if target == "panll" {
+        // Save locally — delegate to feedback module's storage.
+        let dir = dirs::home_dir()
+            .map(|h| h.join(".panll").join("feedback"))
+            .ok_or("Cannot determine home directory")?;
+        let _ = std::fs::create_dir_all(&dir);
+
+        let timestamp = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .unwrap_or_default()
+            .as_millis();
+
+        let wrapped = serde_json::json!({
+            "id": format!("groove-feedback-{timestamp}"),
+            "timestamp": timestamp,
+            "source": "groove",
+            "report": feedback,
+        });
+
+        let filepath = dir.join(format!("{timestamp}.json"));
+        let content = serde_json::to_string_pretty(&wrapped).unwrap_or_default();
+        let _ = std::fs::write(&filepath, content);
+
+        let resp = serde_json::json!({
+            "ok": true,
+            "routed_to": "panll",
+            "id": format!("groove-feedback-{timestamp}"),
+        });
+        send_response(stream, 200, "application/json", &resp.to_string())?;
+    } else {
+        // Route to the target service via the mesh.
+        match route_feedback_to_peer(target, body) {
+            Ok(()) => {
+                let resp = serde_json::json!({
+                    "ok": true,
+                    "routed_to": target,
+                });
+                send_response(stream, 200, "application/json", &resp.to_string())?;
+            }
+            Err(e) => {
+                let resp = serde_json::json!({
+                    "ok": false,
+                    "error": format!("routing failed: {e}"),
+                    "target_service": target,
+                });
+                send_response(stream, 502, "application/json", &resp.to_string())?;
+            }
+        }
+    }
+
+    Ok(())
+}
+
+/// Attempt to route feedback to a peer service in the mesh.
+///
+/// Looks up the peer's port from the cached mesh state and POSTs the
+/// feedback to their `/.well-known/groove/feedback` endpoint.
+fn route_feedback_to_peer(target: &str, body: &str) -> Result<(), Box<dyn std::error::Error>> {
+    let peers = MESH_STATE.lock().map_err(|_| "mesh lock poisoned")?;
+    let peer = peers
+        .iter()
+        .find(|p| p.service_id == target && p.status == "up")
+        .ok_or_else(|| format!("peer '{target}' not found or not up in mesh"))?;
+
+    let addr = format!("127.0.0.1:{}", peer.port);
+    let mut stream = TcpStream::connect(&addr)?;
+    stream.set_write_timeout(Some(std::time::Duration::from_secs(2)))?;
+    stream.set_read_timeout(Some(std::time::Duration::from_secs(2)))?;
+
+    let request = format!(
+        "POST /.well-known/groove/feedback HTTP/1.0\r\nHost: {}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
+        addr,
+        body.len(),
+        body
+    );
+    stream.write_all(request.as_bytes())?;
+    Ok(())
+}
+
+/// Build a JSON string representing the current mesh health view.
+fn build_mesh_json() -> String {
+    let peers = MESH_STATE.lock().unwrap_or_else(|e| e.into_inner());
+    let now_ms = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .unwrap_or_default()
+        .as_millis() as u64;
+
+    let peer_entries: Vec<String> = peers
+        .iter()
+        .map(|p| {
+            format!(
+                r#"{{"service_id":"{}","port":{},"status":"{}","last_seen_ms":{}}}"#,
+                p.service_id, p.port, p.status, p.last_seen_ms
+            )
+        })
+        .collect();
+
+    format!(
+        r#"{{"service_id":"panll","timestamp_ms":{},"peers":[{}]}}"#,
+        now_ms,
+        peer_entries.join(",")
+    )
+}
+
+/// Probe known groove peers and update the cached mesh state.
+///
+/// Called periodically from the mesh monitor thread. For each port in
+/// `MESH_PROBE_PORTS`, sends `GET /.well-known/groove/status` and records
+/// the result (up/down + service_id).
+pub fn probe_mesh_peers() {
+    let now_ms = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .unwrap_or_default()
+        .as_millis() as u64;
+
+    let mut results = Vec::new();
+
+    for &port in MESH_PROBE_PORTS {
+        // Skip our own port.
+        if port == 8000 {
+            continue;
+        }
+
+        let addr = format!("127.0.0.1:{}", port);
+        let status = match TcpStream::connect_timeout(
+            &addr.parse().unwrap(),
+            std::time::Duration::from_millis(500),
+        ) {
+            Ok(mut stream) => {
+                stream
+                    .set_read_timeout(Some(std::time::Duration::from_millis(500)))
+                    .ok();
+                stream
+                    .set_write_timeout(Some(std::time::Duration::from_millis(500)))
+                    .ok();
+
+                let request = format!(
+                    "GET /.well-known/groove/status HTTP/1.0\r\nHost: {}\r\nConnection: close\r\n\r\n",
+                    addr
+                );
+                if stream.write_all(request.as_bytes()).is_ok() {
+                    let mut buf = vec![0u8; 4096];
+                    let service_id = match stream.read(&mut buf) {
+                        Ok(n) if n > 0 => {
+                            let resp = String::from_utf8_lossy(&buf[..n]);
+                            // Extract service_id from JSON body.
+                            extract_service_id_from_response(&resp)
+                        }
+                        _ => "unknown".to_string(),
+                    };
+                    PeerHealth {
+                        service_id,
+                        port,
+                        status: "up".to_string(),
+                        last_seen_ms: now_ms,
+                    }
+                } else {
+                    PeerHealth {
+                        service_id: "unknown".to_string(),
+                        port,
+                        status: "down".to_string(),
+                        last_seen_ms: 0,
+                    }
+                }
+            }
+            Err(_) => PeerHealth {
+                service_id: "unknown".to_string(),
+                port,
+                status: "down".to_string(),
+                last_seen_ms: 0,
+            },
+        };
+
+        // Only include peers that responded (up or previously known).
+        if status.status == "up" {
+            results.push(status);
+        }
+    }
+
+    if let Ok(mut state) = MESH_STATE.lock() {
+        *state = results;
+    }
+}
+
+/// Extract the service_id field from a groove status JSON response.
+///
+/// Performs a simple substring search to avoid pulling in a full JSON parser
+/// for probe responses. Falls back to "unknown" if not found.
+fn extract_service_id_from_response(response: &str) -> String {
+    // Find the JSON body (after headers).
+    let body = if let Some(idx) = response.find("\r\n\r\n") {
+        &response[idx + 4..]
+    } else if let Some(idx) = response.find("\n\n") {
+        &response[idx + 2..]
+    } else {
+        response
+    };
+
+    // Try to parse as JSON for the service_id field.
+    if let Ok(v) = serde_json::from_str::<serde_json::Value>(body) {
+        if let Some(id) = v.get("service").and_then(|s| s.as_str()) {
+            return id.to_string();
+        }
+        if let Some(id) = v.get("service_id").and_then(|s| s.as_str()) {
+            return id.to_string();
+        }
+    }
+
+    "unknown".to_string()
+}
+
+/// Spawn the mesh health monitor thread.
+///
+/// Probes known groove peers every 30 seconds and updates the cached mesh
+/// state. This runs alongside the groove HTTP server.
+pub fn spawn_mesh_monitor() {
+    std::thread::Builder::new()
+        .name("panll-mesh-monitor".into())
+        .spawn(|| {
+            loop {
+                probe_mesh_peers();
+                std::thread::sleep(std::time::Duration::from_secs(30));
+            }
+        })
+        .expect("Failed to spawn mesh monitor thread");
+}
+
 /// Send an HTTP response with the given content type and body.
 fn send_response(
     stream: &mut TcpStream,
@@ -156,8 +490,10 @@ fn send_response(
 ) -> Result<(), Box<dyn std::error::Error>> {
     let status_text = match status {
         200 => "OK",
+        204 => "No Content",
         400 => "Bad Request",
         404 => "Not Found",
+        502 => "Bad Gateway",
         _ => "Unknown",
     };
     let response = format!(