diff --git a/nerve/agent/engine.py b/nerve/agent/engine.py
index 80ae1133..db1be789 100644
--- a/nerve/agent/engine.py
+++ b/nerve/agent/engine.py
@@ -997,6 +997,7 @@ 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.
@@ -1004,12 +1005,28 @@ async def run(
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:
diff --git a/nerve/agent/sessions.py b/nerve/agent/sessions.py
index 72b694de..85b7a02b 100644
--- a/nerve/agent/sessions.py
+++ b/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 #
diff --git a/nerve/agent/tools.py b/nerve/agent/tools.py
index fb088035..222e19b5 100644
--- a/nerve/agent/tools.py
+++ b/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)
diff --git a/nerve/channels/base.py b/nerve/channels/base.py
index 29fef458..3960e851 100644
--- a/nerve/channels/base.py
+++ b/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. #
diff --git a/nerve/channels/router.py b/nerve/channels/router.py
index e66c5fc4..0487ac20 100644
--- a/nerve/channels/router.py
+++ b/nerve/channels/router.py
@@ -48,6 +48,9 @@ def __init__(self, engine: AgentEngine):
self._channels: dict[str, BaseChannel] = {}
# Active stream adapters: (channel_name, target) -> StreamAdapter
self._adapters: dict[tuple[str, str], StreamAdapter] = {}
+ # Per-session inbound message context (for reaction support)
+ # Maps session_id -> {channel_name, target, message_id}
+ self._message_context: dict[str, dict[str, Any]] = {}
# ------------------------------------------------------------------ #
# Channel registry #
@@ -102,6 +105,15 @@ async def handle_message(self, msg: InboundMessage) -> str:
msg.channel_key, source=msg.channel_name,
)
+ # Store message context for reaction support
+ msg_id = msg.metadata.get("message_id") if msg.metadata else None
+ if msg_id is not None:
+ self._message_context[session_id] = {
+ "channel_name": msg.channel_name,
+ "target": msg.sender_id,
+ "message_id": msg_id,
+ }
+
# Show typing indicator if supported
if ChannelCapability.TYPING_INDICATOR in channel.capabilities:
try:
@@ -114,19 +126,59 @@ async def handle_message(self, msg: InboundMessage) -> str:
channel, msg.sender_id, session_id,
)
- try:
- response = await self.engine.run(
+ # Extract images from metadata (e.g. Telegram photos)
+ images = msg.metadata.get("images") if msg.metadata else None
+
+ # Wrap in a Task so stop_session() can cancel it (otherwise
+ # channels that ``await engine.run()`` directly — like Telegram —
+ # have no cancellable task and /stop only sends an SDK interrupt
+ # which may hang indefinitely).
+ task = asyncio.create_task(
+ self.engine.run(
session_id=session_id,
user_message=msg.text,
source=msg.channel_name,
channel=msg.channel_name,
+ images=images,
)
+ )
+ self.engine.register_task(session_id, task)
+ try:
+ response = await task
return response
+ except asyncio.CancelledError:
+ # /stop cancelled the task — _run_inner already handled
+ # cleanup (persisted sdk_session_id, marked stopped, etc.).
+ # Return whatever partial response was captured.
+ if task.done() and not task.cancelled():
+ return task.result()
+ return ""
finally:
await self._teardown_streaming(
channel.name, msg.sender_id, session_id,
)
+ # ------------------------------------------------------------------ #
+ # Reactions #
+ # ------------------------------------------------------------------ #
+
+ async def set_reaction(self, session_id: str, emoji: str) -> bool:
+ """Set a reaction on the last inbound message for a session.
+
+ Returns True if the reaction was set, False if no context or
+ the channel does not support reactions.
+ """
+ ctx = self._message_context.get(session_id)
+ if not ctx:
+ return False
+
+ channel = self._channels.get(ctx["channel_name"])
+ if not channel or ChannelCapability.REACTIONS not in channel.capabilities:
+ return False
+
+ await channel.set_reaction(ctx["target"], ctx["message_id"], emoji)
+ return True
+
# ------------------------------------------------------------------ #
# Interactive tool response routing #
# ------------------------------------------------------------------ #
diff --git a/nerve/channels/stream_adapter.py b/nerve/channels/stream_adapter.py
index 311723f8..cc8b2b77 100644
--- a/nerve/channels/stream_adapter.py
+++ b/nerve/channels/stream_adapter.py
@@ -44,6 +44,12 @@ def __init__(
self._last_edit: float = 0.0
self._edit_lock = asyncio.Lock()
+ # Tool label grouping (collapse consecutive same-tool calls)
+ self._last_tool_name: str | None = None
+ self._tool_run_count: int = 0
+ self._tool_label_start: int = 0
+ self._tool_label_prefix: str = "\n\n"
+
# Precompute capability checks
self._supports_streaming = (
ChannelCapability.STREAMING in channel.capabilities
@@ -72,7 +78,17 @@ async def on_event(self, session_id: str, message: dict[str, Any]) -> None:
elif msg_type == "tool_use":
tool_name = message.get("tool", "unknown")
if self._supports_streaming and self._supports_edit:
- self._buffer += f"\n`[tool: {tool_name}]`\n"
+ if tool_name == self._last_tool_name:
+ self._tool_run_count += 1
+ self._buffer = self._buffer[:self._tool_label_start]
+ self._buffer += f"{self._tool_label_prefix}`[{tool_name}] x{self._tool_run_count}`\n"
+ else:
+ after_tool = self._last_tool_name is not None
+ self._last_tool_name = tool_name
+ self._tool_run_count = 1
+ self._tool_label_start = len(self._buffer)
+ self._tool_label_prefix = "" if after_tool else "\n\n"
+ self._buffer += f"{self._tool_label_prefix}`[{tool_name}]`\n"
elif msg_type == "done":
await self._handle_done()
@@ -92,6 +108,10 @@ async def on_event(self, session_id: str, message: dict[str, Any]) -> None:
# ------------------------------------------------------------------ #
async def _handle_token(self, content: str) -> None:
+ if self._last_tool_name is not None:
+ # Transitioning from tool block to text — add blank line separator
+ self._buffer += "\n"
+ self._last_tool_name = None # Reset tool grouping
self._buffer += content
if not self._supports_streaming or not self._supports_edit:
diff --git a/nerve/channels/telegram.py b/nerve/channels/telegram.py
index ca327ab3..3a0aa5b5 100644
--- a/nerve/channels/telegram.py
+++ b/nerve/channels/telegram.py
@@ -7,12 +7,19 @@
from __future__ import annotations
+import asyncio
+import base64
+import collections
+import html as _html
import logging
+import re
+import socket
+import time
from typing import Any, TYPE_CHECKING
from telegram import Update
from telegram.constants import ChatAction, ParseMode
-from telegram.ext import Application, CallbackQueryHandler, CommandHandler, MessageHandler, filters
+from telegram.ext import Application, CallbackQueryHandler, CommandHandler, MessageHandler, MessageReactionHandler, filters
from nerve.channels.base import (
BaseChannel,
@@ -32,6 +39,112 @@
MAX_MSG_LEN = 4096
# Minimum interval between message edits (seconds) to avoid rate limits
EDIT_INTERVAL = 1.5
+# Watchdog: check every 30s, log heartbeat every ~5 min
+WATCHDOG_INTERVAL = 30
+WATCHDOG_HEARTBEAT_EVERY = 10
+
+# TCP keepalive: prevent NAT/firewall from silently dropping the
+# long-poll connection. These values tell the OS to send a keepalive
+# probe after 60s idle, retry every 10s, give up after 3 failures.
+_TCP_KEEPALIVE_OPTS = (
+ (socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1),
+ (socket.SOL_TCP, socket.TCP_KEEPIDLE, 60),
+ (socket.SOL_TCP, socket.TCP_KEEPINTVL, 10),
+ (socket.SOL_TCP, socket.TCP_KEEPCNT, 3),
+)
+
+
+def _md_to_tg_html(text: str) -> str:
+ """Convert standard Markdown to Telegram-compatible HTML.
+
+ Telegram's legacy ``ParseMode.MARKDOWN`` only supports ``*bold*``,
+ but LLMs emit ``**bold**`` (standard Markdown). This converts the
+ common constructs to HTML so we can use ``ParseMode.HTML`` instead,
+ which is more predictable and doesn't choke on special characters.
+
+ Handles: ``**bold**``, ``*italic*``, `` `code` ``, code fences,
+ and ``[text](url)``. Unmatched markers pass through as-is.
+ """
+ protected: list[str] = []
+
+ def _protect(replacement: str) -> str:
+ idx = len(protected)
+ protected.append(replacement)
+ return f"\x00{idx}\x00"
+
+ # -- protect constructs that contain chars we'd otherwise escape --
+
+ # Code fences: ```lang\n...\n```
+ def _fence(m: re.Match) -> str:
+ return _protect(f"{_html.escape(m.group(2))}")
+ text = re.sub(r"```(\w*)\n?(.*?)```", _fence, text, flags=re.DOTALL)
+
+ # Inline code: `...`
+ def _code(m: re.Match) -> str:
+ return _protect(f"{_html.escape(m.group(1))}")
+ text = re.sub(r"`([^`]+)`", _code, text)
+
+ # Markdown links: [text](url)
+ def _link(m: re.Match) -> str:
+ label = _html.escape(m.group(1))
+ url = m.group(2)
+ return _protect(f'{label}')
+ text = re.sub(r"\[([^\]]+)\]\(([^)]+)\)", _link, text)
+
+ # -- escape remaining HTML entities --
+ text = _html.escape(text, quote=False)
+
+ # -- inline formatting --
+ text = re.sub(r"\*\*(.+?)\*\*", r"\1", text)
+ text = re.sub(r"(?\1", text)
+
+ # -- restore protected spans --
+ for i, repl in enumerate(protected):
+ text = text.replace(f"\x00{i}\x00", repl)
+
+ return text
+
+
+def _format_reply_context(message: Any) -> str:
+ """Extract reply-to context and quote from a Telegram message.
+
+ Returns a prefix string like:
+ [Reply to assistant: "original text here"]
+ [Quoted: "selected portion"]
+
+ Returns empty string if the message is not a reply.
+ """
+ reply = getattr(message, "reply_to_message", None)
+ if not reply:
+ return ""
+
+ parts: list[str] = []
+
+ # Determine sender label
+ from_user = getattr(reply, "from_user", None)
+ if from_user and getattr(from_user, "is_bot", False):
+ sender = "assistant"
+ elif from_user:
+ sender = getattr(from_user, "first_name", None) or "user"
+ else:
+ sender = "user"
+
+ # Original message text
+ original = getattr(reply, "text", None) or getattr(reply, "caption", None) or ""
+ if original:
+ display = original if len(original) <= 500 else original[:500] + "…"
+ parts.append(f'[Reply to {sender}: "{display}"]')
+ else:
+ parts.append(f"[Reply to {sender}'s message]")
+
+ # Quote (manually selected text)
+ quote = getattr(message, "quote", None)
+ if quote:
+ quote_text = getattr(quote, "text", None)
+ if quote_text:
+ parts.append(f'[Quoted: "{quote_text}"]')
+
+ return "\n".join(parts)
class TelegramChannel(BaseChannel):
@@ -47,6 +160,17 @@ def __init__(self, config: NerveConfig, router: ChannelRouter):
self._app: Application | None = None
self._allowed_users: set[int] = set(config.telegram.allowed_users)
self._notification_service = None # Set after service is created
+ self._watchdog_task: asyncio.Task | None = None
+ self._stopping = False
+ self._last_update_time: float = 0.0 # monotonic, set on any incoming update
+ # Media group (album) collection: group_id -> list of updates
+ self._media_groups: dict[str, list[Update]] = {}
+ self._media_group_tasks: dict[str, asyncio.Task] = {}
+ # Message cache for reaction context: message_id -> (chat_id, text_snippet)
+ self._message_cache: collections.OrderedDict[int, tuple[int, str]] = (
+ collections.OrderedDict()
+ )
+ self._message_cache_max = 200
def set_notification_service(self, service) -> None:
"""Wire the notification service for callback query handling."""
@@ -62,6 +186,7 @@ def capabilities(self) -> ChannelCapability:
ChannelCapability.SEND_TEXT
| ChannelCapability.MARKDOWN
| ChannelCapability.TYPING_INDICATOR
+ | ChannelCapability.REACTIONS
)
if self.config.telegram.stream_mode == "partial":
caps |= ChannelCapability.STREAMING
@@ -79,38 +204,230 @@ def constraints(self) -> ChannelConstraints:
# Lifecycle #
# ------------------------------------------------------------------ #
+ def _build_application(self) -> Application:
+ """Build a PTB Application with robust connection settings."""
+ builder = (
+ Application.builder()
+ .token(self.config.telegram.bot_token)
+ # Process updates concurrently so /stop can interrupt a running
+ # message handler instead of being queued behind it.
+ .concurrent_updates(True)
+ # TCP keepalive on the polling connection — prevents NAT/firewall
+ # from silently dropping the long-poll connection
+ .get_updates_socket_options(_TCP_KEEPALIVE_OPTS)
+ # Separate connection pool for polling vs sending, so a stuck
+ # outbound request can't starve the polling connection
+ .get_updates_connection_pool_size(1)
+ .get_updates_read_timeout(10)
+ .get_updates_connect_timeout(10)
+ .get_updates_pool_timeout(1.0)
+ )
+ app = builder.build()
+
+ # Register handlers
+ app.add_handler(CommandHandler("start", self._handle_start))
+ app.add_handler(CommandHandler("session", self._handle_session))
+ app.add_handler(CommandHandler("sessions", self._handle_sessions))
+ app.add_handler(CommandHandler("new", self._handle_new_session))
+ app.add_handler(CommandHandler("stop", self._handle_stop))
+ app.add_handler(CommandHandler("reply", self._handle_reply))
+ app.add_handler(CallbackQueryHandler(self._handle_callback_query))
+ app.add_handler(MessageHandler(
+ (filters.TEXT | filters.PHOTO) & ~filters.COMMAND,
+ self._handle_message,
+ ))
+ app.add_handler(MessageReactionHandler(self._handle_reaction))
+ app.add_error_handler(self._handle_error)
+
+ return app
+
async def start(self) -> None:
"""Start the Telegram bot."""
if not self.config.telegram.bot_token:
logger.warning("Telegram bot token not configured")
return
- self._app = (
- Application.builder()
- .token(self.config.telegram.bot_token)
- .build()
- )
-
- # Register handlers
- self._app.add_handler(CommandHandler("start", self._handle_start))
- self._app.add_handler(CommandHandler("session", self._handle_session))
- self._app.add_handler(CommandHandler("sessions", self._handle_sessions))
- self._app.add_handler(CommandHandler("new", self._handle_new_session))
- self._app.add_handler(CommandHandler("reply", self._handle_reply))
- self._app.add_handler(CallbackQueryHandler(self._handle_callback_query))
- self._app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, self._handle_message))
+ self._stopping = False
+ self._app = self._build_application()
await self._app.initialize()
await self._app.start()
- await self._app.updater.start_polling(drop_pending_updates=True)
- logger.info("Telegram bot started polling")
+ await self._app.updater.start_polling(
+ drop_pending_updates=True,
+ # Retry initial connection indefinitely — don't give up on
+ # transient network errors during startup
+ bootstrap_retries=-1,
+ # Explicitly request all update types — the default set excludes
+ # message_reaction, and auto-detection only works via
+ # Application.run_polling(), not Updater.start_polling().
+ allowed_updates=Update.ALL_TYPES,
+ )
+ self._last_update_time = time.monotonic()
+ logger.info("Telegram bot started polling (with TCP keepalive)")
+
+ # Launch watchdog for monitoring + recovery
+ self._watchdog_task = asyncio.create_task(
+ self._run_watchdog(), name="telegram-polling-watchdog",
+ )
async def stop(self) -> None:
+ self._stopping = True
+ if self._watchdog_task and not self._watchdog_task.done():
+ self._watchdog_task.cancel()
+ try:
+ await self._watchdog_task
+ except asyncio.CancelledError:
+ pass
if self._app:
await self._app.updater.stop()
await self._app.stop()
await self._app.shutdown()
+ # ------------------------------------------------------------------ #
+ # Watchdog — monitor both Updater and Application health #
+ # ------------------------------------------------------------------ #
+
+ async def _run_watchdog(self) -> None:
+ """Monitor Telegram bot health and log diagnostics.
+
+ Checks BOTH the Updater (fetches updates from Telegram) AND the
+ Application (processes updates from queue → handlers). Previous
+ versions only checked the Updater, missing cases where the
+ Application's update-processing task crashed silently.
+ """
+ check_count = 0
+ while not self._stopping:
+ try:
+ await asyncio.sleep(WATCHDOG_INTERVAL)
+ except asyncio.CancelledError:
+ break
+
+ if self._app is None or self._stopping:
+ break
+
+ check_count += 1
+ status = self._get_health_status()
+
+ # Periodic heartbeat
+ if check_count % WATCHDOG_HEARTBEAT_EVERY == 0:
+ logger.info(
+ "Telegram watchdog: %s (check #%d, last_update=%s, queue=%d)",
+ status["summary"], check_count,
+ status["last_update_ago"], status["queue_size"],
+ )
+
+ if not status["healthy"]:
+ logger.warning(
+ "Telegram bot unhealthy: %s (queue=%d, last_update=%s) — rebuilding",
+ status["summary"], status["queue_size"],
+ status["last_update_ago"],
+ )
+ try:
+ await self._rebuild()
+ logger.info("Telegram bot rebuilt successfully")
+ except Exception as e:
+ logger.error("Telegram bot rebuild failed: %s", e, exc_info=True)
+
+ def _get_health_status(self) -> dict:
+ """Check health of both Updater and Application."""
+ now = time.monotonic()
+ stale_sec = now - self._last_update_time if self._last_update_time > 0 else 0
+ last_ago = f"{stale_sec:.0f}s ago" if self._last_update_time > 0 else "never"
+
+ result = {
+ "healthy": True,
+ "summary": "ok",
+ "last_update_ago": last_ago,
+ "queue_size": 0,
+ }
+
+ if not self._app:
+ result.update(healthy=False, summary="no Application")
+ return result
+
+ # --- Check update queue ---
+ # If updates pile up in the queue, the Application isn't consuming them
+ try:
+ result["queue_size"] = self._app.update_queue.qsize()
+ except Exception:
+ pass
+
+ # --- Check Updater (the part that fetches from Telegram) ---
+ updater = self._app.updater
+ if not updater or not updater.running:
+ result.update(healthy=False, summary="updater not running")
+ return result
+
+ polling_task: asyncio.Task | None = getattr(
+ updater, "_Updater__polling_task", None,
+ )
+ if polling_task is not None and polling_task.done():
+ exc = None
+ try:
+ exc = polling_task.exception()
+ except (asyncio.CancelledError, asyncio.InvalidStateError):
+ pass
+ result.update(
+ healthy=False,
+ summary=f"polling task dead (exception: {exc})",
+ )
+ return result
+
+ # --- Check Application (the part that processes updates → handlers) ---
+ fetcher_task: asyncio.Task | None = getattr(
+ self._app, "_Application__update_fetcher_task", None,
+ )
+ if fetcher_task is not None and fetcher_task.done():
+ exc = None
+ try:
+ exc = fetcher_task.exception()
+ except (asyncio.CancelledError, asyncio.InvalidStateError):
+ pass
+ result.update(
+ healthy=False,
+ summary=f"update fetcher task dead (exception: {exc})",
+ )
+ return result
+
+ # --- Check for backed-up queue (Application alive but not consuming) ---
+ if result["queue_size"] > 10:
+ result.update(
+ healthy=False,
+ summary=f"update queue backed up ({result['queue_size']} pending)",
+ )
+ return result
+
+ return result
+
+ async def _rebuild(self) -> None:
+ """Tear down and rebuild the entire PTB Application."""
+ old_app = self._app
+ self._app = self._build_application()
+
+ await self._app.initialize()
+ await self._app.start()
+ await self._app.updater.start_polling(
+ drop_pending_updates=False,
+ bootstrap_retries=-1,
+ allowed_updates=Update.ALL_TYPES,
+ )
+ self._last_update_time = time.monotonic()
+
+ # Best-effort cleanup of the old app
+ if old_app:
+ try:
+ await asyncio.wait_for(old_app.updater.stop(), timeout=5)
+ except Exception:
+ pass
+ try:
+ await asyncio.wait_for(old_app.stop(), timeout=5)
+ except Exception:
+ pass
+ try:
+ await asyncio.wait_for(old_app.shutdown(), timeout=5)
+ except Exception:
+ pass
+
# ------------------------------------------------------------------ #
# Outbound: send complete message #
# ------------------------------------------------------------------ #
@@ -124,11 +441,19 @@ async def send(self, message: OutboundMessage) -> None:
# Split long messages
for i in range(0, len(text), MAX_MSG_LEN):
chunk = text[i:i + MAX_MSG_LEN]
- await self._app.bot.send_message(
- chat_id=chat_id,
- text=chunk,
- parse_mode=ParseMode.MARKDOWN,
- )
+ html_chunk = _md_to_tg_html(chunk)
+ try:
+ sent = await self._app.bot.send_message(
+ chat_id=chat_id,
+ text=html_chunk,
+ parse_mode=ParseMode.HTML,
+ )
+ except Exception:
+ sent = await self._app.bot.send_message(
+ chat_id=chat_id,
+ text=chunk,
+ )
+ self._cache_message(sent.message_id, chat_id, chunk)
def format_response(self, text: str) -> str:
"""Truncate for Telegram if needed."""
@@ -153,11 +478,30 @@ async def edit_message(self, target: str, message_id: str, text: str) -> None:
if self._app is None:
return
chat_id = int(target)
- await self._app.bot.edit_message_text(
- chat_id=chat_id,
- message_id=int(message_id),
- text=text,
- )
+ html_text = _md_to_tg_html(text)
+ try:
+ await self._app.bot.edit_message_text(
+ chat_id=chat_id,
+ message_id=int(message_id),
+ text=html_text,
+ parse_mode=ParseMode.HTML,
+ )
+ except Exception as exc:
+ exc_str = str(exc)
+ if "message is not modified" in exc_str.lower():
+ return # Already up-to-date — don't clobber with plain text
+ logger.warning("edit_message HTML failed: %s", exc)
+ # Fallback: send without formatting if HTML parsing fails
+ try:
+ await self._app.bot.edit_message_text(
+ chat_id=chat_id,
+ message_id=int(message_id),
+ text=text,
+ )
+ except Exception:
+ pass
+ # Update cache with the latest text (streaming overwrites placeholder)
+ self._cache_message(int(message_id), int(target), text)
async def send_typing(self, target: str) -> None:
"""Show typing indicator."""
@@ -168,6 +512,35 @@ async def send_typing(self, target: str) -> None:
action=ChatAction.TYPING,
)
+ # ------------------------------------------------------------------ #
+ # Reactions: set emoji on a message #
+ # ------------------------------------------------------------------ #
+
+ async def set_reaction(self, target: str, message_id: int, emoji: str) -> None:
+ """Set an emoji reaction on a Telegram message."""
+ if self._app is None:
+ return
+ await self._app.bot.set_message_reaction(
+ chat_id=int(target),
+ message_id=message_id,
+ reaction=[emoji],
+ )
+
+ # ------------------------------------------------------------------ #
+ # Message cache (for reaction context) #
+ # ------------------------------------------------------------------ #
+
+ def _cache_message(self, message_id: int, chat_id: int, text: str) -> None:
+ """Store a message snippet in the LRU cache for reaction lookups."""
+ snippet = text[:200] if text else ""
+ if not snippet:
+ return
+ self._message_cache[message_id] = (chat_id, snippet)
+ # Move to end (most recent) and evict oldest if over limit
+ self._message_cache.move_to_end(message_id)
+ while len(self._message_cache) > self._message_cache_max:
+ self._message_cache.popitem(last=False)
+
# ------------------------------------------------------------------ #
# Auth #
# ------------------------------------------------------------------ #
@@ -184,8 +557,13 @@ def _is_authorized(self, user_id: int) -> bool:
# Command handlers — delegate to router for session management #
# ------------------------------------------------------------------ #
+ def _touch(self) -> None:
+ """Record that we received an update from Telegram."""
+ self._last_update_time = time.monotonic()
+
async def _handle_start(self, update: Update, context: Any) -> None:
"""Handle /start command."""
+ self._touch()
user_id = update.effective_user.id
if not self._is_authorized(user_id):
logger.warning("Unauthorized /start from user %d", user_id)
@@ -197,6 +575,7 @@ async def _handle_start(self, update: Update, context: Any) -> None:
async def _handle_session(self, update: Update, context: Any) -> None:
"""Handle /session — switch active session."""
+ self._touch()
if not self._is_authorized(update.effective_user.id):
return
chat_id = update.effective_chat.id
@@ -222,6 +601,7 @@ async def _handle_session(self, update: Update, context: Any) -> None:
async def _handle_sessions(self, update: Update, context: Any) -> None:
"""Handle /sessions — list sessions."""
+ self._touch()
if not self._is_authorized(update.effective_user.id):
return
@@ -236,42 +616,291 @@ async def _handle_sessions(self, update: Update, context: Any) -> None:
await update.message.reply_text("\n".join(lines), parse_mode=ParseMode.MARKDOWN)
async def _handle_new_session(self, update: Update, context: Any) -> None:
- """Handle /new [title] — create and switch to a new session."""
+ """Handle /new [title] — stop current session, create and switch to a new one."""
+ self._touch()
if not self._is_authorized(update.effective_user.id):
return
chat_id = update.effective_chat.id
+ channel_key = f"telegram:{chat_id}"
+
+ # Stop the current session before creating a new one
+ prev = await self.router.get_last_session(channel_key)
+ if prev:
+ stopped = await self.router.engine.stop_session(prev)
+ if stopped:
+ await update.message.reply_text(
+ f"Stopped session `{prev}`.",
+ parse_mode=ParseMode.MARKDOWN,
+ )
title = " ".join(context.args) if context.args else None
session_id = await self.router.create_session(
- f"telegram:{chat_id}", title=title, source="telegram",
+ channel_key, title=title, source="telegram",
)
await update.message.reply_text(
f"New session: `{session_id}`" + (f" — {title}" if title else ""),
parse_mode=ParseMode.MARKDOWN,
)
+ async def _handle_stop(self, update: Update, context: Any) -> None:
+ """Handle /stop — stop the currently running session."""
+ self._touch()
+ if not self._is_authorized(update.effective_user.id):
+ return
+ chat_id = update.effective_chat.id
+ channel_key = f"telegram:{chat_id}"
+
+ session_id = await self.router.get_last_session(channel_key)
+ if not session_id:
+ await update.message.reply_text("No active session.")
+ return
+
+ stopped = await self.router.engine.stop_session(session_id)
+ if stopped:
+ await update.message.reply_text(
+ f"Stopped session `{session_id}`.",
+ parse_mode=ParseMode.MARKDOWN,
+ )
+ else:
+ await update.message.reply_text("Nothing running to stop.")
+
# ------------------------------------------------------------------ #
# Message handler — construct InboundMessage and delegate #
# ------------------------------------------------------------------ #
+ async def _extract_image(self, message: Any) -> dict[str, str] | None:
+ """Download and base64-encode an image from a Telegram message."""
+ if message.photo:
+ # Telegram provides multiple resolutions; pick the largest
+ photo = message.photo[-1]
+ tg_file = await photo.get_file()
+ data = await tg_file.download_as_bytearray()
+ return {
+ "type": "base64",
+ "media_type": "image/jpeg",
+ "data": base64.b64encode(bytes(data)).decode("utf-8"),
+ }
+ return None
+
async def _handle_message(self, update: Update, context: Any) -> None:
- """Handle incoming text messages — delegate to router."""
+ """Handle incoming text and photo messages — delegate to router."""
+ self._touch()
if not self._is_authorized(update.effective_user.id):
return
+
+ # Media group (album) — collect all parts before processing
+ if update.message.media_group_id:
+ await self._collect_media_group(update)
+ return
+
chat_id = update.effective_chat.id
+ text = update.message.text or update.message.caption or ""
+
+ # Cache for reaction lookups (raw text, before reply-context prefix)
+ self._cache_message(update.message.message_id, chat_id, text)
+
+ # Prepend reply-to context and quote (if replying to a message)
+ reply_context = _format_reply_context(update.message)
+ if reply_context:
+ text = f"{reply_context}\n\n{text}" if text else reply_context
+
+ # Extract image if present
+ images: list[dict[str, str]] = []
+ image = await self._extract_image(update.message)
+ if image:
+ images.append(image)
+
+ if not text and not images:
+ return
+
+ logger.info(
+ "Telegram message from %s: %s%s",
+ chat_id,
+ (text[:80] + ("..." if len(text) > 80 else "")) if text else "(no text)",
+ f" [{len(images)} image(s)]" if images else "",
+ )
+
+ metadata: dict[str, Any] = {"message_id": update.message.message_id}
+ if images:
+ metadata["images"] = images
msg = InboundMessage(
channel_name="telegram",
channel_key=f"telegram:{chat_id}",
sender_id=str(chat_id),
- text=update.message.text,
+ text=text,
+ metadata=metadata,
)
try:
await self.router.handle_message(msg)
except Exception as e:
- logger.error("Agent error: %s", e)
- await update.message.reply_text(f"Error: {e}")
+ logger.error("Agent error for chat %s: %s", chat_id, e, exc_info=True)
+ try:
+ await update.message.reply_text(f"Error: {e}")
+ except Exception:
+ logger.error("Failed to send error reply to chat %s", chat_id)
+
+ # ------------------------------------------------------------------ #
+ # Reaction handler #
+ # ------------------------------------------------------------------ #
+
+ async def _handle_reaction(self, update: Update, context: Any) -> None:
+ """Handle message reaction updates — forward as text to router."""
+ self._touch()
+ reaction_update = update.message_reaction
+ if reaction_update is None:
+ return
+
+ user = reaction_update.user
+ if user is None or not self._is_authorized(user.id):
+ return
+
+ chat_id = reaction_update.chat.id
+ message_id = reaction_update.message_id
+
+ # Extract new emoji reactions (standard + premium custom emoji)
+ emojis = []
+ custom_ids = []
+ for r in reaction_update.new_reaction:
+ emoji = getattr(r, "emoji", None)
+ if emoji:
+ emojis.append(emoji)
+ else:
+ custom_id = getattr(r, "custom_emoji_id", None)
+ if custom_id:
+ custom_ids.append(custom_id)
+
+ # Resolve premium custom emoji IDs to their base emoji
+ if custom_ids:
+ try:
+ stickers = await self._app.bot.get_custom_emoji_stickers(custom_ids)
+ for sticker in stickers:
+ emojis.append(sticker.emoji or f"[premium:{sticker.custom_emoji_id}]")
+ except Exception as e:
+ logger.warning("Failed to resolve custom emoji: %s", e)
+ emojis.extend(f"[premium:{cid}]" for cid in custom_ids)
+
+ if not emojis:
+ # Reaction removed — ignore for now
+ return
+
+ emoji_str = " ".join(emojis)
+
+ # Look up original message text from cache
+ cached = self._message_cache.get(message_id)
+ if cached:
+ _, original_text = cached
+ text = f'[Reaction: {emoji_str} on message: "{original_text}"]'
+ else:
+ text = f"[Reaction: {emoji_str}]"
+
+ logger.info("Telegram reaction from %s: %s (msg %d)", chat_id, emoji_str, message_id)
+
+ msg = InboundMessage(
+ channel_name="telegram",
+ channel_key=f"telegram:{chat_id}",
+ sender_id=str(chat_id),
+ text=text,
+ metadata={},
+ )
+
+ try:
+ await self.router.handle_message(msg)
+ except Exception as e:
+ logger.error("Agent error for reaction in chat %s: %s", chat_id, e, exc_info=True)
+
+ # ------------------------------------------------------------------ #
+ # Media group (album) collection #
+ # ------------------------------------------------------------------ #
+
+ async def _collect_media_group(self, update: Update) -> None:
+ """Buffer a media-group message; schedule processing after a short delay."""
+ group_id = update.message.media_group_id
+ if group_id not in self._media_groups:
+ self._media_groups[group_id] = []
+ self._media_groups[group_id].append(update)
+
+ # Reset the timer — wait for remaining album parts
+ task = self._media_group_tasks.get(group_id)
+ if task:
+ task.cancel()
+ self._media_group_tasks[group_id] = asyncio.create_task(
+ self._process_media_group(group_id),
+ )
+
+ async def _process_media_group(self, group_id: str) -> None:
+ """Wait briefly for all album parts, then send as one message."""
+ await asyncio.sleep(0.5)
+
+ updates = self._media_groups.pop(group_id, [])
+ self._media_group_tasks.pop(group_id, None)
+ if not updates:
+ return
+
+ chat_id = updates[0].effective_chat.id
+
+ # Caption is usually on the first message only
+ text = ""
+ for u in updates:
+ caption = u.message.text or u.message.caption or ""
+ if caption:
+ text = caption
+ break
+
+ # Prepend reply-to context (album reply info is on the first message)
+ reply_context = _format_reply_context(updates[0].message)
+ if reply_context:
+ text = f"{reply_context}\n\n{text}" if text else reply_context
+
+ # Download all images
+ images: list[dict[str, str]] = []
+ for u in updates:
+ image = await self._extract_image(u.message)
+ if image:
+ images.append(image)
+
+ if not text and not images:
+ return
+
+ logger.info(
+ "Telegram media group from %s: %d image(s), caption: %s",
+ chat_id, len(images),
+ (text[:80] + "..." if len(text) > 80 else text) if text else "(none)",
+ )
+
+ metadata: dict[str, Any] = {"message_id": updates[0].message.message_id}
+ if images:
+ metadata["images"] = images
+
+ msg = InboundMessage(
+ channel_name="telegram",
+ channel_key=f"telegram:{chat_id}",
+ sender_id=str(chat_id),
+ text=text,
+ metadata=metadata,
+ )
+
+ try:
+ await self.router.handle_message(msg)
+ except Exception as e:
+ logger.error("Agent error for chat %s: %s", chat_id, e, exc_info=True)
+ try:
+ await updates[0].message.reply_text(f"Error: {e}")
+ except Exception:
+ logger.error("Failed to send error reply to chat %s", chat_id)
+
+ # ------------------------------------------------------------------ #
+ # Error handler #
+ # ------------------------------------------------------------------ #
+
+ async def _handle_error(self, update: object, context: Any) -> None:
+ """Log errors from the Telegram bot polling/handler pipeline."""
+ self._touch()
+ logger.error(
+ "Telegram update error: %s (update=%s)",
+ context.error, update, exc_info=context.error,
+ )
# ------------------------------------------------------------------ #
# Notification callback handlers #
@@ -279,6 +908,7 @@ async def _handle_message(self, update: Update, context: Any) -> None:
async def _handle_callback_query(self, update: Update, context: Any) -> None:
"""Handle inline keyboard button presses for notification questions."""
+ self._touch()
query = update.callback_query
if not query or not query.data:
return
@@ -321,6 +951,7 @@ async def _handle_callback_query(self, update: Update, context: Any) -> None:
async def _handle_reply(self, update: Update, context: Any) -> None:
"""Handle /reply — answer the most recent pending question."""
+ self._touch()
if not self._is_authorized(update.effective_user.id):
return
if not context.args:
diff --git a/nerve/cli.py b/nerve/cli.py
index ea251e91..1d40cd33 100644
--- a/nerve/cli.py
+++ b/nerve/cli.py
@@ -330,7 +330,13 @@ def stop(ctx: click.Context) -> None:
@main.command()
@click.pass_context
def restart(ctx: click.Context) -> None:
- """Restart the Nerve daemon."""
+ """Restart the Nerve daemon.
+
+ Spawns a detached helper process that stops the old daemon and starts a
+ new one. This ensures restarts work even when triggered from *inside*
+ the running daemon (e.g. via the Telegram bot / agent), because the
+ helper runs in its own session and survives the parent's death.
+ """
config = ctx.obj["config"]
config_dir = Path(ctx.obj["config_dir"])
@@ -342,32 +348,76 @@ def restart(ctx: click.Context) -> None:
ctx.exit(rc)
return
- running, pid = _get_daemon_status()
- if running:
- click.echo(f"Stopping Nerve (PID {pid})...")
- os.kill(pid, signal.SIGTERM)
-
- # Wait for shutdown
- for _ in range(30):
- time.sleep(0.5)
- if not _is_running(pid):
- break
- else:
- click.echo("Graceful shutdown timed out, sending SIGKILL...")
- try:
- os.kill(pid, signal.SIGKILL)
- time.sleep(0.5)
- except ProcessLookupError:
- pass
-
- _remove_pid()
- click.echo("Nerve stopped")
+ # Build the command that `start` would use to launch the daemon.
+ verbose = ctx.obj["verbose"]
+ nerve_bin = sys.argv[0]
+ start_cmd_parts = [sys.executable, nerve_bin, "-c", str(config_dir)]
+ if verbose:
+ start_cmd_parts.append("-v")
+ start_cmd_parts.extend(["start", "--foreground"])
+
+ running, old_pid = _get_daemon_status()
+
+ # Spawn a detached helper that: waits for old PID to exit, then starts
+ # a new daemon. Written as an inline Python script so we don't need an
+ # external shell script on disk.
+ helper_script = (
+ "import os, signal, subprocess, sys, time\n"
+ f"old_pid = {old_pid if running else 'None'}\n"
+ f"pid_file = {str(PID_FILE)!r}\n"
+ f"log_file = {str(LOG_FILE)!r}\n"
+ f"start_cmd = {start_cmd_parts!r}\n"
+ "if old_pid is not None:\n"
+ " try:\n"
+ " os.kill(old_pid, signal.SIGTERM)\n"
+ " except ProcessLookupError:\n"
+ " pass\n"
+ " for _ in range(30):\n"
+ " time.sleep(0.5)\n"
+ " try:\n"
+ " os.kill(old_pid, 0)\n"
+ " except ProcessLookupError:\n"
+ " break\n"
+ " else:\n"
+ " try:\n"
+ " os.kill(old_pid, signal.SIGKILL)\n"
+ " time.sleep(0.5)\n"
+ " except ProcessLookupError:\n"
+ " pass\n"
+ " # Remove stale PID file\n"
+ " try:\n"
+ " os.unlink(pid_file)\n"
+ " except FileNotFoundError:\n"
+ " pass\n"
+ "time.sleep(0.5)\n"
+ "log_fd = open(log_file, 'a')\n"
+ "proc = subprocess.Popen(\n"
+ " start_cmd,\n"
+ " stdout=log_fd,\n"
+ " stderr=log_fd,\n"
+ " stdin=subprocess.DEVNULL,\n"
+ " start_new_session=True,\n"
+ ")\n"
+ "log_fd.close()\n"
+ "time.sleep(1)\n"
+ "if proc.poll() is not None:\n"
+ " sys.exit(1)\n"
+ )
- # Brief pause before restart
- time.sleep(0.5)
+ log_fd = open(LOG_FILE, "a")
+ subprocess.Popen(
+ [sys.executable, "-c", helper_script],
+ stdout=log_fd,
+ stderr=log_fd,
+ stdin=subprocess.DEVNULL,
+ start_new_session=True,
+ )
+ log_fd.close()
- # Start again — invoke start command
- ctx.invoke(start)
+ if running:
+ click.echo(f"Restarting Nerve (PID {old_pid})... new instance will start shortly.")
+ else:
+ click.echo("Starting Nerve... new instance will start shortly.")
@main.command()
diff --git a/nerve/gateway/server.py b/nerve/gateway/server.py
index 662879cb..d9a69946 100644
--- a/nerve/gateway/server.py
+++ b/nerve/gateway/server.py
@@ -91,13 +91,13 @@ async def lifespan(app: FastAPI):
set_notification_service(notification_service)
# Start Telegram bot if enabled
- telegram_task = None
+ telegram_channel = None
if config.telegram.enabled and config.telegram.bot_token:
from nerve.channels.telegram import TelegramChannel
- telegram = TelegramChannel(config, _engine.router)
- telegram.set_notification_service(notification_service)
- _engine.register_channel(telegram)
- telegram_task = asyncio.create_task(telegram.start())
+ telegram_channel = TelegramChannel(config, _engine.router)
+ telegram_channel.set_notification_service(notification_service)
+ _engine.register_channel(telegram_channel)
+ await telegram_channel.start()
logger.info("Telegram bot started")
# Start cron service
@@ -187,16 +187,19 @@ async def _periodic_notify_expiry():
yield
+ # Shutdown: stop telegram FIRST, before cancelling background tasks.
+ # Background task cancellation propagates through anyio cancel scopes
+ # (Starlette runs the lifespan in an anyio context), which can kill
+ # the telegram polling task before we get a chance to stop it cleanly.
+ if telegram_channel:
+ await telegram_channel.stop()
+ if cron_task:
+ await cron_task.stop()
+
notify_expiry_task.cancel()
idle_sweep_task.cancel()
memorize_task.cancel()
cleanup_task.cancel()
-
- # Shutdown
- if cron_task:
- await cron_task.stop()
- if telegram_task:
- telegram_task.cancel()
await _engine.shutdown()
await close_db()
if proxy_service:
diff --git a/nerve/memory/memu_bridge.py b/nerve/memory/memu_bridge.py
index 17e91307..b3170a0f 100644
--- a/nerve/memory/memu_bridge.py
+++ b/nerve/memory/memu_bridge.py
@@ -1121,6 +1121,10 @@ async def _timeout_chat(
prompt, *, max_tokens=None, system_prompt=None, temperature=0.2,
_orig=original_chat, _prof=profile,
):
+ # Anthropic API requires max_tokens >= 1; memU sometimes
+ # omits it. Default to 4096 to prevent 400 errors.
+ if max_tokens is None:
+ max_tokens = 4096
t0 = time.monotonic()
try:
return await asyncio.wait_for(
@@ -1546,6 +1550,9 @@ def _get_anthropic_client(self) -> Any:
return self._anthropic_client
import anthropic
base_url = self.config.anthropic_api_base_url.rstrip("/")
+ # Strip /v1/ suffix — Anthropic SDK prepends it internally.
+ if base_url.endswith("/v1"):
+ base_url = base_url[:-3]
self._anthropic_client = anthropic.Anthropic(
api_key=self.config.effective_api_key,
base_url=base_url,
diff --git a/nerve/sources/registry.py b/nerve/sources/registry.py
index a3c7cfde..4aa72c00 100644
--- a/nerve/sources/registry.py
+++ b/nerve/sources/registry.py
@@ -7,7 +7,7 @@
from __future__ import annotations
import logging
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
from nerve.sources.runner import SourceRunner
@@ -34,12 +34,13 @@ def build_source_runners(
ttl_days = config.sync.message_ttl_days
# Build condense config from API credentials
- condense_cfg: dict[str, str] | None = None
+ condense_cfg: dict[str, Any] | None = None
if config.effective_api_key and config.memory.fast_model:
condense_cfg = {
"api_key": config.effective_api_key,
"model": config.memory.fast_model,
"base_url": config.anthropic_api_base_url,
+ "use_proxy": config.proxy.enabled,
}
# Telegram
diff --git a/nerve/sources/runner.py b/nerve/sources/runner.py
index f0b61ee9..6af0918c 100644
--- a/nerve/sources/runner.py
+++ b/nerve/sources/runner.py
@@ -173,7 +173,12 @@ async def _update_processed_content(self, processed_map: dict[str, str]) -> None
# ------------------------------------------------------------------
async def _get_condense_client(self) -> Any | None:
- """Get or create the shared AsyncAnthropic client for condensation."""
+ """Get or create the shared AsyncAnthropic client for condensation.
+
+ Returns None when proxy mode is active (uses httpx directly).
+ """
+ if self.condense_config.get("use_proxy"):
+ return None # Proxy uses OpenAI-compatible httpx calls
if self._condense_client is not None:
return self._condense_client
api_key = self.condense_config.get("api_key")
@@ -187,10 +192,44 @@ async def _get_condense_client(self) -> Any | None:
base_url = self.condense_config.get("base_url")
kwargs: dict[str, Any] = {"api_key": api_key}
if base_url:
- kwargs["base_url"] = base_url.rstrip("/")
+ # Strip /v1/ suffix — Anthropic SDK prepends it internally,
+ # so including it here causes /v1/v1/messages (404).
+ url = base_url.rstrip("/")
+ if url.endswith("/v1"):
+ url = url[:-3]
+ kwargs["base_url"] = url
self._condense_client = anthropic.AsyncAnthropic(**kwargs)
return self._condense_client
+ async def _condense_via_proxy(
+ self, content: str, model: str,
+ ) -> str:
+ """Condense content via the OpenAI-compatible proxy endpoint."""
+ import httpx
+
+ base_url = self.condense_config.get("base_url", "")
+ if not base_url.endswith("/"):
+ base_url += "/"
+ api_key = self.condense_config.get("api_key", "")
+
+ async with httpx.AsyncClient() as http_client:
+ resp = await http_client.post(
+ f"{base_url}chat/completions",
+ headers={
+ "Authorization": f"Bearer {api_key}",
+ "Content-Type": "application/json",
+ },
+ json={
+ "model": model,
+ "max_tokens": 1024,
+ "messages": [{"role": "user", "content": content}],
+ },
+ timeout=30,
+ )
+ resp.raise_for_status()
+ data = resp.json()
+ return data["choices"][0]["message"]["content"]
+
async def _condense_long_content(
self, records: list[SourceRecord],
) -> list[SourceRecord]:
@@ -211,13 +250,21 @@ async def _condense_long_content(
if not long_records:
return records
- client = await self._get_condense_client()
- if not client:
- return records
+ use_proxy = self.condense_config.get("use_proxy", False)
+
+ if not use_proxy:
+ client = await self._get_condense_client()
+ if not client:
+ return records
+ else:
+ client = None
+ if not self.condense_config.get("api_key"):
+ return records
logger.info(
- "Source %s: condensing %d/%d records with %s",
+ "Source %s: condensing %d/%d records with %s%s",
self.source.source_name, len(long_records), len(records), model,
+ " (via proxy)" if use_proxy else "",
)
sem = asyncio.Semaphore(5)
@@ -225,22 +272,28 @@ async def condense_one(record: SourceRecord) -> None:
async with sem:
original_len = len(record.content)
try:
- response = await asyncio.wait_for(
- client.messages.create(
- model=model,
- max_tokens=1024,
- messages=[{
- "role": "user",
- "content": (
- f"{_CONDENSE_PROMPT}\n\n"
- f"---\n\n"
- f"{record.content}"
- ),
- }],
- ),
- timeout=30,
+ prompt_content = (
+ f"{_CONDENSE_PROMPT}\n\n"
+ f"---\n\n"
+ f"{record.content}"
)
- condensed = response.content[0].text
+ if use_proxy:
+ condensed = await self._condense_via_proxy(
+ prompt_content, model,
+ )
+ else:
+ response = await asyncio.wait_for(
+ client.messages.create(
+ model=model,
+ max_tokens=1024,
+ messages=[{
+ "role": "user",
+ "content": prompt_content,
+ }],
+ ),
+ timeout=30,
+ )
+ condensed = response.content[0].text
record.content = condensed
logger.debug(
"Condensed %s: %d → %d chars",
diff --git a/tests/test_bootstrap.py b/tests/test_bootstrap.py
index c9eb972b..9172fbee 100644
--- a/tests/test_bootstrap.py
+++ b/tests/test_bootstrap.py
@@ -544,11 +544,13 @@ def test_resolve_claude_from_oauth_env(self) -> None:
assert token == "sk-ant-oat01-test"
assert source == "CLAUDE_CODE_OAUTH_TOKEN env var"
- def test_resolve_claude_from_api_key_env(self) -> None:
+ def test_resolve_claude_from_api_key_env(self, tmp_path: Path) -> None:
"""ANTHROPIC_API_KEY should be last resort in waterfall."""
env = {"ANTHROPIC_API_KEY": "sk-ant-api03-test"}
- # Clear OAuth token to ensure it doesn't interfere
- with patch.dict(os.environ, env, clear=False):
+ # Point credentials file at a non-existent path so the real one isn't found
+ fake_creds = tmp_path / "nonexistent" / ".credentials.json"
+ with patch.dict(os.environ, env, clear=False), \
+ patch("nerve.bootstrap.Path.expanduser", return_value=fake_creds):
# Remove CLAUDE_CODE_OAUTH_TOKEN if set
os.environ.pop("CLAUDE_CODE_OAUTH_TOKEN", None)
token, source, debug = _resolve_claude_credential()
@@ -569,10 +571,13 @@ def test_resolve_claude_from_credentials_file(self, tmp_path: Path) -> None:
assert token == "sk-ant-oat01-file"
assert "credentials.json" in source
- def test_resolve_claude_none(self) -> None:
+ def test_resolve_claude_none(self, tmp_path: Path) -> None:
"""Should return empty when no credentials found."""
+ # Point credentials file at a non-existent path so the real one isn't found
+ fake_creds = tmp_path / "nonexistent" / ".credentials.json"
with patch.dict(os.environ, {}, clear=True), \
- patch("nerve.bootstrap.sys.platform", "linux"):
+ patch("nerve.bootstrap.sys.platform", "linux"), \
+ patch("nerve.bootstrap.Path.expanduser", return_value=fake_creds):
token, source, debug = _resolve_claude_credential()
assert token == ""
assert source == "none"
diff --git a/tests/test_proxy.py b/tests/test_proxy.py
index f5ee2791..e0c88718 100644
--- a/tests/test_proxy.py
+++ b/tests/test_proxy.py
@@ -534,3 +534,68 @@ def test_registry_no_proxy_uses_raw_key(self) -> None:
assert len(runners) >= 1
assert runners[0].condense_config["api_key"] == "sk-ant-real-key"
assert "api.anthropic.com" in runners[0].condense_config["base_url"]
+
+
+class TestCondenseClientBaseUrl:
+ """Verify condense client strips /v1 suffix to avoid /v1/v1/messages."""
+
+ @pytest.mark.asyncio
+ async def test_condense_client_strips_v1_from_proxy_url(self) -> None:
+ """Proxy base_url includes /v1/ — SDK must not double it."""
+ from nerve.sources.runner import SourceRunner
+
+ runner = SourceRunner(
+ source=MagicMock(),
+ db=MagicMock(),
+ condense=True,
+ condense_config={
+ "api_key": "sk-test",
+ "model": "claude-haiku-4-5-20251001",
+ "base_url": "http://127.0.0.1:8317/v1/",
+ },
+ )
+ client = await runner._get_condense_client()
+ assert client is not None
+ base = str(client.base_url)
+ # Must NOT contain /v1/v1
+ assert "/v1/v1" not in base
+ # Must end with the proxy host (no /v1 path)
+ assert base.rstrip("/") == "http://127.0.0.1:8317"
+
+ @pytest.mark.asyncio
+ async def test_condense_client_strips_v1_from_direct_url(self) -> None:
+ """Direct API base_url also includes /v1/ — same fix applies."""
+ from nerve.sources.runner import SourceRunner
+
+ runner = SourceRunner(
+ source=MagicMock(),
+ db=MagicMock(),
+ condense=True,
+ condense_config={
+ "api_key": "sk-test",
+ "model": "claude-haiku-4-5-20251001",
+ "base_url": "https://api.anthropic.com/v1/",
+ },
+ )
+ client = await runner._get_condense_client()
+ assert client is not None
+ base = str(client.base_url)
+ assert "/v1/v1" not in base
+ assert base.rstrip("/") == "https://api.anthropic.com"
+
+ @pytest.mark.asyncio
+ async def test_condense_client_no_base_url(self) -> None:
+ """Without base_url, SDK uses its default — no crash."""
+ from nerve.sources.runner import SourceRunner
+
+ runner = SourceRunner(
+ source=MagicMock(),
+ db=MagicMock(),
+ condense=True,
+ condense_config={
+ "api_key": "sk-test",
+ "model": "claude-haiku-4-5-20251001",
+ },
+ )
+ client = await runner._get_condense_client()
+ assert client is not None