diff --git a/Cargo.lock b/Cargo.lock index 8cf27ab..e2740cd 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -89,6 +89,15 @@ dependencies = [ "hybrid-array", ] +[[package]] +name = "block2" +version = "0.6.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cdeb9d870516001442e364c5220d3574d2da8dc765554b4a617230d33fa58ef5" +dependencies = [ + "objc2", +] + [[package]] name = "bumpalo" version = "3.20.3" @@ -259,6 +268,16 @@ dependencies = [ "windows-sys 0.48.0", ] +[[package]] +name = "dispatch2" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e0e367e4e7da84520dedcac1901e4da967309406d1e51017ae1abfb97adbd38" +dependencies = [ + "bitflags", + "objc2", +] + [[package]] name = "displaydoc" version = "0.2.6" @@ -948,6 +967,56 @@ dependencies = [ "libc", ] +[[package]] +name = "objc2" +version = "0.6.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3a12a8ed07aefc768292f076dc3ac8c48f3781c8f2d5851dd3d98950e8c5a89f" +dependencies = [ + "objc2-encode", +] + +[[package]] +name = "objc2-contacts" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b034b578389f89a85c055eacc8d8b368be5f04a6c1b07f672bf3aec21d0ef621" +dependencies = [ + "block2", + "objc2", + "objc2-foundation", +] + +[[package]] +name = "objc2-core-foundation" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2a180dd8642fa45cdb7dd721cd4c11b1cadd4929ce112ebd8b9f5803cc79d536" +dependencies = [ + "bitflags", + "dispatch2", + "objc2", +] + +[[package]] +name = "objc2-encode" +version = "4.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ef25abbcd74fb2609453eb695bd2f860d389e457f67dc17cafc8b8cbc89d0c33" + +[[package]] +name = "objc2-foundation" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e3e0adef53c21f888deb4fa59fc59f7eb17404926ee8a6f59f5df0fd7f9f3272" +dependencies = [ + "bitflags", + "block2", + "libc", + "objc2", + "objc2-core-foundation", +] + [[package]] name = "once_cell" version = "1.21.4" @@ -1648,6 +1717,7 @@ version = "0.1.1" dependencies = [ "anyhow", "async-trait", + "block2", "chrono", "dirs", "dotenvy", @@ -1655,6 +1725,9 @@ dependencies = [ "git2", "hex", "log", + "objc2", + "objc2-contacts", + "objc2-foundation", "parking_lot", "rand", "regex", diff --git a/Cargo.toml b/Cargo.toml index 059a23b..3a737ce 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -79,6 +79,22 @@ wiki-git = ["dep:git2", "dep:hex"] # about (its `obsidian.json` registry). obsidian = ["dep:dirs"] +# Contact resolution and scoring (`memory::people`): a SQLite store of people, +# handle aliases and interactions, a deterministic handle → `PersonId` resolver, +# and a recency × frequency × reciprocity × depth scorer. +# +# Implies `tokio` because the store shares its connection as an +# `Arc>` across tasks. It adds no other +# dependency — `rusqlite`, `chrono`, `uuid` and `serde` are already core. +people = ["tokio"] + +# The macOS system address book (`memory::people::address_book`) as a seed +# source for the people store. Gated on the target *and* this feature: the four +# objc2 crates it needs are macOS-only and are not worth compiling for a host +# that never seeds from Contacts. Off (or non-macOS) leaves a stub that returns +# an empty contact list, so a refresh seeds nothing rather than failing. +contacts = ["people", "dep:objc2", "dep:objc2-foundation", "dep:objc2-contacts", "dep:block2"] + [dependencies] anyhow = "1" log = "0.4" @@ -144,6 +160,28 @@ tokio = { version = "1", features = [ "net", ], optional = true } +# macOS system address book, read by `memory::people::address_book` behind the +# `contacts` feature. Declared under a `cfg(target_os = "macos")` target table +# so these never enter a Linux or Windows dependency graph at all — the feature +# is a no-op there and the module falls back to its empty-contacts stub. +[target.'cfg(target_os = "macos")'.dependencies] +objc2 = { version = "0.6", optional = true } +objc2-foundation = { version = "0.3", features = [ + "NSArray", + "NSError", + "NSObject", + "NSString", + "NSPredicate", +], optional = true } +objc2-contacts = { version = "0.3.2", features = [ + "CNContact", + "CNContactFetchRequest", + "CNContactStore", + "CNLabeledValue", + "CNPhoneNumber", +], optional = true } +block2 = { version = "0.6", optional = true } + [dev-dependencies] dotenvy = "0.15" tempfile = "3" diff --git a/src/memory/mod.rs b/src/memory/mod.rs index a07a56e..503316f 100644 --- a/src/memory/mod.rs +++ b/src/memory/mod.rs @@ -107,6 +107,19 @@ pub mod providers; /// the git-history reader additionally requires `git-diff`. #[cfg(feature = "persona")] pub mod persona; + +/// Contact resolution and scoring: a SQLite store of people, handle aliases and +/// interactions, a deterministic (handle | email | display name) → `PersonId` +/// resolver, and a recency × frequency × reciprocity × depth scorer. +/// +/// Gated behind the default-off `people` feature, which implies `tokio` — the +/// store shares its connection across tasks. The macOS address-book seed +/// source additionally requires `contacts`. +/// +/// This is storage, so it belongs to the engine rather than to the memory +/// contract: an engine bound in TinyCortex's place brings its own. +#[cfg(feature = "people")] +pub mod people; // ── Re-exports ────────────────────────────────────────────────────────────── pub use config::{MemoryConfig, WeightProfile}; pub use error::{MemoryEngineResult, MemoryError as MemoryEngineError}; diff --git a/src/memory/people/README.md b/src/memory/people/README.md new file mode 100644 index 0000000..e9142b2 --- /dev/null +++ b/src/memory/people/README.md @@ -0,0 +1,85 @@ +# people + +Contact resolution + relationship scoring (the "A5" module). Maps any of three handle kinds — iMessage handle, email, or display name — to a single stable `PersonId`, and ranks known people by a deterministic composite score (recency × frequency × reciprocity × depth) derived from observed interaction rows. Backed by its own SQLite database (people / handle aliases / interactions). Can seed itself from the macOS system Address Book (`CNContactStore`). Intentionally self-contained — per its module docstring it has no dependency on `life_capture`, `chronicle`, `nudges`, or UI; downstream integration is left to later slices. + +## Responsibilities + +- Canonicalize handles (lowercase/trim emails + email-style iMessage handles; whitespace-collapse display names) so the same person resolves consistently across case and spacing. +- Deterministically resolve a `Handle` to an existing `PersonId`, or mint a new `Person` skeleton on first sight (`create_if_missing`). +- Link handles together (`link`) so an email + phone + display name can be attached to one person — without ever *auto*-merging distinct identities that share only a display name or an unverified handle. +- Record interactions and aggregate them into a per-person composite score plus an explainable component breakdown. +- Rank all known people by score for `people.list`. +- Seed the store from the macOS Address Book, distinguishing "permission denied" from "no contacts". +- Persist people, handle aliases, and interactions in a dedicated SQLite DB with idempotent migrations. + +## Key files + +| File | Role | +| --- | --- | +| `src/openhuman/memory/people/mod.rs` | Export-focused. Declares submodules and re-exports `all_people_controller_schemas` / `all_people_registered_controllers`. | +| `src/openhuman/memory/people/types.rs` | Domain types: `PersonId`, `Handle` (with `canonicalize` / `as_key`), `Person`, `Interaction`, `ScoreComponents`, `AddressBookContact`. | +| `src/openhuman/memory/people/resolver.rs` | `HandleResolver` — `resolve`, `resolve_or_create(_with_status)`, `link`, `seed_from_address_book`. The deterministic handle→PersonId logic + cross-source merge-safety contract. | +| `src/openhuman/memory/people/scorer.rs` | Pure `score(interactions, now) -> ScoreComponents`. Recency half-life, frequency window/cap, reciprocity balance, depth cap as module constants. | +| `src/openhuman/memory/people/store.rs` | SQLite-backed `PeopleStore` (`Arc>`) + rebindable process-global accessor (`init_from_workspace` / `get`). CRUD, lookup, interaction read/write, batched interaction fetch. | +| `src/openhuman/memory/people/address_book.rs` | `ContactsSource` trait + `SystemContactsSource` (macOS `CNContactStore` FFI via objc2) and non-mac stub; `MockContactsSource` for tests; `AddressBookError`. | +| `src/openhuman/memory/people/rpc.rs` | Domain RPC handlers (`handle_list`, `handle_resolve`, `handle_score`, `handle_refresh_address_book`) returning `RpcOutcome`; callable directly in tests with a constructed `PeopleStore`. | +| `src/openhuman/memory/people/schemas.rs` | Controller schemas + param-parsing adapter handlers that fetch the global store and delegate to `rpc.rs`. | +| `src/openhuman/memory/people/migrations.rs` | Idempotent migration runner (bookkeeping table `_people_migrations`, per-migration transaction). | +| `src/openhuman/memory/people/migrations/0001_init.sql` | Schema: `people`, `handle_aliases`, `interactions` + indexes. | +| `src/openhuman/memory/people/tests.rs` | Cross-file integration tests for the domain. | + +## Public surface + +- Types: `PersonId`, `Handle` (`IMessage` / `Email` / `DisplayName`), `Person`, `Interaction`, `ScoreComponents`, `AddressBookContact`. +- `HandleResolver::{resolve, resolve_or_create, resolve_or_create_with_status, link, seed_from_address_book}`. +- `scorer::score` + tunable constants `RECENCY_HALF_LIFE_DAYS`, `FREQUENCY_WINDOW_DAYS`, `FREQUENCY_CAP`, `DEPTH_CAP_CHARS`. +- `store::{PeopleStore, init, get}` and `ConnHandle`. +- `address_book::{ContactsSource, SystemContactsSource, read, read_with, AddressBookError}`. +- `mod.rs` re-exports `all_people_controller_schemas` / `all_people_registered_controllers` for the controller registry. + +## RPC / controllers + +Registered via the controller registry (wired in `src/core/all.rs`). Four controllers in the `people` namespace: + +| Method | Inputs | Output | +| --- | --- | --- | +| `people.list` | `limit?` (default 100, capped at 500) | `people[]` ranked by score desc — each with `person_id`, `display_name?`, `primary_email?`, `primary_phone?`, `handles[]`, `score`, `components`, `interaction_count`. | +| `people.resolve` | `kind` (`imessage`/`email`/`display_name`), `value`, `create_if_missing?` | `person_id?` (null when unknown and not creating), `created`. | +| `people.score` | `person_id` (UUID) | `person_id`, `score`, `components`, `interaction_count`. Errors if person not found. | +| `people.refresh_address_book` | — | `seeded`, `skipped`, `permission_denied`. | + +`score` / composite is `recency * frequency * reciprocity * depth`, each clamped to `[0,1]`. + +## Persistence + +Dedicated SQLite DB managed by `PeopleStore` (open via `open_at(path)` or `open_in_memory()`; migrations run on open). Three tables (see `0001_init.sql`): + +- `people` — one row per resolved person (uuid id, display name, primary email/phone, timestamps). +- `handle_aliases` — `(kind, value)` primary key → `person_id` (FK, `ON DELETE CASCADE`); `value` is the canonicalized form. This table *is* the resolver index. +- `interactions` — `(person_id, ts, is_outbound, length)` rows the scorer aggregates; indexed by `(person_id, ts DESC)` and `ts DESC`. + +Migrations are tracked in `_people_migrations` and applied idempotently in a transaction. The store is exposed process-globally through a workspace-tagged `RwLock>` slot (`get` from controller handlers). `store::init_from_workspace(workspace_dir)` seeds it, opening `/people/people.db`; it is called at core boot (`src/core/jsonrpc.rs`, alongside `memory::global` and `whatsapp_data::global`) and again on active-user switch (`credentials::ops`, `app_state::ops`), where a **different** workspace rebinds the store — mirroring `memory::global` so people never keeps writing the pre-login workspace. Same-workspace calls are a no-op. Tests construct stores directly with `open_in_memory`. + +## Dependencies + +- the host's `core::all::{ControllerFuture, RegisteredController}` — controller registry types, used by the RPC surface that stayed in the host. +- the host's `core::{ControllerSchema, FieldSchema, TypeSchema}` — controller schema definitions, used by the RPC surface that stayed in the host. +- `crate::rpc::RpcOutcome` — standard RPC result envelope (`RpcOutcome`). +- External crates: `rusqlite` (storage), `tokio` (async + `spawn_blocking` for sync SQL, `Mutex`), `chrono` (timestamps/scoring), `uuid` (`PersonId`), `serde`; on macOS, `block2` / `objc2` / `objc2-contacts` / `objc2-foundation` for the `CNContactStore` FFI in `address_book.rs`. The global store slot uses `std::sync::{OnceLock, RwLock}`. + +Notably it depends on **no other `openhuman` domain** — consistent with its "self-contained" docstring. + +## Used by + +- `src/core/all.rs` — registers the people controllers and schemas, and routes the `"people"` namespace. +- `src/openhuman/memory/store/` — reuses `people::types::{Person, PersonId, Handle}` (e.g. `Person` aliased as `Contact` in `kinds.rs`, and in `traits.rs`). + +## Notes / gotchas + +- **Cross-source merge safety (issue #1538):** two identities that share only a display name or only an unverified handle from different sources are **never** auto-merged. Merging only happens via explicit `link()`. Resolver tests lock this contract in. +- **Idempotent seeding:** `seed_from_address_book` re-runs as a no-op for already-known handles; on `PermissionDenied` it writes nothing (no partial state). The "primary" link target is first email, else first phone, else display name. +- **macOS Address Book FFI must not run on the main thread** — `CNContactStore` access requests deadlock there; `request_access` blocks on a completion-handler channel. Non-mac builds return an empty contact list. +- **Scoring constants are module-level**, not config-driven yet — kept fixed so tests stay stable; the docstring notes they can move to config later without breaking the API. +- **Composite is a product:** any zero component (e.g. a one-sided conversation → reciprocity 0) zeroes the whole score. +- **SQL runs on `spawn_blocking`:** the connection is sync `rusqlite` behind `Arc>`; `JoinError`s from blocking tasks are mapped into a synthetic `rusqlite` IO error. +- **Tests bypass the global store:** they construct `PeopleStore::open_in_memory()` and call `rpc::*` / `HandleResolver` directly rather than going through the schema adapters (which require the workspace-seeded global). diff --git a/src/memory/people/address_book.rs b/src/memory/people/address_book.rs new file mode 100644 index 0000000..b02b286 --- /dev/null +++ b/src/memory/people/address_book.rs @@ -0,0 +1,382 @@ +//! macOS Address Book read via `CNContactStore`. +//! +//! Uses the documented Contacts framework API (`CNContactStore`) which: +//! - Triggers the TCC Contacts permission prompt (sandboxed builds work correctly). +//! - Returns a structured error for "permission denied" so callers can distinguish +//! that case from "no contacts". +//! +//! A trait (`ContactsSource`) provides a mockable seam so unit tests can inject a +//! canned list or a permission-denied error without any FFI calls. +//! +//! On non-mac platforms `read()` returns an empty vec (stub path). + +use crate::memory::people::types::AddressBookContact; + +/// Result type distinguishing permission errors from other failures. +#[derive(Debug, PartialEq)] +pub enum AddressBookError { + /// The user denied or restricted Contacts access. + PermissionDenied, + /// Any other error (typically returned as a descriptive string). + Other(String), +} + +impl std::fmt::Display for AddressBookError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + AddressBookError::PermissionDenied => { + write!( + f, + "contacts access denied — grant access in System Settings > Privacy > Contacts" + ) + } + AddressBookError::Other(s) => write!(f, "{s}"), + } + } +} + +/// Mockable seam for contact fetching. The real impl calls CNContactStore; +/// tests inject a `MockContactsSource`. +pub trait ContactsSource: Send + Sync { + fn fetch_contacts(&self) -> Result, AddressBookError>; +} + +/// Real implementation backed by CNContactStore (macOS only). +/// On non-mac this is an empty struct whose `fetch_contacts` always returns `Ok(vec![])`. +pub struct SystemContactsSource; + +impl ContactsSource for SystemContactsSource { + fn fetch_contacts(&self) -> Result, AddressBookError> { + imp::fetch_via_cn_contact_store() + } +} + +/// Fetch all contacts using the provided `ContactsSource`. +/// +/// Errors are logged at `warn` level and surfaced to the caller so RPC +/// handlers can distinguish "permission denied" from "no contacts found". +pub fn read_with(source: &dyn ContactsSource) -> Result, AddressBookError> { + match source.fetch_contacts() { + Ok(v) => { + log::debug!("[people::address_book] fetched {} contacts", v.len()); + Ok(v) + } + Err(AddressBookError::PermissionDenied) => { + log::warn!( + "[people::address_book] contacts access denied — \ + grant access in System Settings > Privacy > Contacts" + ); + Err(AddressBookError::PermissionDenied) + } + Err(AddressBookError::Other(ref e)) => { + log::warn!("[people::address_book] fetch error: {e}"); + Err(AddressBookError::Other(e.clone())) + } + } +} + +/// Convenience wrapper using the real `SystemContactsSource`. +pub fn read() -> Result, AddressBookError> { + read_with(&SystemContactsSource) +} + +// ── macOS implementation ────────────────────────────────────────────────────── +// +// Gated on `contacts` as well as the target: the four objc2 crates this needs +// are exclusive to this module, so a slim macOS build sheds the whole cohort. + +#[cfg(all(target_os = "macos", feature = "contacts"))] +mod imp { + use super::{AddressBookContact, AddressBookError}; + + use block2::RcBlock; + use core::ptr::NonNull; + use objc2::runtime::Bool; + use objc2::runtime::ProtocolObject; + use objc2::AnyThread as _; + use objc2_contacts::{ + CNAuthorizationStatus, CNContact, CNContactFetchRequest, CNContactStore, CNEntityType, + }; + use objc2_foundation::{NSArray, NSError, NSString}; + use std::sync::{Arc, Mutex}; + + // CNKeyDescriptor is a protocol; NSString conforms to it. + // We build the keys array as NSArray>. + use objc2_contacts::CNKeyDescriptor; + + /// Build the keys array used for CNContactFetchRequest. + /// + /// # Safety + /// NSString::from_str is safe; casting to ProtocolObject is safe because + /// `NSString: CNKeyDescriptor` (confirmed by the objc2-contacts bindings). + unsafe fn make_keys_array() -> objc2::rc::Retained>> + { + let given = NSString::from_str("givenName"); + let family = NSString::from_str("familyName"); + let emails = NSString::from_str("emailAddresses"); + let phones = NSString::from_str("phoneNumbers"); + + // NSString conforms to CNKeyDescriptor, so we can cast the refs. + let refs: &[&ProtocolObject] = &[ + ProtocolObject::from_ref(&*given), + ProtocolObject::from_ref(&*family), + ProtocolObject::from_ref(&*emails), + ProtocolObject::from_ref(&*phones), + ]; + NSArray::from_slice(refs) + } + + /// Request contacts access from TCC. Blocks on the calling thread until + /// the completion handler fires. Must not be called from the main thread + /// on macOS (CNContactStore will deadlock). + fn request_access(store: &CNContactStore) -> Result<(), AddressBookError> { + unsafe { + let status = CNContactStore::authorizationStatusForEntityType(CNEntityType::Contacts); + match status { + CNAuthorizationStatus::Authorized | CNAuthorizationStatus::Limited => { + log::debug!("[people::address_book] contacts access already authorized"); + return Ok(()); + } + CNAuthorizationStatus::Denied | CNAuthorizationStatus::Restricted => { + return Err(AddressBookError::PermissionDenied); + } + _ => { + log::debug!( + "[people::address_book] requesting contacts access (status={status:?})" + ); + } + } + + let (tx, rx) = std::sync::mpsc::channel::>(); + let tx = Arc::new(Mutex::new(Some(tx))); + let tx_clone = Arc::clone(&tx); + + let block = RcBlock::new(move |granted: Bool, _error: *mut NSError| { + let mut slot = tx_clone.lock().unwrap(); + if let Some(sender) = slot.take() { + let result = if granted.as_bool() { + Ok(()) + } else { + Err(AddressBookError::PermissionDenied) + }; + let _ = sender.send(result); + } + }); + + store.requestAccessForEntityType_completionHandler(CNEntityType::Contacts, &block); + + rx.recv().map_err(|_| { + AddressBookError::Other("contacts permission callback never fired".into()) + })? + } + } + + pub fn fetch_via_cn_contact_store() -> Result, AddressBookError> { + log::debug!("[people::address_book] fetch_via_cn_contact_store entry"); + unsafe { + let store = CNContactStore::new(); + request_access(&store)?; + + let keys_array = make_keys_array(); + let request = CNContactFetchRequest::initWithKeysToFetch( + CNContactFetchRequest::alloc(), + &keys_array, + ); + + let mut contacts: Vec = Vec::new(); + + // We use a raw pointer to the vec inside the block so that we can + // push from within the block. The block runs synchronously within + // enumerateContactsWithFetchRequest (it blocks until done), so the + // pointer is valid throughout. + let contacts_ptr: *mut Vec = &mut contacts; + + let block = RcBlock::new( + move |contact_nn: NonNull, _stop: NonNull| { + let contact: &CNContact = contact_nn.as_ref(); + + let given = contact.givenName().to_string(); + let family = contact.familyName().to_string(); + let full = { + let g = given.trim(); + let f = family.trim(); + match (g.is_empty(), f.is_empty()) { + (true, true) => None, + (false, true) => Some(g.to_string()), + (true, false) => Some(f.to_string()), + (false, false) => Some(format!("{g} {f}")), + } + }; + + let emails: Vec = { + let arr = contact.emailAddresses(); + let mut v = Vec::new(); + for i in 0..arr.len() { + let lv = arr.objectAtIndex(i); + // CNLabeledValue.value() → Retained + let email = lv.value().to_string(); + let trimmed = email.trim().to_string(); + if !trimmed.is_empty() { + v.push(trimmed); + } + } + v + }; + + let phones: Vec = { + let arr = contact.phoneNumbers(); + let mut v = Vec::new(); + for i in 0..arr.len() { + let lv = arr.objectAtIndex(i); + // CNLabeledValue.value() → Retained + let num = lv.value().stringValue().to_string(); + let trimmed = num.trim().to_string(); + if !trimmed.is_empty() { + v.push(trimmed); + } + } + v + }; + + if full.is_none() && emails.is_empty() && phones.is_empty() { + return; + } + + (*contacts_ptr).push(AddressBookContact { + display_name: full, + emails, + phones, + }); + }, + ); + + let mut error: Option> = None; + let ok = store.enumerateContactsWithFetchRequest_error_usingBlock( + &request, + Some(&mut error), + &block, + ); + if !ok { + let msg = error + .map(|e| e.localizedDescription().to_string()) + .unwrap_or_else(|| "unknown error from CNContactStore".into()); + return Err(AddressBookError::Other(msg)); + } + + log::debug!( + "[people::address_book] enumerated {} contacts", + contacts.len() + ); + Ok(contacts) + } + } +} + +// ── stub: non-macOS, or macOS with `contacts` compiled out ─────────────────── +// +// Pre-dates the gate — it already existed for Linux/Windows. Widening its cfg +// is the whole off-state: `read()`, `read_with()`, `AddressBookError` and +// `SystemContactsSource` stay compiled everywhere, so the `people` RPC surface +// is identical and an address-book refresh seeds nothing rather than failing. + +#[cfg(not(all(target_os = "macos", feature = "contacts")))] +mod imp { + use super::{AddressBookContact, AddressBookError}; + + pub fn fetch_via_cn_contact_store() -> Result, AddressBookError> { + Ok(vec![]) + } +} + +// ── tests ───────────────────────────────────────────────────────────────────── + +#[cfg(test)] +pub mod tests { + use super::*; + + /// Test double that returns a canned list without any FFI calls. + pub struct MockContactsSource { + pub result: Result, AddressBookError>, + } + + impl MockContactsSource { + pub fn ok(contacts: Vec) -> Self { + Self { + result: Ok(contacts), + } + } + + pub fn permission_denied() -> Self { + Self { + result: Err(AddressBookError::PermissionDenied), + } + } + } + + impl ContactsSource for MockContactsSource { + fn fetch_contacts(&self) -> Result, AddressBookError> { + match &self.result { + Ok(v) => Ok(v.clone()), + Err(AddressBookError::PermissionDenied) => Err(AddressBookError::PermissionDenied), + Err(AddressBookError::Other(s)) => Err(AddressBookError::Other(s.clone())), + } + } + } + + fn mk_contact(name: &str, email: &str) -> AddressBookContact { + AddressBookContact { + display_name: Some(name.into()), + emails: vec![email.into()], + phones: vec![], + } + } + + #[test] + fn mock_source_returns_canned_contacts() { + let source = MockContactsSource::ok(vec![ + mk_contact("Alice", "alice@example.com"), + mk_contact("Bob", "bob@example.com"), + ]); + let result = read_with(&source).unwrap(); + assert_eq!(result.len(), 2); + assert_eq!(result[0].display_name.as_deref(), Some("Alice")); + assert_eq!(result[1].emails[0], "bob@example.com"); + } + + #[test] + fn mock_source_permission_denied_is_distinguished() { + let source = MockContactsSource::permission_denied(); + let err = read_with(&source).unwrap_err(); + assert_eq!(err, AddressBookError::PermissionDenied); + } + + #[test] + fn system_source_non_mac_returns_empty() { + // Mirrors the `imp` cfgs above: the stub is what compiles whenever the + // real CNContactStore path is absent, whether by target or by gate. + #[cfg(not(all(target_os = "macos", feature = "contacts")))] + { + let source = SystemContactsSource; + let result = read_with(&source).unwrap(); + assert!(result.is_empty()); + } + #[cfg(all(target_os = "macos", feature = "contacts"))] + { + // TCC state is environment-dependent; just verify no panic. + let source = SystemContactsSource; + let _ = read_with(&source); + } + } + + #[test] + fn contact_with_no_fields_is_excluded_by_mock() { + let source = MockContactsSource::ok(vec![AddressBookContact { + display_name: Some("Sarah Lee".into()), + emails: vec![], + phones: vec!["+1 555 000 0001".into()], + }]); + let result = read_with(&source).unwrap(); + assert_eq!(result.len(), 1); + assert_eq!(result[0].phones[0], "+1 555 000 0001"); + } +} diff --git a/src/memory/people/migrations.rs b/src/memory/people/migrations.rs new file mode 100644 index 0000000..57d153e --- /dev/null +++ b/src/memory/people/migrations.rs @@ -0,0 +1,93 @@ +//! SQLite migrations for the people module. Mirrors the life_capture +//! migration style: idempotent, per-migration transaction, recorded in a +//! dedicated bookkeeping table. + +use rusqlite::{Connection, Result}; + +const MIGRATIONS: &[(&str, &str)] = &[("0001_init", include_str!("migrations/0001_init.sql"))]; + +pub fn run(conn: &Connection) -> Result<()> { + conn.execute_batch( + "CREATE TABLE IF NOT EXISTS _people_migrations ( + name TEXT PRIMARY KEY, + applied_at INTEGER NOT NULL + )", + )?; + + for (name, sql) in MIGRATIONS { + let already: bool = conn.query_row( + "SELECT EXISTS(SELECT 1 FROM _people_migrations WHERE name = ?1)", + rusqlite::params![name], + |row| row.get(0), + )?; + if already { + continue; + } + + conn.execute_batch("BEGIN")?; + let result = (|| -> Result<()> { + conn.execute_batch(sql)?; + conn.execute( + "INSERT INTO _people_migrations(name, applied_at) \ + VALUES (?1, CAST(strftime('%s','now') AS INTEGER))", + rusqlite::params![name], + )?; + Ok(()) + })(); + match result { + Ok(()) => conn.execute_batch("COMMIT")?, + Err(e) => { + let _ = conn.execute_batch("ROLLBACK"); + return Err(e); + } + } + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn fresh() -> Connection { + Connection::open_in_memory().unwrap() + } + + #[test] + fn migrations_create_expected_tables() { + let conn = fresh(); + run(&conn).unwrap(); + let mut stmt = conn + .prepare("SELECT name FROM sqlite_master WHERE type='table' ORDER BY name") + .unwrap(); + let names: Vec = stmt + .query_map([], |row| row.get(0)) + .unwrap() + .map(|r| r.unwrap()) + .collect(); + for expected in [ + "people", + "handle_aliases", + "interactions", + "_people_migrations", + ] { + assert!( + names.iter().any(|n| n == expected), + "missing {expected}: {names:?}" + ); + } + } + + #[test] + fn migrations_are_idempotent() { + let conn = fresh(); + run(&conn).unwrap(); + run(&conn).unwrap(); + let count: i64 = conn + .query_row("SELECT count(*) FROM _people_migrations", [], |row| { + row.get(0) + }) + .unwrap(); + assert_eq!(count, MIGRATIONS.len() as i64); + } +} diff --git a/src/memory/people/migrations/0001_init.sql b/src/memory/people/migrations/0001_init.sql new file mode 100644 index 0000000..ee692b9 --- /dev/null +++ b/src/memory/people/migrations/0001_init.sql @@ -0,0 +1,37 @@ +-- People module schema. +-- +-- `people` holds one row per resolved person. `handle_aliases` holds all +-- known (kind, canonical_value) handles that map to that person; the +-- resolver is a lookup on `(kind, value)` → `person_id`. +-- +-- `interactions` records observed exchanges for scoring. Single-user v1; +-- each row is attributed to (local-user, person_id). + +CREATE TABLE IF NOT EXISTS people ( + id TEXT PRIMARY KEY, -- uuid + display_name TEXT, + primary_email TEXT, + primary_phone TEXT, + created_at INTEGER NOT NULL, -- unix seconds + updated_at INTEGER NOT NULL +); + +CREATE TABLE IF NOT EXISTS handle_aliases ( + kind TEXT NOT NULL, -- 'imessage' | 'email' | 'display_name' + value TEXT NOT NULL, -- canonicalized (lowercase / trimmed) + person_id TEXT NOT NULL REFERENCES people(id) ON DELETE CASCADE, + created_at INTEGER NOT NULL, + PRIMARY KEY (kind, value) +); + +CREATE INDEX IF NOT EXISTS handle_aliases_person_idx ON handle_aliases(person_id); + +CREATE TABLE IF NOT EXISTS interactions ( + person_id TEXT NOT NULL REFERENCES people(id) ON DELETE CASCADE, + ts INTEGER NOT NULL, -- unix seconds + is_outbound INTEGER NOT NULL, -- 1 = user sent, 0 = received + length INTEGER NOT NULL DEFAULT 0 +); + +CREATE INDEX IF NOT EXISTS interactions_person_idx ON interactions(person_id, ts DESC); +CREATE INDEX IF NOT EXISTS interactions_ts_idx ON interactions(ts DESC); diff --git a/src/memory/people/mod.rs b/src/memory/people/mod.rs new file mode 100644 index 0000000..a0ee412 --- /dev/null +++ b/src/memory/people/mod.rs @@ -0,0 +1,18 @@ +//! People: contact resolution + scoring. +//! +//! A5 module. Deterministic resolver maps (imessage handle | email | display +//! name) to a stable `PersonId`. Scoring blends recency × frequency × +//! reciprocity × depth from interaction rows into a ranked `people.list`. +//! +//! Intentionally self-contained: no dependency on `life_capture`, +//! `chronicle`, `nudges`, or UI. Integration happens in later slices. + +pub mod address_book; +pub mod migrations; +pub mod resolver; +pub mod scorer; +pub mod store; +pub mod types; + +#[cfg(test)] +mod tests; diff --git a/src/memory/people/resolver.rs b/src/memory/people/resolver.rs new file mode 100644 index 0000000..4428345 --- /dev/null +++ b/src/memory/people/resolver.rs @@ -0,0 +1,527 @@ +//! HandleResolver — deterministic mapping (Handle) → PersonId. +//! +//! Given the same store contents, resolving the same handle twice returns +//! the same `PersonId`. If the handle is unknown and `create_if_missing` +//! is set, the resolver mints a new `PersonId`, inserts a `Person` skeleton +//! with the handle attached, and returns the new id. +//! +//! `seed_from_address_book` wires the `address_book` read path into the +//! resolver so that contacts from the system address book are pre-populated +//! as `Person` rows (and their handles are registered for future resolution). + +use chrono::Utc; + +use crate::memory::people::address_book::{self, AddressBookError, ContactsSource}; +use crate::memory::people::store::PeopleStore; +use crate::memory::people::types::{Handle, Person, PersonId}; + +pub struct HandleResolver<'a> { + store: &'a PeopleStore, +} + +impl<'a> HandleResolver<'a> { + pub fn new(store: &'a PeopleStore) -> Self { + Self { store } + } + + /// Look up the person for a handle. Returns `None` if unknown. + pub async fn resolve(&self, handle: &Handle) -> Result, String> { + let canonical = handle.canonicalize(); + self.store + .lookup(&canonical) + .await + .map_err(|e| format!("lookup: {e}")) + } + + /// Look up or mint. Display-name / email fields on the newly-minted + /// `Person` are populated from the handle itself so the UI has + /// something to render before any enrichment runs. + pub async fn resolve_or_create(&self, handle: &Handle) -> Result { + self.resolve_or_create_with_status(handle) + .await + .map(|(id, _created)| id) + } + + pub async fn resolve_or_create_with_status( + &self, + handle: &Handle, + ) -> Result<(PersonId, bool), String> { + let canonical = handle.canonicalize(); + let id = PersonId::new(); + let (display_name, primary_email, primary_phone) = match &canonical { + Handle::DisplayName(s) => (Some(s.clone()), None, None), + Handle::Email(s) => (None, Some(s.clone()), None), + Handle::IMessage(s) => { + if s.contains('@') { + (None, Some(s.clone()), None) + } else { + (None, None, Some(s.clone())) + } + } + }; + let now = Utc::now(); + let person = Person { + id, + display_name, + primary_email, + primary_phone, + handles: vec![canonical.clone()], + created_at: now, + updated_at: now, + }; + self.store + .resolve_or_insert_person(&person, &canonical) + .await + .map_err(|e| format!("resolve_or_insert_person: {e}")) + } + + /// Merge: attach `other` as an alias on the person `primary` resolves to. + /// Useful for the sync path that learns "this email and this phone + /// belong to the same contact". + pub async fn link(&self, primary: &Handle, other: Handle) -> Result { + let pid = self.resolve_or_create(primary).await?; + let other = other.canonicalize(); + self.store + .add_alias(pid, other) + .await + .map_err(|e| format!("add_alias: {e}"))?; + Ok(pid) + } + + /// Seed the people store from the system address book. + /// + /// For each contact returned by `source`: + /// - Pick the first email or phone as the "primary" handle and look it + /// up or mint a `PersonId`. + /// - Link any additional emails / phones as aliases on the same person. + /// - If only a display name is present, mint via display name. + /// + /// Contacts that produce no handles at all are skipped. This is + /// idempotent: re-running on the same contact list is a no-op because + ///`lookup` finds existing handle rows. + /// + /// Returns `(seeded, skipped)` counts, and propagates `AddressBookError` + /// to let callers distinguish permission-denied from other failures. + pub async fn seed_from_address_book( + &self, + source: &dyn ContactsSource, + ) -> Result<(usize, usize), AddressBookError> { + let contacts = address_book::read_with(source)?; + let mut seeded = 0usize; + let mut skipped = 0usize; + + for c in contacts { + // Build a flat list of all handles for this contact. + let mut handles: Vec = Vec::new(); + for email in &c.emails { + let trimmed = email.trim(); + if !trimmed.is_empty() { + handles.push(Handle::Email(trimmed.to_string())); + } + } + for phone in &c.phones { + let trimmed = phone.trim(); + if !trimmed.is_empty() { + handles.push(Handle::IMessage(trimmed.to_string())); + } + } + if let Some(ref name) = c.display_name { + let trimmed = name.trim(); + if !trimmed.is_empty() { + handles.push(Handle::DisplayName(trimmed.to_string())); + } + } + + if handles.is_empty() { + skipped += 1; + continue; + } + + // The "primary" handle is the first email if present, otherwise + // the first phone, otherwise the display name. This gives the + // most stable link target for future interactions. + let primary = handles[0].clone(); + + // mint or look up the primary handle + match self.resolve_or_create(&primary).await { + Err(e) => { + log::warn!( + "[people::resolver] seed_from_address_book: failed to upsert primary handle {:?}: {e}", + primary.as_key() + ); + skipped += 1; + continue; + } + Ok(pid) => { + // link all additional handles as aliases + for alias in handles.into_iter().skip(1) { + if let Err(e) = self.store.add_alias(pid, alias.canonicalize()).await { + log::warn!( + "[people::resolver] seed_from_address_book: add_alias failed: {e}" + ); + } + } + seeded += 1; + } + } + } + + log::debug!( + "[people::resolver] seed_from_address_book done: seeded={seeded} skipped={skipped}" + ); + Ok((seeded, skipped)) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::memory::people::address_book::tests::MockContactsSource; + use crate::memory::people::types::AddressBookContact; + + #[tokio::test] + async fn resolve_returns_none_for_unknown_handle() { + let s = PeopleStore::open_in_memory().unwrap(); + let r = HandleResolver::new(&s); + let got = r.resolve(&Handle::Email("x@y.z".into())).await.unwrap(); + assert!(got.is_none()); + } + + #[tokio::test] + async fn resolve_or_create_is_deterministic_across_case_and_whitespace() { + let s = PeopleStore::open_in_memory().unwrap(); + let r = HandleResolver::new(&s); + let a = r + .resolve_or_create(&Handle::Email("Sarah@Example.COM".into())) + .await + .unwrap(); + let b = r + .resolve_or_create(&Handle::Email(" sarah@example.com ".into())) + .await + .unwrap(); + assert_eq!(a, b, "canonicalization must collapse case+whitespace"); + } + + #[tokio::test] + async fn concurrent_resolve_or_create_returns_one_database_id() { + let s = PeopleStore::open_in_memory().unwrap(); + let r = HandleResolver::new(&s); + let handles: Vec<_> = (0..16) + .map(|_| Handle::Email("Race@Example.COM".into())) + .collect(); + + let ids = futures::future::join_all(handles.iter().map(|h| r.resolve_or_create(h))).await; + let first = ids[0].as_ref().unwrap(); + for id in &ids { + assert_eq!(id.as_ref().unwrap(), first); + } + + let people = s.list().await.unwrap(); + assert_eq!(people.len(), 1); + assert_eq!(people[0].id, *first); + } + + #[tokio::test] + async fn same_email_different_display_name_resolve_same_id() { + let s = PeopleStore::open_in_memory().unwrap(); + let r = HandleResolver::new(&s); + let via_email = r + .resolve_or_create(&Handle::Email("a@b.c".into())) + .await + .unwrap(); + // Linking a display name to the same email must not mint a second id. + let via_linked = r + .link( + &Handle::Email("a@b.c".into()), + Handle::DisplayName("Alice".into()), + ) + .await + .unwrap(); + assert_eq!(via_email, via_linked); + // And now resolving the display name returns the same id. + let via_name = r + .resolve(&Handle::DisplayName("Alice".into())) + .await + .unwrap(); + assert_eq!(via_name, Some(via_email)); + } + + #[tokio::test] + async fn distinct_handles_without_linking_produce_distinct_ids() { + let s = PeopleStore::open_in_memory().unwrap(); + let r = HandleResolver::new(&s); + let a = r + .resolve_or_create(&Handle::Email("a@b.c".into())) + .await + .unwrap(); + let b = r + .resolve_or_create(&Handle::Email("x@y.z".into())) + .await + .unwrap(); + assert_ne!(a, b); + } + + #[tokio::test] + async fn seed_from_address_book_populates_store() { + let s = PeopleStore::open_in_memory().unwrap(); + let r = HandleResolver::new(&s); + + let source = MockContactsSource::ok(vec![ + AddressBookContact { + display_name: Some("Alice Smith".into()), + emails: vec!["alice@example.com".into()], + phones: vec!["+1 555 000 0001".into()], + }, + AddressBookContact { + display_name: Some("Bob Jones".into()), + emails: vec!["bob@example.com".into()], + phones: vec![], + }, + ]); + + let (seeded, skipped) = r.seed_from_address_book(&source).await.unwrap(); + assert_eq!(seeded, 2, "both contacts should be seeded"); + assert_eq!(skipped, 0); + + // Alice is resolvable by email + let alice_id = r + .resolve(&Handle::Email("alice@example.com".into())) + .await + .unwrap(); + assert!(alice_id.is_some(), "alice must be resolvable after seed"); + + // Alice is also resolvable by phone (linked as alias) + let alice_via_phone = r + .resolve(&Handle::IMessage("+1 555 000 0001".into())) + .await + .unwrap(); + assert_eq!( + alice_id, alice_via_phone, + "email and phone must resolve to same person" + ); + + // Bob is resolvable + let bob_id = r + .resolve(&Handle::Email("bob@example.com".into())) + .await + .unwrap(); + assert!(bob_id.is_some()); + assert_ne!(alice_id, bob_id, "distinct contacts must have distinct ids"); + } + + #[tokio::test] + async fn seed_from_address_book_permission_denied_is_propagated() { + let s = PeopleStore::open_in_memory().unwrap(); + let r = HandleResolver::new(&s); + + let source = MockContactsSource::permission_denied(); + let err = r.seed_from_address_book(&source).await.unwrap_err(); + assert_eq!(err, AddressBookError::PermissionDenied); + + // Store must still be empty — no partial writes. + let people = s.list().await.unwrap(); + assert!( + people.is_empty(), + "no people should be inserted on permission denied" + ); + } + + #[tokio::test] + async fn seed_is_idempotent() { + let s = PeopleStore::open_in_memory().unwrap(); + let r = HandleResolver::new(&s); + + let source = MockContactsSource::ok(vec![AddressBookContact { + display_name: Some("Carol".into()), + emails: vec!["carol@example.com".into()], + phones: vec![], + }]); + + let (s1, _) = r.seed_from_address_book(&source).await.unwrap(); + let (s2, _) = r.seed_from_address_book(&source).await.unwrap(); + assert_eq!(s1, 1); + assert_eq!(s2, 1, "second seed call should still report 1 (upsert)"); + + // Only one person in store. + let people = s.list().await.unwrap(); + assert_eq!(people.len(), 1, "idempotent — must not duplicate"); + } + + #[tokio::test] + async fn contact_with_only_display_name_is_seeded() { + let s = PeopleStore::open_in_memory().unwrap(); + let r = HandleResolver::new(&s); + + let source = MockContactsSource::ok(vec![AddressBookContact { + display_name: Some("No Email Person".into()), + emails: vec![], + phones: vec![], + }]); + let (seeded, skipped) = r.seed_from_address_book(&source).await.unwrap(); + assert_eq!(seeded, 1); + assert_eq!(skipped, 0); + } + + #[tokio::test] + async fn contact_with_no_fields_is_skipped() { + let s = PeopleStore::open_in_memory().unwrap(); + let r = HandleResolver::new(&s); + + let source = MockContactsSource::ok(vec![AddressBookContact { + display_name: None, + emails: vec![], + phones: vec![], + }]); + let (seeded, skipped) = r.seed_from_address_book(&source).await.unwrap(); + assert_eq!(seeded, 0); + assert_eq!(skipped, 1); + } + + // ── Cross-source merge safety tests (issue#1538) ────────────────────────── + // + // The people resolver must NOT silently merge two distinct identities that + // happen to share only a display name or only an unverified handle from + // different sources. These tests lock in the "ambiguous cross-source" + // contract: two handles from unrelated sources remain distinct unless + // explicitly linked via `link()`. + + /// Two contacts that share only a display name (no email or phone overlap) + /// must NOT be merged — they may be homonymous individuals. + #[tokio::test] + async fn same_display_name_from_different_sources_does_not_merge() { + let s = PeopleStore::open_in_memory().unwrap(); + let r = HandleResolver::new(&s); + + // Source A — email-backed identity + let id_a = r + .resolve_or_create(&Handle::Email("alice@company-a.com".into())) + .await + .unwrap(); + r.link( + &Handle::Email("alice@company-a.com".into()), + Handle::DisplayName("Alice Smith".into()), + ) + .await + .unwrap(); + + // Source B — different email; the same display name surfaces again, + // but as a *separate* DisplayName-backed mint (NOT linked to either + // email). This is the actual collision scenario: two ingestion paths + // both encounter "Alice Smith" without any cross-source identifier. + let id_b = r + .resolve_or_create(&Handle::Email("alice@company-b.com".into())) + .await + .unwrap(); + // The display-name resolver must already pin to id_a (linked above), + // so a second mint of the same DisplayName does NOT spawn a third + // identity — but crucially it also does NOT silently merge id_b into id_a. + let id_name_again = r + .resolve_or_create(&Handle::DisplayName("Alice Smith".into())) + .await + .unwrap(); + + // The two email-backed identities must be distinct. + assert_ne!( + id_a, id_b, + "two email handles with identical display names must not be merged without explicit link" + ); + + // The repeated DisplayName mint resolves to the linked identity (id_a), + // NOT to id_b. If display names auto-merged, id_b would have collapsed + // into id_a; if they minted fresh on every call, this would be a third id. + assert_eq!( + id_name_again, id_a, + "repeated DisplayName mint should resolve to the existing linked identity" + ); + assert_ne!( + id_name_again, id_b, + "DisplayName collision must not silently merge id_b into id_a" + ); + + // Resolving the display name returns the ONE identity that was explicitly linked. + let via_name = r + .resolve(&Handle::DisplayName("Alice Smith".into())) + .await + .unwrap(); + assert_eq!( + via_name, + Some(id_a), + "display name resolves to the explicitly linked identity" + ); + + // company-b Alice is still addressable by email only. + let via_b_email = r + .resolve(&Handle::Email("alice@company-b.com".into())) + .await + .unwrap(); + assert_eq!(via_b_email, Some(id_b)); + } + + /// Minting the same email handle from two logically distinct call sites + /// must always collapse to one `PersonId` (idempotent mint). This is the + /// safe side of cross-source: we never mint duplicates for an identical + /// canonical handle. + #[tokio::test] + async fn same_email_from_two_sources_collapses_to_one_person() { + let s = PeopleStore::open_in_memory().unwrap(); + let r = HandleResolver::new(&s); + + // Simulate two different ingestion paths (gmail vs slack) that both + // surface the same email address. + let from_gmail = r + .resolve_or_create(&Handle::Email("shared@example.com".into())) + .await + .unwrap(); + let from_slack = r + .resolve_or_create(&Handle::Email("shared@example.com".into())) + .await + .unwrap(); + + assert_eq!( + from_gmail, from_slack, + "identical canonical email from two ingestion paths must resolve to one PersonId" + ); + + // Exactly one person in the store. + let people = s.list().await.unwrap(); + assert_eq!( + people.len(), + 1, + "no duplicate person rows must exist for the same canonical email" + ); + } + + /// An iMessage phone handle from one source and an email from a different + /// source for the SAME real person must stay distinct until explicitly linked. + /// Memory must not unsafely merge the same person's identities across sources + /// (issue#1538). + #[tokio::test] + async fn phone_and_email_from_different_sources_are_not_merged_without_link() { + let s = PeopleStore::open_in_memory().unwrap(); + let r = HandleResolver::new(&s); + + // iMessage source sees only a phone. + let id_phone = r + .resolve_or_create(&Handle::IMessage("+15550001234".into())) + .await + .unwrap(); + + // Gmail source sees only an email. + let id_email = r + .resolve_or_create(&Handle::Email("sam@example.com".into())) + .await + .unwrap(); + + // Without an explicit link these are separate identities. This is the + // contract under test — cross-source handles for the same real person + // must NOT auto-merge. Asserting post-link merge semantics is out of + // scope: link()'s exact propagation rule (does the email handle + // afterwards canonically resolve to the phone PersonId, or remain + // independent with only the link table updated?) is a separate + // behavior tested in store_tests.rs. + assert_ne!( + id_phone, id_email, + "phone and email from unrelated sources must not be auto-merged" + ); + } +} diff --git a/src/memory/people/scorer.rs b/src/memory/people/scorer.rs new file mode 100644 index 0000000..e9a455e --- /dev/null +++ b/src/memory/people/scorer.rs @@ -0,0 +1,210 @@ +//! Scoring: recency × frequency × reciprocity × depth. +//! +//! Each component is deterministic given the same interaction list + `now` +//! timestamp, and each is clamped to `[0,1]`. The composite is the product; +//! clamping the product is redundant but kept for defense-in-depth. +//! +//! Weights (half-life / caps) are module constants so tests are stable. +//! They can move to config later without breaking the API. + +use chrono::{DateTime, Utc}; + +use crate::memory::people::types::{Interaction, ScoreComponents}; + +/// Recency half-life in days. An interaction this many days old contributes +/// 0.5 to the recency signal; older interactions decay exponentially. +pub const RECENCY_HALF_LIFE_DAYS: f32 = 14.0; + +/// Frequency is measured within this rolling window (days). Only interactions +/// more recent than `now - FREQUENCY_WINDOW_DAYS` count toward frequency. +pub const FREQUENCY_WINDOW_DAYS: u32 = 30; + +/// Frequency saturates at this many interactions inside `FREQUENCY_WINDOW_DAYS`. +/// 50+ qualifying interactions yields frequency = 1.0. +pub const FREQUENCY_CAP: f32 = 50.0; + +/// Depth saturates when the mean message length reaches this many chars. +pub const DEPTH_CAP_CHARS: f32 = 500.0; + +/// Compute component scores for a person given their interaction list. +/// `now` is passed in so tests can fix time. +pub fn score(interactions: &[Interaction], now: DateTime) -> ScoreComponents { + if interactions.is_empty() { + return ScoreComponents { + recency: 0.0, + frequency: 0.0, + reciprocity: 0.0, + depth: 0.0, + score: 0.0, + }; + } + + // Recency: highest-signal (= most recent) interaction drives the score. + let newest = interactions.iter().map(|i| i.ts).max().unwrap_or(now); + let age_days = ((now - newest).num_seconds() as f32 / 86_400.0).max(0.0); + let recency = (-(age_days * 2f32.ln() / RECENCY_HALF_LIFE_DAYS)) + .exp() + .clamp(0.0, 1.0); + + // Frequency: count within the rolling window, saturated at FREQUENCY_CAP. + // Using a window (rather than total-ever) prevents an old burst of + // messages from inflating the score of a now-silent contact. + let window_cutoff = now - chrono::Duration::days(FREQUENCY_WINDOW_DAYS as i64); + let window_count = interactions + .iter() + .filter(|i| i.ts >= window_cutoff) + .count() as f32; + let frequency = (window_count / FREQUENCY_CAP).clamp(0.0, 1.0); + + // Reciprocity: balance of outbound vs inbound — perfect balance = 1.0, + // all-one-direction = 0.0. Uses all interactions (not windowed) so that + // the long-term pattern is captured even when recent volume is low. + let (out_n, in_n) = interactions.iter().fold((0u32, 0u32), |(o, i), x| { + if x.is_outbound { + (o + 1, i) + } else { + (o, i + 1) + } + }); + let reciprocity = if out_n + in_n == 0 { + 0.0 + } else { + let o = out_n as f32; + let i = in_n as f32; + let min = o.min(i); + let max = o.max(i); + (min / max).clamp(0.0, 1.0) + }; + + // Depth: mean interaction length, saturated at DEPTH_CAP_CHARS. + let count = interactions.len() as f32; + let total_len: u64 = interactions.iter().map(|x| x.length as u64).sum(); + let mean_len = total_len as f32 / count.max(1.0); + let depth = (mean_len / DEPTH_CAP_CHARS).clamp(0.0, 1.0); + + let composite = (recency * frequency * reciprocity * depth).clamp(0.0, 1.0); + + ScoreComponents { + recency, + frequency, + reciprocity, + depth, + score: composite, + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::memory::people::types::PersonId; + use chrono::Duration; + + fn mk(ts: DateTime, outbound: bool, length: u32) -> Interaction { + Interaction { + person_id: PersonId::new(), + ts, + is_outbound: outbound, + length, + } + } + + #[test] + fn empty_interactions_score_zero() { + let s = score(&[], Utc::now()); + assert_eq!(s.score, 0.0); + assert_eq!(s.recency, 0.0); + assert_eq!(s.frequency, 0.0); + } + + #[test] + fn recency_half_life_matches_config() { + let now = Utc::now(); + let half_ago = now - Duration::days(RECENCY_HALF_LIFE_DAYS as i64); + let s = score(&[mk(half_ago, true, 100)], now); + // Half-life point → recency ≈ 0.5 (allow small float slack). + assert!((s.recency - 0.5).abs() < 0.05, "got {}", s.recency); + } + + #[test] + fn all_components_clamped_to_unit_interval() { + let now = Utc::now(); + let interactions: Vec = (0..200) + .map(|i| mk(now - Duration::hours(i), i % 2 == 0, 10_000)) + .collect(); + let s = score(&interactions, now); + for c in [s.recency, s.frequency, s.reciprocity, s.depth, s.score] { + assert!((0.0..=1.0).contains(&c), "component out of range: {c}"); + } + // 200 interactions all within a few days → window_count ≥ FREQUENCY_CAP + assert_eq!(s.frequency, 1.0); + assert_eq!(s.depth, 1.0); + } + + #[test] + fn one_sided_conversation_has_zero_reciprocity() { + let now = Utc::now(); + let v: Vec<_> = (0..5) + .map(|i| mk(now - Duration::hours(i), true, 100)) + .collect(); + let s = score(&v, now); + assert_eq!(s.reciprocity, 0.0); + assert_eq!( + s.score, 0.0, + "composite must be zero when any factor is zero" + ); + } + + #[test] + fn deterministic_given_same_inputs() { + let now = Utc::now(); + let v = vec![ + mk(now - Duration::days(1), true, 100), + mk(now - Duration::days(2), false, 150), + mk(now - Duration::days(3), true, 200), + ]; + let a = score(&v, now); + let b = score(&v, now); + assert_eq!(a.score, b.score); + assert_eq!(a.recency, b.recency); + } + + #[test] + fn old_burst_does_not_inflate_frequency_score() { + // 100 interactions from 90 days ago (outside FREQUENCY_WINDOW_DAYS=30) + // should contribute 0 to frequency; 1 interaction today should give + // 1/FREQUENCY_CAP. + let now = Utc::now(); + let mut v: Vec = (0..100) + .map(|i| mk(now - Duration::days(90 + i), true, 100)) + .collect(); + // Add one recent interaction to avoid zero reciprocity forcing score=0 + v.push(mk(now - Duration::hours(1), false, 100)); + let s = score(&v, now); + // Only 1 interaction falls within the 30-day window. + let expected_frequency = 1.0 / FREQUENCY_CAP; + assert!( + (s.frequency - expected_frequency).abs() < 0.001, + "frequency should be {expected_frequency}, got {}", + s.frequency + ); + } + + #[test] + fn interactions_exactly_at_window_boundary_are_included() { + let now = Utc::now(); + // Interaction exactly FREQUENCY_WINDOW_DAYS ago — should be included + // (boundary is inclusive via >=). + let boundary = now - Duration::days(FREQUENCY_WINDOW_DAYS as i64); + let v = vec![ + mk(boundary, true, 100), + mk(now - Duration::hours(1), false, 100), + ]; + let s = score(&v, now); + let expected = 2.0 / FREQUENCY_CAP; + assert!( + (s.frequency - expected).abs() < 0.001, + "expected {expected} got {}", + s.frequency + ); + } +} diff --git a/src/memory/people/store.rs b/src/memory/people/store.rs new file mode 100644 index 0000000..3855d93 --- /dev/null +++ b/src/memory/people/store.rs @@ -0,0 +1,653 @@ +//! SQLite-backed store for people + handle aliases + interactions. +//! +//! Connection is wrapped in `Arc>` so handlers and tests +//! can share ownership across tokio tasks; operations are synchronous and +//! fast (all single-row CRUD or small aggregates). + +use std::collections::HashMap; +use std::path::{Path, PathBuf}; +use std::sync::{Arc, OnceLock, RwLock}; + +use chrono::{DateTime, TimeZone, Utc}; +use rusqlite::{params, Connection, OptionalExtension, Result as SqlResult}; +use tokio::sync::Mutex; + +use crate::memory::people::migrations; +use crate::memory::people::types::{Handle, Interaction, Person, PersonId}; + +pub type ConnHandle = Arc>; +type PersonRow = ( + String, + Option, + Option, + Option, + i64, + i64, +); + +/// Process-global handle to the `PeopleStore`, tagged with the workspace it is +/// bound to. Controller handlers are free functions with no `&self`, so they +/// fetch the store via `get()`. Seeded at core boot and re-bound on active-user +/// switch via [`init_from_workspace`]. Absent at test time unless a test seeds +/// it; most tests construct stores directly with `open_in_memory`. +#[derive(Clone)] +struct GlobalPeopleStore { + workspace_dir: PathBuf, + store: Arc, +} + +type GlobalStoreSlot = RwLock>; + +static GLOBAL: OnceLock = OnceLock::new(); + +fn global_slot() -> &'static GlobalStoreSlot { + GLOBAL.get_or_init(GlobalStoreSlot::default) +} + +/// Initialise or re-bind the process-global people store from a workspace +/// directory, opening `/people/people.db` (schema migrations run on +/// open). +/// +/// Safe to call repeatedly. +/// A call for the **same** workspace returns the existing store; a call for a +/// **different** workspace replaces the global handle so a post-login +/// active-user switch (or `restart_core_process`, which restarts the embedded +/// core in the same Tauri process) does not keep people controllers/tools +/// reading and writing the pre-login (or a previous user's) workspace. +/// +/// Wired into core boot (`src/core/jsonrpc.rs`) and the active-user rebind +/// sites (`credentials::ops`, `app_state::ops`) alongside `memory::global`. +/// Without the boot seed every people controller / `people_*` tool fails with +/// "people store not initialised" (Sentry TAURI-RUST-8NM); without the rebind +/// they'd write the wrong workspace after login (#4378). +pub fn init_from_workspace(workspace_dir: &Path) -> Result, String> { + let slot = global_slot(); + if let Some(existing) = slot + .read() + .map_err(|e| format!("[people:store] read lock poisoned: {e}"))? + .as_ref() + { + if existing.workspace_dir == workspace_dir { + log::debug!("[people:store] already initialised for current workspace"); + return Ok(Arc::clone(&existing.store)); + } + } + + let db_path = workspace_dir.join("people").join("people.db"); + let store = Arc::new( + PeopleStore::open_at(&db_path).map_err(|e| format!("people store open failed: {e}"))?, + ); + + let mut guard = slot + .write() + .map_err(|e| format!("[people:store] write lock poisoned: {e}"))?; + // Re-check under the write lock: a concurrent caller may have seeded the + // same workspace while we were opening — reuse theirs. A different-workspace + // entry is replaced (rebind). + if let Some(existing) = guard.as_ref() { + if existing.workspace_dir == workspace_dir { + return Ok(Arc::clone(&existing.store)); + } + } + log::info!( + "[people:store] bound store workspace={}", + workspace_dir.display() + ); + *guard = Some(GlobalPeopleStore { + workspace_dir: workspace_dir.to_path_buf(), + store: Arc::clone(&store), + }); + Ok(store) +} + +pub fn get() -> Result, &'static str> { + global_slot() + .read() + .ok() + .and_then(|guard| guard.as_ref().map(|entry| Arc::clone(&entry.store))) + .ok_or("people store not initialised — core startup hasn't completed") +} + +/// Per-workspace store cache keyed by workspace dir. Backs [`for_workspace`], +/// the context-scoped accessor (the host's context-scoped `CoreContext::people`). +/// Distinct from the single `GLOBAL` slot above (which tracks the one +/// active-user workspace for the legacy free-function handlers): this map lets +/// multiple workspaces' stores coexist in one process, which is what per-context +/// isolation (Phase 3) needs. +static STORES: OnceLock>>> = + OnceLock::new(); + +/// Open (or return the cached) people store for a specific workspace dir. Unlike +/// [`get`], this is not tied to the single active-user global — two different +/// workspaces resolve to two isolated stores, and the same workspace always +/// resolves to the same cached `Arc`. Opening `/people/people.db` +/// runs schema migrations. +pub fn for_workspace(workspace_dir: &Path) -> Result, String> { + let cache = STORES.get_or_init(Default::default); + if let Some(store) = cache + .read() + .map_err(|e| format!("[people:store] cache read lock poisoned: {e}"))? + .get(workspace_dir) + { + return Ok(Arc::clone(store)); + } + + let db_path = workspace_dir.join("people").join("people.db"); + let store = Arc::new( + PeopleStore::open_at(&db_path).map_err(|e| format!("people store open failed: {e}"))?, + ); + + let mut guard = cache + .write() + .map_err(|e| format!("[people:store] cache write lock poisoned: {e}"))?; + // Re-check under the write lock: a concurrent caller may have opened the + // same workspace while we were opening — reuse theirs so callers always + // share one store per workspace. + let entry = guard + .entry(workspace_dir.to_path_buf()) + .or_insert_with(|| Arc::clone(&store)); + Ok(Arc::clone(entry)) +} + +pub struct PeopleStore { + pub conn: ConnHandle, +} + +impl PeopleStore { + pub fn open_in_memory() -> SqlResult { + let conn = Connection::open_in_memory()?; + migrations::run(&conn)?; + Ok(Self { + conn: Arc::new(Mutex::new(conn)), + }) + } + + pub fn open_at(path: &std::path::Path) -> SqlResult { + if let Some(parent) = path.parent() { + let _ = std::fs::create_dir_all(parent); + } + let conn = Connection::open(path)?; + migrations::run(&conn)?; + Ok(Self { + conn: Arc::new(Mutex::new(conn)), + }) + } + + /// Insert a new person and its initial set of handles, atomically. + pub async fn insert_person(&self, person: &Person, handles: &[Handle]) -> SqlResult<()> { + let conn = self.conn.clone(); + let person = person.clone(); + let handles: Vec = handles.iter().map(|h| h.canonicalize()).collect(); + tokio::task::spawn_blocking(move || { + let mut guard = conn.blocking_lock(); + let tx = guard.transaction()?; + tx.execute( + "INSERT INTO people(id, display_name, primary_email, primary_phone, created_at, updated_at) \ + VALUES (?1, ?2, ?3, ?4, ?5, ?6)", + params![ + person.id.to_string(), + person.display_name, + person.primary_email, + person.primary_phone, + person.created_at.timestamp(), + person.updated_at.timestamp(), + ], + )?; + for h in &handles { + let (kind, value) = h.as_key(); + tx.execute( + "INSERT OR IGNORE INTO handle_aliases(kind, value, person_id, created_at) \ + VALUES (?1, ?2, ?3, CAST(strftime('%s','now') AS INTEGER))", + params![kind, value, person.id.to_string()], + )?; + } + tx.commit() + }) + .await + .map_err(|e| rusqlite::Error::SqliteFailure( + rusqlite::ffi::Error { + code: rusqlite::ffi::ErrorCode::SystemIoFailure, + extended_code: 0, + }, + Some(e.to_string()), + ))? + } + + /// Resolve an existing canonical handle or insert a new person and alias + /// under one connection lock. Returns the database-authoritative id plus + /// whether this call created the row. + pub async fn resolve_or_insert_person( + &self, + person: &Person, + handle: &Handle, + ) -> SqlResult<(PersonId, bool)> { + let conn = self.conn.clone(); + let person = person.clone(); + let handle = handle.canonicalize(); + tokio::task::spawn_blocking(move || -> SqlResult<(PersonId, bool)> { + let mut guard = conn.blocking_lock(); + let tx = guard.transaction()?; + let (kind, value) = handle.as_key(); + let existing: Option = tx + .query_row( + "SELECT person_id FROM handle_aliases WHERE kind = ?1 AND value = ?2", + params![kind, value], + |row| row.get(0), + ) + .optional()?; + if let Some(id) = existing { + let id = uuid::Uuid::parse_str(&id) + .map(PersonId) + .map_err(|e| rusqlite::Error::InvalidColumnName(e.to_string()))?; + return Ok((id, false)); + } + + tx.execute( + "INSERT INTO people(id, display_name, primary_email, primary_phone, created_at, updated_at) \ + VALUES (?1, ?2, ?3, ?4, ?5, ?6)", + params![ + person.id.to_string(), + person.display_name, + person.primary_email, + person.primary_phone, + person.created_at.timestamp(), + person.updated_at.timestamp(), + ], + )?; + tx.execute( + "INSERT INTO handle_aliases(kind, value, person_id, created_at) \ + VALUES (?1, ?2, ?3, CAST(strftime('%s','now') AS INTEGER))", + params![kind, value, person.id.to_string()], + )?; + tx.commit()?; + Ok((person.id, true)) + }) + .await + .map_err(|e| { + rusqlite::Error::SqliteFailure( + rusqlite::ffi::Error { + code: rusqlite::ffi::ErrorCode::SystemIoFailure, + extended_code: 0, + }, + Some(e.to_string()), + ) + })? + } + + /// Attach a handle alias to an existing person. Idempotent via + /// `INSERT OR IGNORE` on `(kind, value)`. + pub async fn add_alias(&self, person_id: PersonId, handle: Handle) -> SqlResult<()> { + let conn = self.conn.clone(); + let handle = handle.canonicalize(); + tokio::task::spawn_blocking(move || { + let guard = conn.blocking_lock(); + let (kind, value) = handle.as_key(); + guard.execute( + "INSERT OR IGNORE INTO handle_aliases(kind, value, person_id, created_at) \ + VALUES (?1, ?2, ?3, CAST(strftime('%s','now') AS INTEGER))", + params![kind, value, person_id.to_string()], + )?; + Ok(()) + }) + .await + .map_err(|e| { + rusqlite::Error::SqliteFailure( + rusqlite::ffi::Error { + code: rusqlite::ffi::ErrorCode::SystemIoFailure, + extended_code: 0, + }, + Some(e.to_string()), + ) + })? + } + + /// Resolve a canonicalized handle to a `PersonId`, or `None` if unknown. + pub async fn lookup(&self, handle: &Handle) -> SqlResult> { + let conn = self.conn.clone(); + let handle = handle.canonicalize(); + tokio::task::spawn_blocking(move || { + let guard = conn.blocking_lock(); + let (kind, value) = handle.as_key(); + let id: Option = guard + .query_row( + "SELECT person_id FROM handle_aliases WHERE kind = ?1 AND value = ?2", + params![kind, value], + |row| row.get(0), + ) + .optional()?; + Ok(id.and_then(|s| uuid::Uuid::parse_str(&s).ok().map(PersonId))) + }) + .await + .map_err(|e| { + rusqlite::Error::SqliteFailure( + rusqlite::ffi::Error { + code: rusqlite::ffi::ErrorCode::SystemIoFailure, + extended_code: 0, + }, + Some(e.to_string()), + ) + })? + } + + /// Load a person and all their aliases. + pub async fn get(&self, person_id: PersonId) -> SqlResult> { + let conn = self.conn.clone(); + tokio::task::spawn_blocking(move || -> SqlResult> { + let guard = conn.blocking_lock(); + let row: Option = + guard + .query_row( + "SELECT id, display_name, primary_email, primary_phone, created_at, updated_at \ + FROM people WHERE id = ?1", + params![person_id.to_string()], + |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?, r.get(4)?, r.get(5)?)), + ) + .optional()?; + let Some((id_str, display_name, primary_email, primary_phone, created, updated)) = row + else { + return Ok(None); + }; + let id = uuid::Uuid::parse_str(&id_str) + .map(PersonId) + .map_err(|e| rusqlite::Error::InvalidColumnName(e.to_string()))?; + let handles = load_handles(&guard, &id)?; + Ok(Some(Person { + id, + display_name, + primary_email, + primary_phone, + handles, + created_at: ts_to_dt(created), + updated_at: ts_to_dt(updated), + })) + }) + .await + .map_err(|e| rusqlite::Error::SqliteFailure( + rusqlite::ffi::Error { + code: rusqlite::ffi::ErrorCode::SystemIoFailure, + extended_code: 0, + }, + Some(e.to_string()), + ))? + } + + /// List all people (unordered — scorer applies ranking separately). + pub async fn list(&self) -> SqlResult> { + let conn = self.conn.clone(); + tokio::task::spawn_blocking(move || -> SqlResult> { + let guard = conn.blocking_lock(); + let mut stmt = guard.prepare( + "SELECT id, display_name, primary_email, primary_phone, created_at, updated_at \ + FROM people ORDER BY display_name", + )?; + let rows = stmt.query_map([], |r| { + Ok(( + r.get::<_, String>(0)?, + r.get::<_, Option>(1)?, + r.get::<_, Option>(2)?, + r.get::<_, Option>(3)?, + r.get::<_, i64>(4)?, + r.get::<_, i64>(5)?, + )) + })?; + let mut out = Vec::new(); + for r in rows { + let (id_str, display_name, primary_email, primary_phone, created, updated) = r?; + let id = uuid::Uuid::parse_str(&id_str) + .map(PersonId) + .map_err(|e| rusqlite::Error::InvalidColumnName(e.to_string()))?; + let handles = load_handles(&guard, &id)?; + out.push(Person { + id, + display_name, + primary_email, + primary_phone, + handles, + created_at: ts_to_dt(created), + updated_at: ts_to_dt(updated), + }); + } + Ok(out) + }) + .await + .map_err(|e| { + rusqlite::Error::SqliteFailure( + rusqlite::ffi::Error { + code: rusqlite::ffi::ErrorCode::SystemIoFailure, + extended_code: 0, + }, + Some(e.to_string()), + ) + })? + } + + /// Record a single interaction. + pub async fn record_interaction(&self, i: Interaction) -> SqlResult<()> { + let conn = self.conn.clone(); + tokio::task::spawn_blocking(move || { + let guard = conn.blocking_lock(); + guard.execute( + "INSERT INTO interactions(person_id, ts, is_outbound, length) \ + VALUES (?1, ?2, ?3, ?4)", + params![ + i.person_id.to_string(), + i.ts.timestamp(), + if i.is_outbound { 1_i64 } else { 0_i64 }, + i.length as i64, + ], + )?; + Ok(()) + }) + .await + .map_err(|e| { + rusqlite::Error::SqliteFailure( + rusqlite::ffi::Error { + code: rusqlite::ffi::ErrorCode::SystemIoFailure, + extended_code: 0, + }, + Some(e.to_string()), + ) + })? + } + + /// Fetch all interactions for a person, newest first. + pub async fn interactions_for(&self, person_id: PersonId) -> SqlResult> { + let conn = self.conn.clone(); + tokio::task::spawn_blocking(move || -> SqlResult> { + let guard = conn.blocking_lock(); + let mut stmt = guard.prepare( + "SELECT ts, is_outbound, length FROM interactions \ + WHERE person_id = ?1 ORDER BY ts DESC", + )?; + let rows = stmt.query_map(params![person_id.to_string()], |r| { + Ok(( + r.get::<_, i64>(0)?, + r.get::<_, i64>(1)?, + r.get::<_, i64>(2)?, + )) + })?; + let mut out = Vec::new(); + for r in rows { + let (ts, is_out, length) = r?; + out.push(Interaction { + person_id, + ts: ts_to_dt(ts), + is_outbound: is_out != 0, + length: length.max(0) as u32, + }); + } + Ok(out) + }) + .await + .map_err(|e| { + rusqlite::Error::SqliteFailure( + rusqlite::ffi::Error { + code: rusqlite::ffi::ErrorCode::SystemIoFailure, + extended_code: 0, + }, + Some(e.to_string()), + ) + })? + } + + /// Fetch interactions for several people in one query, keyed by person id. + pub async fn batch_interactions_for( + &self, + person_ids: &[PersonId], + ) -> SqlResult>> { + if person_ids.is_empty() { + return Ok(HashMap::new()); + } + let conn = self.conn.clone(); + let ids: Vec = person_ids.to_vec(); + tokio::task::spawn_blocking(move || -> SqlResult>> { + let guard = conn.blocking_lock(); + let placeholders = std::iter::repeat_n("?", ids.len()) + .collect::>() + .join(","); + let sql = format!( + "SELECT person_id, ts, is_outbound, length FROM interactions \ + WHERE person_id IN ({placeholders}) ORDER BY person_id, ts DESC" + ); + let id_strings: Vec = ids.iter().map(ToString::to_string).collect(); + let mut stmt = guard.prepare(&sql)?; + let rows = stmt.query_map(rusqlite::params_from_iter(id_strings.iter()), |r| { + Ok(( + r.get::<_, String>(0)?, + r.get::<_, i64>(1)?, + r.get::<_, i64>(2)?, + r.get::<_, i64>(3)?, + )) + })?; + let mut out: HashMap> = HashMap::new(); + for r in rows { + let (id_str, ts, is_out, length) = r?; + let person_id = uuid::Uuid::parse_str(&id_str) + .map(PersonId) + .map_err(|e| rusqlite::Error::InvalidColumnName(e.to_string()))?; + out.entry(person_id).or_default().push(Interaction { + person_id, + ts: ts_to_dt(ts), + is_outbound: is_out != 0, + length: length.max(0) as u32, + }); + } + Ok(out) + }) + .await + .map_err(|e| { + rusqlite::Error::SqliteFailure( + rusqlite::ffi::Error { + code: rusqlite::ffi::ErrorCode::SystemIoFailure, + extended_code: 0, + }, + Some(e.to_string()), + ) + })? + } +} + +fn load_handles(conn: &Connection, id: &PersonId) -> SqlResult> { + let mut stmt = conn.prepare( + "SELECT kind, value FROM handle_aliases WHERE person_id = ?1 ORDER BY kind, value", + )?; + let rows = stmt.query_map(params![id.to_string()], |r| { + Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?)) + })?; + let mut out = Vec::new(); + for r in rows { + let (kind, value) = r?; + let h = match kind.as_str() { + "imessage" => Handle::IMessage(value), + "email" => Handle::Email(value), + "display_name" => Handle::DisplayName(value), + other => { + return Err(rusqlite::Error::InvalidColumnName(format!( + "unknown handle kind: {other}" + ))); + } + }; + out.push(h); + } + Ok(out) +} + +fn ts_to_dt(ts: i64) -> DateTime { + Utc.timestamp_opt(ts, 0) + .single() + .unwrap_or_else(|| Utc.timestamp_opt(0, 0).unwrap()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn insert_list_and_lookup_round_trip() { + let s = PeopleStore::open_in_memory().unwrap(); + let now = Utc::now(); + let p = Person { + id: PersonId::new(), + display_name: Some("Sarah Lee".into()), + primary_email: Some("sarah@example.com".into()), + primary_phone: None, + handles: vec![], + created_at: now, + updated_at: now, + }; + s.insert_person( + &p, + &[ + Handle::Email("Sarah@Example.com".into()), + Handle::DisplayName("Sarah Lee".into()), + ], + ) + .await + .unwrap(); + + let got = s + .lookup(&Handle::Email("sarah@example.com".into())) + .await + .unwrap(); + assert_eq!(got, Some(p.id)); + + let list = s.list().await.unwrap(); + assert_eq!(list.len(), 1); + assert_eq!(list[0].handles.len(), 2); + } + + #[tokio::test] + async fn interactions_round_trip() { + let s = PeopleStore::open_in_memory().unwrap(); + let now = Utc::now(); + let pid = PersonId::new(); + let p = Person { + id: pid, + display_name: Some("X".into()), + primary_email: None, + primary_phone: None, + handles: vec![], + created_at: now, + updated_at: now, + }; + s.insert_person(&p, &[]).await.unwrap(); + s.record_interaction(Interaction { + person_id: pid, + ts: now, + is_outbound: true, + length: 100, + }) + .await + .unwrap(); + s.record_interaction(Interaction { + person_id: pid, + ts: now, + is_outbound: false, + length: 50, + }) + .await + .unwrap(); + let ints = s.interactions_for(pid).await.unwrap(); + assert_eq!(ints.len(), 2); + } +} diff --git a/src/memory/people/tests.rs b/src/memory/people/tests.rs new file mode 100644 index 0000000..722ae14 --- /dev/null +++ b/src/memory/people/tests.rs @@ -0,0 +1,116 @@ +//! Cross-file integration tests for the people domain. + +use std::sync::Arc; + +use chrono::Utc; + +#[cfg(not(target_os = "macos"))] +use crate::memory::people::address_book; +use crate::memory::people::resolver::HandleResolver; +use crate::memory::people::store::PeopleStore; +use crate::memory::people::types::{Handle, PersonId}; + +/// Serialises every test that mutates the process-global people store. +/// +/// The slot is process-wide, so two tests rebinding it concurrently race and +/// either may observe the other's store through `get()`. +static GLOBAL_STORE_LOCK: parking_lot::Mutex<()> = parking_lot::Mutex::new(()); + +#[tokio::test] +async fn resolver_and_store_cooperate_across_handle_kinds() { + let s = PeopleStore::open_in_memory().unwrap(); + let r = HandleResolver::new(&s); + + // Email mints. + let id = r + .resolve_or_create(&Handle::Email("a@b.c".into())) + .await + .unwrap(); + // iMessage handle linked to same person. + let id2 = r + .link( + &Handle::Email("a@b.c".into()), + Handle::IMessage("+15551234".into()), + ) + .await + .unwrap(); + assert_eq!(id, id2); + + // Resolving by the linked iMessage handle returns the same id. + let via_imsg = r + .resolve(&Handle::IMessage("+15551234".into())) + .await + .unwrap(); + assert_eq!(via_imsg, Some(id)); +} + +#[cfg(not(target_os = "macos"))] +#[test] +fn address_book_is_empty_on_non_mac() { + assert!(address_book::read().unwrap().is_empty()); +} + +/// Regression for Sentry TAURI-RUST-8NM (store never seeded → `get()` always +/// errored) and its #4378 follow-up (store stayed bound to the pre-login +/// workspace after an active-user switch). Verify `init_from_workspace` seeds +/// the global + creates the on-disk db, is an idempotent no-op for the same +/// workspace, and **rebinds** to a different workspace like `memory::global`. +/// +/// Takes [`GLOBAL_STORE_LOCK`] because it mutates the process-global store +/// slot that any test may observe through `get()`. `#[test]` alone does **not** +/// serialise anything — the harness runs tests in parallel by default — so the +/// lock is what makes that true, and it must be taken by every future test that +/// touches the global. +#[test] +fn init_from_workspace_seeds_and_rebinds_global_store() { + use crate::memory::people::store; + + let _serial = GLOBAL_STORE_LOCK.lock(); + + let ws_a = tempfile::tempdir().unwrap(); + let store_a = store::init_from_workspace(ws_a.path()).unwrap(); + assert!( + ws_a.path().join("people").join("people.db").exists(), + "seed must create /people/people.db" + ); + + // Previously-dead global is now reachable — the 8NM fix. + let via_global = store::get().expect("people store reachable after seed"); + assert!(Arc::ptr_eq(&store_a, &via_global)); + + // Same workspace → idempotent no-op, returns the same instance. + let again = store::init_from_workspace(ws_a.path()).unwrap(); + assert!(Arc::ptr_eq(&store_a, &again)); + + // Different workspace (active-user switch) → rebind to a new store. #4378. + let ws_b = tempfile::tempdir().unwrap(); + let store_b = store::init_from_workspace(ws_b.path()).unwrap(); + assert!( + !Arc::ptr_eq(&store_a, &store_b), + "a new workspace must rebind to a fresh store, not reuse the old one" + ); + let after_switch = store::get().expect("people store reachable after rebind"); + assert!( + Arc::ptr_eq(&store_b, &after_switch), + "get() must return the rebound (workspace B) store after a switch" + ); + + // Leave the global bound to a workspace that outlives this test. Both temp + // dirs above are deleted when they drop, and the global would keep pointing + // at whichever was bound last — so a later `get()` would hand back a store + // over a database file that no longer exists. Leaking one directory is the + // cheap way to keep the slot valid for the rest of the process; there is no + // unbind, because production never unbinds either. + let keep: &'static tempfile::TempDir = Box::leak(Box::new(tempfile::tempdir().unwrap())); + store::init_from_workspace(keep.path()).unwrap(); +} + +#[test] +fn person_id_uuid_format() { + let id = PersonId::new(); + // Round-trips through a string. + let s = id.to_string(); + let parsed: uuid::Uuid = s.parse().unwrap(); + assert_eq!(parsed, id.0); + let _now = Utc::now(); +} diff --git a/src/memory/people/types.rs b/src/memory/people/types.rs new file mode 100644 index 0000000..34a0ec7 --- /dev/null +++ b/src/memory/people/types.rs @@ -0,0 +1,159 @@ +//! Core types for the people domain. + +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; +use uuid::Uuid; + +/// Canonical, stable identifier for a person across handles. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] +#[serde(transparent)] +pub struct PersonId(pub Uuid); + +impl PersonId { + pub fn new() -> Self { + Self(Uuid::new_v4()) + } +} + +impl Default for PersonId { + fn default() -> Self { + Self::new() + } +} + +impl std::fmt::Display for PersonId { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{}", self.0) + } +} + +/// A handle is an opaque label by which the user or a source knows a person. +/// `IMessage(h)` is an iMessage chat handle (phone in E.164, or apple id +/// email). `Email(e)` and `DisplayName(n)` are the other two kinds the A5 +/// resolver accepts. +#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)] +#[serde(tag = "kind", content = "value", rename_all = "snake_case")] +pub enum Handle { + IMessage(String), + Email(String), + DisplayName(String), +} + +impl Handle { + /// Return a canonical, case-folded, whitespace-trimmed form used both + /// for storage and for the resolver lookup key. Emails are lowercased; + /// iMessage handles strip surrounding whitespace and lowercase email- + /// style handles; display names are whitespace-collapsed and trimmed. + pub fn canonicalize(&self) -> Handle { + match self { + Handle::IMessage(s) => { + let t = s.trim(); + // An apple id email handle ("foo@bar.com") is treated the + // same regardless of case; phone-style handles ("+1…") have + // no case. Lowercasing is safe for both. + Handle::IMessage(t.to_lowercase()) + } + Handle::Email(s) => Handle::Email(s.trim().to_lowercase()), + Handle::DisplayName(s) => { + let collapsed: String = s.split_whitespace().collect::>().join(" "); + Handle::DisplayName(collapsed) + } + } + } + + /// `(kind, value)` tuple suitable for use as a SQL key. + pub fn as_key(&self) -> (&'static str, &str) { + match self { + Handle::IMessage(s) => ("imessage", s.as_str()), + Handle::Email(s) => ("email", s.as_str()), + Handle::DisplayName(s) => ("display_name", s.as_str()), + } + } +} + +/// Stored representation of a person plus display metadata. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct Person { + pub id: PersonId, + pub display_name: Option, + pub primary_email: Option, + pub primary_phone: Option, + pub handles: Vec, + pub created_at: DateTime, + pub updated_at: DateTime, +} + +/// A single interaction observed with a person. The scorer aggregates +/// these. `is_outbound = true` means the user sent it; that's what drives +/// reciprocity. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct Interaction { + pub person_id: PersonId, + pub ts: DateTime, + pub is_outbound: bool, + /// Token or character count used as a proxy for "depth". Clamped in + /// scoring; callers may pass e.g. message body length. + pub length: u32, +} + +/// Per-component breakdown of a person-score in `[0,1]`. Exposed so that +/// callers (UI, nudge engine) can explain ranking. +#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)] +pub struct ScoreComponents { + pub recency: f32, + pub frequency: f32, + pub reciprocity: f32, + pub depth: f32, + /// Final composite score. `recency * frequency * reciprocity * depth`, + /// clamped to `[0,1]`. + pub score: f32, +} + +/// Lightweight row returned from the macOS Address Book. We keep this a +/// plain data struct so `address_book::read()` can return the same shape +/// on every OS (empty on non-mac). +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct AddressBookContact { + pub display_name: Option, + pub emails: Vec, + pub phones: Vec, +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn handle_canonicalize_lowercases_emails_and_imessage() { + assert_eq!( + Handle::Email(" Foo@Example.COM ".into()).canonicalize(), + Handle::Email("foo@example.com".into()) + ); + assert_eq!( + Handle::IMessage("+1 (555) 123".into()).canonicalize(), + Handle::IMessage("+1 (555) 123".into()) + ); + assert_eq!( + Handle::IMessage(" Foo@Bar.com ".into()).canonicalize(), + Handle::IMessage("foo@bar.com".into()) + ); + } + + #[test] + fn handle_canonicalize_collapses_display_name_whitespace() { + assert_eq!( + Handle::DisplayName(" Sarah Lee ".into()).canonicalize(), + Handle::DisplayName("Sarah Lee".into()) + ); + } + + #[test] + fn handle_as_key_returns_correct_kind() { + assert_eq!(Handle::Email("a@b.c".into()).as_key(), ("email", "a@b.c")); + assert_eq!(Handle::IMessage("+1".into()).as_key(), ("imessage", "+1")); + assert_eq!( + Handle::DisplayName("X".into()).as_key(), + ("display_name", "X") + ); + } +}