diff --git a/clients/typescript/fractal-client/Models.generated.ts b/clients/typescript/fractal-client/Models.generated.ts index 19a491ae870..5df80b08923 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 b6b6a73dc95..9ab8e4e9c47 100644 --- a/clients/typescript/fractal-client/ReadSchema.generated.ts +++ b/clients/typescript/fractal-client/ReadSchema.generated.ts @@ -2284,7 +2284,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 6ab95ba9567..d3be51551b6 100644 --- a/crates/fractal-api-core/src/api.rs +++ b/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(¤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 769b22d4ef4..fb1af0f061c 100644 --- a/crates/fractal-api-core/src/api/tests.rs +++ b/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(); diff --git a/crates/fractal-client/tests/fixtures.rs b/crates/fractal-client/tests/fixtures.rs index a1e6a329224..b771f876dfe 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 person_option_notes_round_trip_and_empty_omission() { let legacy = serde_json::json!({"id": "land", "text": "Ship it."}); 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 473d4628b8e..c39539fec08 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( - /* 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(()) } 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 3bc139b4994..3bba0a195cb 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 3daf13f9002..2ab25c3e50c 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/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/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/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."} +} 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."} +}