fix(desktop): bound HTTP event submission

desktop/src-tauri/src/commands/engram_submit_response.rs

@@ -0,0 +1,73 @@
+/// Apply the shared relay-body timeout classification while preserving the
+/// snapshot paths' established detail for every non-timeout read failure.
+pub(crate) async fn read_engram_submit_response(
+    response: reqwest::Response,
+) -> Result<String, String> {
+    response.text().await.map_err(|error| {
+        crate::relay::classify_body_read_error(&error, || {
+            format!("failed to read relay response: {error}")
+        })
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn body_stall_surfaces_classified_timeout() {
+        use std::io::{Read as _, Write as _};
+        use std::time::Duration;
+
+        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
+        let addr = listener.local_addr().unwrap();
+        let handle = std::thread::spawn(move || {
+            if let Ok((mut stream, _)) = listener.accept() {
+                let mut buf = [0u8; 4096];
+                let _ = stream.read(&mut buf);
+                let _ = stream.write_all(
+                    b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: 64\r\n\r\n",
+                );
+                let _ = stream.flush();
+                std::thread::sleep(Duration::from_secs(1));
+            }
+        });
+        let response = reqwest::Client::new()
+            .get(format!("http://{addr}/events"))
+            .timeout(Duration::from_millis(200))
+            .send()
+            .await
+            .expect("headers must arrive before the deadline");
+        let err = read_engram_submit_response(response).await.unwrap_err();
+
+        assert_eq!(err, "relay unreachable: request timed out");
+        let _ = handle.join();
+    }
+
+    #[tokio::test]
+    async fn incomplete_body_keeps_existing_non_timeout_prefix() {
+        use std::io::{Read as _, Write as _};
+
+        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
+        let addr = listener.local_addr().unwrap();
+        let handle = std::thread::spawn(move || {
+            if let Ok((mut stream, _)) = listener.accept() {
+                let mut buf = [0u8; 4096];
+                let _ = stream.read(&mut buf);
+                let _ = stream.write_all(
+                    b"HTTP/1.1 200 OK\r\nContent-Length: 64\r\nConnection: close\r\n\r\nshort",
+                );
+                let _ = stream.flush();
+            }
+        });
+        let response = reqwest::Client::new()
+            .get(format!("http://{addr}/events"))
+            .send()
+            .await
+            .expect("headers must arrive");
+        let err = read_engram_submit_response(response).await.unwrap_err();
+
+        assert!(err.starts_with("failed to read relay response: "), "{err}");
+        let _ = handle.join();
+    }
+}

desktop/src-tauri/src/commands/mod.rs

@@ -18,6 +18,7 @@ mod channel_window;
 mod channels;
 mod clipboard;
 mod dms;
+pub(crate) mod engram_submit_response;
 mod engrams;
 mod export_util;
 mod global_agent_config;

desktop/src-tauri/src/commands/personas/snapshot/import.rs

@@ -811,39 +811,29 @@ pub(crate) async fn submit_engram_event(
     url: &str,
     auth_tag: Option<&str>,
 ) -> Result<(), String> {
-    use crate::relay::build_nip98_auth_header_for_keys;
-    use reqwest::Method;
-
     crate::egress_guard::assert_no_key_backup_bytes(event_json, "persona snapshot engram submit")?;
 
     // Wait before signing: the relay enforces NIP-98 freshness (±60s) and the
     // gate may hold for up to MAX_HINT_SECONDS (300s). Building auth before the
     // wait produces a stale `created_at` that the relay will reject.
     crate::relay_admission::wait_for_rate_limit().await;
-    let auth = build_nip98_auth_header_for_keys(agent_keys, &Method::POST, url, event_json)?;
-    let mut request = state
-        .http_client
-        .post(url)
-        .header("Authorization", auth)
-        .header("Content-Type", "application/json");
-    if let Some(tag) = auth_tag {
-        request = request.header("x-auth-tag", tag);
-    }
-    let response = request
-        .body(event_json.to_vec())
-        .send()
-        .await
-        .map_err(|e| crate::relay::classify_request_error(&e))?;
-
-    if !response.status().is_success() {
-        let msg = crate::relay::relay_error_message(response).await;
-        return Err(format!("relay rejected engram: {msg}"));
-    }
+    let body = match crate::relay::submit_event_text_with_keys(
+        &state.http_client,
+        url,
+        agent_keys,
+        auth_tag,
+        event_json,
+    )
+    .await
+    {
+        Ok(body) => body,
+        Err(crate::relay::EventSubmitHttpError::Auth(error)) => return Err(error),
+        Err(crate::relay::EventSubmitHttpError::Rejected(error)) => {
+            return Err(format!("relay rejected engram: {error}"));
+        }
+        Err(error) => return Err(error.into_message()),
+    };
 
-    let body = response
-        .text()
-        .await
-        .map_err(|e| format!("failed to read relay response: {e}"))?;
     let parsed: serde_json::Value =
         serde_json::from_str(&body).map_err(|e| format!("relay response not JSON: {e}"))?;
     let accepted = parsed

desktop/src-tauri/src/commands/team_snapshot.rs

@@ -911,39 +911,29 @@ pub(crate) async fn submit_engram_event(
     url: &str,
     auth_tag: Option<&str>,
 ) -> Result<(), String> {
-    use crate::relay::build_nip98_auth_header_for_keys;
-    use reqwest::Method;
-
     crate::egress_guard::assert_no_key_backup_bytes(event_json, "team snapshot engram submit")?;
 
     // Wait before signing: the relay enforces NIP-98 freshness (±60s) and the
     // gate may hold for up to MAX_HINT_SECONDS (300s). Building auth before the
     // wait produces a stale `created_at` that the relay will reject.
     crate::relay_admission::wait_for_rate_limit().await;
-    let auth = build_nip98_auth_header_for_keys(agent_keys, &Method::POST, url, event_json)?;
-    let mut request = state
-        .http_client
-        .post(url)
-        .header("Authorization", auth)
-        .header("Content-Type", "application/json");
-    if let Some(tag) = auth_tag {
-        request = request.header("x-auth-tag", tag);
-    }
-    let response = request
-        .body(event_json.to_vec())
-        .send()
-        .await
-        .map_err(|e| crate::relay::classify_request_error(&e))?;
-
-    if !response.status().is_success() {
-        let msg = crate::relay::relay_error_message(response).await;
-        return Err(format!("relay rejected engram: {msg}"));
-    }
+    let body = match crate::relay::submit_event_text_with_keys(
+        &state.http_client,
+        url,
+        agent_keys,
+        auth_tag,
+        event_json,
+    )
+    .await
+    {
+        Ok(body) => body,
+        Err(crate::relay::EventSubmitHttpError::Auth(error)) => return Err(error),
+        Err(crate::relay::EventSubmitHttpError::Rejected(error)) => {
+            return Err(format!("relay rejected engram: {error}"));
+        }
+        Err(error) => return Err(error.into_message()),
+    };
 
-    let body = response
-        .text()
-        .await
-        .map_err(|e| format!("failed to read relay response: {e}"))?;
     let parsed: serde_json::Value =
         serde_json::from_str(&body).map_err(|e| format!("relay response not JSON: {e}"))?;
     let accepted = parsed

desktop/src-tauri/src/egress_guard_tests.rs

@@ -249,47 +249,47 @@ fn src_rust_files() -> Vec<std::path::PathBuf> {
 }
 
 /// Site-granular `/events` inventory: `(file suffix, expected non-comment
-/// `/events` occurrences, expected guard call sites — full-path calls into
-/// the egress-guard module)`.
+/// `/events` occurrences, expected guard call sites, expected calls into the
+/// bounded, idempotent HTTP event-submit funnel)`.
 ///
-/// Every entry pairs the URL-construction count with the guard-call count for
-/// that file, so BOTH of these fail the scan (not just a brand-new file):
+/// Every production entry pairs the URL-construction count with both safety
+/// boundaries for that file, so all of these fail the scan (not just a
+/// brand-new file):
 ///   - adding an unguarded ninth `/events` site inside an already-listed file
-///     (count goes up without a matching table update), and
-///   - removing/refactoring away a guard call while its egress site remains.
+///     (count goes up without a matching table update),
+///   - removing/refactoring away a guard call while its egress site remains,
+///   - bypassing the retrying event-submit funnel and its per-request deadline.
 ///
-/// Updating a row here is the deliberate act that must accompany wiring the
-/// guard + adding an injection test for the new site.
-const EVENTS_INVENTORY: &[(&str, usize, usize)] = &[
+/// Updating a row here is the deliberate act that must accompany wiring both
+/// boundaries and adding an injection test for a new production site.
+const EVENTS_INVENTORY: &[(&str, usize, usize, usize)] = &[
     // Production egress boundaries (see egress_guard.rs table):
-    ("src/relay.rs", 2, 2),                             // boundaries 2, 4
-    ("src/relay/submit.rs", 1, 1),                      // boundaries 1 + 3 (shared funnel)
-    ("src/huddle/pipeline.rs", 1, 1),                   // boundary 5
-    ("src/commands/team_snapshot.rs", 1, 1),            // boundary 6
-    ("src/commands/personas/snapshot/import.rs", 2, 1), // boundary 7 + its in-file injection-test fixture URL
-    ("src/native_websocket.rs", 0, 2),                  // boundary 8 (WS frames; no events URL)
-    // Test-only fixtures — no production egress, no guard:
-    ("src/relay_admission.rs", 1, 0),
-    ("src/archive/mod_tests.rs", 1, 0),
-    ("src/managed_agents/persona_events/tests.rs", 1, 0),
-    ("src/commands/team_snapshot/tests.rs", 1, 0),
+    ("src/relay.rs", 2, 2, 2),                  // boundaries 2, 4
+    ("src/relay/submit.rs", 1, 1, 1),           // boundaries 1 + 3 (shared funnel)
+    ("src/huddle/pipeline.rs", 1, 1, 1),        // boundary 5
+    ("src/commands/team_snapshot.rs", 1, 1, 1), // boundary 6
+    ("src/commands/personas/snapshot/import.rs", 2, 1, 1), // boundary 7 + its in-file injection-test fixture URL
+    ("src/native_websocket.rs", 0, 2, 0),                  // boundary 8 (WS frames; no events URL)
+    // Test-only fixtures — no production egress, no guard or production submit call:
+    ("src/relay/tests.rs", 5, 0, 0),
+    ("src/commands/engram_submit_response.rs", 2, 0, 0),
+    ("src/relay_admission.rs", 1, 0, 0),
+    ("src/archive/mod_tests.rs", 1, 0, 0),
+    ("src/managed_agents/persona_events/tests.rs", 1, 0, 0),
+    ("src/commands/team_snapshot/tests.rs", 1, 0, 0),
     // Mock-relay route in its in-file tests; production publish goes through
     // the guarded boundary-1 funnel (`submit_signed_event_at_with_keys`).
-    ("src/commands/personas/sharing.rs", 1, 0),
+    ("src/commands/personas/sharing.rs", 1, 0, 0),
     // Loopback submit relay in `identity_archive.rs`'s in-file regen tests;
     // production archive/unarchive publish through the guarded boundary-1
     // funnel via `submit_event`.
-    ("src/commands/identity_archive.rs", 1, 0),
-    // Mock-relay routes in team-sharing tests (accept/reject stub +
-    // recording stub for the delete-then-share gate + gated recording stub for
-    // the two-flush serialization gate + stalling stub for the per-scope
-    // isolation and bounded-stall gates); same pattern as persona sharing
-    // above — production publish goes through the guarded boundary-1 funnel via
-    // the flush loop.
-    ("src/commands/teams/sharing/tests.rs", 4, 0),
+    ("src/commands/identity_archive.rs", 1, 0, 0),
+    // Mock-relay routes in team-sharing tests; production publish goes through
+    // the guarded boundary-1 funnel via the flush loop.
+    ("src/commands/teams/sharing/tests.rs", 4, 0, 0),
     // Stub-relay route in the tombstone-flush gate tests; production flush
     // publishes through the guarded boundary-1 funnel.
-    ("src/commands/teams/pending/tests/gate.rs", 1, 0),
+    ("src/commands/teams/pending/tests/gate.rs", 1, 0, 0),
 ];
 
 // Needles are assembled at runtime so this scan file itself contains no
@@ -300,11 +300,27 @@ fn events_needle() -> String {
 fn guard_needle() -> String {
     ["egress_guard::", "assert_no_key_backup"].concat()
 }
+fn event_submit_call_count(content: &str) -> usize {
+    let needles = [
+        ["send_event_", "http_request_with_keys("].concat(),
+        ["submit_event_", "json_with_keys("].concat(),
+        ["submit_event_", "text_with_keys("].concat(),
+    ];
+    content
+        .lines()
+        .filter(|line| {
+            needles.iter().any(|needle| line.contains(needle))
+                && !line.trim_start().starts_with("//")
+                && !line.contains("fn send_event_")
+                && !line.contains("fn submit_event_")
+        })
+        .count()
+}
 
 /// Pure scan core over `(relative path, content)` pairs. Returns violations;
 /// empty means every file matches its inventory row exactly (files absent
-/// from the table are expected to have zero `/events` sites and zero guard
-/// calls).
+/// from the table are expected to have zero `/events` sites, guard calls, and
+/// bounded HTTP event-submit calls).
 fn events_inventory_violations(files: &[(String, String)]) -> Vec<String> {
     let events = events_needle();
     let guard = guard_needle();
@@ -313,9 +329,9 @@ fn events_inventory_violations(files: &[(String, String)]) -> Vec<String> {
     for (rel, content) in files {
         let expected = EVENTS_INVENTORY
             .iter()
-            .find(|(suffix, _, _)| rel.ends_with(suffix))
-            .map(|&(_, e, g)| (e, g))
-            .unwrap_or((0, 0));
+            .find(|(suffix, _, _, _)| rel.ends_with(suffix))
+            .map(|&(_, e, g, s)| (e, g, s))
+            .unwrap_or((0, 0, 0));
 
         let mut event_sites = Vec::new();
         for (i, line) in content.lines().enumerate() {
@@ -327,15 +343,18 @@ fn events_inventory_violations(files: &[(String, String)]) -> Vec<String> {
             }
         }
         let guard_count = content.matches(&guard).count();
+        let event_submit_count = event_submit_call_count(content);
 
-        if (event_sites.len(), guard_count) != expected {
+        if (event_sites.len(), guard_count, event_submit_count) != expected {
             violations.push(format!(
-                "{rel}: found {} events-URL site(s) + {} guard call(s), inventory \
-                 expects {} + {}. Sites found:\n{}",
+                "{rel}: found {} events-URL site(s) + {} guard call(s) + {} bounded-submit \
+                 call(s), inventory expects {} + {} + {}. Sites found:\n{}",
                 event_sites.len(),
                 guard_count,
+                event_submit_count,
                 expected.0,
                 expected.1,
+                expected.2,
                 if event_sites.is_empty() {
                     "  (none)".to_string()
                 } else {
@@ -365,9 +384,9 @@ fn read_src_files() -> Vec<(String, String)> {
 
 /// Inventory completeness: every `/events` URL-construction site in
 /// `desktop/src-tauri/src` must match the site-granular inventory above. A
-/// future ninth submission path — in a NEW file or an ALREADY-LISTED one —
-/// fails this test until its guard is wired, its injection test exists, and
-/// its inventory row is updated.
+/// future ninth production submission path — in a NEW file or an
+/// ALREADY-LISTED one — fails this test until its guard and bounded HTTP funnel
+/// are wired, its injection test exists, and its inventory row is updated.
 #[test]
 fn events_url_inventory_is_fully_guarded() {
     let violations = events_inventory_violations(&read_src_files());
@@ -416,6 +435,30 @@ fn inventory_scan_catches_removed_guard_call() {
     );
 }
 
+/// The timeout/retry boundary also fires in reverse: a production event path
+/// that bypasses the bounded idempotent funnel while retaining its `/events`
+/// URL and egress guard is caught.
+#[test]
+fn inventory_scan_catches_removed_bounded_submit_call() {
+    let mut files = read_src_files();
+    let huddle = files
+        .iter_mut()
+        .find(|(rel, _)| rel.ends_with("src/huddle/pipeline.rs"))
+        .expect("huddle pipeline must be in the scan set");
+    huddle.1 = huddle.1.replacen(
+        &["send_event_", "http_request_with_keys("].concat(),
+        "unbounded_event_submit(",
+        1,
+    );
+    let violations = events_inventory_violations(&files);
+    assert!(
+        violations
+            .iter()
+            .any(|v| v.contains("src/huddle/pipeline.rs")),
+        "a removed bounded event-submit call must trip the scan: {violations:?}"
+    );
+}
+
 /// A brand-new file with an `/events` site (no inventory row) is caught.
 #[test]
 fn inventory_scan_catches_new_unlisted_file() {