feat(mailbox): senders see the message.expired outcome (#6128 b2, 3/5)

crates/fractal-api-core/src/api.rs

@@ -5400,14 +5400,40 @@ fn message_delivery_for(
         })
     };
     let mut status = message.status.clone();
-    if let Some(host) = message_reads::relay_host(store, message)? {
-        match relayed {
-            Some(message_reads::Relayed::Answered { status: Some(relayed), .. }) => {
-                if crate::store::message_marks::rank(relayed) > crate::store::message_marks::rank(&status) {
-                    status = (*relayed).to_owned();
-                }
+    let host = message_reads::relay_host(store, message)?;
+    if host.is_some()
+        && let Some(message_reads::Relayed::Answered { status: Some(relayed), .. }) = relayed
+        && crate::store::message_marks::rank(relayed) > crate::store::message_marks::rank(&status)
+    {
+        status = (*relayed).to_owned();
+    }
+    // Durable expiry wins over seat diagnostics and an unavailable remote, but
+    // never over actual consumption, including a receipt just relayed by its host.
+    if let Some(expiry) = store.message_expiry(&message.subject)? {
+        if expiry.superseded
+            && let Some(current) = store.message(&message.subject)?
+            && crate::store::message_marks::rank(&current.status) > crate::store::message_marks::rank(&status)
+        {
+            status = current.status;
+        }
+        if !expiry.superseded && !matches!(status.as_str(), "delivered" | "read") {
+            if host.is_some()
+                && let Some(message_reads::Relayed::Answered { consumed: Some(phase), .. }) = relayed
+            {
+                return message_delivery_value(store, &message.to, phase, sent_at, now);
             }
-            Some(message_reads::Relayed::Answered { status: None, .. }) => {}
+            return Ok(json!({
+                "state": "expired", "phase": "expired",
+                "reason": expiry.claim.body["fields"]["reason"],
+                "age_ms": now.saturating_sub(sent_at),
+                "recipient_delivery": delivery_presence::known(&message.to).map(|path| path.to_value()),
+            }));
+        }
+        return message_delivery_value(store, &message.to, &status, sent_at, now);
+    }
+    if let Some(host) = host {
+        match relayed {
+            Some(message_reads::Relayed::Answered { .. }) => {}
             Some(message_reads::Relayed::Unknown { reason, .. }) => {
                 return Ok(unknown(reason.clone()));
             }

crates/fractal-api-core/src/api/message_reads.rs

@@ -52,6 +52,8 @@ pub enum Relayed {
         host: String,
         held: bool,
         status: Option<&'static str>,
+        /// Furthest actual delivery/read mark, retained even when status is closed.
+        consumed: Option<&'static str>,
         latest: Option<(String, u64)>,
     },
     /// The host did not answer: the reason, naming it.
@@ -221,7 +223,7 @@ fn latest_of(answer: &MessageMarksAnswer, seats: &BTreeSet<String>) -> Option<(S
 /// How far `answer`'s marks took the message, keeping only those of seats this member places on
 /// `host`: a host speaks for its own seats. A staging counts, below every mark: a host that
 /// records its seats' stagings as marks answers them here, and nowhere else.
-fn status_of(answer: &MessageMarksAnswer, seats: &BTreeSet<String>) -> Option<&'static str> {
+fn status_of(answer: &MessageMarksAnswer, seats: &BTreeSet<String>) -> (Option<&'static str>, Option<&'static str>) {
     answer
         .marks
         .iter()
@@ -231,8 +233,13 @@ fn status_of(answer: &MessageMarksAnswer, seats: &BTreeSet<String>) -> Option<&'
                 .chain(&message_marks::MARKS)
                 .find(|kind| **kind == mark.kind)
         })
-        .max_by_key(|kind| message_marks::rank(kind.trim_start_matches("message.")))
         .map(|kind| kind.trim_start_matches("message."))
+        .fold((None, None), |(status, consumed), phase| {
+            let further = |previous: Option<&'static str>| Some(previous.filter(|previous| {
+                message_marks::rank(previous) >= message_marks::rank(phase)
+            }).unwrap_or(phase));
+            (further(status), if matches!(phase, "delivered" | "read") { further(consumed) } else { consumed })
+        })
 }
 
 /// Ask each host about its messages, all hosts at once, within `deadline`. Every message asked
@@ -348,7 +355,7 @@ type Heard = (String, Result<MessageMarksAnswer, (String, String)>);
 /// What each message's host said, as a status over the seats this member places on that host. A
 /// close a host answered is queued for the client selectors (`relayed_closes`), which this read
 /// does not wait for.
-fn settle(
+pub(super) fn settle(
     store: &Arc<Store>,
     heard: Vec<Heard>,
 ) -> BTreeMap<String, Relayed> {
@@ -361,12 +368,13 @@ fn settle(
                     seats_on(store, &host).map_err(|error| format!("{error:#}"))
                 });
                 let relayed = match on_host {
-                    Ok(on_host) => Relayed::Answered {
-                        host,
-                        held: answer.held,
-                        status: status_of(&answer, on_host),
-                        latest: latest_of(&answer, on_host),
-                    },
+                    Ok(on_host) => {
+                        let (status, consumed) = status_of(&answer, on_host);
+                        Relayed::Answered {
+                            host, held: answer.held, status, consumed,
+                            latest: latest_of(&answer, on_host),
+                        }
+                    }
                     Err(reason) => Relayed::Unknown {
                         host,
                         reason: reason.clone(),
@@ -380,7 +388,7 @@ fn settle(
     let closes = settled
         .iter()
         .filter_map(|(subject, relayed)| match relayed {
-            Relayed::Answered { host, held: true, status: Some("closed"), latest: Some((receipt, at)) } => {
+            Relayed::Answered { host, held: true, status: Some("closed"), latest: Some((receipt, at)), .. } => {
                 Some(relayed_closes::RelayedClose {
                     message: subject.clone(),
                     host: host.clone(),

crates/fractal-api-core/src/api/tests.rs

@@ -2102,6 +2102,122 @@ fn message_delivery_distinguishes_pending_acceptance_and_recipient_read() {
     );
 }
 
+#[tokio::test]
+async fn message_delivery_expiry_is_terminal_before_human_and_path_diagnostics() {
+    let root = tempfile::tempdir().unwrap();
+    let app = state(root.path());
+    let store = &app.store;
+    let seat = "agent/expiry-delivery-recipient";
+    let append = |subject: &str, kind: &str, actor: &str, fields: Value| {
+        store.append_claim(&ClaimInput {
+            subject: subject.into(), kind: kind.into(), actor: Some(actor.into()),
+            fields: serde_json::from_value(fields).unwrap(), evidence: vec![],
+            expected_subject: None, idempotency_key: None,
+        }).unwrap()
+    };
+    append(seat, "runtime.observed", seat, json!({"status":"running", "incarnation_id":"one"}));
+    append(seat, "harness.observed", seat, json!({
+        "state":"working", "driver":"claude", "incarnation_id":"one", "blocked_on":"human", "ask":"permission"
+    }));
+    let now = client_now_ms();
+    assert!(message_delivery_value(store, seat, "sent", now, now).unwrap()["reason"]
+        .as_str().unwrap().contains("held outside"));
+    delivery_presence::record(seat, &json!({
+        "transport":"omp-channel", "pid":4242, "ready":false, "reason":"native handoff refused"
+    }).to_string());
+    assert!(delivery_presence::known(seat).unwrap().stale());
+    for reason in ["cutoff-never-offered", "cutoff-uncertain-attempt", "no-live-recipient"] {
+        let subject = format!("message/{reason}");
+        let sent = append(&subject, "message.sent", "person/sender", json!({
+            "from":"person/sender", "to":seat, "content":"expiry contract", "status":"sent"
+        }));
+        append(&subject, "message.expired", "daemon/runtime", json!({"status":"expired", "reason":reason}));
+        let (status, view) = get_request(router(app.clone()), &format!("/v1/messages/delivery/{subject}")).await;
+        assert_eq!(status, StatusCode::OK, "{view}");
+        let delivery = &view["delivery"];
+        assert_eq!(delivery["state"], "expired", "{view}");
+        assert_eq!(delivery["phase"], "expired", "{view}");
+        assert_eq!(delivery["reason"], reason, "{view}");
+        assert!(delivery["age_ms"].is_number());
+        assert_eq!(delivery["recipient_delivery"]["state"], "stale");
+        assert!(delivery.get("history").is_none());
+        // A receipt arriving after the endpoint captured its view still wins.
+        let expired = store.message(&subject).unwrap().unwrap();
+        for phase in ["delivered", "read"] {
+            append(&subject, &format!("message.{phase}"), seat, json!({"status":phase}));
+            assert!(store.message_expiry(&subject).unwrap().unwrap().superseded);
+            let delivery = message_delivery_for(store, &expired, sent.accepted_at_unix_ms, now, None).unwrap();
+            assert_eq!(delivery["phase"], phase);
+            let (_, view) = get_request(router(app.clone()), &format!("/v1/messages/delivery/{subject}")).await;
+            assert_eq!(view["delivery"]["phase"], phase);
+            assert_ne!(view["delivery"]["state"], "expired");
+        }
+    }
+}
+
+#[test]
+fn message_delivery_expiry_survives_remote_unknown_but_not_real_relayed_receipts() {
+    let store = Arc::new(Store::open_memory("expiry-sender").unwrap());
+    let source = "version 2\nagent \"recipient\" { host \"expiry-remote\"; command \"true\"; }";
+    let intent = crate::graph::parse_intent(source, "expiry-sender").unwrap();
+    store.apply_internal(&intent, "expiry-remote-delivery").unwrap();
+    // The remote host's advertisement, as retained by replication.
+    {
+        let mut connection = store.connection.lock().unwrap();
+        let transaction = connection.transaction().unwrap();
+        crate::store::append_claim_tx(&transaction, "expiry-remote", "daemon/expiry-remote",
+            "daemon.started", Some("daemon/expiry-remote"), &json!({"fields":{
+                "status":"running", "features":{"seat_freshness_reads":1, "message_offsets":1}
+            }}), &[], None).unwrap();
+        transaction.commit().unwrap();
+    }
+    let seat = store.desired_subjects().unwrap().into_iter()
+        .find(|seat| seat.member.as_ref().is_some_and(|member| member.host == "expiry-remote"))
+        .unwrap().subject;
+    let subject = "message/expiry-relay";
+    let append = |kind: &str, actor: &str, fields: Value| {
+        store.append_claim(&ClaimInput {
+            subject: subject.into(), kind: kind.into(), actor: Some(actor.into()),
+            fields: serde_json::from_value(fields).unwrap(), evidence: vec![],
+            expected_subject: None, idempotency_key: None,
+        }).unwrap()
+    };
+    let sent = append("message.sent", "person/sender", json!({
+        "from":"person/sender", "to":seat, "content":"remote expiry", "status":"sent"
+    }));
+    append("message.expired", "daemon/runtime", json!({"status":"expired", "reason":"no-live-recipient"}));
+    let expired = store.message(subject).unwrap().unwrap();
+    assert_eq!(message_reads::relay_host(&store, &expired).unwrap().as_deref(), Some("expiry-remote"));
+    let unknown = message_reads::Relayed::Unknown { host:"expiry-remote".into(), reason:"unreachable".into() };
+    let delivery = |relayed: Option<&message_reads::Relayed>| message_delivery_for(
+        &store, &expired, sent.accepted_at_unix_ms, sent.accepted_at_unix_ms, relayed,
+    ).unwrap();
+    for relayed in [None, Some(&unknown)] {
+        assert_eq!(delivery(relayed)["state"], "expired");
+        assert_eq!(delivery(relayed)["reason"], "no-live-recipient");
+    }
+    for (phases, expected) in [
+        (&["delivered"][..], "delivered"), (&["read"][..], "read"),
+        (&["read", "closed"][..], "read"), (&["closed"][..], "expired"),
+    ] {
+        let marks = phases.iter().map(|phase| (format!("message.{phase}"), seat.clone()))
+            .chain(std::iter::once(("message.read".into(), "agent/not-on-remote".into())))
+            .map(|(kind, consumer)| crate::peer::RelayedMark {
+                kind, consumer, recorded_at_unix_ms: 1, receipt: "remote-receipt".into(),
+            }).collect();
+        let answers = message_reads::settle(&store, vec![("expiry-remote".into(), Ok(crate::peer::MessageMarksAnswer {
+            subject: subject.into(), held: true, marks,
+        }))]);
+        let answer = answers.get(subject).unwrap();
+        assert_eq!(answer.status(), phases.last().copied());
+        assert_eq!(delivery(Some(answer))["phase"], expected);
+        assert_eq!(delivery(Some(answer))["state"] == "expired", expected == "expired");
+        assert!(!store.message_expiry(subject).unwrap().unwrap().superseded);
+    }
+    append("message.read", &seat, json!({"status":"read"}));
+    assert_eq!(delivery(Some(&unknown))["state"], "read");
+}
+
 #[test]
 fn message_delivery_keeps_expired_claude_hook_mail_held_until_native_closure() {
     let store = Store::open_memory("expired-claude-hook-mail-hold").unwrap();

crates/fractal-client/tests/fixtures.rs

@@ -16,6 +16,23 @@ fn decode<T: serde::de::DeserializeOwned>(name: &str) -> T {
     serde_json::from_slice(&fixture(name)).unwrap_or_else(|error| panic!("decode {name}: {error}"))
 }
 
+#[test]
+fn sender_outcomes_keep_expired_delivered_and_read_as_open_string_states() {
+    let cases: serde_json::Value = decode("message-sender-outcomes.json");
+    let schema: serde_json::Value = serde_json::from_str(include_str!(
+        "../../../docs/st3/client-v0/schemas/client-v0.schema.json")).unwrap();
+    let validator = jsonschema::validator_for(&serde_json::json!({
+        "$defs":schema["$defs"], "$ref":"#/$defs/Page"
+    })).unwrap();
+    assert!(validator.is_valid(&cases));
+    for raw in cases["items"].as_array().unwrap() {
+        let message: Message = serde_json::from_value(raw.clone()).unwrap();
+        assert_eq!(message.state, raw["state"].as_str().unwrap());
+        let encoded = serde_json::to_value(message).unwrap();
+        assert_eq!(encoded["state"], raw["state"]);
+    }
+}
+
 #[test]
 fn person_option_notes_round_trip_and_empty_omission() {
     let legacy = serde_json::json!({"id": "land", "text": "Ship it."});

crates/fractal-daemon/src/peer_tests/seat_freshness_reads.rs

@@ -299,6 +299,7 @@ async fn a_viewer_reads_a_relay_hosts_seat_from_it_and_unknown_when_it_does_not_
             host: OWNER.into(),
             held: true,
             status: Some("read"),
+            consumed: Some("read"),
             latest: answered.get(&message.subject).and_then(|relayed| match relayed {
                 Relayed::Answered { latest, .. } => latest.clone(),
                 Relayed::Unknown { .. } => None,

crates/fractal-store/src/store/unread_mail.rs

@@ -125,14 +125,21 @@ pub fn create_schema(connection: &Connection) -> Result<()> {
         )?;
     }
     // Upgrade only expiry subjects once, not retained sent history on each open/read.
-    connection.execute_batch(
-        /* sqlite-large-value: allow-for-now lv-h5a7c3523223f owner=fractal_store::store::unread_mail TODO=prepare-body-and-predecessor-values */ "BEGIN IMMEDIATE;
-         INSERT OR IGNORE INTO unread_mail_pending
-         SELECT subject FROM claims WHERE kind='message.expired'
-           AND NOT EXISTS(SELECT 1 FROM meta WHERE key='unread_mail_expiry_v1');
-         INSERT OR IGNORE INTO meta(key,value) VALUES('unread_mail_expiry_v1','1');
-         COMMIT;",
+    let expiry_migrated: bool = connection.query_row(
+        "SELECT EXISTS(SELECT 1 FROM meta WHERE key='unread_mail_expiry_v1')",
+        [],
+        |row| row.get(0),
     )?;
+    if !expiry_migrated {
+        connection.execute_batch(
+            /* sqlite-large-value: allow-for-now lv-hb7fe422f0e26 owner=fractal_store::store::unread_mail TODO=prepare-body-and-predecessor-values */ "BEGIN IMMEDIATE;
+             INSERT OR IGNORE INTO unread_mail_pending
+             SELECT subject FROM claims WHERE kind='message.expired'
+               AND NOT EXISTS(SELECT 1 FROM meta WHERE key='unread_mail_expiry_v1');
+             INSERT OR IGNORE INTO meta(key,value) VALUES('unread_mail_expiry_v1','1');
+             COMMIT;",
+        )?;
+    }
     Ok(())
 }
 

crates/fractal/src/conversations_cli.rs

@@ -298,7 +298,8 @@ pub(super) async fn run_message(
                         let unknown = message
                             .status_from
                             .as_ref()
-                            .is_some_and(|from| from.state == "unknown");
+                            .is_some_and(|from| from.state == "unknown")
+                            && message.status != "expired";
                         rows.push(format!(
                             "{}\t{}\t{}\t{}",
                             message.subject,