From 418b225159e5fe7feaa6b9a73c48dbb1203693ba Mon Sep 17 00:00:00 2001 From: schickling-assistant <261620128+schickling-assistant@users.noreply.github.com> Date: Tue, 10 Nov 2026 15:01:12 +0100 Subject: [PATCH 1/2] Expose durable expiry on sender delivery status --- .../fractal-client/Models.generated.ts | 2 +- .../fractal-client/ReadSchema.generated.ts | 2 +- crates/fractal-api-core/src/api.rs | 40 ++++-- .../fractal-api-core/src/api/message_reads.rs | 28 +++-- crates/fractal-api-core/src/api/tests.rs | 116 ++++++++++++++++++ crates/fractal-client/tests/fixtures.rs | 17 +++ .../src/peer_tests/seat_freshness_reads.rs | 1 + crates/fractal-store/src/store/unread_mail.rs | 21 ++-- crates/fractal/src/conversations_cli.rs | 3 +- crates/fractal/tests/client_v0_cli.rs | 63 ++++++++++ docs/st3/client-v0/fixtures/manifest.json | 1 + .../fixtures/message-sender-outcomes.json | 9 ++ .../client-v0/schemas/client-v0.schema.json | 2 +- docs/st3/seat-deploys.md | 15 +++ release-notes/mail-expiry-sender.json | 11 ++ 15 files changed, 303 insertions(+), 28 deletions(-) create mode 100644 docs/st3/client-v0/fixtures/message-sender-outcomes.json create mode 100644 release-notes/mail-expiry-sender.json diff --git a/clients/typescript/fractal-client/Models.generated.ts b/clients/typescript/fractal-client/Models.generated.ts index 688f640cfa2..bb8c1f4ce46 100644 --- a/clients/typescript/fractal-client/Models.generated.ts +++ b/clients/typescript/fractal-client/Models.generated.ts @@ -2360,7 +2360,7 @@ export type Message = ResourceHeader & { kind: "message"; sent_at: Timestamp; session_id?: (Id | null); - state: "sent" | "delivered" | "read" | "closed"; + state: "sent" | "delivered" | "read" | "closed" | "expired"; status_from?: MessageStatusFrom; tags?: Array; title?: string | null; diff --git a/clients/typescript/fractal-client/ReadSchema.generated.ts b/clients/typescript/fractal-client/ReadSchema.generated.ts index 609e7bc5baf..efc83260d41 100644 --- a/clients/typescript/fractal-client/ReadSchema.generated.ts +++ b/clients/typescript/fractal-client/ReadSchema.generated.ts @@ -2276,7 +2276,7 @@ export const Message = /*#__PURE__*/ (() => Schema.Struct({ "revision": Revision, "sent_at": Timestamp, "session_id": Schema.OptionFromOptionalNullOr(Id, NULL_NONE), - "state": Schema.Literals(["sent","delivered","read","closed"]), + "state": Schema.Literals(["sent","delivered","read","closed","expired"]), /** Present when the recipient's seat runs on another host that keeps its receipts: whether that host answered. Absent when this host's own records are the whole answer. */ "status_from": optionalKey(MessageStatusFrom), "tags": optionalKey(Schema.Array(Schema.String).check(Schema.isUnique())), diff --git a/crates/fractal-api-core/src/api.rs b/crates/fractal-api-core/src/api.rs index 3e69887258d..167ddeb1ac4 100644 --- a/crates/fractal-api-core/src/api.rs +++ b/crates/fractal-api-core/src/api.rs @@ -5395,14 +5395,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(¤t.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())); } diff --git a/crates/fractal-api-core/src/api/message_reads.rs b/crates/fractal-api-core/src/api/message_reads.rs index 05163283781..35640e5e2a9 100644 --- a/crates/fractal-api-core/src/api/message_reads.rs +++ b/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) -> 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) -> Option<&'static str> { +fn status_of(answer: &MessageMarksAnswer, seats: &BTreeSet) -> (Option<&'static str>, Option<&'static str>) { answer .marks .iter() @@ -231,8 +233,13 @@ fn status_of(answer: &MessageMarksAnswer, seats: &BTreeSet) -> 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); /// 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, heard: Vec, ) -> BTreeMap { @@ -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(), diff --git a/crates/fractal-api-core/src/api/tests.rs b/crates/fractal-api-core/src/api/tests.rs index d0fc26555a1..890f39c7200 100644 --- a/crates/fractal-api-core/src/api/tests.rs +++ b/crates/fractal-api-core/src/api/tests.rs @@ -1952,6 +1952,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(); diff --git a/crates/fractal-client/tests/fixtures.rs b/crates/fractal-client/tests/fixtures.rs index 33ec5385f56..2d29f88ef88 100644 --- a/crates/fractal-client/tests/fixtures.rs +++ b/crates/fractal-client/tests/fixtures.rs @@ -16,6 +16,23 @@ fn decode(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 related_run_rows_decode_and_validate_server_semantics() { let value: serde_json::Value = serde_json::from_slice(&fixture("agent-related-runs.json")).unwrap(); diff --git a/crates/fractal-daemon/src/peer_tests/seat_freshness_reads.rs b/crates/fractal-daemon/src/peer_tests/seat_freshness_reads.rs index c21d22cf02e..e7b6534387a 100644 --- a/crates/fractal-daemon/src/peer_tests/seat_freshness_reads.rs +++ b/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, diff --git a/crates/fractal-store/src/store/unread_mail.rs b/crates/fractal-store/src/store/unread_mail.rs index 2ab7ca70299..9fcf0902ee8 100644 --- a/crates/fractal-store/src/store/unread_mail.rs +++ b/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( - "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( + "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(()) } diff --git a/crates/fractal/src/conversations_cli.rs b/crates/fractal/src/conversations_cli.rs index 3073d801414..66f892a3932 100644 --- a/crates/fractal/src/conversations_cli.rs +++ b/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, diff --git a/crates/fractal/tests/client_v0_cli.rs b/crates/fractal/tests/client_v0_cli.rs index c49d5b7eef6..a3941c6bad9 100644 --- a/crates/fractal/tests/client_v0_cli.rs +++ b/crates/fractal/tests/client_v0_cli.rs @@ -39,6 +39,69 @@ fn test_state(root: &Path) -> AppState { } } +#[tokio::test] +async fn sender_outcome_fixtures_render_raw_status_and_delivery_reason() { + use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _}; + let cases: Value = serde_json::from_str(include_str!( + "../../../docs/st3/client-v0/fixtures/message-sender-outcomes.json")).unwrap(); + for message in cases["items"].as_array().unwrap() { + let phase = message["state"].as_str().unwrap(); + let reason = match phase { + "expired" => "cutoff-uncertain-attempt", + "delivered" => "the native transport accepted the message; waiting for recipient read", + "read" => "the recipient has read the message", + _ => panic!("unexpected fixture state"), + }; + let case = serde_json::json!({ + "id":message["id"], "subject":message["id"], "from":message["from"], "to":message["to"], + "content":message["content"], "created_index":1, "status":phase, "status_from":message["status_from"], + "delivery":{"phase":phase, "state":if phase == "delivered" { "pending" } else { phase }, "reason":reason} + }); + for list in [false, true] { + for json in [false, true] { + let root = tempfile::tempdir().unwrap(); + let socket = root.path().join("sender.sock"); + let listener = tokio::net::UnixListener::bind(&socket).unwrap(); + let response = if list { + serde_json::json!({"items":[case], "limit":100, "has_more":false, "next_cursor":null}) + } else { case.clone() }; + let response = serde_json::json!({"api_version":"fractal.v1", "value":response}).to_string(); + let server = tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.unwrap(); + let mut request = Vec::new(); + while !request.windows(4).any(|part| part == b"\r\n\r\n") { + let mut byte = [0]; + assert_eq!(stream.read(&mut byte).await.unwrap(), 1); + request.push(byte[0]); + } + stream.write_all(format!( + "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{}", + response.len(), response).as_bytes()).await.unwrap(); + }); + let args = if list { vec!["conversations", "ls", "agent/sender"] } + else { vec!["conversations", "status", case["id"].as_str().unwrap()] }; + let output = if json { run_cli(&socket, &args).await } + else { run_cli_human(&socket, &args).await }; + assert!(output.status.success(), "{}", failure(&output)); + if json { + let value = value(&output); + if list { assert_eq!(value[0]["status"], case["status"]); } + else { assert_eq!(value["delivery"], case["delivery"]); } + } else { + let text = String::from_utf8(output.stdout).unwrap(); + assert!(text.contains(case["status"].as_str().unwrap()), "{text}"); + if !list { + assert!(text.contains(case["delivery"]["reason"].as_str().unwrap()), "{text}"); + } else { + assert!(!text.contains("unknown"), "{text}"); + } + } + server.await.unwrap(); + } + } + } +} + #[tokio::test] async fn ordinary_agent_list_and_tree_cli_send_no_freshness_barrier() { use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _}; diff --git a/docs/st3/client-v0/fixtures/manifest.json b/docs/st3/client-v0/fixtures/manifest.json index ff13c8e9ec1..998eb7b9a9d 100644 --- a/docs/st3/client-v0/fixtures/manifest.json +++ b/docs/st3/client-v0/fixtures/manifest.json @@ -20,6 +20,7 @@ { "file": "mission-detail-step-history.json", "definition": "Mission" }, { "file": "mission-declaration-provenance.json", "definition": "MissionDeclaration" }, { "file": "conversation-search.json", "definition": "Envelope" }, + { "file": "message-sender-outcomes.json", "definition": "Page" }, { "file": "timeline.json", "definition": "Envelope" }, { "file": "timeline-views.json", "definition": "Envelope" }, { "file": "timeline-conformance.json", "definition": "TimelinePage" }, diff --git a/docs/st3/client-v0/fixtures/message-sender-outcomes.json b/docs/st3/client-v0/fixtures/message-sender-outcomes.json new file mode 100644 index 00000000000..9658b08e846 --- /dev/null +++ b/docs/st3/client-v0/fixtures/message-sender-outcomes.json @@ -0,0 +1,9 @@ +{ + "kind":"page", "collection":"messages", "filters":{}, + "items":[ + {"kind":"message","id":"message/expired","revision":"1","updated_at":"2026-11-10T00:00:00Z","from":"agent/sender","to":"agent/recipient","content":"Fixture","state":"expired","sent_at":"2026-11-10T00:00:00Z","status_from":{"host":"host/offline","state":"unknown","reason":"remote-unavailable"}}, + {"kind":"message","id":"message/delivered","revision":"2","updated_at":"2026-11-10T00:00:00Z","from":"agent/sender","to":"agent/recipient","content":"Fixture","state":"delivered","sent_at":"2026-11-10T00:00:00Z"}, + {"kind":"message","id":"message/read","revision":"3","updated_at":"2026-11-10T00:00:00Z","from":"agent/sender","to":"agent/recipient","content":"Fixture","state":"read","sent_at":"2026-11-10T00:00:00Z"} + ], + "page":{"limit":100,"has_more":false} +} diff --git a/docs/st3/client-v0/schemas/client-v0.schema.json b/docs/st3/client-v0/schemas/client-v0.schema.json index 00ec535ae5c..42881f66389 100644 --- a/docs/st3/client-v0/schemas/client-v0.schema.json +++ b/docs/st3/client-v0/schemas/client-v0.schema.json @@ -1201,7 +1201,7 @@ "to": { "$ref": "#/$defs/Id" }, "title": { "type": ["string", "null"] }, "content": { "type": "string" }, - "state": { "enum": ["sent", "delivered", "read", "closed"] }, + "state": { "enum": ["sent", "delivered", "read", "closed", "expired"] }, "sent_at": { "$ref": "#/$defs/Timestamp" }, "session_id": { "oneOf": [{ "$ref": "#/$defs/Id" }, { "type": "null" }] }, "in_reply_to": { "oneOf": [{ "$ref": "#/$defs/Id" }, { "type": "null" }] }, diff --git a/docs/st3/seat-deploys.md b/docs/st3/seat-deploys.md index a96296a4cf4..bd5bfcc6060 100644 --- a/docs/st3/seat-deploys.md +++ b/docs/st3/seat-deploys.md @@ -173,9 +173,24 @@ mail remains unread. The expiry claim stays in history. Harness receipts cannot write expiry, and the messages shard still replicates only `message.sent`. Outcome readers, projections and sender surfaces must be deployed to every pinned host before any expiry writer is enabled; older registries retain the new kind as unknown rather than interpreting its outcome. +Writer activation also requires upgraded TypeScript clients, including fractal-web and generated +consumers, whose message-state decoder accepts `expired`. This is a prerequisite for the +expiry-writer rollout, not a negotiated response format. The writer must also enforce original-accepting-member authority before activation; replicated expiry admission here checks the daemon actor, not that future writer prerequisite. +`GET /v1/messages/delivery/:id` reports active expiry as terminal `state: expired`, +`phase: expired`, with the durable reason above, before human-hold or delivery-path +diagnostics. An unavailable remote receipt host cannot replace that outcome with +`unknown`. Actual delivered/read evidence supersedes expiry and reports normal receipt +progress; this endpoint does not add expiry history. `fractal conversations status` +prints the outcome and reason; `fractal conversations ls` shows the raw message status. +Rust's generated string field can receive the raw `expired` status without regeneration. +Older TypeScript `ReadSchema.Message.state` decoders use a closed literal union and reject +`expired`, even in tolerant mode; upgrade those clients before enabling any expiry writer. +Upgrading the generated union may require consumers to add an exhaustive `expired` arm. +No harness receipt or automatic expiry writer is changed. + Claude distinguishes native queue acceptance (`queue-operation/enqueue`, which records `delivered`) from consumption (a user transcript entry, including `isMeta`, or an explicit `queue-operation/remove` with `reason: absorbed_mid_turn`, which records `read`). Receipt matching diff --git a/release-notes/mail-expiry-sender.json b/release-notes/mail-expiry-sender.json new file mode 100644 index 00000000000..e7175e72d2e --- /dev/null +++ b/release-notes/mail-expiry-sender.json @@ -0,0 +1,11 @@ +{ + "version": 1, + "summary": "Report durable mail expiry on sender delivery status without enabling expiry writers", + "replay": {"status": "none", "detail": "No replay or new receipt is introduced. Real delivered/read evidence supersedes expiry."}, + "schema": {"status": "changed", "detail": "The client Message.state schema accepts expired. Generated string-based message models stay compatible; the registry and receipt wire formats are unchanged."}, + "rules": {"status": "changed", "detail": "Sender delivery status reports terminal expired state and phase with its durable reason before hold/path diagnostics; unavailable remote receipt hosts cannot replace it with unknown. The completed unread expiry migration skips its extra write transaction on reopen."}, + "compatibility": {"status": "changed", "detail": "Rust string-based message models accept expired. Older TypeScript Message.state decoders reject it even in tolerant mode: upgrade fractal-web and generated TypeScript consumers before enabling any expiry writer, alongside sender and mailbox readers. Exhaustive downstream state switches may need an expired arm. No negotiation mechanism, writer or harness receipt change is included."}, + "downtime": {"status": "none", "detail": "No operator downtime is required."}, + "manual": {"status": "none", "detail": "No operator action is required."}, + "recovery": {"status": "none", "detail": "Expiry remains durable history; this sender endpoint adds no history field or recovery action."} +} From d031251ba62a231e0528bb94aed5450c094e8517 Mon Sep 17 00:00:00 2001 From: schickling-assistant <261620128+schickling-assistant@users.noreply.github.com> Date: Tue, 10 Nov 2026 21:14:52 +0100 Subject: [PATCH 2/2] chore: refresh merged mail expiry SQL fingerprint --- crates/fractal-store/src/store/unread_mail.rs | 2 +- docs/st3/large-value-sites.json | 2 +- release-notes/6128-pr3-merged-sqlite-inventory.json | 11 +++++++++++ 3 files changed, 13 insertions(+), 2 deletions(-) create mode 100644 release-notes/6128-pr3-merged-sqlite-inventory.json diff --git a/crates/fractal-store/src/store/unread_mail.rs b/crates/fractal-store/src/store/unread_mail.rs index 9495f2294ce..c39539fec08 100644 --- a/crates/fractal-store/src/store/unread_mail.rs +++ b/crates/fractal-store/src/store/unread_mail.rs @@ -132,7 +132,7 @@ pub fn create_schema(connection: &Connection) -> Result<()> { )?; if !expiry_migrated { 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; + /* 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'); diff --git a/docs/st3/large-value-sites.json b/docs/st3/large-value-sites.json index 43f7529e25b..379ce824b05 100644 --- a/docs/st3/large-value-sites.json +++ b/docs/st3/large-value-sites.json @@ -2336,7 +2336,6 @@ {"id": "lv-h5a3d29781577", "path": "crates/smallclaims/src/store.rs", "kind": "unresolved-sql-sink", "fingerprint": "8c48a6dcfa06487103b62cf1fa6a87f5a79dc91b4660d687cf65cb5ca5679218", "owner": "smallclaims::store", "todo": "trace-statement-and-value-representation", "disposition": "allow-for-now", "line_hint": 9337, "excerpt": "& canonical_sql ( \"WITH RECURSIVE pairs(id) AS ( SELECT (SELECT id FROM claims INDEXED BY claims_transport_pair_accepted_index WHERE kind='transport.observed' ORDER BY origin, subject LIMIT 1) UNION A"}, {"id": "lv-h5a47c2ff7e59", "path": "crates/fractal-store/src/store/client_messages.rs", "kind": "unresolved-sql-sink", "fingerprint": "a2fc0ae6abe631f978a0979e040a1aa21894fdadcced4e72c0b5ea24c538ddad", "owner": "fractal_store::store::client_messages", "todo": "trace-statement-and-value-representation", "disposition": "allow-for-now", "line_hint": 1458, "excerpt": "& SQL [ flags ]"}, {"id": "lv-h5a7af5d0161f", "path": "crates/smallclaims/src/store/checkpoint.rs", "kind": "unresolved-sql-sink", "fingerprint": "3b691426c38c73cbdb64b8aca2b74b9dda8dbada750f6e3524d4df22717e0d32", "owner": "smallclaims::store::checkpoint", "todo": "trace-statement-and-value-representation", "disposition": "allow-for-now", "line_hint": 369, "excerpt": "params ! [ claim . id , claim . writer , claim . sequence , claim . envelope_hash , claim . subject , claim . kind , claim . actor , serde_json : : to_string ( & claim . predecessors ) ? , claim . ope"}, - {"id": "lv-h5a7c3523223f", "path": "crates/fractal-store/src/store/unread_mail.rs", "kind": "sql-template", "fingerprint": "051d3516e53dac393a02410b69c8b1a1b9faca05199ca2041bbf58331a540b8b", "owner": "fractal_store::store::unread_mail", "todo": "prepare-body-and-predecessor-values", "disposition": "allow-for-now", "line_hint": 129, "excerpt": "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 IGNOR"}, {"id": "lv-h5a7d3874506e", "path": "crates/fractal-store/src/store/messages_shard/stream_operations.rs", "kind": "unresolved-sql-sink", "fingerprint": "45a9482391c19256a64fb07899c22a6263cefd6b10cd1e1129a6d31eff97abeb", "owner": "fractal_store::store::messages_shard::stream_operations", "todo": "trace-statement-and-value-representation", "disposition": "allow-for-now", "line_hint": 32, "excerpt": "STREAM_OPERATION , [ operation_id ] , | row | Ok ( ( row . get ( 0 ) ? , row . get ( 1 ) ? ) ) ,"}, {"id": "lv-h5a7e8f605584", "path": "crates/fractal-store/src/store/client_messages.rs", "kind": "unresolved-sql-sink", "fingerprint": "06a53d879d2dc5fb7aae823f9e2aa3329b966bdf4527bd6fa6adcde760a9eb56", "owner": "fractal_store::store::client_messages", "todo": "trace-statement-and-value-representation", "disposition": "allow-for-now", "line_hint": 341, "excerpt": "[ subject ] , | row | Ok ( ( row . get ( 0 ) ? , row . get ( 1 ) ? , row . get ( 2 ) ? , row . get ( 3 ) ? ) )"}, {"id": "lv-h5a84af9fc3b5", "path": "crates/fractal-store/src/store/terminal_gate_evidence/native_source/transaction_hooks/tests.rs", "kind": "sql-template", "fingerprint": "144fd2809504df03bc19ebb5323c703f11350b2f547cc4cfda4c760cb72f2d3e", "owner": "fractal_store::store::terminal_gate_evidence::native_source::transaction_hooks::tests", "todo": "prepare-body-and-predecessor-values", "disposition": "allow-for-now", "line_hint": 453, "excerpt": "UPDATE claims SET body=body WHERE id='foreign/1'"}, @@ -3913,6 +3912,7 @@ {"id": "lv-hb7cf1dc01667", "path": "crates/fractal-api-core/src/api/published_lists_tests.rs", "kind": "sql-template", "fingerprint": "39c0ce46943adc6d3fa133a261008f77c00d77aa927d346dd8295ee44b6ef104", "owner": "fractal_api_core::api::published_lists_tests", "todo": "bound-or-reference-values-and-reclamation", "disposition": "allow-for-now", "line_hint": 363, "excerpt": "UPDATE step_runs SET assignee=?2, created_at_unix_ms=?3 WHERE subject=?1"}, {"id": "lv-hb7ed2976f44f", "path": "crates/fractal-store/src/store.rs", "kind": "unresolved-sql-sink", "fingerprint": "d1f5ab3e6b8522867ac512eae576dbb7ef1e4800b4dbf53a0ec146f1b4054166", "owner": "fractal_store::store", "todo": "trace-statement-and-value-representation", "disposition": "allow-for-now", "line_hint": 18506, "excerpt": "params ! [ claim . subject , to ] ,"}, {"id": "lv-hb7f8923255f2", "path": "crates/smallclaims/src/sqlite/read_lifetime/tests.rs", "kind": "sql-template", "fingerprint": "69029be48c65949b99471da0adf147accfc0c78d768e7131ce9386c27e664aff", "owner": "smallclaims::sqlite::read_lifetime::tests", "todo": "bound-or-reference-values-and-reclamation", "disposition": "allow-for-now", "line_hint": 81, "excerpt": "UPDATE payload SET value=zeroblob(8192)"}, + {"id": "lv-hb7fe422f0e26", "path": "crates/fractal-store/src/store/unread_mail.rs", "kind": "sql-template", "fingerprint": "77a9e33d563de935512f2a40164cf10739760e2441a4a60559f5ece640cc91a4", "owner": "fractal_store::store::unread_mail", "todo": "prepare-body-and-predecessor-values", "disposition": "allow-for-now", "line_hint": 135, "excerpt": "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 IGNOR"}, {"id": "lv-hb806bfc38d25", "path": "crates/smallclaims/src/store/transport_observation_index.rs", "kind": "sql-template", "fingerprint": "0eb11da108f89988de47a90451cafe08bc166a8ab47c59147718245906dfdb64", "owner": "smallclaims::store::transport_observation_index", "todo": "chunk-or-reference-variable-columns", "disposition": "allow-for-now", "line_hint": 26, "excerpt": " CREATE TABLE IF NOT EXISTS transport_observation_index ( claim_id TEXT PRIMARY KEY, subject TEXT NOT NULL, origin TEXT NOT NULL, accepted_at_unix_ms TEXT NOT NULL ); -- This CREATE INDEX only ever se"}, {"id": "lv-hb81297962ca8", "path": "crates/fractal-store/src/store.rs", "kind": "unresolved-sql-sink", "fingerprint": "0bc36077f58b59bfd143fdd0714e71048ebe87fdd42d898da54ddbeaac8a1ae6", "owner": "fractal_store::store", "todo": "trace-statement-and-value-representation", "disposition": "allow-for-now", "line_hint": 15287, "excerpt": "params ! [ step , crate : : mission : : AFTER_RUN_STEP ] , | row | Ok ( ( row . get ( 0 ) ? , row . get ( 1 ) ? ) ) ,"}, {"id": "lv-hb8151383f885", "path": "crates/smallclaims/src/store/repair_operation_tests.rs", "kind": "sql-template", "fingerprint": "acfc05f2296a799e0451c895cdb246716834f6fdf0034b52f9b8e8c56065cc89", "owner": "smallclaims::store::repair_operation_tests", "todo": "bound-or-reference-values-and-reclamation", "disposition": "allow-for-now", "line_hint": 641, "excerpt": "CREATE TEMP TRIGGER fail_late_operation BEFORE INSERT ON operations WHEN NEW.id='op/z' BEGIN SELECT RAISE(ABORT,'late insert failure'); END;"}, diff --git a/release-notes/6128-pr3-merged-sqlite-inventory.json b/release-notes/6128-pr3-merged-sqlite-inventory.json new file mode 100644 index 00000000000..fd37c950f53 --- /dev/null +++ b/release-notes/6128-pr3-merged-sqlite-inventory.json @@ -0,0 +1,11 @@ +{ + "version": 1, + "summary": "Refresh the SQLite exception fingerprint after the mail expiry migration guard merge", + "replay": {"status": "none", "detail": "Inventory metadata and source marker only; replay is unchanged."}, + "schema": {"status": "none", "detail": "No database schema or inventory format changes."}, + "rules": {"status": "none", "detail": "No runtime behavior changes; the moved SQL literal retains its allow-for-now exception without size proof."}, + "compatibility": {"status": "none", "detail": "No wire format or persisted state changes."}, + "downtime": {"status": "none", "detail": "No deployment or migration required."}, + "manual": {"status": "none", "detail": "No operator action required."}, + "recovery": {"status": "none", "detail": "No persistent runtime state changes; no recovery action required."} +}