E3: the Kotlin/Java shell over a JNI bridge into Rust (RUST.md)

Two Java classes (MainActivity, NotificationService) hand their lifecycle
to a new android-shell crate built on client-core; client-core gains
notifications.rs (the /notifications SSE parse and attention_line, ported
from Notifications.kt). Packaged as a new app/shellApp Gradle module
rather than a rewrite of app/androidApp in place, so that module's working
Compose UI is untouched.

Both pass conditions held on the emulator: a notification arrived in
Android's drawer with the app closed, and a shared text share landed as a
real message in a sandbox session's transcript. Found and fixed three
real bugs along the way (a silently-wrong JNI signature from a generic
JObject parameter, a class-by-name lookup failing on this crate's own
background thread for lack of an app ClassLoader, and onStartCommand
opening two /notifications connections per enrollment -- the last a
latent bug in Notifications.kt itself). Full account, exact commands and
what was deliberately cut are in RUST.md's E3 box.

Co-Authored-By: Claude Sonnet <noreply@anthropic.com>

client-core/src/notifications.rs

@@ -0,0 +1,162 @@
+//! `GET /notifications`, the attention stream PLAN.md's "Notifications: two
+//! places, never both" describes. Ported from the parsing half of
+//! `app/.../Notifications.kt`'s `NotificationService` -- the framing
+//! ([`crate::sse`]) and the wire shape ([`SessionNotification`],
+//! [`NotificationKind`], mirroring `server/src/session/mod.rs`'s
+//! `Notification`/`NotificationKind`).
+//!
+//! What is deliberately **not** here, because it is a decision rather than
+//! logic: whether a given notification is shown at all (the session on
+//! screen gets nothing), handed to the app as a banner, or posted to the
+//! platform's own notification drawer. That three-way choice reads
+//! process-wide state (what screen is open, whether the app is in front)
+//! that has no meaning to a pure crate with no UI and no Android in it --
+//! see `android-shell` for where it lives for this port.
+
+use std::io::{BufRead, BufReader};
+
+use serde::Deserialize;
+
+use crate::api::{ApiError, Transport};
+use crate::sse::SseReader;
+
+/// One frame of `GET /notifications`, matching `server/src/session/mod.rs`'s
+/// `Notification` field for field.
+#[derive(Debug, Clone, PartialEq, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct SessionNotification {
+    pub session_id: String,
+    pub title: String,
+    pub kind: NotificationKind,
+    /// Epoch seconds, so a phone that was asleep can say how long ago.
+    pub at: f64,
+}
+
+/// Mirrors `server/src/session/mod.rs`'s `NotificationKind` -- serialized
+/// the same way, so this deserializes the wire's `"awaitingInput"` /
+/// `"finished"` directly rather than through a string match.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub enum NotificationKind {
+    AwaitingInput,
+    Finished,
+}
+
+impl NotificationKind {
+    /// What a notification asks of the reader, in the words they see --
+    /// ported verbatim from `Notifications.kt`'s `attentionLine`. One
+    /// function because the same fact is shown in two places (the
+    /// platform's drawer and the app's own banner) and two mappings of one
+    /// word drift.
+    pub fn attention_line(self) -> &'static str {
+        match self {
+            NotificationKind::AwaitingInput => "Waiting for you",
+            NotificationKind::Finished => "Finished",
+        }
+    }
+}
+
+/// Follows `/notifications`, calling `on_notification` for each frame until
+/// the connection drops or the callback asks to stop (by returning
+/// `false`). Reconnecting is the caller's job -- mirroring
+/// `NotificationService.follow`'s retry loop, which is a platform policy
+/// (how long to wait, whether to give up) rather than parsing logic.
+pub fn follow_notifications(
+    transport: &dyn Transport,
+    mut on_notification: impl FnMut(SessionNotification) -> bool,
+) -> Result<(), ApiError> {
+    let body = transport.stream("/notifications")?;
+    let mut lines = BufReader::new(body).lines();
+    let mut reader = SseReader::new();
+    while let Some(line) = lines.next().transpose().map_err(|e| ApiError {
+        message: format!("Can't reach the server -- retrying. ({e})"),
+        status: None,
+    })? {
+        let Some(frame) = reader.feed_line(&line) else {
+            continue;
+        };
+        if frame.data.is_empty() {
+            continue;
+        }
+        let notification: SessionNotification =
+            serde_json::from_str(&frame.data).map_err(|e| ApiError {
+                message: format!("The server sent a notification this build couldn't parse: {e}"),
+                status: None,
+            })?;
+        if !on_notification(notification) {
+            return Ok(());
+        }
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::api::{Body, RawResponse};
+    use std::io::Cursor;
+
+    struct FixtureTransport {
+        body: &'static str,
+    }
+
+    impl Transport for FixtureTransport {
+        fn request(
+            &self,
+            _method: &str,
+            _path: &str,
+            _body: Option<Body>,
+        ) -> Result<RawResponse, ApiError> {
+            unimplemented!("this fixture only serves a stream")
+        }
+
+        fn stream(&self, _path: &str) -> Result<Box<dyn std::io::Read + Send>, ApiError> {
+            Ok(Box::new(Cursor::new(self.body.as_bytes().to_vec())))
+        }
+    }
+
+    #[test]
+    fn a_notification_frame_parses_both_kinds() {
+        let transport = FixtureTransport {
+            body: "data:{\"sessionId\":\"s1\",\"title\":\"fix the bug\",\"kind\":\"awaitingInput\",\"at\":1.0}\n\n\
+                   data:{\"sessionId\":\"s2\",\"title\":\"add tests\",\"kind\":\"finished\",\"at\":2.0}\n\n",
+        };
+        let mut seen = Vec::new();
+        follow_notifications(&transport, |n| {
+            seen.push((n.session_id, n.kind));
+            true
+        })
+        .unwrap();
+        assert_eq!(
+            seen,
+            vec![
+                ("s1".to_string(), NotificationKind::AwaitingInput),
+                ("s2".to_string(), NotificationKind::Finished),
+            ]
+        );
+    }
+
+    #[test]
+    fn the_caller_can_stop_early() {
+        let transport = FixtureTransport {
+            body: "data:{\"sessionId\":\"s1\",\"title\":\"a\",\"kind\":\"finished\",\"at\":1.0}\n\n\
+                   data:{\"sessionId\":\"s2\",\"title\":\"b\",\"kind\":\"finished\",\"at\":2.0}\n\n",
+        };
+        let mut count = 0;
+        follow_notifications(&transport, |_| {
+            count += 1;
+            count < 1
+        })
+        .unwrap();
+        assert_eq!(count, 1);
+    }
+
+    #[test]
+    fn attention_line_matches_the_kotlin_original() {
+        assert_eq!(
+            NotificationKind::AwaitingInput.attention_line(),
+            "Waiting for you"
+        );
+        assert_eq!(NotificationKind::Finished.attention_line(), "Finished");
+    }
+}