feat(grpc): add authenticated Subscribe event stream (push channel)

Add System.Subscribe server-streaming RPC on the authenticated System service,
mapping every CoreEvent variant to a nuncio.v1.Event oneof. The daemon streams
live EventBus broadcasts (skip-on-lag with a warning; clean close when the bus
drops). Add a client subscribe helper in nuncio-proto. Regression test proves
20 concurrent unary GetStatus calls and 20 streamed events arrive uncorrupted
and in order -- the property the old hand-rolled JSON-RPC IPC demux lacked.

Refs #151

Cargo.lock

@@ -4386,6 +4386,7 @@ dependencies = [
  "futures-core",
  "pin-project-lite",
  "tokio",
+ "tokio-util",
 ]
 
 [[package]]

crates/nuncio-cli/src/runner.rs

@@ -1084,7 +1084,7 @@ mod tests {
     #[tokio::test]
     async fn system_status_reports_live_daemon_status_over_grpc_when_reachable() {
         use nuncio_proto::v1::system_server::{System as SystemService, SystemServer};
-        use nuncio_proto::v1::{GetStatusRequest, GetStatusResponse};
+        use nuncio_proto::v1::{Event, GetStatusRequest, GetStatusResponse, SubscribeRequest};
 
         /// Minimal test-only stub of the `nuncio.v1.System` service: no
         /// auth interceptor, just a fixed status. Bearer-token acceptance
@@ -1104,6 +1104,25 @@ mod tests {
                     version: "9.9.9".to_string(),
                 }))
             }
+
+            // `Subscribe` streaming is exercised by `nunciod`'s own
+            // `grpc::tests` (backlog story 1.A.4 / GH-151); this stub only
+            // needs to satisfy the trait so the CLI's `GetStatus` happy
+            // path above can compile against the real `System` service
+            // definition, so it deliberately returns `unimplemented` rather
+            // than fabricating stream behavior no test here relies on.
+            type SubscribeStream = std::pin::Pin<
+                Box<dyn tokio_stream::Stream<Item = Result<Event, tonic::Status>> + Send + 'static>,
+            >;
+
+            async fn subscribe(
+                &self,
+                _request: tonic::Request<SubscribeRequest>,
+            ) -> Result<tonic::Response<Self::SubscribeStream>, tonic::Status> {
+                Err(tonic::Status::unimplemented(
+                    "subscribe is not exercised by this stub",
+                ))
+            }
         }
 
         let listener = tokio::net::TcpListener::bind("127.0.0.1:0")

crates/nuncio-proto/proto/nuncio/v1/nuncio.proto

@@ -9,9 +9,19 @@ package nuncio.v1;
 // RPCs will be added under the `nuncio.v1` package as later stories land;
 // breaking changes will be introduced under a new `nuncio.v2` package rather
 // than mutating this one.
+//
+// `Subscribe` is deliberately defined on THIS service (rather than a separate
+// one) so that it is automatically covered by the same server-side bearer
+// auth interceptor that guards `GetStatus`, instead of risking a second,
+// un-intercepted service being mounted by mistake (see GH #165).
 service System {
   // GetStatus reports basic daemon/engine health information.
   rpc GetStatus(GetStatusRequest) returns (GetStatusResponse);
+  // Subscribe opens a live, server-streaming push feed of daemon domain
+  // events (see `nuncio_core::CoreEvent`) for the lifetime of the call. The
+  // stream ends when the client disconnects or the daemon's event bus is
+  // shut down.
+  rpc Subscribe(SubscribeRequest) returns (stream Event);
 }
 
 // GetStatusRequest currently carries no parameters. Fields may be added
@@ -26,3 +36,96 @@ message GetStatusResponse {
   // The nunciod build/crate version serving this response.
   string version = 2;
 }
+
+// SubscribeRequest currently carries no parameters. Fields may be added
+// here in future stories (e.g. event-kind filtering).
+message SubscribeRequest {}
+
+// Event faithfully mirrors `nuncio_core::CoreEvent`: exactly one variant
+// message is populated per event, carrying that variant's fields. New
+// `CoreEvent` variants MUST be added here as a new oneof case (never by
+// repurposing an existing field number) to keep the wire contract stable.
+message Event {
+  oneof kind {
+    SyncStarted sync_started = 1;
+    SyncCompleted sync_completed = 2;
+    MessageFlagsChanged message_flags_changed = 3;
+    FilterExecuted filter_executed = 4;
+    BatchFilterProgress batch_filter_progress = 5;
+    DatabaseRecovered database_recovered = 6;
+    UpdateAvailable update_available = 7;
+    EventError error = 8;
+    ShuttingDown shutting_down = 9;
+  }
+}
+
+// Mirrors `CoreEvent::SyncStarted`.
+message SyncStarted {
+  // Account ID if scoped to a specific account; absent when synchronizing
+  // all accounts.
+  optional string account_id = 1;
+}
+
+// Mirrors `CoreEvent::SyncCompleted`.
+message SyncCompleted {
+  // Account ID if scoped to a specific account; absent when synchronizing
+  // all accounts.
+  optional string account_id = 1;
+}
+
+// Mirrors `CoreEvent::MessageFlagsChanged`.
+message MessageFlagsChanged {
+  // Message identifier.
+  string message_id = 1;
+  // New read state.
+  bool read = 2;
+}
+
+// Mirrors `CoreEvent::FilterExecuted`.
+message FilterExecuted {
+  // Filter rule identifier.
+  string rule_id = 1;
+  // Email message identifier.
+  string message_id = 2;
+  // Action description.
+  string action_taken = 3;
+}
+
+// Mirrors `CoreEvent::BatchFilterProgress`.
+message BatchFilterProgress {
+  // Messages processed so far in the batch.
+  uint64 processed = 1;
+  // Total message count in batch.
+  uint64 total = 2;
+  // Total matching rule executions.
+  uint64 matched = 3;
+}
+
+// Mirrors `CoreEvent::DatabaseRecovered`.
+message DatabaseRecovered {
+  // Forensic backup path.
+  string backup_path = 1;
+  // Number of filter rules salvaged from corrupted DB.
+  uint64 salvaged_rules_count = 2;
+  // Whether automated resync was triggered.
+  bool resync_triggered = 3;
+}
+
+// Mirrors `CoreEvent::UpdateAvailable`.
+message UpdateAvailable {
+  // Latest available release version (e.g. "0.2.0").
+  string version = 1;
+  // Release notes / changelog.
+  string release_notes = 2;
+}
+
+// Mirrors `CoreEvent::Error`. Named `EventError` (rather than `Error`) to
+// avoid colliding with the `Error` trait/derive macro name that is
+// ubiquitous in hand-written Rust using this generated module.
+message EventError {
+  // Error message text.
+  string message = 1;
+}
+
+// Mirrors `CoreEvent::ShuttingDown`, which carries no fields.
+message ShuttingDown {}

crates/nuncio-proto/src/client.rs

@@ -10,7 +10,9 @@
 //! server-side interceptor in `nunciod::grpc`.
 
 use crate::v1::system_client::SystemClient;
+use crate::v1::{Event, SubscribeRequest};
 use thiserror::Error;
+use tonic::codec::Streaming;
 use tonic::metadata::errors::InvalidMetadataValue;
 use tonic::metadata::{Ascii, MetadataValue};
 use tonic::service::interceptor::InterceptedService;
@@ -110,6 +112,24 @@ pub async fn connect_system(
     Ok(SystemClient::with_interceptor(channel, interceptor))
 }
 
+/// Opens the `nuncio.v1.System/Subscribe` server-streaming RPC on an already
+/// authenticated client (as returned by [`connect_system`]), reusing the
+/// bearer-token interceptor already attached to `client` rather than
+/// requiring callers to re-authenticate. Returns the live event stream;
+/// callers pull events with `Streaming::message` (or the `Stream`/`StreamExt`
+/// combinators) until the daemon closes the stream or the connection drops.
+///
+/// This is the client-side counterpart of `nunciod::grpc`'s `Subscribe`
+/// implementation (backlog story 1.A.4 / GH-151) and keeps thin
+/// presentation-shell clients free of any dependency on the `nunciod`
+/// binary crate itself.
+pub async fn subscribe_events(
+    client: &mut AuthenticatedSystemClient,
+) -> Result<Streaming<Event>, Status> {
+    let response = client.subscribe(SubscribeRequest {}).await?;
+    Ok(response.into_inner())
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

crates/nunciod/Cargo.toml

@@ -20,7 +20,7 @@ arc-swap.workspace = true
 
 
 tokio = { workspace = true, features = ["full"] }
-tokio-stream = { workspace = true, features = ["net"] }
+tokio-stream = { workspace = true, features = ["net", "sync"] }
 serde.workspace = true
 serde_json.workspace = true
 thiserror.workspace = true

crates/nunciod/src/grpc.rs

@@ -7,14 +7,22 @@
 //! (`nuncio_core::ipc::IpcDaemonServer`); migrating callers off the JSON-RPC
 //! transport is out of scope here and lands in a later story.
 
-use nuncio_core::EventBus;
+use nuncio_core::{CoreEvent, EventBus};
+use nuncio_proto::v1::event::Kind;
 use nuncio_proto::v1::system_server::{System, SystemServer};
-use nuncio_proto::v1::{GetStatusRequest, GetStatusResponse};
+use nuncio_proto::v1::{
+    BatchFilterProgress, DatabaseRecovered, Event, EventError, FilterExecuted, GetStatusRequest,
+    GetStatusResponse, MessageFlagsChanged, ShuttingDown, SubscribeRequest, SyncCompleted,
+    SyncStarted, UpdateAvailable,
+};
+use std::pin::Pin;
 use std::sync::Arc;
 use subtle::ConstantTimeEq;
 use thiserror::Error;
 use tokio::net::TcpListener;
-use tokio_stream::wrappers::TcpListenerStream;
+use tokio_stream::wrappers::errors::BroadcastStreamRecvError;
+use tokio_stream::wrappers::{BroadcastStream, TcpListenerStream};
+use tokio_stream::{Stream, StreamExt};
 use tonic::transport::Server;
 use tonic::{Request, Response, Status};
 
@@ -43,6 +51,64 @@ pub enum GrpcServeError {
     Transport(#[from] tonic::transport::Error),
 }
 
+/// Maps a `nuncio_core::CoreEvent` domain event onto its wire-format
+/// `nuncio.v1.Event` representation, faithfully carrying every variant's
+/// fields into the matching `oneof` case (see `proto/nuncio/v1/nuncio.proto`
+/// for the wire contract, backlog story 1.A.4 / GH-151).
+///
+/// `usize` fields (`processed`, `total`, `matched`, `salvaged_rules_count`)
+/// are narrowed to `u64` for the wire; these are in-process counters that
+/// cannot realistically approach `u64::MAX`, and protobuf has no native
+/// `usize` type.
+fn map_core_event(event: CoreEvent) -> Event {
+    let kind = match event {
+        CoreEvent::SyncStarted { account_id } => Kind::SyncStarted(SyncStarted { account_id }),
+        CoreEvent::SyncCompleted { account_id } => {
+            Kind::SyncCompleted(SyncCompleted { account_id })
+        }
+        CoreEvent::MessageFlagsChanged { message_id, read } => {
+            Kind::MessageFlagsChanged(MessageFlagsChanged { message_id, read })
+        }
+        CoreEvent::FilterExecuted {
+            rule_id,
+            message_id,
+            action_taken,
+        } => Kind::FilterExecuted(FilterExecuted {
+            rule_id,
+            message_id,
+            action_taken,
+        }),
+        CoreEvent::BatchFilterProgress {
+            processed,
+            total,
+            matched,
+        } => Kind::BatchFilterProgress(BatchFilterProgress {
+            processed: processed as u64,
+            total: total as u64,
+            matched: matched as u64,
+        }),
+        CoreEvent::DatabaseRecovered {
+            backup_path,
+            salvaged_rules_count,
+            resync_triggered,
+        } => Kind::DatabaseRecovered(DatabaseRecovered {
+            backup_path,
+            salvaged_rules_count: salvaged_rules_count as u64,
+            resync_triggered,
+        }),
+        CoreEvent::UpdateAvailable {
+            version,
+            release_notes,
+        } => Kind::UpdateAvailable(UpdateAvailable {
+            version,
+            release_notes,
+        }),
+        CoreEvent::Error { message } => Kind::Error(EventError { message }),
+        CoreEvent::ShuttingDown => Kind::ShuttingDown(ShuttingDown {}),
+    };
+    Event { kind: Some(kind) }
+}
+
 /// `nuncio.v1.System` gRPC service implementation backed by the daemon's live
 /// [`EventBus`] state.
 struct SystemGrpcService {
@@ -61,6 +127,47 @@ impl System for SystemGrpcService {
             version: env!("CARGO_PKG_VERSION").to_string(),
         }))
     }
+
+    /// Server-streaming push feed of daemon domain events (backlog story
+    /// 1.A.4 / GH-151). Defined on `System` (not a separate service) so this
+    /// RPC is automatically covered by the same `BearerAuthInterceptor` that
+    /// guards `GetStatus`, rather than risking a second, un-intercepted
+    /// service being mounted by mistake (see GH #165).
+    ///
+    /// The subscription is registered (`event_bus.subscribe_events()`)
+    /// synchronously before this method returns, so once a client's
+    /// `subscribe` call resolves, any event published afterwards is
+    /// guaranteed to be observed on the returned stream — no fixed sleep is
+    /// needed by callers to avoid a race between "start subscribing" and
+    /// "event published".
+    ///
+    /// A lagging subscriber (`BroadcastStreamRecvError::Lagged`) is handled
+    /// by logging a warning and skipping to the next available event rather
+    /// than terminating the stream or panicking; the daemon's event bus is
+    /// a best-effort broadcast feed, not a durable log, so a slow subscriber
+    /// missing some events is preferable to it never recovering. The stream
+    /// ends (returns `None`) only when the event bus itself is closed (all
+    /// `EventBus` senders dropped) or the client disconnects.
+    type SubscribeStream = Pin<Box<dyn Stream<Item = Result<Event, Status>> + Send + 'static>>;
+
+    async fn subscribe(
+        &self,
+        _request: Request<SubscribeRequest>,
+    ) -> Result<Response<Self::SubscribeStream>, Status> {
+        let receiver = self.event_bus.subscribe_events();
+        let mapped = BroadcastStream::new(receiver).filter_map(|item| match item {
+            Ok(core_event) => Some(Ok(map_core_event(core_event))),
+            Err(BroadcastStreamRecvError::Lagged(skipped)) => {
+                tracing::warn!(
+                    "gRPC Subscribe stream lagged behind the daemon event bus; skipped {} \
+                     event(s); resuming from the next available event",
+                    skipped
+                );
+                None
+            }
+        });
+        Ok(Response::new(Box::pin(mapped)))
+    }
 }
 
 /// Bearer-token authentication interceptor for the loopback `nuncio.v1` gRPC
@@ -310,4 +417,155 @@ mod tests {
             .expect_err("invalid bind address must fail");
         assert!(matches!(err, GrpcServeError::Bind { .. }));
     }
+
+    fn authed_bearer_request<T>(payload: T) -> Request<T> {
+        let mut request = Request::new(payload);
+        request.metadata_mut().insert(
+            "authorization",
+            "Bearer correct-token"
+                .parse()
+                .expect("valid ascii metadata value"),
+        );
+        request
+    }
+
+    #[tokio::test]
+    async fn subscribe_streams_mapped_core_events_from_the_live_event_bus() {
+        let event_bus = Arc::new(EventBus::new());
+        let (addr, _handle) = spawn_test_server(event_bus.clone(), "correct-token").await;
+
+        let mut client = SystemClient::connect(format!("http://{addr}"))
+            .await
+            .expect("client connects");
+
+        // `client.subscribe(..).await` only resolves once the server-side
+        // `subscribe` method has returned `Response::new(stream)`, and that
+        // method registers `event_bus.subscribe_events()` before returning.
+        // So once this `.await` completes, the subscription is guaranteed
+        // to already be live: publishing an event afterwards cannot race
+        // ahead of the subscriber registering, and no fixed sleep is
+        // needed to make this deterministic.
+        let mut stream = client
+            .subscribe(authed_bearer_request(SubscribeRequest {}))
+            .await
+            .expect("subscribe is accepted for a valid bearer token")
+            .into_inner();
+
+        event_bus.process_command(CoreCommand::SyncAccount {
+            account_id: "acct-42".to_string(),
+        });
+
+        let event = stream
+            .message()
+            .await
+            .expect("stream yields without a transport error")
+            .expect("stream produces an event rather than ending");
+
+        match event.kind {
+            Some(Kind::SyncStarted(SyncStarted { account_id })) => {
+                assert_eq!(account_id, Some("acct-42".to_string()));
+            }
+            other => panic!("expected a mapped SyncStarted event, got {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn subscribe_rejects_missing_bearer_token() {
+        // Confirms `Subscribe` inherits the same `BearerAuthInterceptor` as
+        // `GetStatus` because both are defined on the single `System`
+        // service (GH #165: no separate, un-intercepted service).
+        let event_bus = Arc::new(EventBus::new());
+        let (addr, _handle) = spawn_test_server(event_bus, "correct-token").await;
+
+        let mut client = SystemClient::connect(format!("http://{addr}"))
+            .await
+            .expect("client connects");
+        let err = client
+            .subscribe(SubscribeRequest {})
+            .await
+            .expect_err("missing bearer token must be rejected");
+        assert_eq!(err.code(), Code::Unauthenticated);
+    }
+
+    #[tokio::test]
+    async fn concurrent_unary_and_streaming_calls_do_not_interleave_or_corrupt() {
+        // Regression test for the property the old hand-rolled JSON-RPC IPC
+        // transport lacked: a response/notification demux race that could
+        // corrupt reads when a streamed notification and a unary response
+        // were in flight at the same time. gRPC/HTTP2 multiplexes streaming
+        // and unary calls on independent logical streams, so this asserts
+        // both an active event subscription and many concurrent unary
+        // `GetStatus` calls complete correctly and without corrupting one
+        // another.
+        let event_bus = Arc::new(EventBus::new());
+        let (addr, _handle) = spawn_test_server(event_bus.clone(), "correct-token").await;
+
+        // Open (and await) the subscription first, on its own connection,
+        // so the server-side registration has deterministically happened
+        // before any event below is published (see the comment in the
+        // single-event test above for why this ordering is race-free).
+        let mut sub_client = SystemClient::connect(format!("http://{addr}"))
+            .await
+            .expect("subscribe client connects");
+        let mut stream = sub_client
+            .subscribe(authed_bearer_request(SubscribeRequest {}))
+            .await
+            .expect("subscribe is accepted for a valid bearer token")
+            .into_inner();
+
+        const ROUNDS: usize = 20;
+        let mut unary_client = SystemClient::connect(format!("http://{addr}"))
+            .await
+            .expect("unary client connects");
+        let unary_task = tokio::spawn(async move {
+            let mut responses = Vec::with_capacity(ROUNDS);
+            for _ in 0..ROUNDS {
+                let response = unary_client
+                    .get_status(authed_bearer_request(GetStatusRequest {}))
+                    .await
+                    .expect("unary call succeeds")
+                    .into_inner();
+                responses.push(response);
+            }
+            responses
+        });
+
+        for i in 0..ROUNDS {
+            event_bus.process_command(CoreCommand::SyncAccount {
+                account_id: format!("acct-{i}"),
+            });
+        }
+
+        let unary_responses = unary_task.await.expect("unary task does not panic");
+        assert_eq!(unary_responses.len(), ROUNDS);
+        for response in &unary_responses {
+            // Every unary response must be a well-formed, uncorrupted
+            // `GetStatusResponse`: the correct build version, and a status
+            // string matching a real `EngineStatus` variant -- never bytes
+            // bled in from the concurrently-running event stream.
+            assert_eq!(response.version, env!("CARGO_PKG_VERSION"));
+            assert!(
+                matches!(
+                    response.engine_status.as_str(),
+                    "Idle" | "Syncing" | "ShuttingDown"
+                ),
+                "unexpected/corrupted engine_status: {:?}",
+                response.engine_status
+            );
+        }
+
+        for i in 0..ROUNDS {
+            let event = stream
+                .message()
+                .await
+                .expect("stream yields without a transport error")
+                .expect("stream produces an event rather than ending early");
+            match event.kind {
+                Some(Kind::SyncStarted(SyncStarted { account_id })) => {
+                    assert_eq!(account_id, Some(format!("acct-{i}")));
+                }
+                other => panic!("expected mapped SyncStarted(acct-{i}) event, got {other:?}"),
+            }
+        }
+    }
 }