Add event bus for live UI updates

Introduces a system for broadcasting real-time events to the web UI.
This enables features like automatic page reloads when case data changes
in the background, improving the user experience by keeping the UI
synchronized with the backend.

Key components:
- `events` module: Contains the `CaseEventKind` enum, `CaseEvent`
  struct, and `EventSender` type for managing the broadcast channel.
- SSE endpoint (`/web/events`): Streams events to connected browsers.
- Client-side JavaScript: Listens for events and triggers debounced page
  reloads.
- Integration points: Workers and route handlers now emit events when
  relevant state changes occur.

Cargo.lock

@@ -1210,6 +1210,7 @@ dependencies = [
  "doctate-common",
  "dotenvy",
  "filetime",
+ "futures-util",
  "pulldown-cmark",
  "rand 0.8.6",
  "reqwest",
@@ -1221,6 +1222,7 @@ dependencies = [
  "tempfile",
  "time",
  "tokio",
+ "tokio-stream",
  "toml",
  "toml_edit 0.25.11+spec-1.1.0",
  "tower",
@@ -4294,6 +4296,18 @@ dependencies = [
  "tokio",
 ]
 
+[[package]]
+name = "tokio-stream"
+version = "0.1.18"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "32da49809aab5c3bc678af03902d4ccddea2a87d028d86392a4b1560c6906c70"
+dependencies = [
+ "futures-core",
+ "pin-project-lite",
+ "tokio",
+ "tokio-util",
+]
+
 [[package]]
 name = "tokio-util"
 version = "0.7.18"

server/Cargo.toml

@@ -20,6 +20,8 @@ serde.workspace = true
 serde_json.workspace = true
 toml.workspace = true
 tower-http = { version = "0.6", features = ["limit", "trace"] }
+tokio-stream = { version = "0.1", features = ["sync"] }
+futures-util = "0.3"
 rand = "0.8"
 bcrypt = "0.15"
 axum-extra = { version = "0.12", features = ["cookie", "form"] }

server/src/analyze/worker.rs

@@ -7,6 +7,7 @@ use tracing::{error, info, warn};
 
 use super::{ANALYSIS_INPUT_FILE, AnalysisInput, AnalyzeReceiver, DOCUMENT_FILE, llm, prompt};
 use crate::config::Config;
+use crate::events::{self, CaseEventKind, EventSender};
 use crate::gazetteer::Gazetteer;
 use crate::{BusyGuard, WorkerBusy};
 
@@ -19,13 +20,14 @@ pub async fn run(
     client: reqwest::Client,
     worker_busy: WorkerBusy,
     vocab: Arc<Gazetteer>,
+    events_tx: EventSender,
 ) {
     info!(vocab_entries = vocab.len(), "Analyze worker started");
     let timeout = Duration::from_secs(config.llm_timeout_seconds);
 
     while let Some(job) = rx.recv().await {
         let _guard = BusyGuard::new(worker_busy.clone());
-        process(&job.case_dir, &config, &client, &vocab, timeout).await;
+        process(&job.case_dir, &config, &client, &vocab, timeout, &events_tx).await;
     }
 
     warn!("Analyze worker stopped (channel closed)");
@@ -37,6 +39,7 @@ async fn process(
     client: &reqwest::Client,
     vocab: &Gazetteer,
     timeout: Duration,
+    events_tx: &EventSender,
 ) {
     let input_path = case_dir.join(ANALYSIS_INPUT_FILE);
     let document_path = case_dir.join(DOCUMENT_FILE);
@@ -72,6 +75,12 @@ async fn process(
             warn!(path = %input_path.display(), error = %e, "removing analysis input failed");
         }
         info!(case = %case_dir.display(), "analysis done (stub, no recordings)");
+        events::emit(
+            events_tx,
+            events::user_slug_of(case_dir),
+            events::case_id_of(case_dir),
+            CaseEventKind::DocumentReady,
+        );
         return;
     }
 
@@ -134,6 +143,13 @@ async fn process(
         bytes = document.len(),
         "analysis done"
     );
+
+    events::emit(
+        events_tx,
+        events::user_slug_of(case_dir),
+        events::case_id_of(case_dir),
+        CaseEventKind::DocumentReady,
+    );
 }
 
 /// Write `bytes` to `path` atomically: write to `<path>.tmp`, fsync, rename.

server/src/events.rs

@@ -0,0 +1,143 @@
+//! Broadcast bus for UI live-update events.
+//!
+//! The web UI subscribes via the `/web/events` SSE route (see
+//! `routes::events`). Background workers and request handlers call
+//! [`emit`] whenever they mutate case data on disk; subscribed browsers
+//! then schedule a debounced reload. The filesystem remains the source
+//! of truth — these events are pure triggers, never state.
+
+use std::path::Path;
+
+use serde::Serialize;
+use tokio::sync::broadcast;
+
+/// Broadcast capacity. Oversized on purpose: even a "bulk analyze 50 cases"
+/// operation emits at most ~5 events per case (Queued/Transcript/Oneliner/
+/// Document/Done), well under 256. Subscribers more than 256 events behind
+/// receive a `Lagged(n)` on their next recv; they are re-synced by their
+/// next page reload anyway, so skipping is safe.
+pub const EVENT_CHANNEL_CAPACITY: usize = 256;
+
+/// All UI-observable mutations on a case. Serialized as a `kind` field
+/// in snake_case so the client can filter with e.g. `evt.kind === "transcript_ready"`.
+#[derive(Debug, Clone, Serialize)]
+#[serde(rename_all = "snake_case")]
+pub enum CaseEventKind {
+    RecordingUploaded,
+    TranscriptReady,
+    TranscriptFailed,
+    OnelinerUpdated,
+    AnalysisQueued,
+    DocumentReady,
+    CaseDeleted,
+    CaseRestored,
+    CaseReset,
+}
+
+/// A single state-change notification. Clone is required because
+/// `tokio::sync::broadcast` hands every subscriber its own copy.
+#[derive(Debug, Clone, Serialize)]
+pub struct CaseEvent {
+    pub user_slug: String,
+    pub case_id: String,
+    pub kind: CaseEventKind,
+    pub ts: String,
+}
+
+/// Multi-producer, multi-consumer sender shared via `AppState`.
+pub type EventSender = broadcast::Sender<CaseEvent>;
+
+/// Construct a fresh broadcast channel sized with [`EVENT_CHANNEL_CAPACITY`].
+/// The unused receiver is dropped — subscribers call
+/// [`EventSender::subscribe`] on the returned sender.
+pub fn channel() -> EventSender {
+    let (tx, _rx) = broadcast::channel(EVENT_CHANNEL_CAPACITY);
+    tx
+}
+
+/// Best-effort publish. `broadcast::Sender::send` returns `Err` only when
+/// there are zero subscribers — which is the normal state before any
+/// browser connects — so we swallow the error silently. Never blocks.
+pub fn emit(
+    tx: &EventSender,
+    user_slug: impl Into<String>,
+    case_id: impl Into<String>,
+    kind: CaseEventKind,
+) {
+    let _ = tx.send(CaseEvent {
+        user_slug: user_slug.into(),
+        case_id: case_id.into(),
+        kind,
+        ts: doctate_common::now_rfc3339(),
+    });
+}
+
+/// Extract `<user_slug>` from a `<data_path>/<user_slug>/<case_id>/` path.
+/// Returns the empty string on malformed input. Empty slugs still flow
+/// through the admin filter (admins see them) but are filtered out for
+/// non-admins, which is the safe default.
+pub fn user_slug_of(case_dir: &Path) -> String {
+    case_dir
+        .parent()
+        .and_then(|p| p.file_name())
+        .and_then(|s| s.to_str())
+        .map(str::to_owned)
+        .unwrap_or_default()
+}
+
+/// Extract `<case_id>` from a `<data_path>/<user_slug>/<case_id>/` path.
+/// Returns the empty string on malformed input.
+pub fn case_id_of(case_dir: &Path) -> String {
+    case_dir
+        .file_name()
+        .and_then(|s| s.to_str())
+        .map(str::to_owned)
+        .unwrap_or_default()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn two_subscribers_both_receive() {
+        let tx = channel();
+        let mut a = tx.subscribe();
+        let mut b = tx.subscribe();
+        emit(&tx, "alice", "c1", CaseEventKind::RecordingUploaded);
+        let ea = a.recv().await.unwrap();
+        let eb = b.recv().await.unwrap();
+        assert_eq!(ea.user_slug, "alice");
+        assert_eq!(eb.case_id, "c1");
+        assert!(matches!(ea.kind, CaseEventKind::RecordingUploaded));
+    }
+
+    #[tokio::test]
+    async fn slow_subscriber_sees_lagged() {
+        let tx = channel();
+        let mut rx = tx.subscribe();
+        // Fill the channel past capacity without recv'ing.
+        for _ in 0..(EVENT_CHANNEL_CAPACITY + 5) {
+            emit(&tx, "u", "c", CaseEventKind::TranscriptReady);
+        }
+        let err = rx.recv().await.unwrap_err();
+        assert!(matches!(err, broadcast::error::RecvError::Lagged(_)));
+    }
+
+    #[test]
+    fn user_slug_extracted_from_case_dir() {
+        let p = Path::new("/data/alice/550e8400-e29b-41d4-a716-446655440000");
+        assert_eq!(user_slug_of(p), "alice");
+    }
+
+    #[test]
+    fn user_slug_empty_on_root_path() {
+        assert_eq!(user_slug_of(Path::new("/")), "");
+    }
+
+    #[test]
+    fn emit_with_no_subscribers_does_not_panic() {
+        let tx = channel();
+        emit(&tx, "u", "c", CaseEventKind::DocumentReady);
+    }
+}

server/src/lib.rs

@@ -3,6 +3,7 @@ pub mod auth;
 pub mod case_id;
 pub mod config;
 pub mod error;
+pub mod events;
 pub mod gazetteer;
 pub mod magic_link;
 pub mod models;
@@ -68,7 +69,7 @@ pub struct PipelineState {
 }
 
 /// Shared application state. Clonable — all fields are cheap to clone
-/// (`Arc` and `mpsc::Sender`).
+/// (`Arc` and `mpsc::Sender` / `broadcast::Sender`).
 #[derive(Clone)]
 pub struct AppState {
     pub config: Arc<Config>,
@@ -78,6 +79,7 @@ pub struct AppState {
     pub magic_link_store: MagicLinkStore,
     pub analyze_busy: AnalyzeBusy,
     pub transcribe_busy: TranscribeBusy,
+    pub events_tx: events::EventSender,
 }
 
 impl FromRef<AppState> for Arc<Config> {
@@ -133,6 +135,12 @@ impl FromRef<AppState> for PipelineState {
     }
 }
 
+impl FromRef<AppState> for events::EventSender {
+    fn from_ref(state: &AppState) -> Self {
+        state.events_tx.clone()
+    }
+}
+
 /// Test/simple entrypoint: jobs pushed into either channel are dropped
 /// because the receivers are not retained. Use [`create_router_with_state`]
 /// from `main.rs` where real workers own the receivers.
@@ -147,6 +155,7 @@ pub fn create_router(config: Arc<Config>) -> Router {
         magic_link_store: magic_link::new_store(),
         analyze_busy: AnalyzeBusy(Arc::new(AtomicBool::new(false))),
         transcribe_busy: TranscribeBusy(Arc::new(AtomicBool::new(false))),
+        events_tx: events::channel(),
     })
 }
 

server/src/main.rs

@@ -11,6 +11,7 @@ use tracing_subscriber::util::SubscriberInitExt;
 use doctate_server::AppState;
 use doctate_server::analyze;
 use doctate_server::config::Config;
+use doctate_server::events;
 use doctate_server::gazetteer::{Gazetteer, SpellbookDict};
 use doctate_server::transcribe;
 
@@ -110,6 +111,11 @@ async fn main() {
         }
     });
 
+    // Live-update bus: background workers and handlers publish case
+    // events here; the SSE route at `/web/events` fans them out to
+    // subscribed browsers.
+    let events_tx = events::channel();
+
     // Transcription pipeline: channel + worker + recovery scan.
     let (transcribe_tx, transcribe_rx) = transcribe::channel();
     let transcribe_busy: doctate_server::WorkerBusy =
@@ -120,6 +126,7 @@ async fn main() {
         http_client.clone(),
         transcribe_busy.clone(),
         vocab.clone(),
+        events_tx.clone(),
     ));
     {
         let tx = transcribe_tx.clone();
@@ -127,12 +134,15 @@ async fn main() {
         let client = http_client.clone();
         let cfg = config.clone();
         let vocab = vocab.clone();
+        let events = events_tx.clone();
         tokio::spawn(async move {
             transcribe::recovery::scan_and_enqueue(&data_path, &tx).await;
             // Then catch up oneliners for cases that crashed between
             // transcript-write and oneliner-write in a previous run.
-            transcribe::recovery::regenerate_missing_oneliners(&data_path, &client, &cfg, &vocab)
-                .await;
+            transcribe::recovery::regenerate_missing_oneliners(
+                &data_path, &client, &cfg, &vocab, &events,
+            )
+            .await;
         });
     }
 
@@ -146,6 +156,7 @@ async fn main() {
         http_client.clone(),
         analyze_busy.clone(),
         vocab.clone(),
+        events_tx.clone(),
     ));
     {
         let tx = analyze_tx.clone();
@@ -163,6 +174,7 @@ async fn main() {
         magic_link_store: doctate_server::magic_link::new_store(),
         analyze_busy: doctate_server::AnalyzeBusy(analyze_busy),
         transcribe_busy: doctate_server::TranscribeBusy(transcribe_busy),
+        events_tx,
     };
 
     let addr = format!("0.0.0.0:{}", config.server_port);

server/src/routes/bulk.rs

@@ -12,6 +12,7 @@ use crate::analyze::{ANALYSIS_INPUT_FILE, AnalyzeJob, AnalyzeSender, DOCUMENT_FI
 use crate::auth::AuthenticatedWebUser;
 use crate::config::Config;
 use crate::error::AppError;
+use crate::events::{self, CaseEventKind, EventSender};
 use crate::paths::{DeleteMarker, write_delete_marker};
 use crate::routes::case_actions::{
     build_analysis_input, reset_case_artefacts, write_input_create_new,
@@ -34,21 +35,30 @@ pub async fn handle_bulk_action(
     user: AuthenticatedWebUser,
     State(config): State<Arc<Config>>,
     State(analyze_tx): State<AnalyzeSender>,
+    State(events_tx): State<EventSender>,
     Form(form): Form<BulkForm>,
 ) -> Result<Redirect, AppError> {
     let user_root = config.data_path.join(&user.slug);
 
     match form.action.as_str() {
         "analyze" => {
-            bulk_analyze(&config, &analyze_tx, &user.slug, &user_root, &form.case_ids).await
+            bulk_analyze(
+                &config,
+                &analyze_tx,
+                &events_tx,
+                &user.slug,
+                &user_root,
+                &form.case_ids,
+            )
+            .await
         }
-        "delete" => bulk_delete(&user.slug, &user_root, &form.case_ids).await,
+        "delete" => bulk_delete(&events_tx, &user.slug, &user_root, &form.case_ids).await,
         "reset" => {
             if !user.is_admin() {
                 warn!(slug = %user.slug, "bulk-reset: not admin");
                 return Err(AppError::Forbidden("Nur für Admins".into()));
             }
-            bulk_reset(&user.slug, &user_root, &form.case_ids).await
+            bulk_reset(&events_tx, &user.slug, &user_root, &form.case_ids).await
         }
         other => {
             warn!(slug = %user.slug, action = %other, "bulk: unknown action");
@@ -62,6 +72,7 @@ pub async fn handle_bulk_action(
 async fn bulk_analyze(
     config: &Config,
     analyze_tx: &AnalyzeSender,
+    events_tx: &EventSender,
     slug: &str,
     user_root: &std::path::Path,
     case_ids: &[String],
@@ -117,12 +128,18 @@ async fn bulk_analyze(
         {
             warn!(slug = %slug, case_id = %case_id, "bulk-analyze: send failed (worker gone)");
         }
+        events::emit(events_tx, slug, case_id, CaseEventKind::AnalysisQueued);
         ok += 1;
     }
     info!(slug = %slug, ok, skipped, "bulk-analyze completed");
 }
 
-async fn bulk_delete(slug: &str, user_root: &std::path::Path, case_ids: &[String]) {
+async fn bulk_delete(
+    events_tx: &EventSender,
+    slug: &str,
+    user_root: &std::path::Path,
+    case_ids: &[String],
+) {
     // One batch UUID + one timestamp shared across the entire bulk action,
     // so undo can restore exactly this group.
     let batch = uuid::Uuid::new_v4();
@@ -155,12 +172,18 @@ async fn bulk_delete(slug: &str, user_root: &std::path::Path, case_ids: &[String
             skipped += 1;
             continue;
         }
+        events::emit(events_tx, slug, case_id, CaseEventKind::CaseDeleted);
         ok += 1;
     }
     info!(slug = %slug, batch = %batch, ok, skipped, "bulk-delete completed");
 }
 
-async fn bulk_reset(slug: &str, user_root: &std::path::Path, case_ids: &[String]) {
+async fn bulk_reset(
+    events_tx: &EventSender,
+    slug: &str,
+    user_root: &std::path::Path,
+    case_ids: &[String],
+) {
     let mut ok = 0usize;
     let mut skipped = 0usize;
     for case_id in case_ids {
@@ -179,6 +202,7 @@ async fn bulk_reset(slug: &str, user_root: &std::path::Path, case_ids: &[String]
             skipped += 1;
             continue;
         }
+        events::emit(events_tx, slug, case_id, CaseEventKind::CaseReset);
         ok += 1;
     }
     info!(slug = %slug, ok, skipped, "bulk-reset completed");

server/src/routes/case_actions.rs

@@ -18,6 +18,7 @@ use crate::auth::AuthenticatedWebUser;
 use crate::case_id::CaseIdPath;
 use crate::config::Config;
 use crate::error::AppError;
+use crate::events::{self, CaseEventKind, EventSender};
 use crate::paths::{DELETE_MARKER, DeleteMarker, read_delete_marker, write_delete_marker};
 use crate::routes::user_web::locate_case_or_404;
 
@@ -32,6 +33,7 @@ pub async fn handle_analyze_case(
     user: AuthenticatedWebUser,
     State(config): State<Arc<Config>>,
     State(analyze_tx): State<AnalyzeSender>,
+    State(events_tx): State<EventSender>,
     CaseIdPath(case_id): CaseIdPath,
 ) -> Result<Redirect, AppError> {
     if !config.llm_configured() {
@@ -76,6 +78,13 @@ pub async fn handle_analyze_case(
         "analyze requested; analysis enqueued"
     );
 
+    events::emit(
+        &events_tx,
+        &user.slug,
+        case_id.to_string(),
+        CaseEventKind::AnalysisQueued,
+    );
+
     Ok(Redirect::to("/web/cases"))
 }
 
@@ -242,6 +251,7 @@ pub(crate) async fn reset_case_artefacts(case_dir: &Path) -> std::io::Result<()>
 pub async fn handle_delete_case(
     user: AuthenticatedWebUser,
     State(config): State<Arc<Config>>,
+    State(events_tx): State<EventSender>,
     CaseIdPath(case_id): CaseIdPath,
 ) -> Result<Redirect, AppError> {
     let user_root = config.data_path.join(&user.slug);
@@ -260,6 +270,13 @@ pub async fn handle_delete_case(
         "case soft-deleted"
     );
 
+    events::emit(
+        &events_tx,
+        &user.slug,
+        case_id.to_string(),
+        CaseEventKind::CaseDeleted,
+    );
+
     Ok(Redirect::to("/web/cases"))
 }
 
@@ -273,6 +290,7 @@ pub async fn handle_delete_case(
 pub async fn handle_reset_case(
     user: AuthenticatedWebUser,
     State(config): State<Arc<Config>>,
+    State(events_tx): State<EventSender>,
     CaseIdPath(case_id): CaseIdPath,
 ) -> Result<Redirect, AppError> {
     if !user.is_admin() {
@@ -287,6 +305,12 @@ pub async fn handle_reset_case(
         .map_err(|e| AppError::Internal(format!("reset: {e}")))?;
 
     info!(slug = %user.slug, case_id = %case_id, "case reset (admin)");
+    events::emit(
+        &events_tx,
+        &user.slug,
+        case_id.to_string(),
+        CaseEventKind::CaseReset,
+    );
     Ok(Redirect::to("/web/cases"))
 }
 
@@ -299,6 +323,7 @@ pub async fn handle_reset_case(
 pub async fn handle_undo_delete(
     user: AuthenticatedWebUser,
     State(config): State<Arc<Config>>,
+    State(events_tx): State<EventSender>,
 ) -> Result<Redirect, AppError> {
     let user_root = config.data_path.join(&user.slug);
 
@@ -308,7 +333,7 @@ pub async fn handle_undo_delete(
         return Ok(Redirect::to("/web/cases"));
     };
 
-    let restored = restore_batch(&user_root, batch).await;
+    let restored = restore_batch(&user_root, batch, &events_tx, &user.slug).await;
     info!(slug = %user.slug, batch = %batch, restored, "undo-delete completed");
 
     Ok(Redirect::to("/web/cases"))
@@ -363,8 +388,14 @@ async fn latest_delete_batch(user_root: &Path) -> Option<(uuid::Uuid, String)> {
 }
 
 /// Remove `.deleted` markers from every case whose marker carries `batch`.
-/// Returns the number of restored cases.
-async fn restore_batch(user_root: &Path, batch: uuid::Uuid) -> usize {
+/// Returns the number of restored cases. Emits a `CaseRestored` event per
+/// case so subscribed browsers can update their listings.
+async fn restore_batch(
+    user_root: &Path,
+    batch: uuid::Uuid,
+    events_tx: &EventSender,
+    user_slug: &str,
+) -> usize {
     let mut count = 0;
     let mut entries = match tokio::fs::read_dir(user_root).await {
         Ok(r) => r,
@@ -385,6 +416,12 @@ async fn restore_batch(user_root: &Path, batch: uuid::Uuid) -> usize {
             warn!(case_dir = ?case_path, error = %e, "failed to remove .deleted marker");
             continue;
         }
+        events::emit(
+            events_tx,
+            user_slug,
+            events::case_id_of(&case_path),
+            CaseEventKind::CaseRestored,
+        );
         count += 1;
     }
     count

server/src/routes/events.rs

@@ -0,0 +1,55 @@
+//! SSE endpoint streaming case events to the web UI.
+//!
+//! One long-lived connection per browser tab. Non-admins only see events
+//! for their own user_slug; admins see events across all users
+//! (useful for support/monitoring). A 15s keep-alive comment keeps idle
+//! connections alive through reverse proxies / NAT.
+
+use std::convert::Infallible;
+use std::time::Duration;
+
+use axum::extract::State;
+use axum::response::sse::{Event, KeepAlive, Sse};
+use futures_util::stream::Stream;
+use tokio_stream::StreamExt;
+use tokio_stream::wrappers::BroadcastStream;
+use tokio_stream::wrappers::errors::BroadcastStreamRecvError;
+
+use crate::auth::AuthenticatedWebUser;
+use crate::events::EventSender;
+
+/// `GET /web/events` — `text/event-stream`, one connection per tab.
+pub async fn handle_events(
+    user: AuthenticatedWebUser,
+    State(events_tx): State<EventSender>,
+) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
+    let rx = events_tx.subscribe();
+    let is_admin = user.is_admin();
+    let slug = user.slug.clone();
+
+    let stream = BroadcastStream::new(rx).filter_map(move |res| match res {
+        Ok(evt) => {
+            // Non-admins only see events for their own user_slug.
+            if !is_admin && evt.user_slug != slug {
+                return None;
+            }
+            // Serialization of `CaseEvent` is infallible in practice (plain
+            // struct with String + enum fields); on the unexpected error
+            // path we drop the event rather than tear down the stream.
+            let json = serde_json::to_string(&evt).ok()?;
+            Some(Ok(Event::default().event("case").data(json)))
+        }
+        // Lagged subscribers: skip silently. The next real event or the
+        // user's next navigation will resync state from the filesystem.
+        Err(BroadcastStreamRecvError::Lagged(n)) => {
+            tracing::debug!(skipped = n, "sse subscriber lagged");
+            None
+        }
+    });
+
+    Sse::new(stream).keep_alive(
+        KeepAlive::new()
+            .interval(Duration::from_secs(15))
+            .text("keepalive"),
+    )
+}

server/src/routes/mod.rs

@@ -1,6 +1,7 @@
 mod bulk;
 mod case_actions;
 mod debug;
+mod events;
 mod health;
 mod login;
 mod magic;
@@ -54,4 +55,5 @@ pub fn api_router() -> Router<AppState> {
             "/web/audio/{user}/{case_id}/{filename}",
             get(web::handle_audio),
         )
+        .route("/web/events", get(events::handle_events))
 }

server/src/routes/upload.rs

@@ -8,12 +8,14 @@ use tracing::{info, warn};
 
 use crate::auth::AuthenticatedUser;
 use crate::error::AppError;
+use crate::events::{self, CaseEventKind, EventSender};
 use crate::models::{AckResponse, AckStatus};
 use crate::transcribe::{TranscribeJob, TranscribeSender};
 
 pub async fn handle_upload(
     user: AuthenticatedUser,
     State(transcribe_tx): State<TranscribeSender>,
+    State(events_tx): State<EventSender>,
     mut multipart: Multipart,
 ) -> Result<Json<AckResponse>, AppError> {
     let mut case_id: Option<String> = None;
@@ -85,6 +87,12 @@ pub async fn handle_upload(
             case_id = %case_id,
             "Reopened soft-deleted case via new upload"
         );
+        events::emit(
+            &events_tx,
+            &user.slug,
+            &case_id,
+            CaseEventKind::CaseRestored,
+        );
     }
 
     // Build a filesystem-safe filename from the timestamp (colons → hyphens).
@@ -100,6 +108,13 @@ pub async fn handle_upload(
         "Recording received"
     );
 
+    events::emit(
+        &events_tx,
+        &user.slug,
+        &case_id,
+        CaseEventKind::RecordingUploaded,
+    );
+
     // Enqueue transcription. Failure here is non-fatal: on restart the recovery
     // scan will re-enqueue any .m4a without a .transcript.txt sibling.
     match transcribe_tx.try_send(TranscribeJob {

server/src/transcribe/recovery.rs

@@ -4,6 +4,7 @@ use tracing::{info, warn};
 
 use super::{TranscribeJob, TranscribeSender};
 use crate::config::Config;
+use crate::events::EventSender;
 use crate::gazetteer::Gazetteer;
 /// Walk `data_path/*/` and enqueue every recording pending transcription for
 /// every user. Intended for startup; the same primitive backs the per-user
@@ -149,6 +150,7 @@ pub async fn regenerate_missing_oneliners(
     client: &reqwest::Client,
     config: &Config,
     vocab: &Gazetteer,
+    events_tx: &EventSender,
 ) {
     let cases = cases_needing_oneliner(data_path).await;
     if cases.is_empty() {
@@ -156,7 +158,7 @@ pub async fn regenerate_missing_oneliners(
     }
     info!(count = cases.len(), "Regenerating missing oneliners");
     for (case_dir, slug) in cases {
-        super::worker::update_oneliner(&case_dir, &slug, client, config, vocab).await;
+        super::worker::update_oneliner(&case_dir, &slug, client, config, vocab, events_tx).await;
     }
 }
 

server/src/transcribe/worker.rs

@@ -6,6 +6,7 @@ use tracing::{error, info, warn};
 
 use super::{TranscribeReceiver, ffmpeg, ollama, whisper};
 use crate::config::{Config, WhisperUserSettings};
+use crate::events::{self, CaseEventKind, EventSender};
 use crate::gazetteer::Gazetteer;
 use crate::{BusyGuard, WorkerBusy};
 
@@ -19,6 +20,7 @@ pub async fn run(
     client: reqwest::Client,
     worker_busy: WorkerBusy,
     vocab: Arc<Gazetteer>,
+    events_tx: EventSender,
 ) {
     info!("Transcription worker started");
     let timeout = Duration::from_secs(config.whisper_timeout_seconds);
@@ -48,7 +50,7 @@ pub async fn run(
             Ok(f) => f,
             Err(e) => {
                 error!(audio = %audio_path.display(), error = %e, "ffmpeg remux failed");
-                mark_failed(&audio_path).await;
+                mark_failed(&audio_path, &events_tx, &job.user_slug).await;
                 continue;
             }
         };
@@ -71,7 +73,7 @@ pub async fn run(
             Ok(t) => t,
             Err(e) => {
                 error!(audio = %audio_path.display(), error = %e, "whisper call failed");
-                mark_failed(&audio_path).await;
+                mark_failed(&audio_path, &events_tx, &job.user_slug).await;
                 continue;
             }
         };
@@ -92,6 +94,17 @@ pub async fn run(
             "Transcript written"
         );
 
+        // Notify the live-update bus so subscribed browsers can refresh the
+        // recordings view. case_id = parent dir of the audio file.
+        if let Some(case_dir) = audio_path.parent() {
+            events::emit(
+                &events_tx,
+                &job.user_slug,
+                events::case_id_of(case_dir),
+                CaseEventKind::TranscriptReady,
+            );
+        }
+
         // Regenerate the oneliner, but only after the *last* pending
         // recording in the batch — otherwise we'd burn LLM calls on
         // incomplete inputs that the next job would redo. Later
@@ -102,7 +115,15 @@ pub async fn run(
         if let Some(case_dir) = audio_path.parent()
             && !has_pending_recordings(case_dir).await
         {
-            update_oneliner(case_dir, &job.user_slug, &client, &config, &vocab).await;
+            update_oneliner(
+                case_dir,
+                &job.user_slug,
+                &client,
+                &config,
+                &vocab,
+                &events_tx,
+            )
+            .await;
         }
     }
 
@@ -113,7 +134,7 @@ pub async fn run(
 /// so the recovery scan no longer picks it up. The UI still surfaces these
 /// files (with a "failed" flag) so the user can listen to the audio and
 /// manually decide whether to retry (by renaming back) or to delete.
-async fn mark_failed(audio_path: &Path) {
+async fn mark_failed(audio_path: &Path, events_tx: &EventSender, user_slug: &str) {
     let failed_path = audio_path.with_extension("m4a.failed");
     if let Err(e) = tokio::fs::rename(audio_path, &failed_path).await {
         // Nothing to roll back — if rename fails, the worst case is that we
@@ -129,6 +150,14 @@ async fn mark_failed(audio_path: &Path) {
             failed = %failed_path.display(),
             "Recording marked as failed",
         );
+        if let Some(case_dir) = audio_path.parent() {
+            events::emit(
+                events_tx,
+                user_slug,
+                events::case_id_of(case_dir),
+                CaseEventKind::TranscriptFailed,
+            );
+        }
     }
 }
 
@@ -162,6 +191,7 @@ pub(crate) async fn update_oneliner(
     client: &reqwest::Client,
     config: &Config,
     vocab: &Gazetteer,
+    events_tx: &EventSender,
 ) {
     let path = case_dir.join("oneliner.txt");
 
@@ -191,6 +221,12 @@ pub(crate) async fn update_oneliner(
                     chars = line.chars().count(),
                     "Oneliner updated"
                 );
+                events::emit(
+                    events_tx,
+                    user_slug,
+                    events::case_id_of(case_dir),
+                    CaseEventKind::OnelinerUpdated,
+                );
             }
         }
         Err(e) => {

server/templates/case_page.html

@@ -147,6 +147,22 @@ try {
     document.documentElement.classList.toggle('admin-view-off', !on);
   });
 })();
+
+// Live-update bus: only events for THIS case trigger a debounced reload.
+(() => {
+  const MY_CASE_ID = "{{ case_id }}";
+  let timer = null;
+  const scheduleReload = () => {
+    clearTimeout(timer);
+    timer = setTimeout(() => location.reload(), 300);
+  };
+  const es = new EventSource('/web/events');
+  es.addEventListener('case', (msg) => {
+    let evt;
+    try { evt = JSON.parse(msg.data); } catch (_) { return; }
+    if (evt.case_id === MY_CASE_ID) scheduleReload();
+  });
+})();
 </script>
 </body>
 </html>

server/templates/case_recordings.html

@@ -245,6 +245,42 @@ try {
     });
   });
 })();
+
+// Live-update bus with audio-playback protection: only events for THIS
+// case trigger a reload, and if any audio is currently playing, show a
+// dismissible "Updates verfügbar"-banner instead of interrupting playback.
+// The `.play.is-pause` class is set by the player script above while audio
+// plays, so we can probe it via DOM selector without extra wiring.
+(() => {
+  const MY_CASE_ID = "{{ case_id }}";
+  let timer = null;
+  const anyAudioPlaying = () => !!document.querySelector('.player .play.is-pause');
+  const showBanner = () => {
+    if (document.getElementById('update-banner')) return;
+    const b = document.createElement('button');
+    b.id = 'update-banner';
+    b.type = 'button';
+    b.textContent = '\u21bb Aktualisierungen verf\u00fcgbar \u2014 neu laden';
+    b.style.cssText = 'position:fixed;bottom:1em;right:1em;z-index:9999;'
+      + 'padding:0.6em 1em;background:#fff3cd;border:1px solid #d39e00;'
+      + 'border-radius:4px;cursor:pointer;font-family:inherit;';
+    b.addEventListener('click', () => location.reload());
+    document.body.appendChild(b);
+  };
+  const scheduleReload = () => {
+    clearTimeout(timer);
+    timer = setTimeout(() => {
+      if (anyAudioPlaying()) showBanner();
+      else location.reload();
+    }, 300);
+  };
+  const es = new EventSource('/web/events');
+  es.addEventListener('case', (msg) => {
+    let evt;
+    try { evt = JSON.parse(msg.data); } catch (_) { return; }
+    if (evt.case_id === MY_CASE_ID) scheduleReload();
+  });
+})();
 </script>
 </body>
 </html>

server/templates/my_cases.html

@@ -151,6 +151,19 @@ try {
     document.documentElement.classList.toggle('admin-view-off', !on);
   });
 })();
+
+// Live-update bus: any case event for this user triggers a debounced
+// page reload. Debounce collapses event bursts (e.g. bulk-delete) into
+// one reload. EventSource auto-reconnects on network drops.
+(() => {
+  let timer = null;
+  const scheduleReload = () => {
+    clearTimeout(timer);
+    timer = setTimeout(() => location.reload(), 300);
+  };
+  const es = new EventSource('/web/events');
+  es.addEventListener('case', () => scheduleReload());
+})();
 </script>
 </body>
 </html>

server/tests/analyze_test.rs

@@ -369,7 +369,14 @@ async fn analyze_worker_writes_document_via_wiremock() {
     let client = reqwest::Client::new();
     let busy = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
     let vocab = std::sync::Arc::new(doctate_server::gazetteer::Gazetteer::empty());
-    let handle = tokio::spawn(analyze::worker::run(rx, config, client, busy, vocab));
+    let handle = tokio::spawn(analyze::worker::run(
+        rx,
+        config,
+        client,
+        busy,
+        vocab,
+        doctate_server::events::channel(),
+    ));
 
     tx.send(analyze::AnalyzeJob {
         case_dir: case_dir.clone(),
@@ -447,7 +454,14 @@ async fn analyze_worker_normalizes_llm_output() {
     let (tx, rx) = analyze::channel();
     let client = reqwest::Client::new();
     let busy = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
-    let handle = tokio::spawn(analyze::worker::run(rx, config, client, busy, vocab));
+    let handle = tokio::spawn(analyze::worker::run(
+        rx,
+        config,
+        client,
+        busy,
+        vocab,
+        doctate_server::events::channel(),
+    ));
 
     tx.send(analyze::AnalyzeJob {
         case_dir: case_dir.clone(),
@@ -998,7 +1012,14 @@ async fn analyze_works_against_ollama_style_endpoint_without_api_key() {
     let client = reqwest::Client::new();
     let busy = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
     let vocab = std::sync::Arc::new(doctate_server::gazetteer::Gazetteer::empty());
-    let handle = tokio::spawn(analyze::worker::run(rx, config, client, busy, vocab));
+    let handle = tokio::spawn(analyze::worker::run(
+        rx,
+        config,
+        client,
+        busy,
+        vocab,
+        doctate_server::events::channel(),
+    ));
 
     tx.send(analyze::AnalyzeJob {
         case_dir: case_dir.clone(),

server/tests/magic_link_test.rs

@@ -54,6 +54,7 @@ fn build_state() -> (AppState, Arc<Config>) {
         magic_link_store,
         analyze_busy: AnalyzeBusy(Arc::new(AtomicBool::new(false))),
         transcribe_busy: TranscribeBusy(Arc::new(AtomicBool::new(false))),
+        events_tx: doctate_server::events::channel(),
     };
 
     (state, config)

server/tests/sse_integration.rs

@@ -0,0 +1,136 @@
+//! Integration tests for the `/web/events` SSE route.
+//!
+//! We test two things end-to-end:
+//! 1. Without a session cookie, the route redirects to the login page
+//!    (the web-auth extractor's standard rejection).
+//! 2. With a valid session, the route returns a streaming response with
+//!    `Content-Type: text/event-stream`.
+//!
+//! Actual event-filtering logic (per-user scoping, admin sees-all) is
+//! covered at the unit level in `src/events.rs`; wiring the broadcast
+//! channel through an HTTP test would require reading a never-ending
+//! body, which these tests deliberately avoid.
+
+use std::collections::HashMap;
+use std::path::PathBuf;
+use std::sync::Arc;
+
+use axum::body::Body;
+use axum::http::{Request, StatusCode, header};
+use tower::util::ServiceExt;
+
+use doctate_server::config::{Config, User};
+
+fn unique_data_path() -> PathBuf {
+    std::env::temp_dir().join(format!(
+        "doctate-sse-test-{}-{}",
+        std::process::id(),
+        uuid::Uuid::new_v4()
+    ))
+}
+
+fn make_user(slug: &str) -> User {
+    User {
+        slug: slug.into(),
+        api_key: format!("key-{slug}"),
+        // Password "s" — matches the login helper below.
+        web_password: bcrypt::hash("s", 4).unwrap(),
+        role: "doctor".into(),
+        whisper: Default::default(),
+    }
+}
+
+fn build_config(data_path: PathBuf) -> Arc<Config> {
+    let users = vec![make_user("alice")];
+    let api_keys: HashMap<String, String> = users
+        .iter()
+        .map(|u| (u.api_key.clone(), u.slug.clone()))
+        .collect();
+    Arc::new(Config {
+        data_path,
+        users,
+        api_keys,
+        ..Config::test_default()
+    })
+}
+
+/// Log in via POST /web/login and return the resulting `session=…` cookie
+/// header value. Mirrors the helper in the other integration tests.
+async fn login(app: axum::Router, slug: &str) -> String {
+    let body = format!("slug={slug}&password=s");
+    let resp = app
+        .oneshot(
+            Request::builder()
+                .method("POST")
+                .uri("/web/login")
+                .header(header::CONTENT_TYPE, "application/x-www-form-urlencoded")
+                .body(Body::from(body))
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+    for v in resp.headers().get_all(header::SET_COOKIE).iter() {
+        let s = v.to_str().unwrap();
+        if let Some(pair) = s.split(';').next()
+            && pair.starts_with("session=")
+        {
+            return pair.to_string();
+        }
+    }
+    panic!("no session cookie after login");
+}
+
+#[tokio::test]
+async fn events_without_session_redirects_to_login() {
+    let config = build_config(unique_data_path());
+    let app = doctate_server::create_router(config);
+
+    let resp = app
+        .oneshot(
+            Request::builder()
+                .method("GET")
+                .uri("/web/events")
+                .body(Body::empty())
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+
+    assert_eq!(resp.status(), StatusCode::FOUND);
+    let location = resp
+        .headers()
+        .get(header::LOCATION)
+        .and_then(|v| v.to_str().ok())
+        .unwrap_or("");
+    assert_eq!(location, "/web/login");
+}
+
+#[tokio::test]
+async fn events_with_session_returns_sse_content_type() {
+    let config = build_config(unique_data_path());
+    let app = doctate_server::create_router(config);
+    let cookie = login(app.clone(), "alice").await;
+
+    let resp = app
+        .oneshot(
+            Request::builder()
+                .method("GET")
+                .uri("/web/events")
+                .header(header::COOKIE, cookie)
+                .body(Body::empty())
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+
+    assert_eq!(resp.status(), StatusCode::OK);
+    let ct = resp
+        .headers()
+        .get(header::CONTENT_TYPE)
+        .and_then(|v| v.to_str().ok())
+        .unwrap_or("");
+    assert!(
+        ct.starts_with("text/event-stream"),
+        "expected text/event-stream, got {ct:?}"
+    );
+}

server/tests/transcribe_test.rs

@@ -304,7 +304,15 @@ async fn worker_renames_audio_to_failed_on_whisper_error() {
     let client = reqwest::Client::new();
     let busy = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
     let vocab = std::sync::Arc::new(doctate_server::gazetteer::Gazetteer::empty());
-    transcribe::worker::run(rx, config, client, busy, vocab).await;
+    transcribe::worker::run(
+        rx,
+        config,
+        client,
+        busy,
+        vocab,
+        doctate_server::events::channel(),
+    )
+    .await;
 
     // Audio must have been renamed so recovery skips it next time.
     assert!(!audio.exists(), "original .m4a still present");
@@ -353,7 +361,15 @@ async fn transcribe_worker_normalizes_whisper_output() {
 
     let client = reqwest::Client::new();
     let busy = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
-    transcribe::worker::run(rx, config, client, busy, vocab).await;
+    transcribe::worker::run(
+        rx,
+        config,
+        client,
+        busy,
+        vocab,
+        doctate_server::events::channel(),
+    )
+    .await;
 
     let transcript_path = case_dir.join("2026-04-13T10-30-00Z.transcript.txt");
     assert!(transcript_path.exists(), "transcript missing");