Fix CLI sync, Telegram error handling, and memU max_tokens

nerve/agent/engine.py

@@ -997,19 +997,36 @@ async def run(
         channel: str | None = None,
         model: str | None = None,
         internal: bool = False,
+        images: list[dict[str, Any]] | None = None,
     ) -> str:
         """Run the agent for a user message and return the final text response.
 
         Args:
             internal: If True, the user_message is a system-generated trigger
                       (e.g., background task completion) and won't be stored in
                       DB or shown in the UI.
+            images: Optional list of image dicts with keys ``type``,
+                    ``media_type``, and ``data`` (base64-encoded).
         """
+        # If the session is still running (e.g. /stop cleanup in progress),
+        # wait briefly instead of failing immediately.
         if self.sessions.is_running(session_id):
-            raise RuntimeError(f"Session {session_id} is already running")
+            for _ in range(10):
+                await asyncio.sleep(0.3)
+                if not self.sessions.is_running(session_id):
+                    break
+            else:
+                raise RuntimeError(f"Session {session_id} is already running")
 
         broadcaster.start_buffering(session_id)
         async with self._semaphore:
+            # Clear any stale deferred-stop flag left over from a *previous*
+            # turn.  If /stop arrived while the old turn was still cleaning up
+            # (mark_not_running hadn't run yet), the flag lingers and would
+            # immediately kill this brand-new turn.  Flags set *during* this
+            # turn's client init are unaffected — they're created after
+            # mark_running below.
+            self.sessions.pop_stop_request(session_id)
             self.sessions.mark_running(session_id)
             # Notify all connected clients that this session started running
             await broadcaster.broadcast("__global__", {
@@ -1020,7 +1037,7 @@ async def run(
             try:
                 return await self._run_inner(
                     session_id, user_message, source, channel, model,
-                    internal=internal,
+                    internal=internal, images=images,
                 )
             finally:
                 self.sessions.mark_not_running(session_id)
@@ -1040,6 +1057,7 @@ async def _run_inner(
         channel: str | None,
         model: str | None,
         internal: bool = False,
+        images: list[dict[str, Any]] | None = None,
     ) -> str:
         # Ensure session exists in DB
         await self.sessions.get_or_create(session_id, source=source)
@@ -1066,9 +1084,13 @@ async def _run_inner(
                     self._generate_session_title(session_id, user_message),
                 )
 
-            # Store user message in DB
+            # Store user message in DB (note attached images for display)
+            db_text = user_message
+            if images:
+                suffix = f"\n[{len(images)} image(s) attached]"
+                db_text = (user_message + suffix) if user_message else suffix.strip()
             await self.sessions.add_message(
-                session_id, "user", user_message, channel=channel,
+                session_id, "user", db_text, channel=channel,
             )
 
         full_response_text = ""
@@ -1096,10 +1118,48 @@ async def _run_inner(
                 session_id, source, model, fork_from=fork_from,
             )
 
+            # Check for deferred /stop that arrived while we were setting up
+            if self.sessions.pop_stop_request(session_id):
+                logger.info("Stop requested before agent turn — aborting session %s", session_id)
+                return ""
+
             # Send message — the client preserves conversation history internally
-            await client.query(user_message)
+            if images:
+                # Build multi-modal content blocks (text + images)
+                content_blocks: list[dict[str, Any]] = []
+                if user_message:
+                    content_blocks.append({"type": "text", "text": user_message})
+                for img in images:
+                    content_blocks.append({
+                        "type": "image",
+                        "source": {
+                            "type": img["type"],
+                            "media_type": img["media_type"],
+                            "data": img["data"],
+                        },
+                    })
+
+                async def _image_prompt():
+                    yield {
+                        "type": "user",
+                        "message": {"role": "user", "content": content_blocks},
+                        "parent_tool_use_id": None,
+                    }
+
+                await client.query(_image_prompt())
+            else:
+                await client.query(user_message)
 
             async for message in client.receive_response():
+                # Early-capture sdk_session_id from first message that
+                # carries it so it survives /stop cancellation (ResultMessage
+                # — the normal source — never arrives when the turn is
+                # interrupted).
+                if not sdk_session_id:
+                    msg_sid = getattr(message, "session_id", None)
+                    if msg_sid:
+                        sdk_session_id = msg_sid
+
                 if isinstance(message, AssistantMessage):
                     # Extract parent_tool_use_id — set when this message
                     # comes from a sub-agent (Task) rather than the main agent
@@ -1195,26 +1255,41 @@ async def _run_inner(
                 if full_response_text
                 else "[Stopped by user]"
             )
-            # Merge available tool results
-            self._merge_tool_results(tool_calls_log, tool_results_map)
-            await self.sessions.add_message(
-                session_id, "assistant", partial,
-                channel=channel,
-                thinking=thinking_text if thinking_text else None,
-                tool_calls=tool_calls_log if tool_calls_log else None,
-                blocks=ordered_blocks if ordered_blocks else None,
-            )
-            await broadcaster.broadcast(session_id, {
-                "type": "stopped", "session_id": session_id,
-            })
-            # Memorize before discarding client
-            await self._memorize_session(session_id)
-            # Keep sdk_session_id for resume — stop is user-initiated
+
+            # --- Critical cleanup first (must succeed for resume) ----------
+            # Persist sdk_session_id so the session can be resumed later.
+            # For new sessions the DB still has NULL because mark_active()
+            # was called before the SDK emitted any messages.
+            if sdk_session_id:
+                await self.db.update_session_fields(
+                    session_id, {"sdk_session_id": sdk_session_id},
+                )
             await self.sessions.mark_stopped(session_id)
             unregister_handler(session_id)
             client = self.sessions.remove_client(session_id)
             if client:
                 await self._safe_disconnect(client)
+
+            # --- Non-critical: save message, broadcast, memorize -----------
+            try:
+                self._merge_tool_results(tool_calls_log, tool_results_map)
+                await self.sessions.add_message(
+                    session_id, "assistant", partial,
+                    channel=channel,
+                    thinking=thinking_text if thinking_text else None,
+                    tool_calls=tool_calls_log if tool_calls_log else None,
+                    blocks=ordered_blocks if ordered_blocks else None,
+                )
+                await broadcaster.broadcast(session_id, {
+                    "type": "stopped", "session_id": session_id,
+                })
+            except Exception as cleanup_err:
+                logger.warning(
+                    "Non-critical stop cleanup failed for %s: %s",
+                    session_id, cleanup_err,
+                )
+            # Memorize in background — don't block the stop path
+            asyncio.create_task(self._memorize_session(session_id))
             return partial
 
         except Exception as e:

nerve/agent/sessions.py

@@ -60,6 +60,9 @@ def __init__(self, db: Database, sticky_period_minutes: int = 120):
         # Running task tracking
         self._running_tasks: dict[str, asyncio.Task] = {}
         self._running_sessions: set[str] = set()
+        # Stop-requested flag: set when /stop arrives before the SDK client
+        # is registered.  _run_inner() checks this after creating the client.
+        self._stop_requested: set[str] = set()
         # Serialize status transitions
         self._transition_lock = asyncio.Lock()
         # Callback for memorizing session before close/archive/delete.
@@ -323,33 +326,69 @@ def get_running_ids(self) -> set[str]:
         """Return the set of currently running session IDs."""
         return set(self._running_sessions)
 
+    def request_stop(self, session_id: str) -> None:
+        """Set a deferred stop flag (checked by _run_inner after client init)."""
+        self._stop_requested.add(session_id)
+
+    def pop_stop_request(self, session_id: str) -> bool:
+        """Return True and clear if a stop was requested for *session_id*."""
+        try:
+            self._stop_requested.remove(session_id)
+            return True
+        except KeyError:
+            return False
+
     async def stop_session(self, session_id: str) -> bool:
-        """Stop a running session.
+        """Stop the current turn of a running session.
 
-        Uses SDK client.interrupt() first for clean stop, falls back to
-        asyncio task cancellation.  Returns True if something was stopped.
+        Sends SDK interrupt first — this gracefully ends the current turn
+        while keeping the client alive for the next message.  Falls back
+        to asyncio task cancellation (which disconnects the client) only
+        if the interrupt doesn't complete within a timeout.
         """
-        # Try SDK interrupt first (cleanly stops the current turn)
         client = self._clients.get(session_id)
+        interrupted = False
         if client:
             try:
                 await client.interrupt()
+                interrupted = True
                 logger.info("Interrupted SDK client for session %s", session_id)
-                return True
             except Exception as e:
                 logger.warning(
-                    "SDK interrupt failed for %s: %s, falling back to cancel",
+                    "SDK interrupt failed for %s: %s",
                     session_id, e,
                 )
 
-        # Fallback: cancel the asyncio task
         task = self._running_tasks.get(session_id)
         if task and not task.done():
+            if interrupted:
+                # Give interrupt time to complete the turn gracefully.
+                # When it works, receive_response() yields a ResultMessage,
+                # _run_inner exits via the normal path, and the client
+                # stays alive for the next message.
+                done, _ = await asyncio.wait({task}, timeout=5.0)
+                if done:
+                    logger.info(
+                        "Session %s stopped gracefully via interrupt",
+                        session_id,
+                    )
+                    return True
+
+            # Interrupt failed or timed out — force cancel.
+            # The CancelledError handler in _run_inner disconnects the
+            # client since its state is inconsistent.
             task.cancel()
-            logger.info("Cancelled task for session %s", session_id)
+            logger.info("Cancelled task for session %s (interrupt timed out)", session_id)
+            return True
+
+        # Session is running but client/task not registered yet — set a
+        # deferred stop flag that _run_inner() will check after client init.
+        if self.is_running(session_id):
+            self.request_stop(session_id)
+            logger.info("Deferred stop requested for session %s", session_id)
             return True
 
-        return False
+        return client is not None  # interrupt was sent even if no task
 
     # ------------------------------------------------------------------ #
     #  Fork & resume                                                       #

nerve/agent/tools.py

@@ -1623,6 +1623,24 @@ async def _ask_user_impl(args: dict, session_id: str) -> dict:
         return {"content": [{"type": "text", "text": f"Failed to ask question: {e}"}]}
 
 
+async def _react_impl(args: dict, session_id: str) -> dict:
+    """Core implementation for the react tool."""
+    if not _engine:
+        return {"content": [{"type": "text", "text": "Engine not available."}]}
+
+    emoji = args["emoji"]
+
+    try:
+        success = await _engine.router.set_reaction(session_id, emoji)
+        if success:
+            return {"content": [{"type": "text", "text": f"Reaction set: {emoji}"}]}
+        else:
+            return {"content": [{"type": "text", "text": "Cannot set reaction: no message context or channel does not support reactions."}]}
+    except Exception as e:
+        logger.error("react tool failed: %s", e)
+        return {"content": [{"type": "text", "text": f"Failed to set reaction: {e}"}]}
+
+
 _nerve_asgi_app = None  # Cached mini FastAPI app for in-process API calls
 
 
@@ -1739,6 +1757,20 @@ async def notify(args: dict) -> dict:
     return await _notify_impl(args, _current_session_id)
 
 
+@tool(
+    "react",
+    "Set an emoji reaction on the user's last message. "
+    "Use to acknowledge messages, express emotions, or respond non-verbally. "
+    "Works on channels that support reactions (e.g., Telegram).",
+    {
+        "emoji": {"type": "string", "description": "Emoji to react with (e.g., '👍', '❤', '🔥', '😂')"},
+    },
+)
+async def react_tool(args: dict) -> dict:
+    """Set a reaction (fallback — uses deprecated global)."""
+    return await _react_impl(args, _current_session_id)
+
+
 @tool(
     "ask_user",
     "Ask the user a question via async notification. "
@@ -1795,11 +1827,15 @@ def create_nerve_mcp_server():
     "priority": {"type": "string", "description": "Priority: 'low', 'normal', 'high', 'urgent'. Default: 'normal'", "default": "normal"},
 }
 
+_REACT_SCHEMA = {
+    "emoji": {"type": "string", "description": "Emoji to react with (e.g., '👍', '❤', '🔥', '😂')"},
+}
+
 
 def create_session_mcp_server(session_id: str):
-    """Create an MCP server with session_id bound for notification tools.
+    """Create an MCP server with session_id bound for session-scoped tools.
 
-    Each session gets its own MCP server instance so notify/ask_user
+    Each session gets its own MCP server instance so notify/ask_user/react
     tools always reference the correct session — no shared global needed.
     """
 
@@ -1825,8 +1861,19 @@ async def session_ask_user(args: dict) -> dict:
         # session_id captured from enclosing scope — race-free
         return await _ask_user_impl(args, session_id)
 
+    @tool(
+        "react",
+        "Set an emoji reaction on the user's last message. "
+        "Use to acknowledge messages, express emotions, or respond non-verbally. "
+        "Works on channels that support reactions (e.g., Telegram).",
+        _REACT_SCHEMA,
+    )
+    async def session_react(args: dict) -> dict:
+        # session_id captured from enclosing scope — race-free
+        return await _react_impl(args, session_id)
+
     # Shared tools (don't need session context) + session-scoped tools
-    shared_tools = [t for t in ALL_TOOLS if t.name not in ("notify", "ask_user")]
-    all_tools = shared_tools + [session_notify, session_ask_user]
+    shared_tools = [t for t in ALL_TOOLS if t.name not in ("notify", "ask_user", "react")]
+    all_tools = shared_tools + [session_notify, session_ask_user, session_react]
 
     return create_sdk_mcp_server(name="nerve", version="1.0.0", tools=all_tools)

nerve/channels/base.py

@@ -27,6 +27,7 @@ class ChannelCapability(Flag):
     MARKDOWN = auto()           # Renders markdown natively
     INTERACTIVE = auto()        # Can route interactive tool responses back (AskUserQuestion, etc.)
     TYPING_INDICATOR = auto()   # Can show "typing…" status
+    REACTIONS = auto()          # Can set emoji reactions on messages
 
 
 @dataclass(frozen=True)
@@ -141,6 +142,17 @@ async def send_typing(self, target: str) -> None:
         Only called if channel declares TYPING_INDICATOR capability.
         """
 
+    # ------------------------------------------------------------------ #
+    #  Optional: reactions support                                          #
+    #  Only called if channel declares ChannelCapability.REACTIONS.         #
+    # ------------------------------------------------------------------ #
+
+    async def set_reaction(self, target: str, message_id: int, emoji: str) -> None:
+        """Set an emoji reaction on a message.
+
+        Only called if channel declares REACTIONS capability.
+        """
+
     # ------------------------------------------------------------------ #
     #  Optional: interactive tool support                                   #
     #  Only called if channel declares ChannelCapability.INTERACTIVE.       #