diff --git a/nanobot/agent/loop.py b/nanobot/agent/loop.py index 7ac786d1..0a851b67 100644 --- a/nanobot/agent/loop.py +++ b/nanobot/agent/loop.py @@ -73,7 +73,7 @@ from nanobot.session.goal_state import ( sustained_goal_active, ) from nanobot.session.history_visibility import HIDDEN_HISTORY_META -from nanobot.session.keys import UNIFIED_SESSION_KEY +from nanobot.session.keys import UNIFIED_SESSION_KEY, remember_last_channel from nanobot.session.manager import ( Session, SessionManager, @@ -798,6 +798,27 @@ class AgentLoop: return UNIFIED_SESSION_KEY return msg.session_key + def _remember_unified_session_route( + self, + session: Session, + msg: InboundMessage, + *, + is_user_turn: bool, + ) -> None: + """Remember the latest user-facing route for unified-session delivery.""" + if ( + not self._unified_session + or session.key != UNIFIED_SESSION_KEY + or not is_user_turn + or msg.channel in {"cli", "system"} + or msg.sender_id == "subagent" + ): + return + _, automation_metadata = automation_history_overrides(msg.metadata) + if automation_metadata: + return + remember_last_channel(session.metadata, msg.channel, msg.chat_id) + @staticmethod def _replay_token_budget(runtime: LLMRuntime) -> int: """Derive a token budget for session history replay from the context window.""" @@ -1490,6 +1511,11 @@ class AgentLoop: # ensure it exists in case this handler is invoked independently. if ctx.session is None: ctx.session = self.sessions.get_or_create(ctx.session_key) + self._remember_unified_session_route( + ctx.session, + msg, + is_user_turn=ctx.kind is TurnKind.USER, + ) await ctx.delivery.started() if ctx.kind is TurnKind.USER: self.workspace_scopes.persist_message_scope(ctx.session, msg) diff --git a/nanobot/cli/commands.py b/nanobot/cli/commands.py index 71dd978d..a91f408d 100644 --- a/nanobot/cli/commands.py +++ b/nanobot/cli/commands.py @@ -77,6 +77,10 @@ from nanobot.cli.stream import StreamRenderer, ThinkingSpinner # noqa: E402 from nanobot.config.paths import get_workspace_path, is_default_workspace # noqa: E402 from nanobot.config.schema import Config # noqa: E402 from nanobot.security.network import is_loopback_host # noqa: E402 +from nanobot.session.keys import ( # noqa: E402 + UNIFIED_SESSION_KEY, + last_channel_from_metadata, +) from nanobot.utils.evaluator import evaluate_response, resolve_evaluator_prompt # noqa: E402 from nanobot.utils.helpers import ( # noqa: E402 sanitize_surrogates as _sanitize_surrogates, @@ -264,6 +268,7 @@ def _pick_heartbeat_target_from_sessions( enabled_channels: Iterable[str], sessions: Iterable[dict[str, Any]], archived_keys: Iterable[str], + unified_session_metadata: dict[str, Any] | None = None, ) -> tuple[str, str]: enabled = set(enabled_channels) archived = set(archived_keys) @@ -271,6 +276,13 @@ def _pick_heartbeat_target_from_sessions( key = item.get("key") or "" if key in archived: continue + if key == UNIFIED_SESSION_KEY: + route = last_channel_from_metadata(unified_session_metadata) + if route is not None: + channel, chat_id = route + if channel not in {"cli", "system"} and channel in enabled: + return channel, chat_id + continue if ":" not in key: continue channel, chat_id = key.split(":", 1) @@ -1984,10 +1996,16 @@ def _run_gateway( def _pick_heartbeat_target() -> tuple[str, str]: """Pick a routable channel/chat target for heartbeat-triggered messages.""" sidebar_state = read_webui_sidebar_state() + unified_metadata = None + if config.agents.defaults.unified_session: + record = session_manager.read_session_metadata(UNIFIED_SESSION_KEY) + if isinstance(record, dict) and isinstance(record.get("metadata"), dict): + unified_metadata = record["metadata"] return _pick_heartbeat_target_from_sessions( enabled_channels=channels.enabled_channels, sessions=session_manager.list_sessions(), archived_keys=sidebar_state.get("archived_keys", []), + unified_session_metadata=unified_metadata, ) if channels.enabled_channels: diff --git a/nanobot/session/keys.py b/nanobot/session/keys.py index ce581bdc..45f6d1cf 100644 --- a/nanobot/session/keys.py +++ b/nanobot/session/keys.py @@ -2,7 +2,11 @@ from __future__ import annotations +from collections.abc import Mapping, MutableMapping +from typing import Any + UNIFIED_SESSION_KEY = "unified:default" +LAST_CHANNEL_METADATA_KEY = "last_channel" def session_key_for_channel(channel: str, chat_id: str, *, unified_session: bool = False) -> str: @@ -10,3 +14,29 @@ def session_key_for_channel(channel: str, chat_id: str, *, unified_session: bool if unified_session: return UNIFIED_SESSION_KEY return f"{channel}:{chat_id}" + + +def remember_last_channel( + metadata: MutableMapping[str, Any], + channel: str, + chat_id: str, +) -> None: + """Persist the latest concrete delivery route in session metadata.""" + if not channel or not chat_id: + return + metadata[LAST_CHANNEL_METADATA_KEY] = f"{channel}:{chat_id}" + + +def last_channel_from_metadata( + metadata: Mapping[str, Any] | None, +) -> tuple[str, str] | None: + """Return a concrete delivery route from persisted session metadata.""" + if not isinstance(metadata, Mapping): + return None + route = metadata.get(LAST_CHANNEL_METADATA_KEY) + if not isinstance(route, str) or ":" not in route: + return None + channel, chat_id = route.split(":", 1) + if not channel or not chat_id: + return None + return channel, chat_id diff --git a/tests/agent/test_loop_save_turn.py b/tests/agent/test_loop_save_turn.py index a87600a5..ed53b2c2 100644 --- a/tests/agent/test_loop_save_turn.py +++ b/tests/agent/test_loop_save_turn.py @@ -30,6 +30,10 @@ from nanobot.runtime_context import ( ) from nanobot.session.automation_turns import AUTOMATION_HISTORY_META from nanobot.session.goal_state import GOAL_STATE_KEY +from nanobot.session.keys import ( + LAST_CHANNEL_METADATA_KEY, + UNIFIED_SESSION_KEY, +) from nanobot.session.manager import Session, SessionManager from nanobot.session.turn_continuation import ( INTERNAL_CONTINUATION_META, @@ -682,6 +686,28 @@ async def test_process_message_persists_user_message_before_turn_completes(tmp_p assert persisted.updated_at >= persisted.created_at +@pytest.mark.asyncio +async def test_process_message_persists_unified_session_delivery_route(tmp_path: Path) -> None: + loop = _make_full_loop(tmp_path) + loop._unified_session = True + loop.consolidator.maybe_consolidate_by_tokens = AsyncMock(return_value=False) # type: ignore[method-assign] + loop._run_agent_loop = AsyncMock(side_effect=RuntimeError("boom")) # type: ignore[method-assign] + + msg = InboundMessage( + channel="feishu", + sender_id="u1", + chat_id="oc_123", + content="persist my route", + session_key_override=UNIFIED_SESSION_KEY, + ) + with pytest.raises(RuntimeError, match="boom"): + await loop._process_message(msg) + + loop.sessions.invalidate(UNIFIED_SESSION_KEY) + persisted = loop.sessions.get_or_create(UNIFIED_SESSION_KEY) + assert persisted.metadata[LAST_CHANNEL_METADATA_KEY] == "feishu:oc_123" + + # 1x1 PNG used by the media-persistence tests. ``extract_documents`` runs # at the top of ``_process_message`` and filters ``msg.media`` down to # paths that magic-byte-sniff as images, so the test fixture needs real diff --git a/tests/cli/test_commands.py b/tests/cli/test_commands.py index 9f94dbba..34c9c674 100644 --- a/tests/cli/test_commands.py +++ b/tests/cli/test_commands.py @@ -1787,6 +1787,42 @@ def test_heartbeat_target_skips_archived_webui_sessions(): assert target == ("websocket", "active") +def test_heartbeat_target_uses_last_channel_for_unified_session(): + from nanobot.cli.commands import _pick_heartbeat_target_from_sessions + from nanobot.session.keys import LAST_CHANNEL_METADATA_KEY, UNIFIED_SESSION_KEY + + target = _pick_heartbeat_target_from_sessions( + enabled_channels=["telegram", "discord"], + archived_keys=[], + sessions=[{"key": UNIFIED_SESSION_KEY}], + unified_session_metadata={LAST_CHANNEL_METADATA_KEY: "discord:chat-42"}, + ) + + assert target == ("discord", "chat-42") + + +@pytest.mark.parametrize( + "metadata", + [ + {"last_channel": "telegram:chat-42"}, + {"last_channel": "cli:direct"}, + {"last_channel": "invalid"}, + ], +) +def test_heartbeat_target_rejects_unroutable_unified_metadata(metadata): + from nanobot.cli.commands import _pick_heartbeat_target_from_sessions + from nanobot.session.keys import UNIFIED_SESSION_KEY + + target = _pick_heartbeat_target_from_sessions( + enabled_channels=["discord"], + archived_keys=[], + sessions=[{"key": UNIFIED_SESSION_KEY}], + unified_session_metadata=metadata, + ) + + assert target == ("cli", "direct") + + def _write_instance_config(tmp_path: Path) -> Path: config_file = tmp_path / "instance" / "config.json" config_file.parent.mkdir(parents=True)