Something went wrong. Try again.
A personal email screening utility using Himalaya, Jev, and Ratatui
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884188518861887188818891890189118921893189418951896189718981899mod action;mod lesson;mod policy;mod review;
use std::{ collections::{HashMap, HashSet}, env, fmt, ops::Range, path::PathBuf, process::{Command, ExitCode, Stdio}, sync::{ Mutex, atomic::{AtomicUsize, Ordering}, }, thread, time::Duration,};
use action::ClassificationAction;use anyhow::{Context, Result, anyhow, bail};use chrono::{DateTime, Utc};use clap::Parser;use lesson::{LessonStore, MAX_LESSON_EXCERPT_BYTES, PromptLesson, RELEVANT_LESSON_LIMIT};use policy::{ACTIONS, ClassificationPolicy, PolicySource};use reqwest::blocking::Client;use review::{ReviewItem, ReviewSelection};use serde::{Deserialize, Serialize};use serde_json::json;use tracing::{debug, error, info, warn};use tracing_subscriber::EnvFilter;
const DEFAULT_MESSAGE_COUNT: usize = 10;const ENVELOPE_PAGE_SIZE: usize = 100;const CLASSIFICATION_BATCH_SIZE: usize = 10;const MAX_JEV_REQUEST_BYTES: usize = 24 * 1024;const MIN_JEV_CONTENT_BYTES: usize = 512;const DEFAULT_CLASSIFICATION_CONCURRENCY: usize = 4;const MAX_CLASSIFICATION_CONCURRENCY: usize = 16;const MAILBOX_ACTION_BATCH_SIZE: usize = 100;const MAX_CONTENT_BYTES: usize = 4_000;const TYPESAFE_API_URL: &str = "https://api.typesafe.ai/v1/systemone";
#[derive(Parser)]#[command( version, about = "Classify inbox messages with Jev, learn from corrections, and bulk-review mailbox actions", after_help = "EXAMPLES:\n TYPESAFE_API_KEY=... screener\n screener --count 100\n screener --all\n screener --older-than '1 week ago'\n screener --all --older-than '30 minutes ago'\n screener --classification-concurrency 8\n screener --api-token-command 'op read op://Private/TypeSafe/credential'\n screener --policy work\n screener --policy ./policies/custom.json\n screener --lessons-file ./private-lessons.json\n\nPOLICIES:\n A configured policies/default.json overrides the embedded default policy.\n --policy NAME loads policies/NAME.json from the user configuration directory.\n --policy PATH still loads an explicit policy file.\n --install-default-policy writes an editable default.json without overwriting an existing file.\n\nFILTERING:\n --older-than filters before message bodies are read or classified.\n --count limits matching messages after the age filter is applied.\n\nPERFORMANCE:\n Jev batches contain up to 10 messages and stay within a conservative request-size budget.\n Up to 4 batches run concurrently by default.\n --classification-concurrency N sets a bounded concurrency from 1 through 16.\n\nREVIEW:\n Archive and delete recommendations are shown first; press v to show every classification.\n All mailbox actions default to keep. Press t to teach a corrected class and optional note.\n Corrections are stored locally and relevant lessons are included in future Jev requests.\n\nLOGGING:\n RUST_LOG=screener=debug screener")]struct Cli { /// Shell command whose stdout contains the TypeSafe API key. #[arg(long, value_name = "COMMAND")] api_token_command: Option<String>,
/// Select a configured policy by name, or load an explicit policy file path. #[arg(long, value_name = "NAME_OR_PATH")] policy: Option<PathBuf>,
/// Install an editable copy of the built-in default policy and exit. #[arg(long, exclusive = true)] install_default_policy: bool,
/// Override the local JSON file used to persist reviewer lessons. #[arg(long, value_name = "PATH")] lessons_file: Option<PathBuf>,
/// Maximum number of Jev classification batches to run concurrently. #[arg( long, value_name = "N", default_value_t = DEFAULT_CLASSIFICATION_CONCURRENCY, value_parser = parse_classification_concurrency )] classification_concurrency: usize,
/// Number of recent messages to classify. Defaults to 10 unless --all is used. #[arg( short = 'n', long, value_parser = parse_message_count, conflicts_with = "all" )] count: Option<usize>,
/// Classify every message in the inbox. #[arg(long, conflicts_with = "count")] all: bool,
/// Only classify messages older than this relative age, such as "1 week ago". #[arg(long, value_name = "AGE", value_parser = parse_older_than)] older_than: Option<Duration>,}
fn parse_message_count(value: &str) -> Result<usize, String> { let count = value .parse::<usize>() .map_err(|_| "message count must be a positive integer".to_owned())?; if count == 0 { return Err("message count must be greater than zero".to_owned()); } Ok(count)}
fn parse_classification_concurrency(value: &str) -> Result<usize, String> { let concurrency = value.parse::<usize>().map_err(|_| { format!( "classification concurrency must be an integer from 1 through {MAX_CLASSIFICATION_CONCURRENCY}" ) })?; if !(1..=MAX_CLASSIFICATION_CONCURRENCY).contains(&concurrency) { return Err(format!( "classification concurrency must be from 1 through {MAX_CLASSIFICATION_CONCURRENCY}" )); } Ok(concurrency)}
fn parse_older_than(value: &str) -> Result<Duration, String> { let normalized = value.trim().to_ascii_lowercase(); let mut parts = normalized.split_whitespace(); let Some(amount) = parts.next() else { return Err("age must look like \"1 week ago\" or \"30 minutes ago\"".to_owned()); }; let Some(unit) = parts.next() else { return Err("age must look like \"1 week ago\" or \"30 minutes ago\"".to_owned()); }; if parts.next() != Some("ago") || parts.next().is_some() { return Err("age must look like \"1 week ago\" or \"30 minutes ago\"".to_owned()); }
let amount = amount .parse::<u64>() .map_err(|_| "age amount must be a positive integer".to_owned())?; if amount == 0 { return Err("age amount must be greater than zero".to_owned()); }
let seconds_per_unit = match unit { "second" | "seconds" | "sec" | "secs" => 1, "minute" | "minutes" | "min" | "mins" => 60, "hour" | "hours" | "hr" | "hrs" => 60 * 60, "day" | "days" => 24 * 60 * 60, "week" | "weeks" => 7 * 24 * 60 * 60, _ => { return Err("age unit must be seconds, minutes, hours, days, or weeks".to_owned()); } }; let seconds = amount .checked_mul(seconds_per_unit) .ok_or_else(|| "age is too large".to_owned())?;
Ok(Duration::from_secs(seconds))}
fn age_cutoff(now: DateTime<Utc>, age: Duration) -> Result<DateTime<Utc>> { let age = chrono::Duration::from_std(age).context("older-than age is too large")?; now.checked_sub_signed(age) .context("older-than cutoff is outside the supported date range")}
impl Cli { fn message_limit(&self) -> Option<usize> { if self.all { None } else { Some(self.count.unwrap_or(DEFAULT_MESSAGE_COUNT)) } }}
fn envelope_page_size(message_limit: Option<usize>) -> usize { message_limit .unwrap_or(ENVELOPE_PAGE_SIZE) .min(ENVELOPE_PAGE_SIZE)}
#[derive(Debug, Deserialize)]struct MailboxList { mailboxes: Vec<Mailbox>,}
#[derive(Debug, Deserialize)]struct Mailbox { id: String, name: String,}
#[derive(Debug, Deserialize)]struct EnvelopeList { envelopes: Vec<Envelope>,}
#[derive(Debug, Deserialize, Serialize)]struct Envelope { id: String, subject: String, from: Vec<Address>, date: String,}
fn content_read_batches(envelopes: &[Envelope]) -> std::slice::Chunks<'_, Envelope> { envelopes.chunks(CLASSIFICATION_BATCH_SIZE)}
fn classification_worker_count(batch_count: usize, concurrency: usize) -> usize { batch_count.min(concurrency)}
#[derive(Debug, Deserialize, Serialize)]struct Address { name: Option<String>, email: String,}
#[derive(Serialize)]struct MessageState<'a> { id: &'a str, subject: &'a str, from: &'a [Address], content: &'a str, relevant_lessons: Vec<PromptLesson<'a>>,}
#[derive(Deserialize)]struct EvaluationResponse { model: String, answers: HashMap<String, ChoiceAnswer>,}
#[derive(Deserialize)]struct ChoiceAnswer { choice: String, confidence: f64, probabilities: HashMap<String, f64>,}
#[derive(Debug)]struct ClassificationBatch { model: String, items: Vec<ReviewItem>,}
#[derive(Clone, Debug, Eq, PartialEq)]struct ClassificationBatchPlan { range: Range<usize>, lesson_limit: usize, content_limit: usize, request_bytes: usize,}
impl ClassificationBatchPlan { fn len(&self) -> usize { self.range.len() }}
#[derive(Debug)]enum ClassificationCallError { TokenLimit { status: reqwest::StatusCode, body: String, }, Other(anyhow::Error),}
impl fmt::Display for ClassificationCallError { fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { match self { Self::TokenLimit { status, body } => { write!(formatter, "TypeSafe API returned {status}: {}", body.trim()) } Self::Other(error) => fmt::Display::fmt(error, formatter), } }}
#[derive(Clone, Copy)]struct ClassificationContext<'a> { client: &'a Client, api_url: &'a str, api_key: &'a str, policy: &'a ClassificationPolicy, lesson_store: &'a LessonStore, policy_scope: &'a str, concurrency: usize,}
fn main() -> ExitCode { init_tracing();
match run() { Ok(()) => ExitCode::SUCCESS, Err(error) => { error!(error = ?error, "screener failed"); ExitCode::FAILURE } }}
fn init_tracing() { let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("screener=info")); tracing_subscriber::fmt() .with_env_filter(filter) .with_target(false) .with_writer(std::io::stderr) .init();}
fn run() -> Result<()> { let cli = Cli::parse(); if cli.install_default_policy { let (path, created) = PolicySource::install_default()?; if created { println!("Installed editable default policy at {}.", path.display()); } else { println!("Preserved existing default policy at {}.", path.display()); } return Ok(()); } let message_limit = cli.message_limit(); let token_source = if cli.api_token_command.is_some() { "command" } else { "environment" };
match message_limit { Some(limit) => info!(message_limit = limit, "starting inbox screening"), None => info!("starting full inbox screening"), } if let Some(age) = cli.older_than { info!( older_than_seconds = age.as_secs(), "filtering inbox messages by age" ); }
let policy_source = PolicySource::resolve(cli.policy.as_deref())?; let policy = policy_source.load()?; let policy_scope = policy_source.scope()?; debug!(source = %policy_source, "loaded classification policy");
let cutoff = cli .older_than .map(|age| age_cutoff(Utc::now(), age)) .transpose()?; let inbox_id = find_mailbox("Inbox")?; debug!("resolved inbox mailbox"); let envelopes = list_selected_envelopes(&inbox_id, message_limit, cutoff.as_ref())?; info!( message_count = envelopes.len(), "loaded selected inbox envelopes" ); if envelopes.is_empty() { if cutoff.is_some() { println!("No inbox messages matched the --older-than filter."); } else { println!("No messages found in the inbox."); } return Ok(()); }
let lesson_path = match cli.lessons_file.clone() { Some(path) => path, None => LessonStore::default_path()?, }; let mut lesson_store = LessonStore::load(lesson_path)?; debug!( lesson_count = lesson_store.len(), path = %lesson_store.path().display(), "loaded reviewer lessons" );
let api_key = load_api_key(cli.api_token_command.as_deref())?; debug!(source = token_source, "loaded TypeSafe API key"); let client = build_typesafe_client()?; let classification = ClassificationContext { client: &client, api_url: TYPESAFE_API_URL, api_key: &api_key, policy: &policy, lesson_store: &lesson_store, policy_scope: &policy_scope, concurrency: cli.classification_concurrency, }; let (model, items) = classify_all_messages(&classification, &inbox_id, &envelopes)?; info!( message_count = items.len(), "completed inbox classification" );
let Some(selection) = review::run(items, model)? else { println!("Review cancelled. No mailbox changes or lessons saved."); return Ok(()); }; if selection.delete_ids.is_empty() && selection.archive_ids.is_empty() && selection.lessons.is_empty() { println!("No mailbox actions or lessons selected."); return Ok(()); }
let lesson_count = lesson_store.record(&policy_scope, selection.lessons.clone())?; if lesson_count > 0 { println!( "Saved {lesson_count} reviewer lesson(s) to {}.", lesson_store.path().display() ); }
let archive_id = if selection.archive_ids.is_empty() { None } else { Some(find_mailbox("Archive")?) }; apply_review_selection(&inbox_id, archive_id.as_deref(), &selection)?; if !selection.delete_ids.is_empty() || !selection.archive_ids.is_empty() { println!( "Applied mailbox actions: {} deleted, {} archived.", selection.delete_ids.len(), selection.archive_ids.len() ); } info!( delete_count = selection.delete_ids.len(), archive_count = selection.archive_ids.len(), lesson_count, "inbox screening complete" ); Ok(())}
fn load_api_key(command: Option<&str>) -> Result<String> { let key = match command { Some(command) => { let output = Command::new("sh") .args(["-c", command]) .stdin(Stdio::null()) .output() .with_context(|| format!("failed to run API token command: {command}"))?;
if !output.status.success() { let stderr = String::from_utf8_lossy(&output.stderr); bail!( "API token command failed with {}: {}", output.status, stderr.trim() ); }
String::from_utf8(output.stdout) .context("API token command returned non-UTF-8 output")? } None => env::var("TYPESAFE_API_KEY").context( "missing TypeSafe API key; set TYPESAFE_API_KEY or pass --api-token-command", )?, };
let key = key.trim().to_owned(); if key.is_empty() { bail!("TypeSafe API key is empty"); } Ok(key)}
fn find_mailbox(name: &str) -> Result<String> { let output = run_himalaya(&["mailbox", "list", "--json"])?; let list: MailboxList = serde_json::from_str(&output).context("failed to parse Himalaya mailbox list")?;
list.mailboxes .into_iter() .find(|mailbox| mailbox.name.eq_ignore_ascii_case(name)) .map(|mailbox| mailbox.id) .with_context(|| format!("default account has no mailbox named {name}"))}
fn list_envelopes(mailbox_id: &str, page: usize, page_size: usize) -> Result<Vec<Envelope>> { let page = page.to_string(); let page_size = page_size.to_string(); let output = run_himalaya(&[ "envelope", "list", "--mailbox", mailbox_id, "--page", &page, "--page-size", &page_size, "--json", ])?; let list: EnvelopeList = serde_json::from_str(&output).context("failed to parse Himalaya envelope list")?; Ok(list.envelopes)}
fn filter_envelopes_by_age( envelopes: Vec<Envelope>, cutoff: Option<&DateTime<Utc>>,) -> Result<Vec<Envelope>> { let Some(cutoff) = cutoff else { return Ok(envelopes); };
envelopes .into_iter() .filter_map(|envelope| { let received_at = match DateTime::parse_from_rfc3339(&envelope.date) { Ok(received_at) => received_at.with_timezone(&Utc), Err(error) => { return Some(Err(error).with_context(|| { format!( "failed to parse date {:?} for message {}", envelope.date, envelope.id ) })); } }; (received_at < *cutoff).then_some(Ok(envelope)) }) .collect()}
fn list_selected_envelopes( mailbox_id: &str, message_limit: Option<usize>, cutoff: Option<&DateTime<Utc>>,) -> Result<Vec<Envelope>> { collect_selected_envelopes(message_limit, cutoff, |page, page_size| { list_envelopes(mailbox_id, page, page_size) })}
fn collect_selected_envelopes( message_limit: Option<usize>, cutoff: Option<&DateTime<Utc>>, mut load_page: impl FnMut(usize, usize) -> Result<Vec<Envelope>>,) -> Result<Vec<Envelope>> { let page_size = envelope_page_size(message_limit); let mut envelopes = Vec::new(); let mut seen_ids = HashSet::new(); let mut page = 1;
loop { let page_envelopes = load_page(page, page_size)?; let fetched_count = page_envelopes.len(); let page_envelopes = filter_envelopes_by_age(page_envelopes, cutoff)?; for envelope in page_envelopes { if seen_ids.insert(envelope.id.clone()) { envelopes.push(envelope); if message_limit.is_some_and(|limit| envelopes.len() >= limit) { break; } } }
info!( page, fetched_count, selected_message_count = envelopes.len(), "loaded inbox page" ); if message_limit.is_some_and(|limit| envelopes.len() >= limit) || fetched_count < page_size { break; } page += 1; }
Ok(envelopes)}
fn read_message_contents( mailbox_id: &str, envelopes: &[Envelope], processed_count: usize,) -> Result<Vec<String>> { thread::scope(|scope| { let handles: Vec<_> = envelopes .iter() .enumerate() .map(|(offset, envelope)| { scope.spawn(move || { debug!( message_number = processed_count + offset + 1, "reading message" ); let mut content = run_himalaya(&["message", "read", "--mailbox", mailbox_id, &envelope.id])?; let original_bytes = content.len(); truncate_utf8(&mut content, MAX_CONTENT_BYTES); debug!( message_number = processed_count + offset + 1, original_bytes, submitted_bytes = content.len(), "prepared message content" ); Ok(content) }) }) .collect();
handles .into_iter() .map(|handle| { handle .join() .map_err(|_| anyhow!("message reader thread panicked"))? }) .collect() })}
fn run_himalaya(args: &[&str]) -> Result<String> { let output = Command::new("himalaya") .args(args) .stdin(Stdio::null()) .output() .context("failed to run Himalaya; ensure `himalaya` is installed and on PATH")?;
if !output.status.success() { let stderr = String::from_utf8_lossy(&output.stderr); let stdout = String::from_utf8_lossy(&output.stdout); let detail = if stderr.trim().is_empty() { stdout.trim() } else { stderr.trim() }; bail!("Himalaya failed with {}: {detail}", output.status); }
String::from_utf8(output.stdout).context("Himalaya returned non-UTF-8 output")}
fn truncate_utf8(value: &mut String, max_bytes: usize) { if value.len() <= max_bytes { return; }
let mut end = max_bytes; while !value.is_char_boundary(end) { end -= 1; } value.truncate(end);}
fn sender_label(envelope: &Envelope) -> String { envelope .from .first() .map(|address| match address.name.as_deref().map(str::trim) { Some(name) if !name.is_empty() => format!("{name} <{}>", address.email), _ => address.email.clone(), }) .unwrap_or_else(|| "unknown sender".to_owned())}
fn utf8_prefix(value: &str, max_bytes: usize) -> &str { if value.len() <= max_bytes { return value; }
let mut end = max_bytes; while !value.is_char_boundary(end) { end -= 1; } &value[..end]}
fn limited_contents(contents: &[String], max_bytes: usize) -> Vec<&str> { contents .iter() .map(|content| utf8_prefix(content, max_bytes)) .collect()}
fn build_evaluation_request( policy: &ClassificationPolicy, lesson_store: &LessonStore, policy_scope: &str, envelopes: &[Envelope], contents: &[&str], lesson_limit: usize,) -> serde_json::Value { let evaluated_at = Utc::now().to_rfc3339(); let messages: Vec<_> = envelopes .iter() .zip(contents) .map(|(envelope, content)| MessageState { id: &envelope.id, subject: &envelope.subject, from: &envelope.from, content, relevant_lessons: lesson_store.relevant_to_with_limit( policy_scope, &envelope.subject, &sender_label(envelope), content, lesson_limit, ), }) .collect();
let criteria: serde_json::Map<String, serde_json::Value> = ACTIONS .into_iter() .map(|action| { ( action.to_owned(), serde_json::Value::String(format!( "Choose when `policy.actions.{action}` applies." )), ) }) .collect();
let questions: serde_json::Map<String, serde_json::Value> = (0..messages.len()) .map(|index| { ( format!("message_{index}"), json!({ "type": "choice", "instructions": format!( "Apply the complete `policy` to `messages[{index}]` and choose exactly one action. Follow `policy.decision_order` in order and use `policy.examples` as boundary guidance. Use `evaluated_at` when the policy depends on whether a dated message is still current or has expired. Consult `messages[{index}].relevant_lessons`: when a lesson is meaningfully similar, its `user_corrected_to` is stronger boundary evidence than the generic policy. Ignore unrelated lessons and never generalize from sender identity alone. Classification is advisory; mailbox changes require separate user approval." ), "criteria": criteria.clone() }), ) }) .collect();
json!({ "state": { "evaluated_at": evaluated_at, "messages": messages, "policy": policy }, "model": "jev-latest", "questions": questions })}
fn batch_plan( context: &ClassificationContext<'_>, envelopes: &[Envelope], contents: &[String], range: Range<usize>, lesson_limit: usize, content_limit: usize,) -> Result<ClassificationBatchPlan> { let request_contents = limited_contents(&contents[range.clone()], content_limit); let request = build_evaluation_request( context.policy, context.lesson_store, context.policy_scope, &envelopes[range.clone()], &request_contents, lesson_limit, ); let request_bytes = serde_json::to_vec(&request) .context("failed to measure TypeSafe request size")? .len(); Ok(ClassificationBatchPlan { range, lesson_limit, content_limit, request_bytes, })}
fn fit_single_message_batch( context: &ClassificationContext<'_>, envelopes: &[Envelope], contents: &[String], index: usize,) -> Result<ClassificationBatchPlan> { for lesson_limit in (0..=RELEVANT_LESSON_LIMIT).rev() { let plan = batch_plan( context, envelopes, contents, index..index + 1, lesson_limit, MAX_CONTENT_BYTES, )?; if plan.request_bytes <= MAX_JEV_REQUEST_BYTES { return Ok(plan); } }
let mut content_limit = MAX_CONTENT_BYTES / 2; loop { let plan = batch_plan( context, envelopes, contents, index..index + 1, 0, content_limit, )?; if plan.request_bytes <= MAX_JEV_REQUEST_BYTES { return Ok(plan); } if content_limit == MIN_JEV_CONTENT_BYTES { bail!( "message {} cannot fit within the conservative {}-byte Jev request budget even with no lessons and {} content bytes; simplify the selected policy", envelopes[index].id, MAX_JEV_REQUEST_BYTES, MIN_JEV_CONTENT_BYTES ); } content_limit = (content_limit / 2).max(MIN_JEV_CONTENT_BYTES); }}
fn plan_classification_batches( context: &ClassificationContext<'_>, envelopes: &[Envelope], contents: &[String],) -> Result<Vec<ClassificationBatchPlan>> { if envelopes.len() != contents.len() { bail!("message envelope and content counts differ"); }
let mut plans = Vec::new(); let mut start = 0; while start < envelopes.len() { let mut best = None; let max_end = (start + CLASSIFICATION_BATCH_SIZE).min(envelopes.len()); for end in start + 1..=max_end { let plan = batch_plan( context, envelopes, contents, start..end, RELEVANT_LESSON_LIMIT, MAX_CONTENT_BYTES, )?; if plan.request_bytes > MAX_JEV_REQUEST_BYTES { break; } best = Some(plan); }
let plan = match best { Some(plan) => plan, None => fit_single_message_batch(context, envelopes, contents, start)?, }; start = plan.range.end; plans.push(plan); } Ok(plans)}
fn build_typesafe_client() -> Result<Client> { Client::builder() .timeout(Duration::from_secs(60)) .build() .context("failed to build TypeSafe HTTP client")}
fn is_token_limit_error(status: reqwest::StatusCode, body: &str) -> bool { status == reqwest::StatusCode::BAD_REQUEST && serde_json::from_str::<serde_json::Value>(body) .ok() .and_then(|value| { value .pointer("/detail/error_type") .and_then(serde_json::Value::as_str) .map(str::to_owned) }) .as_deref() == Some("max_tokens_exceeded")}
fn classify_messages( context: &ClassificationContext<'_>, envelopes: &[Envelope], contents: &[&str], lesson_limit: usize,) -> std::result::Result<EvaluationResponse, ClassificationCallError> { let request = build_evaluation_request( context.policy, context.lesson_store, context.policy_scope, envelopes, contents, lesson_limit, ); let response = context .client .post(context.api_url) .bearer_auth(context.api_key) .json(&request) .send() .map_err(|error| { ClassificationCallError::Other( anyhow::Error::new(error).context("failed to call the TypeSafe API"), ) })?;
let status = response.status(); let body = response.text().map_err(|error| { ClassificationCallError::Other( anyhow::Error::new(error).context("failed to read the TypeSafe API response"), ) })?; if is_token_limit_error(status, &body) { return Err(ClassificationCallError::TokenLimit { status, body }); } if !status.is_success() { return Err(ClassificationCallError::Other(anyhow!( "TypeSafe API returned {status}: {}", body.trim() ))); }
serde_json::from_str(&body).map_err(|error| { ClassificationCallError::Other( anyhow::Error::new(error).context("failed to parse the TypeSafe API response"), ) })}
fn read_all_message_contents( context: &ClassificationContext<'_>, mailbox_id: &str, envelopes: &[Envelope],) -> Result<Vec<String>> { let batches: Vec<_> = content_read_batches(envelopes).collect(); let next_batch = AtomicUsize::new(0); let results: Mutex<Vec<Option<Result<Vec<String>>>>> = Mutex::new((0..batches.len()).map(|_| None).collect());
thread::scope(|scope| -> Result<()> { let worker_count = classification_worker_count(batches.len(), context.concurrency); let handles: Vec<_> = (0..worker_count) .map(|_| { scope.spawn(|| -> Result<()> { loop { let batch_index = next_batch.fetch_add(1, Ordering::Relaxed); let Some(batch) = batches.get(batch_index).copied() else { break; }; let result = read_message_contents( mailbox_id, batch, batch_index * CLASSIFICATION_BATCH_SIZE, ); results .lock() .map_err(|_| anyhow!("message content result lock poisoned"))? [batch_index] = Some(result); } Ok(()) }) }) .collect();
for handle in handles { handle .join() .map_err(|_| anyhow!("message content worker panicked"))??; } Ok(()) })?;
let results = results .into_inner() .map_err(|_| anyhow!("message content result lock poisoned"))?; let mut contents = Vec::with_capacity(envelopes.len()); for (batch_index, result) in results.into_iter().enumerate() { let mut batch = result .with_context(|| format!("message content worker skipped batch {}", batch_index + 1))? .with_context(|| format!("message content batch {} failed", batch_index + 1))?; contents.append(&mut batch); } Ok(contents)}
fn smaller_single_message_plan( context: &ClassificationContext<'_>, envelopes: &[Envelope], contents: &[String], plan: &ClassificationBatchPlan,) -> Result<Option<ClassificationBatchPlan>> { if plan.lesson_limit > 0 { return batch_plan( context, envelopes, contents, plan.range.clone(), plan.lesson_limit - 1, plan.content_limit, ) .map(Some); } if plan.content_limit > MIN_JEV_CONTENT_BYTES { return batch_plan( context, envelopes, contents, plan.range.clone(), 0, (plan.content_limit / 2).max(MIN_JEV_CONTENT_BYTES), ) .map(Some); } Ok(None)}
fn classify_planned_batch( context: &ClassificationContext<'_>, envelopes: &[Envelope], contents: &[String], plan: ClassificationBatchPlan,) -> Result<Vec<ClassificationBatch>> { let batch_envelopes = &envelopes[plan.range.clone()]; let batch_contents = &contents[plan.range.clone()]; let request_contents = limited_contents(batch_contents, plan.content_limit); info!( first_message = plan.range.start + 1, batch_size = plan.len(), request_bytes = plan.request_bytes, lesson_limit = plan.lesson_limit, content_limit = plan.content_limit, "requesting token-safe Jev classification batch" );
match classify_messages( context, batch_envelopes, &request_contents, plan.lesson_limit, ) { Ok(evaluation) => { let items = build_review_items(batch_envelopes, batch_contents, &evaluation)?; Ok(vec![ClassificationBatch { model: evaluation.model, items, }]) } Err(ClassificationCallError::Other(error)) => Err(error), Err(ClassificationCallError::TokenLimit { .. }) if plan.len() > 1 => { let midpoint = plan.range.start + plan.len() / 2; warn!( first_message = plan.range.start + 1, batch_size = plan.len(), "TypeSafe rejected a preflighted batch for token length; splitting and retrying" ); let left = batch_plan( context, envelopes, contents, plan.range.start..midpoint, plan.lesson_limit, plan.content_limit, )?; let right = batch_plan( context, envelopes, contents, midpoint..plan.range.end, plan.lesson_limit, plan.content_limit, )?; let mut classified = classify_planned_batch(context, envelopes, contents, left)?; classified.extend(classify_planned_batch(context, envelopes, contents, right)?); Ok(classified) } Err(ClassificationCallError::TokenLimit { status, body }) => { let Some(smaller) = smaller_single_message_plan(context, envelopes, contents, &plan)? else { bail!( "TypeSafe rejected message {} at the minimum safe request size: {status}: {}", batch_envelopes[0].id, body.trim() ); }; warn!( message_id = %batch_envelopes[0].id, lesson_limit = smaller.lesson_limit, content_limit = smaller.content_limit, "TypeSafe rejected a single-message request for token length; reducing optional context and retrying" ); classify_planned_batch(context, envelopes, contents, smaller) } }}
fn classify_all_messages( context: &ClassificationContext<'_>, mailbox_id: &str, envelopes: &[Envelope],) -> Result<(String, Vec<ReviewItem>)> { let contents = read_all_message_contents(context, mailbox_id, envelopes)?; let batches = plan_classification_batches(context, envelopes, &contents)?; info!( batch_count = batches.len(), request_byte_budget = MAX_JEV_REQUEST_BYTES, "planned token-safe Jev classification batches" ); let next_batch = AtomicUsize::new(0); let results: Mutex<Vec<Option<Result<Vec<ClassificationBatch>>>>> = Mutex::new((0..batches.len()).map(|_| None).collect());
thread::scope(|scope| -> Result<()> { let worker_count = classification_worker_count(batches.len(), context.concurrency); let handles: Vec<_> = (0..worker_count) .map(|_| { scope.spawn(|| -> Result<()> { loop { let batch_index = next_batch.fetch_add(1, Ordering::Relaxed); let Some(batch) = batches.get(batch_index).cloned() else { break; }; let result = classify_planned_batch(context, envelopes, &contents, batch); results .lock() .map_err(|_| anyhow!("classification result lock poisoned"))? [batch_index] = Some(result); } Ok(()) }) }) .collect();
for handle in handles { handle .join() .map_err(|_| anyhow!("classification worker panicked"))??; } Ok(()) })?;
let results = results .into_inner() .map_err(|_| anyhow!("classification result lock poisoned"))?; let mut model: Option<String> = None; let mut items = Vec::with_capacity(envelopes.len()); for (batch_index, result) in results.into_iter().enumerate() { let batches = result .with_context(|| format!("classification worker skipped batch {}", batch_index + 1))? .with_context(|| format!("classification batch {} failed", batch_index + 1))?; for batch in batches { if let Some(previous_model) = &model && previous_model != &batch.model { warn!( previous_model = %previous_model, model = %batch.model, "Jev model changed during inbox screening" ); } model.get_or_insert_with(|| batch.model.clone()); items.extend(batch.items); } }
Ok(( model.context("classification produced no model result")?, items, ))}
fn build_review_items( envelopes: &[Envelope], contents: &[String], evaluation: &EvaluationResponse,) -> Result<Vec<ReviewItem>> { let mut items = Vec::new(); for (index, envelope) in envelopes.iter().enumerate() { let answer = evaluation .answers .get(&format!("message_{index}")) .with_context(|| format!("TypeSafe response is missing message_{index}"))?; let action = answer .choice .parse::<ClassificationAction>() .map_err(|error| anyhow!(error))?; let content = contents .get(index) .with_context(|| format!("message content is missing for message_{index}"))?; let probability = answer .probabilities .get(&answer.choice) .copied() .unwrap_or_default(); let sender = sender_label(envelope); let mut content_excerpt = content.clone(); truncate_utf8(&mut content_excerpt, MAX_LESSON_EXCERPT_BYTES); items.push(ReviewItem::new( envelope.id.clone(), envelope.subject.clone(), sender, content_excerpt, action, probability, answer.confidence, )); } Ok(items)}
fn archive_command_args(source: &str, destination: &str, ids: &[String]) -> Vec<String> { let mut args = vec![ "message".to_owned(), "move".to_owned(), "--from".to_owned(), source.to_owned(), "--to".to_owned(), destination.to_owned(), ]; args.extend(ids.iter().cloned()); args}
fn delete_command_args(mailbox_id: &str, ids: &[String]) -> Vec<String> { let mut args = vec![ "message".to_owned(), "delete".to_owned(), "--mailbox".to_owned(), mailbox_id.to_owned(), ]; args.extend(ids.iter().cloned()); args}
fn run_himalaya_owned(args: &[String]) -> Result<String> { let args: Vec<_> = args.iter().map(String::as_str).collect(); run_himalaya(&args)}
fn apply_review_selection( inbox_id: &str, archive_id: Option<&str>, actions: &ReviewSelection,) -> Result<()> { if !actions.archive_ids.is_empty() { let archive_id = archive_id.context("archive mailbox was not resolved")?; for ids in actions.archive_ids.chunks(MAILBOX_ACTION_BATCH_SIZE) { let args = archive_command_args(inbox_id, archive_id, ids); run_himalaya_owned(&args).context("failed to archive approved messages")?; } info!( archive_count = actions.archive_ids.len(), "archived approved messages" ); }
if !actions.delete_ids.is_empty() { for ids in actions.delete_ids.chunks(MAILBOX_ACTION_BATCH_SIZE) { let args = delete_command_args(inbox_id, ids); run_himalaya_owned(&args).context("failed to delete approved messages")?; } info!( delete_count = actions.delete_ids.len(), "deleted approved messages" ); }
Ok(())}
#[cfg(test)]mod tests { use super::*; use std::{ io::{Read, Write}, net::{TcpListener, TcpStream}, sync::mpsc, thread, };
fn test_envelope(id: &str, date: &str) -> Envelope { Envelope { id: id.to_owned(), subject: format!("Subject for {id}"), from: vec![Address { name: Some("Sender".to_owned()), email: "sender@example.com".to_owned(), }], date: date.to_owned(), } }
fn empty_lesson_store() -> LessonStore { let path = std::env::temp_dir().join(format!( "screener-empty-lessons-{}-{:?}.json", std::process::id(), thread::current().id() )); LessonStore::load(path).unwrap() }
fn read_http_json(stream: &mut TcpStream) -> serde_json::Value { let mut request = Vec::new(); let mut buffer = [0; 4096]; let (body_start, content_length) = loop { let read = stream.read(&mut buffer).unwrap(); assert_ne!(read, 0); request.extend_from_slice(&buffer[..read]); if let Some(header_end) = request.windows(4).position(|part| part == b"\r\n\r\n") { let headers = String::from_utf8_lossy(&request[..header_end]); let content_length = headers .lines() .find_map(|line| { let (name, value) = line.split_once(':')?; name.eq_ignore_ascii_case("content-length") .then(|| value.trim().parse::<usize>().unwrap()) }) .unwrap(); break (header_end + 4, content_length); } }; while request.len() < body_start + content_length { let read = stream.read(&mut buffer).unwrap(); assert_ne!(read, 0); request.extend_from_slice(&buffer[..read]); } serde_json::from_slice(&request[body_start..body_start + content_length]).unwrap() }
fn write_http_json(stream: &mut TcpStream, status: &str, body: &serde_json::Value) { let body = serde_json::to_string(body).unwrap(); write!( stream, "HTTP/1.1 {status}\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{}", body.len(), body ) .unwrap(); }
#[test] fn token_command_output_is_trimmed() { let key = load_api_key(Some("printf 'test-key\\n'")).unwrap(); assert_eq!(key, "test-key"); }
#[test] fn message_selection_supports_limits_and_full_inbox() { let limited = Cli::try_parse_from(["screener", "--count", "250"]).unwrap(); assert_eq!(limited.message_limit(), Some(250));
let defaulted = Cli::try_parse_from(["screener"]).unwrap(); assert_eq!(defaulted.message_limit(), Some(DEFAULT_MESSAGE_COUNT));
let all = Cli::try_parse_from(["screener", "--all"]).unwrap(); assert_eq!(all.message_limit(), None);
assert!(Cli::try_parse_from(["screener", "--count", "0"]).is_err()); assert!(Cli::try_parse_from(["screener", "--all", "--count", "5"]).is_err()); }
#[test] fn classification_concurrency_is_configurable_and_bounded() { let defaulted = Cli::try_parse_from(["screener"]).unwrap(); assert_eq!( defaulted.classification_concurrency, DEFAULT_CLASSIFICATION_CONCURRENCY );
let configured = Cli::try_parse_from(["screener", "--classification-concurrency", "8"]).unwrap(); assert_eq!(configured.classification_concurrency, 8);
assert!(Cli::try_parse_from(["screener", "--classification-concurrency", "0"]).is_err()); assert!(Cli::try_parse_from(["screener", "--classification-concurrency", "17"]).is_err()); assert_eq!(classification_worker_count(2, 8), 2); assert_eq!(classification_worker_count(20, 8), 8); }
#[test] fn parses_supported_older_than_expressions() { assert_eq!( parse_older_than("1 week ago").unwrap(), Duration::from_secs(7 * 24 * 60 * 60) ); assert_eq!( parse_older_than("1 hour ago").unwrap(), Duration::from_secs(60 * 60) ); assert_eq!( parse_older_than("30 minutes ago").unwrap(), Duration::from_secs(30 * 60) ); assert_eq!( parse_older_than("2 DAYS AGO").unwrap(), Duration::from_secs(2 * 24 * 60 * 60) );
assert!(parse_older_than("0 minutes ago").is_err()); assert!(parse_older_than("1 month ago").is_err()); assert!(parse_older_than("1 week").is_err());
let cli = Cli::try_parse_from(["screener", "--older-than", "30 minutes ago"]).unwrap(); assert_eq!(cli.older_than, Some(Duration::from_secs(30 * 60))); }
#[test] fn filters_envelopes_strictly_older_than_cutoff() { let cutoff = DateTime::parse_from_rfc3339("2026-09-19T12:00:00Z") .unwrap() .with_timezone(&Utc); let envelopes = vec![ test_envelope("older", "2026-09-19T11:59:59Z"), test_envelope("boundary", "2026-09-19T12:00:00Z"), test_envelope("newer", "2026-09-19T12:00:01Z"), ];
let filtered = filter_envelopes_by_age(envelopes, Some(&cutoff)).unwrap(); assert_eq!( filtered .iter() .map(|envelope| envelope.id.as_str()) .collect::<Vec<_>>(), ["older"] ); }
#[test] fn message_count_applies_after_age_filtering() { let cutoff = DateTime::parse_from_rfc3339("2026-09-19T12:00:00Z") .unwrap() .with_timezone(&Utc); let mut pages = vec![ vec![ test_envelope("newer-1", "2026-09-19T12:01:00Z"), test_envelope("newer-2", "2026-09-19T12:02:00Z"), ], vec![ test_envelope("older-1", "2026-09-19T11:59:00Z"), test_envelope("older-2", "2026-09-19T11:58:00Z"), ], ] .into_iter();
let selected = collect_selected_envelopes(Some(2), Some(&cutoff), |_, page_size| { assert_eq!(page_size, 2); Ok(pages.next().unwrap_or_default()) }) .unwrap();
assert_eq!( selected .iter() .map(|envelope| envelope.id.as_str()) .collect::<Vec<_>>(), ["older-1", "older-2"] ); }
#[test] fn rejects_unparseable_envelope_dates_when_filtering() { let cutoff = DateTime::parse_from_rfc3339("2026-09-19T12:00:00Z") .unwrap() .with_timezone(&Utc); let error = filter_envelopes_by_age(vec![test_envelope("broken", "not-a-date")], Some(&cutoff)) .unwrap_err();
assert!(error.to_string().contains("message broken")); }
#[test] fn builds_review_items_for_every_classification() { let envelopes: Vec<_> = ["delete-id", "archive-id", "reply-id", "review-id"] .into_iter() .map(|id| test_envelope(id, "2026-09-19T09:00:25-04:00")) .collect(); let contents: Vec<_> = envelopes .iter() .map(|envelope| format!("Content for {}", envelope.id)) .collect(); let answers = ["delete", "archive", "reply", "review"] .into_iter() .enumerate() .map(|(index, choice)| { ( format!("message_{index}"), ChoiceAnswer { choice: choice.to_owned(), confidence: 0.75, probabilities: HashMap::from([(choice.to_owned(), 0.8)]), }, ) }) .collect(); let evaluation = EvaluationResponse { model: "jev-test".to_owned(), answers, };
let items = build_review_items(&envelopes, &contents, &evaluation).unwrap(); assert_eq!(items.len(), 4); assert_eq!(items[0].action, ClassificationAction::Delete); assert_eq!(items[1].action, ClassificationAction::Archive); assert_eq!(items[2].action, ClassificationAction::Reply); assert_eq!(items[3].action, ClassificationAction::Review); assert_eq!(items[0].content_excerpt, "Content for delete-id"); }
#[test] fn builds_himalaya_batch_mutation_commands() { let ids = vec!["one".to_owned(), "two".to_owned()]; assert_eq!( archive_command_args("inbox-id", "archive-id", &ids), [ "message", "move", "--from", "inbox-id", "--to", "archive-id", "one", "two" ] ); assert_eq!( delete_command_args("inbox-id", &ids), ["message", "delete", "--mailbox", "inbox-id", "one", "two"] ); }
#[test] fn bounds_page_and_classification_batch_sizes() { assert_eq!(envelope_page_size(None), ENVELOPE_PAGE_SIZE); assert_eq!(envelope_page_size(Some(250)), ENVELOPE_PAGE_SIZE); assert_eq!(envelope_page_size(Some(10)), 10);
let envelopes: Vec<_> = (0..12) .map(|id| test_envelope(&id.to_string(), "2026-09-19T09:00:25-04:00")) .collect(); let batch_sizes: Vec<_> = content_read_batches(&envelopes) .map(|batch| batch.len()) .collect(); assert_eq!(batch_sizes, [10, 2]); }
#[test] fn plans_classification_batches_within_conservative_request_budget() { let envelopes: Vec<_> = (0..8) .map(|id| test_envelope(&id.to_string(), "2026-09-19T09:00:25-04:00")) .collect(); let contents = vec!["x".repeat(MAX_CONTENT_BYTES); envelopes.len()]; let policy = PolicySource::EmbeddedDefault.load().unwrap(); let lesson_store = empty_lesson_store(); let client = build_typesafe_client().unwrap(); let context = ClassificationContext { client: &client, api_url: "http://unused.invalid", api_key: "unused", policy: &policy, lesson_store: &lesson_store, policy_scope: "embedded:default", concurrency: DEFAULT_CLASSIFICATION_CONCURRENCY, };
let plans = plan_classification_batches(&context, &envelopes, &contents).unwrap();
assert!(plans.len() > 1); assert!(plans.iter().all(|plan| { plan.request_bytes <= MAX_JEV_REQUEST_BYTES && plan.len() <= CLASSIFICATION_BATCH_SIZE })); assert_eq!(plans.first().unwrap().range.start, 0); assert_eq!(plans.last().unwrap().range.end, envelopes.len()); assert!( plans .windows(2) .all(|pair| pair[0].range.end == pair[1].range.start) ); }
#[test] fn reduces_lessons_when_one_message_would_exceed_request_budget() { let path = std::env::temp_dir().join(format!( "screener-large-prompt-lessons-{}-{}.json", std::process::id(), Utc::now().timestamp_micros() )); let mut lesson_store = LessonStore::load(path.clone()).unwrap(); let lessons = (0..RELEVANT_LESSON_LIMIT) .map(|index| crate::lesson::LessonInput { message_id: format!("lesson-{index}"), subject: format!("Dog food promotion {index}"), sender: "Store <offers@example.com>".to_owned(), content_excerpt: "dog food promotion ".repeat(100), predicted_action: ClassificationAction::Archive, corrected_action: ClassificationAction::Delete, note: Some("historical preference context ".repeat(250)), }) .collect(); lesson_store.record("embedded:default", lessons).unwrap(); let envelopes = vec![test_envelope("current", "2026-09-19T09:00:25-04:00")]; let contents = vec!["dog food promotion ".repeat(200)]; let policy = PolicySource::EmbeddedDefault.load().unwrap(); let client = build_typesafe_client().unwrap(); let context = ClassificationContext { client: &client, api_url: "http://unused.invalid", api_key: "unused", policy: &policy, lesson_store: &lesson_store, policy_scope: "embedded:default", concurrency: DEFAULT_CLASSIFICATION_CONCURRENCY, };
let plans = plan_classification_batches(&context, &envelopes, &contents).unwrap();
assert_eq!(plans.len(), 1); assert!(plans[0].request_bytes <= MAX_JEV_REQUEST_BYTES); assert!(plans[0].lesson_limit < RELEVANT_LESSON_LIMIT); assert_eq!(plans[0].content_limit, MAX_CONTENT_BYTES); std::fs::remove_file(path).unwrap(); }
#[test] fn truncation_preserves_utf8_boundaries() { let mut value = "abécd".to_owned(); truncate_utf8(&mut value, 3); assert_eq!(value, "ab"); }
#[test] fn parses_observed_himalaya_envelope_shape() { let list: EnvelopeList = serde_json::from_str( r#"{"envelopes":[{"id":"42","subject":"Hello","from":[{"name":"Ada","email":"ada@example.com"}],"date":"2026-09-19T09:00:25-04:00"}]}"#, ) .unwrap();
assert_eq!(list.envelopes[0].id, "42"); assert_eq!(list.envelopes[0].from[0].email, "ada@example.com"); assert_eq!(list.envelopes[0].date, "2026-09-19T09:00:25-04:00"); }
#[test] fn injects_relevant_reviewer_lessons_into_message_state() { let path = std::env::temp_dir().join(format!( "screener-prompt-lessons-{}-{}.json", std::process::id(), Utc::now().timestamp_micros() )); let mut lesson_store = LessonStore::load(path.clone()).unwrap(); lesson_store .record( "embedded:default", vec![crate::lesson::LessonInput { message_id: "old-message".to_owned(), subject: "Last chance dog food sale".to_owned(), sender: "Pet Store <offers@example.com>".to_owned(), content_excerpt: "Save on dog food before the promotion ends.".to_owned(), predicted_action: ClassificationAction::Archive, corrected_action: ClassificationAction::Delete, note: Some("Promotional urgency is disposable.".to_owned()), }], ) .unwrap(); let envelopes = vec![Envelope { id: "new-message".to_owned(), subject: "Final hours for dog food savings".to_owned(), from: vec![Address { name: Some("Another Pet Store".to_owned()), email: "pets@example.net".to_owned(), }], date: "2026-09-19T09:00:25-04:00".to_owned(), }]; let contents = vec!["This dog food promotion ends tonight.".to_owned()]; let policy = PolicySource::EmbeddedDefault.load().unwrap(); let content_refs = limited_contents(&contents, MAX_CONTENT_BYTES);
let request = build_evaluation_request( &policy, &lesson_store, "embedded:default", &envelopes, &content_refs, RELEVANT_LESSON_LIMIT, );
let lessons = request["state"]["messages"][0]["relevant_lessons"] .as_array() .unwrap(); assert_eq!(lessons.len(), 1); assert_eq!(lessons[0]["model_suggested"], "archive"); assert_eq!(lessons[0]["user_corrected_to"], "delete"); assert_eq!( lessons[0]["user_note"], "Promotional urgency is disposable." ); assert!( request["questions"]["message_0"]["instructions"] .as_str() .unwrap() .contains("stronger boundary evidence") );
std::fs::remove_file(path).unwrap(); }
#[test] fn splits_and_retries_when_typesafe_reports_a_token_limit() { let listener = TcpListener::bind("127.0.0.1:0").unwrap(); let api_url = format!("http://{}/v1/systemone", listener.local_addr().unwrap()); let server = thread::spawn(move || { for request_index in 0..3 { let (mut stream, _) = listener.accept().unwrap(); let request = read_http_json(&mut stream); if request_index == 0 { assert_eq!(request["state"]["messages"].as_array().unwrap().len(), 2); write_http_json( &mut stream, "400 Bad Request", &json!({"detail": {"error_type": "max_tokens_exceeded"}}), ); continue; }
let messages = request["state"]["messages"].as_array().unwrap(); assert_eq!(messages.len(), 1); let id = messages[0]["id"].as_str().unwrap(); let choice = match id { "first" => "delete", "second" => "archive", _ => panic!("unexpected message id {id}"), }; write_http_json( &mut stream, "200 OK", &json!({ "model": "jev-test", "answers": { "message_0": { "type": "choice", "choice": choice, "confidence": 0.8, "probabilities": { "reply": 0.0, "review": 0.0, "archive": if choice == "archive" { 0.9 } else { 0.1 }, "delete": if choice == "delete" { 0.9 } else { 0.1 } } } }, "usage": {"input_tokens": 1, "output_tokens": 1} }), ); } }); let envelopes = vec![ test_envelope("first", "2026-09-19T09:00:25-04:00"), test_envelope("second", "2026-09-19T09:00:25-04:00"), ]; let contents = vec!["First message".to_owned(), "Second message".to_owned()]; let policy = PolicySource::EmbeddedDefault.load().unwrap(); let lesson_store = empty_lesson_store(); let client = build_typesafe_client().unwrap(); let context = ClassificationContext { client: &client, api_url: &api_url, api_key: "test-key", policy: &policy, lesson_store: &lesson_store, policy_scope: "embedded:default", concurrency: DEFAULT_CLASSIFICATION_CONCURRENCY, }; let plan = batch_plan( &context, &envelopes, &contents, 0..2, RELEVANT_LESSON_LIMIT, MAX_CONTENT_BYTES, ) .unwrap();
let classified = classify_planned_batch(&context, &envelopes, &contents, plan).unwrap();
let actions: Vec<_> = classified .iter() .flat_map(|batch| batch.items.iter().map(|item| item.action)) .collect(); assert_eq!( actions, [ClassificationAction::Delete, ClassificationAction::Archive] ); server.join().unwrap(); }
#[test] fn sends_batched_message_questions_to_typesafe() { let listener = TcpListener::bind("127.0.0.1:0").unwrap(); let api_url = format!("http://{}/v1/systemone", listener.local_addr().unwrap()); let (request_tx, request_rx) = mpsc::channel(); let server = thread::spawn(move || { let (mut stream, _) = listener.accept().unwrap(); let mut request = Vec::new(); let mut buffer = [0; 4096]; let (body_start, content_length) = loop { let read = stream.read(&mut buffer).unwrap(); assert_ne!(read, 0); request.extend_from_slice(&buffer[..read]);
if let Some(header_end) = request.windows(4).position(|part| part == b"\r\n\r\n") { let headers = String::from_utf8_lossy(&request[..header_end]); let content_length = headers .lines() .find_map(|line| { let (name, value) = line.split_once(':')?; name.eq_ignore_ascii_case("content-length") .then(|| value.trim().parse::<usize>().unwrap()) }) .unwrap(); break (header_end + 4, content_length); } };
while request.len() < body_start + content_length { let read = stream.read(&mut buffer).unwrap(); assert_ne!(read, 0); request.extend_from_slice(&buffer[..read]); } request_tx .send(request[body_start..body_start + content_length].to_vec()) .unwrap();
let response = r#"{"model":"jev-test","answers":{"message_0":{"type":"choice","choice":"reply","confidence":0.8,"probabilities":{"reply":0.9,"review":0.0,"archive":0.1,"delete":0.0}}},"usage":{"input_tokens":1,"output_tokens":1}}"#; write!( stream, "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{}", response.len(), response ) .unwrap(); });
let envelopes = vec![Envelope { id: "42".to_owned(), subject: "Can we talk?".to_owned(), from: vec![Address { name: Some("Ada".to_owned()), email: "ada@example.com".to_owned(), }], date: "2026-09-19T09:00:25-04:00".to_owned(), }]; let policy = PolicySource::EmbeddedDefault.load().unwrap(); let contents = vec!["Please reply when you can.".to_owned()]; let client = build_typesafe_client().unwrap(); let lesson_store = empty_lesson_store(); let classification = ClassificationContext { client: &client, api_url: &api_url, api_key: "test-key", policy: &policy, lesson_store: &lesson_store, policy_scope: "embedded:default", concurrency: DEFAULT_CLASSIFICATION_CONCURRENCY, }; let content_refs = limited_contents(&contents, MAX_CONTENT_BYTES); let evaluation = classify_messages( &classification, &envelopes, &content_refs, RELEVANT_LESSON_LIMIT, ) .unwrap();
assert_eq!(evaluation.model, "jev-test"); assert_eq!(evaluation.answers["message_0"].choice, "reply");
let request: serde_json::Value = serde_json::from_slice(&request_rx.recv().unwrap()).unwrap(); assert_eq!(request["model"], "jev-latest"); assert_eq!(request["state"]["messages"][0]["id"], "42"); DateTime::parse_from_rfc3339(request["state"]["evaluated_at"].as_str().unwrap()).unwrap(); assert!( request["state"]["messages"][0]["relevant_lessons"] .as_array() .unwrap() .is_empty() ); assert!(request["state"]["policy"]["decision_order"].is_array()); assert_eq!( request["questions"]["message_0"]["criteria"] .as_object() .unwrap() .len(), ACTIONS.len() ); assert!( request["questions"]["message_0"]["criteria"]["review"] .as_str() .is_some() ); server.join().unwrap(); }}