fix(agent): improve cross-channel session persist robustness

When session A (e.g. websocket) uses the `message` tool to send to
channel B (e.g. feishu), the outbound message is delivered to the user
but was never recorded in session B's history. This caused session B to
lose context when the user replied on that channel.

Add `_persist_cross_channel_calls()` to detect cross-channel `message`
tool calls during `_save_turn()` and append a lightweight assistant
entry (with `_cross_channel: True` marker) to the target session.

nanobot/agent/loop.py

@@ -843,8 +843,81 @@ class AgentLoop:
                     entry["content"] = filtered
             entry.setdefault("timestamp", datetime.now().isoformat())
             session.messages.append(entry)
+
+        # Persist cross-channel message tool calls into target sessions so
+        # that the target session has context when the user replies there.
+        self._persist_cross_channel_calls(session, messages[skip:])
         session.updated_at = datetime.now()
 
+    def _persist_cross_channel_calls(
+        self, source_session: Session, new_messages: list[dict[str, Any]]
+    ) -> None:
+        """Record cross-channel ``message`` tool calls into the target session.
+
+        When session A (e.g. websocket) uses the *message* tool to send to
+        channel B (e.g. feishu), the outbound message is delivered to the user
+        but is not recorded in session B's history.  This causes session B to
+        lose context when the user replies on channel B.
+
+        This method detects such cross-channel sends and appends a lightweight
+        assistant entry to the target session so it has the necessary context.
+
+        Improvements over the initial implementation:
+        - Use ``sessions.get_or_create()`` instead of accessing ``_cache``
+          directly, so sessions persisted on disk but evicted from memory are
+          still found.
+        - Persist ``media`` file paths alongside ``content`` so the target
+          session retains full context about attachments.
+        - Record ``_source_session`` to make the provenance traceable.
+        """
+        from datetime import datetime
+
+        for m in new_messages:
+            if m.get("role") != "assistant":
+                continue
+            tool_calls = m.get("tool_calls") or []
+            for tc in tool_calls:
+                func = tc.get("function", {})
+                if func.get("name") != "message":
+                    continue
+                try:
+                    args = json.loads(func.get("arguments", "{}"))
+                except (json.JSONDecodeError, TypeError):
+                    continue
+
+                target_channel = args.get("channel") or source_session.key.split(":", 1)[0]
+                target_chat_id = args.get("chat_id") or source_session.key.split(":", 1)[-1]
+                target_key = f"{target_channel}:{target_chat_id}"
+
+                if target_key == source_session.key:
+                    continue  # same session, nothing to do
+
+                content = args.get("content", "")
+                media = args.get("media")
+                if not content and not media:
+                    continue
+
+                # Use the public API so disk-persisted sessions are loaded too.
+                target_session = self.sessions.get_or_create(target_key)
+
+                entry: dict[str, Any] = {
+                    "role": "assistant",
+                    "content": content,
+                    "timestamp": datetime.now().isoformat(),
+                    "_cross_channel": True,
+                    "_source_session": source_session.key,
+                }
+                if media:
+                    entry["_media"] = media
+
+                target_session.messages.append(entry)
+                target_session.updated_at = datetime.now()
+                self.sessions.save(target_session)
+                logger.info(
+                    "Cross-channel message persisted: {} -> {}",
+                    source_session.key, target_key,
+                )
+
     def _set_runtime_checkpoint(self, session: Session, payload: dict[str, Any]) -> None:
         """Persist the latest in-flight turn state into session metadata."""
         session.metadata[self._RUNTIME_CHECKPOINT_KEY] = payload

nanobot/session/manager.py

@@ -57,6 +57,12 @@ class Session:
             for key in ("tool_calls", "tool_call_id", "name", "reasoning_content"):
                 if key in message:
                     entry[key] = message[key]
+            # Annotate cross-channel messages so the LLM knows the provenance,
+            # but keep the entry clean of internal metadata keys.
+            if message.get("_cross_channel"):
+                source = message.get("_source_session", "unknown")
+                prefix = f"[Sent from {source}] "
+                entry["content"] = prefix + (entry.get("content") or "")
             out.append(entry)
         return out
 

tests/agent/test_loop_save_turn.py

@@ -1,3 +1,4 @@
+import json
 from pathlib import Path
 from unittest.mock import AsyncMock, MagicMock
 
@@ -308,3 +309,297 @@ async def test_next_turn_after_crash_closes_pending_user_turn_before_new_input(t
         {"role": "assistant", "content": "new answer"},
     ]
     assert AgentLoop._PENDING_USER_TURN_KEY not in session.metadata
+
+
+def _cross_channel_messages(
+    source_channel: str = "websocket",
+    source_chat_id: str = "ws-uuid-123",
+    target_channel: str = "feishu",
+    target_chat_id: str = "ou_abc123",
+) -> list[dict]:
+    """Build a message list with a cross-channel message tool call."""
+    return [
+        {"role": "user", "content": "send report to feishu"},
+        {
+            "role": "assistant",
+            "content": "",
+            "tool_calls": [
+                {
+                    "id": "call_x1",
+                    "type": "function",
+                    "function": {
+                        "name": "message",
+                        "arguments": json.dumps({
+                            "content": "Report: audit complete",
+                            "channel": target_channel,
+                            "chat_id": target_chat_id,
+                        }),
+                    },
+                }
+            ],
+        },
+        {
+            "role": "tool",
+            "tool_call_id": "call_x1",
+            "name": "message",
+            "content": f"Message sent to {target_channel}:{target_chat_id}",
+        },
+        {"role": "assistant", "content": "Done, sent to feishu."},
+    ]
+
+
+def test_cross_channel_message_persisted_in_target_session(tmp_path: Path) -> None:
+    loop = _make_full_loop(tmp_path)
+    source_key = "websocket:ws-uuid-123"
+    target_key = "feishu:ou_abc123"
+
+    # Pre-create the target session (simulate an existing feishu conversation)
+    target_session = loop.sessions.get_or_create(target_key)
+    target_session.add_message("user", "hello from feishu")
+    loop.sessions.save(target_session)
+
+    source_session = loop.sessions.get_or_create(source_key)
+    msgs = _cross_channel_messages()
+    loop._save_turn(source_session, msgs, skip=1)  # skip user message
+
+    # Source session has its own messages
+    source_session = loop.sessions.get_or_create(source_key)
+    assert len(source_session.messages) >= 2  # assistant + tool + final
+
+    # Target session now has the cross-channel message appended
+    loop.sessions.invalidate(target_key)
+    target = loop.sessions.get_or_create(target_key)
+    cross_msg = [m for m in target.messages if m.get("_cross_channel")]
+    assert len(cross_msg) == 1
+    assert cross_msg[0]["content"] == "Report: audit complete"
+    assert cross_msg[0]["role"] == "assistant"
+
+
+def test_cross_channel_same_session_not_duplicated(tmp_path: Path) -> None:
+    loop = _make_full_loop(tmp_path)
+    key = "feishu:ou_same"
+
+    session = loop.sessions.get_or_create(key)
+    # message tool call targeting the same session — should NOT create a duplicate
+    msgs = [
+        {"role": "user", "content": "hello"},
+        {
+            "role": "assistant",
+            "content": "",
+            "tool_calls": [
+                {
+                    "id": "call_s1",
+                    "type": "function",
+                    "function": {
+                        "name": "message",
+                        "arguments": json.dumps({
+                            "content": "same channel msg",
+                            "channel": "feishu",
+                            "chat_id": "ou_same",
+                        }),
+                    },
+                }
+            ],
+        },
+        {"role": "tool", "tool_call_id": "call_s1", "name": "message", "content": "ok"},
+    ]
+    loop._save_turn(session, msgs, skip=1)
+
+    # No _cross_channel entries should exist
+    assert all(not m.get("_cross_channel") for m in session.messages)
+
+
+def test_cross_channel_target_session_not_exist_creates_session(tmp_path: Path) -> None:
+    """When the target session does not exist yet, get_or_create will create it
+    and the cross-channel message should still be persisted."""
+    loop = _make_full_loop(tmp_path)
+    source_session = loop.sessions.get_or_create("websocket:ws-xyz")
+
+    msgs = _cross_channel_messages(
+        target_channel="feishu", target_chat_id="ou_nonexistent"
+    )
+    loop._save_turn(source_session, msgs, skip=1)
+
+    # Target session is now auto-created with the cross-channel message
+    target = loop.sessions.get_or_create("feishu:ou_nonexistent")
+    cross_msgs = [m for m in target.messages if m.get("_cross_channel")]
+    assert len(cross_msgs) == 1
+    assert cross_msgs[0]["content"] == "Report: audit complete"
+
+
+def test_cross_channel_persists_media_attachments(tmp_path: Path) -> None:
+    """When the message tool call includes media, the cross-channel entry
+    should preserve the media paths so the target session has full context."""
+    loop = _make_full_loop(tmp_path)
+    source_key = "websocket:ws-media"
+    target_key = "telegram:tg_user1"
+
+    target_session = loop.sessions.get_or_create(target_key)
+    target_session.add_message("user", "waiting for report")
+    loop.sessions.save(target_session)
+
+    source_session = loop.sessions.get_or_create(source_key)
+    msgs = [
+        {"role": "user", "content": "send chart to telegram"},
+        {
+            "role": "assistant",
+            "content": "",
+            "tool_calls": [
+                {
+                    "id": "call_m1",
+                    "type": "function",
+                    "function": {
+                        "name": "message",
+                        "arguments": json.dumps({
+                            "content": "Here is the chart",
+                            "channel": "telegram",
+                            "chat_id": "tg_user1",
+                            "media": ["/tmp/chart.png", "/tmp/data.csv"],
+                        }),
+                    },
+                }
+            ],
+        },
+        {
+            "role": "tool",
+            "tool_call_id": "call_m1",
+            "name": "message",
+            "content": "Message sent to telegram:tg_user1",
+        },
+    ]
+    loop._save_turn(source_session, msgs, skip=1)
+
+    loop.sessions.invalidate(target_key)
+    target = loop.sessions.get_or_create(target_key)
+    cross_msgs = [m for m in target.messages if m.get("_cross_channel")]
+    assert len(cross_msgs) == 1
+    assert cross_msgs[0]["content"] == "Here is the chart"
+    assert cross_msgs[0]["_media"] == ["/tmp/chart.png", "/tmp/data.csv"]
+
+
+def test_cross_channel_records_source_session(tmp_path: Path) -> None:
+    """Cross-channel entries should include _source_session for traceability."""
+    loop = _make_full_loop(tmp_path)
+    source_key = "cron:heartbeat"
+    target_key = "feishu:ou_trace"
+
+    target_session = loop.sessions.get_or_create(target_key)
+    target_session.add_message("user", "hi")
+    loop.sessions.save(target_session)
+
+    source_session = loop.sessions.get_or_create(source_key)
+    msgs = _cross_channel_messages(
+        source_channel="cron", source_chat_id="heartbeat",
+        target_channel="feishu", target_chat_id="ou_trace",
+    )
+    loop._save_turn(source_session, msgs, skip=1)
+
+    loop.sessions.invalidate(target_key)
+    target = loop.sessions.get_or_create(target_key)
+    cross_msgs = [m for m in target.messages if m.get("_cross_channel")]
+    assert len(cross_msgs) == 1
+    assert cross_msgs[0]["_source_session"] == "cron:heartbeat"
+
+
+def test_cross_channel_media_only_no_content(tmp_path: Path) -> None:
+    """A message with media but empty content should still be persisted."""
+    loop = _make_full_loop(tmp_path)
+    target_key = "discord:ch_img"
+
+    target_session = loop.sessions.get_or_create(target_key)
+    loop.sessions.save(target_session)
+
+    source_session = loop.sessions.get_or_create("websocket:ws-img")
+    msgs = [
+        {"role": "user", "content": "send image"},
+        {
+            "role": "assistant",
+            "content": "",
+            "tool_calls": [
+                {
+                    "id": "call_img",
+                    "type": "function",
+                    "function": {
+                        "name": "message",
+                        "arguments": json.dumps({
+                            "content": "",
+                            "channel": "discord",
+                            "chat_id": "ch_img",
+                            "media": ["/tmp/photo.jpg"],
+                        }),
+                    },
+                }
+            ],
+        },
+        {"role": "tool", "tool_call_id": "call_img", "name": "message", "content": "ok"},
+    ]
+    loop._save_turn(source_session, msgs, skip=1)
+
+    loop.sessions.invalidate(target_key)
+    target = loop.sessions.get_or_create(target_key)
+    cross_msgs = [m for m in target.messages if m.get("_cross_channel")]
+    assert len(cross_msgs) == 1
+    assert cross_msgs[0]["_media"] == ["/tmp/photo.jpg"]
+
+
+def test_cross_channel_non_message_tools_ignored(tmp_path: Path) -> None:
+    loop = _make_full_loop(tmp_path)
+    source_session = loop.sessions.get_or_create("websocket:ws-abc")
+    target_session = loop.sessions.get_or_create("feishu:ou_tgt")
+    target_session.add_message("user", "hi")
+    loop.sessions.save(target_session)
+
+    msgs = [
+        {"role": "user", "content": "do stuff"},
+        {
+            "role": "assistant",
+            "content": "",
+            "tool_calls": [
+                {
+                    "id": "call_e1",
+                    "type": "function",
+                    "function": {
+                        "name": "exec",
+                        "arguments": json.dumps({"command": "echo hi"}),
+                    },
+                }
+            ],
+        },
+        {"role": "tool", "tool_call_id": "call_e1", "name": "exec", "content": "hi"},
+    ]
+    loop._save_turn(source_session, msgs, skip=1)
+
+    # exec tool should NOT produce cross-channel entries
+    target = loop.sessions.get_or_create("feishu:ou_tgt")
+    cross_msgs = [m for m in target.messages if m.get("_cross_channel")]
+    assert len(cross_msgs) == 0
+
+
+def test_cross_channel_get_history_annotates_provenance(tmp_path: Path) -> None:
+    """get_history() should prefix cross-channel messages with source info
+    so the LLM knows where the message came from."""
+    loop = _make_full_loop(tmp_path)
+    target_key = "feishu:ou_hist"
+
+    target_session = loop.sessions.get_or_create(target_key)
+    target_session.add_message("user", "hello")
+    # Simulate a cross-channel entry as _persist_cross_channel_calls would create
+    target_session.messages.append({
+        "role": "assistant",
+        "content": "Daily report ready",
+        "_cross_channel": True,
+        "_source_session": "cron:daily-report",
+    })
+    loop.sessions.save(target_session)
+
+    loop.sessions.invalidate(target_key)
+    target = loop.sessions.get_or_create(target_key)
+    history = target.get_history()
+
+    # Find the annotated message
+    annotated = [m for m in history if "cron:daily-report" in m.get("content", "")]
+    assert len(annotated) == 1
+    assert annotated[0]["content"] == "[Sent from cron:daily-report] Daily report ready"
+    # Internal metadata keys should NOT leak into the history output
+    assert "_cross_channel" not in annotated[0]
+    assert "_source_session" not in annotated[0]