first pass at platform agnostic notifications

* wip/local_ids:
  first attempt at trying to keep track of locally send id

README.md

@@ -64,16 +64,3 @@ Below are brief notes. Each subproject’s README has more detail.
 - Android: open `android/` in Android Studio and build. See `android/README.md` for configuration.
 - Mock server (Go): `cd mock && go run ./...` or `make`.
 
-## Security and Entitlements
-
-The macOS server uses private APIs and restricted entitlements. On production macOS builds, processes with restricted entitlements can be killed by the kernel; development requires workarounds (e.g., swizzling, hooking `imagent`) and careful code signing. See `server/README.md` for instructions and caveats.
-
-## Status
-
-- Android client: ships its own API client (not yet using Rust `core`).
-- GTK + macOS clients: use the Rust `core` library and integrate with the `kordophoned` client daemon for caching/IPC.
-- Mock server: useful for development; implements common endpoints and WebSocket updates.
-
-## Contributing
-
-Issues and PRs are welcome. If you add a new client or endpoint, please update relevant READMEs and link it from this root README. Prefer small, focused changes and keep style consistent with the existing code.

core/kordophone-db/migrations/2025-09-12-000001_add_message_aliases/down.sql

@@ -0,0 +1,3 @@
+-- Drop the alias mapping table
+DROP TABLE IF EXISTS `message_aliases`;
+

core/kordophone-db/migrations/2025-09-12-000001_add_message_aliases/up.sql

@@ -0,0 +1,7 @@
+-- Add table to map local (client) IDs to server message GUIDs
+CREATE TABLE IF NOT EXISTS `message_aliases` (
+    `local_id` TEXT NOT NULL PRIMARY KEY,
+    `server_id` TEXT NOT NULL UNIQUE,
+    `conversation_id` TEXT NOT NULL
+);
+

core/kordophone-db/src/repository.rs

@@ -307,8 +307,11 @@ impl<'a> Repository<'a> {
     }
 
     pub fn delete_all_messages(&mut self) -> Result<()> {
-        use crate::schema::messages::dsl::*;
-        diesel::delete(messages).execute(self.connection)?;
+        use crate::schema::message_aliases::dsl as aliases_dsl;
+        use crate::schema::messages::dsl as messages_dsl;
+
+        diesel::delete(messages_dsl::messages).execute(self.connection)?;
+        diesel::delete(aliases_dsl::message_aliases).execute(self.connection)?;
         Ok(())
     }
 
@@ -359,6 +362,57 @@ impl<'a> Repository<'a> {
         )
     }
 
+    /// Create or update an alias mapping between a local (client) message id and a server message id.
+    pub fn set_message_alias(
+        &mut self,
+        local_id_in: &str,
+        server_id_in: &str,
+        conversation_id_in: &str,
+    ) -> Result<()> {
+        use crate::schema::message_aliases::dsl::*;
+        diesel::replace_into(message_aliases)
+            .values((
+                local_id.eq(local_id_in),
+                server_id.eq(server_id_in),
+                conversation_id.eq(conversation_id_in),
+            ))
+            .execute(self.connection)?;
+        Ok(())
+    }
+
+    /// Returns the local id for a given server id, if any.
+    pub fn get_local_id_for(&mut self, server_id_in: &str) -> Result<Option<String>> {
+        use crate::schema::message_aliases::dsl::*;
+        let result = message_aliases
+            .filter(server_id.eq(server_id_in))
+            .select(local_id)
+            .first::<String>(self.connection)
+            .optional()?;
+        Ok(result)
+    }
+
+    /// Batch lookup: returns a map server_id -> local_id for the provided server ids.
+    pub fn get_local_ids_for(
+        &mut self,
+        server_ids_in: Vec<String>,
+    ) -> Result<HashMap<String, String>> {
+        use crate::schema::message_aliases::dsl::*;
+        if server_ids_in.is_empty() {
+            return Ok(HashMap::new());
+        }
+
+        let rows: Vec<(String, String)> = message_aliases
+            .filter(server_id.eq_any(&server_ids_in))
+            .select((server_id, local_id))
+            .load::<(String, String)>(self.connection)?;
+
+        let mut map = HashMap::new();
+        for (sid, lid) in rows {
+            map.insert(sid, lid);
+        }
+        Ok(map)
+    }
+
     /// Update the contact_id for an existing participant record.
     pub fn update_participant_contact(
         &mut self,

core/kordophone-db/src/schema.rs

@@ -44,6 +44,14 @@ diesel::table! {
     }
 }
 
+diesel::table! {
+    message_aliases (local_id) {
+        local_id -> Text,
+        server_id -> Text,
+        conversation_id -> Text,
+    }
+}
+
 diesel::table! {
     settings (key) {
         key -> Text,
@@ -62,5 +70,6 @@ diesel::allow_tables_to_appear_in_same_query!(
     conversation_participants,
     messages,
     conversation_messages,
+    message_aliases,
     settings,
 );

core/kordophone/src/api/http_client.rs

@@ -394,65 +394,73 @@ impl<K: AuthenticationStore + Send + Sync> APIInterface for HTTPAPIClient<K> {
             None => "updates".to_string(),
         };
 
-        let uri = self
-            .uri_for_endpoint(&endpoint, Some(self.websocket_scheme()))?;
+        let uri = self.uri_for_endpoint(&endpoint, Some(self.websocket_scheme()))?;
 
-        log::debug!("Connecting to websocket: {:?}", uri);
+        loop {
+            log::debug!("Connecting to websocket: {:?}", uri);
 
-        let auth = self.auth_store.get_token().await;
-        let host = uri.authority().unwrap().host();
-        let mut request = TungsteniteRequest::builder()
-            .header("Host", host)
-            .header("Connection", "Upgrade")
-            .header("Upgrade", "websocket")
-            .header("Sec-WebSocket-Version", "13")
-            .header("Sec-WebSocket-Key", generate_key())
-            .uri(uri.to_string())
-            .body(())
-            .expect("Unable to build websocket request");
+            let auth = self.auth_store.get_token().await;
+            let host = uri.authority().unwrap().host();
+            let mut request = TungsteniteRequest::builder()
+                .header("Host", host)
+                .header("Connection", "Upgrade")
+                .header("Upgrade", "websocket")
+                .header("Sec-WebSocket-Version", "13")
+                .header("Sec-WebSocket-Key", generate_key())
+                .uri(uri.to_string())
+                .body(())
+                .expect("Unable to build websocket request");
 
-        match &auth {
-            Some(token) => {
-                request.headers_mut().insert(
-                    "Authorization",
-                    format!("Bearer: {}", token).parse().unwrap(),
-                );
+            match &auth {
+                Some(token) => {
+                    request.headers_mut().insert(
+                        "Authorization",
+                        format!("Bearer: {}", token).parse().unwrap(),
+                    );
+                }
+                None => {
+                    log::warn!(target: "websocket", "Proceeding without auth token.");
+                }
             }
-            None => {
-                log::warn!(target: "websocket", "Proceeding without auth token.");
-            }
-        }
 
-        log::debug!("Websocket request: {:?}", request);
+            log::debug!("Websocket request: {:?}", request);
 
-        match connect_async(request).await.map_err(Error::from) {
-            Ok((socket, response)) => {
-                log::debug!("Websocket connected: {:?}", response.status());
-                Ok(WebsocketEventSocket::new(socket))
-            }
-            Err(e) => match &e {
-                Error::ClientError(ce) => match ce.as_str() {
-                    "HTTP error: 401 Unauthorized" | "Unauthorized" => {
-                        // Try to authenticate
-                        if let Some(credentials) = &self.auth_store.get_credentials().await {
-                            log::warn!("Websocket connection failed, attempting to authenticate");
-                            let new_token = self.authenticate(credentials.clone()).await?;
-                            self.auth_store.set_token(new_token.to_string()).await;
+            let mut should_retry = true; // retry once after authenticating.
+            match connect_async(request).await.map_err(Error::from) {
+                Ok((socket, response)) => {
+                    log::debug!("Websocket connected: {:?}", response.status());
+                    break Ok(WebsocketEventSocket::new(socket));
+                }
+                Err(e) => match &e {
+                    Error::ClientError(ce) => match ce.as_str() {
+                        "HTTP error: 401 Unauthorized" | "Unauthorized" => {
+                            // Try to authenticate
+                            if let Some(credentials) = &self.auth_store.get_credentials().await {
+                                log::warn!(
+                                    "Websocket connection failed, attempting to authenticate"
+                                );
+                                let new_token = self.authenticate(credentials.clone()).await?;
+                                self.auth_store.set_token(new_token.to_string()).await;
 
-                            // try again on the next attempt.
-                            return Err(Error::Unauthorized);
-                        } else {
-                            log::error!("Websocket unauthorized, no credentials provided");
-                            return Err(Error::ClientError(
-                                "Unauthorized, no credentials provided".into(),
-                            ));
+                                if should_retry {
+                                    // try again on the next attempt.
+                                    continue;
+                                } else {
+                                    break Err(e);
+                                }
+                            } else {
+                                log::error!("Websocket unauthorized, no credentials provided");
+                                break Err(Error::ClientError(
+                                    "Unauthorized, no credentials provided".into(),
+                                ));
+                            }
                         }
-                    }
-                    _ => Err(e),
-                },
+                        _ => break Err(e),
+                    },
 
-                _ => Err(e),
-            },
+                    _ => break Err(e),
+                },
+            }
         }
     }
 }
@@ -462,16 +470,16 @@ impl<K: AuthenticationStore + Send + Sync> HTTPAPIClient<K> {
         let https = HttpsConnector::new();
         let client = Client::builder().build::<_, Body>(https);
 
-        HTTPAPIClient { base_url, auth_store, client }
+        HTTPAPIClient {
+            base_url,
+            auth_store,
+            client,
+        }
     }
 
     fn uri_for_endpoint(&self, endpoint: &str, scheme: Option<&str>) -> Result<Uri, Error> {
         let mut parts = self.base_url.clone().into_parts();
-        let root_path: PathBuf = parts
-            .path_and_query
-            .ok_or(Error::URLError)?
-            .path()
-            .into();
+        let root_path: PathBuf = parts.path_and_query.ok_or(Error::URLError)?.path().into();
 
         let path = root_path.join(endpoint);
         let path_str = path.to_str().ok_or(Error::URLError)?;
@@ -618,6 +626,27 @@ impl<K: AuthenticationStore + Send + Sync> HTTPAPIClient<K> {
 
         Ok(response)
     }
+
+    // Fetch an attachment while preserving response headers (e.g., Content-Type).
+    // Returns the streaming body and the Content-Type header if present.
+    pub async fn fetch_attachment_with_metadata(
+        &mut self,
+        guid: &str,
+        preview: bool,
+    ) -> Result<(ResponseStream, Option<String>), Error> {
+        let endpoint = format!("attachment?guid={}&preview={}", guid, preview);
+        let response = self
+            .response_with_body_retry(&endpoint, Method::GET, Body::empty, true)
+            .await?;
+
+        let content_type = response
+            .headers()
+            .get(hyper::header::CONTENT_TYPE)
+            .and_then(|v| v.to_str().ok())
+            .map(|s| s.to_string());
+
+        Ok((ResponseStream::from(response.into_body()), content_type))
+    }
 }
 
 #[cfg(test)]

core/kordophoned/Cargo.toml

@@ -22,6 +22,8 @@ tokio = { version = "1", features = ["full"] }
 tokio-condvar = "0.3.0"
 uuid = "1.16.0"
 once_cell = "1.19.0"
+mime_guess = "2.0"
+notify = { package = "notify-rust", version = "4.10.0" }
 
 # D-Bus dependencies only on Linux
 [target.'cfg(target_os = "linux")'.dependencies]
@@ -49,4 +51,3 @@ assets = [
     { source = "../target/release/kpcli", dest = "/usr/bin/kpcli", mode = "755" },
     { source = "include/net.buzzert.kordophonecd.service", dest = "/usr/share/dbus-1/services/net.buzzert.kordophonecd.service", mode = "644" },
 ]
-

core/kordophoned/include/net.buzzert.kordophonecd.Server.xml

@@ -103,6 +103,13 @@
                         "/>
         </method>
 
+        <method name="TestNotification">
+            <arg type="s" name="summary" direction="in"/>
+            <arg type="s" name="body" direction="in"/>
+            <annotation name="org.freedesktop.DBus.DocString"
+                        value="Displays a test desktop notification with the provided summary and body."/>
+        </method>
+
         <signal name="MessagesUpdated">
             <arg type="s" name="conversation_id" direction="in"/>
             <annotation name="org.freedesktop.DBus.DocString"

core/kordophoned/src/daemon/attachment_store.rs

@@ -17,7 +17,7 @@ use crate::daemon::Daemon;
 
 use std::sync::Arc;
 use tokio::sync::mpsc::{Receiver, Sender};
-use tokio::sync::Mutex;
+use tokio::sync::{Mutex, Semaphore};
 
 use uuid::Uuid;
 
@@ -63,6 +63,9 @@ pub struct AttachmentStore {
 
     event_source: Receiver<AttachmentStoreEvent>,
     event_sink: Option<Sender<AttachmentStoreEvent>>,
+
+    // Limits concurrent downloads to avoid overloading server and local I/O
+    download_limit: Arc<Semaphore>,
 }
 
 impl AttachmentStore {
@@ -84,12 +87,16 @@ impl AttachmentStore {
 
         let (event_sink, event_source) = tokio::sync::mpsc::channel(100);
 
+        // Limit to at most 5 concurrent downloads by default
+        let download_limit = Arc::new(Semaphore::new(5));
+
         AttachmentStore {
             store_path: store_path,
             database: database,
             daemon_event_sink: daemon_event_sink,
             event_source: event_source,
             event_sink: Some(event_sink),
+            download_limit,
         }
     }
 
@@ -103,11 +110,49 @@ impl AttachmentStore {
 
     pub fn get_attachment_impl(store_path: &PathBuf, guid: &String) -> Attachment {
         let base_path = store_path.join(guid);
-        Attachment {
+        let mut attachment = Attachment {
             guid: guid.to_owned(),
             base_path: base_path,
             metadata: None,
+            mime_type: None,
+        };
+
+        // Best-effort: if a file already exists, try to infer MIME type from extension
+        let kind = "full";
+        let stem = attachment
+            .base_path
+            .file_name()
+            .map(|s| s.to_string_lossy().to_string())
+            .unwrap_or_default();
+        let legacy = attachment.base_path.with_extension(kind);
+        let existing_path = if legacy.exists() {
+            Some(legacy)
+        } else {
+            let prefix = format!("{}.{}.", stem, kind);
+            let parent = attachment
+                .base_path
+                .parent()
+                .unwrap_or_else(|| std::path::Path::new("."));
+            let mut found: Option<PathBuf> = None;
+            if let Ok(entries) = std::fs::read_dir(parent) {
+                for entry in entries.flatten() {
+                    let name = entry.file_name().to_string_lossy().to_string();
+                    if name.starts_with(&prefix) && !name.ends_with(".download") {
+                        found = Some(entry.path());
+                        break;
+                    }
+                }
+            }
+            found
+        };
+
+        if let Some(existing) = existing_path {
+            if let Some(m) = mime_guess::from_path(&existing).first_raw() {
+                attachment.mime_type = Some(m.to_string());
+            }
         }
+
+        attachment
     }
 
     async fn download_attachment_impl(
@@ -124,22 +169,45 @@ impl AttachmentStore {
             return Err(AttachmentStoreError::AttachmentAlreadyDownloaded.into());
         }
 
-        let temporary_path = attachment.get_path_for_preview_scratch(preview, true);
-        if std::fs::exists(&temporary_path).unwrap_or(false) {
-            log::warn!(target: target::ATTACHMENTS, "Temporary file already exists: {}, assuming download is in progress", temporary_path.display());
+        // Check if any in-progress temporary file exists already for this attachment
+        if let Some(in_progress) = Self::find_in_progress_download(&attachment, preview) {
+            log::warn!(target: target::ATTACHMENTS, "Temporary file already exists: {}, assuming download is in progress", in_progress.display());
+            // Treat as a non-fatal condition so we don't spam errors.
             return Err(AttachmentStoreError::DownloadAlreadyInProgress.into());
         }
 
         log::debug!(target: target::ATTACHMENTS, "Starting download for attachment: {}", attachment.guid);
 
-        let file = std::fs::File::create(&temporary_path)?;
-        let mut writer = BufWriter::new(&file);
         let mut client = Daemon::get_client_impl(database).await?;
-        let mut stream = client
-            .fetch_attachment_data(&attachment.guid, preview)
+        let (mut stream, content_type) = client
+            .fetch_attachment_with_metadata(&attachment.guid, preview)
             .await
             .map_err(|e| AttachmentStoreError::APIClientError(format!("{:?}", e)))?;
 
+        let kind = if preview { "preview" } else { "full" };
+        let normalized_mime = content_type
+            .as_deref()
+            .map(|s| s.split(';').next().unwrap_or(s).trim().to_string());
+
+        let guessed_ext = normalized_mime
+            .as_deref()
+            .and_then(|m| Attachment::preferred_extension_for_mime(m))
+            .unwrap_or("bin");
+
+        let final_path = attachment
+            .base_path
+            .with_extension(format!("{}.{}", kind, guessed_ext));
+        let temporary_path = final_path.with_extension(format!(
+            "{}.download",
+            final_path
+                .extension()
+                .and_then(|e| e.to_str())
+                .unwrap_or("")
+        ));
+
+        let file = std::fs::File::create(&temporary_path)?;
+        let mut writer = BufWriter::new(&file);
+
         log::trace!(target: target::ATTACHMENTS, "Writing attachment {:?} data to temporary file {:?}", &attachment.guid, &temporary_path);
         while let Some(Ok(data)) = stream.next().await {
             writer.write(data.as_ref())?;
@@ -150,10 +218,7 @@ impl AttachmentStore {
         file.sync_all()?;
 
         // Atomically move the temporary file to the final location
-        std::fs::rename(
-            &temporary_path,
-            &attachment.get_path_for_preview_scratch(preview, false),
-        )?;
+        std::fs::rename(&temporary_path, &final_path)?;
 
         log::debug!(target: target::ATTACHMENTS, "Completed download for attachment: {}", attachment.guid);
 
@@ -164,6 +229,41 @@ impl AttachmentStore {
         Ok(())
     }
 
+    fn find_in_progress_download(attachment: &Attachment, preview: bool) -> Option<PathBuf> {
+        let kind = if preview { "preview" } else { "full" };
+
+        // Legacy temp path: guid.<kind>.download
+        let legacy = attachment
+            .base_path
+            .with_extension(format!("{}.download", kind));
+        if legacy.exists() {
+            return Some(legacy);
+        }
+
+        // Scan for any guid.<kind>.<ext>.download
+        let parent = attachment
+            .base_path
+            .parent()
+            .unwrap_or_else(|| std::path::Path::new("."));
+        let stem = attachment
+            .base_path
+            .file_name()
+            .map(|s| s.to_string_lossy().to_string())
+            .unwrap_or_default();
+        let prefix = format!("{}.{}.", stem, kind);
+
+        if let Ok(entries) = std::fs::read_dir(parent) {
+            for entry in entries.flatten() {
+                let name = entry.file_name().to_string_lossy().to_string();
+                if name.starts_with(&prefix) && name.ends_with(".download") {
+                    return Some(entry.path());
+                }
+            }
+        }
+
+        None
+    }
+
     async fn upload_attachment_impl(
         store_path: &PathBuf,
         incoming_path: &PathBuf,
@@ -222,9 +322,12 @@ impl AttachmentStore {
                                 let mut database = self.database.clone();
                                 let daemon_event_sink = self.daemon_event_sink.clone();
                                 let _guid = guid.clone();
+                                let limiter = self.download_limit.clone();
 
                                 // Spawn a new task here so we don't block incoming queue events.
                                 tokio::spawn(async move {
+                                    // Acquire a slot in the concurrent download limiter.
+                                    let _permit = limiter.acquire_owned().await.expect("Semaphore closed");
                                     let result = Self::download_attachment_impl(
                                         &store_path,
                                         &mut database,
@@ -234,7 +337,23 @@ impl AttachmentStore {
                                     ).await;
 
                                     if let Err(e) = result {
-                                        log::error!(target: target::ATTACHMENTS, "Error downloading attachment {}: {}", &_guid, e);
+                                        // Downgrade noise for expected cases
+                                        if let Some(kind) = e.downcast_ref::<AttachmentStoreError>() {
+                                            match kind {
+                                                AttachmentStoreError::AttachmentAlreadyDownloaded => {
+                                                    log::debug!(target: target::ATTACHMENTS, "Attachment already downloaded: {}", &_guid);
+                                                }
+                                                AttachmentStoreError::DownloadAlreadyInProgress => {
+                                                    // Already logged a warning where detected
+                                                    log::debug!(target: target::ATTACHMENTS, "Download already in progress: {}", &_guid);
+                                                }
+                                                _ => {
+                                                    log::error!(target: target::ATTACHMENTS, "Error downloading attachment {}: {}", &_guid, kind);
+                                                }
+                                            }
+                                        } else {
+                                            log::error!(target: target::ATTACHMENTS, "Error downloading attachment {}: {}", &_guid, e);
+                                        }
                                     }
                                 });
 

core/kordophoned/src/daemon/events.rs

@@ -61,6 +61,9 @@ pub enum Event {
     /// - reply: The outgoing message ID (not the server-assigned message ID).
     SendMessage(String, String, Vec<String>, Reply<Uuid>),
 
+    /// Triggers a manual test notification.
+    TestNotification(String, String, Reply<Result<(), String>>),
+
     /// Notifies the daemon that a message has been sent.
     /// Parameters:
     /// - message: The message that was sent.

core/kordophoned/src/daemon/mod.rs

@@ -17,8 +17,11 @@ use std::path::PathBuf;
 use std::sync::Arc;
 
 use thiserror::Error;
-use tokio::sync::mpsc::{Receiver, Sender};
-use tokio::sync::Mutex;
+use tokio::sync::{
+    broadcast,
+    mpsc::{Receiver, Sender},
+    Mutex,
+};
 use uuid::Uuid;
 
 use kordophone_db::{
@@ -41,6 +44,9 @@ mod post_office;
 use post_office::Event as PostOfficeEvent;
 use post_office::PostOffice;
 
+mod notifier;
+use notifier::NotificationService;
+
 mod models;
 pub use models::Attachment;
 pub use models::Message;
@@ -76,8 +82,7 @@ pub struct Daemon {
     pub event_sender: Sender<Event>,
     event_receiver: Receiver<Event>,
 
-    signal_receiver: Option<Receiver<Signal>>,
-    signal_sender: Sender<Signal>,
+    signal_sender: broadcast::Sender<Signal>,
 
     post_office_sink: Sender<PostOfficeEvent>,
     post_office_source: Option<Receiver<PostOfficeEvent>>,
@@ -87,6 +92,7 @@ pub struct Daemon {
     attachment_store_sink: Option<Sender<AttachmentStoreEvent>>,
     update_monitor_command_tx: Option<Sender<UpdateMonitorCommand>>,
 
+    notifier: Arc<NotificationService>,
     version: String,
     database: Arc<Mutex<Database>>,
     runtime: tokio::runtime::Runtime,
@@ -103,7 +109,7 @@ impl Daemon {
 
         // Create event channels
         let (event_sender, event_receiver) = tokio::sync::mpsc::channel(100);
-        let (signal_sender, signal_receiver) = tokio::sync::mpsc::channel(100);
+        let (signal_sender, _) = tokio::sync::broadcast::channel(100);
         let (post_office_sink, post_office_source) = tokio::sync::mpsc::channel(100);
 
         // Create background task runtime
@@ -114,13 +120,14 @@ impl Daemon {
 
         let database_impl = Database::new(&database_path.to_string_lossy())?;
         let database = Arc::new(Mutex::new(database_impl));
+        let notifier = Arc::new(NotificationService::new());
 
         Ok(Self {
             version: env!("CARGO_PKG_VERSION").to_string(),
+            notifier,
             database,
             event_receiver,
             event_sender,
-            signal_receiver: Some(signal_receiver),
             signal_sender,
             post_office_sink,
             post_office_source: Some(post_office_source),
@@ -165,6 +172,16 @@ impl Daemon {
             attachment_store.run().await;
         });
 
+        // Notification listener
+        {
+            let notifier = self.notifier.clone();
+            let mut signal_rx = self.signal_sender.subscribe();
+            let database = self.database.clone();
+            tokio::spawn(async move {
+                notifier.listen(signal_rx, database).await;
+            });
+        }
+
         while let Some(event) = self.event_receiver.recv().await {
             log::debug!(target: target::EVENT, "Received event: {:?}", event);
             self.handle_event(event).await;
@@ -260,10 +277,11 @@ impl Daemon {
                 self.spawn_conversation_list_sync();
 
                 // Send signal to the client that the update stream has been reconnected.
-                self.signal_sender
-                    .send(Signal::UpdateStreamReconnected)
-                    .await
-                    .unwrap();
+                Self::send_signal(
+                    &self.signal_sender,
+                    Signal::UpdateStreamReconnected,
+                    target::UPDATES,
+                );
             }
 
             Event::GetAllConversations(limit, offset, reply) => {
@@ -326,17 +344,13 @@ impl Daemon {
             }
 
             Event::SendMessage(conversation_id, text, attachment_guids, reply) => {
-                let conversation_id = conversation_id.clone();
                 let uuid = self
                     .enqueue_outgoing_message(text, conversation_id.clone(), attachment_guids)
                     .await;
                 reply.send(uuid).unwrap();
 
-                // Send message updated signal, we have a placeholder message we will return.
-                self.signal_sender
-                    .send(Signal::MessagesUpdated(conversation_id.clone()))
-                    .await
-                    .unwrap();
+                // Notify clients that messages have changed (e.g., to refresh placeholders).
+                self.emit_messages_updated(conversation_id);
             }
 
             Event::MessageSent(message, outgoing_message, conversation_id) => {
@@ -347,7 +361,16 @@ impl Daemon {
                 self.database
                     .lock()
                     .await
-                    .with_repository(|r| r.insert_message(&conversation_id, message.into()))
+                    .with_repository(|r| {
+                        // 1) Insert the server message
+                        r.insert_message(&conversation_id, message.clone().into())?;
+                        // 2) Persist alias local -> server for stable UI ids
+                        r.set_message_alias(
+                            &outgoing_message.guid.to_string(),
+                            &message.id,
+                            &conversation_id,
+                        )
+                    })
                     .await
                     .unwrap();
 
@@ -357,11 +380,20 @@ impl Daemon {
                     .get_mut(&conversation_id)
                     .map(|messages| messages.retain(|m| m.guid != outgoing_message.guid));
 
-                // Send message updated signal.
-                self.signal_sender
-                    .send(Signal::MessagesUpdated(conversation_id))
-                    .await
-                    .unwrap();
+                // Notify clients to refresh the conversation after the final message arrives.
+                self.emit_messages_updated(conversation_id);
+            }
+
+            Event::TestNotification(summary, body, reply) => {
+                let result = self
+                    .signal_sender
+                    .send(Signal::Internal(InternalSignal::TestNotification {
+                        summary,
+                        body,
+                    }))
+                    .map(|_| ())
+                    .map_err(|e| e.to_string());
+                reply.send(result).unwrap();
             }
 
             Event::GetAttachment(guid, reply) => {
@@ -393,10 +425,11 @@ impl Daemon {
                 log::debug!(target: target::ATTACHMENTS, "Daemon: attachment downloaded: {}, sending signal", attachment_id);
 
                 // Send signal to the client that the attachment has been downloaded.
-                self.signal_sender
-                    .send(Signal::AttachmentDownloaded(attachment_id))
-                    .await
-                    .unwrap();
+                Self::send_signal(
+                    &self.signal_sender,
+                    Signal::AttachmentDownloaded(attachment_id),
+                    target::ATTACHMENTS,
+                );
             }
 
             Event::UploadAttachment(path, reply) => {
@@ -411,17 +444,17 @@ impl Daemon {
             Event::AttachmentUploaded(upload_guid, attachment_guid) => {
                 log::info!(target: target::ATTACHMENTS, "Daemon: attachment uploaded: {}, {}", upload_guid, attachment_guid);
 
-                self.signal_sender
-                    .send(Signal::AttachmentUploaded(upload_guid, attachment_guid))
-                    .await
-                    .unwrap();
+                Self::send_signal(
+                    &self.signal_sender,
+                    Signal::AttachmentUploaded(upload_guid, attachment_guid),
+                    target::ATTACHMENTS,
+                );
             }
         }
     }
 
-    /// Panics if the signal receiver has already been taken.
-    pub fn obtain_signal_receiver(&mut self) -> Receiver<Signal> {
-        self.signal_receiver.take().unwrap()
+    pub fn subscribe_signals(&self) -> broadcast::Receiver<Signal> {
+        self.signal_sender.subscribe()
     }
 
     async fn get_conversations_limit_offset(
@@ -436,6 +469,10 @@ impl Daemon {
             .await
     }
 
+    fn emit_messages_updated(&self, conversation_id: String) {
+        Self::send_messages_updated(&self.signal_sender, conversation_id);
+    }
+
     async fn get_messages(
         &mut self,
         conversation_id: String,
@@ -448,18 +485,37 @@ impl Daemon {
             .get(&conversation_id)
             .unwrap_or(&empty_vec);
 
-        self.database
+        // Fetch DB messages and an alias map (server_id -> local_id) in one DB access.
+        let (db_messages, alias_map) = self
+            .database
             .lock()
             .await
             .with_repository(|r| {
-                r.get_messages_for_conversation(&conversation_id)
-                    .unwrap()
-                    .into_iter()
-                    .map(|m| m.into()) // Convert db::Message to daemon::Message
-                    .chain(outgoing_messages.into_iter().map(|m| m.into()))
-                    .collect()
+                let msgs = r.get_messages_for_conversation(&conversation_id).unwrap();
+                let ids: Vec<String> = msgs.iter().map(|m| m.id.clone()).collect();
+                let map = r.get_local_ids_for(ids).unwrap_or_default();
+                (msgs, map)
             })
-            .await
+            .await;
+
+        // Convert DB messages to daemon model, substituting local_id when an alias exists.
+        let mut result: Vec<Message> =
+            Vec::with_capacity(db_messages.len() + outgoing_messages.len());
+        for m in db_messages.into_iter() {
+            let server_id = m.id.clone();
+            let mut dm: Message = m.into();
+            if let Some(local_id) = alias_map.get(&server_id) {
+                dm.id = local_id.clone();
+            }
+            result.push(dm);
+        }
+
+        // Append pending outgoing messages (these already use local_id)
+        for om in outgoing_messages.iter() {
+            result.push(om.into());
+        }
+
+        result
     }
 
     async fn enqueue_outgoing_message(
@@ -492,7 +548,7 @@ impl Daemon {
 
     async fn sync_conversation_list(
         database: &mut Arc<Mutex<Database>>,
-        signal_sender: &Sender<Signal>,
+        signal_sender: &broadcast::Sender<Signal>,
     ) -> Result<()> {
         log::info!(target: target::SYNC, "Starting list conversation sync");
 
@@ -544,7 +600,7 @@ impl Daemon {
         }
 
         // Send conversations updated signal
-        signal_sender.send(Signal::ConversationsUpdated).await?;
+        Self::send_signal(signal_sender, Signal::ConversationsUpdated, target::SYNC);
 
         log::info!(target: target::SYNC, "List synchronized: {} conversations", num_conversations);
         Ok(())
@@ -552,7 +608,7 @@ impl Daemon {
 
     async fn sync_all_conversations_impl(
         database: &mut Arc<Mutex<Database>>,
-        signal_sender: &Sender<Signal>,
+        signal_sender: &broadcast::Sender<Signal>,
     ) -> Result<()> {
         log::info!(target: target::SYNC, "Starting full conversation sync");
 
@@ -580,7 +636,7 @@ impl Daemon {
         }
 
         // Send conversations updated signal.
-        signal_sender.send(Signal::ConversationsUpdated).await?;
+        Self::send_signal(signal_sender, Signal::ConversationsUpdated, target::SYNC);
 
         log::info!(target: target::SYNC, "Full sync complete, {} conversations processed", num_conversations);
         Ok(())
@@ -588,7 +644,7 @@ impl Daemon {
 
     async fn sync_conversation_impl(
         database: &mut Arc<Mutex<Database>>,
-        signal_sender: &Sender<Signal>,
+        signal_sender: &broadcast::Sender<Signal>,
         conversation_id: String,
     ) -> Result<()> {
         log::debug!(target: target::SYNC, "Starting conversation sync for {}", conversation_id);
@@ -646,9 +702,7 @@ impl Daemon {
 
         // Send messages updated signal, if we actually inserted any messages.
         if num_messages > 0 {
-            signal_sender
-                .send(Signal::MessagesUpdated(conversation_id.clone()))
-                .await?;
+            Self::send_messages_updated(signal_sender, conversation_id.clone());
         }
 
         log::debug!(target: target::SYNC, "Synchronized {} messages for conversation {}", num_messages, &conversation_id);
@@ -669,14 +723,14 @@ impl Daemon {
     async fn update_conversation_metadata_impl(
         database: &mut Arc<Mutex<Database>>,
         conversation: Conversation,
-        signal_sender: &Sender<Signal>,
+        signal_sender: &broadcast::Sender<Signal>,
     ) -> Result<()> {
         log::debug!(target: target::DAEMON, "Updating conversation metadata: {}", conversation.guid);
         let updated = database
             .with_repository(|r| r.merge_conversation_metadata(conversation))
             .await?;
         if updated {
-            signal_sender.send(Signal::ConversationsUpdated).await?;
+            Self::send_signal(signal_sender, Signal::ConversationsUpdated, target::DAEMON);
         }
 
         Ok(())
@@ -691,6 +745,40 @@ impl Daemon {
         self.database.with_settings(|s| settings.save(s)).await
     }
 
+    fn send_signal(sender: &broadcast::Sender<Signal>, signal: Signal, log_target: &str) {
+        if let Err(error) = sender.send(signal) {
+            log::trace!(
+                target: log_target,
+                "Signal delivery skipped (no listeners?): {}",
+                error
+            );
+        }
+    }
+
+    fn send_internal(sender: &broadcast::Sender<Signal>, signal: InternalSignal) {
+        if let Err(error) = sender.send(Signal::Internal(signal)) {
+            log::trace!(
+                target: target::DAEMON,
+                "Internal signal delivery skipped: {}",
+                error
+            );
+        }
+    }
+
+    fn send_messages_updated(sender: &broadcast::Sender<Signal>, conversation_id: String) {
+        Self::send_internal(
+            sender,
+            InternalSignal::MessagesUpdated(conversation_id.clone()),
+        );
+        if let Err(error) = sender.send(Signal::MessagesUpdated(conversation_id)) {
+            log::warn!(
+                target: target::DAEMON,
+                "Failed to send MessagesUpdated signal: {}",
+                error
+            );
+        }
+    }
+
     async fn get_client_impl(
         database: &mut Arc<Mutex<Database>>,
     ) -> Result<HTTPAPIClient<DatabaseAuthenticationStore>> {
@@ -723,9 +811,11 @@ impl Daemon {
             })
             .await?;
 
-        self.signal_sender
-            .send(Signal::ConversationsUpdated)
-            .await?;
+        Self::send_signal(
+            &self.signal_sender,
+            Signal::ConversationsUpdated,
+            target::SYNC,
+        );
 
         Ok(())
     }

core/kordophoned/src/daemon/models/attachment.rs

@@ -1,4 +1,4 @@
-use std::path::PathBuf;
+use std::path::{Path, PathBuf};
 
 #[derive(Debug, Clone)]
 pub struct AttachmentMetadata {
@@ -16,9 +16,25 @@ pub struct Attachment {
     pub guid: String,
     pub base_path: PathBuf,
     pub metadata: Option<AttachmentMetadata>,
+    pub mime_type: Option<String>,
 }
 
 impl Attachment {
+    pub(crate) fn preferred_extension_for_mime(mime: &str) -> Option<&'static str> {
+        let normalized = mime.split(';').next().unwrap_or(mime).trim();
+        // Prefer common, user-friendly extensions over obscure ones
+        match normalized {
+            "image/jpeg" | "image/pjpeg" => Some("jpg"),
+            _ => mime_guess::get_mime_extensions_str(normalized).and_then(|list| {
+                // If jpg is one of the candidates, prefer it
+                if list.iter().any(|e| *e == "jpg") {
+                    Some("jpg")
+                } else {
+                    list.first().copied()
+                }
+            }),
+        }
+    }
     pub fn get_path(&self) -> PathBuf {
         self.get_path_for_preview_scratch(false, false)
     }
@@ -28,12 +44,34 @@ impl Attachment {
     }
 
     pub fn get_path_for_preview_scratch(&self, preview: bool, scratch: bool) -> PathBuf {
-        let extension = if preview { "preview" } else { "full" };
+        // Determine whether this is a preview or full attachment.
+        let kind = if preview { "preview" } else { "full" };
+
+        // If not a scratch path, and a file already exists on disk with a concrete
+        // file extension (e.g., guid.full.jpg), return that existing path.
+        if !scratch {
+            if let Some(existing) = self.find_existing_path(preview) {
+                return existing;
+            }
+        }
+
+        // Fall back to constructing a path using known info. If we know the MIME type,
+        // prefer an extension guessed from it; otherwise keep legacy naming.
+        let ext_from_mime = self
+            .mime_type
+            .as_ref()
+            .and_then(|m| Self::preferred_extension_for_mime(m));
+
+        let base_ext = match ext_from_mime {
+            Some(ext) => format!("{}.{}", kind, ext),
+            None => kind.to_string(),
+        };
+
         if scratch {
             self.base_path
-                .with_extension(format!("{}.download", extension))
+                .with_extension(format!("{}.download", base_ext))
         } else {
-            self.base_path.with_extension(extension)
+            self.base_path.with_extension(base_ext)
         }
     }
 
@@ -46,6 +84,36 @@ impl Attachment {
             .as_str(),
         )
     }
+
+    fn find_existing_path(&self, preview: bool) -> Option<PathBuf> {
+        let kind = if preview { "preview" } else { "full" };
+
+        // First, check legacy path without a concrete file extension.
+        let legacy = self.base_path.with_extension(kind);
+        if legacy.exists() {
+            return Some(legacy);
+        }
+
+        // Next, search for a filename like: <guid>.<kind>.<ext>
+        let file_stem = self
+            .base_path
+            .file_name()
+            .map(|s| s.to_string_lossy().to_string())?;
+        let prefix = format!("{}.{}.", file_stem, kind);
+
+        let parent = self.base_path.parent().unwrap_or_else(|| Path::new("."));
+        if let Ok(dir) = std::fs::read_dir(parent) {
+            for entry in dir.flatten() {
+                let file_name = entry.file_name();
+                let name = file_name.to_string_lossy();
+                if name.starts_with(&prefix) && !name.ends_with(".download") {
+                    return Some(entry.path());
+                }
+            }
+        }
+
+        None
+    }
 }
 
 impl From<kordophone::model::message::AttachmentMetadata> for AttachmentMetadata {

core/kordophoned/src/daemon/notifier.rs

@@ -0,0 +1,288 @@
+use super::contact_resolver::{ContactResolver, DefaultContactResolverBackend};
+use super::models::message::Participant;
+use super::signals::{InternalSignal, Signal};
+use super::{target, Message};
+
+use kordophone_db::{
+    database::{Database, DatabaseAccess},
+    models::Conversation,
+    models::Participant as DbParticipant,
+};
+use notify::Notification;
+use std::sync::Arc;
+use tokio::sync::{broadcast, Mutex};
+
+/// Centralised notification helper used by platform transports (D-Bus, XPC, …).
+pub struct NotificationService {
+    resolver: Mutex<ContactResolver<DefaultContactResolverBackend>>,
+}
+
+impl NotificationService {
+    pub fn new() -> Self {
+        Self {
+            resolver: Mutex::new(ContactResolver::new(
+                DefaultContactResolverBackend::default(),
+            )),
+        }
+    }
+
+    pub async fn listen(
+        self: Arc<Self>,
+        mut signal_rx: broadcast::Receiver<Signal>,
+        database: Arc<Mutex<Database>>,
+    ) {
+        log::trace!(target: target::DAEMON, "NotificationService listener started");
+        loop {
+            match signal_rx.recv().await {
+                Ok(Signal::Internal(InternalSignal::MessagesUpdated(conversation_id))) => {
+                    log::trace!(
+                        target: target::DAEMON,
+                        "NotificationService received MessagesUpdated for {}",
+                        conversation_id
+                    );
+                    self.notify_new_messages(&database, &conversation_id).await;
+                }
+                Ok(Signal::Internal(InternalSignal::TestNotification { summary, body })) => {
+                    log::trace!(
+                        target: target::DAEMON,
+                        "NotificationService received TestNotification"
+                    );
+                    if let Err(error) = self.send_manual(&summary, &body) {
+                        log::warn!(
+                            target: target::DAEMON,
+                            "Failed to display test notification: {}",
+                            error
+                        );
+                    }
+                }
+                Ok(other) => {
+                    log::trace!(
+                        target: target::DAEMON,
+                        "NotificationService ignoring signal: {:?}",
+                        other
+                    );
+                }
+                Err(broadcast::error::RecvError::Lagged(skipped)) => {
+                    log::warn!(
+                        target: target::DAEMON,
+                        "NotificationService lagged; skipped {} signals",
+                        skipped
+                    );
+                }
+                Err(broadcast::error::RecvError::Closed) => {
+                    log::trace!(target: target::DAEMON, "NotificationService listener exiting");
+                    break;
+                }
+            }
+        }
+    }
+
+    /// Checks whether a new user-visible notification should be shown for the
+    /// given conversation and displays it if appropriate.
+    pub async fn notify_new_messages(
+        &self,
+        database: &Arc<Mutex<Database>>,
+        conversation_id: &str,
+    ) {
+        log::trace!(
+            target: target::DAEMON,
+            "NotificationService preparing payload for {}",
+            conversation_id
+        );
+        if let Some((summary, body)) = self.prepare_payload(database, conversation_id).await {
+            log::trace!(
+                target: target::DAEMON,
+                "NotificationService displaying notification for {}",
+                conversation_id
+            );
+            if let Err(error) = self.show_notification(&summary, &body) {
+                log::warn!(
+                    target: target::DAEMON,
+                    "Failed to display notification for conversation {}: {}",
+                    conversation_id,
+                    error
+                );
+            }
+        } else {
+            log::trace!(
+                target: target::DAEMON,
+                "NotificationService skipping notification for {}",
+                conversation_id
+            );
+        }
+    }
+
+    /// Displays a manual test notification.
+    pub fn send_manual(&self, summary: &str, body: &str) -> Result<(), notify::error::Error> {
+        log::trace!(
+            target: target::DAEMON,
+            "NotificationService sending manual notification"
+        );
+        self.show_notification(summary, body)
+    }
+
+    async fn prepare_payload(
+        &self,
+        database: &Arc<Mutex<Database>>,
+        conversation_id: &str,
+    ) -> Option<(String, String)> {
+        let conversation_opt = database
+            .lock()
+            .await
+            .with_repository(|r| r.get_conversation_by_guid(conversation_id))
+            .await;
+
+        let conversation = match conversation_opt {
+            Ok(Some(conv)) => conv,
+            Ok(None) => {
+                log::trace!(
+                    target: target::DAEMON,
+                    "NotificationService: conversation {} not found",
+                    conversation_id
+                );
+                return None;
+            }
+            Err(err) => {
+                log::warn!(
+                    target: target::DAEMON,
+                    "Notification lookup failed for conversation {}: {}",
+                    conversation_id,
+                    err
+                );
+                return None;
+            }
+        };
+
+        if conversation.unread_count == 0 {
+            log::trace!(
+                target: target::DAEMON,
+                "NotificationService: conversation {} has no unread messages",
+                conversation_id
+            );
+            return None;
+        }
+
+        let last_message_opt = database
+            .lock()
+            .await
+            .with_repository(|r| r.get_last_message_for_conversation(conversation_id))
+            .await;
+
+        let last_message: Message = match last_message_opt {
+            Ok(Some(message)) => message.into(),
+            Ok(None) => {
+                log::trace!(
+                    target: target::DAEMON,
+                    "NotificationService: conversation {} has no messages",
+                    conversation_id
+                );
+                return None;
+            }
+            Err(err) => {
+                log::warn!(
+                    target: target::DAEMON,
+                    "Notification lookup failed for conversation {}: {}",
+                    conversation_id,
+                    err
+                );
+                return None;
+            }
+        };
+
+        if matches!(last_message.sender, Participant::Me) {
+            log::trace!(
+                target: target::DAEMON,
+                "NotificationService: last message in {} was sent by self",
+                conversation_id
+            );
+            return None;
+        }
+
+        let mut resolver = self.resolver.lock().await;
+        let summary = self.conversation_display_name(&conversation, &mut resolver);
+        let sender_display_name =
+            self.resolve_participant_display_name(&last_message.sender, &mut resolver);
+
+        let mut message_text = last_message.text.replace('\u{FFFC}', "");
+        if message_text.trim().is_empty() {
+            if !last_message.attachments.is_empty() {
+                message_text = "Sent an attachment".to_string();
+            } else {
+                message_text = "Sent a message".to_string();
+            }
+        }
+
+        let body = if sender_display_name.is_empty() {
+            message_text
+        } else {
+            format!("{}: {}", sender_display_name, message_text)
+        };
+
+        Some((summary, body))
+    }
+
+    fn conversation_display_name(
+        &self,
+        conversation: &Conversation,
+        resolver: &mut ContactResolver<DefaultContactResolverBackend>,
+    ) -> String {
+        if let Some(display_name) = &conversation.display_name {
+            if !display_name.trim().is_empty() {
+                return display_name.clone();
+            }
+        }
+
+        let names: Vec<String> = conversation
+            .participants
+            .iter()
+            .filter_map(|participant| match participant {
+                DbParticipant::Me => None,
+                DbParticipant::Remote { handle, contact_id } => {
+                    if let Some(contact_id) = contact_id {
+                        Some(
+                            resolver
+                                .get_contact_display_name(contact_id)
+                                .unwrap_or_else(|| handle.clone()),
+                        )
+                    } else {
+                        Some(handle.clone())
+                    }
+                }
+            })
+            .collect();
+
+        if names.is_empty() {
+            "Kordophone".to_string()
+        } else {
+            names.join(", ")
+        }
+    }
+
+    fn resolve_participant_display_name(
+        &self,
+        participant: &Participant,
+        resolver: &mut ContactResolver<DefaultContactResolverBackend>,
+    ) -> String {
+        match participant {
+            Participant::Me => "".to_string(),
+            Participant::Remote { handle, contact_id } => {
+                if let Some(contact_id) = contact_id {
+                    resolver
+                        .get_contact_display_name(contact_id)
+                        .unwrap_or_else(|| handle.clone())
+                } else {
+                    handle.clone()
+                }
+            }
+        }
+    }
+
+    fn show_notification(&self, summary: &str, body: &str) -> Result<(), notify::error::Error> {
+        Notification::new()
+            .appname("Kordophone")
+            .summary(summary)
+            .body(body)
+            .show()
+            .map(|_| ())
+    }
+}

core/kordophoned/src/daemon/signals.rs

@@ -1,3 +1,12 @@
+#[derive(Debug, Clone)]
+pub enum InternalSignal {
+    /// Notification that new messages are available for a conversation.
+    MessagesUpdated(String),
+
+    /// Manual test notification request.
+    TestNotification { summary: String, body: String },
+}
+
 #[derive(Debug, Clone)]
 pub enum Signal {
     /// Emitted when the list of conversations is updated.
@@ -21,4 +30,7 @@ pub enum Signal {
 
     /// Emitted when the update stream is reconnected after a timeout or configuration change.
     UpdateStreamReconnected,
+
+    /// Internal-only signals consumed by daemon components.
+    Internal(InternalSignal),
 }

core/kordophoned/src/dbus/agent.rs

@@ -2,7 +2,7 @@ use dbus::arg;
 use dbus_tree::MethodErr;
 use std::sync::Arc;
 use std::{future::Future, thread};
-use tokio::sync::{mpsc, oneshot, Mutex};
+use tokio::sync::{broadcast, mpsc, oneshot, Mutex};
 
 use kordophoned::daemon::{
     contact_resolver::{ContactResolver, DefaultContactResolverBackend},
@@ -22,12 +22,15 @@ use dbus_tokio::connection;
 #[derive(Clone)]
 pub struct DBusAgent {
     event_sink: mpsc::Sender<Event>,
-    signal_receiver: Arc<Mutex<Option<mpsc::Receiver<Signal>>>>,
+    signal_receiver: Arc<Mutex<Option<broadcast::Receiver<Signal>>>>,
     contact_resolver: ContactResolver<DefaultContactResolverBackend>,
 }
 
 impl DBusAgent {
-    pub fn new(event_sink: mpsc::Sender<Event>, signal_receiver: mpsc::Receiver<Signal>) -> Self {
+    pub fn new(
+        event_sink: mpsc::Sender<Event>,
+        signal_receiver: broadcast::Receiver<Signal>,
+    ) -> Self {
         Self {
             event_sink,
             signal_receiver: Arc::new(Mutex::new(Some(signal_receiver))),
@@ -75,80 +78,95 @@ impl DBusAgent {
                     .take()
                     .expect("Signal receiver already taken");
 
-                while let Some(signal) = receiver.recv().await {
-                    match signal {
-                        Signal::ConversationsUpdated => {
-                            log::debug!("Sending signal: ConversationsUpdated");
-                            registry
-                                .send_signal(
-                                    interface::OBJECT_PATH,
-                                    DbusSignals::ConversationsUpdated {},
-                                )
-                                .unwrap_or_else(|_| {
-                                    log::error!("Failed to send signal");
-                                    0
-                                });
-                        }
-                        Signal::MessagesUpdated(conversation_id) => {
-                            log::debug!(
-                                "Sending signal: MessagesUpdated for conversation {}",
-                                conversation_id
+                loop {
+                    match receiver.recv().await {
+                        Ok(signal) => match signal {
+                            Signal::ConversationsUpdated => {
+                                log::debug!("Sending signal: ConversationsUpdated");
+                                registry
+                                    .send_signal(
+                                        interface::OBJECT_PATH,
+                                        DbusSignals::ConversationsUpdated {},
+                                    )
+                                    .unwrap_or_else(|_| {
+                                        log::error!("Failed to send signal");
+                                        0
+                                    });
+                            }
+                            Signal::MessagesUpdated(conversation_id) => {
+                                log::debug!(
+                                    "Sending signal: MessagesUpdated for conversation {}",
+                                    conversation_id
+                                );
+                                registry
+                                    .send_signal(
+                                        interface::OBJECT_PATH,
+                                        DbusSignals::MessagesUpdated { conversation_id },
+                                    )
+                                    .unwrap_or_else(|_| {
+                                        log::error!("Failed to send signal");
+                                        0
+                                    });
+                            }
+                            Signal::AttachmentDownloaded(attachment_id) => {
+                                log::debug!(
+                                    "Sending signal: AttachmentDownloaded for attachment {}",
+                                    attachment_id
+                                );
+                                registry
+                                    .send_signal(
+                                        interface::OBJECT_PATH,
+                                        DbusSignals::AttachmentDownloadCompleted { attachment_id },
+                                    )
+                                    .unwrap_or_else(|_| {
+                                        log::error!("Failed to send signal");
+                                        0
+                                    });
+                            }
+                            Signal::AttachmentUploaded(upload_guid, attachment_guid) => {
+                                log::debug!(
+                                    "Sending signal: AttachmentUploaded for upload {}, attachment {}",
+                                    upload_guid,
+                                    attachment_guid
+                                );
+                                registry
+                                    .send_signal(
+                                        interface::OBJECT_PATH,
+                                        DbusSignals::AttachmentUploadCompleted {
+                                            upload_guid,
+                                            attachment_guid,
+                                        },
+                                    )
+                                    .unwrap_or_else(|_| {
+                                        log::error!("Failed to send signal");
+                                        0
+                                    });
+                            }
+                            Signal::UpdateStreamReconnected => {
+                                log::debug!("Sending signal: UpdateStreamReconnected");
+                                registry
+                                    .send_signal(
+                                        interface::OBJECT_PATH,
+                                        DbusSignals::UpdateStreamReconnected {},
+                                    )
+                                    .unwrap_or_else(|_| {
+                                        log::error!("Failed to send signal");
+                                        0
+                                    });
+                            }
+                            Signal::Internal(_) => {
+                                log::trace!("Ignoring internal signal for D-Bus transport");
+                            }
+                        },
+                        Err(broadcast::error::RecvError::Lagged(skipped)) => {
+                            log::warn!(
+                                "Signal receiver lagged; skipped {} daemon signals",
+                                skipped
                             );
-                            registry
-                                .send_signal(
-                                    interface::OBJECT_PATH,
-                                    DbusSignals::MessagesUpdated { conversation_id },
-                                )
-                                .unwrap_or_else(|_| {
-                                    log::error!("Failed to send signal");
-                                    0
-                                });
                         }
-                        Signal::AttachmentDownloaded(attachment_id) => {
-                            log::debug!(
-                                "Sending signal: AttachmentDownloaded for attachment {}",
-                                attachment_id
-                            );
-                            registry
-                                .send_signal(
-                                    interface::OBJECT_PATH,
-                                    DbusSignals::AttachmentDownloadCompleted { attachment_id },
-                                )
-                                .unwrap_or_else(|_| {
-                                    log::error!("Failed to send signal");
-                                    0
-                                });
-                        }
-                        Signal::AttachmentUploaded(upload_guid, attachment_guid) => {
-                            log::debug!(
-                                "Sending signal: AttachmentUploaded for upload {}, attachment {}",
-                                upload_guid,
-                                attachment_guid
-                            );
-                            registry
-                                .send_signal(
-                                    interface::OBJECT_PATH,
-                                    DbusSignals::AttachmentUploadCompleted {
-                                        upload_guid,
-                                        attachment_guid,
-                                    },
-                                )
-                                .unwrap_or_else(|_| {
-                                    log::error!("Failed to send signal");
-                                    0
-                                });
-                        }
-                        Signal::UpdateStreamReconnected => {
-                            log::debug!("Sending signal: UpdateStreamReconnected");
-                            registry
-                                .send_signal(
-                                    interface::OBJECT_PATH,
-                                    DbusSignals::UpdateStreamReconnected {},
-                                )
-                                .unwrap_or_else(|_| {
-                                    log::error!("Failed to send signal");
-                                    0
-                                });
+                        Err(broadcast::error::RecvError::Closed) => {
+                            log::warn!("Signal channel closed; stopping D-Bus forwarding");
+                            break;
                         }
                     }
                 }
@@ -398,6 +416,13 @@ impl DbusRepository for DBusAgent {
             .map(|uuid| uuid.to_string())
     }
 
+    fn test_notification(&mut self, summary: String, body: String) -> Result<(), MethodErr> {
+        match self.send_event_sync(|r| Event::TestNotification(summary, body, r))? {
+            Ok(()) => Ok(()),
+            Err(message) => Err(MethodErr::failed(&message)),
+        }
+    }
+
     fn get_attachment_info(
         &mut self,
         attachment_id: String,

core/kordophoned/src/main.rs

@@ -26,7 +26,7 @@ async fn start_ipc_agent(daemon: &mut Daemon) {
     use dbus::agent::DBusAgent;
 
     // Start the D-Bus agent (events in, signals out).
-    let agent = DBusAgent::new(daemon.event_sender.clone(), daemon.obtain_signal_receiver());
+    let agent = DBusAgent::new(daemon.event_sender.clone(), daemon.subscribe_signals());
     tokio::spawn(async move {
         agent.run().await;
     });
@@ -35,8 +35,7 @@ async fn start_ipc_agent(daemon: &mut Daemon) {
 #[cfg(target_os = "macos")]
 async fn start_ipc_agent(daemon: &mut Daemon) {
     // Start the macOS XPC agent (events in, signals out) on a dedicated thread.
-    let agent =
-        xpc::agent::XpcAgent::new(daemon.event_sender.clone(), daemon.obtain_signal_receiver());
+    let agent = xpc::agent::XpcAgent::new(daemon.event_sender.clone(), daemon.subscribe_signals());
     std::thread::spawn(move || {
         // Use a single-threaded Tokio runtime for the XPC agent.
         let rt = tokio::runtime::Builder::new_current_thread()

core/kordophoned/src/xpc/agent.rs

@@ -4,7 +4,7 @@ use std::ffi::CString;
 use std::os::raw::c_char;
 use std::ptr;
 use std::sync::Arc;
-use tokio::sync::{mpsc, oneshot, Mutex};
+use tokio::sync::{broadcast, mpsc, oneshot, Mutex};
 use xpc_connection::{message_to_xpc_object, xpc_object_to_message, Message, MessageError};
 use xpc_connection_sys as xpc_sys;
 
@@ -22,11 +22,14 @@ type Subscribers = Arc<std::sync::Mutex<Vec<XpcConn>>>;
 #[derive(Clone)]
 pub struct XpcAgent {
     event_sink: mpsc::Sender<Event>,
-    signal_receiver: Arc<Mutex<Option<mpsc::Receiver<Signal>>>>,
+    signal_receiver: Arc<Mutex<Option<broadcast::Receiver<Signal>>>>,
 }
 
 impl XpcAgent {
-    pub fn new(event_sink: mpsc::Sender<Event>, signal_receiver: mpsc::Receiver<Signal>) -> Self {
+    pub fn new(
+        event_sink: mpsc::Sender<Event>,
+        signal_receiver: broadcast::Receiver<Signal>,
+    ) -> Self {
         Self {
             event_sink,
             signal_receiver: Arc::new(Mutex::new(Some(signal_receiver))),
@@ -71,7 +74,31 @@ impl XpcAgent {
                     .await
                     .take()
                     .expect("Signal receiver already taken");
-                while let Some(signal) = receiver.recv().await {
+                loop {
+                    let signal = match receiver.recv().await {
+                        Ok(signal) => signal,
+                        Err(broadcast::error::RecvError::Lagged(skipped)) => {
+                            log::warn!(
+                                target: LOG_TARGET,
+                                "XPC agent lagged; skipped {} signals",
+                                skipped
+                            );
+                            continue;
+                        }
+                        Err(broadcast::error::RecvError::Closed) => {
+                            log::warn!(
+                                target: LOG_TARGET,
+                                "Signal channel closed; stopping XPC forwarding"
+                            );
+                            break;
+                        }
+                    };
+
+                    if matches!(signal, Signal::Internal(_)) {
+                        log::trace!(target: LOG_TARGET, "Skipping internal signal for XPC");
+                        continue;
+                    }
+
                     log::trace!(target: LOG_TARGET, "Broadcasting signal: {:?}", signal);
                     let msg = super::util::signal_to_message(signal);
                     let xobj = message_to_xpc_object(msg);
@@ -127,7 +154,7 @@ impl XpcAgent {
 
                                 // Drop any cleanup resource now that payload is constructed and sent.
                                 drop(result.cleanup);
-                                
+
                                 log::trace!(target: LOG_TARGET, "XPC reply sent for method: {}", method);
                             } else {
                                 log::warn!(target: LOG_TARGET, "No reply port for method: {}", method);

core/kordophoned/src/xpc/mod.rs

@@ -15,10 +15,16 @@ pub struct DispatchResult {
 
 impl DispatchResult {
     pub fn new(message: Message) -> Self {
-        Self { message, cleanup: None }
+        Self {
+            message,
+            cleanup: None,
+        }
     }
 
     pub fn with_cleanup<T: Any + Send + 'static>(message: Message, cleanup: T) -> Self {
-        Self { message, cleanup: Some(Box::new(cleanup)) }
+        Self {
+            message,
+            cleanup: Some(Box::new(cleanup)),
+        }
     }
 }

core/kordophoned/src/xpc/rpc.rs

@@ -105,7 +105,12 @@ pub async fn dispatch(
                 .and_then(|m| dict_get_str(m, "conversation_id"))
             {
                 Some(id) => id,
-                None => return DispatchResult::new(make_error_reply("InvalidRequest", "Missing conversation_id")),
+                None => {
+                    return DispatchResult::new(make_error_reply(
+                        "InvalidRequest",
+                        "Missing conversation_id",
+                    ))
+                }
             };
             match agent
                 .send_event(|r| Event::SyncConversation(conversation_id, r))
@@ -122,7 +127,12 @@ pub async fn dispatch(
                 .and_then(|m| dict_get_str(m, "conversation_id"))
             {
                 Some(id) => id,
-                None => return DispatchResult::new(make_error_reply("InvalidRequest", "Missing conversation_id")),
+                None => {
+                    return DispatchResult::new(make_error_reply(
+                        "InvalidRequest",
+                        "Missing conversation_id",
+                    ))
+                }
             };
             match agent
                 .send_event(|r| Event::MarkConversationAsRead(conversation_id, r))
@@ -137,11 +147,21 @@ pub async fn dispatch(
         "GetMessages" => {
             let args = match get_dictionary_field(root, "arguments") {
                 Some(a) => a,
-                None => return DispatchResult::new(make_error_reply("InvalidRequest", "Missing arguments")),
+                None => {
+                    return DispatchResult::new(make_error_reply(
+                        "InvalidRequest",
+                        "Missing arguments",
+                    ))
+                }
             };
             let conversation_id = match dict_get_str(args, "conversation_id") {
                 Some(id) => id,
-                None => return DispatchResult::new(make_error_reply("InvalidRequest", "Missing conversation_id")),
+                None => {
+                    return DispatchResult::new(make_error_reply(
+                        "InvalidRequest",
+                        "Missing conversation_id",
+                    ))
+                }
             };
             let last_message_id = dict_get_str(args, "last_message_id");
             match agent
@@ -158,13 +178,10 @@ pub async fn dispatch(
                         dict_put_str(&mut m, "sender", &msg.sender.display_name());
 
                         // Include attachment GUIDs for the client to resolve/download
-                        let attachment_guids: Vec<String> = msg
-                            .attachments
-                            .iter()
-                            .map(|a| a.guid.clone())
-                            .collect();
+                        let attachment_guids: Vec<String> =
+                            msg.attachments.iter().map(|a| a.guid.clone()).collect();
                         m.insert(cstr("attachment_guids"), array_from_strs(attachment_guids));
-                        
+
                         // Full attachments array with metadata (mirrors DBus fields)
                         let mut attachments_items: Vec<Message> = Vec::new();
                         for attachment in msg.attachments.iter() {
@@ -193,12 +210,23 @@ pub async fn dispatch(
                                 if let Some(attribution_info) = &metadata.attribution_info {
                                     let mut attribution_map: XpcMap = HashMap::new();
                                     if let Some(width) = attribution_info.width {
-                                        dict_put_i64_as_str(&mut attribution_map, "width", width as i64);
+                                        dict_put_i64_as_str(
+                                            &mut attribution_map,
+                                            "width",
+                                            width as i64,
+                                        );
                                     }
                                     if let Some(height) = attribution_info.height {
-                                        dict_put_i64_as_str(&mut attribution_map, "height", height as i64);
+                                        dict_put_i64_as_str(
+                                            &mut attribution_map,
+                                            "height",
+                                            height as i64,
+                                        );
                                     }
-                                    metadata_map.insert(cstr("attribution_info"), Message::Dictionary(attribution_map));
+                                    metadata_map.insert(
+                                        cstr("attribution_info"),
+                                        Message::Dictionary(attribution_map),
+                                    );
                                 }
                                 if !metadata_map.is_empty() {
                                     a.insert(cstr("metadata"), Message::Dictionary(metadata_map));
@@ -208,7 +236,7 @@ pub async fn dispatch(
                             attachments_items.push(Message::Dictionary(a));
                         }
                         m.insert(cstr("attachments"), Message::Array(attachments_items));
-                        
+
                         items.push(Message::Dictionary(m));
                     }
                     let mut reply: XpcMap = HashMap::new();
@@ -230,11 +258,21 @@ pub async fn dispatch(
         "SendMessage" => {
             let args = match get_dictionary_field(root, "arguments") {
                 Some(a) => a,
-                None => return DispatchResult::new(make_error_reply("InvalidRequest", "Missing arguments")),
+                None => {
+                    return DispatchResult::new(make_error_reply(
+                        "InvalidRequest",
+                        "Missing arguments",
+                    ))
+                }
             };
             let conversation_id = match dict_get_str(args, "conversation_id") {
                 Some(v) => v,
-                None => return DispatchResult::new(make_error_reply("InvalidRequest", "Missing conversation_id")),
+                None => {
+                    return DispatchResult::new(make_error_reply(
+                        "InvalidRequest",
+                        "Missing conversation_id",
+                    ))
+                }
             };
             let text = dict_get_str(args, "text").unwrap_or_default();
             let attachment_guids: Vec<String> = match args.get(&cstr("attachment_guids")) {
@@ -265,11 +303,21 @@ pub async fn dispatch(
         "GetAttachmentInfo" => {
             let args = match get_dictionary_field(root, "arguments") {
                 Some(a) => a,
-                None => return DispatchResult::new(make_error_reply("InvalidRequest", "Missing arguments")),
+                None => {
+                    return DispatchResult::new(make_error_reply(
+                        "InvalidRequest",
+                        "Missing arguments",
+                    ))
+                }
             };
             let attachment_id = match dict_get_str(args, "attachment_id") {
                 Some(v) => v,
-                None => return DispatchResult::new(make_error_reply("InvalidRequest", "Missing attachment_id")),
+                None => {
+                    return DispatchResult::new(make_error_reply(
+                        "InvalidRequest",
+                        "Missing attachment_id",
+                    ))
+                }
             };
             match agent
                 .send_event(|r| Event::GetAttachment(attachment_id, r))
@@ -308,11 +356,21 @@ pub async fn dispatch(
         "OpenAttachmentFd" => {
             let args = match get_dictionary_field(root, "arguments") {
                 Some(a) => a,
-                None => return DispatchResult::new(make_error_reply("InvalidRequest", "Missing arguments")),
+                None => {
+                    return DispatchResult::new(make_error_reply(
+                        "InvalidRequest",
+                        "Missing arguments",
+                    ))
+                }
             };
             let attachment_id = match dict_get_str(args, "attachment_id") {
                 Some(v) => v,
-                None => return DispatchResult::new(make_error_reply("InvalidRequest", "Missing attachment_id")),
+                None => {
+                    return DispatchResult::new(make_error_reply(
+                        "InvalidRequest",
+                        "Missing attachment_id",
+                    ))
+                }
             };
             let preview = dict_get_str(args, "preview")
                 .map(|s| s == "true")
@@ -324,7 +382,7 @@ pub async fn dispatch(
             {
                 Ok(attachment) => {
                     use std::os::fd::AsRawFd;
-                    
+
                     let path = attachment.get_path_for_preview(preview);
                     match std::fs::OpenOptions::new().read(true).open(&path) {
                         Ok(file) => {
@@ -335,9 +393,14 @@ pub async fn dispatch(
                             dict_put_str(&mut reply, "type", "OpenAttachmentFdResponse");
                             reply.insert(cstr("fd"), Message::Fd(fd));
 
-                            DispatchResult { message: Message::Dictionary(reply), cleanup: Some(Box::new(file)) }
+                            DispatchResult {
+                                message: Message::Dictionary(reply),
+                                cleanup: Some(Box::new(file)),
+                            }
+                        }
+                        Err(e) => {
+                            DispatchResult::new(make_error_reply("OpenFailed", &format!("{}", e)))
                         }
-                        Err(e) => DispatchResult::new(make_error_reply("OpenFailed", &format!("{}", e))),
                     }
                 }
                 Err(e) => DispatchResult::new(make_error_reply("DaemonError", &format!("{}", e))),
@@ -348,11 +411,21 @@ pub async fn dispatch(
         "DownloadAttachment" => {
             let args = match get_dictionary_field(root, "arguments") {
                 Some(a) => a,
-                None => return DispatchResult::new(make_error_reply("InvalidRequest", "Missing arguments")),
+                None => {
+                    return DispatchResult::new(make_error_reply(
+                        "InvalidRequest",
+                        "Missing arguments",
+                    ))
+                }
             };
             let attachment_id = match dict_get_str(args, "attachment_id") {
                 Some(v) => v,
-                None => return DispatchResult::new(make_error_reply("InvalidRequest", "Missing attachment_id")),
+                None => {
+                    return DispatchResult::new(make_error_reply(
+                        "InvalidRequest",
+                        "Missing attachment_id",
+                    ))
+                }
             };
             let preview = dict_get_str(args, "preview")
                 .map(|s| s == "true")
@@ -371,11 +444,18 @@ pub async fn dispatch(
             use std::path::PathBuf;
             let args = match get_dictionary_field(root, "arguments") {
                 Some(a) => a,
-                None => return DispatchResult::new(make_error_reply("InvalidRequest", "Missing arguments")),
+                None => {
+                    return DispatchResult::new(make_error_reply(
+                        "InvalidRequest",
+                        "Missing arguments",
+                    ))
+                }
             };
             let path = match dict_get_str(args, "path") {
                 Some(v) => v,
-                None => return DispatchResult::new(make_error_reply("InvalidRequest", "Missing path")),
+                None => {
+                    return DispatchResult::new(make_error_reply("InvalidRequest", "Missing path"))
+                }
             };
             match agent
                 .send_event(|r| Event::UploadAttachment(PathBuf::from(path), r))
@@ -413,7 +493,12 @@ pub async fn dispatch(
         "UpdateSettings" => {
             let args = match get_dictionary_field(root, "arguments") {
                 Some(a) => a,
-                None => return DispatchResult::new(make_error_reply("InvalidRequest", "Missing arguments")),
+                None => {
+                    return DispatchResult::new(make_error_reply(
+                        "InvalidRequest",
+                        "Missing arguments",
+                    ))
+                }
             };
             let server_url = dict_get_str(args, "server_url");
             let username = dict_get_str(args, "username");

core/kpcli/src/client/mod.rs

@@ -143,15 +143,23 @@ impl ClientCli {
 
         println!("Listening for raw updates...");
         let mut stream = socket.raw_updates().await;
-        while let Some(Ok(update)) = stream.next().await {
-            match update {
-                SocketUpdate::Update(updates) => {
-                    for update in updates {
-                        println!("Got update: {:?}", update);
+
+        loop {
+            match stream.next().await.unwrap() {
+                Ok(update) => match update {
+                    SocketUpdate::Update(updates) => {
+                        for update in updates {
+                            println!("Got update: {:?}", update);
+                        }
                     }
-                }
-                SocketUpdate::Pong => {
-                    println!("Pong");
+                    SocketUpdate::Pong => {
+                        println!("Pong");
+                    }
+                },
+
+                Err(e) => {
+                    println!("Update error: {:?}", e);
+                    break;
                 }
             }
         }

core/kpcli/src/daemon/dbus.rs

@@ -209,4 +209,9 @@ impl DaemonInterface for DBusDaemonInterface {
         KordophoneRepository::mark_conversation_as_read(&self.proxy(), &conversation_id)
             .map_err(|e| anyhow::anyhow!("Failed to mark conversation as read: {}", e))
     }
+
+    async fn test_notification(&mut self, summary: String, body: String) -> Result<()> {
+        KordophoneRepository::test_notification(&self.proxy(), &summary, &body)
+            .map_err(|e| anyhow::anyhow!("Failed to trigger test notification: {}", e))
+    }
 }

core/kpcli/src/daemon/mod.rs

@@ -32,6 +32,7 @@ pub trait DaemonInterface {
     async fn download_attachment(&mut self, attachment_id: String) -> Result<()>;
     async fn upload_attachment(&mut self, path: String) -> Result<()>;
     async fn mark_conversation_as_read(&mut self, conversation_id: String) -> Result<()>;
+    async fn test_notification(&mut self, summary: String, body: String) -> Result<()>;
 }
 
 struct StubDaemonInterface;
@@ -112,6 +113,11 @@ impl DaemonInterface for StubDaemonInterface {
             "Daemon interface not implemented on this platform"
         ))
     }
+    async fn test_notification(&mut self, _summary: String, _body: String) -> Result<()> {
+        Err(anyhow::anyhow!(
+            "Daemon interface not implemented on this platform"
+        ))
+    }
 }
 
 pub fn new_daemon_interface() -> Result<Box<dyn DaemonInterface>> {
@@ -175,6 +181,9 @@ pub enum Commands {
 
     /// Marks a conversation as read.
     MarkConversationAsRead { conversation_id: String },
+
+    /// Displays a test notification using the daemon.
+    TestNotification { summary: String, body: String },
 }
 
 #[derive(Subcommand)]
@@ -219,6 +228,9 @@ impl Commands {
             Commands::MarkConversationAsRead { conversation_id } => {
                 client.mark_conversation_as_read(conversation_id).await
             }
+            Commands::TestNotification { summary, body } => {
+                client.test_notification(summary, body).await
+            }
         }
     }
 }

core/utilities/src/bin/snoozer.rs

@@ -1,13 +1,13 @@
 use std::env;
 use std::process;
 
-use kordophone::{
-    api::{HTTPAPIClient, InMemoryAuthenticationStore, EventSocket},
-    model::{ConversationID, event::EventData},
-    APIInterface,
-};
-use kordophone::api::http_client::Credentials;
 use kordophone::api::AuthenticationStore;
+use kordophone::api::http_client::Credentials;
+use kordophone::{
+    APIInterface,
+    api::{EventSocket, HTTPAPIClient, InMemoryAuthenticationStore},
+    model::{ConversationID, event::EventData},
+};
 
 use futures_util::StreamExt;
 use hyper::Uri;
@@ -18,7 +18,10 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     let args: Vec<String> = env::args().collect();
     if args.len() < 2 {
-        eprintln!("Usage: {} <conversation_id1> [conversation_id2] [conversation_id3] ...", args[0]);
+        eprintln!(
+            "Usage: {} <conversation_id1> [conversation_id2] [conversation_id3] ...",
+            args[0]
+        );
         eprintln!("Environment variables required:");
         eprintln!("  KORDOPHONE_API_URL - Server URL");
         eprintln!("  KORDOPHONE_USERNAME - Username for authentication");
@@ -30,65 +33,74 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let server_url: Uri = env::var("KORDOPHONE_API_URL")
         .map_err(|_| "KORDOPHONE_API_URL environment variable not set")?
         .parse()?;
-    
+
     let username = env::var("KORDOPHONE_USERNAME")
         .map_err(|_| "KORDOPHONE_USERNAME environment variable not set")?;
-    
+
     let password = env::var("KORDOPHONE_PASSWORD")
         .map_err(|_| "KORDOPHONE_PASSWORD environment variable not set")?;
-    
+
     let credentials = Credentials { username, password };
-    
+
     // Collect all conversation IDs from command line arguments
-    let target_conversation_ids: Vec<ConversationID> = args[1..].iter()
-        .map(|id| id.clone())
-        .collect();
-    
-    println!("Monitoring {} conversation(s) for updates: {:?}", 
-             target_conversation_ids.len(), target_conversation_ids);
-    
+    let target_conversation_ids: Vec<ConversationID> =
+        args[1..].iter().map(|id| id.clone()).collect();
+
+    println!(
+        "Monitoring {} conversation(s) for updates: {:?}",
+        target_conversation_ids.len(),
+        target_conversation_ids
+    );
+
     let auth_store = InMemoryAuthenticationStore::new(Some(credentials.clone()));
     let mut client = HTTPAPIClient::new(server_url, auth_store);
     let _ = client.authenticate(credentials).await?;
-    
+
     // Open event socket
     let event_socket = client.open_event_socket(None).await?;
     let (mut stream, _sink) = event_socket.events().await;
-    
+
     println!("Connected to event stream, waiting for updates...");
-    
+
     // Process events
     while let Some(event_result) = stream.next().await {
         match event_result {
             Ok(socket_event) => {
                 match socket_event {
-                    kordophone::api::event_socket::SocketEvent::Update(event) => {
-                        match event.data {
-                            EventData::MessageReceived(conversation, _message) => {
-                                if target_conversation_ids.contains(&conversation.guid) {
-                                    println!("Message update detected for conversation {}, marking as read...", conversation.guid);
-                                    match client.mark_conversation_as_read(&conversation.guid).await {
-                                        Ok(_) => println!("Successfully marked conversation {} as read", conversation.guid),
-                                        Err(e) => eprintln!("Failed to mark conversation {} as read: {:?}", conversation.guid, e),
-                                    }
+                    kordophone::api::event_socket::SocketEvent::Update(event) => match event.data {
+                        EventData::MessageReceived(conversation, _message) => {
+                            if target_conversation_ids.contains(&conversation.guid) {
+                                println!(
+                                    "Message update detected for conversation {}, marking as read...",
+                                    conversation.guid
+                                );
+                                match client.mark_conversation_as_read(&conversation.guid).await {
+                                    Ok(_) => println!(
+                                        "Successfully marked conversation {} as read",
+                                        conversation.guid
+                                    ),
+                                    Err(e) => eprintln!(
+                                        "Failed to mark conversation {} as read: {:?}",
+                                        conversation.guid, e
+                                    ),
                                 }
-                            },
-
-                            _ => {}
+                            }
                         }
+
+                        _ => {}
                     },
                     kordophone::api::event_socket::SocketEvent::Pong => {
                         // Ignore pong messages
                     }
                 }
-            },
+            }
             Err(e) => {
                 eprintln!("Error receiving event: {:?}", e);
                 break;
             }
         }
     }
-    
+
     println!("Event stream ended");
     Ok(())
 }

gtk/Makefile

@@ -0,0 +1,14 @@
+
+all: setup
+	ninja -C build
+
+setup: build/
+	meson build
+
+VER := 1.0.2
+TMP := $(shell mktemp -d)
+rpm:
+	git -C .. archive --format=tar.gz --prefix=kordophone/ -o $(TMP)/v$(VER).tar.gz HEAD
+	rpmbuild -ba dist/rpm/kordophone.spec --define "_sourcedir $(TMP)"
+
+

gtk/dist/rpm/kordophone.spec

@@ -1,11 +1,11 @@
 Name:           kordophone
-Version:        1.0.1
+Version:        1.0.2
 Release:        1%{?dist}
 Summary:        GTK4/Libadwaita client for Kordophone
 
 License:        GPL
 URL:            https://code.buzzert.dev/buzzert/Kordophone
-Source0:        https://code.buzzert.dev/buzzert/Kordophone/archive/master.tar.gz
+Source0:        %{url}/archive/v%{version}.tar.gz
 
 BuildRequires:  meson >= 0.56.0
 BuildRequires:  vala

osx/kordophone2.xcodeproj/project.pbxproj

@@ -32,29 +32,29 @@
 /* End PBXCopyFilesBuildPhase section */
 
 /* Begin PBXFileReference section */
-		CD41F5972E5B8E7300E0027B /* kordophone2.app */ = {isa = PBXFileReference; explicitFileType = wrapper.application; includeInIndex = 0; path = kordophone2.app; sourceTree = BUILT_PRODUCTS_DIR; };
+		CD41F5972E5B8E7300E0027B /* Kordophone.app */ = {isa = PBXFileReference; explicitFileType = wrapper.application; includeInIndex = 0; path = Kordophone.app; sourceTree = BUILT_PRODUCTS_DIR; };
 /* End PBXFileReference section */
 
 /* Begin PBXFileSystemSynchronizedBuildFileExceptionSet section */
-		CD41F5DA2E62850100E0027B /* Exceptions for "kordophone2" folder in "kordophone2" target */ = {
+		CD41F5DA2E62850100E0027B /* Exceptions for "kordophone2" folder in "Kordophone" target */ = {
 			isa = PBXFileSystemSynchronizedBuildFileExceptionSet;
 			membershipExceptions = (
 				Daemon/kordophoned,
 				Daemon/net.buzzert.kordophonecd.plist,
 			);
-			target = CD41F5962E5B8E7300E0027B /* kordophone2 */;
+			target = CD41F5962E5B8E7300E0027B /* Kordophone */;
 		};
 /* End PBXFileSystemSynchronizedBuildFileExceptionSet section */
 
 /* Begin PBXFileSystemSynchronizedGroupBuildPhaseMembershipExceptionSet section */
-		CD41F5DC2E62853800E0027B /* Exceptions for "kordophone2" folder in "Copy Files" phase from "kordophone2" target */ = {
+		CD41F5DC2E62853800E0027B /* Exceptions for "kordophone2" folder in "Copy Files" phase from "Kordophone" target */ = {
 			isa = PBXFileSystemSynchronizedGroupBuildPhaseMembershipExceptionSet;
 			buildPhase = CD41F5D92E6284FD00E0027B /* CopyFiles */;
 			membershipExceptions = (
 				Daemon/net.buzzert.kordophonecd.plist,
 			);
 		};
-		CD41F5E12E62860700E0027B /* Exceptions for "kordophone2" folder in "Copy Files" phase from "kordophone2" target */ = {
+		CD41F5E12E62860700E0027B /* Exceptions for "kordophone2" folder in "Copy Files" phase from "Kordophone" target */ = {
 			isa = PBXFileSystemSynchronizedGroupBuildPhaseMembershipExceptionSet;
 			attributesByRelativePath = {
 				Daemon/kordophoned = (CodeSignOnCopy, );
@@ -70,9 +70,9 @@
 		CD41F5992E5B8E7300E0027B /* kordophone2 */ = {
 			isa = PBXFileSystemSynchronizedRootGroup;
 			exceptions = (
-				CD41F5DA2E62850100E0027B /* Exceptions for "kordophone2" folder in "kordophone2" target */,
-				CD41F5DC2E62853800E0027B /* Exceptions for "kordophone2" folder in "Copy Files" phase from "kordophone2" target */,
-				CD41F5E12E62860700E0027B /* Exceptions for "kordophone2" folder in "Copy Files" phase from "kordophone2" target */,
+				CD41F5DA2E62850100E0027B /* Exceptions for "kordophone2" folder in "Kordophone" target */,
+				CD41F5DC2E62853800E0027B /* Exceptions for "kordophone2" folder in "Copy Files" phase from "Kordophone" target */,
+				CD41F5E12E62860700E0027B /* Exceptions for "kordophone2" folder in "Copy Files" phase from "Kordophone" target */,
 			);
 			path = kordophone2;
 			sourceTree = "<group>";
@@ -102,7 +102,7 @@
 		CD41F5982E5B8E7300E0027B /* Products */ = {
 			isa = PBXGroup;
 			children = (
-				CD41F5972E5B8E7300E0027B /* kordophone2.app */,
+				CD41F5972E5B8E7300E0027B /* Kordophone.app */,
 			);
 			name = Products;
 			sourceTree = "<group>";
@@ -110,9 +110,9 @@
 /* End PBXGroup section */
 
 /* Begin PBXNativeTarget section */
-		CD41F5962E5B8E7300E0027B /* kordophone2 */ = {
+		CD41F5962E5B8E7300E0027B /* Kordophone */ = {
 			isa = PBXNativeTarget;
-			buildConfigurationList = CD41F5A32E5B8E7400E0027B /* Build configuration list for PBXNativeTarget "kordophone2" */;
+			buildConfigurationList = CD41F5A32E5B8E7400E0027B /* Build configuration list for PBXNativeTarget "Kordophone" */;
 			buildPhases = (
 				CD41F5932E5B8E7300E0027B /* Sources */,
 				CD41F5942E5B8E7300E0027B /* Frameworks */,
@@ -127,12 +127,12 @@
 			fileSystemSynchronizedGroups = (
 				CD41F5992E5B8E7300E0027B /* kordophone2 */,
 			);
-			name = kordophone2;
+			name = Kordophone;
 			packageProductDependencies = (
 				CD41F5D22E62431D00E0027B /* KeychainAccess */,
 			);
 			productName = kordophone2;
-			productReference = CD41F5972E5B8E7300E0027B /* kordophone2.app */;
+			productReference = CD41F5972E5B8E7300E0027B /* Kordophone.app */;
 			productType = "com.apple.product-type.application";
 		};
 /* End PBXNativeTarget section */
@@ -167,7 +167,7 @@
 			projectDirPath = "";
 			projectRoot = "";
 			targets = (
-				CD41F5962E5B8E7300E0027B /* kordophone2 */,
+				CD41F5962E5B8E7300E0027B /* Kordophone */,
 			);
 		};
 /* End PBXProject section */
@@ -322,7 +322,7 @@
 				CODE_SIGN_STYLE = Automatic;
 				COMBINE_HIDPI_IMAGES = YES;
 				CURRENT_PROJECT_VERSION = 1;
-				DEVELOPMENT_TEAM = DQQH5H6GBD;
+				DEVELOPMENT_TEAM = 3SJALV9BQ7;
 				ENABLE_HARDENED_RUNTIME = NO;
 				ENABLE_PREVIEWS = YES;
 				GENERATE_INFOPLIST_FILE = YES;
@@ -349,7 +349,7 @@
 				CODE_SIGN_STYLE = Automatic;
 				COMBINE_HIDPI_IMAGES = YES;
 				CURRENT_PROJECT_VERSION = 1;
-				DEVELOPMENT_TEAM = DQQH5H6GBD;
+				DEVELOPMENT_TEAM = 3SJALV9BQ7;
 				ENABLE_HARDENED_RUNTIME = NO;
 				ENABLE_PREVIEWS = YES;
 				GENERATE_INFOPLIST_FILE = YES;
@@ -379,7 +379,7 @@
 			defaultConfigurationIsVisible = 0;
 			defaultConfigurationName = Release;
 		};
-		CD41F5A32E5B8E7400E0027B /* Build configuration list for PBXNativeTarget "kordophone2" */ = {
+		CD41F5A32E5B8E7400E0027B /* Build configuration list for PBXNativeTarget "Kordophone" */ = {
 			isa = XCConfigurationList;
 			buildConfigurations = (
 				CD41F5A42E5B8E7400E0027B /* Debug */,

osx/kordophone2.xcodeproj/xcshareddata/xcschemes/Kordophone App.xcscheme

@@ -0,0 +1,78 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<Scheme
+   LastUpgradeVersion = "1640"
+   version = "1.7">
+   <BuildAction
+      parallelizeBuildables = "YES"
+      buildImplicitDependencies = "YES"
+      buildArchitectures = "Automatic">
+      <BuildActionEntries>
+         <BuildActionEntry
+            buildForTesting = "YES"
+            buildForRunning = "YES"
+            buildForProfiling = "YES"
+            buildForArchiving = "YES"
+            buildForAnalyzing = "YES">
+            <BuildableReference
+               BuildableIdentifier = "primary"
+               BlueprintIdentifier = "CD41F5962E5B8E7300E0027B"
+               BuildableName = "Kordophone.app"
+               BlueprintName = "Kordophone"
+               ReferencedContainer = "container:kordophone2.xcodeproj">
+            </BuildableReference>
+         </BuildActionEntry>
+      </BuildActionEntries>
+   </BuildAction>
+   <TestAction
+      buildConfiguration = "Debug"
+      selectedDebuggerIdentifier = "Xcode.DebuggerFoundation.Debugger.LLDB"
+      selectedLauncherIdentifier = "Xcode.DebuggerFoundation.Launcher.LLDB"
+      shouldUseLaunchSchemeArgsEnv = "YES"
+      shouldAutocreateTestPlan = "YES">
+   </TestAction>
+   <LaunchAction
+      buildConfiguration = "Debug"
+      selectedDebuggerIdentifier = "Xcode.DebuggerFoundation.Debugger.LLDB"
+      selectedLauncherIdentifier = "Xcode.DebuggerFoundation.Launcher.LLDB"
+      launchStyle = "0"
+      useCustomWorkingDirectory = "NO"
+      ignoresPersistentStateOnLaunch = "NO"
+      debugDocumentVersioning = "YES"
+      debugServiceExtension = "internal"
+      allowLocationSimulation = "YES">
+      <BuildableProductRunnable
+         runnableDebuggingMode = "0">
+         <BuildableReference
+            BuildableIdentifier = "primary"
+            BlueprintIdentifier = "CD41F5962E5B8E7300E0027B"
+            BuildableName = "Kordophone.app"
+            BlueprintName = "Kordophone"
+            ReferencedContainer = "container:kordophone2.xcodeproj">
+         </BuildableReference>
+      </BuildableProductRunnable>
+   </LaunchAction>
+   <ProfileAction
+      buildConfiguration = "Release"
+      shouldUseLaunchSchemeArgsEnv = "YES"
+      savedToolIdentifier = ""
+      useCustomWorkingDirectory = "NO"
+      debugDocumentVersioning = "YES">
+      <BuildableProductRunnable
+         runnableDebuggingMode = "0">
+         <BuildableReference
+            BuildableIdentifier = "primary"
+            BlueprintIdentifier = "CD41F5962E5B8E7300E0027B"
+            BuildableName = "Kordophone.app"
+            BlueprintName = "Kordophone"
+            ReferencedContainer = "container:kordophone2.xcodeproj">
+         </BuildableReference>
+      </BuildableProductRunnable>
+   </ProfileAction>
+   <AnalyzeAction
+      buildConfiguration = "Debug">
+   </AnalyzeAction>
+   <ArchiveAction
+      buildConfiguration = "Release"
+      revealArchiveInOrganizer = "YES">
+   </ArchiveAction>
+</Scheme>

osx/kordophone2/App.swift

@@ -14,7 +14,14 @@ struct KordophoneApp: App
         WindowGroup {
             SplitView()
         }
-        
+        .commands {
+            TextEditingCommands()
+        }
+
+        WindowGroup(id: .transcriptWindow, for: Display.Conversation.self) { selectedConversation in
+            TranscriptWindowView(conversation: selectedConversation)
+        }
+
         Settings {
             PreferencesView()
         }
@@ -25,3 +32,42 @@ struct KordophoneApp: App
         print("Error: \(e.localizedDescription)")
     }
 }
+
+struct TranscriptWindowView: View
+{
+    @State private var transcriptViewModel = TranscriptView.ViewModel()
+    @State private var entryViewModel = MessageEntryView.ViewModel()
+    private let displayedConversation: Binding<Display.Conversation?>
+
+    public init(conversation: Binding<Display.Conversation?>) {
+        self.displayedConversation = conversation
+        transcriptViewModel.displayedConversation = conversation.wrappedValue
+        observeDisplayedConversationChanges()
+    }
+
+    private func observeDisplayedConversationChanges() {
+        withObservationTracking {
+            _ = displayedConversation.wrappedValue
+        } onChange: {
+            Task { @MainActor in
+                guard let displayedConversation = self.displayedConversation.wrappedValue else { return }
+                transcriptViewModel.displayedConversation = displayedConversation
+
+                observeDisplayedConversationChanges()
+            }
+        }
+    }
+
+    var body: some View {
+        VStack {
+            ConversationView(transcriptModel: $transcriptViewModel, entryModel: $entryViewModel)
+                .navigationTitle(displayedConversation.wrappedValue?.displayName ?? "Kordophone")
+                .selectedConversation(displayedConversation.wrappedValue)
+        }
+    }
+}
+
+extension String
+{
+    static let transcriptWindow = "TranscriptWindow"
+}

osx/kordophone2/ConversationListView.swift

@@ -11,9 +11,10 @@ struct ConversationListView: View
 {
     @Binding var model: ViewModel
     @Environment(\.xpcClient) private var xpcClient
-    
+    @Environment(\.openWindow) private var openWindow
+
     var body: some View {
-        List($model.conversations, selection: $model.selectedConversations) { conv in
+        List($model.conversations, selection: $model.selectedConversation) { conv in
             let isUnread = conv.wrappedValue.unreadCount > 0
             
             HStack(spacing: 0.0) {
@@ -64,14 +65,14 @@ struct ConversationListView: View
     class ViewModel
     {
         var conversations: [Display.Conversation]
-        var selectedConversations: Set<Display.Conversation.ID>
-        
+        var selectedConversation: Display.Conversation.ID?
+
         private var needsReload: Bool = true
         private let client = XPCClient()
         
         public init(conversations: [Display.Conversation] = []) {
             self.conversations = conversations
-            self.selectedConversations = Set()
+            self.selectedConversation = nil
             setNeedsReload()
         }
         
@@ -101,6 +102,11 @@ struct ConversationListView: View
                     .map { Display.Conversation(from: $0) }
                 
                 self.conversations = clientConversations
+
+                let unreadConversations = clientConversations.filter(\.isUnread)
+                await MainActor.run {
+                    NSApplication.shared.dockTile.badgeLabel = unreadConversations.isEmpty ? nil : "\(unreadConversations.count)"
+                }
             } catch {
                 print("Error reloading conversations: \(error)")
             }

osx/kordophone2/ConversationView.swift

@@ -14,7 +14,7 @@ struct ConversationView: View
     @Binding var entryModel: MessageEntryView.ViewModel
     
     var body: some View {
-        VStack {
+        VStack(spacing: 0.0) {
             TranscriptView(model: $transcriptModel)
             MessageEntryView(viewModel: $entryModel)
         }
@@ -23,5 +23,10 @@ struct ConversationView: View
             entryModel.handleDroppedProviders(providers)
             return true
         }
+
+        .onHover { isHovering in
+            guard isHovering else { return } 
+            transcriptModel.setNeedsMarkAsRead()
+        }
     }
 }

osx/kordophone2/MessageEntryView.swift

@@ -36,6 +36,7 @@ struct MessageEntryView: View
                         .font(.body)
                         .scrollDisabled(true)
                         .disabled(selectedConversation == nil)
+                        .id("messageEntry")
                 }
                 .padding(8.0)
                 .background {

osx/kordophone2/Models.swift

@@ -10,7 +10,7 @@ import XPC
 
 enum Display
 {
-    struct Conversation: Identifiable, Hashable
+    struct Conversation: Identifiable, Hashable, Codable
     {
         let id: String
         let name: String?
@@ -26,7 +26,11 @@ enum Display
         var isGroupChat: Bool {
             participants.count > 1
         }
-        
+
+        var isUnread: Bool {
+            unreadCount > 0
+        }
+
         init(from c: Serialized.Conversation) {
             self.id = c.guid
             self.name = c.displayName
@@ -111,7 +115,15 @@ enum Display
         var previewPath: String {
             data.previewPath
         }
-        
+
+        var isFullsizeDownloaded: Bool {
+            data.isDownloaded
+        }
+
+        var fullsizePath: String {
+            data.path
+        }
+
         init(from serialized: Serialized.Attachment, dateSent: Date, sender: Sender) {
             self.id = serialized.guid
             self.sender = sender

osx/kordophone2/SplitView.swift

@@ -15,7 +15,7 @@ struct SplitView: View
     
     private let xpcClient = XPCClient()
     private var selectedConversation: Display.Conversation? {
-        guard let id = conversationListModel.selectedConversations.first else { return nil }
+        guard let id = conversationListModel.selectedConversation else { return nil }
         return conversationListModel.conversations.first { $0.id == id }
     }
 
@@ -28,10 +28,10 @@ struct SplitView: View
             ConversationView(transcriptModel: $transcriptViewModel, entryModel: $entryViewModel)
                 .xpcClient(xpcClient)
                 .selectedConversation(selectedConversation)
-                .navigationTitle("Kordophone")
-                .navigationSubtitle(selectedConversation?.displayName ?? "")
-                .onChange(of: conversationListModel.selectedConversations) { oldValue, newValue in
-                    transcriptViewModel.displayedConversation = conversationListModel.conversations.first { $0.id == newValue.first }
+                .navigationTitle(selectedConversation?.displayName ?? "Kordophone")
+                .navigationSubtitle(selectedConversation?.participants.joined(separator: ", ") ?? "")
+                .onChange(of: conversationListModel.selectedConversation) { oldValue, newValue in
+                    transcriptViewModel.displayedConversation = conversationListModel.conversations.first { $0.id == newValue }
                 }
         }
     }

osx/kordophone2/Transcript/PreviewPanel.swift

@@ -0,0 +1,39 @@
+//
+//  PreviewPanel.swift
+//  Kordophone
+//
+//  Created by James Magahern on 9/12/25.
+//
+
+import AppKit
+import QuickLook
+import QuickLookUI
+
+internal class PreviewPanel
+{
+    static let shared = PreviewPanel()
+
+    private var displayedURL: URL? = nil
+    private var impl: QLPreviewPanel { QLPreviewPanel.shared() }
+
+    private init() {
+        impl.dataSource = self
+    }
+
+    public func show(url: URL) {
+        self.displayedURL = url
+        impl.makeKeyAndOrderFront(self)
+    }
+}
+
+
+extension PreviewPanel: QLPreviewPanelDataSource
+{
+    func numberOfPreviewItems(in panel: QLPreviewPanel!) -> Int {
+        1
+    }
+
+    func previewPanel(_ panel: QLPreviewPanel!, previewItemAt index: Int) -> (any QLPreviewItem)! {
+        return displayedURL! as NSURL
+    }
+}

osx/kordophone2/Transcript/TranscriptDisplayItemViews.swift

@@ -58,7 +58,7 @@ struct TextBubbleItemView: View
     let date: Date
     
     private var isFromMe: Bool { sender.isMe }
-    
+
     var body: some View {
         let bubbleColor: Color = isFromMe ? .blue : Color(NSColor(name: "grayish", dynamicProvider: { appearance in
             appearance.name == .darkAqua ? .darkGray : NSColor(white: 0.78, alpha: 1.0)
@@ -67,7 +67,7 @@ struct TextBubbleItemView: View
 
         BubbleView(sender: sender, date: date) {
             HStack {
-                Text(text)
+                Text(text.linkifiedAttributedString())
                     .foregroundStyle(textColor)
                     .multilineTextAlignment(.leading)
             }
@@ -75,6 +75,7 @@ struct TextBubbleItemView: View
             .padding(.horizontal, 16.0)
             .padding(.vertical, 10.0)
             .background(bubbleColor)
+            .textSelection(.enabled)
         }
     }
 }
@@ -89,7 +90,8 @@ struct ImageItemView: View
     @Environment(\.xpcClient) var xpcClient
     
     @State private var containerWidth: CGFloat? = nil
-    
+    @State private var isDownloadingFullAttachment: Bool = false
+
     private var aspectRatio: CGFloat {
         attachment.size?.aspectRatio ?? 1.0
     }
@@ -101,17 +103,36 @@ struct ImageItemView: View
     var body: some View {
         BubbleView(sender: sender, date: date) {
             let maxWidth = CGFloat.minimum(.imageMaxWidth, containerWidth ?? .imageMaxWidth)
-            if let img {
-                Image(nsImage: img)
-                    .resizable()
-                    .scaledToFit()
-                    .frame(maxWidth: maxWidth)
-            } else {
-                Rectangle()
-                    .fill(.gray.opacity(0.4))
-                    .frame(width: preferredWidth, height: preferredWidth / aspectRatio)
-                    .frame(maxWidth: maxWidth)
+
+            Group {
+                if let img {
+                    Image(nsImage: img)
+                        .resizable()
+                        .scaledToFit()
+                        .frame(maxWidth: maxWidth)
+                } else {
+                    Rectangle()
+                        .fill(.gray.opacity(0.4))
+                        .frame(width: preferredWidth, height: preferredWidth / aspectRatio)
+                        .frame(maxWidth: maxWidth)
+                }
             }
+
+            // Download indicator
+            .overlay {
+                if isDownloadingFullAttachment {
+                    ZStack {
+                        Rectangle()
+                            .fill(.black.opacity(0.2))
+
+                        ProgressView()
+                            .progressViewStyle(.circular)
+                    }
+                }
+            }
+        }
+        .onTapGesture(count: 2) {
+            openAttachment()
         }
         .onGeometryChange(for: CGFloat.self,
                           of: { $0.size.width },
@@ -135,6 +156,24 @@ struct ImageItemView: View
             }
         }
     }
+
+    private func openAttachment() {
+        Task {
+            var path = attachment.fullsizePath
+            if !attachment.isFullsizeDownloaded {
+                isDownloadingFullAttachment = true
+                try await xpcClient.downloadAttachment(attachmentId: attachment.id, preview: false, awaitCompletion: true)
+
+                // Need to re-fetch this -- the extension may have changed.
+                let info = try await xpcClient.getAttachmentInfo(attachmentId: attachment.id)
+                path = info.path
+
+                isDownloadingFullAttachment = false
+            }
+
+            PreviewPanel.shared.show(url: URL(filePath: path))
+        }
+    }
 }
 
 struct PlaceholderImageItemView: View
@@ -219,14 +258,16 @@ struct SenderAttributionView: View
     }
 }
 
-fileprivate extension CGFloat {
+fileprivate extension CGFloat
+{
     static let dominantCornerRadius = 16.0
     static let minorCornerRadius = 4.0
     static let minimumBubbleHorizontalPadding = 80.0
     static let imageMaxWidth = 380.0
 }
 
-fileprivate extension CGSize {
+fileprivate extension CGSize
+{
     var aspectRatio: CGFloat { width / height }
 }
 
@@ -239,3 +280,28 @@ fileprivate func preferredBubbleWidth(forAttachmentSize attachmentSize: CGSize?,
         return 200.0 // fallback
     }
 }
+
+fileprivate extension String
+{
+    func linkifiedAttributedString() -> AttributedString {
+        var attributed = AttributedString(self)
+        guard let detector = try? NSDataDetector(types: NSTextCheckingResult.CheckingType.link.rawValue) else {
+            return attributed
+        }
+
+        let nsText = self as NSString
+        let fullRange = NSRange(location: 0, length: nsText.length)
+        detector.enumerateMatches(in: self, options: [], range: fullRange) { result, _, _ in
+            guard let result, let url = result.url,
+                  let swiftRange = Range(result.range, in: self),
+                  let start = AttributedString.Index(swiftRange.lowerBound, within: attributed),
+                  let end = AttributedString.Index(swiftRange.upperBound, within: attributed) else { return }
+
+            attributed[start..<end].link = url
+            attributed[start..<end].foregroundColor = NSColor.textColor
+            attributed[start..<end].underlineStyle = .single
+        }
+
+        return attributed
+    }
+}

osx/kordophone2/Transcript/TranscriptDisplayItems.swift

@@ -10,7 +10,7 @@ import SwiftUI
 
 extension TranscriptView.ViewModel
 {
-    internal func rebuildDisplayItems(animated: Bool = false) {
+    internal func rebuildDisplayItems(animated: Bool = false, completion: () -> Void = {}) {
         var displayItems: [DisplayItem] = []
         var lastDate: Date = .distantPast
         var lastSender: Display.Sender? = nil
@@ -53,6 +53,7 @@ extension TranscriptView.ViewModel
         let animation: Animation? = animated ? .default : nil
         withAnimation(animation) {
             self.displayItems = displayItems
+            completion()
         }
     }
 }

osx/kordophone2/Transcript/TranscriptView.swift

@@ -12,25 +12,57 @@ struct TranscriptView: View
     @Binding var model: ViewModel
     
     @Environment(\.xpcClient) private var xpcClient
-    
+
+    init(model: Binding<ViewModel>) {
+        self._model = model
+    }
+
     var body: some View {
-        ScrollView {
-            LazyVStack(spacing: 6.0) {
-                ForEach($model.displayItems.reversed()) { item in
-                    displayItemView(item.wrappedValue)
-                        .id(item.id)
-                        .scaleEffect(CGSize(width: 1.0, height: -1.0))
-                        .transition(
-                            .push(from: .top)
-                            .combined(with: .opacity)
-                        )
+        ScrollViewReader { proxy in
+            ScrollView {
+                // For resetting scroll position to the "bottom"
+                EmptyView()
+                    .id(ViewID.bottomAnchor)
+
+                LazyVStack(spacing: 6.0) {
+                    ForEach($model.displayItems.reversed()) { item in
+                        displayItemView(item.wrappedValue)
+                            .id(item.id)
+                            .scaleEffect(CGSize(width: 1.0, height: -1.0))
+                            .transition(
+                                .push(from: .top)
+                                .combined(with: .opacity)
+                            )
+                    }
+                }
+                .padding()
+            }
+
+            // Flip vertically so newest messages are at the bottom.
+            .scaleEffect(CGSize(width: 1.0, height: -1.0))
+
+            // Watch for xpc events
+            .task { await watchForMessageListChanges() }
+
+            // On conversation change, reload displayed messages and mark as read.
+            .onChange(of: model.displayedConversation) { oldValue, newValue in
+                Task {
+                    guard oldValue != newValue else { return }
+
+                    // Reload NOW
+                    await model.reloadMessages(animated: false) {
+                        // Once that's done, scroll to the "bottom" (actually top)
+                        proxy.scrollTo(ViewID.bottomAnchor, anchor: .top)
+                    }
+                }
+
+                Task.detached {
+                    // Mark as read on server, and trigger a sync.
+                    await model.markAsRead()
+                    await model.triggerSync()
                 }
             }
-            .padding()
         }
-        .scaleEffect(CGSize(width: 1.0, height: -1.0))
-        .id(model.displayedConversation?.id)
-        .task { await watchForMessageListChanges() }
     }
     
     private func watchForMessageListChanges() async {
@@ -79,22 +111,31 @@ struct TranscriptView: View
             }
         }
     }
-    
+
     // MARK: - Types
-    
+
+    enum ViewID: String
+    {
+        case bottomAnchor
+    }
+
+    // MARK: - View Model
+
     @Observable
     class ViewModel
     {
         var displayItems: [DisplayItem] = []
         var displayedConversation: Display.Conversation? = nil
-        
+
         internal var needsReload: NeedsReload = .no
         internal var messages: [Display.Message]
         internal let client = XPCClient()
 
+        private var needsMarkAsRead: Bool = false
+        private var lastMarkAsRead: Date = .now
+
         init(messages: [Display.Message] = []) {
             self.messages = messages
-            observeDisplayedConversation()
             rebuildDisplayItems()
         }
         
@@ -106,31 +147,28 @@ struct TranscriptView: View
             needsReload = .yes(animated)
             Task { @MainActor [weak self] in
                 guard let self else { return }
-                await reloadMessages()
+                await reloadIfNeeded()
             }
         }
-        
+
+        func setNeedsMarkAsRead() {
+            guard needsMarkAsRead == false else { return }
+            guard Date.now.timeIntervalSince(lastMarkAsRead) > 5.0 else { return }
+
+            needsMarkAsRead = true
+            Task { @MainActor [weak self] in
+                guard let self else { return }
+                await markAsRead()
+                needsMarkAsRead = false
+                lastMarkAsRead = .now
+            }
+        }
+
         func attachmentDownloaded(id: String) {
             // TODO: should be smarter here
             setNeedsReload(animated: false)
         }
         
-        private func observeDisplayedConversation() {
-            withObservationTracking {
-                _ = displayedConversation
-            } onChange: {
-                Task { @MainActor [weak self] in
-                    guard let self else { return }
-                    
-                    await markAsRead()
-                    await triggerSync()
-                    
-                    setNeedsReload(animated: false)
-                    observeDisplayedConversation()
-                }
-            }
-        }
-        
         func markAsRead() async {
             guard let displayedConversation else { return }
             
@@ -150,11 +188,15 @@ struct TranscriptView: View
                 print("Error triggering sync: \(error)")
             }
         }
-        
-        private func reloadMessages() async {
+
+        func reloadIfNeeded(completion: () -> Void = {}) async {
             guard case .yes(let animated) = needsReload else { return }
             needsReload = .no
-            
+
+            await reloadMessages(animated: animated, completion: completion)
+        }
+
+        func reloadMessages(animated: Bool, completion: () -> Void) async {
             guard let displayedConversation else { return }
             
             do {
@@ -167,8 +209,10 @@ struct TranscriptView: View
                 // Only animate for incoming messages.
                 let shouldAnimate = (newIds.count == 1)
 
-                self.messages = clientMessages
-                self.rebuildDisplayItems(animated: animated && shouldAnimate)
+                await MainActor.run {
+                    self.messages = clientMessages
+                    self.rebuildDisplayItems(animated: animated && shouldAnimate, completion: completion)
+                }
             } catch {
                 print("Message fetch error: \(error)")
             }

osx/kordophone2/XPC/XPCClient.swift

@@ -146,13 +146,33 @@ final class XPCClient
         guard let reply = try await sendSync(req), xpc_get_type(reply) == XPC_TYPE_DICTIONARY else { throw Error.typeError }
     }
 
-    public func downloadAttachment(attachmentId: String, preview: Bool) async throws {
+    public func getAttachmentInfo(attachmentId: String) async throws -> AttachmentInfo {
+        var args: [String: xpc_object_t] = [:]
+        args["attachment_id"] = xpcString(attachmentId)
+
+        let req = makeRequest(method: "GetAttachmentInfo", arguments: args)
+        guard let reply = try await sendSync(req), xpc_get_type(reply) == XPC_TYPE_DICTIONARY else { throw Error.typeError }
+        
+        return AttachmentInfo(
+            path: reply["path"] ?? "",
+            previewPath: reply["preview_path"] ?? "",
+            isDownloaded: reply["is_downloaded"] ?? false,
+            isPreviewDownloaded: reply["is_preview_downloaded"] ?? false
+        )
+    }
+
+    public func downloadAttachment(attachmentId: String, preview: Bool, awaitCompletion: Bool = false) async throws {
         var args: [String: xpc_object_t] = [:]
         args["attachment_id"] = xpcString(attachmentId)
         args["preview"] = xpcString(preview ? "true" : "false")
 
         let req = makeRequest(method: "DownloadAttachment", arguments: args)
         _ = try await sendSync(req)
+
+        if awaitCompletion {
+            // Wait for downloaded event
+            let _ = await eventStream().first { $0 == .attachmentDownloaded(attachmentId: attachmentId) }
+        }
     }
     
     public func uploadAttachment(path: String) async throws -> String {
@@ -200,6 +220,14 @@ final class XPCClient
     }
     
     // MARK: - Types
+
+    struct AttachmentInfo: Decodable
+    {
+        let path: String
+        let previewPath: String
+        let isDownloaded: Bool
+        let isPreviewDownloaded: Bool
+    }
     
     enum Error: Swift.Error
     {
@@ -209,7 +237,7 @@ final class XPCClient
         case connectionError
     }
     
-    enum Signal
+    enum Signal: Equatable
     {
         case conversationsUpdated
         case messagesUpdated(conversationId: String)

osx/kordophone2/XPC/XPCConvertible.swift

@@ -76,6 +76,22 @@ extension Array: XPCConvertible where Element: XPCConvertible
     }
 }
 
+extension Bool: XPCConvertible
+{
+    static func fromXPC(_ value: xpc_object_t) -> Bool? {
+        if xpc_get_type(value) == XPC_TYPE_BOOL {
+            return xpc_bool_get_value(value)
+        }
+
+        if xpc_get_type(value) == XPC_TYPE_STRING {
+            guard let cstr = xpc_string_get_string_ptr(value) else { return nil }
+            return strcmp(cstr, "true") == 0
+        }
+
+        return nil
+    }
+}
+
 extension xpc_object_t
 {
     func getObject(_ key: String) -> xpc_object_t? {