Persist Matrix key requests safely and pin the local master

.github/workflows/ci.yaml

@@ -46,13 +46,21 @@ jobs:
             R CMD INSTALL "$RUNNER_TEMP/$pkg"
           done
 
-      # mx.crypto needs a Rust toolchain to build from source, so it is
-      # Linux-only here, off the r2u binary rapt already has sources for.
-      # Its one test -- that matrix_crypto_init() binds the store before
-      # opening the account -- is conditional and does not run on macOS.
+      # The development crypto APIs come from drat after the upstream PR
+      # lands. Bypass rapt's older r2u binary and build the vendored sources
+      # with the Linux runner's Rust toolchain. Keep the existing platform
+      # split: macOS exercises the Matrix adapter without optional crypto.
       - name: Install mx.crypto (Linux)
         if: runner.os == 'Linux'
-        run: Rscript -e 'install.packages("mx.crypto")'
+        run: |
+          Rscript -e '
+            if (isTRUE(getOption("rapt.enabled"))) rapt::disable()
+            utils::install.packages("mx.crypto",
+                             repos = "https://cornball-ai.github.io/drat",
+                             type = "source", dependencies = FALSE)
+            if (!requireNamespace("mx.crypto", quietly = TRUE) ||
+                utils::packageVersion("mx.crypto") < "0.2.1.1")
+              stop("CI needs mx.crypto >= 0.2.1.1; publish it to drat first")'
 
       # Fail loudly rather than letting the suite skip. mx.crypto is
       # reported but not required, since the macOS leg does without it.

DESCRIPTION

@@ -1,8 +1,8 @@
 Package: chat.api
 Type: Package
 Title: Transport-Agnostic Chat Contract for R Agents
-Version: 0.0.1.24
-Date: 2026-08-20
+Version: 0.0.1.25
+Date: 2026-09-04
 Authors@R: c(
     person("Troy", "Hernandez", role = c("aut", "cre"),
            email = "troy@cornball.ai",
@@ -22,8 +22,8 @@ BugReports: https://github.com/cornball-ai/chat.api/issues
 Suggests:
     httr,
     mx.api,
-    mx.client (>= 0.2.0.6),
-    mx.crypto,
+    mx.client (>= 0.2.0.8),
+    mx.crypto (>= 0.2.1.1),
     slackr,
     tinytest
 Encoding: UTF-8

NEWS.md

@@ -0,0 +1,10 @@
+# chat.api 0.0.1.25
+
+* Matrix E2EE saves ratchet state before room-key request transport, retries
+  unsent requests with their stable ids, and treats transport failures as
+  warnings so a decrypted sync batch is not lost or replayed.
+* Same-user forwarded-key recovery requires a cross-signing chain matching
+  the master key in the local crypto store. Missing or unreadable local keys
+  leave the user's devices untrusted for recovery.
+* Matrix E2EE now requires mx.client >= 0.2.0.8 and mx.crypto >= 0.2.1.1 for
+  durable request handling and local cross-signing key access.

R/matrix-crypto.R

@@ -463,6 +463,73 @@ matrix_detect_encrypted_rooms <- function(sync) {
     out
 }
 
+# Persist ratchet state before any request transport. Olm and Megolm handles
+# mutate in place during sync processing, so a network error must never unwind
+# past this point and cause the same batch to replay against advanced ratchets.
+matrix_crypto_commit_key_requests <- function(
+    crypto, res, mx,
+    .send = mx.client::mx_crypto_send_key_requests,
+    .mark = mx.client::mx_crypto_mark_key_requests_sent,
+    .save = mx.client::mx_crypto_sessions_save) {
+    crypto$sessions <- res$sessions
+    .save(crypto$sessions, crypto$store)
+
+    sent <- list()
+    for (request in res$key_requests %||% list()) {
+        ok <- tryCatch({
+            .send(mx, list(request))
+            TRUE
+        }, error = function(e) {
+            warning("chat.api: Matrix room-key request send failed; it remains ",
+                    "queued: ", conditionMessage(e), call. = FALSE)
+            FALSE
+        })
+        if (ok) sent[[length(sent) + 1L]] <- request
+    }
+
+    if (length(sent)) {
+        crypto$sessions <- .mark(crypto$sessions, sent)
+        # Failure here is safe: the first save contains sent = FALSE, so a
+        # restart can resend the same stable request id. Keep the live copy
+        # marked to avoid a duplicate during this process.
+        tryCatch(
+            .save(crypto$sessions, crypto$store),
+            error = function(e) warning(
+                "chat.api: room-key request sent marker was not persisted; ",
+                "a restart may safely retry it: ", conditionMessage(e),
+                call. = FALSE))
+    }
+
+    for (cancellation in res$key_request_cancellations %||% list()) {
+        tryCatch(
+            .send(mx, list(cancellation)),
+            error = function(e) warning(
+                "chat.api: Matrix room-key request cancellation failed and ",
+                "was dropped: ", conditionMessage(e), call. = FALSE))
+    }
+    invisible(NULL)
+}
+
+# Reload the local authority when querying devices so bootstrap performed
+# after context initialization is picked up too. A missing/unreadable store
+# leaves our own devices untrusted without losing other users' device keys.
+matrix_crypto_known_devices <- function(
+    crypto, mx, user_ids,
+    .load = mx.client::mx_crypto_cross_signing_load,
+    .public = mx.crypto::mxc_signing_key_public,
+    .query = mx.client::mx_crypto_known_devices) {
+    master <- tryCatch({
+        keys <- .load(crypto$store)
+        if (is.null(keys)) NULL else .public(keys$master)
+    }, error = function(e) {
+        warning("chat.api: cannot load local cross-signing master; this ",
+                "user's devices will not be trusted for room-key recovery: ",
+                conditionMessage(e), call. = FALSE)
+        NULL
+    })
+    .query(mx, user_ids, self_master_key = master)
+}
+
 # Decrypt a sync: recover room keys from to-device, decrypt encrypted
 # timeline events, refresh the encrypted-room set. Mutates `crypto`.
 #
@@ -480,17 +547,17 @@ matrix_crypto_decrypt <- function(crypto, sync, mx) {
     }
     senders <- matrix_encrypted_senders(sync)
     devices <- if (length(senders)) {
-        tryCatch(mx.client::mx_crypto_known_devices(mx, senders),
+        tryCatch(matrix_crypto_known_devices(crypto, mx, senders),
                  error = function(e) NULL)
     } else {
         NULL
     }
 
     res <- mx.client::mx_crypto_process_sync(
         crypto$account, crypto$sessions, sync, crypto$self_curve,
-        self_id = mx$user_id, devices = devices)
-    crypto$sessions <- res$sessions
-    mx.client::mx_crypto_sessions_save(crypto$sessions, crypto$store)
+        self_id = mx$user_id, devices = devices,
+        self_device_id = mx$device_id)
+    matrix_crypto_commit_key_requests(crypto, res, mx)
     res$events
 }
 

inst/tinytest/test_matrix_mxclient.R

@@ -10,6 +10,32 @@ if (!requireNamespace("mx.client", quietly = TRUE)) {
     exit_file("mx.client not installed")
 }
 
+# Device queries use the local master even when bootstrap happens after the
+# crypto context was built. Missing/corrupt keys must never trust a server pin.
+if (requireNamespace("mx.crypto", quietly = TRUE) &&
+    utils::packageVersion("mx.crypto") >= "0.2.1.1") local({
+    store <- tempfile("pin-store-")
+    dir.create(store)
+    on.exit(unlink(store, recursive = TRUE), add = TRUE)
+    crypto <- list(store = store)
+    mx <- list(user_id = "@bot:example.org")
+    query <- function(client, user_ids, self_master_key) {
+        expect_identical(client, mx)
+        expect_identical(user_ids, mx$user_id)
+        list(pin = self_master_key)
+    }
+    lookup <- function() chat.api:::matrix_crypto_known_devices(
+        crypto, mx, mx$user_id, .query = query)
+    expect_null(lookup()$pin)
+    keys <- mx.client:::mx_crypto_cross_signing_new()
+    mx.client:::mx_crypto_cross_signing_save(keys, store)
+    expect_identical(lookup()$pin,
+                     mx.crypto::mxc_signing_key_public(keys$master))
+    writeLines("corrupt", file.path(store, "cross-signing.json"))
+    expect_warning(untrusted <- lookup(), "cannot load local cross-signing")
+    expect_null(untrusted$pin)
+})
+
 # ---- API drift ----
 # Membership is not enough: the adapter passes the first one or two
 # arguments positionally, so an upstream reorder would keep every name
@@ -1083,3 +1109,81 @@ local({
     expect_error(chat_set_identity(cl, "x"), "M_LIMIT_EXCEEDED")
     expect_identical(cl$env$mx$token, "rotated")
 })
+
+# Crypto state is committed before request transport. Failed requests remain
+# unsent and retryable; successful requests are marked only after transport;
+# cancellation failure is warning-only.
+local({
+    req_ok <- list(request_id = "ok", content = list(action = "request"))
+    req_fail <- list(request_id = "fail", content = list(action = "request"))
+    cancel <- list(request_id = "cancel",
+                   content = list(action = "request_cancellation"))
+    sessions <- list(key_requests = list(
+        a = c(req_ok, list(sent = FALSE)),
+        b = c(req_fail, list(sent = FALSE))))
+    res <- list(sessions = sessions, key_requests = list(req_ok, req_fail),
+                key_request_cancellations = list(cancel))
+    crypto <- new.env(parent = emptyenv())
+    crypto$sessions <- list()
+    crypto$store <- "unused"
+    ops <- character()
+    snapshots <- list()
+    save <- function(sessions, store) {
+        ops <<- c(ops, "save")
+        snapshots[[length(snapshots) + 1L]] <<- sessions
+        invisible(store)
+    }
+    send <- function(mx, requests) {
+        id <- requests[[1L]]$request_id
+        ops <<- c(ops, paste0("send:", id))
+        if (id %in% c("fail", "cancel")) stop("offline")
+        invisible(list())
+    }
+    mark <- function(sessions, requests) {
+        ids <- vapply(requests, `[[`, character(1), "request_id")
+        for (key in names(sessions$key_requests)) {
+            if (sessions$key_requests[[key]]$request_id %in% ids)
+                sessions$key_requests[[key]]$sent <- TRUE
+        }
+        sessions
+    }
+    warnings <- character()
+    withCallingHandlers(
+        chat.api:::matrix_crypto_commit_key_requests(
+            crypto, res, list(), .send = send, .mark = mark, .save = save),
+        warning = function(w) {
+            warnings <<- c(warnings, conditionMessage(w))
+            invokeRestart("muffleWarning")
+        })
+    expect_identical(ops,
+        c("save", "send:ok", "send:fail", "save", "send:cancel"))
+    expect_equal(length(snapshots), 2L)
+    expect_false(snapshots[[1L]]$key_requests$a$sent)
+    expect_true(snapshots[[2L]]$key_requests$a$sent)
+    expect_false(snapshots[[2L]]$key_requests$b$sent)
+    expect_true(crypto$sessions$key_requests$a$sent)
+    expect_false(crypto$sessions$key_requests$b$sent)
+    expect_equal(length(warnings), 2L)
+})
+
+# A post-send marker-save failure is warning-only. The first snapshot remains
+# safely retryable and the live state avoids an immediate duplicate.
+local({
+    request <- list(request_id = "ok", content = list(action = "request"))
+    sessions <- list(key_requests = list(
+        a = c(request, list(sent = FALSE))))
+    res <- list(sessions = sessions, key_requests = list(request),
+                key_request_cancellations = list())
+    crypto <- new.env(parent = emptyenv())
+    crypto$store <- "unused"
+    saves <- 0L
+    save <- function(...) {
+        saves <<- saves + 1L
+        if (saves == 2L) stop("disk full")
+    }
+    expect_warning(chat.api:::matrix_crypto_commit_key_requests(
+        crypto, res, list(), .send = function(...) NULL,
+        .mark = function(s, r) { s$key_requests$a$sent <- TRUE; s },
+        .save = save), "restart may safely retry")
+    expect_true(crypto$sessions$key_requests$a$sent)
+})