From 2bb20faa557ef5968ea78ac955f7fa0c239c11e2 Mon Sep 17 00:00:00 2001
From: Steven Enamakel <31011319+senamakel@users.noreply.github.com>
Date: Thu, 16 Apr 2026 10:16:04 -0700
Subject: [PATCH] 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.
---
 app/src/components/BottomTabBar.tsx           |  19 -
 app/src/pages/Conversations.tsx               | 174 +++++++--
 app/src/services/api/threadApi.test.ts        |   8 +-
 app/src/services/api/threadApi.ts             |  31 +-
 app/src/store/threadSlice.ts                  | 185 +++------
 src/core/all.rs                               |   4 +
 .../agent/harness/session/runtime.rs          |   8 +
 src/openhuman/channels/providers/web.rs       |  32 ++
 src/openhuman/memory/ops.rs                   | 174 +--------
 src/openhuman/memory/schemas.rs               | 243 +-----------
 src/openhuman/mod.rs                          |   1 +
 src/openhuman/threads/mod.rs                  |  13 +
 src/openhuman/threads/ops.rs                  | 237 ++++++++++++
 src/openhuman/threads/schemas.rs              | 360 ++++++++++++++++++
 14 files changed, 883 insertions(+), 606 deletions(-)
 create mode 100644 src/openhuman/threads/mod.rs
 create mode 100644 src/openhuman/threads/ops.rs
 create mode 100644 src/openhuman/threads/schemas.rs

diff --git a/app/src/components/BottomTabBar.tsx b/app/src/components/BottomTabBar.tsx
index 9f10feaac..205de87e5 100644
--- a/app/src/components/BottomTabBar.tsx
+++ b/app/src/components/BottomTabBar.tsx
@@ -1,7 +1,6 @@
 import { useLocation, useNavigate } from 'react-router-dom';
 
 import { useCoreState } from '../providers/CoreStateProvider';
-import { useAppSelector } from '../store/hooks';
 
 const tabs = [
   {
@@ -108,16 +107,6 @@ const BottomTabBar = () => {
   const { snapshot } = useCoreState();
   const token = snapshot.sessionToken;
 
-  const conversationsUnreadCount = useAppSelector(state => {
-    const { threads, lastViewedAt } = state.thread;
-    if (threads.length === 0) return 0;
-    return threads.filter(t => {
-      const viewed = lastViewedAt[t.id];
-      const lastMsg = new Date(t.lastMessageAt || t.createdAt).getTime();
-      return viewed == null || lastMsg > viewed;
-    }).length;
-  });
-
   const hiddenPaths = ['/', '/login'];
   if (
     !token ||
@@ -146,7 +135,6 @@ const BottomTabBar = () => {
       <nav className="pointer-events-auto inline-flex items-center gap-2 rounded-sm border border-stone-300 bg-stone-200 shadow-soft px-1 py-1">
         {tabs.map(tab => {
           const active = isActive(tab.path);
-          const showBadge = tab.id === 'chat' && conversationsUnreadCount > 0;
           return (
             <button
               key={tab.id}
@@ -159,13 +147,6 @@ const BottomTabBar = () => {
               aria-label={tab.label}>
               {tab.icon}
               <span>{tab.label}</span>
-              {showBadge && (
-                <span
-                  className="absolute -top-1 left-5 min-w-[16px] h-[16px] px-1 flex items-center justify-center rounded-full bg-coral-500 text-white text-[9px] font-medium"
-                  aria-label={`${conversationsUnreadCount} unread`}>
-                  {conversationsUnreadCount > 99 ? '99+' : conversationsUnreadCount}
-                </span>
-              )}
             </button>
           );
         })}
diff --git a/app/src/pages/Conversations.tsx b/app/src/pages/Conversations.tsx
index ba94d6669..e63b57c18 100644
--- a/app/src/pages/Conversations.tsx
+++ b/app/src/pages/Conversations.tsx
@@ -18,13 +18,13 @@ import { useAppDispatch, useAppSelector } from '../store/hooks';
 import { selectSocketStatus } from '../store/socketSelectors';
 import {
   addMessageLocal,
-  createThreadLocal,
+  createNewThread,
+  deleteThread,
   fetchSuggestedQuestions,
   loadThreadMessages,
   loadThreads,
   persistReaction,
   setActiveThread,
-  setLastViewed,
   setSelectedThread,
 } from '../store/threadSlice';
 import type { ThreadMessage } from '../types/thread';
@@ -38,9 +38,9 @@ import {
   openhumanVoiceTts,
 } from '../utils/tauriCommands';
 
-const DEFAULT_THREAD_ID = 'default-thread';
-const DEFAULT_THREAD_TITLE = 'Conversation';
-const AGENTIC_MODEL_ID = 'agentic-v1';
+// Chat uses the reasoning model; `agentic-v1` is reserved for sub-agents
+// that execute tool calls, not the primary user-facing conversation.
+const CHAT_MODEL_ID = 'reasoning-v1';
 /** Maximum trailing characters rendered in the live-streaming assistant
  *  preview bubble. The full response is revealed via `addInferenceResponse`
  *  on `chat_done` — this is purely a ticker-tape affordance to signal
@@ -150,6 +150,7 @@ const Conversations = () => {
   const dispatch = useAppDispatch();
   const navigate = useNavigate();
   const {
+    threads,
     selectedThreadId,
     messages,
     isLoadingMessages,
@@ -159,6 +160,7 @@ const Conversations = () => {
     activeThreadId,
   } = useAppSelector(state => state.thread);
 
+  const [showSidebar, setShowSidebar] = useState(true);
   const [inputValue, setInputValue] = useState('');
   const [copiedMessageId, setCopiedMessageId] = useState<string | null>(null);
   const [inputMode, setInputMode] = useState<InputMode>('text');
@@ -222,23 +224,26 @@ const Conversations = () => {
     typeof navigator.mediaDevices !== 'undefined' &&
     typeof navigator.mediaDevices.getUserMedia === 'function';
 
-  useEffect(() => {
-    void dispatch(loadThreads());
-    void dispatch(
-      createThreadLocal({
-        id: DEFAULT_THREAD_ID,
-        title: DEFAULT_THREAD_TITLE,
-        createdAt: new Date().toISOString(),
-      })
-    ).then(() => {
-      dispatch(setSelectedThread(DEFAULT_THREAD_ID));
-      void dispatch(loadThreadMessages(DEFAULT_THREAD_ID));
-    });
-  }, [dispatch]);
+  const handleCreateNewThread = async () => {
+    const thread = await dispatch(createNewThread()).unwrap();
+    dispatch(setSelectedThread(thread.id));
+    void dispatch(loadThreadMessages(thread.id));
+  };
 
   useEffect(() => {
-    if (selectedThreadId) dispatch(setLastViewed(selectedThreadId));
-  }, [selectedThreadId, dispatch]);
+    void dispatch(loadThreads())
+      .unwrap()
+      .then(data => {
+        if (data.threads.length > 0) {
+          const mostRecent = data.threads[0];
+          dispatch(setSelectedThread(mostRecent.id));
+          void dispatch(loadThreadMessages(mostRecent.id));
+        } else {
+          void handleCreateNewThread();
+        }
+      });
+    // eslint-disable-next-line react-hooks/exhaustive-deps
+  }, [dispatch]);
 
   useEffect(() => {
     if (selectedThreadId) {
@@ -376,11 +381,24 @@ const Conversations = () => {
     };
   }, [inputMode, rustChat]);
 
+  const handleSlashCommand = (command: string): boolean => {
+    const cmd = command.toLowerCase();
+    if (cmd === '/new' || cmd === '/clear') {
+      setInputValue('');
+      void handleCreateNewThread();
+      return true;
+    }
+    return false;
+  };
+
   const handleSendMessage = async (text?: string) => {
     const normalized = text ?? inputValue;
     const trimmed = normalized.trim();
 
     if (!trimmed || !selectedThreadId || composerBlocked) return;
+
+    if (handleSlashCommand(trimmed)) return;
+
     if (isAtLimit) {
       setShowLimitModal(true);
       setSendError(
@@ -437,7 +455,7 @@ const Conversations = () => {
     // Local model (Ollama) is used only for supplementary features
     // (auto-react, autocomplete, etc.) — never as a primary chat path.
     try {
-      await chatSend({ threadId: sendingThreadId, message: trimmed, model: AGENTIC_MODEL_ID });
+      await chatSend({ threadId: sendingThreadId, message: trimmed, model: CHAT_MODEL_ID });
 
       // Active-thread reset happens in the global ChatRuntimeProvider events.
     } catch (err) {
@@ -685,9 +703,121 @@ const Conversations = () => {
       inferenceTurnLifecycleByThread[selectedThreadId] === 'streaming')
   );
 
+  const sortedThreads = [...threads].sort(
+    (a, b) => new Date(b.lastMessageAt).getTime() - new Date(a.lastMessageAt).getTime()
+  );
+
   return (
-    <div className="h-full relative z-10 flex justify-center overflow-hidden p-4 pt-6">
+    <div className="h-full relative z-10 flex overflow-hidden p-4 pt-6 gap-3">
+      {/* Thread sidebar */}
+      {showSidebar && (
+        <div className="w-64 flex-shrink-0 flex flex-col bg-white rounded-2xl shadow-soft border border-stone-200 overflow-hidden">
+          <div className="flex items-center justify-between px-4 py-3 border-b border-stone-100">
+            <h2 className="text-sm font-semibold text-stone-700">Threads</h2>
+            <button
+              onClick={() => void handleCreateNewThread()}
+              className="w-7 h-7 flex items-center justify-center rounded-lg hover:bg-stone-100 text-stone-500 hover:text-stone-700 transition-colors"
+              title="New thread">
+              <svg className="w-4 h-4" fill="none" stroke="currentColor" viewBox="0 0 24 24">
+                <path
+                  strokeLinecap="round"
+                  strokeLinejoin="round"
+                  strokeWidth={2}
+                  d="M12 4v16m8-8H4"
+                />
+              </svg>
+            </button>
+          </div>
+          <div className="flex-1 overflow-y-auto">
+            {sortedThreads.length === 0 ? (
+              <p className="px-4 py-6 text-xs text-stone-400 text-center">No threads yet</p>
+            ) : (
+              sortedThreads.map(thread => (
+                <button
+                  key={thread.id}
+                  onClick={() => {
+                    dispatch(setSelectedThread(thread.id));
+                    void dispatch(loadThreadMessages(thread.id));
+                  }}
+                  className={`w-full text-left px-4 py-3 border-b border-stone-50 transition-colors group ${
+                    selectedThreadId === thread.id
+                      ? 'bg-primary-50 border-l-2 border-l-primary-500'
+                      : 'hover:bg-stone-50'
+                  }`}>
+                  <div className="flex items-center justify-between">
+                    <p
+                      className={`text-sm truncate flex-1 ${
+                        selectedThreadId === thread.id
+                          ? 'font-medium text-primary-700'
+                          : 'text-stone-700'
+                      }`}>
+                      {thread.title}
+                    </p>
+                    <button
+                      onClick={e => {
+                        e.stopPropagation();
+                        void dispatch(deleteThread(thread.id));
+                      }}
+                      className="ml-2 p-1 rounded opacity-0 group-hover:opacity-100 hover:bg-stone-200 text-stone-400 hover:text-coral-500 transition-all flex-shrink-0"
+                      title="Delete thread">
+                      <svg
+                        className="w-3 h-3"
+                        fill="none"
+                        stroke="currentColor"
+                        viewBox="0 0 24 24">
+                        <path
+                          strokeLinecap="round"
+                          strokeLinejoin="round"
+                          strokeWidth={2}
+                          d="M6 18L18 6M6 6l12 12"
+                        />
+                      </svg>
+                    </button>
+                  </div>
+                  <div className="flex items-center gap-2 mt-0.5">
+                    <span className="text-[10px] text-stone-400">
+                      {formatRelativeTime(thread.lastMessageAt)}
+                    </span>
+                    {thread.messageCount > 0 && (
+                      <span className="text-[10px] text-stone-400">
+                        {thread.messageCount} msg{thread.messageCount !== 1 ? 's' : ''}
+                      </span>
+                    )}
+                  </div>
+                </button>
+              ))
+            )}
+          </div>
+        </div>
+      )}
+
+      {/* Main chat area */}
       <div className="flex-1 flex flex-col min-w-0 max-w-2xl bg-white rounded-2xl shadow-soft border border-stone-200 overflow-hidden">
+        {/* Chat header */}
+        <div className="flex items-center gap-2 px-4 py-2.5 border-b border-stone-100">
+          <button
+            onClick={() => setShowSidebar(prev => !prev)}
+            className="w-7 h-7 flex items-center justify-center rounded-lg hover:bg-stone-100 text-stone-500 hover:text-stone-700 transition-colors"
+            title={showSidebar ? 'Hide sidebar' : 'Show sidebar'}>
+            <svg className="w-4 h-4" fill="none" stroke="currentColor" viewBox="0 0 24 24">
+              <path
+                strokeLinecap="round"
+                strokeLinejoin="round"
+                strokeWidth={2}
+                d="M4 6h16M4 12h16M4 18h16"
+              />
+            </svg>
+          </button>
+          <h3 className="text-sm font-medium text-stone-700 truncate flex-1">
+            {threads.find(t => t.id === selectedThreadId)?.title ?? 'Select a thread'}
+          </h3>
+          <button
+            onClick={() => void handleCreateNewThread()}
+            className="px-2.5 py-1 rounded-lg text-xs font-medium text-primary-600 hover:bg-primary-50 transition-colors"
+            title="New thread (/new)">
+            + New
+          </button>
+        </div>
         <div className="flex-1 overflow-y-auto px-5 py-4 bg-stone-50">
           {isLoadingMessages ? (
             <div className="space-y-4">
diff --git a/app/src/services/api/threadApi.test.ts b/app/src/services/api/threadApi.test.ts
index c28d8d5c8..635c5208d 100644
--- a/app/src/services/api/threadApi.test.ts
+++ b/app/src/services/api/threadApi.test.ts
@@ -11,7 +11,7 @@ describe('threadApi', () => {
     mockCallCoreRpc.mockReset();
   });
 
-  it('loads threads from the memory RPC store', async () => {
+  it('loads threads from the threads RPC store', async () => {
     mockCallCoreRpc.mockResolvedValueOnce({
       data: {
         threads: [
@@ -32,12 +32,12 @@ describe('threadApi', () => {
     const { threadApi } = await import('./threadApi');
     const result = await threadApi.getThreads();
 
-    expect(mockCallCoreRpc).toHaveBeenCalledWith({ method: 'openhuman.memory_threads_list' });
+    expect(mockCallCoreRpc).toHaveBeenCalledWith({ method: 'openhuman.threads_list' });
     expect(result.count).toBe(1);
     expect(result.threads[0].id).toBe('default-thread');
   });
 
-  it('appends a message via memory RPC', async () => {
+  it('appends a message via threads RPC', async () => {
     const message = {
       id: 'm1',
       content: 'hello',
@@ -52,7 +52,7 @@ describe('threadApi', () => {
     const result = await threadApi.appendMessage('default-thread', message);
 
     expect(mockCallCoreRpc).toHaveBeenCalledWith({
-      method: 'openhuman.memory_message_append',
+      method: 'openhuman.threads_message_append',
       params: { thread_id: 'default-thread', message },
     });
     expect(result).toEqual(message);
diff --git a/app/src/services/api/threadApi.ts b/app/src/services/api/threadApi.ts
index c270f2c0e..d1fdceb41 100644
--- a/app/src/services/api/threadApi.ts
+++ b/app/src/services/api/threadApi.ts
@@ -1,7 +1,6 @@
 import type {
   PurgeResultData,
   Thread,
-  ThreadCreateData,
   ThreadDeleteData,
   ThreadMessage,
   ThreadMessagesData,
@@ -21,29 +20,23 @@ function unwrapEnvelope<T>(response: Envelope<T> | T): T {
 }
 
 export const threadApi = {
-  getThreads: async (): Promise<ThreadsListData> => {
-    const response = await callCoreRpc<Envelope<ThreadsListData>>({
-      method: 'openhuman.memory_threads_list',
+  createNewThread: async (): Promise<Thread> => {
+    const response = await callCoreRpc<Envelope<Thread>>({
+      method: 'openhuman.threads_create_new',
     });
     return unwrapEnvelope(response);
   },
 
-  createThread: async (input: {
-    id: string;
-    title: string;
-    createdAt: string;
-  }): Promise<ThreadCreateData> => {
-    const response = await callCoreRpc<Envelope<Thread>>({
-      method: 'openhuman.memory_thread_upsert',
-      params: { id: input.id, title: input.title, created_at: input.createdAt },
+  getThreads: async (): Promise<ThreadsListData> => {
+    const response = await callCoreRpc<Envelope<ThreadsListData>>({
+      method: 'openhuman.threads_list',
     });
-    const thread = unwrapEnvelope(response);
-    return { id: thread.id };
+    return unwrapEnvelope(response);
   },
 
   getThreadMessages: async (threadId: string): Promise<ThreadMessagesData> => {
     const response = await callCoreRpc<Envelope<ThreadMessagesData>>({
-      method: 'openhuman.memory_messages_list',
+      method: 'openhuman.threads_messages_list',
       params: { thread_id: threadId },
     });
     return unwrapEnvelope(response);
@@ -51,7 +44,7 @@ export const threadApi = {
 
   appendMessage: async (threadId: string, message: ThreadMessage): Promise<ThreadMessage> => {
     const response = await callCoreRpc<Envelope<ThreadMessage>>({
-      method: 'openhuman.memory_message_append',
+      method: 'openhuman.threads_message_append',
       params: { thread_id: threadId, message },
     });
     return unwrapEnvelope(response);
@@ -63,7 +56,7 @@ export const threadApi = {
     extraMetadata: Record<string, unknown>
   ): Promise<ThreadMessage> => {
     const response = await callCoreRpc<Envelope<ThreadMessage>>({
-      method: 'openhuman.memory_message_update',
+      method: 'openhuman.threads_message_update',
       params: { thread_id: threadId, message_id: messageId, extra_metadata: extraMetadata },
     });
     return unwrapEnvelope(response);
@@ -71,7 +64,7 @@ export const threadApi = {
 
   deleteThread: async (threadId: string): Promise<ThreadDeleteData> => {
     const response = await callCoreRpc<Envelope<ThreadDeleteData>>({
-      method: 'openhuman.memory_thread_delete',
+      method: 'openhuman.threads_delete',
       params: { thread_id: threadId, deleted_at: new Date().toISOString() },
     });
     return unwrapEnvelope(response);
@@ -79,7 +72,7 @@ export const threadApi = {
 
   purge: async (): Promise<PurgeResultData> => {
     const response = await callCoreRpc<Envelope<PurgeResultData>>({
-      method: 'openhuman.memory_threads_purge',
+      method: 'openhuman.threads_purge',
     });
     return unwrapEnvelope(response);
   },
diff --git a/app/src/store/threadSlice.ts b/app/src/store/threadSlice.ts
index 1977b1e98..2b52395e4 100644
--- a/app/src/store/threadSlice.ts
+++ b/app/src/store/threadSlice.ts
@@ -7,41 +7,27 @@ import { isTauri, openhumanLocalAiSuggestQuestions } from '../utils/tauriCommand
 interface ThreadState {
   threads: Thread[];
   selectedThreadId: string | null;
-  panelWidth: number;
-  lastViewedAt: Record<string, number>;
   activeThreadId: string | null;
   messagesByThreadId: Record<string, ThreadMessage[]>;
   messages: ThreadMessage[];
   isLoadingThreads: boolean;
   isLoadingMessages: boolean;
   messagesError: string | null;
-  sendStatus: 'idle' | 'loading' | 'success' | 'error';
-  sendError: string | null;
-  deleteStatus: 'idle' | 'loading' | 'success' | 'error';
-  purgeStatus: 'idle' | 'loading' | 'success' | 'error';
   suggestedQuestions: Array<{ text: string; confidence: number }>;
   isLoadingSuggestions: boolean;
-  suggestError: string | null;
 }
 
 const initialState: ThreadState = {
   threads: [],
   selectedThreadId: null,
-  panelWidth: 320,
-  lastViewedAt: {},
   activeThreadId: null,
   messagesByThreadId: {},
   messages: [],
   isLoadingThreads: false,
   isLoadingMessages: false,
   messagesError: null,
-  sendStatus: 'idle',
-  sendError: null,
-  deleteStatus: 'idle',
-  purgeStatus: 'idle',
   suggestedQuestions: [],
   isLoadingSuggestions: false,
-  suggestError: null,
 };
 
 function appendMessageToCache(
@@ -51,37 +37,18 @@ function appendMessageToCache(
   replaceExisting = false
 ) {
   const existing = state.messagesByThreadId[threadId] ?? [];
-  const nextStored = replaceExisting
-    ? existing.map(entry => (entry.id === message.id ? message : entry))
+  const next = replaceExisting
+    ? existing.map(e => (e.id === message.id ? message : e))
     : [...existing, message];
-  state.messagesByThreadId[threadId] = nextStored;
-
+  state.messagesByThreadId[threadId] = next;
   if (threadId === state.selectedThreadId) {
     state.messages = replaceExisting
-      ? state.messages.map(entry => (entry.id === message.id ? message : entry))
+      ? state.messages.map(e => (e.id === message.id ? message : e))
       : [...state.messages, message];
   }
-
-  const thread = state.threads.find(entry => entry.id === threadId);
-  if (thread) {
-    thread.messageCount = nextStored.length;
-    thread.lastMessageAt =
-      nextStored.length > 0 ? nextStored[nextStored.length - 1].createdAt : thread.createdAt;
-  }
 }
 
-function replaceMessagesForThread(state: ThreadState, threadId: string, messages: ThreadMessage[]) {
-  state.messagesByThreadId[threadId] = messages;
-  if (threadId === state.selectedThreadId) {
-    state.messages = messages;
-  }
-  const thread = state.threads.find(entry => entry.id === threadId);
-  if (thread) {
-    thread.messageCount = messages.length;
-    thread.lastMessageAt =
-      messages.length > 0 ? messages[messages.length - 1].createdAt : thread.createdAt;
-  }
-}
+// ── Async thunks (thin RPC wrappers) ──────────────────────────────
 
 export const loadThreads = createAsyncThunk(
   'thread/loadThreads',
@@ -94,22 +61,41 @@ export const loadThreads = createAsyncThunk(
   }
 );
 
-export const createThreadLocal = createAsyncThunk(
-  'thread/createThreadLocal',
-  async (
-    payload: { id: string; title: string; createdAt: string },
-    { dispatch, rejectWithValue }
-  ) => {
+export const createNewThread = createAsyncThunk(
+  'thread/createNewThread',
+  async (_, { dispatch, rejectWithValue }) => {
     try {
-      const created = await threadApi.createThread(payload);
+      const thread = await threadApi.createNewThread();
       await dispatch(loadThreads()).unwrap();
-      return created;
+      return thread;
     } catch (error) {
       return rejectWithValue(error instanceof Error ? error.message : 'Failed to create thread');
     }
   }
 );
 
+export const deleteThread = createAsyncThunk(
+  'thread/deleteThread',
+  async (threadId: string, { dispatch, getState, rejectWithValue }) => {
+    try {
+      await threadApi.deleteThread(threadId);
+      const state = getState() as { thread: ThreadState };
+      if (state.thread.selectedThreadId === threadId) {
+        const remaining = state.thread.threads.filter(t => t.id !== threadId);
+        if (remaining.length > 0) {
+          dispatch(setSelectedThread(remaining[0].id));
+        } else {
+          dispatch(clearSelectedThread());
+        }
+      }
+      await dispatch(loadThreads()).unwrap();
+      return { threadId };
+    } catch (error) {
+      return rejectWithValue(error instanceof Error ? error.message : 'Failed to delete thread');
+    }
+  }
+);
+
 export const loadThreadMessages = createAsyncThunk(
   'thread/loadThreadMessages',
   async (threadId: string, { rejectWithValue }) => {
@@ -142,9 +128,7 @@ export const addInferenceResponse = createAsyncThunk(
   ) => {
     const state = getState() as { thread: ThreadState };
     const targetThreadId = payload.threadId ?? state.thread.activeThreadId;
-    if (!targetThreadId) {
-      return rejectWithValue('No target thread for inference response');
-    }
+    if (!targetThreadId) return rejectWithValue('No target thread');
 
     const message: ThreadMessage = {
       id: payload.messageId ?? `inference-${Date.now()}-${Math.random()}`,
@@ -172,15 +156,12 @@ export const persistReaction = createAsyncThunk(
   ) => {
     const state = getState() as { thread: ThreadState };
     const stored = state.thread.messagesByThreadId[payload.threadId] ?? [];
-    const message = stored.find(entry => entry.id === payload.messageId);
-    if (!message) {
-      return rejectWithValue('Message not found for reaction update');
-    }
+    const message = stored.find(e => e.id === payload.messageId);
+    if (!message) return rejectWithValue('Message not found');
 
     const prev = (message.extraMetadata['myReactions'] as string[] | undefined) ?? [];
     const idx = prev.indexOf(payload.emoji);
-    const next =
-      idx >= 0 ? prev.filter(entry => entry !== payload.emoji) : [...prev, payload.emoji];
+    const next = idx >= 0 ? prev.filter(e => e !== payload.emoji) : [...prev, payload.emoji];
     const extraMetadata = { ...message.extraMetadata, myReactions: next };
 
     try {
@@ -214,19 +195,16 @@ export const fetchSuggestedQuestions = createAsyncThunk(
   async (conversationId: string | undefined, { getState, rejectWithValue }) => {
     try {
       const state = getState() as { thread: ThreadState };
-      const selectedThreadId = conversationId ?? state.thread.selectedThreadId ?? undefined;
-      const threadMessages = selectedThreadId
-        ? (state.thread.messagesByThreadId[selectedThreadId] ?? [])
-        : [];
+      const tid = conversationId ?? state.thread.selectedThreadId ?? undefined;
+      const msgs = tid ? (state.thread.messagesByThreadId[tid] ?? []) : [];
 
       if (isTauri()) {
-        const lines = threadMessages
+        const lines = msgs
           .slice(-24)
-          .map(msg => `${msg.sender === 'user' ? 'User' : 'Assistant'}: ${msg.content}`);
+          .map(m => `${m.sender === 'user' ? 'User' : 'Assistant'}: ${m.content}`);
         const local = await openhumanLocalAiSuggestQuestions(undefined, lines);
         return local.result;
       }
-
       return [];
     } catch (error) {
       return rejectWithValue(
@@ -236,6 +214,8 @@ export const fetchSuggestedQuestions = createAsyncThunk(
   }
 );
 
+// ── Slice ─────────────────────────────────────────────────────────
+
 const threadSlice = createSlice({
   name: 'thread',
   initialState,
@@ -245,45 +225,23 @@ const threadSlice = createSlice({
       state.messages = state.messagesByThreadId[action.payload] ?? [];
       state.messagesError = null;
       state.suggestedQuestions = [];
-      state.suggestError = null;
     },
     clearSelectedThread: state => {
       state.selectedThreadId = null;
       state.messages = [];
       state.messagesError = null;
       state.suggestedQuestions = [];
-      state.suggestError = null;
     },
-    clearSuggestedQuestions: state => {
-      state.suggestedQuestions = [];
-      state.suggestError = null;
-    },
-    clearDeleteStatus: state => {
-      state.deleteStatus = 'idle';
-    },
-    clearPurgeStatus: state => {
-      state.purgeStatus = 'idle';
-    },
-    clearSendError: state => {
-      state.sendError = null;
-    },
-    setPanelWidth: (state, action: { payload: number }) => {
-      state.panelWidth = action.payload;
-    },
-    setLastViewed: (state, action: { payload: string }) => {
-      state.lastViewedAt[action.payload] = Date.now();
+    setActiveThread: (state, action: { payload: string | null }) => {
+      state.activeThreadId = action.payload;
     },
     clearAllThreads: state => {
       state.threads = [];
       state.messagesByThreadId = {};
       state.selectedThreadId = null;
       state.messages = [];
-      state.lastViewedAt = {};
       state.activeThreadId = null;
     },
-    setActiveThread: (state, action: { payload: string | null }) => {
-      state.activeThreadId = action.payload;
-    },
   },
   extraReducers: builder => {
     builder
@@ -297,48 +255,31 @@ const threadSlice = createSlice({
       .addCase(loadThreads.rejected, state => {
         state.isLoadingThreads = false;
       })
-      .addCase(createThreadLocal.pending, state => {
-        state.isLoadingThreads = true;
-      })
-      .addCase(createThreadLocal.fulfilled, state => {
-        state.isLoadingThreads = false;
-      })
-      .addCase(createThreadLocal.rejected, (state, action) => {
-        state.isLoadingThreads = false;
-        state.messagesError = action.payload as string;
-      })
       .addCase(loadThreadMessages.pending, state => {
         state.isLoadingMessages = true;
         state.messagesError = null;
       })
       .addCase(loadThreadMessages.fulfilled, (state, action) => {
         state.isLoadingMessages = false;
-        replaceMessagesForThread(state, action.payload.threadId, action.payload.messages);
+        state.messagesByThreadId[action.payload.threadId] = action.payload.messages;
+        if (action.payload.threadId === state.selectedThreadId) {
+          state.messages = action.payload.messages;
+        }
       })
       .addCase(loadThreadMessages.rejected, (state, action) => {
         state.isLoadingMessages = false;
         state.messagesError = action.payload as string;
       })
-      .addCase(addMessageLocal.pending, state => {
-        state.sendStatus = 'loading';
-        state.sendError = null;
-      })
       .addCase(addMessageLocal.fulfilled, (state, action) => {
-        state.sendStatus = 'success';
         appendMessageToCache(state, action.payload.threadId, action.payload.message);
       })
-      .addCase(addMessageLocal.rejected, (state, action) => {
-        state.sendStatus = 'error';
-        state.sendError = action.payload as string;
-      })
       .addCase(addInferenceResponse.fulfilled, (state, action) => {
         appendMessageToCache(state, action.payload.threadId, action.payload.message);
         // Do not clear activeThreadId here: streaming sends many segment append
         // thunks; clearing each time would re-enable the composer mid-turn.
         // ChatRuntimeProvider clears it on chat_done / chat_error.
       })
-      .addCase(addInferenceResponse.rejected, (state, action) => {
-        state.sendError = action.payload as string;
+      .addCase(addInferenceResponse.rejected, () => {
         // Do NOT clear activeThreadId here — ChatRuntimeProvider clears it on
         // chat_done / chat_error. Clearing on every rejected segment append
         // would re-enable the composer while the turn is still in-flight.
@@ -346,42 +287,24 @@ const threadSlice = createSlice({
       .addCase(persistReaction.fulfilled, (state, action) => {
         appendMessageToCache(state, action.payload.threadId, action.payload.message, true);
       })
-      .addCase(purgeThreads.pending, state => {
-        state.purgeStatus = 'loading';
-      })
-      .addCase(purgeThreads.fulfilled, state => {
-        state.purgeStatus = 'success';
-      })
-      .addCase(purgeThreads.rejected, state => {
-        state.purgeStatus = 'error';
+      .addCase(deleteThread.fulfilled, (state, action) => {
+        delete state.messagesByThreadId[action.payload.threadId];
       })
       .addCase(fetchSuggestedQuestions.pending, state => {
         state.isLoadingSuggestions = true;
-        state.suggestError = null;
       })
       .addCase(fetchSuggestedQuestions.fulfilled, (state, action) => {
         state.isLoadingSuggestions = false;
         state.suggestedQuestions = action.payload;
       })
-      .addCase(fetchSuggestedQuestions.rejected, (state, action) => {
+      .addCase(fetchSuggestedQuestions.rejected, state => {
         state.isLoadingSuggestions = false;
-        state.suggestError = action.payload as string;
         state.suggestedQuestions = [];
       });
   },
 });
 
-export const {
-  setSelectedThread,
-  clearSelectedThread,
-  clearDeleteStatus,
-  clearPurgeStatus,
-  clearSendError,
-  clearSuggestedQuestions,
-  setPanelWidth,
-  setLastViewed,
-  clearAllThreads,
-  setActiveThread,
-} = threadSlice.actions;
+export const { setSelectedThread, clearSelectedThread, setActiveThread, clearAllThreads } =
+  threadSlice.actions;
 
 export default threadSlice.reducer;
diff --git a/src/core/all.rs b/src/core/all.rs
index ce65c35b8..d529db0f7 100644
--- a/src/core/all.rs
+++ b/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
 }
 
diff --git a/src/openhuman/agent/harness/session/runtime.rs b/src/openhuman/agent/harness/session/runtime.rs
index 7cb6c1440..90c387a8a 100644
--- a/src/openhuman/agent/harness/session/runtime.rs
+++ b/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
diff --git a/src/openhuman/channels/providers/web.rs b/src/openhuman/channels/providers/web.rs
index d92a32ee2..480c1f09f 100644
--- a/src/openhuman/channels/providers/web.rs
+++ b/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())
diff --git a/src/openhuman/memory/ops.rs b/src/openhuman/memory/ops.rs
index c28a3803c..82c30db66 100644
--- a/src/openhuman/memory/ops.rs
+++ b/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,
diff --git a/src/openhuman/memory/schemas.rs b/src/openhuman/memory/schemas.rs
index 297104ede..42859855d 100644
--- a/src/openhuman/memory/schemas.rs
+++ b/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",
diff --git a/src/openhuman/mod.rs b/src/openhuman/mod.rs
index b7a677ed6..c30c92a0f 100644
--- a/src/openhuman/mod.rs
+++ b/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;
diff --git a/src/openhuman/threads/mod.rs b/src/openhuman/threads/mod.rs
new file mode 100644
index 000000000..731276b6b
--- /dev/null
+++ b/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,
+};
diff --git a/src/openhuman/threads/ops.rs b/src/openhuman/threads/ops.rs
new file mode 100644
index 000000000..823f5ae6c
--- /dev/null
+++ b/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,
+    ))
+}
diff --git a/src/openhuman/threads/schemas.rs b/src/openhuman/threads/schemas.rs
new file mode 100644
index 000000000..94b1868aa
--- /dev/null
+++ b/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");
+    }
+}