diff --git a/crates/cccc-cli/src/args/messaging.rs b/crates/cccc-cli/src/args/messaging.rs index 91d42a0b1..89055e76f 100644 --- a/crates/cccc-cli/src/args/messaging.rs +++ b/crates/cccc-cli/src/args/messaging.rs @@ -148,6 +148,19 @@ pub struct TailArgs { pub follow: bool, } +#[derive(Debug, Args)] +pub struct InboxSummaryArgs { + /// Target Group ID; omit to use the current Group. + #[arg(long = "group")] + pub group_id: Option, + /// Inclusive original-Mail timestamp boundary (RFC3339, not in the future). + #[arg(long)] + pub cutoff: String, + /// Author identity: user or the Group's Foreman. + #[arg(long, default_value = "user")] + pub by: String, +} + #[derive(Debug, Args)] pub struct InboxArgs { /// Target Group ID; omit to use the current Group. diff --git a/crates/cccc-cli/src/args/mod.rs b/crates/cccc-cli/src/args/mod.rs index b33b513bd..bc01d47e2 100644 --- a/crates/cccc-cli/src/args/mod.rs +++ b/crates/cccc-cli/src/args/mod.rs @@ -12,8 +12,8 @@ pub use integrations::{ SpaceCredentialAction, SpaceJobsAction, }; pub use messaging::{ - CancelReplyArgs, ConnectArgs, DeliverArgs, InboxArgs, LedgerAction, LedgerArgs, ReplyArgs, - SendArgs, TailArgs, TrackedSendArgs, + CancelReplyArgs, ConnectArgs, DeliverArgs, InboxArgs, InboxSummaryArgs, LedgerAction, + LedgerArgs, ReplyArgs, SendArgs, TailArgs, TrackedSendArgs, }; use clap::{Args, Parser, Subcommand, ValueEnum}; @@ -137,6 +137,8 @@ pub enum CommandKind { Space(SpaceArgs), /// Read and consume the next unread Mail batch for an actor. Inbox(InboxArgs), + /// Summarize current awaiting Mail across a Group without consuming it. + InboxSummary(InboxSummaryArgs), /// Read cached same-account and Group connection directories. Does not connect or start Actors. Connect(ConnectArgs), /// Send a message to a Group. diff --git a/crates/cccc-cli/src/commands/messaging.rs b/crates/cccc-cli/src/commands/messaging.rs index be79097d5..31753be77 100644 --- a/crates/cccc-cli/src/commands/messaging.rs +++ b/crates/cccc-cli/src/commands/messaging.rs @@ -4,8 +4,8 @@ use cccc_core::HomeLayout; use serde_json::json; use crate::args::{ - CancelReplyArgs, ConnectArgs, DeliverArgs, InboxArgs, LedgerAction, LedgerArgs, ReplyArgs, - SendArgs, TailArgs, TrackedSendArgs, + CancelReplyArgs, ConnectArgs, DeliverArgs, InboxArgs, InboxSummaryArgs, LedgerAction, + LedgerArgs, ReplyArgs, SendArgs, TailArgs, TrackedSendArgs, }; use crate::commands::common::{call, group, print}; @@ -209,6 +209,21 @@ pub async fn tail(client: &DaemonClient, home: &HomeLayout, args: TailArgs) -> R } } +pub async fn inbox_summary( + client: &DaemonClient, + home: &HomeLayout, + args: InboxSummaryArgs, +) -> Result<()> { + let group_id = group(home, args.group_id)?; + let response = call( + client, + "inbox_summary", + json!({"group_id":group_id,"cutoff":args.cutoff,"by":args.by}), + ) + .await?; + print(response) +} + pub async fn inbox(client: &DaemonClient, home: &HomeLayout, args: InboxArgs) -> Result<()> { let group_id = group(home, args.group_id)?; let response = call( diff --git a/crates/cccc-cli/src/main.rs b/crates/cccc-cli/src/main.rs index 47c8b1cad..e0c437da6 100644 --- a/crates/cccc-cli/src/main.rs +++ b/crates/cccc-cli/src/main.rs @@ -143,6 +143,9 @@ async fn main() -> Result<()> { } Some(CommandKind::Tail(args)) => commands::messaging::tail(&client, &home, args).await, Some(CommandKind::Inbox(args)) => commands::messaging::inbox(&client, &home, args).await, + Some(CommandKind::InboxSummary(args)) => { + commands::messaging::inbox_summary(&client, &home, args).await + } Some(CommandKind::Ledger(args)) => commands::messaging::ledger(&client, &home, args).await, Some(CommandKind::Daemon { action }) => daemon(action, home, &client).await, Some(CommandKind::Runtime { action }) => runtime(&client, action).await, diff --git a/crates/cccc-cli/tests/inbox_projection.rs b/crates/cccc-cli/tests/inbox_projection.rs index dbb0e5982..23323d056 100644 --- a/crates/cccc-cli/tests/inbox_projection.rs +++ b/crates/cccc-cli/tests/inbox_projection.rs @@ -2,7 +2,7 @@ use cccc_client::DaemonClient; use cccc_contracts::DaemonRequest; -use cccc_core::HomeLayout; +use cccc_core::{GroupStore, HomeLayout, ledger}; use serde_json::{Value, json}; use std::process::{Child, Command, Stdio}; use std::time::{Duration, Instant}; @@ -29,6 +29,351 @@ async fn call(client: &DaemonClient, op: &str, args: Value) -> Value { Value::Object(response.result) } +fn summary_row<'a>(summary: &'a Value, actor_id: &str) -> &'a Value { + summary["actors"] + .as_array() + .expect("actors") + .iter() + .find(|actor| actor["actor_id"] == actor_id) + .expect("actor row") +} + +#[tokio::test] +async fn overdue_summary_distinguishes_worker_mail_delivery_from_notice_delivery() { + let temp = tempfile::tempdir().expect("tempdir"); + let home = HomeLayout::from_path(temp.path().join("home")).expect("home"); + home.initialize().expect("initialize"); + let mut daemon = Daemon( + Command::new(env!("CARGO_BIN_EXE_cccc")) + .args(["daemon", "run", "--host", "127.0.0.1", "--port", "0"]) + .env("CCCC_HOME", home.root()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .expect("daemon"), + ); + let deadline = Instant::now() + Duration::from_secs(20); + while !home.daemon_dir().join("ccccd.addr.json").exists() { + assert!(daemon.0.try_wait().expect("check daemon status").is_none()); + assert!(Instant::now() < deadline); + tokio::time::sleep(Duration::from_millis(20)).await; + } + let client = DaemonClient::new(home.clone()).with_timeout(Duration::from_secs(30)); + let runtime_home = home.clone(); + // Always shut down the real daemon (and its owned local actor), including + // when a test assertion panics. No provider or external account is used. + let outcome = tokio::spawn(async move { + let created = call(&client, "group_create", json!({"title":"Delivery summary","by":"user"})).await; + let group = created["group"]["group_id"].as_str().expect("created group ID"); + let project = runtime_home.root().join("project"); + std::fs::create_dir(&project).expect("create fixture project directory"); + call(&client, "attach", json!({"group_id":group,"path":project,"by":"user"})).await; + call(&client, "group_preamble_set", json!({"group_id":group,"content":"Use the delivery protocol.","by":"user"})).await; + call(&client, "actor_add", json!({"group_id":group,"actor_id":"reader","runtime":"custom","submit":"newline", + "command":["sh","-c","stty -echo -icanon min 1 time 0; while IFS= read -r line; do printf '%s\\n' \"$line\"; done"],"by":"user"})).await; + call(&client, "actor_start", json!({"group_id":group,"actor_id":"reader","by":"user"})).await; + let mail = call(&client, "message_send", json!({"group_id":group,"to":["reader"],"by":"user","message_mode":"mail","text":"Please review the delivery result."})).await; + let mail_id = mail["event"]["id"].as_str().expect("Mail event ID"); + let cutoff = mail["event"]["ts"].as_str().expect("original Mail timestamp"); + let before = call(&client, "inbox_summary", json!({"group_id":group,"cutoff":cutoff})).await; + assert_eq!(summary_row(&before, "reader")["awaiting_count"], 1); + let notice = call(&client, "system_notify", json!({"group_id":group,"by":"system","to":["reader"],"text":"A separate delivery notice."})).await; + let notice_id = notice["event"]["id"].as_str().expect("notice event ID"); + let path = GroupStore::new(runtime_home.clone()).expect("open fixture group store").ledger_path(group).expect("fixture group ledger path"); + let deadline = Instant::now() + Duration::from_secs(20); + loop { + let events = ledger::read_all(&path).expect("read notice-delivery evidence"); + if events.iter().any(|event| event.kind == "runtime.delivery" + && event.data["source_event_id"] == notice_id && event.data["state"] == "accepted") { + break; + } + assert!(Instant::now() < deadline, "notice worker did not settle"); + tokio::time::sleep(Duration::from_millis(20)).await; + } + let noticed = call(&client, "inbox_summary", json!({"group_id":group,"cutoff":cutoff})).await; + assert_eq!(summary_row(¬iced, "reader")["awaiting_count"], 1, "notice delivery settled the Mail itself"); + assert_eq!(summary_row(¬iced, "reader")["oldest_awaiting_at"], summary_row(&before, "reader")["oldest_awaiting_at"]); + call(&client, "actor_stop", json!({"group_id":group,"actor_id":"reader","by":"user"})).await; + let parked = call(&client, "inbox_summary", json!({"group_id":group,"cutoff":cutoff})).await; + assert_eq!(summary_row(&parked, "reader")["awaiting_count"], 1, "stopping an Actor must not hide its awaiting Mail"); + assert_eq!(summary_row(&parked, "reader")["oldest_awaiting_at"], summary_row(&before, "reader")["oldest_awaiting_at"]); + call(&client, "actor_start", json!({"group_id":group,"actor_id":"reader","by":"user"})).await; + let restarted = call(&client, "inbox_summary", json!({"group_id":group,"cutoff":cutoff})).await; + assert_eq!(summary_row(&restarted, "reader")["awaiting_count"], 1); + assert_eq!(summary_row(&restarted, "reader")["oldest_awaiting_at"], summary_row(&before, "reader")["oldest_awaiting_at"], "restart must not reset Mail age"); + call(&client, "message_deliver", json!({"group_id":group,"source_event_id":mail_id,"actor_ids":["reader"],"by":"user"})).await; + let deadline = Instant::now() + Duration::from_secs(20); + loop { + let events = ledger::read_all(&path).expect("read Mail-delivery evidence"); + if events.iter().any(|event| event.kind == "runtime.delivery" + && event.data["source_event_id"] == mail_id && event.data["state"] == "accepted") { + break; + } + assert!(Instant::now() < deadline, "Mail worker did not settle"); + tokio::time::sleep(Duration::from_millis(20)).await; + } + let delivered = call(&client, "inbox_summary", json!({"group_id":group,"cutoff":cutoff})).await; + assert_eq!(summary_row(&delivered, "reader")["awaiting_count"], 0); + let peek = call(&client, "inbox_peek", json!({"group_id":group,"actor_id":"reader"})).await; + assert_eq!(peek["messages"][0]["id"], mail_id, "delivery is distinct from reading Mail"); + call(&client, "actor_stop", json!({"group_id":group,"actor_id":"reader","by":"user"})).await; + let stopped = call(&client, "inbox_summary", json!({"group_id":group,"cutoff":cutoff})).await; + assert_eq!(summary_row(&stopped, "reader")["awaiting_count"], 0); + }).await; + let cleanup = DaemonClient::new(home); + call(&cleanup, "shutdown", json!({})).await; + outcome.expect("delivery test task"); +} + +#[tokio::test] +async fn overdue_mail_summary_is_passive_authorized_and_generation_scoped() { + let temp = tempfile::tempdir().expect("tempdir"); + let home = HomeLayout::from_path(temp.path().join("home")).expect("home"); + home.initialize().expect("initialize"); + let mut daemon = Daemon( + Command::new(env!("CARGO_BIN_EXE_cccc")) + .args(["daemon", "run", "--host", "127.0.0.1", "--port", "0"]) + .env("CCCC_HOME", home.root()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .expect("daemon"), + ); + let deadline = Instant::now() + Duration::from_secs(20); + while !home.daemon_dir().join("ccccd.addr.json").exists() { + assert!(daemon.0.try_wait().expect("status").is_none()); + assert!(Instant::now() < deadline, "startup timed out"); + tokio::time::sleep(Duration::from_millis(20)).await; + } + let client = DaemonClient::new(home.clone()).with_timeout(Duration::from_secs(30)); + let created = call( + &client, + "group_create", + json!({"title":"Mail summary","by":"user"}), + ) + .await; + let group = created["group"]["group_id"].as_str().expect("group"); + // These Actors are never started. The first visible Actor is Foreman. + for actor in ["coordinator", "reader", "other"] { + call( + &client, + "actor_add", + json!({"group_id":group,"actor_id":actor,"runtime":"custom","by":"user"}), + ) + .await; + } + let sent = call(&client, "message_send", json!({"group_id":group,"by":"user","to":["reader"],"message_mode":"mail","text":"Please review this change."})).await; + let id = sent["event"]["id"].as_str().expect("id"); + let timestamp = sent["event"]["ts"].as_str().expect("original timestamp"); + let before = call( + &client, + "inbox_summary", + json!({"group_id":group,"cutoff":"1970-01-01T00:00:00Z"}), + ) + .await; + assert_eq!(summary_row(&before, "reader")["awaiting_count"], 1); + assert_eq!(summary_row(&before, "reader")["overdue_count"], 0); + let summary = call( + &client, + "inbox_summary", + json!({"group_id":group,"cutoff":timestamp}), + ) + .await; + assert_eq!( + summary["actors"] + .as_array() + .expect("actor summary rows") + .len(), + 3 + ); + assert_eq!( + summary_row(&summary, "reader")["overdue_count"], + 1, + "cutoff equality is inclusive" + ); + assert_eq!(summary_row(&summary, "coordinator")["awaiting_count"], 0); + assert_eq!(summary_row(&summary, "other")["awaiting_count"], 0); + let after = call( + &client, + "inbox_summary", + json!({"group_id":group,"cutoff":summary["observed_at"],"by":"coordinator"}), + ) + .await; + assert_eq!(summary_row(&after, "reader")["overdue_count"], 1); + assert_eq!( + summary_row(&after, "reader")["oldest_awaiting_at"], + summary_row(&summary, "reader")["oldest_awaiting_at"] + ); + for by in ["reader", "unknown"] { + let response = client + .call(&DaemonRequest { + v: 1, + op: "inbox_summary".into(), + args: json!({"group_id":group,"cutoff":timestamp,"by":by}) + .as_object() + .expect("permission-test request arguments") + .clone(), + }) + .await + .expect("permission-test daemon response"); + assert!(!response.ok, "{by} obtained a Group-wide Inbox view"); + } + for cutoff in ["not-a-time", "9999-01-01T00:00:00Z"] { + let response = client + .call(&DaemonRequest { + v: 1, + op: "inbox_summary".into(), + args: json!({"group_id":group,"cutoff":cutoff}) + .as_object() + .expect("cutoff-test request arguments") + .clone(), + }) + .await + .expect("cutoff-test daemon response"); + assert!(!response.ok); + } + let peek = call( + &client, + "inbox_peek", + json!({"group_id":group,"actor_id":"reader"}), + ) + .await; + assert_eq!(peek["messages"][0]["id"], id, "summary consumed Mail"); + let disabled = call( + &client, + "actor_update", + json!({"group_id":group,"actor_id":"reader","patch":{"enabled":false},"by":"user"}), + ) + .await; + assert_eq!(disabled["actor"]["enabled"], false); + let disabled_summary = call( + &client, + "inbox_summary", + json!({"group_id":group,"cutoff":timestamp}), + ) + .await; + assert_eq!( + summary_row(&disabled_summary, "reader")["awaiting_count"], + 1 + ); + call(&client, "reply", json!({"group_id":group,"by":"reader","to":["user"],"reply_to":id,"message_mode":"send","text":"Reviewed."})).await; + let replied = call( + &client, + "inbox_summary", + json!({"group_id":group,"cutoff":timestamp}), + ) + .await; + assert_eq!(summary_row(&replied, "reader")["awaiting_count"], 0); + let mail_reply = call(&client, "reply", json!({"group_id":group,"by":"reader","to":["coordinator"],"reply_to":id,"message_mode":"mail","text":"Review result for the coordinator."})).await; + let reply_summary = call( + &client, + "inbox_summary", + json!({"group_id":group,"cutoff":mail_reply["event"]["ts"]}), + ) + .await; + assert_eq!( + summary_row(&reply_summary, "coordinator")["awaiting_count"], + 1 + ); + assert_eq!(summary_row(&reply_summary, "reader")["awaiting_count"], 0); + call(&client, "message_send", json!({"group_id":group,"by":"user","to":["reader"],"message_mode":"send","text":"An immediate non-Mail message."})).await; + let sent_only = call( + &client, + "inbox_summary", + json!({"group_id":group,"cutoff":mail_reply["event"]["ts"]}), + ) + .await; + assert_eq!(summary_row(&sent_only, "reader")["awaiting_count"], 0); + assert_eq!( + call( + &client, + "inbox_peek", + json!({"group_id":group,"actor_id":"reader"}) + ) + .await["messages"][0]["id"], + id, + "a reply is not an Inbox read" + ); + call(&client, "message_send", json!({"group_id":group,"by":"user","to":["reader"],"message_mode":"mail","text":"Please check the next change."})).await; + call( + &client, + "inbox_read", + json!({"group_id":group,"actor_id":"reader"}), + ) + .await; + let read = call( + &client, + "inbox_summary", + json!({"group_id":group,"cutoff":timestamp}), + ) + .await; + assert_eq!(summary_row(&read, "reader")["awaiting_count"], 0); + call(&client, "message_send", json!({"group_id":group,"by":"user","to":["reader"],"message_mode":"mail","text":"Review after the next session."})).await; + call( + &client, + "actor_remove", + json!({"group_id":group,"actor_id":"reader","by":"user"}), + ) + .await; + call( + &client, + "actor_add", + json!({"group_id":group,"actor_id":"reader","runtime":"custom","by":"user"}), + ) + .await; + let generation = call( + &client, + "inbox_summary", + json!({"group_id":group,"cutoff":timestamp}), + ) + .await; + assert_eq!(summary_row(&generation, "reader")["awaiting_count"], 0); + let current = call(&client, "message_send", json!({"group_id":group,"by":"user","to":["reader"],"message_mode":"mail","text":"Review in this session."})).await; + let current_cutoff = current["event"]["ts"] + .as_str() + .expect("current-generation Mail timestamp"); + let warm = call( + &client, + "inbox_summary", + json!({"group_id":group,"cutoff":current_cutoff}), + ) + .await; + call( + &client, + "ledger_compact", + json!({"group_id":group,"reason":"summary snapshot rebuild"}), + ) + .await; + let rebuilt = call( + &client, + "inbox_summary", + json!({"group_id":group,"cutoff":current_cutoff}), + ) + .await; + assert_eq!(summary_row(&warm, "reader")["awaiting_count"], 1); + assert_eq!(summary_row(&rebuilt, "reader")["awaiting_count"], 1); + assert_eq!(summary_row(&rebuilt, "reader")["overdue_count"], 1); + // Exercise the public CLI entry point against the same real daemon. + let output = Command::new(env!("CARGO_BIN_EXE_cccc")) + .args([ + "inbox-summary", + "--group", + group, + "--cutoff", + current_cutoff, + ]) + .env("CCCC_HOME", home.root()) + .output() + .expect("CLI summary"); + assert!( + output.status.success(), + "{}", + String::from_utf8_lossy(&output.stderr) + ); + let cli: Value = serde_json::from_slice(&output.stdout).expect("CLI JSON"); + assert_eq!(summary_row(&cli, "reader")["overdue_count"], 1); +} + #[tokio::test] async fn inbox_projection_refills_overflow_and_follows_compaction() { let temp = tempfile::tempdir().expect("tempdir"); diff --git a/crates/cccc-core/src/inbox.rs b/crates/cccc-core/src/inbox.rs index a23f9eb3c..dfe67ed13 100644 --- a/crates/cccc-core/src/inbox.rs +++ b/crates/cccc-core/src/inbox.rs @@ -20,6 +20,11 @@ pub(crate) mod projection; const MAIL_CURSOR_SCHEMA: u64 = 1; const PENDING_READ_SCHEMA: u64 = 1; +#[cfg(test)] +thread_local! { + static TEST_PENDING_FACT_CHECKS: std::cell::Cell = const { std::cell::Cell::new(0) }; +} + #[derive(Debug, Clone, Default)] struct InboxState { cursors: BTreeMap, @@ -184,6 +189,65 @@ pub fn consume_unread( }) } +/// Payload-free current awaiting state filtered by an inclusive original-Mail +/// timestamp cutoff. This is not historical Inbox reconstruction. +#[derive(Debug, Serialize)] +pub struct OverdueMailSummary { + pub actor_id: String, + pub awaiting_count: usize, + pub overdue_count: usize, + pub oldest_awaiting_at: Option, + pub oldest_age_seconds: Option, +} + +impl OverdueMailSummary { + fn from_timestamps<'a>( + actor_id: &str, + timestamps: impl Iterator, + cutoff: DateTime, + observed_at: DateTime, + ) -> Self { + let mut result = Self { + actor_id: actor_id.to_owned(), + awaiting_count: 0, + overdue_count: 0, + oldest_awaiting_at: None, + oldest_age_seconds: None, + }; + let mut oldest: Option> = None; + let mut unknown_age = false; + for timestamp in timestamps { + result.awaiting_count += 1; + let Ok(created) = DateTime::parse_from_rfc3339(timestamp) else { + unknown_age = true; + continue; + }; + let created = created.with_timezone(&Utc); + if created > observed_at { + unknown_age = true; + continue; + } + if created <= cutoff { + result.overdue_count += 1; + } + oldest = Some(oldest.map_or(created, |oldest| oldest.min(created))); + } + if !unknown_age && let Some(oldest) = oldest { + result.oldest_awaiting_at = Some(oldest.to_rfc3339()); + result.oldest_age_seconds = Some((observed_at - oldest).num_seconds()); + } + result + } +} + +pub fn overdue_summary( + home: &HomeLayout, + group: &GroupDoc, + cutoff: DateTime, +) -> io::Result<(DateTime, Vec)> { + projection::overdue_summary(home, group, cutoff) +} + pub fn mail_pending_summary( home: &HomeLayout, group: &GroupDoc, @@ -393,6 +457,8 @@ fn load_pending(home: &HomeLayout, group_id: &str) -> io::Result io::Result { + #[cfg(test)] + TEST_PENDING_FACT_CHECKS.with(|count| count.set(count.get().saturating_add(1))); let ledger_path = GroupStore::new(home.clone())?.ledger_path(group_id)?; let mut found = false; ledger::visit_newest_first(&ledger_path, |event| { @@ -590,6 +656,146 @@ mod tests { use cccc_contracts::Actor; use std::sync::{Arc, Barrier}; + #[test] + fn overdue_summary_keeps_pending_recovery_out_of_warm_polls() { + // Synthetic recovery controls, separate from the real-daemon IPC tests. + for committed in [false, true] { + let temp = tempfile::tempdir().expect("temporary recovery directory"); + let home = HomeLayout::from_path(temp.path().join("home")).expect("recovery home"); + let store = GroupStore::new(home.clone()).expect("recovery group store"); + let mut group = store + .create("summary recovery", "") + .expect("recovery group"); + group.actors.push(Actor::new("reader")); + store.save(&group).expect("save recovery group"); + let ledger_path = store + .ledger_path(&group.group_id) + .expect("recovery ledger path"); + let mails = ["01", "02", "03"].map(|second| { + let mut mail = Event::new("chat.message", &group.group_id); + mail.ts = format!("2000-01-01T00:00:{second}Z"); + mail.by = "user".into(); + mail.data = json!({"to":["reader"],"message_mode":"mail"}) + .as_object() + .cloned() + .expect("recovery Mail data"); + ledger::append(&ledger_path, &mail).expect("append recovery Mail"); + mail + }); + let cursor_path = path(&home, &group.group_id).expect("durable cursor path"); + write_json( + &cursor_path, + &json!({"schema":MAIL_CURSOR_SCHEMA,"cursors":{ + "reader":{"event_id":mails[0].id,"ts":mails[0].ts,"updated_at":mails[0].ts} + }}), + ) + .expect("write stale durable cursor"); + if committed { + let mut read = Event::new("mail.read", &group.group_id); + read.by = "reader".into(); + read.data = json!({"actor_id":"reader","event_id":mails[1].id}) + .as_object() + .cloned() + .expect("committed read data"); + ledger::append(&ledger_path, &read).expect("append committed read"); + } + let pending = PendingRead { + schema: PENDING_READ_SCHEMA, + group_id: group.group_id.clone(), + actor_id: "reader".into(), + expected: PendingCursor { + event_id: mails[0].id.clone(), + ..PendingCursor::default() + }, + target: PendingCursor { + event_id: mails[1].id.clone(), + ts: mails[1].ts.clone(), + updated_at: mails[1].ts.clone(), + }, + }; + let marker_path = pending_path(&home, &group.group_id).expect("pending marker path"); + write_json(&marker_path, &pending).expect("write persistent marker"); + let cursor_bytes = fs::read(&cursor_path).expect("saved cursor bytes"); + let marker_bytes = fs::read(&marker_path).expect("saved marker bytes"); + + // Warm actor pages under their effective cursor policy first. + list_unread(&home, &group, "reader", 10).expect("warm actor inbox"); + for _ in 0..3 { + let scans = TEST_PENDING_FACT_CHECKS.with(std::cell::Cell::get); + let rebuilds = projection::TEST_REBUILDS.with(std::cell::Cell::get); + let (_, summary) = + overdue_summary(&home, &group, Utc::now()).expect("warm summary"); + assert_eq!(summary[0].awaiting_count, if committed { 1 } else { 2 }); + let oldest_at = chrono::DateTime::parse_from_rfc3339( + summary[0] + .oldest_awaiting_at + .as_deref() + .expect("oldest awaiting Mail timestamp"), + ) + .expect("valid summary timestamp"); + let expected_at = + chrono::DateTime::parse_from_rfc3339(&mails[if committed { 2 } else { 1 }].ts) + .expect("valid Mail fixture timestamp"); + assert_eq!(oldest_at, expected_at); + assert_eq!(TEST_PENDING_FACT_CHECKS.with(std::cell::Cell::get), scans); + assert_eq!( + projection::TEST_REBUILDS.with(std::cell::Cell::get), + rebuilds + ); + assert_eq!( + fs::read(&cursor_path).expect("unchanged cursor"), + cursor_bytes + ); + assert_eq!( + fs::read(&marker_path).expect("unchanged marker"), + marker_bytes + ); + // Legacy readers retain recovery semantics without forcing the + // next summary to rewind and rebuild the shared cache. + list_unread(&home, &group, "reader", 10).expect("interleaved actor inbox"); + assert_eq!( + projection::TEST_REBUILDS.with(std::cell::Cell::get), + rebuilds + ); + } + } + } + + #[test] + fn overdue_timestamp_boundary_and_unknown_ages() { + // Pure timestamp unit controls, not captured or daemon-produced Mail. + let cutoff = DateTime::parse_from_rfc3339("2026-01-01T12:00:00Z") + .expect("valid timestamp boundary fixture") + .with_timezone(&Utc); + let observed = cutoff + chrono::Duration::seconds(60); + let known = OverdueMailSummary::from_timestamps( + "reader", + [ + "2026-01-01T11:59:59.999999Z", + "2026-01-01T12:00:00Z", + "2026-01-01T12:00:00.000001Z", + ] + .into_iter(), + cutoff, + observed, + ); + assert_eq!(known.awaiting_count, 3); + assert_eq!(known.overdue_count, 2); + assert_eq!(known.oldest_age_seconds, Some(60)); + for unknown in ["invalid", "2026-01-01T12:01:00.000001Z"] { + let summary = OverdueMailSummary::from_timestamps( + "reader", + ["2026-01-01T12:00:00Z", unknown].into_iter(), + cutoff, + observed, + ); + assert_eq!(summary.awaiting_count, 2); + assert_eq!(summary.overdue_count, 1); + assert!(summary.oldest_awaiting_at.is_none()); + assert!(summary.oldest_age_seconds.is_none()); + } + } + #[test] fn concurrent_consumers_return_each_message_once() { let temp = tempfile::tempdir().expect("tempdir"); diff --git a/crates/cccc-core/src/inbox/notices.rs b/crates/cccc-core/src/inbox/notices.rs index e30714234..e6bcb106c 100644 --- a/crates/cccc-core/src/inbox/notices.rs +++ b/crates/cccc-core/src/inbox/notices.rs @@ -13,6 +13,82 @@ struct MailClaim { sources: Vec, } +#[cfg(test)] +mod summary_tests { + use super::*; + use cccc_contracts::Actor; + + #[test] + fn unresolved_cache_follows_recipient_generation_and_delivery_facts() { + // Reducer unit controls: these in-memory facts are synthetic, unlike + // the IPC integration tests' daemon-produced events. + let temp = tempfile::tempdir().expect("create temporary test directory"); + let home = HomeLayout::from_path(temp.path().join("home")).expect("create test home"); + let store = GroupStore::new(home).expect("open test group store"); + let mut group = store.create("Mail reducer", "").expect("create test group"); + group.actors.push(Actor::new("reader")); + group.actors.push(Actor::new("other")); + let mut state = NoticeState::default(); + let mut mail = Event::new("chat.message", &group.group_id); + mail.by = "user".into(); + mail.data = json!({"to":["reader"],"message_mode":"mail"}) + .as_object() + .expect("Mail fixture is an object") + .clone(); + state.note(&group, &mail); + let cutoff = Utc::now(); + let summarize = |state: &mut NoticeState| { + state.overdue_summary(&group, &BTreeMap::new(), cutoff, cutoff) + }; + assert_eq!(summarize(&mut state)[0].awaiting_count, 1); + assert_eq!(summarize(&mut state)[1].awaiting_count, 0); + + let mut delivery = Event::new("runtime.delivery", &group.group_id); + delivery.data = + json!({"actor_id":"reader","source_event_id":"notice-id","state":"accepted"}) + .as_object() + .expect("delivery fixture is an object") + .clone(); + state.note(&group, &delivery); + assert_eq!(summarize(&mut state)[0].awaiting_count, 1); + delivery + .data + .insert("source_event_id".into(), json!(mail.id)); + for status in ["accepted", "ambiguous"] { + delivery.data.insert("state".into(), json!(status)); + state.note(&group, &delivery); + assert_eq!(summarize(&mut state)[0].awaiting_count, 0); + delivery.data.insert("state".into(), json!("failed")); + state.note(&group, &delivery); + assert_eq!(summarize(&mut state)[0].awaiting_count, 1); + } + let mut reply = Event::new("chat.message", &group.group_id); + reply.by = "reader".into(); + reply.data.insert("reply_to".into(), json!(mail.id)); + state.note(&group, &reply); + assert_eq!(summarize(&mut state)[0].awaiting_count, 0); + + mail.id = "second-mail".into(); + state.note(&group, &mail); + assert_eq!(summarize(&mut state)[0].awaiting_count, 1); + let cursor = BTreeMap::from([("reader".into(), mail.id.clone())]); + assert_eq!( + state.overdue_summary(&group, &cursor, cutoff, cutoff)[0].awaiting_count, + 0 + ); + mail.id = "third-mail".into(); + state.note(&group, &mail); + assert_eq!(summarize(&mut state)[0].awaiting_count, 1); + let mut added = Event::new("actor.add", &group.group_id); + added.data.insert("actor".into(), json!({"id":"reader"})); + state.note(&group, &added); + assert_eq!(summarize(&mut state)[0].awaiting_count, 0); + mail.id = "current-generation".into(); + state.note(&group, &mail); + assert_eq!(summarize(&mut state)[0].awaiting_count, 1); + } +} + #[derive(Default)] pub(crate) struct NoticeState { pub positions: HashMap, @@ -20,6 +96,10 @@ pub(crate) struct NoticeState { mail_ids: HashSet, pub next_position: usize, sources: Vec, + mail_sources: HashMap, + // Payload-free references to unresolved Mail only. Resolved history is + // removed on each fact, so a summary poll never walks historical sources. + awaiting_mail: BTreeMap>, generations: HashMap, mail_reads: HashMap>, replies: HashMap<(String, String), usize>, @@ -35,6 +115,7 @@ struct Source { ts: String, position: usize, mail: bool, + notice: bool, recipients: Vec, } @@ -90,6 +171,7 @@ impl NoticeState { } if let Some(actor) = actor_add_id(event) { self.generations.insert(actor.to_owned(), position); + self.awaiting_mail.remove(actor); } if event.kind == "group.start" || (event.kind == "group.set_state" @@ -115,6 +197,11 @@ impl NoticeState { .entry(actor.to_owned()) .or_default() .push((position, boundary.to_owned())); + if let Some(boundary) = self.positions.get(boundary) + && let Some(pending) = self.awaiting_mail.get_mut(actor) + { + pending.retain(|position, _| position > boundary); + } } } "chat.message" => { @@ -122,6 +209,7 @@ impl NoticeState { self.replies .entry((id.to_owned(), event.by.clone())) .or_insert(position); + self.refresh_mail(id, &event.by); } let broadcast = event .data @@ -134,7 +222,7 @@ impl NoticeState { .filter_map(Value::as_str) .any(|id| matches!(id, "@all" | "@peers" | "@foreman")) }); - if source && (mode != "mail" || !broadcast) { + if source { let recipients = group .actors .iter() @@ -142,13 +230,21 @@ impl NoticeState { .map(|actor| actor.id.clone()) .collect::>(); if !recipients.is_empty() { + let index = self.sources.len(); self.sources.push(Source { id: event.id.clone(), ts: event.ts.clone(), position, mail: mode == "mail", + notice: mode != "mail" || !broadcast, recipients, }); + if mode == "mail" { + self.mail_sources.insert(event.id.clone(), index); + for actor in self.sources[index].recipients.clone() { + self.refresh_mail(&event.id, &actor); + } + } } } } @@ -167,6 +263,7 @@ impl NoticeState { (id.to_owned(), actor.to_owned()), (state.to_owned(), event.ts.clone(), position), ); + self.refresh_mail(id, actor); } } "system.notify" => { @@ -215,6 +312,63 @@ impl NoticeState { } } + fn refresh_mail(&mut self, id: &str, actor: &str) { + let Some(&index) = self.mail_sources.get(id) else { + return; + }; + let source = &self.sources[index]; + let generation = self.generations.get(actor).copied().unwrap_or(0); + let read = self.mail_reads.get(actor).is_some_and(|reads| { + reads.iter().any(|(_, boundary)| { + self.positions + .get(boundary) + .is_some_and(|position| *position >= source.position) + }) + }); + let key = (id.to_owned(), actor.to_owned()); + let awaiting = source.position >= generation + && source.recipients.iter().any(|recipient| recipient == actor) + && crate::mail_pending::awaiting( + read, + self.replies.contains_key(&key), + self.deliveries.get(&key).map(|value| value.0.as_str()), + ); + let pending = self.awaiting_mail.entry(actor.to_owned()).or_default(); + if awaiting { + pending.insert(source.position, index); + } else { + pending.remove(&source.position); + } + } + + pub fn overdue_summary( + &mut self, + group: &GroupDoc, + cursors: &BTreeMap, + cutoff: DateTime, + observed_at: DateTime, + ) -> Vec { + group + .actors + .iter() + .map(|actor| { + let cursor = cursors.get(&actor.id).and_then(|id| self.positions.get(id)); + let pending = self.awaiting_mail.entry(actor.id.clone()).or_default(); + if let Some(cursor) = cursor { + pending.retain(|position, _| position > cursor); + } + OverdueMailSummary::from_timestamps( + &actor.id, + pending + .values() + .map(|index| self.sources[*index].ts.as_str()), + cutoff, + observed_at, + ) + }) + .collect() + } + fn prune_mail_claims(&mut self, group: &GroupDoc) { let repeat = crate::automation::delivery_timing_value(group, "mail_notice_repeat_seconds", 0); @@ -325,7 +479,9 @@ impl NoticeState { let mut mail = Vec::new(); let mut replies = Vec::new(); for source in self.sources.iter().filter(|source| { - source.position >= generation && source.recipients.contains(&actor.id) + source.notice + && source.position >= generation + && source.recipients.contains(&actor.id) }) { let key = (source.id.clone(), actor.id.clone()); let replied = self.replies.contains_key(&key); diff --git a/crates/cccc-core/src/inbox/projection.rs b/crates/cccc-core/src/inbox/projection.rs index 7d4e02454..2f686a6be 100644 --- a/crates/cccc-core/src/inbox/projection.rs +++ b/crates/cccc-core/src/inbox/projection.rs @@ -196,12 +196,50 @@ impl Projection { // The query mutex serializes cache publication, never append. Snapshot identity // includes source order and inode; growth is decoded from committed offsets. +#[derive(Clone, Copy)] +enum CursorPolicy { + Effective, + Summary, +} + +impl CursorPolicy { + fn load(self, home: &HomeLayout, group_id: &str) -> io::Result { + match self { + Self::Effective => load(home, group_id), + Self::Summary => load_raw(home, group_id), + } + } +} + +#[cfg(test)] +thread_local! { + pub(super) static TEST_REBUILDS: std::cell::Cell = const { std::cell::Cell::new(0) }; +} + fn with_projection( home: &HomeLayout, group: &GroupDoc, actor_ids: &[String], state: &InboxState, mut inspect: impl FnMut(&mut Projection, &Snapshot) -> io::Result, +) -> io::Result { + with_projection_policy( + home, + group, + actor_ids, + state, + CursorPolicy::Effective, + |projection, snapshot, _| inspect(projection, snapshot), + ) +} + +fn with_projection_policy( + home: &HomeLayout, + group: &GroupDoc, + actor_ids: &[String], + state: &InboxState, + cursor_policy: CursorPolicy, + mut inspect: impl FnMut(&mut Projection, &Snapshot, &InboxState) -> io::Result, ) -> io::Result { let path = GroupStore::new(home.clone())?.ledger_path(&group.group_id)?; let entry = entry(&path); @@ -226,7 +264,12 @@ fn with_projection( unknown |= !projection.deliveries.can_append(&event); ControlFlow::Continue(()) })?; - if unknown || !projection.move_cursors(&state, &snapshot)? { + // Summary filtering uses captured durable cursors and read + // facts, without rewinding another reader's mailbox cache. + if unknown + || (matches!(cursor_policy, CursorPolicy::Effective) + && !projection.move_cursors(&state, &snapshot)?) + { true } else { let old_offsets = projection.offsets.clone(); @@ -241,9 +284,15 @@ fn with_projection( _ => true, }; if rebuild { + #[cfg(test)] + TEST_REBUILDS.with(|count| count.set(count.get().saturating_add(1))); *guard = Some(Projection::build(group, &state, &snapshot, actor_ids)?); } - inspect(guard.as_mut().expect("projection initialized"), &snapshot) + inspect( + guard.as_mut().expect("projection initialized"), + &snapshot, + &state, + ) })(); if result.is_err() { // A failed suffix read/refill may have reduced only a prefix. @@ -257,7 +306,7 @@ fn with_projection( // and must not starve publication. A missed try_lock callback merges // by byte offset on the next query. let current = Snapshot::open(&path)?; - let current_state = load(home, &group.group_id)?; + let current_state = cursor_policy.load(home, &group.group_id)?; if !current.extends(&snapshot.sources) || current_state.cursors != state.cursors || entry.cursor_epoch.load(Ordering::Acquire) != cursor_epoch @@ -269,6 +318,59 @@ fn with_projection( } } +#[cfg(test)] +mod summary_cursor_tests { + use super::*; + use cccc_contracts::Actor; + + #[test] + fn summary_publication_retry_receives_the_new_raw_cursor_state() { + // Synthetic publication-race control, not a daemon-produced read. + let temp = tempfile::tempdir().expect("publication test directory"); + let home = HomeLayout::from_path(temp.path().join("home")).expect("publication home"); + let store = GroupStore::new(home.clone()).expect("publication group store"); + let mut group = store + .create("summary publication", "") + .expect("publication group"); + group.actors.push(Actor::new("reader")); + store.save(&group).expect("save publication group"); + let ledger_path = store + .ledger_path(&group.group_id) + .expect("publication ledger path"); + let mut mail = Event::new("chat.message", &group.group_id); + mail.data = json!({"to":["reader"],"message_mode":"mail"}) + .as_object() + .cloned() + .expect("publication Mail data"); + ledger::append(&ledger_path, &mail).expect("append publication Mail"); + let state = load_raw(&home, &group.group_id).expect("initial raw cursor state"); + let mut observed = Vec::new(); + let result = with_projection_policy( + &home, + &group, + &[], + &state, + CursorPolicy::Summary, + |_, _, current| { + let cursor = current.cursors.get("reader").cloned(); + observed.push(cursor.clone()); + if observed.len() == 1 { + write_json( + &path(&home, &group.group_id)?, + &json!({ + "schema":MAIL_CURSOR_SCHEMA,"cursors":{"reader":{"event_id":mail.id}} + }), + )?; + } + Ok(cursor) + }, + ) + .expect("retry changed raw cursor state"); + assert_eq!(observed, vec![None, Some(mail.id.clone())]); + assert_eq!(result, Some(mail.id)); + } +} + pub(super) fn pages( home: &HomeLayout, group: &GroupDoc, @@ -332,6 +434,32 @@ pub(super) fn pending( .map(|mail| (mailbox.count, mail.ts.clone()))) }) } +pub(super) fn overdue_summary( + home: &HomeLayout, + group: &GroupDoc, + cutoff: DateTime, +) -> io::Result<(DateTime, Vec)> { + let state = load_raw(home, &group.group_id)?; + with_projection_policy( + home, + group, + &[], + &state, + CursorPolicy::Summary, + |projection, _, state| { + // Sample after loading/replaying the captured prefix, including on a + // publication retry. Cold-load time must not make newly loaded Mail + // appear to have a future timestamp relative to this observation. + let observed_at = Utc::now(); + let actors = + projection + .notices + .overdue_summary(group, &state.cursors, cutoff, observed_at); + Ok((observed_at, actors)) + }, + ) +} + pub(crate) fn due( home: &HomeLayout, group: &GroupDoc, diff --git a/crates/cccc-daemon/src/ops/messaging.rs b/crates/cccc-daemon/src/ops/messaging.rs index 9b7df8122..deaf26ddd 100644 --- a/crates/cccc-daemon/src/ops/messaging.rs +++ b/crates/cccc-daemon/src/ops/messaging.rs @@ -55,6 +55,7 @@ pub(super) fn resolve_operation(request: &DaemonRequest) -> Option { "message_read_status" => Operation::new(Read, super::messaging_status::read_status), "inbox_peek" => Operation::new(Write, messaging_inbox::peek), "inbox_read" => Operation::new(Write, messaging_inbox::read), + "inbox_summary" => Operation::new(Read, messaging_inbox::summary), "message_history" => Operation::new(Read, messaging_inbox::history), _ => return None, }) diff --git a/crates/cccc-daemon/src/ops/messaging_inbox.rs b/crates/cccc-daemon/src/ops/messaging_inbox.rs index 70587ac9c..e93e80ad7 100644 --- a/crates/cccc-daemon/src/ops/messaging_inbox.rs +++ b/crates/cccc-daemon/src/ops/messaging_inbox.rs @@ -6,6 +6,32 @@ use serde_json::{Value, json}; use crate::dispatch::{OpError, OpResult, object, required_arg, string_arg}; use crate::ops::messaging::load; +pub fn summary(home: &HomeLayout, request: &DaemonRequest) -> OpResult { + let group = load(home, request)?; + let by = string_arg(request, "by").unwrap_or_else(|| "user".into()); + // A Group-wide view must also reject peers in an empty Group; checking + // only each returned row would accidentally make that permission vacuous. + permissions::require_group(&group, &by).map_err(OpError::invalid)?; + let cutoff = required_arg(request, "cutoff")?; + let cutoff = chrono::DateTime::parse_from_rfc3339(&cutoff) + .map_err(|_| OpError::invalid("cutoff must be an RFC3339 timestamp"))? + .with_timezone(&chrono::Utc); + if cutoff > chrono::Utc::now() { + return Err(OpError::invalid("cutoff must not be in the future")); + } + let (observed_at, summary) = + inbox::overdue_summary(home, &group, cutoff).map_err(OpError::io)?; + if cutoff > observed_at { + return Err(OpError::invalid("cutoff must not be after observed_at")); + } + object(json!({ + "group_id": group.group_id, + "observed_at": observed_at.to_rfc3339(), + "cutoff": cutoff.to_rfc3339(), + "actors": summary, + })) +} + pub fn peek(home: &HomeLayout, request: &DaemonRequest) -> OpResult { let group = load(home, request)?; let actor_id = required_arg(request, "actor_id")?; diff --git a/docs/standards/CCCC_DAEMON_IPC_V1.md b/docs/standards/CCCC_DAEMON_IPC_V1.md index b80a4aaae..293514aa8 100644 --- a/docs/standards/CCCC_DAEMON_IPC_V1.md +++ b/docs/standards/CCCC_DAEMON_IPC_V1.md @@ -5172,6 +5172,67 @@ Result: { messages: CCCSEventV1[]; cursor: { event_id: string; ts: string } } ``` +#### `inbox_summary` + +Passively summarize current-generation awaiting Mail across a Group. This is a +Group-wide Inbox view: the existing user/system and Foreman permissions apply; +Peers and unknown actors cannot use it. Stopped and disabled Actors are included. +The operation is a Read and does not start Actors, deliver notices, advance Mail +cursors, write an index, or schedule background sampling. + +Args: +```ts +{ group_id: string; cutoff: string; by?: string } +``` + +`cutoff` MUST be an RFC3339 timestamp no later than the server's `observed_at`. +The boundary is inclusive: an awaiting Mail is overdue when its original +`chat.message.ts <= cutoff`. The cutoff filters **current** awaiting state; it +does not reconstruct Inbox state as of that timestamp. Age is measured from the +original Mail timestamp to `observed_at`, independently of notice delivery or +Actor restart. The observation time is sampled after the captured projection +prefix is loaded, and is resampled if publication validation retries. +A malformed or future Mail timestamp has unknown age and cannot +establish overdue status. Invalid/future request cutoffs are errors. + +Result: +```ts +{ + group_id: string + observed_at: string + cutoff: string + actors: { + actor_id: string + awaiting_count: number + overdue_count: number + oldest_awaiting_at: string | null + oldest_age_seconds: number | null + }[] +} +``` + +Each current Actor has one row, including zero-count rows. Awaiting uses the +existing Mail predicate: unread, not replied, and no `accepted` or `ambiguous` +delivery of the Mail itself to that recipient in the current Actor generation. +Delivery of an Inbox notice is distinct and does not clear its source Mail. +Send, Send + Reply, and `system.notify` are not Mail. The recipient, cursor and +latest-`actor.add` append-order boundaries are the same as Inbox. A reply, +read/cursor advance, or delivery may remove Mail from the current result even +when its original timestamp precedes the cutoff. + +`oldest_awaiting_at` and `oldest_age_seconds` are null for an empty row or when +any awaiting Mail has an invalid/future timestamp, since its oldest age cannot +be established. Known ages are nonnegative whole seconds. No message bodies, +source ids, escalation policy or notification action are returned. + +All Actor rows MUST share one incremental Group projection snapshot. Warm +polls reuse committed projection offsets rather than rescan full ledger history +per Actor or per poll. The projection retains the existing bounded Group cache, +generation/cursor validation and rotation publication checks. + +CLI: `cccc inbox-summary --group --cutoff [--by ]`. +The consuming `cccc inbox` command is unchanged. No default MCP tool is added. + #### `inbox_read` Atomically return and consume the next unread Mail prefix for an actor.