feat(threads): dedicated threads controller with per-thread session scoping (#590)

* feat(conversations): implement thread management features including creation and deletion

- Added functionality to create new conversation threads with unique IDs and titles based on the current date and time.
- Introduced a deleteThread action to remove existing threads, updating the selected thread accordingly.
- Enhanced the UI to display threads in a sidebar, allowing users to select and delete threads easily.
- Updated the Conversations component to handle thread loading and selection more efficiently, ensuring a smoother user experience.

Also updated the OpenHuman package version to 0.52.15 in Cargo.lock files.

* refactor(conversations): rename createThreadLocal to createNewThread and streamline thread creation logic

- Updated the function name from `createThreadLocal` to `createNewThread` for clarity and consistency.
- Simplified the thread creation process by removing the manual ID and title generation, leveraging the new API method for automatic thread creation.
- Adjusted the Conversations component to utilize the new `handleCreateNewThread` function, enhancing readability and maintainability.
- Removed unused thread ID and title generation functions to clean up the codebase.

This refactor improves the overall structure and clarity of the thread management functionality.

* refactor(memory): remove deprecated conversation thread management functions

- Eliminated unused functions related to listing, creating, updating, appending, and deleting conversation threads in memory operations.
- This cleanup enhances code maintainability and reduces complexity in the memory module.
- The refactor focuses on streamlining the conversation management logic, aligning with recent changes in the thread handling API.

* refactor(api): update thread API method names and remove deprecated functions

- Renamed thread-related API methods to align with the new naming convention, improving clarity and consistency.
- Removed deprecated functions related to thread creation, streamlining the API and enhancing maintainability.
- Adjusted the implementation of existing methods to reflect the updated API structure, ensuring proper functionality.

* refactor(thread): simplify thread state management and remove unused properties

- Streamlined the thread state by removing the lastViewedAt property and related logic, enhancing clarity in unread message counting.
- Updated the BottomTabBar component to reflect the new unread count logic, which now simply returns the length of threads.
- Removed the setLastViewed action from the Conversations component, further simplifying the thread management logic.
- Introduced a new deleteThread action to handle thread deletion, ensuring proper state updates and selection handling.
- Overall, these changes improve maintainability and reduce complexity in the thread management system.

* style(threads): apply cargo fmt

* refactor(tabbar): remove conversations unread badge

* update(Cargo.lock): bump OpenHuman package version to 0.52.15

* test(threadApi): update RPC method names to threads namespace

* refactor(conversations): update model ID for chat functionality

- Changed the model ID used in the chatSend function from `agentic-v1` to `reasoning-v1`, clarifying the purpose of each model.
- Added comments to explain the distinction between the reasoning model and the agentic model, enhancing code readability and maintainability.

src/core/all.rs

@@ -137,6 +137,8 @@ fn build_registered_controllers() -> Vec<RegisteredController> {
         .extend(crate::openhuman::tree_summarizer::all_tree_summarizer_registered_controllers());
     // Self-learning and user context enrichment
     controllers.extend(crate::openhuman::learning::all_learning_registered_controllers());
+    // Conversation thread and message management
+    controllers.extend(crate::openhuman::threads::all_threads_registered_controllers());
     controllers
 }
 
@@ -182,6 +184,8 @@ fn build_declared_controller_schemas() -> Vec<ControllerSchema> {
     schemas.extend(crate::openhuman::update::all_update_controller_schemas());
     schemas.extend(crate::openhuman::tree_summarizer::all_tree_summarizer_controller_schemas());
     schemas.extend(crate::openhuman::learning::all_learning_controller_schemas());
+    // Conversation thread and message management
+    schemas.extend(crate::openhuman::threads::all_threads_controller_schemas());
     schemas
 }
 

src/openhuman/agent/harness/session/runtime.rs

@@ -140,6 +140,14 @@ impl Agent {
         self.event_channel = channel.into();
     }
 
+    /// Override the agent definition name used for session transcript
+    /// file paths. Callers (e.g. the web channel) use this to scope
+    /// transcripts per thread so each conversation thread gets its own
+    /// transcript namespace instead of sharing one by agent type.
+    pub fn set_agent_definition_name(&mut self, name: impl Into<String>) {
+        self.agent_definition_name = name.into();
+    }
+
     /// Attach a progress event sender for real-time turn updates.
     ///
     /// When set, the turn loop emits [`AgentProgress`] events so

src/openhuman/channels/providers/web.rs

@@ -251,6 +251,28 @@ pub async fn start_chat(
     Ok(request_id)
 }
 
+/// Invalidate all cached agent sessions for the given thread ID.
+/// Called when a thread is deleted so stale sessions don't leak
+/// into reused thread IDs.
+pub async fn invalidate_thread_sessions(thread_id: &str) {
+    let mut sessions = THREAD_SESSIONS.lock().await;
+    let keys_to_remove: Vec<String> = sessions
+        .keys()
+        .filter(|k| k.ends_with(&format!("::{thread_id}")))
+        .cloned()
+        .collect();
+    for key in &keys_to_remove {
+        sessions.remove(key);
+    }
+    if !keys_to_remove.is_empty() {
+        log::debug!(
+            "[web-channel] invalidated {} cached session(s) for thread_id={}",
+            keys_to_remove.len(),
+            thread_id
+        );
+    }
+}
+
 pub async fn cancel_chat(client_id: &str, thread_id: &str) -> Result<Option<String>, String> {
     let client_id = client_id.trim();
     let thread_id = thread_id.trim();
@@ -826,6 +848,16 @@ fn build_session_agent(
     Agent::from_config_for_agent(&effective, target_agent_id)
         .map(|mut agent| {
             agent.set_event_context(event_session_id_for(client_id, thread_id), "web_channel");
+            // Scope session transcripts per thread so each conversation
+            // gets its own transcript file instead of sharing one by
+            // agent type. Without this, new threads load the latest
+            // transcript for the agent name and inherit prior messages.
+            let short_thread = if thread_id.len() > 12 {
+                &thread_id[..12]
+            } else {
+                thread_id
+            };
+            agent.set_agent_definition_name(format!("{target_agent_id}_{short_thread}"));
             agent
         })
         .map_err(|e| e.to_string())

src/openhuman/memory/ops.rs

@@ -6,26 +6,18 @@
 //! for formatting and filtering memory results.
 
 use crate::openhuman::config::Config;
-use crate::openhuman::memory::conversations::{
-    self, ConversationMessage, ConversationMessagePatch, ConversationThread,
-    CreateConversationThread,
-};
 use crate::openhuman::memory::store::GraphRelationRecord;
 use crate::openhuman::memory::{
-    ApiEnvelope, ApiError, ApiMeta, AppendConversationMessageRequest, ConversationMessageRecord,
-    ConversationMessagesRequest, ConversationMessagesResponse, ConversationThreadSummary,
-    ConversationThreadsListResponse, DeleteConversationThreadRequest,
-    DeleteConversationThreadResponse, DeleteDocumentRequest, DeleteDocumentResponse, EmptyRequest,
+    ApiEnvelope, ApiError, ApiMeta, DeleteDocumentRequest, DeleteDocumentResponse, EmptyRequest,
     ListDocumentsRequest, ListDocumentsResponse, ListMemoryFilesRequest, ListMemoryFilesResponse,
     ListNamespacesResponse, MemoryClient, MemoryClientRef, MemoryDocumentSummary,
     MemoryIngestionConfig, MemoryIngestionRequest, MemoryIngestionResult, MemoryInitRequest,
     MemoryInitResponse, MemoryItemKind, MemoryRecallItem, MemoryRetrievalChunk,
     MemoryRetrievalContext, MemoryRetrievalEntity, MemoryRetrievalRelation, NamespaceDocumentInput,
-    NamespaceMemoryHit, NamespaceRetrievalContext, PaginationMeta,
-    PurgeConversationThreadsResponse, QueryNamespaceRequest, QueryNamespaceResponse,
-    ReadMemoryFileRequest, ReadMemoryFileResponse, RecallContextRequest, RecallContextResponse,
-    RecallMemoriesRequest, RecallMemoriesResponse, UpdateConversationMessageRequest,
-    UpsertConversationThreadRequest, WriteMemoryFileRequest, WriteMemoryFileResponse,
+    NamespaceMemoryHit, NamespaceRetrievalContext, PaginationMeta, QueryNamespaceRequest,
+    QueryNamespaceResponse, ReadMemoryFileRequest, ReadMemoryFileResponse, RecallContextRequest,
+    RecallContextResponse, RecallMemoriesRequest, RecallMemoriesResponse, WriteMemoryFileRequest,
+    WriteMemoryFileResponse,
 };
 use crate::rpc::RpcOutcome;
 use chrono::TimeZone;
@@ -699,40 +691,6 @@ fn default_category() -> String {
     "core".to_string()
 }
 
-fn conversation_thread_to_summary(thread: ConversationThread) -> ConversationThreadSummary {
-    ConversationThreadSummary {
-        id: thread.id,
-        title: thread.title,
-        chat_id: thread.chat_id,
-        is_active: thread.is_active,
-        message_count: thread.message_count,
-        last_message_at: thread.last_message_at,
-        created_at: thread.created_at,
-    }
-}
-
-fn conversation_message_to_record(message: ConversationMessage) -> ConversationMessageRecord {
-    ConversationMessageRecord {
-        id: message.id,
-        content: message.content,
-        message_type: message.message_type,
-        extra_metadata: message.extra_metadata,
-        sender: message.sender,
-        created_at: message.created_at,
-    }
-}
-
-fn conversation_record_to_message(record: ConversationMessageRecord) -> ConversationMessage {
-    ConversationMessage {
-        id: record.id,
-        content: record.content,
-        message_type: record.message_type,
-        extra_metadata: record.extra_metadata,
-        sender: record.sender,
-        created_at: record.created_at,
-    }
-}
-
 /// Lists all namespaces in the memory system.
 pub async fn namespace_list() -> Result<RpcOutcome<Vec<String>>, String> {
     let client = active_memory_client().await?;
@@ -1011,128 +969,6 @@ pub async fn memory_delete_document(
     ))
 }
 
-/// Lists workspace-backed conversation threads from JSONL storage.
-pub async fn memory_threads_list(
-    _request: EmptyRequest,
-) -> Result<RpcOutcome<ApiEnvelope<ConversationThreadsListResponse>>, String> {
-    let workspace_dir = current_workspace_dir().await?;
-    let threads = conversations::list_threads(workspace_dir)?
-        .into_iter()
-        .map(conversation_thread_to_summary)
-        .collect::<Vec<_>>();
-    let count = threads.len();
-    Ok(envelope(
-        ConversationThreadsListResponse { threads, count },
-        Some(memory_counts([("num_threads", count)])),
-        None,
-    ))
-}
-
-/// Ensures a workspace-backed conversation thread exists.
-pub async fn memory_thread_upsert(
-    request: UpsertConversationThreadRequest,
-) -> Result<RpcOutcome<ApiEnvelope<ConversationThreadSummary>>, String> {
-    let workspace_dir = current_workspace_dir().await?;
-    let thread = conversations::ensure_thread(
-        workspace_dir,
-        CreateConversationThread {
-            id: request.id,
-            title: request.title,
-            created_at: request.created_at,
-        },
-    )?;
-    Ok(envelope(
-        conversation_thread_to_summary(thread),
-        Some(memory_counts([("num_threads", 1)])),
-        None,
-    ))
-}
-
-/// Lists persisted messages for a workspace-backed conversation thread.
-pub async fn memory_messages_list(
-    request: ConversationMessagesRequest,
-) -> Result<RpcOutcome<ApiEnvelope<ConversationMessagesResponse>>, String> {
-    let workspace_dir = current_workspace_dir().await?;
-    let messages = conversations::get_messages(workspace_dir, &request.thread_id)?
-        .into_iter()
-        .map(conversation_message_to_record)
-        .collect::<Vec<_>>();
-    let count = messages.len();
-    Ok(envelope(
-        ConversationMessagesResponse { messages, count },
-        Some(memory_counts([("num_messages", count)])),
-        None,
-    ))
-}
-
-/// Appends a persisted message to a workspace-backed conversation thread.
-pub async fn memory_message_append(
-    request: AppendConversationMessageRequest,
-) -> Result<RpcOutcome<ApiEnvelope<ConversationMessageRecord>>, String> {
-    let workspace_dir = current_workspace_dir().await?;
-    let message = conversations::append_message(
-        workspace_dir,
-        &request.thread_id,
-        conversation_record_to_message(request.message),
-    )?;
-    Ok(envelope(
-        conversation_message_to_record(message),
-        Some(memory_counts([("num_messages", 1)])),
-        None,
-    ))
-}
-
-/// Updates persisted metadata for an existing conversation message.
-pub async fn memory_message_update(
-    request: UpdateConversationMessageRequest,
-) -> Result<RpcOutcome<ApiEnvelope<ConversationMessageRecord>>, String> {
-    let workspace_dir = current_workspace_dir().await?;
-    let message = conversations::update_message(
-        workspace_dir,
-        &request.thread_id,
-        &request.message_id,
-        ConversationMessagePatch {
-            extra_metadata: request.extra_metadata,
-        },
-    )?;
-    Ok(envelope(
-        conversation_message_to_record(message),
-        Some(memory_counts([("num_messages", 1)])),
-        None,
-    ))
-}
-
-/// Deletes a workspace-backed conversation thread and its JSONL message log.
-pub async fn memory_thread_delete(
-    request: DeleteConversationThreadRequest,
-) -> Result<RpcOutcome<ApiEnvelope<DeleteConversationThreadResponse>>, String> {
-    let workspace_dir = current_workspace_dir().await?;
-    let deleted = conversations::ConversationStore::new(workspace_dir)
-        .delete_thread(&request.thread_id, &request.deleted_at)?;
-    Ok(envelope(
-        DeleteConversationThreadResponse { deleted },
-        None,
-        None,
-    ))
-}
-
-/// Purges all workspace-backed conversation JSONL state.
-pub async fn memory_threads_purge(
-    _request: EmptyRequest,
-) -> Result<RpcOutcome<ApiEnvelope<PurgeConversationThreadsResponse>>, String> {
-    let workspace_dir = current_workspace_dir().await?;
-    let stats = conversations::purge_threads(workspace_dir)?;
-    Ok(envelope(
-        PurgeConversationThreadsResponse {
-            messages_deleted: stats.message_count,
-            agent_threads_deleted: stats.thread_count,
-            agent_messages_deleted: stats.message_count,
-        },
-        None,
-        None,
-    ))
-}
-
 /// Performs a semantic query against a namespace, returning a retrieval context.
 pub async fn memory_query_namespace(
     request: QueryNamespaceRequest,

src/openhuman/memory/schemas.rs

@@ -15,11 +15,9 @@ use crate::openhuman::memory::rpc::{
     QueryNamespaceParams, RecallNamespaceParams,
 };
 use crate::openhuman::memory::{
-    AppendConversationMessageRequest, ConversationMessagesRequest, DeleteConversationThreadRequest,
     DeleteDocumentRequest, EmptyRequest, ListDocumentsRequest, ListMemoryFilesRequest,
     MemoryInitRequest, QueryNamespaceRequest, ReadMemoryFileRequest, RecallContextRequest,
-    RecallMemoriesRequest, UpdateConversationMessageRequest, UpsertConversationThreadRequest,
-    WriteMemoryFileRequest,
+    RecallMemoriesRequest, WriteMemoryFileRequest,
 };
 use crate::rpc::RpcOutcome;
 
@@ -40,13 +38,6 @@ pub fn all_controller_schemas() -> Vec<ControllerSchema> {
         schemas("list_files"),
         schemas("read_file"),
         schemas("write_file"),
-        schemas("threads_list"),
-        schemas("thread_upsert"),
-        schemas("messages_list"),
-        schemas("message_append"),
-        schemas("message_update"),
-        schemas("thread_delete"),
-        schemas("threads_purge"),
         schemas("namespace_list"),
         schemas("doc_put"),
         schemas("doc_ingest"),
@@ -107,34 +98,6 @@ pub fn all_registered_controllers() -> Vec<RegisteredController> {
             schema: schemas("write_file"),
             handler: handle_write_file,
         },
-        RegisteredController {
-            schema: schemas("threads_list"),
-            handler: handle_threads_list,
-        },
-        RegisteredController {
-            schema: schemas("thread_upsert"),
-            handler: handle_thread_upsert,
-        },
-        RegisteredController {
-            schema: schemas("messages_list"),
-            handler: handle_messages_list,
-        },
-        RegisteredController {
-            schema: schemas("message_append"),
-            handler: handle_message_append,
-        },
-        RegisteredController {
-            schema: schemas("message_update"),
-            handler: handle_message_update,
-        },
-        RegisteredController {
-            schema: schemas("thread_delete"),
-            handler: handle_thread_delete,
-        },
-        RegisteredController {
-            schema: schemas("threads_purge"),
-            handler: handle_threads_purge,
-        },
         RegisteredController {
             schema: schemas("namespace_list"),
             handler: handle_namespace_list,
@@ -471,160 +434,6 @@ pub fn schemas(function: &str) -> ControllerSchema {
                 required: true,
             }],
         },
-        "threads_list" => ControllerSchema {
-            namespace: "memory",
-            function: "threads_list",
-            description: "List workspace-backed conversation threads stored as JSONL files.",
-            inputs: vec![],
-            outputs: vec![FieldSchema {
-                name: "result",
-                ty: TypeSchema::Json,
-                comment: "Envelope with thread summaries and count.",
-                required: true,
-            }],
-        },
-        "thread_upsert" => ControllerSchema {
-            namespace: "memory",
-            function: "thread_upsert",
-            description: "Create or refresh a workspace-backed conversation thread entry.",
-            inputs: vec![
-                FieldSchema {
-                    name: "id",
-                    ty: TypeSchema::String,
-                    comment: "Stable thread identifier.",
-                    required: true,
-                },
-                FieldSchema {
-                    name: "title",
-                    ty: TypeSchema::String,
-                    comment: "Human-readable thread title.",
-                    required: true,
-                },
-                FieldSchema {
-                    name: "created_at",
-                    ty: TypeSchema::String,
-                    comment: "RFC3339 timestamp for first thread creation.",
-                    required: true,
-                },
-            ],
-            outputs: vec![FieldSchema {
-                name: "result",
-                ty: TypeSchema::Json,
-                comment: "Envelope with the resulting thread summary.",
-                required: true,
-            }],
-        },
-        "messages_list" => ControllerSchema {
-            namespace: "memory",
-            function: "messages_list",
-            description: "List persisted messages for a workspace-backed conversation thread.",
-            inputs: vec![FieldSchema {
-                name: "thread_id",
-                ty: TypeSchema::String,
-                comment: "Thread identifier.",
-                required: true,
-            }],
-            outputs: vec![FieldSchema {
-                name: "result",
-                ty: TypeSchema::Json,
-                comment: "Envelope with messages and count.",
-                required: true,
-            }],
-        },
-        "message_append" => ControllerSchema {
-            namespace: "memory",
-            function: "message_append",
-            description: "Append a persisted message record to a workspace-backed conversation thread.",
-            inputs: vec![
-                FieldSchema {
-                    name: "thread_id",
-                    ty: TypeSchema::String,
-                    comment: "Thread identifier.",
-                    required: true,
-                },
-                FieldSchema {
-                    name: "message",
-                    ty: TypeSchema::Json,
-                    comment: "Message payload to append.",
-                    required: true,
-                },
-            ],
-            outputs: vec![FieldSchema {
-                name: "result",
-                ty: TypeSchema::Json,
-                comment: "Envelope with the appended message payload.",
-                required: true,
-            }],
-        },
-        "message_update" => ControllerSchema {
-            namespace: "memory",
-            function: "message_update",
-            description: "Patch persisted metadata for an existing conversation message.",
-            inputs: vec![
-                FieldSchema {
-                    name: "thread_id",
-                    ty: TypeSchema::String,
-                    comment: "Thread identifier.",
-                    required: true,
-                },
-                FieldSchema {
-                    name: "message_id",
-                    ty: TypeSchema::String,
-                    comment: "Message identifier.",
-                    required: true,
-                },
-                FieldSchema {
-                    name: "extra_metadata",
-                    ty: TypeSchema::Option(Box::new(TypeSchema::Json)),
-                    comment: "Replacement message metadata object.",
-                    required: false,
-                },
-            ],
-            outputs: vec![FieldSchema {
-                name: "result",
-                ty: TypeSchema::Json,
-                comment: "Envelope with the updated message payload.",
-                required: true,
-            }],
-        },
-        "thread_delete" => ControllerSchema {
-            namespace: "memory",
-            function: "thread_delete",
-            description: "Delete a workspace-backed conversation thread and its message log.",
-            inputs: vec![
-                FieldSchema {
-                    name: "thread_id",
-                    ty: TypeSchema::String,
-                    comment: "Thread identifier.",
-                    required: true,
-                },
-                FieldSchema {
-                    name: "deleted_at",
-                    ty: TypeSchema::String,
-                    comment: "RFC3339 deletion timestamp.",
-                    required: true,
-                },
-            ],
-            outputs: vec![FieldSchema {
-                name: "result",
-                ty: TypeSchema::Json,
-                comment: "Envelope with deletion status.",
-                required: true,
-            }],
-        },
-        "threads_purge" => ControllerSchema {
-            namespace: "memory",
-            function: "threads_purge",
-            description: "Remove all workspace-backed conversation JSONL files.",
-            inputs: vec![],
-            outputs: vec![FieldSchema {
-                name: "result",
-                ty: TypeSchema::Json,
-                comment: "Envelope with deleted thread/message counts.",
-                required: true,
-            }],
-        },
-
         // ----- unified memory API methods -----
         "namespace_list" => ControllerSchema {
             namespace: "memory",
@@ -1202,49 +1011,6 @@ fn handle_write_file(params: Map<String, Value>) -> ControllerFuture {
     })
 }
 
-fn handle_threads_list(_params: Map<String, Value>) -> ControllerFuture {
-    Box::pin(async move { to_json(rpc::memory_threads_list(EmptyRequest {}).await?) })
-}
-
-fn handle_thread_upsert(params: Map<String, Value>) -> ControllerFuture {
-    Box::pin(async move {
-        let payload = parse_params::<UpsertConversationThreadRequest>(params)?;
-        to_json(rpc::memory_thread_upsert(payload).await?)
-    })
-}
-
-fn handle_messages_list(params: Map<String, Value>) -> ControllerFuture {
-    Box::pin(async move {
-        let payload = parse_params::<ConversationMessagesRequest>(params)?;
-        to_json(rpc::memory_messages_list(payload).await?)
-    })
-}
-
-fn handle_message_append(params: Map<String, Value>) -> ControllerFuture {
-    Box::pin(async move {
-        let payload = parse_params::<AppendConversationMessageRequest>(params)?;
-        to_json(rpc::memory_message_append(payload).await?)
-    })
-}
-
-fn handle_message_update(params: Map<String, Value>) -> ControllerFuture {
-    Box::pin(async move {
-        let payload = parse_params::<UpdateConversationMessageRequest>(params)?;
-        to_json(rpc::memory_message_update(payload).await?)
-    })
-}
-
-fn handle_thread_delete(params: Map<String, Value>) -> ControllerFuture {
-    Box::pin(async move {
-        let payload = parse_params::<DeleteConversationThreadRequest>(params)?;
-        to_json(rpc::memory_thread_delete(payload).await?)
-    })
-}
-
-fn handle_threads_purge(_params: Map<String, Value>) -> ControllerFuture {
-    Box::pin(async move { to_json(rpc::memory_threads_purge(EmptyRequest {}).await?) })
-}
-
 fn handle_namespace_list(_params: Map<String, Value>) -> ControllerFuture {
     Box::pin(async move { to_json(rpc::namespace_list().await?) })
 }
@@ -1374,13 +1140,6 @@ mod tests {
         "list_files",
         "read_file",
         "write_file",
-        "threads_list",
-        "thread_upsert",
-        "messages_list",
-        "message_append",
-        "message_update",
-        "thread_delete",
-        "threads_purge",
         "namespace_list",
         "doc_put",
         "doc_ingest",

src/openhuman/mod.rs

@@ -51,6 +51,7 @@ pub mod socket;
 pub mod subconscious;
 pub mod team;
 pub mod text_input;
+pub mod threads;
 pub mod tool_timeout;
 pub mod tools;
 pub mod tree_summarizer;

src/openhuman/threads/mod.rs

@@ -0,0 +1,13 @@
+//! Conversation thread and message management.
+//!
+//! Thread lifecycle (create, list, delete, purge) and per-thread message
+//! CRUD. Storage delegates to `memory::conversations` JSONL files; this
+//! module owns the RPC surface and controller registry.
+
+pub mod ops;
+pub mod schemas;
+
+pub use schemas::{
+    all_controller_schemas as all_threads_controller_schemas,
+    all_registered_controllers as all_threads_registered_controllers,
+};

src/openhuman/threads/ops.rs

@@ -0,0 +1,237 @@
+//! RPC operations for conversation thread management.
+
+use crate::openhuman::channels::providers::web as web_channel;
+use crate::openhuman::config::Config;
+use crate::openhuman::memory::conversations::{
+    self, ConversationMessage, ConversationMessagePatch, ConversationThread,
+    CreateConversationThread,
+};
+use crate::openhuman::memory::{
+    ApiEnvelope, ApiMeta, AppendConversationMessageRequest, ConversationMessageRecord,
+    ConversationMessagesRequest, ConversationMessagesResponse, ConversationThreadSummary,
+    ConversationThreadsListResponse, DeleteConversationThreadRequest,
+    DeleteConversationThreadResponse, EmptyRequest, PaginationMeta,
+    PurgeConversationThreadsResponse, UpdateConversationMessageRequest,
+    UpsertConversationThreadRequest,
+};
+use crate::rpc::RpcOutcome;
+use serde::Serialize;
+use std::collections::BTreeMap;
+use std::path::PathBuf;
+
+fn request_id() -> String {
+    uuid::Uuid::new_v4().to_string()
+}
+
+fn counts(entries: impl IntoIterator<Item = (&'static str, usize)>) -> BTreeMap<String, usize> {
+    entries
+        .into_iter()
+        .map(|(k, v)| (k.to_string(), v))
+        .collect()
+}
+
+fn envelope<T: Serialize>(
+    data: T,
+    counts: Option<BTreeMap<String, usize>>,
+    pagination: Option<PaginationMeta>,
+) -> RpcOutcome<ApiEnvelope<T>> {
+    RpcOutcome::new(
+        ApiEnvelope {
+            data: Some(data),
+            error: None,
+            meta: ApiMeta {
+                request_id: request_id(),
+                latency_seconds: None,
+                cached: None,
+                counts,
+                pagination,
+            },
+        },
+        vec![],
+    )
+}
+
+async fn workspace_dir() -> Result<PathBuf, String> {
+    Config::load_or_init()
+        .await
+        .map(|c| c.workspace_dir)
+        .map_err(|e| format!("load config: {e}"))
+}
+
+fn thread_to_summary(thread: ConversationThread) -> ConversationThreadSummary {
+    ConversationThreadSummary {
+        id: thread.id,
+        title: thread.title,
+        chat_id: thread.chat_id,
+        is_active: thread.is_active,
+        message_count: thread.message_count,
+        last_message_at: thread.last_message_at,
+        created_at: thread.created_at,
+    }
+}
+
+fn message_to_record(message: ConversationMessage) -> ConversationMessageRecord {
+    ConversationMessageRecord {
+        id: message.id,
+        content: message.content,
+        message_type: message.message_type,
+        extra_metadata: message.extra_metadata,
+        sender: message.sender,
+        created_at: message.created_at,
+    }
+}
+
+fn record_to_message(record: ConversationMessageRecord) -> ConversationMessage {
+    ConversationMessage {
+        id: record.id,
+        content: record.content,
+        message_type: record.message_type,
+        extra_metadata: record.extra_metadata,
+        sender: record.sender,
+        created_at: record.created_at,
+    }
+}
+
+/// Lists all conversation threads.
+pub async fn threads_list(
+    _request: EmptyRequest,
+) -> Result<RpcOutcome<ApiEnvelope<ConversationThreadsListResponse>>, String> {
+    let dir = workspace_dir().await?;
+    let threads = conversations::list_threads(dir)?
+        .into_iter()
+        .map(thread_to_summary)
+        .collect::<Vec<_>>();
+    let count = threads.len();
+    Ok(envelope(
+        ConversationThreadsListResponse { threads, count },
+        Some(counts([("num_threads", count)])),
+        None,
+    ))
+}
+
+/// Creates or refreshes a conversation thread.
+pub async fn thread_upsert(
+    request: UpsertConversationThreadRequest,
+) -> Result<RpcOutcome<ApiEnvelope<ConversationThreadSummary>>, String> {
+    let dir = workspace_dir().await?;
+    let thread = conversations::ensure_thread(
+        dir,
+        CreateConversationThread {
+            id: request.id,
+            title: request.title,
+            created_at: request.created_at,
+        },
+    )?;
+    Ok(envelope(
+        thread_to_summary(thread),
+        Some(counts([("num_threads", 1)])),
+        None,
+    ))
+}
+
+/// Creates a new conversation thread with auto-generated ID and title.
+pub async fn thread_create_new(
+    _request: EmptyRequest,
+) -> Result<RpcOutcome<ApiEnvelope<ConversationThreadSummary>>, String> {
+    let dir = workspace_dir().await?;
+    let id = format!("thread-{}", uuid::Uuid::new_v4());
+    let now = chrono::Local::now();
+    let title = format!("Chat {} {}", now.format("%b %-d"), now.format("%-I:%M %p"));
+    let created_at = chrono::Utc::now().to_rfc3339();
+    let thread = conversations::ensure_thread(
+        dir,
+        CreateConversationThread {
+            id,
+            title,
+            created_at,
+        },
+    )?;
+    Ok(envelope(
+        thread_to_summary(thread),
+        Some(counts([("num_threads", 1)])),
+        None,
+    ))
+}
+
+/// Lists messages for a conversation thread.
+pub async fn messages_list(
+    request: ConversationMessagesRequest,
+) -> Result<RpcOutcome<ApiEnvelope<ConversationMessagesResponse>>, String> {
+    let dir = workspace_dir().await?;
+    let messages = conversations::get_messages(dir, &request.thread_id)?
+        .into_iter()
+        .map(message_to_record)
+        .collect::<Vec<_>>();
+    let count = messages.len();
+    Ok(envelope(
+        ConversationMessagesResponse { messages, count },
+        Some(counts([("num_messages", count)])),
+        None,
+    ))
+}
+
+/// Appends a message to a conversation thread.
+pub async fn message_append(
+    request: AppendConversationMessageRequest,
+) -> Result<RpcOutcome<ApiEnvelope<ConversationMessageRecord>>, String> {
+    let dir = workspace_dir().await?;
+    let message =
+        conversations::append_message(dir, &request.thread_id, record_to_message(request.message))?;
+    Ok(envelope(
+        message_to_record(message),
+        Some(counts([("num_messages", 1)])),
+        None,
+    ))
+}
+
+/// Updates metadata on an existing conversation message.
+pub async fn message_update(
+    request: UpdateConversationMessageRequest,
+) -> Result<RpcOutcome<ApiEnvelope<ConversationMessageRecord>>, String> {
+    let dir = workspace_dir().await?;
+    let message = conversations::update_message(
+        dir,
+        &request.thread_id,
+        &request.message_id,
+        ConversationMessagePatch {
+            extra_metadata: request.extra_metadata,
+        },
+    )?;
+    Ok(envelope(
+        message_to_record(message),
+        Some(counts([("num_messages", 1)])),
+        None,
+    ))
+}
+
+/// Deletes a conversation thread and its message log.
+pub async fn thread_delete(
+    request: DeleteConversationThreadRequest,
+) -> Result<RpcOutcome<ApiEnvelope<DeleteConversationThreadResponse>>, String> {
+    let dir = workspace_dir().await?;
+    let deleted = conversations::ConversationStore::new(dir)
+        .delete_thread(&request.thread_id, &request.deleted_at)?;
+    web_channel::invalidate_thread_sessions(&request.thread_id).await;
+    Ok(envelope(
+        DeleteConversationThreadResponse { deleted },
+        None,
+        None,
+    ))
+}
+
+/// Purges all conversation threads and messages.
+pub async fn threads_purge(
+    _request: EmptyRequest,
+) -> Result<RpcOutcome<ApiEnvelope<PurgeConversationThreadsResponse>>, String> {
+    let dir = workspace_dir().await?;
+    let stats = conversations::purge_threads(dir)?;
+    Ok(envelope(
+        PurgeConversationThreadsResponse {
+            messages_deleted: stats.message_count,
+            agent_threads_deleted: stats.thread_count,
+            agent_messages_deleted: stats.message_count,
+        },
+        None,
+        None,
+    ))
+}

src/openhuman/threads/schemas.rs

@@ -0,0 +1,360 @@
+//! RPC schemas and controller registration for conversation threads.
+
+use serde::de::DeserializeOwned;
+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, UpdateConversationMessageRequest, UpsertConversationThreadRequest,
+};
+
+use super::ops;
+
+pub fn all_controller_schemas() -> Vec<ControllerSchema> {
+    vec![
+        schemas("list"),
+        schemas("upsert"),
+        schemas("create_new"),
+        schemas("messages_list"),
+        schemas("message_append"),
+        schemas("message_update"),
+        schemas("delete"),
+        schemas("purge"),
+    ]
+}
+
+pub fn all_registered_controllers() -> Vec<RegisteredController> {
+    vec![
+        RegisteredController {
+            schema: schemas("list"),
+            handler: handle_list,
+        },
+        RegisteredController {
+            schema: schemas("upsert"),
+            handler: handle_upsert,
+        },
+        RegisteredController {
+            schema: schemas("create_new"),
+            handler: handle_create_new,
+        },
+        RegisteredController {
+            schema: schemas("messages_list"),
+            handler: handle_messages_list,
+        },
+        RegisteredController {
+            schema: schemas("message_append"),
+            handler: handle_message_append,
+        },
+        RegisteredController {
+            schema: schemas("message_update"),
+            handler: handle_message_update,
+        },
+        RegisteredController {
+            schema: schemas("delete"),
+            handler: handle_delete,
+        },
+        RegisteredController {
+            schema: schemas("purge"),
+            handler: handle_purge,
+        },
+    ]
+}
+
+pub fn schemas(function: &str) -> ControllerSchema {
+    match function {
+        "list" => ControllerSchema {
+            namespace: "threads",
+            function: "list",
+            description: "List conversation threads.",
+            inputs: vec![],
+            outputs: vec![FieldSchema {
+                name: "result",
+                ty: TypeSchema::Json,
+                comment: "Envelope with thread summaries and count.",
+                required: true,
+            }],
+        },
+        "upsert" => ControllerSchema {
+            namespace: "threads",
+            function: "upsert",
+            description: "Create or refresh a conversation thread.",
+            inputs: vec![
+                FieldSchema {
+                    name: "id",
+                    ty: TypeSchema::String,
+                    comment: "Stable thread identifier.",
+                    required: true,
+                },
+                FieldSchema {
+                    name: "title",
+                    ty: TypeSchema::String,
+                    comment: "Human-readable thread title.",
+                    required: true,
+                },
+                FieldSchema {
+                    name: "created_at",
+                    ty: TypeSchema::String,
+                    comment: "RFC3339 timestamp for first thread creation.",
+                    required: true,
+                },
+            ],
+            outputs: vec![FieldSchema {
+                name: "result",
+                ty: TypeSchema::Json,
+                comment: "Envelope with the resulting thread summary.",
+                required: true,
+            }],
+        },
+        "create_new" => ControllerSchema {
+            namespace: "threads",
+            function: "create_new",
+            description: "Create a new conversation thread with auto-generated ID and title.",
+            inputs: vec![],
+            outputs: vec![FieldSchema {
+                name: "result",
+                ty: TypeSchema::Json,
+                comment: "Envelope with the created thread summary.",
+                required: true,
+            }],
+        },
+        "messages_list" => ControllerSchema {
+            namespace: "threads",
+            function: "messages_list",
+            description: "List messages for a conversation thread.",
+            inputs: vec![FieldSchema {
+                name: "thread_id",
+                ty: TypeSchema::String,
+                comment: "Thread identifier.",
+                required: true,
+            }],
+            outputs: vec![FieldSchema {
+                name: "result",
+                ty: TypeSchema::Json,
+                comment: "Envelope with messages and count.",
+                required: true,
+            }],
+        },
+        "message_append" => ControllerSchema {
+            namespace: "threads",
+            function: "message_append",
+            description: "Append a message to a conversation thread.",
+            inputs: vec![
+                FieldSchema {
+                    name: "thread_id",
+                    ty: TypeSchema::String,
+                    comment: "Thread identifier.",
+                    required: true,
+                },
+                FieldSchema {
+                    name: "message",
+                    ty: TypeSchema::Json,
+                    comment: "Message payload to append.",
+                    required: true,
+                },
+            ],
+            outputs: vec![FieldSchema {
+                name: "result",
+                ty: TypeSchema::Json,
+                comment: "Envelope with the appended message payload.",
+                required: true,
+            }],
+        },
+        "message_update" => ControllerSchema {
+            namespace: "threads",
+            function: "message_update",
+            description: "Patch metadata on an existing conversation message.",
+            inputs: vec![
+                FieldSchema {
+                    name: "thread_id",
+                    ty: TypeSchema::String,
+                    comment: "Thread identifier.",
+                    required: true,
+                },
+                FieldSchema {
+                    name: "message_id",
+                    ty: TypeSchema::String,
+                    comment: "Message identifier.",
+                    required: true,
+                },
+                FieldSchema {
+                    name: "extra_metadata",
+                    ty: TypeSchema::Option(Box::new(TypeSchema::Json)),
+                    comment: "Replacement message metadata object.",
+                    required: false,
+                },
+            ],
+            outputs: vec![FieldSchema {
+                name: "result",
+                ty: TypeSchema::Json,
+                comment: "Envelope with the updated message payload.",
+                required: true,
+            }],
+        },
+        "delete" => ControllerSchema {
+            namespace: "threads",
+            function: "delete",
+            description: "Delete a conversation thread and its message log.",
+            inputs: vec![
+                FieldSchema {
+                    name: "thread_id",
+                    ty: TypeSchema::String,
+                    comment: "Thread identifier.",
+                    required: true,
+                },
+                FieldSchema {
+                    name: "deleted_at",
+                    ty: TypeSchema::String,
+                    comment: "RFC3339 deletion timestamp.",
+                    required: true,
+                },
+            ],
+            outputs: vec![FieldSchema {
+                name: "result",
+                ty: TypeSchema::Json,
+                comment: "Envelope with deletion status.",
+                required: true,
+            }],
+        },
+        "purge" => ControllerSchema {
+            namespace: "threads",
+            function: "purge",
+            description: "Remove all conversation threads and messages.",
+            inputs: vec![],
+            outputs: vec![FieldSchema {
+                name: "result",
+                ty: TypeSchema::Json,
+                comment: "Envelope with deleted thread/message counts.",
+                required: true,
+            }],
+        },
+        _other => ControllerSchema {
+            namespace: "threads",
+            function: "unknown",
+            description: "Unknown threads controller function.",
+            inputs: vec![],
+            outputs: vec![FieldSchema {
+                name: "error",
+                ty: TypeSchema::String,
+                comment: "Lookup error details.",
+                required: true,
+            }],
+        },
+    }
+}
+
+// ── Handlers ─────────────────────────────────────────────────────────
+
+fn handle_list(_params: Map<String, Value>) -> ControllerFuture {
+    Box::pin(async move { to_json(ops::threads_list(EmptyRequest {}).await?) })
+}
+
+fn handle_upsert(params: Map<String, Value>) -> ControllerFuture {
+    Box::pin(async move {
+        let p = parse::<UpsertConversationThreadRequest>(params)?;
+        to_json(ops::thread_upsert(p).await?)
+    })
+}
+
+fn handle_create_new(_params: Map<String, Value>) -> ControllerFuture {
+    Box::pin(async move { to_json(ops::thread_create_new(EmptyRequest {}).await?) })
+}
+
+fn handle_messages_list(params: Map<String, Value>) -> ControllerFuture {
+    Box::pin(async move {
+        let p = parse::<ConversationMessagesRequest>(params)?;
+        to_json(ops::messages_list(p).await?)
+    })
+}
+
+fn handle_message_append(params: Map<String, Value>) -> ControllerFuture {
+    Box::pin(async move {
+        let p = parse::<AppendConversationMessageRequest>(params)?;
+        to_json(ops::message_append(p).await?)
+    })
+}
+
+fn handle_message_update(params: Map<String, Value>) -> ControllerFuture {
+    Box::pin(async move {
+        let p = parse::<UpdateConversationMessageRequest>(params)?;
+        to_json(ops::message_update(p).await?)
+    })
+}
+
+fn handle_delete(params: Map<String, Value>) -> ControllerFuture {
+    Box::pin(async move {
+        let p = parse::<DeleteConversationThreadRequest>(params)?;
+        to_json(ops::thread_delete(p).await?)
+    })
+}
+
+fn handle_purge(_params: Map<String, Value>) -> ControllerFuture {
+    Box::pin(async move { to_json(ops::threads_purge(EmptyRequest {}).await?) })
+}
+
+// ── Helpers ──────────────────────────────────────────────────────────
+
+fn parse<T: DeserializeOwned>(params: Map<String, Value>) -> Result<T, String> {
+    serde_json::from_value(Value::Object(params)).map_err(|e| format!("invalid params: {e}"))
+}
+
+fn to_json<T: serde::Serialize>(outcome: crate::rpc::RpcOutcome<T>) -> Result<Value, String> {
+    outcome.into_cli_compatible_json()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    const ALL_FUNCTIONS: &[&str] = &[
+        "list",
+        "upsert",
+        "create_new",
+        "messages_list",
+        "message_append",
+        "message_update",
+        "delete",
+        "purge",
+    ];
+
+    #[test]
+    fn all_controller_schemas_has_entry_per_function() {
+        let names: Vec<_> = all_controller_schemas()
+            .into_iter()
+            .map(|s| s.function)
+            .collect();
+        assert_eq!(names.len(), ALL_FUNCTIONS.len());
+        for expected in ALL_FUNCTIONS {
+            assert!(names.contains(expected), "missing schema for {expected}");
+        }
+    }
+
+    #[test]
+    fn all_registered_controllers_has_handler_per_schema() {
+        let controllers = all_registered_controllers();
+        assert_eq!(controllers.len(), ALL_FUNCTIONS.len());
+        let names: Vec<_> = controllers.iter().map(|c| c.schema.function).collect();
+        for expected in ALL_FUNCTIONS {
+            assert!(names.contains(expected), "missing handler for {expected}");
+        }
+    }
+
+    #[test]
+    fn every_schema_uses_threads_namespace() {
+        for s in all_controller_schemas() {
+            assert_eq!(
+                s.namespace, "threads",
+                "schema {} wrong namespace",
+                s.function
+            );
+        }
+    }
+
+    #[test]
+    fn unknown_function_returns_fallback() {
+        let s = schemas("no_such_fn");
+        assert_eq!(s.function, "unknown");
+        assert_eq!(s.namespace, "threads");
+    }
+}