diff --git a/app/src/pages/Conversations.tsx b/app/src/pages/Conversations.tsx index a468cf0ee..6a5ba541e 100644 --- a/app/src/pages/Conversations.tsx +++ b/app/src/pages/Conversations.tsx @@ -1,10 +1,11 @@ import { convertFileSrc } from '@tauri-apps/api/core'; -import { useEffect, useRef, useState } from 'react'; +import { useEffect, useMemo, useRef, useState } from 'react'; import { useLocation, useNavigate } from 'react-router-dom'; import { type ChatSendError, chatSendError } from '../chat/chatSendError'; import TokenUsagePill from '../components/chat/TokenUsagePill'; import { ConfirmationModal } from '../components/intelligence/ConfirmationModal'; +import PillTabBar from '../components/PillTabBar'; import UpsellBanner from '../components/upsell/UpsellBanner'; import { dismissBanner, shouldShowBanner } from '../components/upsell/upsellDismissState'; import UsageLimitModal from '../components/upsell/UsageLimitModal'; @@ -151,6 +152,7 @@ const Conversations = ({ variant = 'page' }: ConversationsProps = {}) => { const [isTranscribing, setIsTranscribing] = useState(false); const [voiceStatus, setVoiceStatus] = useState(null); const [isPlayingReply, setIsPlayingReply] = useState(false); + const [selectedLabel, setSelectedLabel] = useState('all'); const [inlineSuggestionValue, setInlineSuggestionValue] = useState(''); const [sendError, setSendError] = useState(null); const socketStatus = useAppSelector(selectSocketStatus); @@ -792,9 +794,38 @@ const Conversations = ({ variant = 'page' }: ConversationsProps = {}) => { const shouldRenderTimelineBeforeLatestAgentMessage = selectedThreadToolTimeline.length > 0 && !isSending && Boolean(latestVisibleAgentMessage); - const sortedThreads = [...threads].sort( - (a, b) => new Date(b.lastMessageAt).getTime() - new Date(a.lastMessageAt).getTime() - ); + const filteredThreads = useMemo(() => { + return threads.filter(t => { + if (selectedLabel === 'all') return true; + return t.labels?.includes(selectedLabel); + }); + }, [threads, selectedLabel]); + + const sortedThreads = useMemo(() => { + return [...filteredThreads].sort( + (a, b) => new Date(b.lastMessageAt).getTime() - new Date(a.lastMessageAt).getTime() + ); + }, [filteredThreads]); + + const allLabels = useMemo(() => { + return Array.from(new Set(threads.flatMap(t => t.labels ?? []))).sort(); + }, [threads]); + + // Fixed tab set so categories don't disappear when empty and the active + // filter state remains unambiguous regardless of what threads exist. + const labelTabs = [ + { label: 'All', value: 'all' }, + { label: 'Work', value: 'work' }, + { label: 'Briefing', value: 'briefing' }, + { label: 'Notification', value: 'notification' }, + ]; + + // Reset stale selectedLabel when the last thread carrying that label is deleted. + useEffect(() => { + if (selectedLabel !== 'all' && !allLabels.includes(selectedLabel)) { + setSelectedLabel('all'); + } + }, [allLabels, selectedLabel]); const isSidebar = variant === 'sidebar'; @@ -827,9 +858,19 @@ const Conversations = ({ variant = 'page' }: ConversationsProps = {}) => { +
+ +
{sortedThreads.length === 0 ? ( -

No threads yet

+

+ {selectedLabel === 'all' ? 'No threads yet' : `No "${selectedLabel}" threads`} +

) : ( sortedThreads.map(thread => (
(response: Envelope | T): T { const generateTitleLog = debug('threadApi.generateTitleIfNeeded'); export const threadApi = { - createNewThread: async (): Promise => { + createNewThread: async (labels?: string[]): Promise => { const response = await callCoreRpc>({ method: 'openhuman.threads_create_new', + params: { labels }, }); return unwrapEnvelope(response); }, @@ -101,4 +102,12 @@ export const threadApi = { }); return unwrapEnvelope(response); }, + + updateLabels: async (threadId: string, labels: string[]): Promise => { + const response = await callCoreRpc>({ + method: 'openhuman.threads_update_labels', + params: { thread_id: threadId, labels }, + }); + return unwrapEnvelope(response); + }, }; diff --git a/app/src/store/__tests__/threadSlice.test.ts b/app/src/store/__tests__/threadSlice.test.ts index 24a6b4e33..d47db2026 100644 --- a/app/src/store/__tests__/threadSlice.test.ts +++ b/app/src/store/__tests__/threadSlice.test.ts @@ -41,6 +41,7 @@ function makeThread(overrides: Partial = {}): Thread { messageCount: 0, lastMessageAt: '2026-01-01T00:00:00.000Z', createdAt: '2026-01-01T00:00:00.000Z', + labels: [], ...overrides, }; } diff --git a/app/src/store/threadSlice.ts b/app/src/store/threadSlice.ts index 9ad8e17e3..d5d86adbd 100644 --- a/app/src/store/threadSlice.ts +++ b/app/src/store/threadSlice.ts @@ -69,9 +69,9 @@ export const loadThreads = createAsyncThunk( export const createNewThread = createAsyncThunk( 'thread/createNewThread', - async (_, { dispatch, rejectWithValue }) => { + async (labels: string[] | undefined, { dispatch, rejectWithValue }) => { try { - const thread = await threadApi.createNewThread(); + const thread = await threadApi.createNewThread(labels); await dispatch(loadThreads()).unwrap(); return thread; } catch (error) { @@ -221,6 +221,21 @@ export const persistReaction = createAsyncThunk( } ); +export const updateThreadLabels = createAsyncThunk( + 'thread/updateThreadLabels', + async (payload: { threadId: string; labels: string[] }, { dispatch, rejectWithValue }) => { + try { + const thread = await threadApi.updateLabels(payload.threadId, payload.labels); + await dispatch(loadThreads()).unwrap(); + return thread; + } catch (error) { + return rejectWithValue( + error instanceof Error ? error.message : 'Failed to update thread labels' + ); + } + } +); + export const purgeThreads = createAsyncThunk( 'thread/purgeThreads', async (_, { dispatch, rejectWithValue }) => { diff --git a/app/src/types/thread.ts b/app/src/types/thread.ts index b1e20bf38..ded5c68d9 100644 --- a/app/src/types/thread.ts +++ b/app/src/types/thread.ts @@ -6,6 +6,7 @@ export interface Thread { messageCount: number; lastMessageAt: string; createdAt: string; + labels: string[]; } export interface ThreadMessage { diff --git a/src/openhuman/about_app/catalog.rs b/src/openhuman/about_app/catalog.rs index 57cdbcb24..1085da5d2 100644 --- a/src/openhuman/about_app/catalog.rs +++ b/src/openhuman/about_app/catalog.rs @@ -116,6 +116,16 @@ const CAPABILITIES: &[Capability] = &[ status: CapabilityStatus::Beta, privacy: None, }, + Capability { + id: "conversation.label_filter", + name: "Thread Label Filters", + domain: "conversation", + category: CapabilityCategory::Conversation, + description: "Filter the thread list by label (Work, Briefing, Notification) using the tab bar at the top of the thread list.", + how_to: "Conversations > Label tabs", + status: CapabilityStatus::Beta, + privacy: None, + }, Capability { id: "intelligence.analyze_actionable_items", name: "Analyze Actionable Items", diff --git a/src/openhuman/memory/conversations/bus.rs b/src/openhuman/memory/conversations/bus.rs index 0b4697da1..4681ba9cb 100644 --- a/src/openhuman/memory/conversations/bus.rs +++ b/src/openhuman/memory/conversations/bus.rs @@ -169,6 +169,7 @@ fn persist_channel_turn( id: thread_id.clone(), title, created_at: created_at.clone(), + labels: Some(vec!["work".to_string()]), }, )?; diff --git a/src/openhuman/memory/conversations/mod.rs b/src/openhuman/memory/conversations/mod.rs index 3b09f93f8..181026cf3 100644 --- a/src/openhuman/memory/conversations/mod.rs +++ b/src/openhuman/memory/conversations/mod.rs @@ -11,7 +11,8 @@ mod types; pub use bus::register_conversation_persistence_subscriber; pub use store::{ append_message, delete_thread, ensure_thread, get_messages, list_threads, purge_threads, - update_message, update_thread_title, ConversationPurgeStats, ConversationStore, + update_message, update_thread_labels, update_thread_title, ConversationPurgeStats, + ConversationStore, }; pub use types::{ ConversationMessage, ConversationMessagePatch, ConversationThread, CreateConversationThread, diff --git a/src/openhuman/memory/conversations/store.rs b/src/openhuman/memory/conversations/store.rs index 05a508a0d..88599084f 100644 --- a/src/openhuman/memory/conversations/store.rs +++ b/src/openhuman/memory/conversations/store.rs @@ -47,6 +47,8 @@ enum ThreadLogEntry { title: String, created_at: String, updated_at: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + labels: Option>, }, Delete { thread_id: String, @@ -74,6 +76,7 @@ impl ConversationStore { title: request.title.clone(), created_at: request.created_at.clone(), updated_at: now, + labels: request.labels.clone(), }, )?; debug!( @@ -141,6 +144,7 @@ impl ConversationStore { title: title.to_string(), created_at: entry.created_at.clone(), updated_at: updated_at.to_string(), + labels: Some(entry.labels.clone()), }, )?; debug!( @@ -153,6 +157,37 @@ impl ConversationStore { .ok_or_else(|| format!("thread {} missing after title update", thread_id)) } + pub fn update_thread_labels( + &self, + thread_id: &str, + labels: Vec, + updated_at: &str, + ) -> Result { + let _guard = CONVERSATION_STORE_LOCK.lock(); + let index = self.thread_index_unlocked()?; + let entry = index + .get(thread_id) + .ok_or_else(|| format!("thread {} does not exist", thread_id))?; + let threads_path = self.ensure_root()?.join(THREADS_FILENAME); + append_jsonl( + &threads_path, + &ThreadLogEntry::Upsert { + thread_id: thread_id.to_string(), + title: entry.title.clone(), + created_at: entry.created_at.clone(), + updated_at: updated_at.to_string(), + labels: Some(labels), + }, + )?; + debug!( + "{LOG_PREFIX} updated thread labels id={} path={}", + thread_id, + threads_path.display() + ); + self.thread_summary_unlocked(thread_id)? + .ok_or_else(|| format!("thread {} missing after labels update", thread_id)) + } + pub fn update_message( &self, thread_id: &str, @@ -297,6 +332,7 @@ impl ConversationStore { message_count, last_message_at, created_at: entry.created_at.clone(), + labels: entry.labels.clone(), })) } @@ -314,17 +350,25 @@ impl ConversationStore { thread_id, title, created_at, + labels, .. } => { - let created_at_value = match index.get(&thread_id) { - Some(existing) => existing.created_at.clone(), - None => created_at, + let (created_at_value, labels_value) = match index.get(&thread_id) { + Some(existing) => ( + existing.created_at.clone(), + labels.unwrap_or_else(|| existing.labels.clone()), + ), + None => { + let inferred = labels.unwrap_or_else(|| infer_labels(&thread_id)); + (created_at, inferred) + } }; index.insert( thread_id, ThreadIndexEntry { title, created_at: created_at_value, + labels: labels_value, }, ); } @@ -350,6 +394,17 @@ impl ConversationStore { struct ThreadIndexEntry { title: String, created_at: String, + labels: Vec, +} + +fn infer_labels(thread_id: &str) -> Vec { + if thread_id == "proactive:morning_briefing" { + vec!["briefing".to_string()] + } else if thread_id.starts_with("proactive:") { + vec!["notification".to_string()] + } else { + vec!["work".to_string()] + } } fn read_jsonl(path: &Path) -> Result, String> @@ -466,6 +521,15 @@ pub fn update_thread_title( ConversationStore::new(workspace_dir).update_thread_title(thread_id, title, updated_at) } +pub fn update_thread_labels( + workspace_dir: PathBuf, + thread_id: &str, + labels: Vec, + updated_at: &str, +) -> Result { + ConversationStore::new(workspace_dir).update_thread_labels(thread_id, labels, updated_at) +} + pub fn update_message( workspace_dir: PathBuf, thread_id: &str, diff --git a/src/openhuman/memory/conversations/store_tests.rs b/src/openhuman/memory/conversations/store_tests.rs index 7273f6c99..234fc9d6c 100644 --- a/src/openhuman/memory/conversations/store_tests.rs +++ b/src/openhuman/memory/conversations/store_tests.rs @@ -18,6 +18,7 @@ fn store_roundtrips_threads_and_messages() { id: "default-thread".to_string(), title: "Conversation".to_string(), created_at: created_at.clone(), + labels: None, }) .expect("ensure thread"); assert_eq!(thread.message_count, 0); @@ -54,6 +55,7 @@ fn store_updates_message_metadata() { id: "default-thread".to_string(), title: "Conversation".to_string(), created_at: "2026-04-10T12:00:00Z".to_string(), + labels: None, }) .expect("ensure thread"); store @@ -93,6 +95,7 @@ fn purge_removes_threads_and_messages() { id: "default-thread".to_string(), title: "Conversation".to_string(), created_at: "2026-04-10T12:00:00Z".to_string(), + labels: None, }) .expect("ensure thread"); store @@ -122,6 +125,7 @@ fn ensure_thread_is_idempotent() { id: "t1".to_string(), title: "Thread".to_string(), created_at: "2026-04-10T12:00:00Z".to_string(), + labels: None, }; store.ensure_thread(req.clone()).unwrap(); store.ensure_thread(req).unwrap(); @@ -137,6 +141,7 @@ fn delete_thread_removes_thread_and_messages() { id: "t1".to_string(), title: "Thread".to_string(), created_at: "2026-04-10T12:00:00Z".to_string(), + labels: None, }) .unwrap(); store @@ -174,6 +179,7 @@ fn get_messages_empty_thread() { id: "t1".to_string(), title: "Empty".to_string(), created_at: "2026-04-10T12:00:00Z".to_string(), + labels: None, }) .unwrap(); let messages = store.get_messages("t1").unwrap(); @@ -196,6 +202,7 @@ fn multiple_threads_and_messages() { id: format!("t{i}"), title: format!("Thread {i}"), created_at: format!("2026-04-10T12:0{i}:00Z"), + labels: None, }) .unwrap(); store @@ -232,6 +239,7 @@ fn update_message_nonexistent_returns_error() { id: "t1".to_string(), title: "Thread".to_string(), created_at: "2026-04-10T12:00:00Z".to_string(), + labels: None, }) .unwrap(); let result = store.update_message( @@ -252,6 +260,7 @@ fn update_thread_title_persists_latest_title() { id: "t1".to_string(), title: "Chat Apr 10 12:00 PM".to_string(), created_at: "2026-04-10T12:00:00Z".to_string(), + labels: None, }) .unwrap(); @@ -265,6 +274,93 @@ fn update_thread_title_persists_latest_title() { assert_eq!(threads[0].created_at, "2026-04-10T12:00:00Z"); } +#[test] +fn store_handles_labels_and_inference() { + let (_temp, store) = make_store(); + + // 1. Explicit labels on ensure + store + .ensure_thread(CreateConversationThread { + id: "t1".to_string(), + title: "Thread 1".to_string(), + created_at: "2026-04-10T12:00:00Z".to_string(), + labels: Some(vec!["custom".to_string()]), + }) + .unwrap(); + + // 2. Inferred labels for morning briefing + store + .ensure_thread(CreateConversationThread { + id: "proactive:morning_briefing".to_string(), + title: "Morning Briefing".to_string(), + created_at: "2026-04-10T12:00:00Z".to_string(), + labels: None, + }) + .unwrap(); + + // 3. Inferred labels for other proactive + store + .ensure_thread(CreateConversationThread { + id: "proactive:system".to_string(), + title: "System Notification".to_string(), + created_at: "2026-04-10T12:00:00Z".to_string(), + labels: None, + }) + .unwrap(); + + // 4. Default inferred labels (work) + store + .ensure_thread(CreateConversationThread { + id: "user-thread".to_string(), + title: "User Chat".to_string(), + created_at: "2026-04-10T12:00:00Z".to_string(), + labels: None, + }) + .unwrap(); + + let threads = store.list_threads().unwrap(); + { + let t1 = threads.iter().find(|t| t.id == "t1").unwrap(); + assert_eq!(t1.labels, vec!["custom"]); + } + { + let mb = threads + .iter() + .find(|t| t.id == "proactive:morning_briefing") + .unwrap(); + assert_eq!(mb.labels, vec!["briefing"]); + } + { + let sys = threads.iter().find(|t| t.id == "proactive:system").unwrap(); + assert_eq!(sys.labels, vec!["notification"]); + } + { + let user = threads.iter().find(|t| t.id == "user-thread").unwrap(); + assert_eq!(user.labels, vec!["work"]); + } + + // 5. Update labels + store + .update_thread_labels("t1", vec!["updated".to_string()], "2026-04-10T12:05:00Z") + .unwrap(); + let threads = store.list_threads().unwrap(); + { + let t1 = threads.iter().find(|t| t.id == "t1").unwrap(); + assert_eq!(t1.labels, vec!["updated"]); + } + + // 6. Title update preserves labels + store + .update_thread_title("t1", "New Title", "2026-04-10T12:06:00Z") + .unwrap(); + let threads = store.list_threads().unwrap(); + { + let t1 = threads.iter().find(|t| t.id == "t1").unwrap(); + assert_eq!(t1.labels, vec!["updated"]); + assert_eq!(t1.title, "New Title"); + } +} + #[test] fn conversation_store_new() { let tmp = TempDir::new().unwrap(); diff --git a/src/openhuman/memory/conversations/types.rs b/src/openhuman/memory/conversations/types.rs index 6fd996228..d4148d650 100644 --- a/src/openhuman/memory/conversations/types.rs +++ b/src/openhuman/memory/conversations/types.rs @@ -12,6 +12,8 @@ pub struct ConversationThread { pub message_count: usize, pub last_message_at: String, pub created_at: String, + #[serde(default)] + pub labels: Vec, } #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] @@ -33,6 +35,8 @@ pub struct CreateConversationThread { pub id: String, pub title: String, pub created_at: String, + #[serde(default)] + pub labels: Option>, } #[derive(Debug, Clone, Serialize, Deserialize, Default)] diff --git a/src/openhuman/memory/rpc_models.rs b/src/openhuman/memory/rpc_models.rs index c49415600..d66f355e9 100644 --- a/src/openhuman/memory/rpc_models.rs +++ b/src/openhuman/memory/rpc_models.rs @@ -67,6 +67,14 @@ pub struct ApiEnvelope { #[serde(deny_unknown_fields)] pub struct EmptyRequest {} +/// Request to create a new conversation thread. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct CreateConversationThreadRequest { + #[serde(default)] + pub labels: Option>, +} + /// Request payload for `openhuman.memory_init`. /// /// `jwt_token` is accepted for backward compatibility but **not used** — memory @@ -102,6 +110,8 @@ pub struct ConversationThreadSummary { pub message_count: usize, pub last_message_at: String, pub created_at: String, + #[serde(default)] + pub labels: Vec, } /// A single persisted conversation message. @@ -125,6 +135,16 @@ pub struct UpsertConversationThreadRequest { pub id: String, pub title: String, pub created_at: String, + #[serde(default)] + pub labels: Option>, +} + +/// Request to update labels for a conversation thread. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct UpdateConversationThreadLabelsRequest { + pub thread_id: String, + pub labels: Vec, } /// Response payload for thread list operations. diff --git a/src/openhuman/threads/ops.rs b/src/openhuman/threads/ops.rs index 87f040cb1..fdcd4179a 100644 --- a/src/openhuman/threads/ops.rs +++ b/src/openhuman/threads/ops.rs @@ -9,9 +9,10 @@ use crate::openhuman::memory::conversations::{ use crate::openhuman::memory::{ ApiEnvelope, ApiMeta, AppendConversationMessageRequest, ConversationMessageRecord, ConversationMessagesRequest, ConversationMessagesResponse, ConversationThreadSummary, - ConversationThreadsListResponse, DeleteConversationThreadRequest, - DeleteConversationThreadResponse, EmptyRequest, GenerateConversationThreadTitleRequest, - PaginationMeta, PurgeConversationThreadsResponse, UpdateConversationMessageRequest, + ConversationThreadsListResponse, CreateConversationThreadRequest, + DeleteConversationThreadRequest, DeleteConversationThreadResponse, EmptyRequest, + GenerateConversationThreadTitleRequest, PaginationMeta, PurgeConversationThreadsResponse, + UpdateConversationMessageRequest, UpdateConversationThreadLabelsRequest, UpsertConversationThreadRequest, }; use crate::openhuman::providers::{self, ProviderRuntimeOptions}; @@ -73,6 +74,7 @@ fn thread_to_summary(thread: ConversationThread) -> ConversationThreadSummary { message_count: thread.message_count, last_message_at: thread.last_message_at, created_at: thread.created_at, + labels: thread.labels, } } @@ -126,6 +128,7 @@ pub async fn thread_upsert( id: request.id, title: request.title, created_at: request.created_at, + labels: request.labels, }, )?; Ok(envelope( @@ -137,7 +140,7 @@ pub async fn thread_upsert( /// Creates a new conversation thread with auto-generated ID and title. pub async fn thread_create_new( - _request: EmptyRequest, + request: CreateConversationThreadRequest, ) -> Result>, String> { let dir = workspace_dir().await?; let id = format!("thread-{}", uuid::Uuid::new_v4()); @@ -150,8 +153,17 @@ pub async fn thread_create_new( id, title, created_at, + // Pass labels through as-is; the store's infer_labels() applies + // the same default on index rebuild, so this is the single source + // of truth for default labels. + labels: request.labels, }, )?; + tracing::debug!( + thread_id = %thread.id, + labels = ?thread.labels, + "[threads] created new thread" + ); Ok(envelope( thread_to_summary(thread), Some(counts([("num_threads", 1)])), @@ -363,6 +375,33 @@ pub async fn thread_generate_title( )) } +/// Updates labels for a conversation thread. +/// +/// An empty `labels` vec is valid and clears all labels from the thread, +/// making it invisible in every non-"All" filter view. Callers should +/// ensure this is intentional. +pub async fn thread_update_labels( + request: UpdateConversationThreadLabelsRequest, +) -> Result>, String> { + let dir = workspace_dir().await?; + let thread = conversations::update_thread_labels( + dir, + &request.thread_id, + request.labels.clone(), + &chrono::Utc::now().to_rfc3339(), + )?; + tracing::debug!( + thread_id = %request.thread_id, + labels = ?request.labels, + "[threads] updated thread labels" + ); + Ok(envelope( + thread_to_summary(thread), + Some(counts([("num_threads", 1)])), + None, + )) +} + /// Updates metadata on an existing conversation message. pub async fn message_update( request: UpdateConversationMessageRequest, diff --git a/src/openhuman/threads/ops_tests.rs b/src/openhuman/threads/ops_tests.rs index beb212605..3b9a85004 100644 --- a/src/openhuman/threads/ops_tests.rs +++ b/src/openhuman/threads/ops_tests.rs @@ -288,6 +288,7 @@ fn sample_thread() -> ConversationThread { message_count: 5, last_message_at: "2026-01-01T00:00:00Z".into(), created_at: "2026-01-01T00:00:00Z".into(), + labels: vec!["work".to_string()], } } @@ -312,6 +313,7 @@ fn thread_to_summary_preserves_all_fields() { assert_eq!(summary.message_count, 5); assert_eq!(summary.last_message_at, "2026-01-01T00:00:00Z"); assert_eq!(summary.created_at, "2026-01-01T00:00:00Z"); + assert_eq!(summary.labels, vec!["work".to_string()]); } #[test] diff --git a/src/openhuman/threads/schemas.rs b/src/openhuman/threads/schemas.rs index c8be1c796..ce3a27645 100644 --- a/src/openhuman/threads/schemas.rs +++ b/src/openhuman/threads/schemas.rs @@ -6,8 +6,9 @@ use serde_json::{Map, Value}; use crate::core::all::{ControllerFuture, RegisteredController}; use crate::core::{ControllerSchema, FieldSchema, TypeSchema}; use crate::openhuman::memory::{ - AppendConversationMessageRequest, ConversationMessagesRequest, DeleteConversationThreadRequest, - EmptyRequest, GenerateConversationThreadTitleRequest, UpdateConversationMessageRequest, + AppendConversationMessageRequest, ConversationMessagesRequest, CreateConversationThreadRequest, + DeleteConversationThreadRequest, EmptyRequest, GenerateConversationThreadTitleRequest, + UpdateConversationMessageRequest, UpdateConversationThreadLabelsRequest, UpsertConversationThreadRequest, }; @@ -21,6 +22,7 @@ pub fn all_controller_schemas() -> Vec { schemas("messages_list"), schemas("message_append"), schemas("generate_title"), + schemas("update_labels"), schemas("message_update"), schemas("delete"), schemas("purge"), @@ -53,6 +55,10 @@ pub fn all_registered_controllers() -> Vec { schema: schemas("generate_title"), handler: handle_generate_title, }, + RegisteredController { + schema: schemas("update_labels"), + handler: handle_update_labels, + }, RegisteredController { schema: schemas("message_update"), handler: handle_message_update, @@ -105,6 +111,12 @@ pub fn schemas(function: &str) -> ControllerSchema { comment: "RFC3339 timestamp for first thread creation.", required: true, }, + FieldSchema { + name: "labels", + ty: TypeSchema::Option(Box::new(TypeSchema::Json)), + comment: "Optional list of labels to assign.", + required: false, + }, ], outputs: vec![FieldSchema { name: "result", @@ -117,7 +129,12 @@ pub fn schemas(function: &str) -> ControllerSchema { namespace: "threads", function: "create_new", description: "Create a new conversation thread with auto-generated ID and title.", - inputs: vec![], + inputs: vec![FieldSchema { + name: "labels", + ty: TypeSchema::Option(Box::new(TypeSchema::Json)), + comment: "Optional labels to assign to the new thread.", + required: false, + }], outputs: vec![FieldSchema { name: "result", ty: TypeSchema::Json, @@ -225,6 +242,31 @@ pub fn schemas(function: &str) -> ControllerSchema { required: true, }], }, + "update_labels" => ControllerSchema { + namespace: "threads", + function: "update_labels", + description: "Update labels for a conversation thread.", + inputs: vec![ + FieldSchema { + name: "thread_id", + ty: TypeSchema::String, + comment: "Thread identifier.", + required: true, + }, + FieldSchema { + name: "labels", + ty: TypeSchema::Json, + comment: "List of labels to assign.", + required: true, + }, + ], + outputs: vec![FieldSchema { + name: "result", + ty: TypeSchema::Json, + comment: "Envelope with the resulting thread summary.", + required: true, + }], + }, "delete" => ControllerSchema { namespace: "threads", function: "delete", @@ -290,8 +332,11 @@ fn handle_upsert(params: Map) -> ControllerFuture { }) } -fn handle_create_new(_params: Map) -> ControllerFuture { - Box::pin(async move { to_json(ops::thread_create_new(EmptyRequest {}).await?) }) +fn handle_create_new(params: Map) -> ControllerFuture { + Box::pin(async move { + let p = parse::(params)?; + to_json(ops::thread_create_new(p).await?) + }) } fn handle_messages_list(params: Map) -> ControllerFuture { @@ -315,6 +360,13 @@ fn handle_generate_title(params: Map) -> ControllerFuture { }) } +fn handle_update_labels(params: Map) -> ControllerFuture { + Box::pin(async move { + let p = parse::(params)?; + to_json(ops::thread_update_labels(p).await?) + }) +} + fn handle_message_update(params: Map) -> ControllerFuture { Box::pin(async move { let p = parse::(params)?; diff --git a/src/openhuman/threads/schemas_tests.rs b/src/openhuman/threads/schemas_tests.rs index 617d529a7..4387d729a 100644 --- a/src/openhuman/threads/schemas_tests.rs +++ b/src/openhuman/threads/schemas_tests.rs @@ -8,6 +8,7 @@ const ALL_FUNCTIONS: &[&str] = &[ "messages_list", "message_append", "generate_title", + "update_labels", "message_update", "delete", "purge", diff --git a/tests/json_rpc_e2e.rs b/tests/json_rpc_e2e.rs index 54cd8c27a..989f9f3eb 100644 --- a/tests/json_rpc_e2e.rs +++ b/tests/json_rpc_e2e.rs @@ -796,6 +796,111 @@ async fn json_rpc_protocol_auth_and_agent_hello() { rpc_join.abort(); } +#[tokio::test] +async fn json_rpc_thread_labels_create_and_update() { + let _env_lock = json_rpc_e2e_env_lock(); + let tmp = tempdir().expect("tempdir"); + let home = tmp.path(); + let openhuman_home = home.join(".openhuman"); + + let _home_guard = EnvVarGuard::set_to_path("HOME", home); + let _workspace_guard = EnvVarGuard::unset("OPENHUMAN_WORKSPACE"); + let _backend_url_guard = EnvVarGuard::unset("BACKEND_URL"); + let _vite_backend_url_guard = EnvVarGuard::unset("VITE_BACKEND_URL"); + let _api_url_guard = EnvVarGuard::unset("OPENHUMAN_API_URL"); + + let (api_addr, api_join) = serve_on_ephemeral(mock_upstream_router()).await; + let api_origin = format!("http://{api_addr}"); + write_min_config(openhuman_home.as_path(), &api_origin); + + let (rpc_addr, rpc_join) = serve_on_ephemeral(build_core_http_router(false)).await; + let rpc_base = format!("http://{rpc_addr}"); + + // 1. Create a thread with an explicit label. + let create = post_json_rpc( + &rpc_base, + 9001, + "openhuman.threads_create_new", + json!({ "labels": ["custom"] }), + ) + .await; + let create_outer = assert_no_jsonrpc_error(&create, "threads_create_new with labels"); + let created = create_outer + .get("data") + .expect("data envelope in create response"); + let thread_id = created + .get("id") + .and_then(Value::as_str) + .expect("id in created thread"); + let created_labels = created + .get("labels") + .and_then(Value::as_array) + .expect("labels in created thread"); + assert_eq!( + created_labels + .iter() + .map(|v| v.as_str().unwrap_or("")) + .collect::>(), + vec!["custom"], + "created thread should have labels=[\"custom\"]" + ); + + // 2. Update labels on the thread. + let update = post_json_rpc( + &rpc_base, + 9002, + "openhuman.threads_update_labels", + json!({ "thread_id": thread_id, "labels": ["work", "briefing"] }), + ) + .await; + let update_outer = assert_no_jsonrpc_error(&update, "threads_update_labels"); + let updated = update_outer + .get("data") + .expect("data envelope in update response"); + let updated_labels = updated + .get("labels") + .and_then(Value::as_array) + .expect("labels in updated thread"); + assert_eq!( + updated_labels + .iter() + .map(|v| v.as_str().unwrap_or("")) + .collect::>(), + vec!["work", "briefing"], + "updated thread should have labels=[\"work\", \"briefing\"]" + ); + + // 3. Verify the updated labels are reflected in threads_list. + let list = post_json_rpc(&rpc_base, 9003, "openhuman.threads_list", json!({})).await; + let list_outer = assert_no_jsonrpc_error(&list, "threads_list after label update"); + let list_result = list_outer + .get("data") + .expect("data envelope in list response"); + let threads = list_result + .get("threads") + .and_then(Value::as_array) + .expect("threads array in list"); + let persisted = threads + .iter() + .find(|t| t.get("id").and_then(Value::as_str) == Some(thread_id)) + .expect("created thread must appear in list"); + let persisted_labels = persisted + .get("labels") + .and_then(Value::as_array) + .expect("labels in persisted thread"); + assert_eq!( + persisted_labels + .iter() + .map(|v| v.as_str().unwrap_or("")) + .collect::>(), + vec!["work", "briefing"], + "threads_list must reflect the updated labels" + ); + + api_join.abort(); + rpc_join.abort(); +} + #[tokio::test] async fn json_rpc_memory_tree_end_to_end() { let _env_lock = json_rpc_e2e_env_lock();