fix(heartbeat): route unified sessions to last channel
This commit is contained in:
+27
-1
@@ -73,7 +73,7 @@ from nanobot.session.goal_state import (
|
|||||||
sustained_goal_active,
|
sustained_goal_active,
|
||||||
)
|
)
|
||||||
from nanobot.session.history_visibility import HIDDEN_HISTORY_META
|
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 (
|
from nanobot.session.manager import (
|
||||||
Session,
|
Session,
|
||||||
SessionManager,
|
SessionManager,
|
||||||
@@ -798,6 +798,27 @@ class AgentLoop:
|
|||||||
return UNIFIED_SESSION_KEY
|
return UNIFIED_SESSION_KEY
|
||||||
return msg.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
|
@staticmethod
|
||||||
def _replay_token_budget(runtime: LLMRuntime) -> int:
|
def _replay_token_budget(runtime: LLMRuntime) -> int:
|
||||||
"""Derive a token budget for session history replay from the context window."""
|
"""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.
|
# ensure it exists in case this handler is invoked independently.
|
||||||
if ctx.session is None:
|
if ctx.session is None:
|
||||||
ctx.session = self.sessions.get_or_create(ctx.session_key)
|
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()
|
await ctx.delivery.started()
|
||||||
if ctx.kind is TurnKind.USER:
|
if ctx.kind is TurnKind.USER:
|
||||||
self.workspace_scopes.persist_message_scope(ctx.session, msg)
|
self.workspace_scopes.persist_message_scope(ctx.session, msg)
|
||||||
|
|||||||
@@ -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.paths import get_workspace_path, is_default_workspace # noqa: E402
|
||||||
from nanobot.config.schema import Config # noqa: E402
|
from nanobot.config.schema import Config # noqa: E402
|
||||||
from nanobot.security.network import is_loopback_host # 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.evaluator import evaluate_response, resolve_evaluator_prompt # noqa: E402
|
||||||
from nanobot.utils.helpers import ( # noqa: E402
|
from nanobot.utils.helpers import ( # noqa: E402
|
||||||
sanitize_surrogates as _sanitize_surrogates,
|
sanitize_surrogates as _sanitize_surrogates,
|
||||||
@@ -264,6 +268,7 @@ def _pick_heartbeat_target_from_sessions(
|
|||||||
enabled_channels: Iterable[str],
|
enabled_channels: Iterable[str],
|
||||||
sessions: Iterable[dict[str, Any]],
|
sessions: Iterable[dict[str, Any]],
|
||||||
archived_keys: Iterable[str],
|
archived_keys: Iterable[str],
|
||||||
|
unified_session_metadata: dict[str, Any] | None = None,
|
||||||
) -> tuple[str, str]:
|
) -> tuple[str, str]:
|
||||||
enabled = set(enabled_channels)
|
enabled = set(enabled_channels)
|
||||||
archived = set(archived_keys)
|
archived = set(archived_keys)
|
||||||
@@ -271,6 +276,13 @@ def _pick_heartbeat_target_from_sessions(
|
|||||||
key = item.get("key") or ""
|
key = item.get("key") or ""
|
||||||
if key in archived:
|
if key in archived:
|
||||||
continue
|
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:
|
if ":" not in key:
|
||||||
continue
|
continue
|
||||||
channel, chat_id = key.split(":", 1)
|
channel, chat_id = key.split(":", 1)
|
||||||
@@ -1984,10 +1996,16 @@ def _run_gateway(
|
|||||||
def _pick_heartbeat_target() -> tuple[str, str]:
|
def _pick_heartbeat_target() -> tuple[str, str]:
|
||||||
"""Pick a routable channel/chat target for heartbeat-triggered messages."""
|
"""Pick a routable channel/chat target for heartbeat-triggered messages."""
|
||||||
sidebar_state = read_webui_sidebar_state()
|
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(
|
return _pick_heartbeat_target_from_sessions(
|
||||||
enabled_channels=channels.enabled_channels,
|
enabled_channels=channels.enabled_channels,
|
||||||
sessions=session_manager.list_sessions(),
|
sessions=session_manager.list_sessions(),
|
||||||
archived_keys=sidebar_state.get("archived_keys", []),
|
archived_keys=sidebar_state.get("archived_keys", []),
|
||||||
|
unified_session_metadata=unified_metadata,
|
||||||
)
|
)
|
||||||
|
|
||||||
if channels.enabled_channels:
|
if channels.enabled_channels:
|
||||||
|
|||||||
@@ -2,7 +2,11 @@
|
|||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from collections.abc import Mapping, MutableMapping
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
UNIFIED_SESSION_KEY = "unified:default"
|
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:
|
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:
|
if unified_session:
|
||||||
return UNIFIED_SESSION_KEY
|
return UNIFIED_SESSION_KEY
|
||||||
return f"{channel}:{chat_id}"
|
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
|
||||||
|
|||||||
@@ -30,6 +30,10 @@ from nanobot.runtime_context import (
|
|||||||
)
|
)
|
||||||
from nanobot.session.automation_turns import AUTOMATION_HISTORY_META
|
from nanobot.session.automation_turns import AUTOMATION_HISTORY_META
|
||||||
from nanobot.session.goal_state import GOAL_STATE_KEY
|
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.manager import Session, SessionManager
|
||||||
from nanobot.session.turn_continuation import (
|
from nanobot.session.turn_continuation import (
|
||||||
INTERNAL_CONTINUATION_META,
|
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
|
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
|
# 1x1 PNG used by the media-persistence tests. ``extract_documents`` runs
|
||||||
# at the top of ``_process_message`` and filters ``msg.media`` down to
|
# 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
|
# paths that magic-byte-sniff as images, so the test fixture needs real
|
||||||
|
|||||||
@@ -1787,6 +1787,42 @@ def test_heartbeat_target_skips_archived_webui_sessions():
|
|||||||
assert target == ("websocket", "active")
|
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:
|
def _write_instance_config(tmp_path: Path) -> Path:
|
||||||
config_file = tmp_path / "instance" / "config.json"
|
config_file = tmp_path / "instance" / "config.json"
|
||||||
config_file.parent.mkdir(parents=True)
|
config_file.parent.mkdir(parents=True)
|
||||||
|
|||||||
Reference in New Issue
Block a user