feat(persona-chat): continue an SMS thread with a persona

Persona conversations can now be anchored on a message in an SMS thread.
Creating one snapshots the thread up to that message (via SMS-API) into the
row; every turn appends it to the persona prompt with guidance to talk like a
text thread rather than write an insight.

Anchored turns get four tools: get_thread_messages (page the thread),
look_at_message_image (local vision on an MMS photo), get_photos_near_time
(library photos with tags and people) and look_at_photo (local vision on a
library photo). The conversation list filters by contact_id.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Cameron CordesandClaude Opus 5.5 committed 2026-10-17 18:07:49 -04:00
1 parent fdbcc75391
commit 4ac574d4d2
11 files changed
+1389 -40

No files matched your search

+9
View File
@@ -577,6 +577,15 @@ POST /insights/chat/stream (SSE: text / tool_call / tool_result / trun
GET /insights/chat/history?path=... (rendered transcript with tool invocations) GET /insights/chat/history?path=... (rendered transcript with tool invocations)
POST /insights/chat/rewind (truncate transcript at a rendered index) POST /insights/chat/rewind (truncate transcript at a rendered index)
// Persona chat — "Continue with persona" from an SMS thread (src/ai/sms_anchor.rs)
POST /persona_chat/conversations (body: { persona_id, anchor?: { contact_id, message_id } };
anchor snapshots ~40 SMS-API messages up to message_id into
the row, 502 if SMS-API can't read the thread)
GET /persona_chat/conversations?contact_id=N (only chats anchored on that thread)
// Anchored turns append the thread + chat-register guidance to the persona prompt and add
// the get_thread_messages / look_at_message_image / get_photos_near_time / look_at_photo
// tools (ToolGateOpts.sms_thread_tools). Image tools use the LOCAL vision model.
// User AI Prefs (owner defaults + passive mirror; src/prefs.rs) // User AI Prefs (owner defaults + passive mirror; src/prefs.rs)
GET /prefs (current row; auth) GET /prefs (current row; auth)
PUT /prefs (partial update: omitted=unchanged, null=clear, value=set; auth) PUT /prefs (partial update: omitted=unchanged, null=clear, value=set; auth)
@@ -0,0 +1,6 @@
DROP INDEX IF EXISTS idx_persona_chat_anchor_contact;
ALTER TABLE persona_chat_conversations DROP COLUMN anchor_context;
ALTER TABLE persona_chat_conversations DROP COLUMN anchor_ts;
ALTER TABLE persona_chat_conversations DROP COLUMN anchor_message_id;
ALTER TABLE persona_chat_conversations DROP COLUMN anchor_contact_name;
ALTER TABLE persona_chat_conversations DROP COLUMN anchor_contact_id;
@@ -0,0 +1,23 @@
-- Anchor a persona conversation on a point in an SMS thread.
--
-- "Continue with persona" starts a persona chat from a long-pressed message
-- in the Messages screen. The conversation remembers which thread and which
-- message it grew out of, so the SMS screen can list the persona chats that
-- belong to a thread and the chat screen can show where it picked up.
--
-- `anchor_context` is a snapshot of the thread leading up to the anchor,
-- rendered once at creation and spliced into the system prompt on every
-- turn. Snapshotting (rather than refetching per turn) keeps the prompt
-- prefix byte-stable for KV-cache reuse and keeps the persona's view of the
-- thread fixed even if SMS-API is re-imported underneath it.
--
-- All nullable: an unanchored (open) persona chat leaves them NULL.
ALTER TABLE persona_chat_conversations ADD COLUMN anchor_contact_id BIGINT;
ALTER TABLE persona_chat_conversations ADD COLUMN anchor_contact_name TEXT;
ALTER TABLE persona_chat_conversations ADD COLUMN anchor_message_id BIGINT;
ALTER TABLE persona_chat_conversations ADD COLUMN anchor_ts BIGINT;
ALTER TABLE persona_chat_conversations ADD COLUMN anchor_context TEXT;
CREATE INDEX idx_persona_chat_anchor_contact
ON persona_chat_conversations (user_id, anchor_contact_id, updated_at DESC);
+393 -1
View File
@@ -286,6 +286,12 @@ pub struct ToolGateOpts {
/// tool name still 404s, and the agent doesn't waste iterations /// tool name still 404s, and the agent doesn't waste iterations
/// trying corrections it isn't allowed to do. /// trying corrections it isn't allowed to do.
pub allow_agent_corrections: bool, pub allow_agent_corrections: bool,
/// Persona chats anchored on an SMS thread ("Continue with persona")
/// get the thread tools: page the thread, look at an MMS photo, find
/// library photos around a moment and look at one. Off everywhere else
/// — the file chat already has its photo, and an open chat has no
/// thread to page.
pub sms_thread_tools: bool,
} }
impl InsightGenerator { impl InsightGenerator {
@@ -421,6 +427,12 @@ impl InsightGenerator {
/// tool-definition gating (no point offering `get_personal_place_at` /// tool-definition gating (no point offering `get_personal_place_at`
/// when Apollo is unreachable) — exposed publicly so `insight_chat` /// when Apollo is unreachable) — exposed publicly so `insight_chat`
/// can call `build_tool_definitions` with the same flag. /// can call `build_tool_definitions` with the same flag.
/// The SMS-API client, for callers outside the tool loop (the anchored
/// persona chat snapshots a thread with it at creation).
pub fn sms_client(&self) -> &SmsApiClient {
&self.sms_client
}
pub fn apollo_enabled(&self) -> bool { pub fn apollo_enabled(&self) -> bool {
self.apollo_client.is_enabled() self.apollo_client.is_enabled()
} }
@@ -485,6 +497,7 @@ impl InsightGenerator {
location_history_present, location_history_present,
faces_present, faces_present,
allow_agent_corrections, allow_agent_corrections,
sms_thread_tools: false,
} }
} }
@@ -1925,6 +1938,13 @@ Return ONLY the summary, nothing else."#,
.await .await
} }
"get_current_datetime" => Self::tool_get_current_datetime(), "get_current_datetime" => Self::tool_get_current_datetime(),
"get_thread_messages" => self.tool_get_thread_messages(arguments).await,
"look_at_message_image" => {
self.tool_look_at_message_image(arguments, backend.local())
.await
}
"get_photos_near_time" => self.tool_get_photos_near_time(arguments).await,
"look_at_photo" => self.tool_look_at_photo(arguments, backend.local()).await,
unknown => format!("Unknown tool: {}", unknown), unknown => format!("Unknown tool: {}", unknown),
}; };
if result.starts_with("Error") || result.starts_with("No ") { if result.starts_with("Error") || result.starts_with("No ") {
@@ -2386,6 +2406,256 @@ Return ONLY the summary, nothing else."#,
} }
} }
/// Tool: get_thread_messages — page the anchored SMS thread from a
/// message id. Lines are rendered exactly like the anchor snapshot so
/// the model can keep paging from any id it sees.
async fn tool_get_thread_messages(&self, args: &serde_json::Value) -> String {
let Some(contact_id) = args.get("contact_id").and_then(|v| v.as_i64()) else {
return "Error: missing required parameter 'contact_id'".to_string();
};
let Some(message_id) = args.get("message_id").and_then(|v| v.as_i64()) else {
return "Error: missing required parameter 'message_id'".to_string();
};
let limit = args
.get("limit")
.and_then(|v| v.as_i64())
.unwrap_or(30)
.clamp(1, 80) as usize;
let (before, after) = match args.get("direction").and_then(|v| v.as_str()) {
Some("after") => (0, limit),
Some("around") => (limit / 2, limit - limit / 2),
_ => (limit, 0),
};
match self
.sms_client
.fetch_thread_window(contact_id, message_id, before, after)
.await
{
Ok(window) if window.messages.len() > 1 => {
let rendered = crate::ai::sms_anchor::render_thread_window(
&window,
&user_display_name(),
self.display_zone(),
);
format!(
"Found {} messages with {} (▶ marks #{}):\n{}",
window.messages.len(),
window.contact.display_name(),
message_id,
rendered
)
}
Ok(_) => "No more messages in that direction.".to_string(),
Err(e) => {
log::warn!("tool_get_thread_messages failed: {}", e);
format!("Error fetching thread messages: {}", e)
}
}
}
/// Tool: look_at_message_image — download an MMS photo from SMS-API and
/// describe it with the local vision model.
async fn tool_look_at_message_image(
&self,
args: &serde_json::Value,
local: &dyn LlmClient,
) -> String {
let Some(contact_id) = args.get("contact_id").and_then(|v| v.as_i64()) else {
return "Error: missing required parameter 'contact_id'".to_string();
};
let Some(message_id) = args.get("message_id").and_then(|v| v.as_i64()) else {
return "Error: missing required parameter 'message_id'".to_string();
};
let media_index = args
.get("media_index")
.and_then(|v| v.as_u64())
.unwrap_or(0) as usize;
let bytes = match self
.sms_client
.fetch_mms_media(contact_id, message_id, media_index)
.await
{
Ok(b) => b,
Err(e) => return format!("Error fetching attachment: {}", e),
};
let image_base64 = match Self::encode_image_bytes_for_vision(&bytes) {
Ok(b64) => b64,
Err(_) => {
return "Error: that attachment isn't a viewable image (videos, audio and some \
formats can't be looked at)."
.to_string();
}
};
match local.describe_image(&image_base64).await {
Ok(desc) => desc,
Err(e) => format!("Error describing attachment: {}", e),
}
}
/// Tool: get_photos_near_time — library photos within ±hours of a
/// moment, closest first, with tags and recognised people.
async fn tool_get_photos_near_time(&self, args: &serde_json::Value) -> String {
let zone = self.display_zone();
let server_zone = crate::camera_tz::server_zone();
let Some(center) = Self::parse_moment(args, zone) else {
return "Error: pass 'timestamp' (unix seconds) or 'datetime' (YYYY-MM-DD HH:MM)"
.to_string();
};
let hours = args
.get("hours_radius")
.and_then(|v| v.as_i64())
.unwrap_or(3)
.clamp(1, 72);
let limit = args
.get("limit")
.and_then(|v| v.as_i64())
.unwrap_or(15)
.clamp(1, 40) as usize;
let range = crate::date_view::InstantRange::new(
Some(center - hours * 3600),
Some(center + hours * 3600),
zone,
server_zone,
);
let rows = {
let mut dao = self.exif_dao.lock().expect("Unable to lock ExifDao");
match dao.query_by_exif(None, None, None, None, None, range) {
Ok(r) => r,
Err(e) => return format!("Error querying photos: {:?}", e),
}
};
// Closest first; the same bytes in two libraries list once.
let mut dated: Vec<(i64, crate::database::models::ImageExif)> = rows
.into_iter()
.filter_map(|r| {
crate::date_view::row_true_instant(&r, zone, server_zone).map(|t| (t, r))
})
.collect();
dated.sort_by_key(|(t, _)| (t - center).abs());
let mut seen = std::collections::HashSet::new();
dated.retain(|(_, r)| match &r.content_hash {
Some(h) => seen.insert(h.clone()),
None => true,
});
let total = dated.len();
if total == 0 {
return format!("No photos within ±{} hours of that moment.", hours);
}
let mut lines = Vec::new();
for (_, row) in dated.into_iter().take(limit) {
let when = crate::date_view::row_display_wall(&row, zone, server_zone)
.map(|w| w.format("%Y-%m-%d %H:%M").to_string())
.unwrap_or_else(|| "unknown time".to_string());
let tags: Vec<String> = {
let mut dao = self.tag_dao.lock().expect("Unable to lock TagDao");
dao.get_tags_for_path(&row.file_path)
.map(|t| t.into_iter().map(|t| t.name).collect())
.unwrap_or_default()
};
let people: Vec<String> = match &row.content_hash {
Some(hash) => {
let mut dao = self.face_dao.lock().expect("Unable to lock FaceDao");
let mut names: Vec<String> = dao
.list_for_content_hash(hash)
.map(|faces| faces.into_iter().filter_map(|f| f.person_name).collect())
.unwrap_or_default();
names.sort();
names.dedup();
names
}
None => Vec::new(),
};
let mut line = format!("- {} — {}", row.file_path, when);
if !tags.is_empty() {
line.push_str(&format!(" · tags: {}", tags.join(", ")));
}
if !people.is_empty() {
line.push_str(&format!(" · people: {}", people.join(", ")));
}
if crate::file_types::is_video_file(std::path::Path::new(&row.file_path)) {
line.push_str(" · video");
}
lines.push(line);
}
format!(
"{}\n{}",
Self::found_header(total, lines.len(), "photos"),
lines.join("\n")
)
}
/// Tool: look_at_photo — describe a library photo by path with the
/// local vision model.
async fn tool_look_at_photo(&self, args: &serde_json::Value, local: &dyn LlmClient) -> String {
let file_path = match args.get("file_path").and_then(|v| v.as_str()) {
Some(p) if Self::photo_path_is_safe(p.trim()) => p.trim().to_string(),
Some(_) => return "Error: file_path must be a library-relative path".to_string(),
None => return "Error: missing required parameter 'file_path'".to_string(),
};
if crate::file_types::is_video_file(std::path::Path::new(&file_path)) {
return "Error: that's a video — only photos can be looked at.".to_string();
}
let image_base64 = match self.load_image_as_base64(&file_path) {
Ok(b) => b,
Err(e) => return format!("Error loading photo: {}", e),
};
match local.describe_image(&image_base64).await {
Ok(desc) => desc,
Err(e) => format!("Error describing photo: {}", e),
}
}
/// A model-supplied path is joined onto each library root, so refuse
/// anything that could climb out of it.
fn photo_path_is_safe(path: &str) -> bool {
use std::path::{Component, Path};
!path.is_empty()
&& Path::new(path)
.components()
.all(|c| matches!(c, Component::Normal(_) | Component::CurDir))
}
/// A tool's centre moment: `timestamp` (unix seconds) wins, else
/// `datetime` read as a wall clock in the owner's display zone. A bare
/// date centres on midday.
fn parse_moment(args: &serde_json::Value, zone: chrono_tz::Tz) -> Option<i64> {
if let Some(ts) = args.get("timestamp").and_then(|v| v.as_i64()) {
return Some(ts);
}
let raw = args.get("datetime").and_then(|v| v.as_str())?.trim();
let naive = chrono::NaiveDateTime::parse_from_str(raw, "%Y-%m-%d %H:%M")
.or_else(|_| chrono::NaiveDateTime::parse_from_str(raw, "%Y-%m-%dT%H:%M"))
.or_else(|_| chrono::NaiveDateTime::parse_from_str(raw, "%Y-%m-%dT%H:%M:%S"))
.ok()
.or_else(|| {
NaiveDate::parse_from_str(raw, "%Y-%m-%d")
.ok()
.and_then(|d| d.and_hms_opt(12, 0, 0))
})?;
zone.from_local_datetime(&naive)
.earliest()
.map(|dt| dt.timestamp())
}
/// Decode arbitrary image bytes (an MMS attachment), shrink to the
/// vision models' 1024px working size and re-encode as base64 JPEG —
/// the same shape `load_image_as_base64` produces for library files.
fn encode_image_bytes_for_vision(bytes: &[u8]) -> Result<String> {
use image::imageops::FilterType;
let img = image::load_from_memory(bytes)?;
// JPEG has no alpha channel; a PNG screenshot would fail to encode.
let resized =
image::DynamicImage::ImageRgb8(img.resize(1024, 1024, FilterType::Lanczos3).to_rgb8());
let mut buffer = Vec::new();
resized.write_to(&mut Cursor::new(&mut buffer), ImageFormat::Jpeg)?;
Ok(base64::engine::general_purpose::STANDARD.encode(&buffer))
}
/// Tool: get_calendar_events — fetch calendar events near a date /// Tool: get_calendar_events — fetch calendar events near a date
async fn tool_get_calendar_events(&self, args: &serde_json::Value) -> String { async fn tool_get_calendar_events(&self, args: &serde_json::Value) -> String {
let date_str = match args.get("date").and_then(|v| v.as_str()) { let date_str = match args.get("date").and_then(|v| v.as_str()) {
@@ -3781,6 +4051,77 @@ Return ONLY the summary, nothing else."#,
}), }),
)); ));
if opts.sms_thread_tools {
tools.push(Tool::function(
"get_thread_messages",
"Read more of the SMS thread this chat was opened from. Pages from a message id in the \
thread (the #numbers in the transcript) — `before` for older messages, `after` for what \
came next, `around` for both sides. Lines carry their own #id, so call again from the \
oldest/newest id to keep paging. \
Example: `{contact_id: 42, message_id: 812, direction: \"before\", limit: 30}`.",
serde_json::json!({
"type": "object",
"required": ["contact_id", "message_id"],
"properties": {
"contact_id": { "type": "integer", "description": "The thread's contact_id (given in the system prompt)." },
"message_id": { "type": "integer", "description": "Message id to page from." },
"direction": { "type": "string", "enum": ["before", "after", "around"], "description": "Which side of message_id to read (default before)." },
"limit": { "type": "integer", "description": "Max messages (default 30, max 80)." }
}
}),
));
tools.push(Tool::function(
"look_at_message_image",
"Look at a photo that was sent in the SMS thread, using the local vision model. Returns a \
visual description. Attachments appear in the transcript as `image #N` on a message line — \
pass that message's id and N as media_index. Videos and audio can't be viewed. \
Example: `{contact_id: 42, message_id: 815, media_index: 0}`.",
serde_json::json!({
"type": "object",
"required": ["contact_id", "message_id"],
"properties": {
"contact_id": { "type": "integer", "description": "The thread's contact_id." },
"message_id": { "type": "integer", "description": "Id of the message carrying the photo." },
"media_index": { "type": "integer", "description": "Which attachment on that message (default 0)." }
}
}),
));
tools.push(Tool::function(
"get_photos_near_time",
"List photos and videos from the user's library taken around a moment, closest first, with \
their user-applied tags and the names of people recognised in them. Use to see what was \
happening around a message (pass its time). Follow up with `look_at_photo` to see one. \
Pass either `timestamp` (unix seconds) or `datetime` (local, \"YYYY-MM-DD HH:MM\" or \
\"YYYY-MM-DD\"). \
Example: `{datetime: \"2019-06-12 18:30\", hours_radius: 4}`.",
serde_json::json!({
"type": "object",
"properties": {
"timestamp": { "type": "integer", "description": "Centre moment, unix seconds." },
"datetime": { "type": "string", "description": "Centre moment in local time, YYYY-MM-DD HH:MM (or YYYY-MM-DD for the whole day)." },
"hours_radius": { "type": "integer", "description": "Hours before and after to include (default 3, max 72)." },
"limit": { "type": "integer", "description": "Max photos (default 15, max 40)." }
}
}),
));
tools.push(Tool::function(
"look_at_photo",
"Look at one photo from the user's library with the local vision model and return a visual \
description — people, place, activity, mood. Pass a file_path from `get_photos_near_time`. \
Example: `{file_path: \"2019/06/IMG_4242.jpg\"}`.",
serde_json::json!({
"type": "object",
"required": ["file_path"],
"properties": {
"file_path": { "type": "string", "description": "Library-relative path of the photo." }
}
}),
));
}
if opts.has_vision { if opts.has_vision {
tools.push(Tool::function( tools.push(Tool::function(
"describe_photo", "describe_photo",
@@ -3915,7 +4256,8 @@ Return ONLY the summary, nothing else."#,
&& let Ok(n) = n_str.parse::<usize>() && let Ok(n) = n_str.parse::<usize>()
{ {
let kind = match tool_name { let kind = match tool_name {
"search_messages" | "get_sms_messages" => "messages", "search_messages" | "get_sms_messages" | "get_thread_messages" => "messages",
"get_photos_near_time" => "photos",
"get_calendar_events" => "events", "get_calendar_events" => "events",
"get_location_history" => "location records", "get_location_history" => "location records",
_ => "results", _ => "results",
@@ -3964,6 +4306,10 @@ Return ONLY the summary, nothing else."#,
.map(|id| format!("stored id={}", id.trim_end_matches(','))) .map(|id| format!("stored id={}", id.trim_end_matches(',')))
.unwrap_or_else(|| "stored".to_string()), .unwrap_or_else(|| "stored".to_string()),
"get_current_datetime" => "time noted".to_string(), "get_current_datetime" => "time noted".to_string(),
"look_at_message_image" | "look_at_photo" => {
let short: String = raw.chars().take(80).collect();
format!("looked: \"{}...\"", short)
}
_ => format!("{} chars", raw.len()), _ => format!("{} chars", raw.len()),
} }
} }
@@ -4882,6 +5228,7 @@ mod tests {
location_history_present: false, location_history_present: false,
faces_present: false, faces_present: false,
allow_agent_corrections: false, allow_agent_corrections: false,
sms_thread_tools: false,
}; };
let tools = InsightGenerator::build_tool_definitions(opts); let tools = InsightGenerator::build_tool_definitions(opts);
let names: Vec<&str> = tools.iter().map(|t| t.function.name.as_str()).collect(); let names: Vec<&str> = tools.iter().map(|t| t.function.name.as_str()).collect();
@@ -4908,6 +5255,11 @@ mod tests {
// Agent-correction tools are absent without the gate. // Agent-correction tools are absent without the gate.
assert!(!names.contains(&"update_fact")); assert!(!names.contains(&"update_fact"));
assert!(!names.contains(&"supersede_fact")); assert!(!names.contains(&"supersede_fact"));
// SMS-thread tools only exist for an anchored persona chat.
assert!(!names.contains(&"get_thread_messages"));
assert!(!names.contains(&"look_at_message_image"));
assert!(!names.contains(&"get_photos_near_time"));
assert!(!names.contains(&"look_at_photo"));
} }
#[test] #[test]
@@ -4920,6 +5272,7 @@ mod tests {
location_history_present: true, location_history_present: true,
faces_present: true, faces_present: true,
allow_agent_corrections: true, allow_agent_corrections: true,
sms_thread_tools: true,
}; };
let tools = InsightGenerator::build_tool_definitions(opts); let tools = InsightGenerator::build_tool_definitions(opts);
let names: Vec<&str> = tools.iter().map(|t| t.function.name.as_str()).collect(); let names: Vec<&str> = tools.iter().map(|t| t.function.name.as_str()).collect();
@@ -4931,6 +5284,45 @@ mod tests {
assert!(names.contains(&"get_faces_in_photo")); assert!(names.contains(&"get_faces_in_photo"));
assert!(names.contains(&"update_fact")); assert!(names.contains(&"update_fact"));
assert!(names.contains(&"supersede_fact")); assert!(names.contains(&"supersede_fact"));
assert!(names.contains(&"get_thread_messages"));
assert!(names.contains(&"look_at_message_image"));
assert!(names.contains(&"get_photos_near_time"));
assert!(names.contains(&"look_at_photo"));
}
#[test]
fn photo_path_is_safe_rejects_escapes() {
assert!(InsightGenerator::photo_path_is_safe("2019/06/IMG_1.jpg"));
assert!(!InsightGenerator::photo_path_is_safe("../etc/passwd"));
assert!(!InsightGenerator::photo_path_is_safe("2019/../../x.jpg"));
assert!(!InsightGenerator::photo_path_is_safe("/etc/passwd"));
assert!(!InsightGenerator::photo_path_is_safe(""));
}
#[test]
fn parse_moment_accepts_unix_and_local_wall_clock() {
let tz = chrono_tz::Tz::America__New_York;
assert_eq!(
InsightGenerator::parse_moment(&serde_json::json!({"timestamp": 1_700_000_000}), tz),
Some(1_700_000_000)
);
// 2023-11-14 17:13 in New York (EST) = 22:13 UTC.
assert_eq!(
InsightGenerator::parse_moment(
&serde_json::json!({"datetime": "2023-11-14 17:13"}),
tz
),
Some(1_699_999_980)
);
// A bare date means midday, the least-wrong centre for a whole day.
assert_eq!(
InsightGenerator::parse_moment(&serde_json::json!({"datetime": "2023-11-14"}), tz),
Some(1_699_981_200)
);
assert_eq!(
InsightGenerator::parse_moment(&serde_json::json!({}), tz),
None
);
} }
fn place(name: &str, description: &str) -> ApolloPlace { fn place(name: &str, description: &str) -> ApolloPlace {
+1
View File
@@ -15,6 +15,7 @@ pub mod ollama;
pub mod openrouter; pub mod openrouter;
pub mod persona_chat; pub mod persona_chat;
pub mod pronunciation; pub mod pronunciation;
pub mod sms_anchor;
pub mod sms_client; pub mod sms_client;
pub mod tts; pub mod tts;
pub mod turn_registry; pub mod turn_registry;
+182 -7
View File
@@ -59,6 +59,9 @@ pub enum PersonaChatError {
EmptyMessage, EmptyMessage,
MessageTooLong, MessageTooLong,
ConcurrentTurn, ConcurrentTurn,
/// "Continue with persona" couldn't read the SMS thread it was asked to
/// anchor on — SMS-API down, or the message isn't in that thread.
SmsThread(String),
Db(anyhow::Error), Db(anyhow::Error),
} }
@@ -74,6 +77,7 @@ impl std::fmt::Display for PersonaChatError {
PersonaChatError::ConcurrentTurn => { PersonaChatError::ConcurrentTurn => {
write!(f, "another turn for this persona is in flight") write!(f, "another turn for this persona is in flight")
} }
PersonaChatError::SmsThread(e) => write!(f, "could not read the SMS thread: {e}"),
PersonaChatError::Db(e) => write!(f, "database error: {e}"), PersonaChatError::Db(e) => write!(f, "database error: {e}"),
} }
} }
@@ -220,6 +224,36 @@ pub struct PersonaChatHistoryView {
/// Empty for a conversation that has never been rewound. /// Empty for a conversation that has never been rewound.
#[serde(skip_serializing_if = "Vec::is_empty")] #[serde(skip_serializing_if = "Vec::is_empty")]
pub fork_info: Vec<Option<crate::ai::handlers::ChatForkInfo>>, pub fork_info: Vec<Option<crate::ai::handlers::ChatForkInfo>>,
/// Where in an SMS thread this conversation picked up. Absent for an
/// open chat.
#[serde(skip_serializing_if = "Option::is_none")]
pub anchor: Option<PersonaChatAnchorView>,
}
/// Wire shape of a conversation's SMS anchor. Enough for the client to show
/// a "picked up from" banner and to jump back to the message in the thread;
/// the full snapshot stays server-side in the system prompt.
#[derive(Debug, Clone, Serialize)]
pub struct PersonaChatAnchorView {
pub contact_id: i64,
pub contact_name: String,
pub message_id: i64,
/// Unix seconds of the anchor message.
pub timestamp: i64,
/// The anchor message as "Speaker: text", recovered from the snapshot.
pub preview: String,
}
impl From<&crate::database::SmsAnchor> for PersonaChatAnchorView {
fn from(a: &crate::database::SmsAnchor) -> Self {
PersonaChatAnchorView {
contact_id: a.contact_id,
contact_name: a.contact_name.clone(),
message_id: a.message_id,
timestamp: a.timestamp,
preview: crate::ai::sms_anchor::anchor_preview(&a.context),
}
}
} }
/// One rendered line of the persona transcript. Mirrors the file chat's /// One rendered line of the persona transcript. Mirrors the file chat's
@@ -420,6 +454,9 @@ pub struct PersonaConversationSummary {
/// Whether the conversation has been forked — the list marks these so a /// Whether the conversation has been forked — the list marks these so a
/// user can tell which transcripts have alternates parked in them. /// user can tell which transcripts have alternates parked in them.
pub has_branches: bool, pub has_branches: bool,
/// SMS anchor for "Continue with persona" chats; absent for open chats.
#[serde(skip_serializing_if = "Option::is_none")]
pub anchor: Option<PersonaChatAnchorView>,
} }
impl PersonaChatHistoryView { impl PersonaChatHistoryView {
@@ -435,6 +472,7 @@ impl PersonaChatHistoryView {
active_leaf_id: 0, active_leaf_id: 0,
viewing_branch_id: 0, viewing_branch_id: 0,
fork_info: Vec::new(), fork_info: Vec::new(),
anchor: None,
} }
} }
} }
@@ -543,6 +581,67 @@ impl PersonaChatSession {
.map_err(|e| PersonaChatError::Db(anyhow!("{e:?}"))) .map_err(|e| PersonaChatError::Db(anyhow!("{e:?}")))
} }
/// Start a conversation anchored on a message in an SMS thread
/// ("Continue with persona").
///
/// The thread leading up to the anchor is fetched once, here, and stored
/// as a rendered snapshot: every turn splices it into the system prompt,
/// so it has to be byte-stable (KV-cache prefix reuse) and has to
/// exist before the first turn. A thread that can't be read fails the
/// create outright — an anchored chat with no thread is just an open
/// chat with a misleading banner.
pub async fn create_anchored_conversation(
&self,
user_id: i32,
persona_id: &str,
contact_id: i64,
message_id: i64,
) -> Result<String, PersonaChatError> {
let persona_id = persona_id.trim();
{
let mut dao = self.persona_dao.lock().expect("persona_dao poisoned");
let persona = dao
.get_persona(user_id, persona_id)
.map_err(|e| PersonaChatError::Db(anyhow!("{e:?}")))?;
if persona.is_none() {
return Err(PersonaChatError::UnknownPersona(persona_id.to_string()));
}
}
let window = self
.generator
.sms_client()
.fetch_thread_window(
contact_id,
message_id,
crate::ai::sms_anchor::ANCHOR_CONTEXT_BEFORE,
0,
)
.await
.map_err(|e| PersonaChatError::SmsThread(format!("{e}")))?;
let context = crate::ai::sms_anchor::render_thread_window(
&window,
&crate::ai::user_display_name(),
self.generator.display_zone(),
);
let anchor = crate::database::SmsAnchor {
contact_id,
contact_name: window.contact.display_name(),
message_id,
timestamp: window.messages[window.anchor_index].timestamp,
context,
};
let mut dao = self.persona_dao.lock().expect("persona_dao poisoned");
dao.create_anchored_persona_chat(
user_id,
persona_id,
Utc::now().timestamp_millis(),
&anchor,
)
.map_err(|e| PersonaChatError::Db(anyhow!("{e:?}")))
}
/// Delete a conversation and its transcript outright. /// Delete a conversation and its transcript outright.
pub fn delete_conversation(&self, user_id: i32, conversation_id: &str) -> Result<()> { pub fn delete_conversation(&self, user_id: i32, conversation_id: &str) -> Result<()> {
let mut dao = self.persona_dao.lock().expect("persona_dao poisoned"); let mut dao = self.persona_dao.lock().expect("persona_dao poisoned");
@@ -577,6 +676,7 @@ impl PersonaChatSession {
conversation_id: row.conversation_id.clone(), conversation_id: row.conversation_id.clone(),
persona_id: row.persona_id.clone(), persona_id: row.persona_id.clone(),
title: row.title.clone(), title: row.title.clone(),
anchor: row.anchor.as_ref().map(PersonaChatAnchorView::from),
..PersonaChatHistoryView::empty() ..PersonaChatHistoryView::empty()
}; };
// A conversation holding only its seed system node renders as the // A conversation holding only its seed system node renders as the
@@ -748,6 +848,17 @@ impl PersonaChatSession {
let system_prompt = resolve_persona_system_prompt(Some(&persona)) let system_prompt = resolve_persona_system_prompt(Some(&persona))
.unwrap_or_else(|| "You are a helpful assistant.".to_string()); .unwrap_or_else(|| "You are a helpful assistant.".to_string());
// An anchored chat carries the SMS thread in its system prompt,
// re-composed each turn like the persona prompt itself.
let system_prompt = match &row.anchor {
Some(anchor) => crate::ai::sms_anchor::anchored_system_prompt(
&system_prompt,
anchor,
&crate::ai::user_display_name(),
self.generator.display_zone(),
),
None => system_prompt,
};
let stored = Some(decode_store(&row.messages_json)?); let stored = Some(decode_store(&row.messages_json)?);
// `seeded` marks a conversation whose messages are NOT yet backed by // `seeded` marks a conversation whose messages are NOT yet backed by
@@ -818,10 +929,11 @@ impl PersonaChatSession {
// Build a no-photo tool catalog. `current_gate_opts_for_persona` // Build a no-photo tool catalog. `current_gate_opts_for_persona`
// already probes each tool's backing table for presence and // already probes each tool's backing table for presence and
// honours the persona's `allow_agent_corrections` toggle. // honours the persona's `allow_agent_corrections` toggle.
let gate_opts = self.generator.current_gate_opts_for_persona( let mut gate_opts = self.generator.current_gate_opts_for_persona(
backend.images_inline, backend.images_inline,
Some((user_id, &persona.persona_id)), Some((user_id, &persona.persona_id)),
); );
gate_opts.sms_thread_tools = row.anchor.is_some();
let tools = InsightGenerator::build_tool_definitions(gate_opts); let tools = InsightGenerator::build_tool_definitions(gate_opts);
// Append the new user turn + apply the per-turn system-prompt // Append the new user turn + apply the per-turn system-prompt
@@ -981,7 +1093,14 @@ impl PersonaChatSession {
/// 404), so listing them would only offer a dead-end tap target. /// 404), so listing them would only offer a dead-end tap target.
/// Conversations that hold nothing but the seed are omitted too — the /// Conversations that hold nothing but the seed are omitted too — the
/// user has never actually said anything in them. /// user has never actually said anything in them.
pub fn list_conversations(&self, user_id: i32) -> Result<Vec<PersonaConversationSummary>> { ///
/// `contact_id` narrows the list to conversations anchored on that SMS
/// thread — the Messages screen's per-thread persona list.
pub fn list_conversations(
&self,
user_id: i32,
contact_id: Option<i64>,
) -> Result<Vec<PersonaConversationSummary>> {
let (rows, personas) = { let (rows, personas) = {
let mut dao = self.persona_dao.lock().expect("persona_dao poisoned"); let mut dao = self.persona_dao.lock().expect("persona_dao poisoned");
let rows = dao let rows = dao
@@ -999,6 +1118,11 @@ impl PersonaChatSession {
let mut out = Vec::with_capacity(rows.len()); let mut out = Vec::with_capacity(rows.len());
for entry in rows { for entry in rows {
if let Some(cid) = contact_id
&& entry.anchor.as_ref().map(|a| a.contact_id) != Some(cid)
{
continue;
}
let Some(persona_name) = names.get(&entry.persona_id).cloned() else { let Some(persona_name) = names.get(&entry.persona_id).cloned() else {
continue; continue;
}; };
@@ -1036,6 +1160,7 @@ impl PersonaChatSession {
created_at: entry.created_at, created_at: entry.created_at,
updated_at: entry.updated_at, updated_at: entry.updated_at,
has_branches: store.leaves().len() > 1, has_branches: store.leaves().len() > 1,
anchor: entry.anchor.as_ref().map(PersonaChatAnchorView::from),
}); });
} }
Ok(out) Ok(out)
@@ -1232,6 +1357,25 @@ pub struct PersonaChatHistoryQuery {
#[derive(Debug, Deserialize)] #[derive(Debug, Deserialize)]
pub struct PersonaChatCreateRequest { pub struct PersonaChatCreateRequest {
pub persona_id: String, pub persona_id: String,
/// Anchor the conversation on a message in an SMS thread ("Continue
/// with persona"). Omitted for an open chat.
#[serde(default)]
pub anchor: Option<PersonaChatAnchorRequest>,
}
/// SMS-API ids of the thread and message to pick up from.
#[derive(Debug, Deserialize)]
pub struct PersonaChatAnchorRequest {
pub contact_id: i64,
pub message_id: i64,
}
/// Query params for GET /persona_chat/conversations.
#[derive(Debug, Deserialize)]
pub struct PersonaChatListQuery {
/// Only conversations anchored on this SMS thread.
#[serde(default)]
pub contact_id: Option<i64>,
} }
/// GET /persona_chat/history?persona_id=... — the rendered rolling /// GET /persona_chat/history?persona_id=... — the rendered rolling
@@ -1315,6 +1459,7 @@ pub async fn persona_chat_turn_handler(
HttpResponse::BadRequest() HttpResponse::BadRequest()
} }
PersonaChatError::ConcurrentTurn => HttpResponse::Conflict(), PersonaChatError::ConcurrentTurn => HttpResponse::Conflict(),
PersonaChatError::SmsThread(_) => HttpResponse::BadGateway(),
PersonaChatError::Db(_) => { PersonaChatError::Db(_) => {
log::error!("persona chat dispatch failed: {e}"); log::error!("persona chat dispatch failed: {e}");
HttpResponse::InternalServerError() HttpResponse::InternalServerError()
@@ -1335,10 +1480,21 @@ pub async fn persona_chat_create_conversation_handler(
app_state: web::Data<AppState>, app_state: web::Data<AppState>,
) -> impl Responder { ) -> impl Responder {
let user_id = claims.sub.parse::<i32>().unwrap_or(1); let user_id = claims.sub.parse::<i32>().unwrap_or(1);
match app_state let session = &app_state.persona_chat_session;
.persona_chat_session let created = match &request.anchor {
.create_conversation(user_id, &request.persona_id) Some(anchor) => {
{ session
.create_anchored_conversation(
user_id,
&request.persona_id,
anchor.contact_id,
anchor.message_id,
)
.await
}
None => session.create_conversation(user_id, &request.persona_id),
};
match created {
Ok(conversation_id) => HttpResponse::Created().json(serde_json::json!({ Ok(conversation_id) => HttpResponse::Created().json(serde_json::json!({
"conversation_id": conversation_id, "conversation_id": conversation_id,
})), })),
@@ -1346,6 +1502,10 @@ pub async fn persona_chat_create_conversation_handler(
PersonaChatError::UnknownPersona(_) => { PersonaChatError::UnknownPersona(_) => {
HttpResponse::NotFound().json(serde_json::json!({ "error": format!("{e}") })) HttpResponse::NotFound().json(serde_json::json!({ "error": format!("{e}") }))
} }
PersonaChatError::SmsThread(_) => {
log::warn!("anchored persona chat create failed: {e}");
HttpResponse::BadGateway().json(serde_json::json!({ "error": format!("{e}") }))
}
_ => { _ => {
log::error!("persona chat create failed: {e}"); log::error!("persona chat create failed: {e}");
HttpResponse::InternalServerError() HttpResponse::InternalServerError()
@@ -1384,10 +1544,14 @@ pub async fn persona_chat_delete_conversation_handler(
#[get("/persona_chat/conversations")] #[get("/persona_chat/conversations")]
pub async fn persona_chat_conversations_handler( pub async fn persona_chat_conversations_handler(
claims: Claims, claims: Claims,
query: web::Query<PersonaChatListQuery>,
app_state: web::Data<AppState>, app_state: web::Data<AppState>,
) -> impl Responder { ) -> impl Responder {
let user_id = claims.sub.parse::<i32>().unwrap_or(1); let user_id = claims.sub.parse::<i32>().unwrap_or(1);
match app_state.persona_chat_session.list_conversations(user_id) { match app_state
.persona_chat_session
.list_conversations(user_id, query.contact_id)
{
Ok(conversations) => HttpResponse::Ok().json(serde_json::json!({ Ok(conversations) => HttpResponse::Ok().json(serde_json::json!({
"conversations": conversations, "conversations": conversations,
})), })),
@@ -1904,6 +2068,7 @@ mod tests {
total: 2, total: 2,
node_id: 3, node_id: 3,
})], })],
anchor: None,
}; };
let json = serde_json::to_value(&view).unwrap(); let json = serde_json::to_value(&view).unwrap();
// The client's shared ChatHistoryView reads snake_case keys. // The client's shared ChatHistoryView reads snake_case keys.
@@ -2102,8 +2267,18 @@ mod tests {
created_at: 1_600_000_000_000, created_at: 1_600_000_000_000,
updated_at: 1_700_000_000_000, updated_at: 1_700_000_000_000,
has_branches: true, has_branches: true,
anchor: Some(PersonaChatAnchorView {
contact_id: 42,
contact_name: "Sam".to_string(),
message_id: 812,
timestamp: 1_700_000_000,
preview: "Sam: see you at 6".to_string(),
}),
}; };
let json = serde_json::to_value(&summary).unwrap(); let json = serde_json::to_value(&summary).unwrap();
assert_eq!(json["anchor"]["contact_id"], 42);
assert_eq!(json["anchor"]["message_id"], 812);
assert_eq!(json["anchor"]["preview"], "Sam: see you at 6");
assert_eq!(json["conversation_id"], "conv-1"); assert_eq!(json["conversation_id"], "conv-1");
assert_eq!(json["title"], "June recap"); assert_eq!(json["title"], "June recap");
assert_eq!(json["persona_id"], "journal"); assert_eq!(json["persona_id"], "journal");
+303
View File
@@ -0,0 +1,303 @@
//! "Continue with persona" — a persona chat anchored on an SMS thread.
//!
//! The user long-presses a message in the Messages screen and picks a
//! persona; the new persona conversation remembers the thread (contact) and
//! the message it picked up from. This module owns the pure parts of that:
//!
//! - rendering a slice of the thread as a compact transcript the model can
//! read (and that names message ids / attachment indices, so the thread
//! tools can page further back or look at a photo),
//! - the conversational guidance appended to the persona's system prompt.
//!
//! The network side (`SmsApiClient::fetch_thread_window`) and the tools
//! (`get_thread_messages`, `look_at_message_image`, `get_photos_near_time`,
//! `look_at_photo` in `insight_generator.rs`) live with their neighbours.
use chrono::{DateTime, TimeZone};
use chrono_tz::Tz;
use crate::ai::sms_client::{ThreadMessage, ThreadWindow};
use crate::database::SmsAnchor;
/// Messages of history before the anchor captured into the system prompt.
/// Enough to pick up tone and the running topic; the model can page
/// further back with `get_thread_messages`.
pub const ANCHOR_CONTEXT_BEFORE: usize = 40;
/// Per-message cap inside the rendered transcript. A pasted article or a
/// wall of text shouldn't crowd out the rest of the thread.
const LINE_MAX_CHARS: usize = 600;
/// Render one thread message as a transcript line:
/// `[#812 · 2019-06-12 14:02] Sam: see you at 6 (image #0 "IMG_1.jpg")`.
///
/// `#id` and the attachment index are exactly what the thread tools take,
/// so the model can act on any line it reads.
pub fn render_thread_line(
msg: &ThreadMessage,
contact_name: &str,
user_name: &str,
tz: Tz,
) -> String {
let ts = DateTime::from_timestamp(msg.timestamp, 0)
.map(|dt| {
tz.from_utc_datetime(&dt.naive_utc())
.format("%Y-%m-%d %H:%M")
.to_string()
})
.unwrap_or_else(|| "unknown time".to_string());
let speaker = if msg.is_sent { user_name } else { contact_name };
let mut body = msg.body.split_whitespace().collect::<Vec<_>>().join(" ");
if body.chars().count() > LINE_MAX_CHARS {
body = format!("{}…", body.chars().take(LINE_MAX_CHARS).collect::<String>());
}
let attachments: Vec<String> = msg
.attachments
.iter()
.map(|a| {
let kind = a.content_type.split('/').next().unwrap_or("file");
if a.filename.is_empty() {
format!("{kind} #{}", a.media_index)
} else {
format!("{kind} #{} \"{}\"", a.media_index, a.filename)
}
})
.collect();
let mut line = format!("[#{} · {}] {}:", msg.id, ts, speaker);
if !body.is_empty() {
line.push(' ');
line.push_str(&body);
}
if !attachments.is_empty() {
line.push_str(&format!(" ({})", attachments.join(", ")));
}
line
}
/// Render a thread window as a transcript, marking the anchor message with
/// `▶` so the model can tell where the user picked the thread up.
pub fn render_thread_window(window: &ThreadWindow, user_name: &str, tz: Tz) -> String {
let contact_name = window.contact.display_name();
window
.messages
.iter()
.enumerate()
.map(|(i, m)| {
let line = render_thread_line(m, &contact_name, user_name, tz);
if i == window.anchor_index {
format!("▶ {line}")
} else {
line
}
})
.collect::<Vec<_>>()
.join("\n")
}
/// The anchor message's text, recovered from the stored snapshot, for the
/// chat screen's "picked up from" banner and the list row. Reads the `▶`
/// line so no extra column (or SMS-API round-trip) is needed.
pub fn anchor_preview(context: &str) -> String {
context
.lines()
.find_map(|l| l.strip_prefix("▶ "))
.map(|l| match l.split_once("] ") {
Some((_, rest)) => rest.to_string(),
None => l.to_string(),
})
.unwrap_or_default()
}
/// The persona's system prompt with the anchored-thread section appended.
///
/// Re-composed every turn from the persona's current prompt and the stored
/// anchor snapshot, so editing a persona still reaches conversations already
/// in flight (as it does for open chats) while the thread snapshot itself
/// stays byte-stable.
pub fn anchored_system_prompt(
persona_prompt: &str,
anchor: &SmsAnchor,
user_name: &str,
tz: Tz,
) -> String {
let contact = if anchor.contact_name.trim().is_empty() {
"someone"
} else {
anchor.contact_name.as_str()
};
let when = DateTime::from_timestamp(anchor.timestamp, 0)
.map(|dt| {
tz.from_utc_datetime(&dt.naive_utc())
.format("%A, %B %-d, %Y at %-I:%M %p")
.to_string()
})
.unwrap_or_else(|| "an unknown date".to_string());
let thread = if anchor.context.trim().is_empty() {
"(The thread could not be loaded — use get_thread_messages to read it.)".to_string()
} else {
anchor.context.clone()
};
let cid = anchor.contact_id;
let mid = anchor.message_id;
format!(
"{persona_prompt}
## Picking up a text thread
{user_name} opened this chat from their text conversation with {contact}. \
Below is that thread up to the message they picked (marked ▶, sent {when}). \
They want to keep talking with you from that point — about what was going on, \
what came of it, or just to carry on the conversation.
How to talk here:
- Keep your own voice and values from above, but this is a chat, not a \
write-up. Reply like a text message: short, warm, one or two thoughts at a \
time. No headings, no bullet lists, no summaries unless asked.
- Match the thread's register — its slang, emoji and punctuation habits, how \
long its messages run. If the thread is playful, be playful.
- You are not {contact}. Never write as them or put words in their mouth; you \
are someone {user_name} trusts who has just read the thread over their shoulder.
- Lead with specifics from the thread — names, plans, inside jokes, open \
questions. Don't recap the thread back to {user_name}.
- The thread is history: today's date is not {when}. If something was left \
unresolved, it's natural to ask how it turned out.
- Don't invent details. When you need more than the thread below shows, look \
it up with your tools first, then answer in your own words — don't narrate \
the lookups. Look at one or two photos that matter, not every one you find.
- You can read and look, but you can't send photos or texts to anyone — \
don't offer to.
Tools for this thread (contact_id {cid}; #numbers below are message ids):
- get_thread_messages {{contact_id: {cid}, message_id: {mid}, direction: \"before\"}} \
— earlier messages; \"after\" for what came next.
- look_at_message_image {{contact_id: {cid}, message_id, media_index}} — see a \
photo sent in the thread (attachments are listed as `image #N`).
- get_photos_near_time {{timestamp}} — photos from {user_name}'s library taken \
around a moment, with their tags and the people in them.
- look_at_photo {{file_path}} — see one of those library photos.
Thread with {contact}:
{thread}"
)
}
#[cfg(test)]
mod tests {
use super::*;
use crate::ai::sms_client::{ThreadAttachment, ThreadContact};
fn msg(id: i64, body: &str, sent: bool) -> ThreadMessage {
ThreadMessage {
id,
body: body.to_string(),
// 2023-11-15 03:13 UTC = 2023-11-14 22:13 in New York.
timestamp: 1_700_018_000 - 1_000 + id,
is_sent: sent,
attachments: Vec::new(),
}
}
#[test]
fn line_names_the_speaker_id_and_local_time() {
let line = render_thread_line(
&msg(1000, "on my way", true),
"Sam",
"Cam",
Tz::America__New_York,
);
assert_eq!(line, "[#1000 · 2023-11-14 22:13] Cam: on my way");
let line = render_thread_line(&msg(1000, "ok", false), "Sam", "Cam", Tz::America__New_York);
assert!(line.ends_with("] Sam: ok"), "{line}");
}
#[test]
fn line_lists_attachments_with_their_media_index() {
let mut m = msg(5, "", false);
m.attachments = vec![
ThreadAttachment {
media_index: 0,
content_type: "image/jpeg".to_string(),
filename: "IMG_1.jpg".to_string(),
},
ThreadAttachment {
media_index: 1,
content_type: "video/mp4".to_string(),
filename: String::new(),
},
];
let line = render_thread_line(&m, "Sam", "Cam", Tz::UTC);
assert!(
line.ends_with("Sam: (image #0 \"IMG_1.jpg\", video #1)"),
"an empty body leaves just the attachments: {line}"
);
}
#[test]
fn line_flattens_and_caps_long_bodies() {
let long = format!("a\n\nb {}", "x".repeat(2000));
let line = render_thread_line(&msg(1, &long, false), "Sam", "Cam", Tz::UTC);
assert!(!line.contains('\n'));
assert!(line.ends_with('…'));
assert!(line.chars().count() < LINE_MAX_CHARS + 60);
}
#[test]
fn window_marks_only_the_anchor() {
let window = ThreadWindow {
contact: ThreadContact {
id: 3,
name: "Sam".to_string(),
address: String::new(),
},
messages: vec![msg(1, "a", false), msg(2, "b", true), msg(3, "c", false)],
anchor_index: 1,
};
let rendered = render_thread_window(&window, "Cam", Tz::UTC);
let lines: Vec<&str> = rendered.lines().collect();
assert_eq!(lines.len(), 3);
assert!(!lines[0].starts_with('▶'));
assert!(lines[1].starts_with("▶ [#2"));
assert!(!lines[2].starts_with('▶'));
}
#[test]
fn anchor_preview_reads_the_marked_line() {
let ctx = "[#1 · 2023-11-14 22:10] Cam: hi\n▶ [#2 · 2023-11-14 22:13] Sam: see you at 6\n";
assert_eq!(anchor_preview(ctx), "Sam: see you at 6");
assert_eq!(anchor_preview("no marker"), "");
}
#[test]
fn system_prompt_keeps_the_persona_first_and_carries_the_ids() {
let anchor = SmsAnchor {
contact_id: 42,
contact_name: "Sam".to_string(),
message_id: 812,
timestamp: 1_700_000_000,
context: "[#812 · …] Sam: see you there".to_string(),
};
let prompt = anchored_system_prompt("You are Journal.", &anchor, "Cam", Tz::UTC);
assert!(prompt.starts_with("You are Journal."));
assert!(prompt.contains("contact_id 42"));
assert!(prompt.contains("message_id: 812"));
assert!(prompt.contains("You are not Sam"));
assert!(prompt.ends_with("Sam: see you there"));
}
#[test]
fn system_prompt_survives_a_missing_snapshot() {
let anchor = SmsAnchor {
contact_id: 42,
contact_name: String::new(),
message_id: 812,
timestamp: 0,
context: String::new(),
};
let prompt = anchored_system_prompt("P", &anchor, "Cam", Tz::UTC);
assert!(prompt.contains("could not be loaded"));
assert!(prompt.contains("conversation with someone"));
}
}
+306
View File
@@ -320,6 +320,138 @@ impl SmsApiClient {
Ok(data.results) Ok(data.results)
} }
/// One page of a contact's thread from SMS-API's conversation endpoint,
/// newest first. With `message_id`, SMS-API ignores `page` and returns
/// the page that contains that message.
pub async fn fetch_thread_page(
&self,
contact_id: i64,
message_id: Option<i64>,
page: usize,
page_size: usize,
) -> Result<ThreadPage> {
let mut url = format!(
"{}/api/conversations/{}/?page={}&page_size={}",
self.base_url, contact_id, page, page_size
);
if let Some(mid) = message_id {
url.push_str(&format!("&message_id={}", mid));
}
let mut request = self.client.get(&url);
if let Some(token) = &self.token {
request = request.header("Authorization", format!("Bearer {}", token));
}
let response = request.send().await?;
if !response.status().is_success() {
let status = response.status();
let body = response.text().await.unwrap_or_default();
return Err(anyhow::anyhow!(
"SMS conversation request failed: {} - {}",
status,
body
));
}
let raw: RawThreadPage = response.json().await?;
Ok(raw.into())
}
/// The slice of a thread around `anchor_message_id`: up to `before`
/// messages older than the anchor, the anchor itself, and up to `after`
/// newer ones — oldest first, ready to render as a transcript.
///
/// SMS-API pages are fixed-size, so the anchor can sit at either edge of
/// the page that holds it; the neighbouring page is pulled in whenever
/// the requested side runs short.
pub async fn fetch_thread_window(
&self,
contact_id: i64,
anchor_message_id: i64,
before: usize,
after: usize,
) -> Result<ThreadWindow> {
let page_size = (before + after + 1).clamp(20, 200);
let first = self
.fetch_thread_page(contact_id, Some(anchor_message_id), 1, page_size)
.await?;
let contact = first.contact.clone();
let Some(idx) = first
.messages
.iter()
.position(|m| m.id == anchor_message_id)
else {
return Err(anyhow::anyhow!(
"message {} not found in conversation {}",
anchor_message_id,
contact_id
));
};
// Newest first: index < idx is newer, > idx is older.
let mut newer: Vec<ThreadMessage> = first.messages[..idx].to_vec();
let anchor = first.messages[idx].clone();
let mut older: Vec<ThreadMessage> = first.messages[idx + 1..].to_vec();
if older.len() < before && first.page < first.total_pages {
let next = self
.fetch_thread_page(contact_id, None, first.page + 1, page_size)
.await?;
older.extend(next.messages);
}
if newer.len() < after && first.page > 1 {
let prev = self
.fetch_thread_page(contact_id, None, first.page - 1, page_size)
.await?;
// The previous page is entirely newer; keep it newest-first so
// the closest messages stay at the end of `newer`.
let mut combined = prev.messages;
combined.extend(newer);
newer = combined;
}
older.truncate(before);
older.reverse();
let newer_start = newer.len().saturating_sub(after);
let mut newer: Vec<ThreadMessage> = newer[newer_start..].to_vec();
newer.reverse();
let mut messages = older;
let anchor_index = messages.len();
messages.push(anchor);
messages.extend(newer);
Ok(ThreadWindow {
contact,
messages,
anchor_index,
})
}
/// Download one MMS attachment. `media_index` is the attachment's
/// position among the message's image/video/audio parts — the same
/// index SMS-API's `/media/{message}/{index}/file/` route takes.
pub async fn fetch_mms_media(
&self,
contact_id: i64,
message_id: i64,
media_index: usize,
) -> Result<Vec<u8>> {
let url = format!(
"{}/api/conversations/{}/media/{}/{}/file/",
self.base_url, contact_id, message_id, media_index
);
let mut request = self.client.get(&url);
if let Some(token) = &self.token {
request = request.header("Authorization", format!("Bearer {}", token));
}
let response = request.send().await?;
if !response.status().is_success() {
return Err(anyhow::anyhow!(
"MMS media request failed: {}",
response.status()
));
}
Ok(response.bytes().await?.to_vec())
}
pub async fn summarize_context( pub async fn summarize_context(
&self, &self,
messages: &[SmsMessage], messages: &[SmsMessage],
@@ -372,6 +504,140 @@ pub struct SmsMessage {
pub is_sent: bool, pub is_sent: bool,
} }
/// A thread participant as SMS-API's conversation endpoint describes it.
#[derive(Debug, Clone, Deserialize)]
pub struct ThreadContact {
pub id: i64,
#[serde(default)]
pub name: String,
#[serde(default)]
pub address: String,
}
impl ThreadContact {
/// Name to show for the other side of the thread — the stored name, or
/// the phone number for a contact that was never named.
pub fn display_name(&self) -> String {
if self.name.trim().is_empty() {
self.address.clone()
} else {
self.name.clone()
}
}
}
/// One message of a thread. `attachments` holds only the media parts
/// (image/video/audio); MMS text parts are folded into `body`.
#[derive(Debug, Clone)]
pub struct ThreadMessage {
pub id: i64,
pub body: String,
/// Unix seconds.
pub timestamp: i64,
pub is_sent: bool,
pub attachments: Vec<ThreadAttachment>,
}
#[derive(Debug, Clone, PartialEq)]
pub struct ThreadAttachment {
/// Position among the message's media parts — the index
/// `fetch_mms_media` takes.
pub media_index: usize,
pub content_type: String,
pub filename: String,
}
#[derive(Debug, Clone)]
pub struct ThreadPage {
pub contact: ThreadContact,
/// Newest first, as SMS-API returns them.
pub messages: Vec<ThreadMessage>,
pub page: usize,
pub total_pages: usize,
}
/// A slice of a thread around an anchor message, oldest first.
#[derive(Debug, Clone)]
pub struct ThreadWindow {
pub contact: ThreadContact,
pub messages: Vec<ThreadMessage>,
/// Index of the anchor message within `messages`.
pub anchor_index: usize,
}
#[derive(Deserialize)]
struct RawThreadPage {
contact: ThreadContact,
messages: Vec<RawThreadMessage>,
page: usize,
total_pages: usize,
}
#[derive(Deserialize)]
struct RawThreadMessage {
id: i64,
#[serde(default)]
body: String,
date: i64,
#[serde(rename = "type")]
type_: i32,
#[serde(default)]
parts: Vec<RawThreadPart>,
}
#[derive(Deserialize)]
struct RawThreadPart {
content_type: String,
#[serde(default)]
filename: String,
#[serde(default)]
text: String,
}
impl From<RawThreadPage> for ThreadPage {
fn from(raw: RawThreadPage) -> Self {
ThreadPage {
contact: raw.contact,
messages: raw.messages.into_iter().map(ThreadMessage::from).collect(),
page: raw.page,
total_pages: raw.total_pages,
}
}
}
impl From<RawThreadMessage> for ThreadMessage {
/// Media indices are counted the way SMS-API's media routes count them:
/// SMIL layout parts are skipped and only image/video/audio parts take
/// an index, in part order.
fn from(raw: RawThreadMessage) -> Self {
let mut attachments = Vec::new();
let mut body = raw.body;
for part in raw.parts {
let ct = part.content_type.to_ascii_lowercase();
if ct == "application/smil" {
continue;
}
if ct.starts_with("image/") || ct.starts_with("video/") || ct.starts_with("audio/") {
attachments.push(ThreadAttachment {
media_index: attachments.len(),
content_type: part.content_type,
filename: part.filename,
});
} else if ct == "text/plain" && body.trim().is_empty() && !part.text.trim().is_empty() {
// MMS keeps its caption in a text part, leaving `body` empty.
body = part.text;
}
}
ThreadMessage {
id: raw.id,
body,
timestamp: raw.date,
is_sent: raw.type_ == 2,
attachments,
}
}
}
#[derive(Deserialize)] #[derive(Deserialize)]
struct SmsApiResponse { struct SmsApiResponse {
messages: Vec<SmsApiMessage>, messages: Vec<SmsApiMessage>,
@@ -451,6 +717,46 @@ mod tests {
assert_eq!(end - 7 * 86400, center); assert_eq!(end - 7 * 86400, center);
} }
#[test]
fn thread_message_indexes_media_parts_like_sms_api() {
// SMIL is skipped and the text part only fills an empty body, so
// the image and video take media indices 0 and 1 — matching the
// `/media/{id}/{index}/file/` route.
let raw: RawThreadMessage = serde_json::from_value(serde_json::json!({
"id": 7,
"body": "",
"date": 1_700_000_000,
"type": 2,
"parts": [
{ "content_type": "application/smil", "filename": "smil.xml" },
{ "content_type": "image/jpeg", "filename": "IMG_1.jpg" },
{ "content_type": "text/plain", "text": "look at this" },
{ "content_type": "video/mp4", "filename": "clip.mp4" }
]
}))
.unwrap();
let msg = ThreadMessage::from(raw);
assert!(msg.is_sent);
assert_eq!(msg.body, "look at this");
assert_eq!(
msg.attachments
.iter()
.map(|a| (a.media_index, a.filename.as_str()))
.collect::<Vec<_>>(),
vec![(0, "IMG_1.jpg"), (1, "clip.mp4")]
);
}
#[test]
fn thread_contact_falls_back_to_the_address() {
let c = ThreadContact {
id: 1,
name: " ".to_string(),
address: "+15551234".to_string(),
};
assert_eq!(c.display_name(), "+15551234");
}
#[test] #[test]
fn window_for_radius_clamps_zero_to_one() { fn window_for_radius_clamps_zero_to_one() {
let (start, end) = SmsApiClient::window_for_radius(100_000, 0); let (start, end) = SmsApiClient::window_for_radius(100_000, 0);
+1 -1
View File
@@ -64,7 +64,7 @@ pub use knowledge_dao::{
KnowledgeDao, PersonaFilter, RecentActivity, SqliteKnowledgeDao, KnowledgeDao, PersonaFilter, RecentActivity, SqliteKnowledgeDao,
}; };
pub use location_dao::{LocationHistoryDao, SqliteLocationHistoryDao}; pub use location_dao::{LocationHistoryDao, SqliteLocationHistoryDao};
pub use persona_dao::{ImportPersona, PersonaDao, PersonaPatch, SqlitePersonaDao}; pub use persona_dao::{ImportPersona, PersonaDao, PersonaPatch, SmsAnchor, SqlitePersonaDao};
pub use precomputed_reel_dao::{PrecomputedReelDao, SqlitePrecomputedReelDao}; pub use precomputed_reel_dao::{PrecomputedReelDao, SqlitePrecomputedReelDao};
pub use preview_dao::{PreviewDao, SqlitePreviewDao}; pub use preview_dao::{PreviewDao, SqlitePreviewDao};
pub use search_dao::{SearchHistoryDao, SqliteSearchHistoryDao}; pub use search_dao::{SearchHistoryDao, SqliteSearchHistoryDao};
+160 -31
View File
@@ -29,21 +29,92 @@ pub struct PersonaChatRow {
pub turn_count: i32, pub turn_count: i32,
pub created_at: i64, pub created_at: i64,
pub updated_at: i64, pub updated_at: i64,
/// The SMS thread position this conversation grew out of, for chats
/// started via "Continue with persona". None for an open chat.
pub anchor: Option<SmsAnchor>,
} }
/// Column tuple → `PersonaChatRow`. Shared by the single-row and list /// Where in an SMS thread an anchored persona conversation picks up.
/// queries so their `select(...)` orders can never drift apart. ///
/// `contact_id` / `message_id` are SMS-API ids. `context` is the rendered
/// snapshot of the thread leading up to the anchor, captured once at
/// creation and spliced into the system prompt on every turn.
#[derive(Clone, Debug, PartialEq)]
pub struct SmsAnchor {
pub contact_id: i64,
pub contact_name: String,
pub message_id: i64,
/// Unix seconds of the anchor message.
pub timestamp: i64,
pub context: String,
}
/// The column tuple every persona-chat read selects. Shared by the
/// single-row and list queries so their `select(...)` orders can never
/// drift apart.
type PersonaChatColumns = (
String,
String,
String,
String,
i32,
i64,
i64,
Option<i64>,
Option<String>,
Option<i64>,
Option<i64>,
Option<String>,
);
macro_rules! persona_chat_select {
() => {
(
conversation_id,
persona_id,
title,
messages_json,
turn_count,
created_at,
updated_at,
anchor_contact_id,
anchor_contact_name,
anchor_message_id,
anchor_ts,
anchor_context,
)
};
}
/// Column tuple → `PersonaChatRow`.
fn persona_chat_row( fn persona_chat_row(
(conversation_id, persona_id, title, messages_json, turn_count, created_at, updated_at): ( (
String, conversation_id,
String, persona_id,
String, title,
String, messages_json,
i32, turn_count,
i64, created_at,
i64, updated_at,
), anchor_contact_id,
anchor_contact_name,
anchor_message_id,
anchor_ts,
anchor_context,
): PersonaChatColumns,
) -> PersonaChatRow { ) -> PersonaChatRow {
// An anchor needs at least the thread and the message; the rest has a
// usable default if a row was written half-way by hand.
let anchor = match (anchor_contact_id, anchor_message_id) {
(Some(contact_id), Some(message_id)) => Some(SmsAnchor {
contact_id,
contact_name: anchor_contact_name.unwrap_or_default(),
message_id,
timestamp: anchor_ts.unwrap_or(0),
context: anchor_context.unwrap_or_default(),
}),
_ => None,
};
PersonaChatRow { PersonaChatRow {
conversation_id, conversation_id,
persona_id, persona_id,
@@ -52,6 +123,7 @@ fn persona_chat_row(
turn_count, turn_count,
created_at, created_at,
updated_at, updated_at,
anchor,
} }
} }
@@ -135,6 +207,17 @@ pub trait PersonaDao: Sync + Send {
created_at: i64, created_at: i64,
) -> Result<String, DbError>; ) -> Result<String, DbError>;
/// Start a new conversation anchored on a point in an SMS thread
/// ("Continue with persona"). Same as `create_persona_chat` plus the
/// anchor columns.
fn create_anchored_persona_chat(
&mut self,
user_id: i32,
persona_id: &str,
created_at: i64,
anchor: &SmsAnchor,
) -> Result<String, DbError>;
/// Replace a conversation's transcript. Called once per completed turn /// Replace a conversation's transcript. Called once per completed turn
/// with the full new `messages_json`; `turn_count` is the cumulative /// with the full new `messages_json`; `turn_count` is the cumulative
/// assistant-turn count on the resulting active branch. Returns the /// assistant-turn count on the resulting active branch. Returns the
@@ -382,16 +465,8 @@ impl PersonaDao for SqlitePersonaDao {
// Scoped by user as well as id: the id is a bearer token for // Scoped by user as well as id: the id is a bearer token for
// someone's transcript, so it never grants access on its own. // someone's transcript, so it never grants access on its own.
.filter(user_id.eq(uid)) .filter(user_id.eq(uid))
.select(( .select(persona_chat_select!())
conversation_id, .first::<PersonaChatColumns>(conn.deref_mut())
persona_id,
title,
messages_json,
turn_count,
created_at,
updated_at,
))
.first::<(String, String, String, String, i32, i64, i64)>(conn.deref_mut())
.optional() .optional()
.map(|opt| opt.map(persona_chat_row)) .map(|opt| opt.map(persona_chat_row))
.map_err(|e| anyhow::anyhow!("Query error: {}", e)) .map_err(|e| anyhow::anyhow!("Query error: {}", e))
@@ -406,16 +481,8 @@ impl PersonaDao for SqlitePersonaDao {
persona_chat_conversations persona_chat_conversations
.filter(user_id.eq(uid)) .filter(user_id.eq(uid))
.order(updated_at.desc()) .order(updated_at.desc())
.select(( .select(persona_chat_select!())
conversation_id, .load::<PersonaChatColumns>(conn.deref_mut())
persona_id,
title,
messages_json,
turn_count,
created_at,
updated_at,
))
.load::<(String, String, String, String, i32, i64, i64)>(conn.deref_mut())
.map(|rows| rows.into_iter().map(persona_chat_row).collect()) .map(|rows| rows.into_iter().map(persona_chat_row).collect())
.map_err(|e| anyhow::anyhow!("Query error: {}", e)) .map_err(|e| anyhow::anyhow!("Query error: {}", e))
}) })
@@ -452,6 +519,40 @@ impl PersonaDao for SqlitePersonaDao {
.map_err(|e| DbError::log(DbErrorKind::InsertError, e)) .map_err(|e| DbError::log(DbErrorKind::InsertError, e))
} }
fn create_anchored_persona_chat(
&mut self,
uid: i32,
pid: &str,
created: i64,
anchor: &SmsAnchor,
) -> Result<String, DbError> {
trace_db_call("insert", "create_anchored_persona_chat", |_span| {
use schema::persona_chat_conversations::dsl::*;
let mut conn = self.connection.lock().expect("PersonaDao lock");
let new_id = uuid::Uuid::new_v4().to_string();
diesel::insert_into(persona_chat_conversations)
.values((
conversation_id.eq(&new_id),
user_id.eq(uid),
persona_id.eq(pid),
title.eq(""),
messages_json.eq("[]"),
turn_count.eq(0),
created_at.eq(created),
updated_at.eq(created),
anchor_contact_id.eq(Some(anchor.contact_id)),
anchor_contact_name.eq(Some(&anchor.contact_name)),
anchor_message_id.eq(Some(anchor.message_id)),
anchor_ts.eq(Some(anchor.timestamp)),
anchor_context.eq(Some(&anchor.context)),
))
.execute(conn.deref_mut())
.map_err(|e| anyhow::anyhow!("Insert error: {}", e))?;
Ok(new_id)
})
.map_err(|e| DbError::log(DbErrorKind::InsertError, e))
}
fn update_persona_chat( fn update_persona_chat(
&mut self, &mut self,
uid: i32, uid: i32,
@@ -766,6 +867,34 @@ mod tests {
); );
} }
#[test]
fn persona_chat_anchor_round_trips_and_open_chats_have_none() {
let (mut dao, uid) = dao_with_user("p10");
let anchor = SmsAnchor {
contact_id: 42,
contact_name: "Sam".to_string(),
message_id: 9001,
timestamp: 1_700_000_000,
context: "[2023-11-14 22:13] Sam: see you there".to_string(),
};
let anchored = dao
.create_anchored_persona_chat(uid, "journal", 100, &anchor)
.unwrap();
let open = dao.create_persona_chat(uid, "journal", 200).unwrap();
let row = dao.get_persona_chat(uid, &anchored).unwrap().unwrap();
assert_eq!(row.anchor, Some(anchor));
let row = dao.get_persona_chat(uid, &open).unwrap().unwrap();
assert_eq!(row.anchor, None);
assert!(
dao.list_persona_chats(uid)
.unwrap()
.iter()
.any(|r| r.anchor.is_some()),
"the list query carries the anchor too"
);
}
#[test] #[test]
fn persona_chat_list_is_empty_for_a_user_who_has_never_chatted() { fn persona_chat_list_is_empty_for_a_user_who_has_never_chatted() {
let (mut dao, uid) = dao_with_user("p8"); let (mut dao, uid) = dao_with_user("p8");
+5
View File
@@ -197,6 +197,11 @@ diesel::table! {
turn_count -> Integer, turn_count -> Integer,
created_at -> BigInt, created_at -> BigInt,
updated_at -> BigInt, updated_at -> BigInt,
anchor_contact_id -> Nullable<BigInt>,
anchor_contact_name -> Nullable<Text>,
anchor_message_id -> Nullable<BigInt>,
anchor_ts -> Nullable<BigInt>,
anchor_context -> Nullable<Text>,
} }
} }