diff --git a/THIRD_PARTY_NOTICES.md b/THIRD_PARTY_NOTICES.md new file mode 100644 index 000000000..9085bfc8e --- /dev/null +++ b/THIRD_PARTY_NOTICES.md @@ -0,0 +1,144 @@ +# Third-Party Notices + +The following third-party components are redistributed as part of the packaged +nanobot Python distribution (`pip install nanobot-ai`). + +--- + +## KaTeX — math rendering (MIT) + +- **Source**: https://github.com/KaTeX/KaTeX +- **Bundled**: `nanobot/web/dist/assets/index-*.{js,css}` + +``` +The MIT License (MIT) + +Copyright (c) 2013-2020 Khan Academy and other contributors + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. +``` + +--- + +## KaTeX Fonts — math typography (SIL OFL 1.1) + +- **Source**: https://github.com/KaTeX/KaTeX/tree/main/src/fonts +- **Bundled**: `nanobot/web/dist/assets/KaTeX_*.{woff2,woff,ttf}` + +The fonts are redistributed unmodified. + +``` +Copyright (c) 2009-2010, Design Science, Inc. () +Copyright (c) 2014-2018 Khan Academy (), +with Reserved Font Names KaTeX_AMS, KaTeX_Caligraphic, KaTeX_Fraktur, +KaTeX_Main, KaTeX_Math, KaTeX_SansSerif, KaTeX_Script, KaTeX_Size1, +KaTeX_Size2, KaTeX_Size3, KaTeX_Size4, KaTeX_Typewriter. + +This Font Software is licensed under the SIL Open Font License, Version 1.1. +This license is copied below, and is also available with a FAQ at: +http://scripts.sil.org/OFL + + +----------------------------------------------------------- +SIL OPEN FONT LICENSE Version 1.1 - 26 February 2007 +----------------------------------------------------------- + +PREAMBLE +The goals of the Open Font License (OFL) are to stimulate worldwide +development of collaborative font projects, to support the font creation +efforts of academic and linguistic communities, and to provide a free and +open framework in which fonts may be shared and improved in partnership +with others. + +The OFL allows the licensed fonts to be used, studied, modified and +redistributed freely as long as they are not sold by themselves. The +fonts, including any derivative works, can be bundled, embedded, +redistributed and/or sold with any software provided that any reserved +names are not used by derivative works. The fonts and derivatives, +however, cannot be released under any other type of license. The +requirement for fonts to remain under this license does not apply +to any document created using the fonts or their derivatives. + +DEFINITIONS +"Font Software" refers to the set of files released by the Copyright +Holder(s) under this license and clearly marked as such. This may +include source files, build scripts and documentation. + +"Reserved Font Name" refers to any names specified as such after the +copyright statement(s). + +"Original Version" refers to the collection of Font Software components as +distributed by the Copyright Holder(s). + +"Modified Version" refers to any derivative made by adding to, deleting, +or substituting -- in part or in whole -- any of the components of the +Original Version, by changing formats or by porting the Font Software to a +new environment. + +"Author" refers to any designer, engineer, programmer, technical +writer or other person who contributed to the Font Software. + +PERMISSION & CONDITIONS +Permission is hereby granted, free of charge, to any person obtaining +a copy of the Font Software, to use, study, copy, merge, embed, modify, +redistribute, and sell modified and unmodified copies of the Font +Software, subject to the following conditions: + +1) Neither the Font Software nor any of its individual components, +in Original or Modified Versions, may be sold by itself. + +2) Original or Modified Versions of the Font Software may be bundled, +redistributed and/or sold with any software, provided that each copy +contains the above copyright notice and this license. These can be +included either as stand-alone text files, human-readable headers or +in the appropriate machine-readable metadata fields within text or +binary files as long as those fields can be easily viewed by the user. + +3) No Modified Version of the Font Software may use the Reserved Font +Name(s) unless explicit written permission is granted by the corresponding +Copyright Holder. This restriction only applies to the primary font name as +presented to the users. + +4) The name(s) of the Copyright Holder(s) or the Author(s) of the Font +Software shall not be used to promote, endorse or advertise any +Modified Version, except to acknowledge the contribution(s) of the +Copyright Holder(s) and the Author(s) or with their explicit written +permission. + +5) The Font Software, modified or unmodified, in part or in whole, +must be distributed entirely under this license, and must not be +distributed under any other license. The requirement for fonts to +remain under this license does not apply to any document created +using the Font Software. + +TERMINATION +This license becomes null and void if any of the above conditions are +not met. + +DISCLAIMER +THE FONT SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, +EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO ANY WARRANTIES OF +MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT +OF COPYRIGHT, PATENT, TRADEMARK, OR OTHER RIGHT. IN NO EVENT SHALL THE +COPYRIGHT HOLDER BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, +INCLUDING ANY GENERAL, SPECIAL, INDIRECT, INCIDENTAL, OR CONSEQUENTIAL +DAMAGES, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING +FROM, OUT OF THE USE OR INABILITY TO USE THE FONT SOFTWARE OR FROM +OTHER DEALINGS IN THE FONT SOFTWARE. +``` diff --git a/nanobot/agent/loop.py b/nanobot/agent/loop.py index 53cb49d71..a3b29fb93 100644 --- a/nanobot/agent/loop.py +++ b/nanobot/agent/loop.py @@ -318,10 +318,16 @@ class AgentLoop: def _set_tool_context(self, channel: str, chat_id: str, message_id: 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}" for name in ("message", "spawn", "cron", "my"): if tool := self.tools.get(name): if hasattr(tool, "set_context"): - tool.set_context(channel, chat_id, *([message_id] if name == "message" else [])) + if name == "spawn": + tool.set_context(channel, chat_id, effective_key=effective_key) + else: + tool.set_context(channel, chat_id, *([message_id] if name == "message" else [])) @staticmethod def _strip_think(text: str | None) -> str | None: diff --git a/nanobot/agent/subagent.py b/nanobot/agent/subagent.py index 388a634c7..7db62dcf4 100644 --- a/nanobot/agent/subagent.py +++ b/nanobot/agent/subagent.py @@ -107,7 +107,7 @@ class SubagentManager: """Spawn a subagent to execute a task in the background.""" task_id = str(uuid.uuid4())[:8] display_label = label or task[:30] + ("..." if len(task) > 30 else "") - origin = {"channel": origin_channel, "chat_id": origin_chat_id} + origin = {"channel": origin_channel, "chat_id": origin_chat_id, "session_key": session_key} status = SubagentStatus( task_id=task_id, @@ -240,12 +240,18 @@ class SubagentManager: result=result, ) - # Inject as system message to trigger main agent + # Inject as system message to trigger main agent. + # Use session_key_override to align with the main agent's effective + # session key (which accounts for unified sessions) so the result is + # routed to the correct pending queue (mid-turn injection) instead of + # being dispatched as a competing independent task. + override = origin.get("session_key") or f"{origin['channel']}:{origin['chat_id']}" msg = InboundMessage( channel="system", sender_id="subagent", chat_id=f"{origin['channel']}:{origin['chat_id']}", content=announce_content, + session_key_override=override, metadata={ "injected_event": "subagent_result", "subagent_task_id": task_id, diff --git a/nanobot/agent/tools/spawn.py b/nanobot/agent/tools/spawn.py index 86319e991..8ffb438bf 100644 --- a/nanobot/agent/tools/spawn.py +++ b/nanobot/agent/tools/spawn.py @@ -25,11 +25,11 @@ class SpawnTool(Tool): self._origin_chat_id = "direct" self._session_key = "cli:direct" - def set_context(self, channel: str, chat_id: str) -> None: + def set_context(self, channel: str, chat_id: str, effective_key: str | None = None) -> None: """Set the origin context for subagent announcements.""" self._origin_channel = channel self._origin_chat_id = chat_id - self._session_key = f"{channel}:{chat_id}" + self._session_key = effective_key or f"{channel}:{chat_id}" @property def name(self) -> str: diff --git a/nanobot/channels/discord.py b/nanobot/channels/discord.py index 60ca06982..9710c5efc 100644 --- a/nanobot/channels/discord.py +++ b/nanobot/channels/discord.py @@ -433,8 +433,15 @@ class DiscordChannel(BaseChannel): raise async def _handle_discord_message(self, message: discord.Message) -> None: - """Handle incoming Discord messages from discord.py.""" - if message.author.bot: + """Handle incoming Discord messages from discord.py. + + Self-loop guard: only drop messages from this bot's own account. Messages + from other bots are allowed through so multi-agent setups (one bot asking + another for help, a bot mentioning another by @name, etc.) can work. + Bot-from-bot loops are still prevented per-instance because each bot + still ignores its own outbound messages. (#3217) + """ + if self._bot_user_id is not None and str(message.author.id) == self._bot_user_id: return sender_id = str(message.author.id) diff --git a/nanobot/cli/commands.py b/nanobot/cli/commands.py index f3b72a72e..ec572ffe9 100644 --- a/nanobot/cli/commands.py +++ b/nanobot/cli/commands.py @@ -726,12 +726,17 @@ def _run_gateway( cron_token = None if isinstance(cron_tool, CronTool): cron_token = cron_tool.set_cron_context(True) + + async def _silent(*_args, **_kwargs): + pass + try: resp = await agent.process_direct( reminder_note, session_key=f"cron:{job.id}", channel=job.payload.channel or "cli", chat_id=job.payload.to or "direct", + on_progress=_silent, ) finally: if isinstance(cron_tool, CronTool) and cron_token is not None: diff --git a/nanobot/session/manager.py b/nanobot/session/manager.py index c91eabcb3..4add4fd3b 100644 --- a/nanobot/session/manager.py +++ b/nanobot/session/manager.py @@ -1,6 +1,7 @@ """Session management for conversation history.""" import json +import os import shutil from dataclasses import dataclass, field from datetime import datetime @@ -187,24 +188,103 @@ class SessionManager: ) except Exception as e: logger.warning("Failed to load session {}: {}", key, e) + repaired = self._repair(key) + if repaired is not None: + logger.info("Recovered session {} from corrupt file ({} messages)", key, len(repaired.messages)) + return repaired + + def _repair(self, key: str) -> Session | None: + """Attempt to recover a session from a corrupt JSONL file.""" + path = self._get_session_path(key) + if not path.exists(): return None - def save(self, session: Session) -> None: - """Save a session to disk.""" - path = self._get_session_path(session.key) + try: + messages: list[dict[str, Any]] = [] + metadata: dict[str, Any] = {} + created_at: datetime | None = None + updated_at: datetime | None = None + last_consolidated = 0 + skipped = 0 - with open(path, "w", encoding="utf-8") as f: - metadata_line = { - "_type": "metadata", - "key": session.key, - "created_at": session.created_at.isoformat(), - "updated_at": session.updated_at.isoformat(), - "metadata": session.metadata, - "last_consolidated": session.last_consolidated - } - f.write(json.dumps(metadata_line, ensure_ascii=False) + "\n") - for msg in session.messages: - f.write(json.dumps(msg, ensure_ascii=False) + "\n") + with open(path, encoding="utf-8") as f: + for line in f: + line = line.strip() + if not line: + continue + try: + data = json.loads(line) + except json.JSONDecodeError: + skipped += 1 + continue + + if data.get("_type") == "metadata": + metadata = data.get("metadata", {}) + if data.get("created_at"): + try: + created_at = datetime.fromisoformat(data["created_at"]) + except (ValueError, TypeError): + pass + if data.get("updated_at"): + try: + updated_at = datetime.fromisoformat(data["updated_at"]) + except (ValueError, TypeError): + pass + last_consolidated = data.get("last_consolidated", 0) + else: + messages.append(data) + + if skipped: + logger.warning("Skipped {} corrupt lines in session {}", skipped, key) + + if not messages and not metadata: + return None + + return Session( + key=key, + messages=messages, + created_at=created_at or datetime.now(), + updated_at=updated_at or datetime.now(), + metadata=metadata, + last_consolidated=last_consolidated + ) + except Exception as e: + logger.warning("Repair failed for session {}: {}", key, e) + return None + + @staticmethod + def _session_payload(session: Session) -> dict[str, Any]: + return { + "key": session.key, + "created_at": session.created_at.isoformat(), + "updated_at": session.updated_at.isoformat(), + "metadata": session.metadata, + "messages": session.messages, + } + + def save(self, session: Session) -> None: + """Save a session to disk atomically.""" + path = self._get_session_path(session.key) + tmp_path = path.with_suffix(".jsonl.tmp") + + try: + with open(tmp_path, "w", encoding="utf-8") as f: + metadata_line = { + "_type": "metadata", + "key": session.key, + "created_at": session.created_at.isoformat(), + "updated_at": session.updated_at.isoformat(), + "metadata": session.metadata, + "last_consolidated": session.last_consolidated + } + f.write(json.dumps(metadata_line, ensure_ascii=False) + "\n") + for msg in session.messages: + f.write(json.dumps(msg, ensure_ascii=False) + "\n") + + os.replace(tmp_path, path) + except BaseException: + tmp_path.unlink(missing_ok=True) + raise self._cache[session.key] = session @@ -265,6 +345,10 @@ class SessionManager: } except Exception as e: logger.warning("Failed to read session {}: {}", key, e) + repaired = self._repair(key) + if repaired is not None: + logger.info("Recovered read-only session view {} from corrupt file", key) + return self._session_payload(repaired) return None def list_sessions(self) -> list[dict[str, Any]]: @@ -277,6 +361,7 @@ class SessionManager: sessions = [] for path in self.sessions_dir.glob("*.jsonl"): + fallback_key = path.stem.replace("_", ":", 1) try: # Read just the metadata line with open(path, encoding="utf-8") as f: @@ -292,6 +377,14 @@ class SessionManager: "path": str(path) }) except Exception: + repaired = self._repair(fallback_key) + if repaired is not None: + sessions.append({ + "key": repaired.key, + "created_at": repaired.created_at.isoformat(), + "updated_at": repaired.updated_at.isoformat(), + "path": str(path) + }) continue return sorted(sessions, key=lambda x: x.get("updated_at", ""), reverse=True) diff --git a/pyproject.toml b/pyproject.toml index 5878570aa..3447ffc30 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -16,6 +16,10 @@ classifiers = [ "Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.12", ] +license-files = [ + "LICENSE", + "THIRD_PARTY_NOTICES.md", +] dependencies = [ "typer>=0.20.0,<1.0.0", @@ -113,6 +117,12 @@ include = [ "nanobot/templates/**/*.md", "nanobot/skills/**/*.md", "nanobot/skills/**/*.sh", +] +# Build-time generated assets that live under .gitignore'd paths but must ship +# in the wheel/sdist. `artifacts` bypasses the VCS filter (unlike `include`). +# The webui is compiled via `bun run build` into nanobot/web/dist/ right before +# `python -m build` runs. +artifacts = [ "nanobot/web/dist/**/*", ] @@ -131,6 +141,8 @@ include = [ "bridge/", "README.md", "LICENSE", + "THIRD_PARTY_NOTICES.md", + "pyproject.toml", ] [tool.ruff] diff --git a/tests/agent/test_session_atomic.py b/tests/agent/test_session_atomic.py new file mode 100644 index 000000000..4720c028a --- /dev/null +++ b/tests/agent/test_session_atomic.py @@ -0,0 +1,263 @@ +"""Tests for atomic session save and corrupt-file repair.""" + +import json +from datetime import datetime +from pathlib import Path + +from nanobot.session.manager import Session, SessionManager + + +class TestAtomicSave: + def test_save_creates_valid_jsonl(self, tmp_path: Path): + mgr = SessionManager(tmp_path) + session = Session(key="test:1") + session.add_message("user", "hello") + session.add_message("assistant", "hi") + + mgr.save(session) + + path = mgr._get_session_path("test:1") + lines = path.read_text(encoding="utf-8").strip().split("\n") + assert len(lines) == 3 + + meta = json.loads(lines[0]) + assert meta["_type"] == "metadata" + assert meta["key"] == "test:1" + + msg1 = json.loads(lines[1]) + assert msg1["role"] == "user" + assert msg1["content"] == "hello" + + def test_no_tmp_file_left_after_successful_save(self, tmp_path: Path): + mgr = SessionManager(tmp_path) + session = Session(key="test:clean") + mgr.save(session) + + tmp_files = list(mgr.sessions_dir.glob("*.tmp")) + assert tmp_files == [] + + def test_tmp_file_cleaned_up_on_write_failure(self, tmp_path: Path): + mgr = SessionManager(tmp_path) + session = Session(key="test:fail") + path = mgr._get_session_path("test:fail") + tmp_path_file = path.with_suffix(".jsonl.tmp") + + path.parent.mkdir(parents=True, exist_ok=True) + tmp_path_file.write_text("stale") + + class BadMessage: + def __init__(self, data): + self.data = data + + original_dumps = json.dumps + + def failing_dumps(obj, **kwargs): + if isinstance(obj, dict) and obj.get("role") == "assistant": + raise OSError("simulated disk full") + return original_dumps(obj, **kwargs) + + session = Session(key="test:fail") + session.messages = [ + {"role": "user", "content": "ok"}, + {"role": "assistant", "content": "will fail"}, + ] + + import unittest.mock + with unittest.mock.patch("nanobot.session.manager.json.dumps", side_effect=failing_dumps): + try: + mgr.save(session) + except OSError: + pass + + assert not tmp_path_file.exists() + + def test_overwrite_preserves_latest_data(self, tmp_path: Path): + mgr = SessionManager(tmp_path) + session = Session(key="test:overwrite") + + session.add_message("user", "first") + mgr.save(session) + + session.add_message("user", "second") + mgr.save(session) + + mgr.invalidate("test:overwrite") + loaded = mgr.get_or_create("test:overwrite") + assert len(loaded.messages) == 2 + assert loaded.messages[0]["content"] == "first" + assert loaded.messages[1]["content"] == "second" + + def test_consecutive_saves_are_consistent(self, tmp_path: Path): + mgr = SessionManager(tmp_path) + session = Session(key="test:consistency") + + for i in range(5): + session.add_message("user", f"msg{i}") + mgr.save(session) + + mgr.invalidate("test:consistency") + loaded = mgr.get_or_create("test:consistency") + assert len(loaded.messages) == 5 + for i in range(5): + assert loaded.messages[i]["content"] == f"msg{i}" + + +class TestRepairCorruptFile: + def _write_corrupt_jsonl(self, path: Path, lines: list[str]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text("\n".join(lines) + "\n", encoding="utf-8") + + def test_truncated_last_line_recovered(self, tmp_path: Path): + mgr = SessionManager(tmp_path) + path = mgr._get_session_path("test:trunc") + + valid_meta = json.dumps({ + "_type": "metadata", + "key": "test:trunc", + "created_at": datetime.now().isoformat(), + "updated_at": datetime.now().isoformat(), + "metadata": {}, + "last_consolidated": 0, + }) + valid_msg = json.dumps({"role": "user", "content": "hello"}) + + self._write_corrupt_jsonl(path, [ + valid_meta, + valid_msg, + '{"role": "assistant", "content": "partial...', + ]) + + session = mgr._load("test:trunc") + assert session is not None + assert len(session.messages) == 1 + assert session.messages[0]["content"] == "hello" + + def test_corrupt_metadata_line_skipped(self, tmp_path: Path): + mgr = SessionManager(tmp_path) + path = mgr._get_session_path("test:badmeta") + + self._write_corrupt_jsonl(path, [ + "NOT VALID JSON!!!", + '{"role": "user", "content": "survived"}', + ]) + + session = mgr._load("test:badmeta") + assert session is not None + assert len(session.messages) == 1 + assert session.messages[0]["content"] == "survived" + + def test_all_corrupt_lines_returns_none(self, tmp_path: Path): + mgr = SessionManager(tmp_path) + path = mgr._get_session_path("test:allbad") + + self._write_corrupt_jsonl(path, [ + "garbage line 1", + "garbage line 2", + "{{invalid json", + ]) + + session = mgr._load("test:allbad") + assert session is None + + def test_empty_file_returns_empty_session(self, tmp_path: Path): + mgr = SessionManager(tmp_path) + path = mgr._get_session_path("test:empty") + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text("", encoding="utf-8") + + session = mgr._load("test:empty") + assert session is not None + assert session.messages == [] + assert session.key == "test:empty" + + def test_repair_preserves_valid_messages_amid_corruption(self, tmp_path: Path): + mgr = SessionManager(tmp_path) + path = mgr._get_session_path("test:mixed") + + self._write_corrupt_jsonl(path, [ + json.dumps({"_type": "metadata", "key": "test:mixed", + "created_at": datetime.now().isoformat(), + "updated_at": datetime.now().isoformat(), + "metadata": {}, "last_consolidated": 0}), + "BROKEN", + json.dumps({"role": "user", "content": "msg1"}), + '{"role": "assistant", "content": "broken', + json.dumps({"role": "user", "content": "msg2"}), + ]) + + session = mgr._load("test:mixed") + assert session is not None + assert len(session.messages) == 2 + assert session.messages[0]["content"] == "msg1" + assert session.messages[1]["content"] == "msg2" + + def test_repair_with_bad_timestamp_uses_fallback(self, tmp_path: Path): + mgr = SessionManager(tmp_path) + path = mgr._get_session_path("test:badts") + + self._write_corrupt_jsonl(path, [ + json.dumps({"_type": "metadata", "key": "test:badts", + "created_at": "not-a-date", + "updated_at": "also-bad", + "metadata": {}, "last_consolidated": 5}), + json.dumps({"role": "user", "content": "hi"}), + ]) + + session = mgr._load("test:badts") + assert session is not None + assert session.last_consolidated == 5 + assert isinstance(session.created_at, datetime) + + def test_read_session_file_repairs_corrupt_jsonl(self, tmp_path: Path): + mgr = SessionManager(tmp_path) + path = mgr._get_session_path("test:read-repair") + + self._write_corrupt_jsonl(path, [ + json.dumps({ + "_type": "metadata", + "key": "test:read-repair", + "created_at": datetime.now().isoformat(), + "updated_at": datetime.now().isoformat(), + "metadata": {"source": "repair"}, + "last_consolidated": 0, + }), + json.dumps({"role": "user", "content": "survived"}), + '{"role": "assistant", "content": "partial...', + ]) + + payload = mgr.read_session_file("test:read-repair") + assert payload is not None + assert payload["key"] == "test:read-repair" + assert payload["metadata"] == {"source": "repair"} + assert payload["messages"] == [{"role": "user", "content": "survived"}] + + def test_list_sessions_keeps_repaired_corrupt_file(self, tmp_path: Path): + mgr = SessionManager(tmp_path) + path = mgr._get_session_path("test:list-repair") + + self._write_corrupt_jsonl(path, [ + "NOT VALID JSON", + json.dumps({ + "_type": "metadata", + "key": "test:list-repair", + "created_at": datetime.now().isoformat(), + "updated_at": datetime.now().isoformat(), + "metadata": {}, + "last_consolidated": 0, + }), + json.dumps({"role": "user", "content": "hello"}), + ]) + + sessions = mgr.list_sessions() + assert any(s["key"] == "test:list-repair" for s in sessions) + + def test_get_or_create_returns_new_session_for_corrupt_file(self, tmp_path: Path): + mgr = SessionManager(tmp_path) + path = mgr._get_session_path("test:fallback") + + self._write_corrupt_jsonl(path, ["{{{{"]) + + session = mgr.get_or_create("test:fallback") + assert session is not None + assert session.messages == [] + assert session.key == "test:fallback" diff --git a/tests/agent/test_task_cancel.py b/tests/agent/test_task_cancel.py index c1c36ca8a..7133554b4 100644 --- a/tests/agent/test_task_cancel.py +++ b/tests/agent/test_task_cancel.py @@ -412,3 +412,91 @@ class TestSubagentCancellation: assert cancelled.is_set() assert task.cancelled() mgr._announce_result.assert_not_awaited() + + +class TestSubagentAnnounceSessionKey: + """Verify _announce_result uses the effective session key for mid-turn routing.""" + + def _make_mgr(self): + """Create a SubagentManager with mocked deps and its bus.""" + from nanobot.agent.subagent import SubagentManager + from nanobot.bus.queue import MessageBus + + bus = MessageBus() + provider = MagicMock() + provider.get_default_model.return_value = "test-model" + mgr = SubagentManager( + provider=provider, + workspace=MagicMock(), + bus=bus, + max_tool_result_chars=_MAX_TOOL_RESULT_CHARS, + ) + return mgr, bus + + @pytest.mark.asyncio + async def test_announce_uses_effective_key_in_unified_mode(self): + """In unified session mode, session_key_override must be 'unified:default' + so the result matches the pending queue key.""" + mgr, bus = self._make_mgr() + + origin = {"channel": "telegram", "chat_id": "111", "session_key": "unified:default"} + await mgr._announce_result("sub-1", "label", "task", "result", origin, "ok") + + msg = await bus.consume_inbound() + assert msg.session_key_override == "unified:default" + assert msg.session_key == "unified:default" + + @pytest.mark.asyncio + async def test_announce_uses_raw_key_in_normal_mode(self): + """Without unified sessions, session_key_override is the raw channel:chat_id.""" + mgr, bus = self._make_mgr() + + origin = {"channel": "telegram", "chat_id": "222", "session_key": "telegram:222"} + await mgr._announce_result("sub-2", "label", "task", "result", origin, "ok") + + msg = await bus.consume_inbound() + assert msg.session_key_override == "telegram:222" + assert msg.session_key == "telegram:222" + + @pytest.mark.asyncio + async def test_announce_falls_back_to_origin_when_no_session_key(self): + """When session_key is None, fallback to f'{channel}:{chat_id}'.""" + mgr, bus = self._make_mgr() + + origin = {"channel": "discord", "chat_id": "333", "session_key": None} + await mgr._announce_result("sub-3", "label", "task", "result", origin, "ok") + + msg = await bus.consume_inbound() + assert msg.session_key_override == "discord:333" + assert msg.channel == "system" + assert msg.chat_id == "discord:333" + + @pytest.mark.asyncio + async def test_session_key_flows_through_run_subagent(self): + """Verify session_key in origin propagates from _run_subagent to _announce_result.""" + from nanobot.agent.subagent import SubagentStatus + + mgr, bus = self._make_mgr() + + async def fake_run(spec): + return SimpleNamespace( + stop_reason="done", + final_content="done", + error=None, + tool_events=[], + ) + + mgr.runner.run = AsyncMock(side_effect=fake_run) + + status = SubagentStatus( + task_id="sub-4", label="label", task_description="task", + started_at=time.monotonic(), + ) + await mgr._run_subagent( + "sub-4", "task", "label", + {"channel": "telegram", "chat_id": "444", "session_key": "unified:default"}, + status, + ) + + msg = await bus.consume_inbound() + assert msg.session_key_override == "unified:default" diff --git a/tests/channels/test_discord_channel.py b/tests/channels/test_discord_channel.py index 82ef6c51b..a0a032270 100644 --- a/tests/channels/test_discord_channel.py +++ b/tests/channels/test_discord_channel.py @@ -273,17 +273,41 @@ async def test_stop_is_safe_after_partial_start(monkeypatch) -> None: @pytest.mark.asyncio -async def test_on_message_ignores_bot_messages() -> None: - # Incoming bot-authored messages must be ignored to prevent feedback loops. +async def test_on_message_ignores_self_messages() -> None: + # Self-loop guard: messages from this bot's own account must be dropped (#3217). channel = DiscordChannel(DiscordConfig(enabled=True, allow_from=["*"]), MessageBus()) + channel._bot_user_id = "999" # simulate bot identity populated in on_ready() handled: list[dict] = [] channel._handle_message = lambda **kwargs: handled.append(kwargs) # type: ignore[method-assign] - await channel._on_message(_make_message(author_bot=True)) + await channel._on_message(_make_message(author_id=999, author_bot=True)) assert handled == [] + +@pytest.mark.asyncio +async def test_on_message_accepts_messages_from_other_bots() -> None: + # Multi-agent setups: messages from OTHER bots must be processed, not dropped (#3217). + channel = DiscordChannel(DiscordConfig(enabled=True, allow_from=["*"]), MessageBus()) + channel._bot_user_id = "999" + handled: list[dict] = [] + + async def capture_handle(**kwargs) -> None: + handled.append(kwargs) + + channel._handle_message = capture_handle # type: ignore[method-assign] + + await channel._on_message(_make_message(author_id=123, author_bot=True)) + + assert len(handled) == 1 + assert handled[0]["sender_id"] == "123" + + +@pytest.mark.asyncio +async def test_on_message_stops_typing_on_handle_exception() -> None: # If inbound handling raises, typing should be stopped for that channel. + channel = DiscordChannel(DiscordConfig(enabled=True, allow_from=["*"]), MessageBus()) + async def fail_handle(**kwargs) -> None: raise RuntimeError("boom") diff --git a/tests/cli/test_commands.py b/tests/cli/test_commands.py index b970d9201..58e6ab2c4 100644 --- a/tests/cli/test_commands.py +++ b/tests/cli/test_commands.py @@ -1032,6 +1032,97 @@ def test_gateway_cron_evaluator_receives_scheduled_reminder_context( ) +def test_gateway_cron_job_suppresses_intermediate_progress( + monkeypatch, tmp_path: Path +) -> None: + """Cron jobs must pass on_progress=_silent to process_direct so that + tool hints and streaming deltas are never leaked to the user channel + before evaluate_response decides whether to deliver.""" + config_file = tmp_path / "instance" / "config.json" + config_file.parent.mkdir(parents=True) + config_file.write_text("{}") + + config = Config() + config.agents.defaults.workspace = str(tmp_path / "config-workspace") + bus = MagicMock() + bus.publish_outbound = AsyncMock() + seen: dict[str, object] = {} + + monkeypatch.setattr("nanobot.config.loader.set_config_path", lambda _path: None) + monkeypatch.setattr("nanobot.config.loader.load_config", lambda _path=None: config) + monkeypatch.setattr("nanobot.cli.commands.sync_workspace_templates", lambda _path: None) + monkeypatch.setattr("nanobot.cli.commands._make_provider", lambda _config: object()) + monkeypatch.setattr("nanobot.bus.queue.MessageBus", lambda: bus) + monkeypatch.setattr("nanobot.session.manager.SessionManager", lambda _workspace: object()) + + class _FakeCron: + def __init__(self, _store_path: Path) -> None: + self.on_job = None + seen["cron"] = self + + class _FakeAgentLoop: + def __init__(self, *args, **kwargs) -> None: + self.model = "test-model" + self.tools = {} + + async def process_direct(self, *_args, on_progress=None, **_kwargs): + seen["on_progress"] = on_progress + return OutboundMessage( + channel="telegram", + chat_id="user-1", + content="Done.", + ) + + async def close_mcp(self) -> None: + return None + + async def run(self) -> None: + return None + + def stop(self) -> None: + return None + + class _StopAfterCronSetup: + def __init__(self, *_args, **_kwargs) -> None: + raise _StopGatewayError("stop") + + async def _always_reject(*_args, **_kwargs) -> bool: + return False + + monkeypatch.setattr("nanobot.cron.service.CronService", _FakeCron) + monkeypatch.setattr("nanobot.agent.loop.AgentLoop", _FakeAgentLoop) + monkeypatch.setattr("nanobot.channels.manager.ChannelManager", _StopAfterCronSetup) + monkeypatch.setattr( + "nanobot.utils.evaluator.evaluate_response", + _always_reject, + ) + + result = runner.invoke(app, ["gateway", "--config", str(config_file)]) + assert isinstance(result.exception, _StopGatewayError) + + cron = seen["cron"] + job = CronJob( + id="cron-silent-test", + name="test-silent", + payload=CronPayload( + message="Run something.", + deliver=True, + channel="telegram", + to="user-1", + ), + ) + response = asyncio.run(cron.on_job(job)) + + assert response == "Done." + # on_progress must be a callable (the _silent noop), not None and not bus_progress + assert seen["on_progress"] is not None + assert callable(seen["on_progress"]) + # Verify it actually swallows calls (no side effects) + asyncio.run(seen["on_progress"]("tool_hint", "🔧 $ echo test")) + # Nothing published to bus since evaluator rejected + bus.publish_outbound.assert_not_awaited() + + def test_gateway_workspace_override_does_not_migrate_legacy_cron( monkeypatch, tmp_path: Path ) -> None: