acpmux: replies and events wait until what they show is durable; the daemon stops for a restart when the journal cannot go on (R-DURABILITY S2c)

cmux-tui/crates/acpmux/README.md

@@ -1482,9 +1482,10 @@ When no config exists, harnesses are imported from `~/.acpx/config.json` (its `a
   could not write into a log that changed without the journal (a run with the switch off after a
   crash, or an older build) go to `journal-unapplied/`. Turning it off is safe: the next start
   writes the journal's lines into the logs first. A failed log sync stops the daemon for a restart,
-  which writes the logs again. With it on, an event line over 64 MiB (the journal's record limit)
-  is refused, and its log does not get it either; with it off, the log takes it. Read when the
-  daemon starts.
+  which writes the logs again: it exits with status 74, and its supervisor (or the next client)
+  starts it again. A reply and an acknowledged event wait until what they show is durable. With it
+  on, an event line over 64 MiB (the journal's record limit) is refused, and its log does not get
+  it either; with it off, the log takes it. Read when the daemon starts.
 - `websocket`: optional. The same protocol over WebSocket text frames, for remote clients such
   as a dashboard or a Durable Object. Put it behind a tunnel; the token is a bearer token.
 

cmux-tui/crates/acpmux/src/cli/entry.rs

@@ -229,7 +229,10 @@ async fn async_main(args: Vec<OsString>, invocation: Invocation) -> Result<()> {
                 )
                 .with_target(false)
                 .init();
-            crate::qos::enter(crate::qos::Role::Daemon);
+            // With `store.journal` on, an acknowledged reply waits on the journal's sync, which a
+            // utility disk tier throttles 15-30x under another app's disk work (STATUS 11554):
+            // the daemon stays at the default tier, and only its background threads yield.
+            crate::qos::enter_daemon(crate::config::store_journal_enabled());
             let result = crate::daemon::run(DaemonOptions {
                 ws_listen: listen,
                 ws_token: token,
@@ -263,8 +266,9 @@ async fn async_main(args: Vec<OsString>, invocation: Invocation) -> Result<()> {
             }
             // The daemon has stopped its agents and synced its store. Exit
             // now rather than wait for the runtime to drain blocking tasks
-            // (a launcher `--version` check can take 20 s).
-            std::process::exit(0)
+            // (a launcher `--version` check can take 20 s); 74 when its store
+            // asked for a restart (`daemon::EXIT_REPAIR`).
+            std::process::exit(crate::daemon::exit_status())
         }
         Some(Command::CliAgent {
             last_block,

cmux-tui/crates/acpmux/src/config.rs

@@ -2050,7 +2050,7 @@ pub use discover::{
 mod peer;
 pub use peer::PeerConfig;
 mod store_config;
-pub use store_config::{StoreConfig, StoreMode};
+pub use store_config::{StoreConfig, StoreMode, store_journal_enabled};
 mod pool;
 pub use pool::PoolConfig;
 mod btw;

cmux-tui/crates/acpmux/src/config/store_config.rs

@@ -57,3 +57,27 @@ impl Default for StoreConfig {
         }
     }
 }
+
+/// Whether `store.journal` is on in this home's config file, read without loading the rest of the
+/// config (no harness discovery): the daemon's process tier is set before it starts any thread
+/// (`qos::enter_daemon`). A file that is missing or does not read counts as off, the default; the
+/// daemon's own config load then reports the error.
+pub fn store_journal_enabled() -> bool {
+    #[derive(Deserialize)]
+    struct File {
+        #[serde(default)]
+        store: Option<Store>,
+    }
+    #[derive(Deserialize)]
+    struct Store {
+        #[serde(default)]
+        journal: bool,
+    }
+    let path = super::Config::path();
+    let Ok(mut file) = super::private_state_file(&path) else { return false };
+    let mut text = String::new();
+    if std::io::Read::read_to_string(&mut file, &mut text).is_err() {
+        return false;
+    }
+    serde_json::from_str::<File>(&text).ok().and_then(|file| file.store).is_some_and(|s| s.journal)
+}

cmux-tui/crates/acpmux/src/daemon.rs

@@ -54,6 +54,7 @@ mod stop_causes;
 #[cfg(unix)]
 use stop_causes::next_signal;
 use stop_causes::{AbortOnDrop, store_asks_to_stop};
+pub use stop_causes::{EXIT_REPAIR, exit_status};
 #[cfg(unix)]
 pub(crate) mod term_sender;
 pub(crate) mod unused_exit;
@@ -719,7 +720,7 @@ pub async fn run(opts: DaemonOptions) -> Result<()> {
     let tailnet = crate::tailnet::embedded_service::Service::spawn(hub.clone());
     // The store asks for a restart when it cannot go on safely (`Store::stop_reason`): a frozen
     // event journal checkpoint, whose logs only the next start's replay writes again.
-    let mut store_stop = std::pin::pin!(store_asks_to_stop(hub.store.stop_reason()));
+    let mut store_stop = std::pin::pin!(store_asks_to_stop(hub.clone()));
     let shutdown = async {
         // Every stop names its cause in the log (`app_orphan` and `unused` name their own).
         #[cfg(unix)]
@@ -743,9 +744,7 @@ pub async fn run(opts: DaemonOptions) -> Result<()> {
                 _ = next_signal(int.as_mut()) => tracing::warn!("stopping: SIGINT"),
                 _ = hub.shutdown.notified() => tracing::warn!("stopping: a client asked it to"),
                 // A restart handoff: the next start repairs what the store could not.
-                () = &mut store_stop => {
-                    hub.restart_requested.store(true, std::sync::atomic::Ordering::SeqCst);
-                },
+                () = &mut store_stop => {},
                 // No build of its app is left to adopt its agents, so they end with it. An app
                 // that only quit or was updated keeps them, as its Quit does
                 // (`daemon/app_orphan.rs`).
@@ -799,9 +798,7 @@ pub async fn run(opts: DaemonOptions) -> Result<()> {
             tokio::select! {
                 _ = ctrl_c => {},
                 _ = hub.shutdown.notified() => {},
-                () = &mut store_stop => {
-                    hub.restart_requested.store(true, std::sync::atomic::Ordering::SeqCst);
-                },
+                () = &mut store_stop => {},
                 // No build of its app is left to adopt its agents, so they end with it. An app
                 // that only quit or was updated keeps them, as its Quit does
                 // (`daemon/app_orphan.rs`).

cmux-tui/crates/acpmux/src/daemon/stop_causes.rs

@@ -1,6 +1,24 @@
 //! What stops the daemon besides a client: a signal, and the store asking for a restart (moved
 //! out of `daemon.rs`, which is over its size budget).
 
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
+
+/// The daemon's exit status after its store asked it to stop for a restart ([`store_asks_to_stop`]:
+/// a failed log sync froze the event journal's checkpoint, or the journal's writer ended): EX_IOERR.
+/// The next start's replay repairs the logs; a supervisor starts it again (launchd `KeepAlive`,
+/// systemd `Restart=always`: the units prevent only 75), and otherwise the next client does.
+pub const EXIT_REPAIR: i32 = 74;
+
+/// The store's stop won the daemon's stop select.
+static STOPPED_FOR_REPAIR: AtomicBool = AtomicBool::new(false);
+
+/// The process's exit status after `daemon::run` returned: [`EXIT_REPAIR`] when the store asked
+/// for the stop, else 0.
+pub fn exit_status() -> i32 {
+    if STOPPED_FOR_REPAIR.load(Ordering::SeqCst) { EXIT_REPAIR } else { 0 }
+}
+
 /// Aborts a task when dropped.
 pub(super) struct AbortOnDrop(pub(super) tokio::task::AbortHandle);
 
@@ -10,18 +28,19 @@ impl Drop for AbortOnDrop {
     }
 }
 
-/// Resolves, with the reason logged, once the store asks the daemon to stop
-/// (`Store::stop_reason`); never for a store that does not ask, or once the store is gone.
-pub(super) async fn store_asks_to_stop(
-    reason: Option<tokio::sync::watch::Receiver<Option<String>>>,
-) {
-    let Some(mut reason) = reason else { return std::future::pending().await };
+/// Resolves once the store asks the daemon to stop (`Store::stop_reason`), with the reason
+/// logged, the stop marked a restart handoff (WebSocket clients get 1012) and the exit status set
+/// to [`EXIT_REPAIR`]; never for a store that does not ask, or once the store is gone.
+pub(super) async fn store_asks_to_stop(hub: Arc<crate::hub::Hub>) {
+    let Some(mut reason) = hub.store.stop_reason() else { return std::future::pending().await };
     match reason.wait_for(Option::is_some).await {
         Ok(why) => {
             tracing::error!("stopping for a restart: {}", why.as_deref().unwrap_or_default());
         }
-        Err(_) => std::future::pending().await,
+        Err(_) => return std::future::pending().await,
     }
+    hub.restart_requested.store(true, Ordering::SeqCst);
+    STOPPED_FOR_REPAIR.store(true, Ordering::SeqCst);
 }
 
 /// The next delivery of a registered signal; never, if registration failed.

cmux-tui/crates/acpmux/src/hub/mod.rs

@@ -2327,14 +2327,22 @@ impl Hub {
         self.store.events(id, after, limit)
     }
 
-    /// Visits the records past `after` in order, one at a time, until `visit` returns false.
+    /// Visits the records past `after` in order, one at a time, until `visit` returns false. A
+    /// caller sends each record as it reads it (`session/load`'s replay), so the visit stops at
+    /// the chat's `last_seq` read first: a line the log holds before its journal record is
+    /// appended (R-DURABILITY S2c) is never sent ahead of that record. A chat no session holds
+    /// takes no appends: read to the end.
     pub fn scan_events(
         &self,
         id: &str,
         after: u64,
         visit: &mut dyn FnMut(EventRecord) -> bool,
     ) -> Result<()> {
-        self.store.scan(id, after, visit)
+        let live = self.sessions.lock().unwrap_or_else(|e| e.into_inner()).get(id).cloned();
+        let Some(last) = live.map(|session| session.meta().last_seq) else {
+            return self.store.scan(id, after, visit);
+        };
+        self.store.scan(id, after, &mut |record| record.seq <= last && visit(record))
     }
 
     pub fn session_dir(&self, id: &str) -> Option<PathBuf> {

cmux-tui/crates/acpmux/src/qos.rs

@@ -102,6 +102,28 @@ fn agent_background() -> bool {
     std::env::var_os(ENV).is_some_and(|v| v == "background")
 }
 
+/// The daemon's tier once its config is read: with `store.journal` on, acknowledged replies wait
+/// on the journal's sync, so the daemon keeps the default disk tier (lead rule 2026-11-04: a reply
+/// the person waits for never waits on a throttled sync) and only its background threads yield
+/// ([`background_thread`]). Without the journal, [`Tier::Serving`] as before.
+pub fn enter_daemon(journal: bool) {
+    if journal {
+        // The default disk tier: nothing to set (the CPU is unchanged for the daemon either way).
+        return;
+    }
+    enter(Role::Daemon);
+}
+
+/// The calling thread, dedicated to background work for its whole life (the event journal's
+/// checkpointer): its disk I/O at the utility tier. Only for a thread that does nothing else: a
+/// thread-scope I/O policy also drops the thread out of QoS for good (see `imp::ThreadOnScreen`),
+/// so a pool thread that is reused must never take it.
+pub fn background_thread() {
+    if enabled() {
+        imp::thread_disk_utility();
+    }
+}
+
 /// Put this process in `role`'s tier. Call it first thing in `main`, before
 /// any thread starts: for [`Tier::Agent`] on macOS it re-executes the process
 /// in place and returns only when that is not done (already clamped, turned
@@ -158,7 +180,6 @@ mod imp {
 
     const IOPOL_TYPE_DISK: libc::c_int = 0;
     const IOPOL_SCOPE_PROCESS: libc::c_int = 0;
-    #[cfg(test)]
     const IOPOL_SCOPE_THREAD: libc::c_int = 1;
     const IOPOL_UTILITY: libc::c_int = 4;
     const PRIO_DARWIN_BG: libc::c_int = 0x1000;
@@ -187,6 +208,14 @@ mod imp {
         }
     }
 
+    /// The calling thread's disk I/O at utility (`background_thread`).
+    pub(super) fn thread_disk_utility() {
+        // SAFETY: plain syscall wrapper on the calling thread with constant arguments.
+        if unsafe { setiopolicy_np(IOPOL_TYPE_DISK, IOPOL_SCOPE_THREAD, IOPOL_UTILITY) } != 0 {
+            tracing::debug!("setiopolicy_np (thread): {}", std::io::Error::last_os_error());
+        }
+    }
+
     /// The clamp setter, looked up at run time: it is not in the public SDK
     /// headers, so a macOS without it starts unclamped instead of failing to load.
     fn clamp_fn() -> Option<ClampFn> {
@@ -362,6 +391,7 @@ mod imp {
 #[cfg(target_os = "linux")]
 mod imp {
     pub(super) fn disk_utility() {}
+    pub(super) fn thread_disk_utility() {}
 
     /// nice +5: inherited by everything the process starts. Not undone (an
     /// unprivileged process cannot lower its nice value), and not needed to be.
@@ -397,6 +427,7 @@ mod imp {
 #[cfg(not(any(target_os = "macos", target_os = "linux")))]
 mod imp {
     pub(super) fn disk_utility() {}
+    pub(super) fn thread_disk_utility() {}
     pub(super) fn clamp_cpu_utility() {}
     pub(super) fn process_background() {
         clamp_cpu_utility();
@@ -417,6 +448,11 @@ mod imp {
     }
 }
 
+// Its tests read macOS disk policies only.
+#[cfg(all(test, target_os = "macos"))]
+#[path = "qos_journal_tests.rs"]
+mod journal_tests;
+
 #[cfg(test)]
 mod tests {
     use super::*;

cmux-tui/crates/acpmux/src/qos_journal_tests.rs

@@ -0,0 +1,79 @@
+//! The daemon's disk tier with the event journal on (R-DURABILITY S2c; lead rule 2026-11-04: a
+//! reply the person waits for never waits on a throttled disk sync). Fleet A/B (STATUS 11554):
+//! at `IOPOL_UTILITY` an acknowledged append waited about 16 ms p50 under another process's disk
+//! stress, against 0.7 to 1.1 ms at the default tier.
+
+use super::*;
+
+/// The process's disk I/O policy (macOS `getiopolicy_np`, process scope).
+#[cfg(target_os = "macos")]
+fn process_io() -> libc::c_int {
+    unsafe extern "C" {
+        fn getiopolicy_np(iotype: libc::c_int, scope: libc::c_int) -> libc::c_int;
+    }
+    // SAFETY: a plain syscall with constant arguments.
+    unsafe { getiopolicy_np(0, 0) }
+}
+
+/// The calling thread's own disk I/O policy (0: it follows the process's).
+#[cfg(target_os = "macos")]
+fn thread_io() -> libc::c_int {
+    unsafe extern "C" {
+        fn getiopolicy_np(iotype: libc::c_int, scope: libc::c_int) -> libc::c_int;
+    }
+    // SAFETY: as above, thread scope.
+    unsafe { getiopolicy_np(0, 1) }
+}
+
+/// With the journal on, the daemon's process keeps the default disk tier; without it, utility as
+/// before. A test process's own policy, so each case runs in a child of this test binary.
+#[cfg(target_os = "macos")]
+#[test]
+fn the_daemon_with_the_journal_keeps_the_default_disk_tier() {
+    if let Ok(case) = std::env::var("ACPMUX_QOS_JOURNAL_CASE") {
+        enter_daemon(case == "journal");
+        println!("QOS-CASE io={}", process_io());
+        return;
+    }
+    let exe = std::env::current_exe().unwrap();
+    let mut seen = Vec::new();
+    for case in ["journal", "nojournal"] {
+        let out = std::process::Command::new(&exe)
+            .args([
+                "--exact",
+                "qos::journal_tests::the_daemon_with_the_journal_keeps_the_default_disk_tier",
+                "--nocapture",
+            ])
+            .env("ACPMUX_QOS_JOURNAL_CASE", case)
+            .env_remove(ENV)
+            .output()
+            .unwrap();
+        let text = String::from_utf8_lossy(&out.stdout).into_owned();
+        let io = text
+            .lines()
+            .find_map(|line| line.strip_prefix("QOS-CASE io="))
+            .and_then(|io| io.trim().parse::<i32>().ok())
+            .unwrap_or_else(|| panic!("case {case}: no QOS-CASE line in {text}"));
+        seen.push((case, io));
+    }
+    // 4 is IOPOL_UTILITY; the default reads 0 (or 1, IOPOL_IMPORTANT, the same tier).
+    assert!(matches!(seen[0], ("journal", 0 | 1)), "the daemon with the journal: {seen:?}");
+    assert_eq!(seen[1], ("nojournal", 4), "the daemon without the journal: {seen:?}");
+}
+
+/// A dedicated background thread (the checkpointer) does its disk work at the utility tier, and
+/// a thread beside it keeps the process's.
+#[cfg(target_os = "macos")]
+#[test]
+fn a_background_thread_yields_its_disk_work() {
+    let (own, other) = std::thread::scope(|scope| {
+        let own = scope.spawn(|| {
+            background_thread();
+            thread_io()
+        });
+        let other = scope.spawn(thread_io);
+        (own.join().unwrap(), other.join().unwrap())
+    });
+    assert_eq!(own, 4, "the background thread's disk tier");
+    assert_eq!(other, 0, "a thread beside it");
+}