Merge remote-tracking branch 'origin/main' into nightly

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. (<www.mathjax.org>)
+Copyright (c) 2014-2018 Khan Academy (<www.khanacademy.org>),
+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.
+```

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:

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,

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:

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."""
+        """Handle incoming Discord messages from discord.py.
-        if message.author.bot:
+
+        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)

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:

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:
+        try:
-        """Save a session to disk."""
+            messages: list[dict[str, Any]] = []
-        path = self._get_session_path(session.key)
+            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:
+            with open(path, encoding="utf-8") as f:
-            metadata_line = {
+                for line in f:
-                "_type": "metadata",
+                    line = line.strip()
-                "key": session.key,
+                    if not line:
-                "created_at": session.created_at.isoformat(),
+                        continue
-                "updated_at": session.updated_at.isoformat(),
+                    try:
-                "metadata": session.metadata,
+                        data = json.loads(line)
-                "last_consolidated": session.last_consolidated
+                    except json.JSONDecodeError:
-            }
+                        skipped += 1
-            f.write(json.dumps(metadata_line, ensure_ascii=False) + "\n")
+                        continue
-            for msg in session.messages:
+
-                f.write(json.dumps(msg, ensure_ascii=False) + "\n")
+                    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)

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]

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"

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"

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:
+async def test_on_message_ignores_self_messages() -> None:
-    # Incoming bot-authored messages must be ignored to prevent feedback loops.
+    # 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")
 

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: