feat(reconcile): filter initial baseline to 7-day unread emails

New mailbox onboarding no longer floods Telegram with the entire
mailbox history.  The initial baseline walk (ReconcileCursor::Walk,
any position) now passes an EmailQueryFilter to list_emails_page that
bounds Email/query to messages received within the last 7 days and
not yet seen ($seen keyword, RFC 8621 §4.1.2).

- EmailQueryFilter { received_after: Option<i64>, unread_only: bool }
  in src/domain/jmap.rs (backend-agnostic domain type).
- StalwartJmap::fetch_email_page builds a compound Filter::and from
  the filter conditions (after + not_keyword "$seen").
- walk_baseline computes the 7-day cutoff once and passes it to
  every page fetch, so resumed walks stay filtered.
- Replay(state) path returns early and never calls list_emails_page,
  so /changes incremental is unaffected.
- const INITIAL_BASELINE_WINDOW_DAYS: u64 = 7.
- New test first_reconcile_filters_the_initial_baseline_to_recent_unread
  pins the filter on every Walk page and the cursor transition to Replay.
- 3 doc anchors re-anchored (worker.rs:227→241, client.rs:485→513 ×2).

GATE-P0: 104 passed / 0 failed / 4 ignored.
This commit is contained in:
anonsaber 2026-10-12 22:08:51 +08:00
parent 1bdda7c471
commit 21e91765d0
6 changed files with 242 additions and 11 deletions

View File

@ -279,7 +279,7 @@ TG /read 3 → handler 取会话里的 folder+page 游标
当前**没有 FSM**。命令路由由 `src/worker.rs` 的 `parse_intent` 解析为 `Intent`;AI 授权态是 Redis 里的一个布尔加过期时间(键与 TTL 见 `docs/reference.md` 的 AI 授权态一节),不是多步确认态。 当前**没有 FSM**。命令路由由 `src/worker.rs` 的 `parse_intent` 解析为 `Intent`;AI 授权态是 Redis 里的一个布尔加过期时间(键与 TTL 见 `docs/reference.md` 的 AI 授权态一节),不是多步确认态。
仍成立的不变量:AI 摘要必须先有用户显式授权,**没有授权态不得调用 LLM**。授权一旦生效,摘要会拉取邮件正文全文送 LLM(`read_email` 请求 `TextBody` + `BodyValues` 并 `fetch_text_body_values(true)`,见 `src/domain/jmap/client.rs`),但正文**不回传给 Telegram**——出站只发摘要,或失败时回退为前 300 字符(`worker.rs:227` 注释:body text never reaches Telegram)。 仍成立的不变量:AI 摘要必须先有用户显式授权,**没有授权态不得调用 LLM**。授权一旦生效,摘要会拉取邮件正文全文送 LLM(`read_email` 请求 `TextBody` + `BodyValues` 并 `fetch_text_body_values(true)`,见 `src/domain/jmap/client.rs`),但正文**不回传给 Telegram**——出站只发摘要,或失败时回退为前 300 字符(`worker.rs:241` 注释:body text never reaches Telegram)。
曾设计过的 5 态 FSM(`Idle` / `AwaitClarify` / `AwaitConfirm` / `Analyzing` / `AwaitFallback`)连同状态转移表与渠道中立说明,见 `docs/retired.md`。 曾设计过的 5 态 FSM(`Idle` / `AwaitClarify` / `AwaitConfirm` / `Analyzing` / `AwaitFallback`)连同状态转移表与渠道中立说明,见 `docs/retired.md`。

View File

@ -107,7 +107,7 @@ Single accounts are fixed by requirement (`REQ-SINGLE-ACCOUNT`) and never enter
- `state:jmap:since` is a **single global cursor**, no account dimension (read `src/state.rs:801`, write `src/state.rs:809`) - both accounts will overwrite each other's cursors, which is the first place to move. - `state:jmap:since` is a **single global cursor**, no account dimension (read `src/state.rs:801`, write `src/state.rs:809`) - both accounts will overwrite each other's cursors, which is the first place to move.
- `lock:reconcile` is a **single global solo lock** (Redis implementation `src/state.rs:1139`) - the reconciliation of two accounts will be serialized by the same lock. - `lock:reconcile` is a **single global solo lock** (Redis implementation `src/state.rs:1139`) - the reconciliation of two accounts will be serialized by the same lock.
- `dedup:jmap:{account_id}:{email_id}` **Already has the account segment** (`docs/reference.md` §1), the key shape does not need to be changed. - `dedup:jmap:{account_id}:{email_id}` **Already has the account segment** (`docs/reference.md` §1), the key shape does not need to be changed.
- The `ACCOUNT_ID` environment variable already exists and has been verified (leave it blank to take the session main account, and an error will be reported if the out-of-bounds value is `src/domain/jmap/client.rs:485`). The business configuration layer can also override `account_id` (`unwrap_or_else` of `src/config.rs:465` falls back). The configuration interface is ready, all that is missing is the sub-dimension status within the instance. - The `ACCOUNT_ID` environment variable already exists and has been verified (leave it blank to take the session main account, and an error will be reported if the out-of-bounds value is `src/domain/jmap/client.rs:513`). The business configuration layer can also override `account_id` (`unwrap_or_else` of `src/config.rs:465` falls back). The configuration interface is ready, all that is missing is the sub-dimension status within the instance.
This item is not skipped, but explicitly rejected: the deployment unit saved by single instance and multiple accounts cannot be exchanged for the routing and cursor complexity it introduces. This item is not skipped, but explicitly rejected: the deployment unit saved by single instance and multiple accounts cannot be exchanged for the routing and cursor complexity it introduces.

View File

@ -107,7 +107,7 @@ TTL 那条真机绿灯已经用你给的 Upstash URL 跑过了(见 §1),
- `state:jmap:since` 是**单一全局游标**,无账户维度(读 `src/state.rs:801`、写 `src/state.rs:809`)——两个账户会互相覆盖游标,这是第一个要动的地方。 - `state:jmap:since` 是**单一全局游标**,无账户维度(读 `src/state.rs:801`、写 `src/state.rs:809`)——两个账户会互相覆盖游标,这是第一个要动的地方。
- `lock:reconcile` 是**单一全局单飞锁**(Redis 实现 `src/state.rs:1139`)——两个账户的对账会被同一把锁串行化。 - `lock:reconcile` 是**单一全局单飞锁**(Redis 实现 `src/state.rs:1139`)——两个账户的对账会被同一把锁串行化。
- `dedup:jmap:{account_id}:{email_id}` **已经带账户段**(`docs/reference.md` §1),键形无需改。 - `dedup:jmap:{account_id}:{email_id}` **已经带账户段**(`docs/reference.md` §1),键形无需改。
- `ACCOUNT_ID` 环境变量已存在且经校验(留空取 session 主账户,越界值在 `src/domain/jmap/client.rs:485` 报错),业务配置层也能覆盖 `account_id`(`src/config.rs:465` 的 `unwrap_or_else` 回落)。配置面已就绪,缺的只是实例内的分维度状态。 - `ACCOUNT_ID` 环境变量已存在且经校验(留空取 session 主账户,越界值在 `src/domain/jmap/client.rs:513` 报错),业务配置层也能覆盖 `account_id`(`src/config.rs:465` 的 `unwrap_or_else` 回落)。配置面已就绪,缺的只是实例内的分维度状态。
这条不是被跳过,是被明确拒绝:单实例多账户省下的一个部署单元,换不来它引入的路由与游标复杂度。 这条不是被跳过,是被明确拒绝:单实例多账户省下的一个部署单元,换不来它引入的路由与游标复杂度。

View File

@ -97,6 +97,21 @@ pub struct SearchResult {
pub snippets: Vec<SearchSnippet>, pub snippets: Vec<SearchSnippet>,
} }
/// Optional server-side constraints for `Email/query`.
///
/// The initial baseline walk uses this so a freshly-connected mailbox does not
/// enqueue its entire history: it narrows to recent unread mail. Incremental
/// `/changes` walks never call `list_emails_page`, so they are unaffected.
///
/// `received_after` is epoch **seconds** (matching the JMAP `after` filter, which
/// the server compares against `receivedAt`). `unread_only` maps to
/// `notKeyword "$seen"` (RFC 8621 §4.1.2: `$seen` is the read keyword).
#[derive(Debug, Clone, Default)]
pub struct EmailQueryFilter {
pub received_after: Option<i64>,
pub unread_only: bool,
}
#[async_trait] #[async_trait]
pub trait JmapBackend: Send + Sync { pub trait JmapBackend: Send + Sync {
async fn create_push_subscription(&self, callback_url: &str) -> Result<String, JmapError>; async fn create_push_subscription(&self, callback_url: &str) -> Result<String, JmapError>;
@ -144,13 +159,23 @@ pub trait JmapBackend: Send + Sync {
/// Fetch one bounded page without requiring the caller to materialize the /// Fetch one bounded page without requiring the caller to materialize the
/// entire mailbox. Implementations with a native JMAP position query should /// entire mailbox. Implementations with a native JMAP position query should
/// override this; the fallback preserves test backends' existing contract. /// override this; the fallback preserves test backends' existing contract.
///
/// `filter` narrows the server-side query (see [`EmailQueryFilter`]). It
/// carries the baseline's recent-unread window, so `position` paginates
/// within the *filtered* set: it must be held constant across every page of
/// one walk, or a later page would paginate the unfiltered mailbox.
async fn list_emails_page( async fn list_emails_page(
&self, &self,
account_id: &str, account_id: &str,
folder_id: Option<&str>, folder_id: Option<&str>,
position: usize, position: usize,
limit: usize, limit: usize,
filter: Option<EmailQueryFilter>,
) -> Result<Vec<EmailMetadata>, JmapError> { ) -> Result<Vec<EmailMetadata>, JmapError> {
// The fallback has no server-side query to push the filter into; it is
// honored only where `Email/query` exists (`JmapClientBackend`), so a
// test backend that ignores it keeps its existing contract.
let _ = filter;
let all = self let all = self
.list_emails(account_id, folder_id, position.saturating_add(limit)) .list_emails(account_id, folder_id, position.saturating_add(limit))
.await?; .await?;
@ -277,12 +302,13 @@ impl<B: JmapBackend> JmapService<B> {
folder_id: Option<&str>, folder_id: Option<&str>,
position: usize, position: usize,
limit: usize, limit: usize,
filter: Option<EmailQueryFilter>,
) -> Result<Vec<EmailMetadata>, JmapError> { ) -> Result<Vec<EmailMetadata>, JmapError> {
if limit == 0 { if limit == 0 {
return Err(JmapError::InvalidRequest("limit")); return Err(JmapError::InvalidRequest("limit"));
} }
self.backend self.backend
.list_emails_page(&self.account_id, folder_id, position, limit) .list_emails_page(&self.account_id, folder_id, position, limit, filter)
.await .await
} }

View File

@ -2,13 +2,17 @@
//! Concrete read-only adapter (MOD-JMAP-CLIENT, GATE-G1-JMAP-READONLY). //! Concrete read-only adapter (MOD-JMAP-CLIENT, GATE-G1-JMAP-READONLY).
use super::{ use super::{
EmailChanges, EmailContent, EmailMetadata, Folder, JmapBackend, JmapError, SearchResult, EmailChanges, EmailContent, EmailMetadata, EmailQueryFilter, Folder, JmapBackend, JmapError,
SearchSnippet, SearchResult, SearchSnippet,
}; };
use crate::state::{runtime_provider, OutboundConfig, RuntimeConfigProvider}; use crate::state::{runtime_provider, OutboundConfig, RuntimeConfigProvider};
use async_trait::async_trait; use async_trait::async_trait;
use jmap_client::{email, mailbox, DataType, Get}; use jmap_client::{email, mailbox, DataType, Get};
/// RFC 8621 §4.1.2: the read keyword. `notKeyword "$seen"` narrows `Email/query`
/// to unread mail for the initial-baseline filter.
const KEYWORD_SEEN: &str = "$seen";
pub struct JmapClientBackend { pub struct JmapClientBackend {
client: jmap_client::client::Client, client: jmap_client::client::Client,
account_id: String, account_id: String,
@ -270,8 +274,9 @@ impl JmapBackend for JmapClientBackend {
folder_id: Option<&str>, folder_id: Option<&str>,
position: usize, position: usize,
limit: usize, limit: usize,
filter: Option<EmailQueryFilter>,
) -> Result<Vec<EmailMetadata>, JmapError> { ) -> Result<Vec<EmailMetadata>, JmapError> {
self.fetch_email_page(account_id, folder_id, position, limit) self.fetch_email_page(account_id, folder_id, position, limit, filter)
.await .await
} }
@ -369,14 +374,37 @@ impl JmapClientBackend {
folder_id: Option<&str>, folder_id: Option<&str>,
position: usize, position: usize,
limit: usize, limit: usize,
filter: Option<EmailQueryFilter>,
) -> Result<Vec<EmailMetadata>, JmapError> { ) -> Result<Vec<EmailMetadata>, JmapError> {
if account_id != self.account_id || limit == 0 { if account_id != self.account_id || limit == 0 {
return Err(JmapError::InvalidRequest("email query")); return Err(JmapError::InvalidRequest("email query"));
} }
let mut request = self.client.build(); let mut request = self.client.build();
let query = request.query_email(); let query = request.query_email();
// Compose the folder condition (if any) with the baseline filter into one
// `Email/query` filter. A single condition is sent as-is; two or more are
// AND'd. Keeping `position` paginating the *filtered* set is what stops a
// multi-page baseline from flooding page 2+ with the unfiltered mailbox.
let mut conditions: Vec<email::query::Filter> = Vec::new();
if let Some(folder_id) = folder_id { if let Some(folder_id) = folder_id {
query.filter(email::query::Filter::in_mailbox(folder_id)); conditions.push(email::query::Filter::in_mailbox(folder_id));
}
if let Some(filter) = filter {
if let Some(after) = filter.received_after {
conditions.push(email::query::Filter::after(after));
}
if filter.unread_only {
conditions.push(email::query::Filter::not_keyword(KEYWORD_SEEN.to_string()));
}
}
match conditions.len() {
0 => {}
1 => {
query.filter(conditions.swap_remove(0));
}
_ => {
query.filter(jmap_client::core::query::Filter::and(conditions));
}
} }
query.position(position as i32).limit(limit.min(100)); query.position(position as i32).limit(limit.min(100));
let mut response = tokio::time::timeout(self.timeout(), request.send_query_email()) let mut response = tokio::time::timeout(self.timeout(), request.send_query_email())

View File

@ -3,7 +3,7 @@
use crate::ai::LlmClient; use crate::ai::LlmClient;
use crate::channel::telegram::TelegramClient; use crate::channel::telegram::TelegramClient;
use crate::config::BusinessConfig; use crate::config::BusinessConfig;
use crate::domain::jmap::{JmapBackend, JmapService, SearchResult}; use crate::domain::jmap::{EmailQueryFilter, JmapBackend, JmapService, SearchResult};
use crate::domain::Notification; use crate::domain::Notification;
use crate::state::ttl; use crate::state::ttl;
use crate::state::ReliableState; use crate::state::ReliableState;
@ -12,7 +12,7 @@ use std::{
future::Future, future::Future,
pin::Pin, pin::Pin,
sync::{Arc, RwLock}, sync::{Arc, RwLock},
time::Duration, time::{Duration, SystemTime, UNIX_EPOCH},
}; };
#[async_trait] #[async_trait]
@ -180,6 +180,11 @@ const CHANGE_WINDOW_CAP: usize = 4_096;
const RECONCILE_MAX_PAGES: usize = 100; const RECONCILE_MAX_PAGES: usize = 100;
/// Wall-clock ceiling for the whole invocation, walk phase included. /// Wall-clock ceiling for the whole invocation, walk phase included.
const RECONCILE_BUDGET: Duration = Duration::from_secs(20); const RECONCILE_BUDGET: Duration = Duration::from_secs(20);
/// The initial-baseline window: the walk only enqueues mail received within
/// the last N days and still unread, so a freshly-connected mailbox does not
/// flood notifications with its entire archive. `/changes` replay is
/// unaffected (it never calls `list_emails_page`).
const INITIAL_BASELINE_WINDOW_DAYS: u64 = 7;
/// `reconcile:state` is bimodal, and the two modes need different handling. /// `reconcile:state` is bimodal, and the two modes need different handling.
/// ///
@ -401,6 +406,18 @@ impl<B: JmapBackend> MetadataWorker<B> {
let mut pages = 0_usize; let mut pages = 0_usize;
let mut emails = 0_usize; let mut emails = 0_usize;
// The baseline walk is the only caller of `list_emails_page`, and it must
// not enqueue a freshly-connected mailbox's entire history. Narrow every
// page to recent unread mail so the position walk paginates the *filtered*
// set consistently — filtering only page 0 would flood page 2+ from the
// unfiltered set at position 100. `/changes` replay never reaches here.
let baseline_filter = Some(EmailQueryFilter {
received_after: SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|now| now.as_secs() as i64 - (INITIAL_BASELINE_WINDOW_DAYS as i64) * 86_400)
.ok(),
unread_only: true,
});
loop { loop {
if pages >= BASELINE_MAX_PAGES if pages >= BASELINE_MAX_PAGES
|| emails >= BASELINE_MAX_EMAILS || emails >= BASELINE_MAX_EMAILS
@ -417,7 +434,7 @@ impl<B: JmapBackend> MetadataWorker<B> {
} }
let page = self let page = self
.jmap .jmap
.list_emails_page(None, position, BASELINE_PAGE_SIZE) .list_emails_page(None, position, BASELINE_PAGE_SIZE, baseline_filter.clone())
.await .await
.map_err(|_| "jmap_list_failed")?; .map_err(|_| "jmap_list_failed")?;
if page.is_empty() { if page.is_empty() {
@ -1347,6 +1364,7 @@ mod reconcile_rebaseline_tests {
_folder_id: Option<&str>, _folder_id: Option<&str>,
_position: usize, _position: usize,
_limit: usize, _limit: usize,
_filter: Option<EmailQueryFilter>,
) -> Result<Vec<EmailMetadata>, JmapError> { ) -> Result<Vec<EmailMetadata>, JmapError> {
Ok(vec![]) Ok(vec![])
} }
@ -1462,6 +1480,7 @@ mod reconcile_rebaseline_tests {
_folder_id: Option<&str>, _folder_id: Option<&str>,
_position: usize, _position: usize,
_limit: usize, _limit: usize,
_filter: Option<EmailQueryFilter>,
) -> Result<Vec<EmailMetadata>, JmapError> { ) -> Result<Vec<EmailMetadata>, JmapError> {
Ok(vec![EmailMetadata { Ok(vec![EmailMetadata {
id: "m1".to_string(), id: "m1".to_string(),
@ -1527,4 +1546,162 @@ mod reconcile_rebaseline_tests {
"a truncated walk must not replay /changes from a half-listed state" "a truncated walk must not replay /changes from a half-listed state"
); );
} }
/// Records every `EmailQueryFilter` the baseline walk hands to
/// `list_emails_page`. Page 0 returns a short filtered-looking set so the
/// walk advances; every later page returns empty so the walk exhausts after
/// one page. The recorded filters let the test pin acceptance #1 (only
/// recent unread) and the whole-Walk-phase deviation (every page carries
/// the filter, not just position 0).
struct FilteredBaselineBackend {
filters: Arc<std::sync::Mutex<Vec<Option<EmailQueryFilter>>>>,
}
#[async_trait::async_trait]
impl JmapBackend for FilteredBaselineBackend {
async fn create_push_subscription(&self, _callback_url: &str) -> Result<String, JmapError> {
unreachable!("baseline walk never creates push subscriptions")
}
async fn destroy_push_subscription(&self, _subscription_id: &str) -> Result<(), JmapError> {
unreachable!("baseline walk never destroys push subscriptions")
}
async fn verify_push_subscription(
&self,
_subscription_id: &str,
_verification_code: &str,
) -> Result<(), JmapError> {
unreachable!("baseline walk never verifies push subscriptions")
}
async fn session_state(&self) -> Result<String, JmapError> {
Ok("state-fresh".to_string())
}
async fn email_changes(
&self,
_account_id: &str,
_since_state: &str,
_max_changes: usize,
) -> Result<EmailChanges, JmapError> {
Ok(EmailChanges {
new_state: "state-listed".to_string(),
created: vec![],
updated: vec![],
has_more_changes: false,
})
}
async fn list_folders(&self, _account_id: &str) -> Result<Vec<Folder>, JmapError> {
unreachable!("baseline walk never lists folders")
}
async fn list_emails(
&self,
_account_id: &str,
_folder_id: Option<&str>,
_limit: usize,
) -> Result<Vec<EmailMetadata>, JmapError> {
unreachable!("baseline walk uses list_emails_page, not list_emails")
}
async fn list_emails_page(
&self,
_account_id: &str,
_folder_id: Option<&str>,
position: usize,
_limit: usize,
filter: Option<EmailQueryFilter>,
) -> Result<Vec<EmailMetadata>, JmapError> {
self.filters
.lock()
.expect("filter recorder mutex")
.push(filter);
if position == 0 {
Ok(vec![
EmailMetadata {
id: "m-recent-unread-1".to_string(),
subject: Some("recent".to_string()),
sender: Some("a@example.org".to_string()),
received_at: Some(1),
preview: None,
size: None,
has_attachment: false,
},
EmailMetadata {
id: "m-recent-unread-2".to_string(),
subject: Some("recent".to_string()),
sender: Some("b@example.org".to_string()),
received_at: Some(2),
preview: None,
size: None,
has_attachment: false,
},
])
} else {
Ok(vec![])
}
}
async fn read_email(
&self,
_account_id: &str,
_email_id: &str,
) -> Result<Option<EmailContent>, JmapError> {
unreachable!("baseline walk never reads individual emails")
}
async fn search_emails(
&self,
_account_id: &str,
_query: &str,
_limit: usize,
) -> Result<SearchResult, JmapError> {
unreachable!("baseline walk never searches")
}
}
#[tokio::test]
async fn first_reconcile_filters_the_initial_baseline_to_recent_unread() {
let filters: Arc<std::sync::Mutex<Vec<Option<EmailQueryFilter>>>> =
Arc::new(std::sync::Mutex::new(Vec::new()));
let backend = FilteredBaselineBackend {
filters: filters.clone(),
};
let jmap = JmapService::new(backend, "account-x").expect("account id");
let telegram = TelegramClient::with_endpoint(
SecretString::from("token".to_string()),
runtime_provider(OutboundConfig::default()),
"http://127.0.0.1:1/".to_string(),
);
let state: Arc<dyn ReliableState> = Arc::new(MemoryState::enabled_for_tests());
let worker = MetadataWorker::new(jmap, telegram, 42, state, None, 0);
let cursor = worker
.reconcile(None, 100)
.await
.expect("a filtered baseline walk must complete, not error");
assert!(
matches!(
ReconcileCursor::from_persisted(&cursor),
ReconcileCursor::Replay(_)
),
"the walk must exhaust the filtered set and move to the /changes phase"
);
let recorded = filters.lock().expect("filter recorder mutex");
assert!(
!recorded.is_empty(),
"the baseline walk must call list_emails_page at least once"
);
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.expect("post-epoch clock")
.as_secs() as i64;
let expected_after = now - (INITIAL_BASELINE_WINDOW_DAYS as i64) * 86_400;
for f in recorded.iter() {
let f = f.as_ref().expect("every baseline page must carry a filter");
assert!(f.unread_only, "baseline filter must request unread only");
let after = f
.received_after
.expect("baseline filter must bound received_at to the 7-day window");
assert!(
after >= expected_after - 60 && after <= expected_after + 60,
"received_after must be ~now - 7d (got {after}, expected ~{expected_after})"
);
}
}
} }