From 008b532306ffd7bf260fe960c57fc433f9c38ddc Mon Sep 17 00:00:00 2001 From: Brent Rager Date: Thu, 9 Jul 2026 13:33:10 -0400 Subject: [PATCH] th-d5b446: Python server list_conversations + resume-by-conversationId MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Mirror the merged Rust/Go/TS reference (pearl th-d5b446) on the Python smooth-operator-server — the conversation-sidebar / resume substrate every client builds against. - New WS action `list_conversations`: most-recent-first, only conversations with messageCount > 0, each with a first-inbound title preview (~60 chars, leading markdown/control stripped, name fallback), ISO-8601 updatedAt, and messageCount. Optional limit (default 50). - `create_conversation_session` gains optional `conversationId`: a known id resumes (reuse id + persisted log so send_message appends and history replays); absent/unknown mints fresh (unchanged). Additive + back-compat. SessionStore gains list_conversations() + optional conversation_id on create_session; in-memory store tracks per-conversation last-activity for the sort key. 16 new tests; full suite green (154). Co-Authored-By: Claude Opus 4.8 (1M context) --- .changeset/python-server-list-resume.md | 9 + .../src/smooth_operator_server/dispatcher.py | 83 +++++++++ .../smooth_operator_server/session_store.py | 90 +++++++-- .../tests/test_list_conversations_resume.py | 172 ++++++++++++++++++ 4 files changed, 342 insertions(+), 12 deletions(-) create mode 100644 .changeset/python-server-list-resume.md create mode 100644 python/server/tests/test_list_conversations_resume.py diff --git a/.changeset/python-server-list-resume.md b/.changeset/python-server-list-resume.md new file mode 100644 index 0000000..fbe8660 --- /dev/null +++ b/.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. diff --git a/python/server/src/smooth_operator_server/dispatcher.py b/python/server/src/smooth_operator_server/dispatcher.py index ca1c34b..de6f9ee 100644 --- a/python/server/src/smooth_operator_server/dispatcher.py +++ b/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() + "…" diff --git a/python/server/src/smooth_operator_server/session_store.py b/python/server/src/smooth_operator_server/session_store.py index 5b410a5..2b09c4a 100644 --- a/python/server/src/smooth_operator_server/session_store.py +++ b/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, []) diff --git a/python/server/tests/test_list_conversations_resume.py b/python/server/tests/test_list_conversations_resume.py new file mode 100644 index 0000000..dd05c26 --- /dev/null +++ b/python/server/tests/test_list_conversations_resume.py @@ -0,0 +1,172 @@ +"""``list_conversations`` + resume-by-``conversationId`` (pearl th-d5b446). + +Mirrors the merged Rust/Go/TS reference: ``list_conversations`` rolls up +conversations most-recent-first, drops empties, and derives a clean title from the +first inbound (user) message; a ``create_conversation_session`` carrying a known +``conversationId`` binds to (resumes) that conversation, while an unknown/absent id +mints a fresh one. + +Driven through the real :class:`FrameDispatcher` (like the OTP flow tests) plus fast +unit cases against the store + the title helper. +""" + +from __future__ import annotations + +import json + +import pytest + +from smooth_operator_server.dispatcher import FrameDispatcher, _conversation_title +from smooth_operator_server.session_store import InMemorySessionStore, MessageDirection + + +async def _dispatch(dispatcher: FrameDispatcher, frame: dict) -> list[dict]: + """Dispatch one frame, collecting every event emitted to the sink.""" + events: list[dict] = [] + await dispatcher.dispatch(json.dumps(frame), events.append) + return events + + +# --------------------------------------------------------------------------- # +# _conversation_title helper +# --------------------------------------------------------------------------- # + + +@pytest.mark.parametrize( + ("first", "fallback", "want"), + [ + ("Hello there", "fb", "Hello there"), + (" spaced ", "fb", "spaced"), + ("### Big title", "fb", "Big title"), + ("- do the thing", "fb", "do the thing"), + ("> _quoted_ line", "fb", "quoted_ line"), + ("", "My Conversation", "My Conversation"), + (None, "My Conversation", "My Conversation"), + ("### ", "Fallback", "Fallback"), + ( + "012345678901234567890123456789012345678901234567890123456789EXTRA", + "fb", + "012345678901234567890123456789012345678901234567890123456789…", + ), + ], +) +def test_conversation_title(first: str | None, fallback: str, want: str) -> None: + assert _conversation_title(first, fallback) == want + + +def test_conversation_title_truncates_to_60_plus_ellipsis() -> None: + got = _conversation_title("x" * 100, "fb") + assert len(got) == 61 # 60 chars + the ellipsis rune + assert got.endswith("…") + + +# --------------------------------------------------------------------------- # +# store: resume + list_conversations +# --------------------------------------------------------------------------- # + + +@pytest.mark.asyncio +async def test_list_conversations_filters_empties_and_previews_title() -> None: + store = InMemorySessionStore() + + # A: empty conversation (created, never messaged) → excluded. + await store.create_session("agent", "U", "u@example.com") + # B: has messages → included, title from first inbound. + b = await store.create_session("agent", "U", "u@example.com") + await store.append_message(b.conversation_id, MessageDirection.INBOUND, "## First user line") + await store.append_message(b.conversation_id, MessageDirection.OUTBOUND, "agent reply") + + summaries = await store.list_conversations() + assert len(summaries) == 1 + got = summaries[0] + assert got.conversation_id == b.conversation_id + assert got.message_count == 2 + assert got.first_inbound_text == "## First user line" + assert got.updated_at is not None + + +@pytest.mark.asyncio +async def test_resume_binds_to_existing_conversation() -> None: + store = InMemorySessionStore() + first = await store.create_session("agent", None, None) + await store.append_message(first.conversation_id, MessageDirection.INBOUND, "original") + + # Resume: a new session bound to the same conversation keeps the log. + resumed = await store.create_session("agent", None, None, first.conversation_id) + assert resumed.conversation_id == first.conversation_id + assert resumed.session_id != first.session_id + log = await store.list_messages(first.conversation_id, 100) + assert [m.text for m in log] == ["original"] # history preserved, not reset + + +@pytest.mark.asyncio +async def test_unknown_conversation_id_mints_fresh() -> None: + store = InMemorySessionStore() + session = await store.create_session("agent", None, None, "does-not-exist") + assert session.conversation_id != "does-not-exist" + # A brand-new empty conversation → nothing to list. + assert await store.list_conversations() == [] + + +# --------------------------------------------------------------------------- # +# dispatcher: list_conversations action + resume via create_conversation_session +# --------------------------------------------------------------------------- # + + +@pytest.mark.asyncio +async def test_dispatch_list_conversations_sorted_and_capped() -> None: + store = InMemorySessionStore() + dispatcher = FrameDispatcher(store, None) + + older = await store.create_session("agent", None, None) + await store.append_message(older.conversation_id, MessageDirection.INBOUND, "older") + newer = await store.create_session("agent", None, None) + await store.append_message(newer.conversation_id, MessageDirection.INBOUND, "newer") + # Touch `older` again so it becomes the most recently active. + await store.append_message(older.conversation_id, MessageDirection.OUTBOUND, "older reply") + + events = await _dispatch(dispatcher, {"action": "list_conversations", "requestId": "r1"}) + assert len(events) == 1 + ev = events[0] + assert ev["type"] == "immediate_response" + assert ev["status"] == 200 + convos = ev["data"]["conversations"] + assert [c["conversationId"] for c in convos] == [older.conversation_id, newer.conversation_id] + assert convos[0]["title"] == "older" + assert convos[0]["messageCount"] == 2 + + # limit caps the result. + capped = await _dispatch(dispatcher, {"action": "list_conversations", "requestId": "r2", "limit": 1}) + assert len(capped[0]["data"]["conversations"]) == 1 + + +@pytest.mark.asyncio +async def test_dispatch_resume_echoes_conversation_id_and_keeps_history() -> None: + store = InMemorySessionStore() + dispatcher = FrameDispatcher(store, None) + + first = await store.create_session("agent", None, None) + await store.append_message(first.conversation_id, MessageDirection.INBOUND, "hi") + + events = await _dispatch( + dispatcher, + {"action": "create_conversation_session", "requestId": "r1", "conversationId": first.conversation_id}, + ) + data = events[0]["data"] + assert data["conversationId"] == first.conversation_id # resumed: same id echoed back + assert data["sessionId"] != first.session_id + # History preserved (resume did not reset the log). + log = await store.list_messages(first.conversation_id, 100) + assert [m.text for m in log] == ["hi"] + + +@pytest.mark.asyncio +async def test_dispatch_create_without_conversation_id_mints_fresh() -> None: + store = InMemorySessionStore() + dispatcher = FrameDispatcher(store, None) + + events = await _dispatch(dispatcher, {"action": "create_conversation_session", "requestId": "r1"}) + convo_id = events[0]["data"]["conversationId"] + assert isinstance(convo_id, str) and convo_id + # Fresh empty conversation → not listed. + assert await store.list_conversations() == []