feat(watch): stream arriving mail as NDJSON over JMAP push

CHANGELOG.md

@@ -1,5 +1,42 @@
 # Changelog
 
+## [Unreleased]
+
+### Added
+
+- **`fastmail watch`.** Blocks and emits one JSON object per line as mail
+  arrives, so incoming email can drive a shell loop instead of a cron job that
+  re-lists the inbox and diffs it by hand. `--mailbox` narrows to one folder,
+  `--full` includes bodies, and `--poll <seconds>` swaps the push connection for
+  periodic checks where a long-lived one will not survive the network.
+
+  JMAP has offered the transport all along: the session advertises an
+  `eventSourceUrl` (RFC 8620 §7.3) that we parsed and threw away. The design
+  decision worth knowing is that push is only ever a wake-up. The `Email` state
+  cursor lives in the CLI, and every notification — or poll tick, or reconnect —
+  runs `Email/changes` against it, so all three paths converge on the same
+  answer and a lost notification costs latency rather than mail. When the server
+  has discarded history back past the cursor, the watcher resyncs and says so on
+  stderr rather than replaying the mailbox as new; stdout stays pure NDJSON
+  either way.
+
+  Only creations are reported. Reporting updates too would replay every flag
+  change and folder move as an arrival, which is not what a mail loop means by
+  "new".
+
+- **The `emails` GraphQL subscription**, at `/graphql/stream` over Server-Sent
+  Events when the HTTP surface is up. Same watcher as `fastmail watch`, so the
+  cursor semantics are identical rather than merely similar — the CLI and the
+  subscription are two front ends on one implementation, which is the only way
+  they stay that way.
+
+  SSE rather than WebSockets: the subscription is a server-to-client firehose,
+  nothing is ever sent back up the socket, and SSE reconnects on its own.
+
+  MCP deliberately has none of this. Tools are request/response and a
+  subscription never returns, so the `graphql` tool's description says so and
+  points at the other two ways to get it.
+
 ## [3.4.0] - 2026-08-17
 
 ### Added

Cargo.toml

@@ -65,6 +65,8 @@ uuid = { version = "1", features = ["v4"] }
 
 [dev-dependencies]
 tempfile = "3"
+# `test-util` for the paused clock, so backoff tests assert timing without waiting it out.
+tokio = { version = "1", features = ["test-util"] }
 wiremock = "0.6"
 
 [profile.release]

README.md

@@ -13,6 +13,7 @@ CLI for Fastmail's JMAP API. Read, search, send, and manage emails from your ter
 | **Text Extraction**   | 56 formats via [kreuzberg](https://github.com/kreuzberg-dev/kreuzberg) |
 | **Image Resizing**    | `--max-size` to resize images on download                              |
 | **Masked Email**      | Create, list, enable/disable aliases                                   |
+| **Watch**             | Stream arriving mail as NDJSON over JMAP push, for real-time loops     |
 | **MCP Server**        | Claude integration via Model Context Protocol                          |
 | **Shell Completions** | Bash, Zsh, Fish, PowerShell                                            |
 | **JSON Output**       | All commands output JSON for scripting                                 |
@@ -169,6 +170,44 @@ fastmail search --from "boss" --has-attachment --after 2024-06-01 --limit 20
 
 Available flags: `--text`, `--from`, `--to`, `--cc`, `--bcc`, `--subject`, `--body`, `--mailbox`, `--has-attachment`, `--min-size`, `--max-size`, `--before`, `--after`, `--unread`, `--flagged`
 
+### Watch for New Mail
+
+Block and emit one JSON object per line as mail arrives, so a shell loop can act
+on it:
+
+```bash
+# Everything that arrives, anywhere in the account
+fastmail watch
+
+# Just the inbox
+fastmail watch --mailbox inbox
+
+# Pipe into a loop
+fastmail watch --mailbox inbox | while read -r line; do
+  echo "$line" | jq -r '.data.subject'
+done
+
+# Include bodies and attachment metadata, not just summaries
+fastmail watch --full
+
+# Fall back to polling every 60s where a long-lived connection won't survive
+fastmail watch --poll 60
+```
+
+Output is the same `{"success":true,"data":{...}}` envelope as every other
+command, one compact line per email, flushed as it is written — so `jq` filters
+and `read` loops both work unbuffered.
+
+It uses JMAP's push channel (`eventSourceUrl`), but treats a notification purely
+as a signal to look again: the state cursor lives in the CLI, and each wake-up
+runs `Email/changes` against it. Dropped connections are reconciled on reconnect
+and `--poll` takes the identical path, so a missed notification costs latency
+rather than mail. Only *new* messages are reported — flag and folder changes to
+existing mail are not arrivals.
+
+Reconnects, and the rare case where the server has discarded change history and
+the cursor has to resync, are reported on stderr; stdout stays pure NDJSON.
+
 ### List Identities
 
 View available sender identities (useful for `--from`):
@@ -448,7 +487,7 @@ Three independent surfaces, each opt-in, sharing one port (default
 | Flag         | Serves                                                      |
 | ------------ | ----------------------------------------------------------- |
 | `--http`     | MCP streamable-HTTP at `/mcp`                               |
-| `--graphql`  | plain GraphQL-over-HTTP at `/graphql`                       |
+| `--graphql`  | plain GraphQL-over-HTTP at `/graphql`, subscriptions at `/graphql/stream` |
 | `--graphiql` | the GraphiQL IDE at `/`, and its `/graphql`                 |
 | `--browser`  | opens the IDE once the port is bound, implying `--graphiql` |
 
@@ -469,6 +508,23 @@ MCP JSON-RPC, which it doesn't. That is why GraphiQL needs its own route rather
 than pointing at the MCP one. Both share the schema, the client cache and the
 credential resolution below, so the IDE sees exactly what a model sees.
 
+`/graphql/stream` carries subscriptions over Server-Sent Events — POST the
+operation, read events off the response:
+
+```bash
+curl -N http://127.0.0.1:8080/graphql/stream \
+  -H 'Content-Type: application/json' \
+  -d '{"query":"subscription { emails(mailbox: \"inbox\") { id subject from { email } } }"}'
+```
+
+SSE rather than WebSockets because the only subscription is a server-to-client
+firehose: nothing is ever sent back up the socket, and SSE reconnects on its
+own. Body fields (`textBody`, `attachments`) want `full: true` — a subscription
+has no request boundary at which the batching loaders reset, so the lazy path
+resolves through a loader that lives as long as the subscription. `pollSeconds`
+is the same fallback as the CLI's `--poll`. MCP has no equivalent: tools are
+request/response, and a subscription never returns.
+
 **Credential resolution is the same everywhere:** the request's own header
 wins, otherwise the local config (or the matching environment variable) is used.
 Running it yourself, that means your own credentials with no ceremony. In a
@@ -748,6 +804,8 @@ size, surfaced in the field descriptions; it never refuses a query. Being told
 
 All operations are available as GraphQL queries and mutations: mailboxes, emails, search, threads, identities (with signatures), attachments (with text extraction and image resizing), contacts, masked email management, and send/reply/forward with the preview/confirm safety pattern.
 
+One subscription, `emails`, streams arrivals over the same machinery as `fastmail watch`.
+
 Token can be set via `FASTMAIL_API_TOKEN` env var or config file.
 
 ## Debug Logging

src/commands/mod.rs

@@ -12,6 +12,7 @@ mod search;
 mod send;
 mod spam;
 mod thread;
+mod watch;
 
 pub use auth::*;
 pub use contacts::*;
@@ -27,3 +28,4 @@ pub use search::*;
 pub use send::*;
 pub use spam::*;
 pub use thread::*;
+pub use watch::*;

src/commands/watch.rs

@@ -0,0 +1,41 @@
+use crate::jmap::{ArrivalWatcher, authenticated_client};
+use crate::models::Output;
+use std::time::Duration;
+
+pub struct WatchOptions {
+    /// Only report mail landing in this mailbox, by name or role.
+    pub mailbox: Option<String>,
+    /// Fetch bodies and attachment metadata rather than summaries.
+    pub full: bool,
+    /// Check every N seconds instead of holding a push connection open.
+    pub poll: Option<u64>,
+}
+
+/// Stream newly arrived emails as newline-delimited JSON, one object per line,
+/// until interrupted.
+pub async fn watch(opts: WatchOptions) -> anyhow::Result<()> {
+    let client = std::sync::Arc::new(tokio::sync::Mutex::new(authenticated_client().await?));
+
+    let mut watcher = ArrivalWatcher::new(
+        client,
+        opts.mailbox.as_deref(),
+        opts.full,
+        opts.poll.map(Duration::from_secs),
+    )
+    .await?;
+
+    loop {
+        let arrivals = watcher.next_arrivals().await?;
+
+        // Mail may have been lost, and stdout is reserved for mail that wasn't.
+        if arrivals.resynced {
+            eprintln!(
+                "watch: server dropped change history; resynced, some arrivals may be missing"
+            );
+        }
+
+        for email in &arrivals.emails {
+            Output::success(email).print_compact();
+        }
+    }
+}

src/jmap/events.rs

@@ -0,0 +1,153 @@
+//! Server-Sent Events parsing for the JMAP push channel (RFC 8620 §7.3).
+//!
+//! The wire format is [SSE]: `field: value` lines, frames separated by a blank
+//! line. Only `event`, `data` and `id` carry meaning here — `id` is what a
+//! reconnect replays from, via the `Last-Event-ID` header.
+//!
+//! [SSE]: https://html.spec.whatwg.org/multipage/server-sent-events.html
+
+/// One complete SSE frame.
+#[derive(Debug, Default, Clone, PartialEq, Eq)]
+pub struct ServerEvent {
+    pub event: Option<String>,
+    pub data: String,
+    pub id: Option<String>,
+}
+
+/// Reassembles frames from arbitrary byte chunks.
+///
+/// Chunk boundaries fall wherever the network puts them — mid-frame, mid-line,
+/// even between the `\r` and `\n` of a line ending — so partial input is held
+/// until the blank line that terminates a frame actually arrives.
+#[derive(Default)]
+pub struct EventParser {
+    buf: String,
+}
+
+impl EventParser {
+    /// Append a chunk and return every frame it completed.
+    pub fn feed(&mut self, chunk: &str) -> Vec<ServerEvent> {
+        self.buf.push_str(chunk);
+
+        let mut out = Vec::new();
+        while let Some((at, len)) = next_frame_end(&self.buf) {
+            let frame: String = self.buf.drain(..at + len).collect();
+            if let Some(event) = parse_frame(&frame) {
+                out.push(event);
+            }
+        }
+        out
+    }
+}
+
+/// Offset and length of the first frame separator: a blank line, in either
+/// line-ending convention. Returns the earliest match so a stream that mixes
+/// them cannot desynchronise.
+fn next_frame_end(buf: &str) -> Option<(usize, usize)> {
+    let lf = buf.find("\n\n").map(|at| (at, 2));
+    let crlf = buf.find("\r\n\r\n").map(|at| (at, 4));
+    match (lf, crlf) {
+        (Some(a), Some(b)) => Some(if a.0 <= b.0 { a } else { b }),
+        (found, None) | (None, found) => found,
+    }
+}
+
+/// A frame with no recognised field is not an event — that is how keep-alive
+/// comments (`: ping`) stay invisible to callers.
+fn parse_frame(frame: &str) -> Option<ServerEvent> {
+    let mut event = ServerEvent::default();
+    let mut data = Vec::new();
+    let mut recognised = false;
+
+    for line in frame.lines() {
+        if line.is_empty() || line.starts_with(':') {
+            continue;
+        }
+        // A line with no colon is a field with an empty value.
+        let (field, value) = match line.split_once(':') {
+            Some((field, value)) => (field, value.strip_prefix(' ').unwrap_or(value)),
+            None => (line, ""),
+        };
+        match field {
+            "event" => event.event = Some(value.to_string()),
+            "data" => data.push(value),
+            "id" => event.id = Some(value.to_string()),
+            // `retry` and unknown fields are ignored: reconnect backoff is the
+            // caller's, and it has better information than the server does.
+            _ => continue,
+        }
+        recognised = true;
+    }
+
+    recognised.then(|| {
+        event.data = data.join("\n");
+        event
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn parses_a_whole_frame() {
+        let mut parser = EventParser::default();
+        let events = parser.feed("event: state\ndata: {\"x\":1}\nid: abc\n\n");
+        assert_eq!(
+            events,
+            vec![ServerEvent {
+                event: Some("state".into()),
+                data: "{\"x\":1}".into(),
+                id: Some("abc".into()),
+            }]
+        );
+    }
+
+    #[test]
+    fn holds_a_frame_split_across_chunks() {
+        let mut parser = EventParser::default();
+        assert!(parser.feed("event: sta").is_empty());
+        assert!(parser.feed("te\ndata: {\"x\":1}").is_empty());
+        let events = parser.feed("\n\n");
+        assert_eq!(events.len(), 1);
+        assert_eq!(events[0].data, "{\"x\":1}");
+    }
+
+    #[test]
+    fn holds_a_frame_split_between_cr_and_lf() {
+        let mut parser = EventParser::default();
+        assert!(parser.feed("data: hi\r\n\r").is_empty());
+        let events = parser.feed("\ndata: there\r\n\r\n");
+        assert_eq!(events.len(), 2);
+        assert_eq!(events[0].data, "hi");
+        assert_eq!(events[1].data, "there");
+    }
+
+    #[test]
+    fn joins_repeated_data_lines() {
+        let mut parser = EventParser::default();
+        let events = parser.feed("data: one\ndata: two\n\n");
+        assert_eq!(events[0].data, "one\ntwo");
+    }
+
+    #[test]
+    fn yields_several_frames_from_one_chunk() {
+        let mut parser = EventParser::default();
+        let events = parser.feed("data: a\n\ndata: b\n\ndata: c\n\n");
+        assert_eq!(events.len(), 3);
+    }
+
+    #[test]
+    fn skips_comment_only_frames() {
+        let mut parser = EventParser::default();
+        assert!(parser.feed(": ping\n\n").is_empty());
+    }
+
+    #[test]
+    fn tolerates_a_missing_space_after_the_colon() {
+        let mut parser = EventParser::default();
+        let events = parser.feed("event:state\ndata:{}\n\n");
+        assert_eq!(events[0].event.as_deref(), Some("state"));
+        assert_eq!(events[0].data, "{}");
+    }
+}