th-d5b446: Python server list_conversations + resume-by-conversationId

.changeset/python-server-list-resume.md

@@ -0,0 +1,9 @@
+---
+"@smooai/smooth-operator": patch
+---
+
+Python server: `list_conversations` + resume-by-`conversationId` (pearl th-d5b446) — Python parity with the merged Rust/Go/TS reference so every client (daemon PWA, `th code` TUI, chat-widget) can build a conversation sidebar + resume against the Python server too.
+
+- New WS action `list_conversations` (`{action, requestId, limit?}`, default limit 50): replies via `immediate_response` (200, "Conversations") with `{ conversations: [ { conversationId, title, updatedAt, messageCount } ] }`, most-recent-first, filtered to conversations with `messageCount > 0` (drops the empty conversations every page-load mints). `title` = a ~60-char preview of the first inbound message with leading markdown/control chars stripped, falling back to a generic name; `updatedAt` = ISO-8601.
+- `create_conversation_session` gains an optional `conversationId`: when it names a known conversation, the new session RESUMES — reuses that conversation's id and keeps its message log, so `send_message` appends to it and the runner replays its history. Absent/unknown id ⇒ a fresh conversation is minted (unchanged behavior).
+- Additive and back-compat: no `conversationId` / no `list_conversations` call = unchanged behavior. `SessionStore` gains `list_conversations()` + an optional `conversation_id` arg on `create_session`; the in-memory store tracks per-conversation last-activity for the sort key.

python/server/src/smooth_operator_server/dispatcher.py

@@ -113,6 +113,8 @@ async def dispatch(self, raw_frame: str, sink: Sink) -> None:
                 sink(protocol.pong(request_id))
             elif action == "create_conversation_session":
                 await self._handle_create_session(frame, request_id, sink)
+            elif action == "list_conversations":
+                await self._handle_list_conversations(frame, request_id, sink)
             elif action == "get_session":
                 await self._handle_get_session(frame, request_id, sink)
             elif action == "send_message":
@@ -132,10 +134,17 @@ async def dispatch(self, raw_frame: str, sink: Sink) -> None:
             sink(protocol.error(request_id, "INTERNAL_ERROR", "Internal error processing the request."))
 
     async def _handle_create_session(self, frame: dict, request_id: str | None, sink: Sink) -> None:
+        # Resume: a `conversationId` naming an existing conversation binds the new session
+        # to it (reuses id + history); absent/unknown → a fresh conversation. The response
+        # echoes conversationId either way, so a resuming client sees the id it passed.
+        # Mirrors the Rust/Go/TS reference. th-d5b446.
+        raw_conv_id = frame.get("conversationId")
+        conversation_id = raw_conv_id if isinstance(raw_conv_id, str) and raw_conv_id else None
         session = await self._store.create_session(
             frame.get("agentId") or "",
             frame.get("userName"),
             frame.get("userEmail"),
+            conversation_id,
         )
         data = {
             "sessionId": session.session_id,
@@ -166,6 +175,35 @@ async def _handle_get_session(self, frame: dict, request_id: str | None, sink: S
         }
         sink(protocol.immediate_response(request_id, 200, "Session", data))
 
+    async def _handle_list_conversations(self, frame: dict, request_id: str | None, sink: Sink) -> None:
+        """``list_conversations`` — the conversation-sidebar / resume substrate. Returns
+        the store's conversations that have at least one message (empties, minted every
+        page-load, are dropped), most-recent-first, each ``{conversationId, title,
+        updatedAt, messageCount}`` where ``title`` is a first-inbound preview. Optional
+        input: ``limit`` (default 50). A client resumes one by passing its
+        ``conversationId`` to ``create_conversation_session``. Mirrors the Rust/Go/TS
+        reference. th-d5b446."""
+        raw_limit = frame.get("limit")
+        limit = (
+            raw_limit
+            if isinstance(raw_limit, int) and not isinstance(raw_limit, bool) and raw_limit > 0
+            else _DEFAULT_LIST_LIMIT
+        )
+
+        summaries = await self._store.list_conversations()
+        # Most-recent-first (empties already dropped by the store), then cap.
+        summaries.sort(key=lambda c: c.updated_at, reverse=True)
+        conversations = [
+            {
+                "conversationId": c.conversation_id,
+                "title": _conversation_title(c.first_inbound_text, f"Conversation {c.conversation_id}"),
+                "updatedAt": c.updated_at.isoformat(),
+                "messageCount": c.message_count,
+            }
+            for c in summaries[:limit]
+        ]
+        sink(protocol.immediate_response(request_id, 200, "Conversations", {"conversations": conversations}))
+
     async def _handle_send_message(self, frame: dict, request_id: str | None, sink: Sink) -> None:
         # requestId is load-bearing for streaming correlation; generate one if the
         # client omitted it (mirrors the C# `requestId ??= Guid`).
@@ -405,3 +443,48 @@ def _handle_confirm_tool_action(self, frame: dict, request_id: str | None, sink:
                 {"sessionId": session_id, "approved": approved},
             )
         )
+
+
+#: Default cap for list_conversations when the caller doesn't ask for a specific limit.
+_DEFAULT_LIST_LIMIT = 50
+
+#: Max characters in a conversation title preview before it's clipped with an ellipsis.
+_TITLE_MAX = 60
+
+#: Leading markdown/control markers stripped off a preview so a raw message body renders
+#: as clean text (heading #, bullets *-, quote >, emphasis _~, code `, cursor ▎). Only
+#: LEADING chars are touched — inline markdown mid-text is left alone.
+_LEADING_MARKUP = "#*>-_~`▎ "
+
+
+def _conversation_title(first_inbound_text: str | None, fallback: str) -> str:
+    """Derive a ``list_conversations`` entry title from the first inbound message text,
+    falling back to ``fallback`` (a conversation name) when there is none. The preview is
+    cleaned (leading markdown/control chars stripped) and clipped to ``_TITLE_MAX`` chars
+    with an ellipsis. Mirrors the Rust ``conversation_title`` + ``truncate_preview``, plus
+    the contract's leading-markdown strip (matching Go/TS). th-d5b446."""
+    cleaned = _strip_leading_markup(first_inbound_text or "")
+    if not cleaned:
+        return fallback
+    return _truncate_preview(cleaned, _TITLE_MAX)
+
+
+def _strip_leading_markup(s: str) -> str:
+    """Trim leading whitespace, control chars, and markdown markers off a preview so
+    ``"### Hi"`` / ``"- do X"`` title as ``"Hi"`` / ``"do X"``."""
+    i = 0
+    for ch in s:
+        if ch.isspace() or (ch.isprintable() is False) or ch in _LEADING_MARKUP:
+            i += 1
+        else:
+            break
+    return s[i:].strip()
+
+
+def _truncate_preview(s: str, max_chars: int) -> str:
+    """Trim ``s`` and clip it to ``max_chars`` (char-safe), appending an ellipsis when
+    clipped. Mirrors the Rust ``truncate_preview``."""
+    s = s.strip()
+    if len(s) <= max_chars:
+        return s
+    return s[:max_chars].rstrip() + "…"

python/server/src/smooth_operator_server/session_store.py

@@ -15,6 +15,7 @@
 import uuid
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
+from datetime import datetime, timezone
 from enum import Enum
 from threading import Lock
 
@@ -53,6 +54,24 @@ class StoredMessage:
     text: str
 
 
+@dataclass(frozen=True)
+class ConversationSummary:
+    """One conversation's roll-up for the ``list_conversations`` action — enough to
+    render a resumable-thread list without pulling every message. The dispatcher turns
+    ``first_inbound_text`` / ``updated_at`` / ``message_count`` into the wire
+    ``{conversationId, title, updatedAt, messageCount}``. The Python analog of the Go
+    ``ConversationSummary`` and the TS ``ConversationSummary``. th-d5b446."""
+
+    conversation_id: str
+    #: Last-activity timestamp (create, then every append) — the sort key + ``updatedAt``.
+    updated_at: datetime
+    #: Total messages in the conversation. The dispatcher drops empties (``0``).
+    message_count: int
+    #: Text of the FIRST inbound (user) message — the dispatcher's title source.
+    #: ``None`` when the conversation has no inbound message (title falls back to a name).
+    first_inbound_text: str | None = None
+
+
 #: The reference agent's display name (mirrors the Rust ``AGENT_NAME`` and the C#
 #: ``InMemorySessionStore`` default).
 AGENT_NAME = "smooth-agent"
@@ -63,11 +82,26 @@ class SessionStore(ABC):
     adapter and the C# ``ISessionStore``)."""
 
     @abstractmethod
-    async def create_session(self, agent_id: str, user_name: str | None, user_email: str | None) -> StoredSession: ...
+    async def create_session(
+        self, agent_id: str, user_name: str | None, user_email: str | None, conversation_id: str | None = None
+    ) -> StoredSession:
+        """Mint a session. When ``conversation_id`` names an EXISTING conversation, the
+        new session binds to it (resume: reuses the id + its persisted message log, so
+        subsequent turns append and history replays). An absent or unknown id mints a
+        fresh conversation (unchanged behavior). th-d5b446."""
+        ...
 
     @abstractmethod
     async def get_session(self, session_id: str) -> StoredSession | None: ...
 
+    @abstractmethod
+    async def list_conversations(self) -> list[ConversationSummary]:
+        """A summary per conversation that has at least one message (empty conversations —
+        every page-load currently mints one — are dropped), in no particular order; the
+        dispatcher sorts most-recent-first and caps. The Python analog of the Rust
+        ``list_conversations_by_org`` + per-conversation peek. th-d5b446."""
+        ...
+
     @abstractmethod
     async def append_message(self, conversation_id: str, direction: MessageDirection, text: str) -> StoredMessage: ...
 
@@ -111,25 +145,38 @@ def __init__(self) -> None:
         self._gate = Lock()
         self._sessions: dict[str, StoredSession] = {}
         self._messages: dict[str, list[StoredMessage]] = {}
+        #: Per-conversation last-activity time (create, then every append) — the sort key
+        #: + updated_at source for list_conversations. th-d5b446.
+        self._updated_at: dict[str, datetime] = {}
         #: Per-conversation workflow-step pointer (absent = fresh start / no workflow).
         self._current_step: dict[str, str] = {}
         #: Per-session OTP-verified bit (absent/False = unverified). Set by a
         #: successful ``verify_otp``; read by the ``end_user`` auth gate.
         self._authenticated: dict[str, bool] = {}
 
-    async def create_session(self, agent_id: str, user_name: str | None, user_email: str | None) -> StoredSession:
-        session = StoredSession(
-            session_id=str(uuid.uuid4()),
-            conversation_id=str(uuid.uuid4()),
-            agent_id=agent_id if agent_id else str(uuid.uuid4()),
-            agent_name=AGENT_NAME,
-            user_participant_id=str(uuid.uuid4()),
-            agent_participant_id=str(uuid.uuid4()),
-            contact_email=(user_email.strip() or None) if isinstance(user_email, str) else None,
-        )
+    async def create_session(
+        self, agent_id: str, user_name: str | None, user_email: str | None, conversation_id: str | None = None
+    ) -> StoredSession:
         with self._gate:
+            # Resume: bind to an existing conversation (reuse its id + persisted log) when
+            # the caller passes a known conversationId. Unknown/absent → mint a fresh one.
+            resume = bool(conversation_id) and conversation_id in self._messages
+            conv_id = conversation_id if resume else str(uuid.uuid4())
+            session = StoredSession(
+                session_id=str(uuid.uuid4()),
+                conversation_id=conv_id,
+                agent_id=agent_id if agent_id else str(uuid.uuid4()),
+                agent_name=AGENT_NAME,
+                user_participant_id=str(uuid.uuid4()),
+                agent_participant_id=str(uuid.uuid4()),
+                contact_email=(user_email.strip() or None) if isinstance(user_email, str) else None,
+            )
             self._sessions[session.session_id] = session
-            self._messages[session.conversation_id] = []
+            # Only seed an empty log + timestamp on a fresh conversation — a resume keeps
+            # its history (and prior last-activity time).
+            if not resume:
+                self._messages[conv_id] = []
+                self._updated_at[conv_id] = datetime.now(timezone.utc)
         return session
 
     async def get_session(self, session_id: str) -> StoredSession | None:
@@ -140,8 +187,27 @@ async def append_message(self, conversation_id: str, direction: MessageDirection
         message = StoredMessage(str(uuid.uuid4()), conversation_id, direction, text)
         with self._gate:
             self._messages.setdefault(conversation_id, []).append(message)
+            self._updated_at[conversation_id] = datetime.now(timezone.utc)
         return message
 
+    async def list_conversations(self) -> list[ConversationSummary]:
+        with self._gate:
+            out: list[ConversationSummary] = []
+            for conv_id, log in self._messages.items():
+                if not log:  # drop empties — every page-load mints one
+                    continue
+                # Messages are stored oldest-first, so the first inbound is the title source.
+                first_inbound = next((m.text for m in log if m.direction is MessageDirection.INBOUND), None)
+                out.append(
+                    ConversationSummary(
+                        conversation_id=conv_id,
+                        updated_at=self._updated_at.get(conv_id, datetime.now(timezone.utc)),
+                        message_count=len(log),
+                        first_inbound_text=first_inbound,
+                    )
+                )
+            return out
+
     async def list_messages(self, conversation_id: str, limit: int) -> list[StoredMessage]:
         with self._gate:
             log = self._messages.get(conversation_id, [])