Grayscale/nanobot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge origin/main into feat/session-replay-file-cap-invariants

Browse Source 
Preserve main's timestamp/tool-context replay semantics while keeping the PR's session history and file-cap budgets.

Made-with: Cursor
This commit is contained in:
Xubin Ren
2026-04-27 07:32:00 +00:00
parent 59dfd74842 311a7fe36e
commit 29ebc2d355
30 changed files with 1297 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nanobot/agent/loop.py
+63 -10
View File
@@ -76,6 +76,8 @@ class _LoopHook(AgentHook):
channel: str = "cli",
chat_id: str = "direct",
message_id: str | None = None,
metadata: dict[str, Any] | None = None,
session_key: str | None = None,
) -> None:
super().__init__(reraise=True)
self._loop = agent_loop
@@ -85,6 +87,8 @@ class _LoopHook(AgentHook):
self._channel = channel
self._chat_id = chat_id
self._message_id = message_id
self._metadata = metadata or {}
self._session_key = session_key
self._stream_buf = ""
def wants_streaming(self) -> bool:
@@ -127,7 +131,13 @@ class _LoopHook(AgentHook):
for tc in context.tool_calls:
args_str = json.dumps(tc.arguments, ensure_ascii=False)
logger.info("Tool call: {}({})", tc.name, args_str[:200])
self._loop._set_tool_context(self._channel, self._chat_id, self._message_id)
self._loop._set_tool_context(
self._channel,
self._chat_id,
self._message_id,
self._metadata,
session_key=self._session_key,
)
async def after_iteration(self, context: AgentHookContext) -> None:
if (
@@ -376,7 +386,7 @@ class AgentLoop:
WebSearchTool(config=self.web_config.search, proxy=self.web_config.proxy)
)
self.tools.register(WebFetchTool(proxy=self.web_config.proxy))
self.tools.register(MessageTool(send_callback=self.bus.publish_outbound))
self.tools.register(MessageTool(send_callback=self.bus.publish_outbound, workspace=self.workspace))
self.tools.register(SpawnTool(manager=self.subagents))
if self.cron_service:
self.tools.register(
@@ -405,18 +415,33 @@ class AgentLoop:
finally:
self._mcp_connecting = False
def _set_tool_context(self, channel: str, chat_id: str, message_id: str | None = None) -> None:
def _set_tool_context(
self, channel: str, chat_id: str,
message_id: str | None = None, metadata: dict | None = None,
session_key: str | None = None,
) -> None:
"""Update context for all tools that need routing info."""
# Compute the effective session key (accounts for unified sessions)
# so that subagent results route to the correct pending queue.
effective_key = UNIFIED_SESSION_KEY if self._unified_session else f"{channel}:{chat_id}"
# When the caller threads a thread-scoped session_key (e.g. slack with
# reply_in_thread: true), honor it so spawn announces route back to
# the originating thread session. Falls back to unified mode or
# channel:chat_id for callers that don't have a thread-scoped key.
if session_key is not None:
effective_key = session_key
elif self._unified_session:
effective_key = UNIFIED_SESSION_KEY
else:
effective_key = f"{channel}:{chat_id}"
for name in ("message", "spawn", "cron", "my"):
if tool := self.tools.get(name):
if hasattr(tool, "set_context"):
if name == "spawn":
tool.set_context(channel, chat_id, effective_key=effective_key)
elif name == "cron":
tool.set_context(channel, chat_id, metadata=metadata, session_key=session_key)
elif name == "message":
tool.set_context(channel, chat_id, message_id, metadata=metadata)
else:
tool.set_context(channel, chat_id, *([message_id] if name == "message" else []))
tool.set_context(channel, chat_id)
@staticmethod
def _strip_think(text: str | None) -> str | None:
@@ -525,6 +550,8 @@ class AgentLoop:
channel: str = "cli",
chat_id: str = "direct",
message_id: str | None = None,
metadata: dict[str, Any] | None = None,
session_key: str | None = None,
pending_queue: asyncio.Queue | None = None,
) -> tuple[str | None, list[str], list[dict], str, bool]:
"""Run the agent iteration loop.
@@ -544,6 +571,8 @@ class AgentLoop:
channel=channel,
chat_id=chat_id,
message_id=message_id,
metadata=metadata,
session_key=session_key,
)
hook: AgentHook = (
CompositeHook([loop_hook] + self._extra_hooks) if self._extra_hooks else loop_hook
@@ -871,7 +900,10 @@ class AgentLoop:
msg.chat_id.split(":", 1) if ":" in msg.chat_id else ("cli", msg.chat_id)
)
logger.info("Processing system message from {}", msg.sender_id)
key = f"{channel}:{chat_id}"
# Honor session_key_override so subagent announces from threaded
# callers route to the originating thread session, not the
# channel-level session derived from chat_id.
key = msg.session_key_override or f"{channel}:{chat_id}"
session = self.sessions.get_or_create(key)
if self._restore_runtime_checkpoint(session):
self.sessions.save(session)
@@ -892,10 +924,14 @@ class AgentLoop:
is_subagent = msg.sender_id == "subagent"
if is_subagent and self._persist_subagent_followup(session, msg):
self.sessions.save(session)
self._set_tool_context(channel, chat_id, msg.metadata.get("message_id"))
self._set_tool_context(
channel, chat_id, msg.metadata.get("message_id"),
msg.metadata, session_key=key,
)
history = session.get_history(
max_messages=self.session_history_max_messages,
max_tokens=self._history_token_budget(),
include_timestamps=True,
)
current_role = "assistant" if is_subagent else "user"
@@ -912,6 +948,8 @@ class AgentLoop:
final_content, _, all_msgs, stop_reason, _ = await self._run_agent_loop(
messages, session=session, channel=channel, chat_id=chat_id,
message_id=msg.metadata.get("message_id"),
metadata=msg.metadata,
session_key=key,
pending_queue=pending_queue,
)
self._save_turn(session, all_msgs, 1 + len(history))
@@ -925,11 +963,20 @@ class AgentLoop:
options,
channel,
)
# Reconstruct channel-specific metadata from session.key so the
# outbound reply lands in the originating thread (not the channel
# top-level). The announce InboundMessage carries only
# injected_event metadata; we recover thread_ts from the session
# key, which slack writes as "slack:<chat_id>:<thread_ts>".
outbound_metadata: dict[str, Any] = {}
if channel == "slack" and key.startswith("slack:") and key.count(":") >= 2:
outbound_metadata["slack"] = {"thread_ts": key.split(":", 2)[2]}
return OutboundMessage(
channel=channel,
chat_id=chat_id,
content=content,
buttons=buttons,
metadata=outbound_metadata,
)
# Extract document text from media at the processing boundary so all
@@ -961,7 +1008,10 @@ class AgentLoop:
session_summary=pending,
)
self._set_tool_context(msg.channel, msg.chat_id, msg.metadata.get("message_id"))
self._set_tool_context(
msg.channel, msg.chat_id, msg.metadata.get("message_id"),
msg.metadata, session_key=key,
)
if message_tool := self.tools.get("message"):
if isinstance(message_tool, MessageTool):
message_tool.start_turn()
@@ -969,6 +1019,7 @@ class AgentLoop:
history = session.get_history(
max_messages=self.session_history_max_messages,
max_tokens=self._history_token_budget(),
include_timestamps=True,
)
pending_ask_id = pending_ask_user_id(history)
@@ -1046,6 +1097,8 @@ class AgentLoop:
channel=msg.channel,
chat_id=msg.chat_id,
message_id=msg.metadata.get("message_id"),
metadata=msg.metadata,
session_key=key,
pending_queue=pending_queue,
)
nanobot/agent/memory.py
+1 -1
View File
@@ -494,7 +494,7 @@ class Consolidator:
session_summary: str | None = None,
) -> tuple[int, str]:
"""Estimate current prompt size for the normal session history view."""
history = session.get_history(max_messages=0)
history = session.get_history(max_messages=0, include_timestamps=True)
channel, chat_id = (session.key.split(":", 1) if ":" in session.key else (None, None))
probe_messages = self._build_messages(
history=history,
nanobot/agent/tools/cron.py
+10 -1
View File
@@ -60,12 +60,19 @@ class CronTool(Tool):
self._default_timezone = default_timezone
self._channel: ContextVar[str] = ContextVar("cron_channel", default="")
self._chat_id: ContextVar[str] = ContextVar("cron_chat_id", default="")
self._metadata: ContextVar[dict] = ContextVar("cron_metadata", default={})
self._session_key: ContextVar[str] = ContextVar("cron_session_key", default="")
self._in_cron_context: ContextVar[bool] = ContextVar("cron_in_context", default=False)
def set_context(self, channel: str, chat_id: str) -> None:
def set_context(
self, channel: str, chat_id: str,
metadata: dict | None = None, session_key: str | None = None,
) -> None:
"""Set the current session context for delivery."""
self._channel.set(channel)
self._chat_id.set(chat_id)
self._metadata.set(metadata or {})
self._session_key.set(session_key or f"{channel}:{chat_id}")
def set_cron_context(self, active: bool):
"""Mark whether the tool is executing inside a cron job callback."""
@@ -199,6 +206,8 @@ class CronTool(Tool):
channel=channel,
to=chat_id,
delete_after_run=delete_after,
channel_meta=self._metadata.get(),
session_key=self._session_key.get() or None,
)
return f"Created job '{job.name}' (id: {job.id})"
nanobot/agent/tools/mcp.py
+15 -5
View File
@@ -2,6 +2,7 @@
import asyncio
import os
import re
import shutil
from contextlib import AsyncExitStack
from typing import Any
@@ -28,6 +29,15 @@ _TRANSIENT_EXC_NAMES: frozenset[str] = frozenset((
_WINDOWS_SHELL_LAUNCHERS: frozenset[str] = frozenset(("npx", "npm", "pnpm", "yarn", "bunx"))
# Characters allowed in tool names by model providers (Anthropic, OpenAI, etc.).
# Replace anything outside [a-zA-Z0-9_-] with underscore and collapse runs.
_SANITIZE_RE = re.compile(r"_+")
def _sanitize_name(name: str) -> str:
"""Sanitize an MCP-derived name for model API compatibility."""
return _SANITIZE_RE.sub("_", re.sub(r"[^a-zA-Z0-9_-]", "_", name))
def _is_transient(exc: BaseException) -> bool:
"""Check if an exception looks like a transient connection error."""
@@ -137,7 +147,7 @@ class MCPToolWrapper(Tool):
def __init__(self, session, server_name: str, tool_def, tool_timeout: int = 30):
self._session = session
self._original_name = tool_def.name
self._name = f"mcp_{server_name}_{tool_def.name}"
self._name = _sanitize_name(f"mcp_{server_name}_{tool_def.name}")
self._description = tool_def.description or tool_def.name
raw_schema = tool_def.inputSchema or {"type": "object", "properties": {}}
self._parameters = _normalize_schema_for_openai(raw_schema)
@@ -221,7 +231,7 @@ class MCPResourceWrapper(Tool):
def __init__(self, session, server_name: str, resource_def, resource_timeout: int = 30):
self._session = session
self._uri = resource_def.uri
self._name = f"mcp_{server_name}_resource_{resource_def.name}"
self._name = _sanitize_name(f"mcp_{server_name}_resource_{resource_def.name}")
desc = resource_def.description or resource_def.name
self._description = f"[MCP Resource] {desc}\nURI: {self._uri}"
self._parameters: dict[str, Any] = {
@@ -311,7 +321,7 @@ class MCPPromptWrapper(Tool):
def __init__(self, session, server_name: str, prompt_def, prompt_timeout: int = 30):
self._session = session
self._prompt_name = prompt_def.name
self._name = f"mcp_{server_name}_prompt_{prompt_def.name}"
self._name = _sanitize_name(f"mcp_{server_name}_prompt_{prompt_def.name}")
desc = prompt_def.description or prompt_def.name
self._description = (
f"[MCP Prompt] {desc}\n"
@@ -514,9 +524,9 @@ async def connect_mcp_servers(
registered_count = 0
matched_enabled_tools: set[str] = set()
available_raw_names = [tool_def.name for tool_def in tools.tools]
available_wrapped_names = [f"mcp_{name}_{tool_def.name}" for tool_def in tools.tools]
available_wrapped_names = [_sanitize_name(f"mcp_{name}_{tool_def.name}") for tool_def in tools.tools]
for tool_def in tools.tools:
wrapped_name = f"mcp_{name}_{tool_def.name}"
wrapped_name = _sanitize_name(f"mcp_{name}_{tool_def.name}")
if (
not allow_all_tools
and tool_def.name not in enabled_tools
nanobot/agent/tools/message.py
+31 -5
View File
@@ -1,11 +1,14 @@
"""Message tool for sending messages to users."""
import os
from contextvars import ContextVar
from pathlib import Path
from typing import Any, Awaitable, Callable
from nanobot.agent.tools.base import Tool, tool_parameters
from nanobot.agent.tools.schema import ArraySchema, StringSchema, tool_parameters_schema
from nanobot.bus.events import OutboundMessage
from nanobot.config.paths import get_workspace_path
@tool_parameters(
@@ -33,25 +36,38 @@ class MessageTool(Tool):
default_channel: str = "",
default_chat_id: str = "",
default_message_id: str | None = None,
workspace: str | Path | None = None,
):
self._send_callback = send_callback
self._workspace = Path(workspace).expanduser() if workspace is not None else get_workspace_path()
self._default_channel: ContextVar[str] = ContextVar("message_default_channel", default=default_channel)
self._default_chat_id: ContextVar[str] = ContextVar("message_default_chat_id", default=default_chat_id)
self._default_message_id: ContextVar[str | None] = ContextVar(
"message_default_message_id",
default=default_message_id,
)
self._default_metadata: ContextVar[dict[str, Any]] = ContextVar(
"message_default_metadata",
default={},
)
self._sent_in_turn_var: ContextVar[bool] = ContextVar("message_sent_in_turn", default=False)
self._record_channel_delivery_var: ContextVar[bool] = ContextVar(
"message_record_channel_delivery",
default=False,
)
def set_context(self, channel: str, chat_id: str, message_id: str | None = None) -> None:
def set_context(
self,
channel: str,
chat_id: str,
message_id: str | None = None,
metadata: dict[str, Any] | None = None,
) -> None:
"""Set the current message context."""
self._default_channel.set(channel)
self._default_chat_id.set(chat_id)
self._default_message_id.set(message_id)
self._default_metadata.set(metadata or {})
def set_send_callback(self, callback: Callable[[OutboundMessage], Awaitable[None]]) -> None:
"""Set the callback for sending messages."""
@@ -118,7 +134,8 @@ class MessageTool(Tool):
# some channels (e.g. Feishu) use it to determine the target
# conversation via their Reply API, which would route the message
# to the wrong chat entirely.
if channel == default_channel and chat_id == default_chat_id:
same_target = channel == default_channel and chat_id == default_chat_id
if same_target:
message_id = message_id or self._default_message_id.get()
else:
message_id = None
@@ -129,9 +146,18 @@ class MessageTool(Tool):
if not self._send_callback:
return "Error: Message sending not configured"
metadata = {
"message_id": message_id,
} if message_id else {}
if media:
resolved = []
for p in media:
if p.startswith(("http://", "https://")) or os.path.isabs(p):
resolved.append(p)
else:
resolved.append(str(self._workspace / p))
media = resolved
metadata = dict(self._default_metadata.get()) if same_target else {}
if message_id:
metadata["message_id"] = message_id
if self._record_channel_delivery_var.get():
metadata["_record_channel_delivery"] = True
nanobot/channels/manager.py
+1
View File
@@ -172,6 +172,7 @@ class ChannelManager:
channel=notice.channel,
chat_id=notice.chat_id,
content=format_restart_completed_message(notice.started_at_raw),
metadata=dict(notice.metadata or {}),
),
))
nanobot/channels/slack.py
+232 -13
View File
@@ -2,8 +2,10 @@
import asyncio
import re
from pathlib import Path
from typing import Any
import httpx
from loguru import logger
from pydantic import Field
from slack_sdk.socket_mode.request import SocketModeRequest
@@ -15,7 +17,9 @@ from slackify_markdown import slackify_markdown
from nanobot.bus.events import OutboundMessage
from nanobot.bus.queue import MessageBus
from nanobot.channels.base import BaseChannel
from nanobot.config.paths import get_media_dir
from nanobot.config.schema import Base
from nanobot.utils.helpers import safe_filename, split_message
class SlackDMConfig(Base):
@@ -38,12 +42,19 @@ class SlackConfig(Base):
reply_in_thread: bool = True
react_emoji: str = "eyes"
done_emoji: str = "white_check_mark"
include_thread_context: bool = True
thread_context_limit: int = 20
allow_from: list[str] = Field(default_factory=list)
group_policy: str = "mention"
group_allow_from: list[str] = Field(default_factory=list)
dm: SlackDMConfig = Field(default_factory=SlackDMConfig)
SLACK_MAX_MESSAGE_LEN = 39_000 # Slack API allows ~40k; leave margin
SLACK_DOWNLOAD_TIMEOUT = 30.0
_HTML_DOWNLOAD_PREFIXES = (b"<!doctype html", b"<html")
class SlackChannel(BaseChannel):
"""Slack channel using Socket Mode."""
@@ -57,6 +68,8 @@ class SlackChannel(BaseChannel):
def default_config(cls) -> dict[str, Any]:
return SlackConfig().model_dump(by_alias=True)
_THREAD_CONTEXT_CACHE_LIMIT = 10_000
def __init__(self, config: Any, bus: MessageBus):
if isinstance(config, dict):
config = SlackConfig.model_validate(config)
@@ -66,6 +79,7 @@ class SlackChannel(BaseChannel):
self._socket_client: SocketModeClient | None = None
self._bot_user_id: str | None = None
self._target_cache: dict[str, str] = {}
self._thread_context_attempted: set[str] = set()
async def start(self) -> None:
"""Start the Slack Socket Mode client."""
@@ -128,14 +142,17 @@ class SlackChannel(BaseChannel):
else None
)
# Slack rejects empty text payloads. Keep media-only messages media-only,
# but send a single blank message when the bot has no text or files to send.
if msg.content or not (msg.media or []):
await self._web_client.chat_postMessage(
channel=target_chat_id,
text=self._to_mrkdwn(msg.content) if msg.content else " ",
thread_ts=thread_ts_param,
)
mrkdwn = self._to_mrkdwn(msg.content) if msg.content else " "
buttons = getattr(msg, "buttons", None) or []
chunks = split_message(mrkdwn, SLACK_MAX_MESSAGE_LEN)
for index, chunk in enumerate(chunks):
kwargs: dict[str, Any] = dict(
channel=target_chat_id, text=chunk, thread_ts=thread_ts_param,
)
if buttons and index == len(chunks) - 1:
kwargs["blocks"] = self._build_button_blocks(chunk, buttons)
await self._web_client.chat_postMessage(**kwargs)
for media_path in msg.media or []:
try:
@@ -273,6 +290,9 @@ class SlackChannel(BaseChannel):
req: SocketModeRequest,
) -> None:
"""Handle incoming Socket Mode requests."""
if req.type == "interactive":
await self._on_block_action(client, req)
return
if req.type != "events_api":
return
@@ -292,8 +312,10 @@ class SlackChannel(BaseChannel):
sender_id = event.get("user")
chat_id = event.get("channel")
# Ignore bot/system messages (any subtype = not a normal user message)
if event.get("subtype"):
subtype = event.get("subtype")
# Slack uses subtype=file_share for user messages with attachments.
# Ignore other subtypes such as bot_message / message_changed / deleted.
if subtype and subtype != "file_share":
return
if self._bot_user_id and sender_id == self._bot_user_id:
return
@@ -308,7 +330,7 @@ class SlackChannel(BaseChannel):
logger.debug(
"Slack event: type={} subtype={} user={} channel={} channel_type={} text={}",
event_type,
event.get("subtype"),
subtype,
sender_id,
chat_id,
event.get("channel_type"),
@@ -327,9 +349,11 @@ class SlackChannel(BaseChannel):
text = self._strip_bot_mention(text)
thread_ts = event.get("thread_ts")
event_ts = event.get("ts")
raw_thread_ts = event.get("thread_ts")
thread_ts = raw_thread_ts
if self.config.reply_in_thread and not thread_ts:
thread_ts = event.get("ts")
thread_ts = event_ts
# Add :eyes: reaction to the triggering message (best-effort)
try:
if self._web_client and event.get("ts"):
@@ -343,12 +367,37 @@ class SlackChannel(BaseChannel):
# Thread-scoped session key for channel/group messages
session_key = f"slack:{chat_id}:{thread_ts}" if thread_ts and channel_type != "im" else None
media_paths: list[str] = []
file_markers: list[str] = []
for file_info in event.get("files") or []:
if not isinstance(file_info, dict):
continue
file_path, marker = await self._download_slack_file(file_info)
if file_path:
media_paths.append(file_path)
if marker:
file_markers.append(marker)
is_slash = text.strip().startswith("/")
content = text if is_slash else await self._with_thread_context(
text,
chat_id=chat_id,
channel_type=channel_type,
thread_ts=thread_ts,
raw_thread_ts=raw_thread_ts,
current_ts=event_ts,
)
if file_markers:
content = "\n".join(part for part in [content, *file_markers] if part)
if not content and not media_paths:
return
try:
await self._handle_message(
sender_id=sender_id,
chat_id=chat_id,
content=text,
content=content,
media=media_paths,
metadata={
"slack": {
"event": event,
@@ -361,6 +410,163 @@ class SlackChannel(BaseChannel):
except Exception:
logger.exception("Error handling Slack message from {}", sender_id)
async def _download_slack_file(self, file_info: dict[str, Any]) -> tuple[str | None, str]:
"""Download a Slack private file to the local media directory."""
file_id = str(file_info.get("id") or "file")
name = str(
file_info.get("name")
or file_info.get("title")
or file_info.get("id")
or "slack-file"
)
marker_type = "image" if str(file_info.get("mimetype") or "").startswith("image/") else "file"
marker = f"[{marker_type}: {name}]"
url = str(file_info.get("url_private_download") or file_info.get("url_private") or "")
if not url:
return None, f"[{marker_type}: {name}: missing download url]"
if not self.config.bot_token:
return None, f"[{marker_type}: {name}: missing bot token]"
filename = safe_filename(f"{file_id}_{name}")
path = Path(get_media_dir("slack")) / filename
try:
async with httpx.AsyncClient(timeout=SLACK_DOWNLOAD_TIMEOUT, follow_redirects=True) as client:
response = await client.get(
url,
headers={"Authorization": f"Bearer {self.config.bot_token}"},
)
response.raise_for_status()
if self._looks_like_html_download(response):
raise ValueError("Slack returned HTML instead of file content")
path.write_bytes(response.content)
return str(path), marker
except Exception as e:
logger.warning("Failed to download Slack file {}: {}", file_id, e)
return None, f"[{marker_type}: {name}: download failed]"
@staticmethod
def _looks_like_html_download(response: httpx.Response) -> bool:
content_type = response.headers.get("content-type", "").lower()
if "text/html" in content_type:
return True
preview = response.content[:256].lstrip().lower()
return preview.startswith(_HTML_DOWNLOAD_PREFIXES)
async def _on_block_action(self, client: SocketModeClient, req: SocketModeRequest) -> None:
"""Handle button clicks from ask_user blocks."""
await client.send_socket_mode_response(SocketModeResponse(envelope_id=req.envelope_id))
payload = req.payload or {}
actions = payload.get("actions") or []
if not actions:
return
value = str(actions[0].get("value") or "")
user_info = payload.get("user") or {}
sender_id = str(user_info.get("id") or "")
channel_info = payload.get("channel") or {}
chat_id = str(channel_info.get("id") or "")
if not sender_id or not chat_id or not value:
return
message_info = payload.get("message") or {}
thread_ts = message_info.get("thread_ts") or message_info.get("ts")
channel_type = self._infer_channel_type(chat_id)
if not self._is_allowed(sender_id, chat_id, channel_type):
return
session_key = f"slack:{chat_id}:{thread_ts}" if thread_ts else None
try:
await self._handle_message(
sender_id=sender_id,
chat_id=chat_id,
content=value,
metadata={"slack": {"thread_ts": thread_ts, "channel_type": channel_type}},
session_key=session_key,
)
except Exception:
logger.exception("Error handling Slack button click from {}", sender_id)
async def _with_thread_context(
self,
text: str,
*,
chat_id: str,
channel_type: str,
thread_ts: str | None,
raw_thread_ts: str | None,
current_ts: str | None,
) -> str:
"""Include thread history the first time the bot is pulled into a Slack thread."""
if (
not self.config.include_thread_context
or not self._web_client
or channel_type == "im"
or not raw_thread_ts
or not thread_ts
or current_ts == thread_ts
):
return text
key = f"{chat_id}:{thread_ts}"
if key in self._thread_context_attempted:
return text
if len(self._thread_context_attempted) >= self._THREAD_CONTEXT_CACHE_LIMIT:
self._thread_context_attempted.clear()
self._thread_context_attempted.add(key)
try:
response = await self._web_client.conversations_replies(
channel=chat_id,
ts=thread_ts,
limit=max(1, self.config.thread_context_limit),
)
except Exception as e:
logger.warning("Slack thread context unavailable for {}: {}", key, e)
return text
lines = self._format_thread_context(
response.get("messages", []),
current_ts=current_ts,
)
if not lines:
return text
return "Slack thread context before this mention:\n" + "\n".join(lines) + f"\n\nCurrent message:\n{text}"
def _format_thread_context(self, messages: list[dict[str, Any]], *, current_ts: str | None) -> list[str]:
lines: list[str] = []
for item in messages:
if item.get("ts") == current_ts:
continue
if item.get("subtype"):
continue
sender = str(item.get("user") or item.get("bot_id") or "unknown")
is_bot = self._bot_user_id is not None and sender == self._bot_user_id
label = "bot" if is_bot else f"<@{sender}>"
text = str(item.get("text") or "").strip()
if not text:
continue
text = self._strip_bot_mention(text)
if len(text) > 500:
text = text[:500] + "…"
lines.append(f"- {label}: {text}")
return lines
@staticmethod
def _build_button_blocks(text: str, buttons: list[list[str]]) -> list[dict[str, Any]]:
"""Build Slack Block Kit blocks with action buttons for ask_user choices."""
blocks: list[dict[str, Any]] = [
{"type": "section", "text": {"type": "mrkdwn", "text": text[:3000]}},
]
elements = []
for row in buttons:
for label in row:
elements.append({
"type": "button",
"text": {"type": "plain_text", "text": label[:75]},
"value": label[:75],
"action_id": f"ask_user_{label[:50]}",
})
if elements:
blocks.append({"type": "actions", "elements": elements[:25]})
return blocks
async def _update_react_emoji(self, chat_id: str, ts: str | None) -> None:
"""Remove the in-progress reaction and optionally add a done reaction."""
if not self._web_client or not ts:
@@ -407,6 +613,19 @@ class SlackChannel(BaseChannel):
return chat_id in self.config.group_allow_from
return False
def is_allowed(self, sender_id: str) -> bool:
# Slack needs channel-aware policy checks, so _on_socket_request and
# _on_block_action call _is_allowed before handing off to BaseChannel.
return True
@staticmethod
def _infer_channel_type(chat_id: str) -> str:
if chat_id.startswith("D"):
return "im"
if chat_id.startswith("G"):
return "group"
return "channel"
def _strip_bot_mention(self, text: str) -> str:
if not text or not self._bot_user_id:
return text
nanobot/cli/commands.py
+12 -5
View File
@@ -672,7 +672,9 @@ def _run_gateway(
else f"{channel}:{chat_id}"
)
async def _deliver_to_channel(msg: OutboundMessage, *, record: bool = False) -> None:
async def _deliver_to_channel(
msg: OutboundMessage, *, record: bool = False, session_key: str | None = None,
) -> None:
"""Publish a user-visible message and mirror it into that channel's session."""
metadata = dict(msg.metadata or {})
record = record or bool(metadata.pop("_record_channel_delivery", False))
@@ -693,7 +695,8 @@ def _run_gateway(
and hasattr(session_manager, "get_or_create")
and hasattr(session_manager, "save")
):
session = session_manager.get_or_create(_channel_session_key(msg.channel, msg.chat_id))
key = session_key or _channel_session_key(msg.channel, msg.chat_id)
session = session_manager.get_or_create(key)
session.add_message("assistant", msg.content, _channel_delivery=True)
session_manager.save(session)
await bus.publish_outbound(msg)
@@ -717,9 +720,11 @@ def _run_gateway(
from nanobot.utils.evaluator import evaluate_response
reminder_note = (
"[Scheduled Task] Timer finished.\n\n"
f"Task '{job.name}' has been triggered.\n"
f"Scheduled instruction: {job.payload.message}"
"The scheduled time has arrived. Deliver this reminder to the user now, "
"as a brief and natural message in their language. Speak directly to them — "
"do not narrate progress, summarize, include user IDs, or add status reports "
"like 'Done' or 'Reminded'.\n\n"
f"Reminder: {job.payload.message}"
)
cron_tool = agent.tools.get("cron")
@@ -763,8 +768,10 @@ def _run_gateway(
channel=job.payload.channel or "cli",
chat_id=job.payload.to,
content=response,
metadata=dict(job.payload.channel_meta),
),
record=True,
session_key=job.payload.session_key,
)
return response
nanobot/command/builtin.py
+5 -1
View File
@@ -28,7 +28,11 @@ async def cmd_stop(ctx: CommandContext) -> OutboundMessage:
async def cmd_restart(ctx: CommandContext) -> OutboundMessage:
"""Restart the process in-place via os.execv."""
msg = ctx.msg
set_restart_notice_to_env(channel=msg.channel, chat_id=msg.chat_id)
set_restart_notice_to_env(
channel=msg.channel,
chat_id=msg.chat_id,
metadata=dict(msg.metadata or {}),
)
async def _do_restart():
await asyncio.sleep(1)
nanobot/cron/service.py
+12
View File
@@ -109,6 +109,12 @@ class CronService:
deliver=j["payload"].get("deliver", False),
channel=j["payload"].get("channel"),
to=j["payload"].get("to"),
channel_meta=(
j["payload"].get("channelMeta")
or j["payload"].get("channel_meta")
or {}
),
session_key=j["payload"].get("sessionKey") or j["payload"].get("session_key"),
),
state=CronJobState(
next_run_at_ms=j.get("state", {}).get("nextRunAtMs"),
@@ -210,6 +216,8 @@ class CronService:
"deliver": j.payload.deliver,
"channel": j.payload.channel,
"to": j.payload.to,
"channelMeta": j.payload.channel_meta,
"sessionKey": j.payload.session_key,
},
"state": {
"nextRunAtMs": j.state.next_run_at_ms,
@@ -379,6 +387,8 @@ class CronService:
channel: str | None = None,
to: str | None = None,
delete_after_run: bool = False,
channel_meta: dict | None = None,
session_key: str | None = None,
) -> CronJob:
"""Add a new job."""
_validate_schedule_for_add(schedule)
@@ -395,6 +405,8 @@ class CronService:
deliver=deliver,
channel=channel,
to=to,
channel_meta=channel_meta or {},
session_key=session_key,
),
state=CronJobState(next_run_at_ms=_compute_next_run(schedule, now)),
created_at_ms=now,
nanobot/cron/types.py
+2
View File
@@ -27,6 +27,8 @@ class CronPayload:
deliver: bool = False
channel: str | None = None # e.g. "whatsapp"
to: str | None = None # e.g. phone number
channel_meta: dict = field(default_factory=dict) # channel-specific routing (e.g. Slack thread_ts)
session_key: str | None = None # original session key for correct session recording
@dataclass
nanobot/session/manager.py
+29
View File
@@ -31,6 +31,32 @@ class Session:
metadata: dict[str, Any] = field(default_factory=dict)
last_consolidated: int = 0 # Number of messages already consolidated to files
@staticmethod
def _annotate_message_time(message: dict[str, Any], content: Any) -> Any:
"""Expose persisted turn timestamps to the model for relative-date reasoning.
Annotating *every* assistant turn trains the model (via in-context
demonstrations) to start its own replies with the same
``[Message Time: ...]`` prefix, which leaks metadata back to the user.
We therefore only annotate:
* ``user`` turns — needed so the model can pin the conversation in time.
* proactive deliveries (``_channel_delivery=True``) — cron / heartbeat
assistant pushes that may sit hours away from the next user reply,
and are too infrequent to act as parroting demonstrations.
"""
timestamp = message.get("timestamp")
if not timestamp or not isinstance(content, str):
return content
role = message.get("role")
if role == "user":
pass
elif role == "assistant" and message.get("_channel_delivery"):
pass
else:
return content
return f"[Message Time: {timestamp}]\n{content}"
def add_message(self, role: str, content: str, **kwargs: Any) -> None:
"""Add a message to the session."""
msg = {
@@ -47,6 +73,7 @@ class Session:
max_messages: int = 500,
*,
max_tokens: int = 0,
include_timestamps: bool = False,
) -> list[dict[str, Any]]:
"""Return unconsolidated messages for LLM input.
@@ -85,6 +112,8 @@ class Session:
image_placeholder_text(p) for p in media if isinstance(p, str) and p
)
content = f"{content}\n{breadcrumbs}" if content else breadcrumbs
if include_timestamps:
content = self._annotate_message_time(message, content)
entry: dict[str, Any] = {"role": message["role"], "content": content}
for key in ("tool_calls", "tool_call_id", "name", "reasoning_content"):
if key in message:
nanobot/utils/restart.py
+30 -3
View File
@@ -2,12 +2,15 @@
from __future__ import annotations
import json
import os
import time
from dataclasses import dataclass
from dataclasses import dataclass, field
from typing import Any
RESTART_NOTIFY_CHANNEL_ENV = "NANOBOT_RESTART_NOTIFY_CHANNEL"
RESTART_NOTIFY_CHAT_ID_ENV = "NANOBOT_RESTART_NOTIFY_CHAT_ID"
RESTART_NOTIFY_METADATA_ENV = "NANOBOT_RESTART_NOTIFY_METADATA"
RESTART_STARTED_AT_ENV = "NANOBOT_RESTART_STARTED_AT"
@@ -16,6 +19,7 @@ class RestartNotice:
channel: str
chat_id: str
started_at_raw: str
metadata: dict[str, Any] = field(default_factory=dict)
def format_restart_completed_message(started_at_raw: str) -> str:
@@ -30,11 +34,20 @@ def format_restart_completed_message(started_at_raw: str) -> str:
return f"Restart completed{elapsed_suffix}."
def set_restart_notice_to_env(*, channel: str, chat_id: str) -> None:
def set_restart_notice_to_env(
*, channel: str, chat_id: str, metadata: dict[str, Any] | None = None,
) -> None:
"""Write restart notice env values for the next process."""
os.environ[RESTART_NOTIFY_CHANNEL_ENV] = channel
os.environ[RESTART_NOTIFY_CHAT_ID_ENV] = chat_id
os.environ[RESTART_STARTED_AT_ENV] = str(time.time())
if metadata:
try:
os.environ[RESTART_NOTIFY_METADATA_ENV] = json.dumps(metadata, default=str)
except (TypeError, ValueError):
os.environ.pop(RESTART_NOTIFY_METADATA_ENV, None)
else:
os.environ.pop(RESTART_NOTIFY_METADATA_ENV, None)
def consume_restart_notice_from_env() -> RestartNotice | None:
@@ -42,9 +55,23 @@ def consume_restart_notice_from_env() -> RestartNotice | None:
channel = os.environ.pop(RESTART_NOTIFY_CHANNEL_ENV, "").strip()
chat_id = os.environ.pop(RESTART_NOTIFY_CHAT_ID_ENV, "").strip()
started_at_raw = os.environ.pop(RESTART_STARTED_AT_ENV, "").strip()
metadata_raw = os.environ.pop(RESTART_NOTIFY_METADATA_ENV, "").strip()
if not (channel and chat_id):
return None
return RestartNotice(channel=channel, chat_id=chat_id, started_at_raw=started_at_raw)
metadata: dict[str, Any] = {}
if metadata_raw:
try:
parsed = json.loads(metadata_raw)
except (TypeError, ValueError):
parsed = None
if isinstance(parsed, dict):
metadata = parsed
return RestartNotice(
channel=channel,
chat_id=chat_id,
started_at_raw=started_at_raw,
metadata=metadata,
)
def should_show_cli_restart_notice(notice: RestartNotice, session_id: str) -> bool: