Grayscale/nanobot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(agent): add persistent runtime context providers

Browse Source
This commit is contained in:
chengyongru
2026-07-12 00:35:17 +08:00
committed by Xubin Ren
parent 7f8c3453e1
commit f75d3519db
27 changed files with 789 additions and 371 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/agent/test_context_builder.py
+19 -128
View File
@@ -5,8 +5,7 @@ from pathlib import Path
import pytest
from nanobot.agent.context import ContextBuilder
from nanobot.bus.events import InboundMessage
from nanobot.session.goal_state import GOAL_STATE_KEY
from nanobot.runtime_context import RuntimeContextBlock
# ---------------------------------------------------------------------------
# Helpers
@@ -16,42 +15,6 @@ def _builder(tmp_path: Path, **kw) -> ContextBuilder:
return ContextBuilder(workspace=tmp_path, **kw)
# ---------------------------------------------------------------------------
# _build_runtime_context (static)
# ---------------------------------------------------------------------------
class TestBuildRuntimeContext:
def test_time_only(self):
ctx = ContextBuilder._build_runtime_context(None, None)
assert "[Runtime Context" in ctx
assert "[/Runtime Context]" in ctx
assert "Current Time:" in ctx
assert "Channel:" not in ctx
def test_with_channel_and_chat_id(self):
ctx = ContextBuilder._build_runtime_context("telegram", "chat123")
assert "Channel: telegram" in ctx
assert "Chat ID: chat123" in ctx
def test_with_sender_id(self):
ctx = ContextBuilder._build_runtime_context("cli", "direct", sender_id="user1")
assert "Sender ID: user1" in ctx
def test_with_timezone(self):
ctx = ContextBuilder._build_runtime_context(None, None, timezone="Asia/Shanghai")
assert "Current Time:" in ctx
def test_no_channel_no_chat_id_omits_both(self):
ctx = ContextBuilder._build_runtime_context(None, None)
assert "Channel:" not in ctx
assert "Chat ID:" not in ctx
def test_no_sender_id_omits(self):
ctx = ContextBuilder._build_runtime_context("cli", "direct")
assert "Sender ID:" not in ctx
# ---------------------------------------------------------------------------
# _merge_message_content (static)
# ---------------------------------------------------------------------------
@@ -315,118 +278,46 @@ class TestBuildMessages:
assert messages[1]["role"] == "user"
assert "hello" in str(messages[1]["content"])
def test_runtime_context_injected(self, tmp_path):
def test_runtime_context_is_not_injected_by_default(self, tmp_path):
builder = _builder(tmp_path)
messages = builder.build_messages([], "hello", channel="cli", chat_id="direct")
user_msg = str(messages[-1]["content"])
assert "[Runtime Context" in user_msg
assert "hello" in user_msg
assert user_msg == "hello"
assert "Runtime Context" not in user_msg
assert "Current Time:" not in user_msg
assert "Chat ID:" not in user_msg
def test_session_metadata_injects_active_goal_state(self, tmp_path):
def test_session_metadata_does_not_inject_context_without_provider(self, tmp_path):
builder = _builder(tmp_path)
meta = {
GOAL_STATE_KEY: {"status": "active", "objective": "Finish docs migration."},
}
messages = builder.build_messages(
[],
"hi",
channel="cli",
chat_id="x",
session_metadata=meta,
session_metadata={"goal_state": {"status": "active", "objective": "hidden"}},
)
user_msg = str(messages[-1]["content"])
assert ContextBuilder._GOAL_RUNTIME_GUIDANCE_TAG in user_msg
assert "Execute sustained work" in user_msg
assert "Start or replace the sustained goal" not in user_msg
assert "Goal (active):" in user_msg
assert "Finish docs migration." in user_msg
assert messages[-1]["content"] == "hi"
def test_goal_start_turn_injects_objective_guidance_after_user_text(self, tmp_path):
builder = _builder(tmp_path)
normal_messages = builder.build_messages([], "hi", channel="cli", chat_id="direct")
messages = builder.build_messages(
[],
"/goal audit the repo",
channel="cli",
chat_id="direct",
goal_start_requested=True,
)
stale_messages = builder.build_messages(
[],
"/goal stale request",
channel="cli",
chat_id="direct",
inbound_message=InboundMessage(
channel="cli",
sender_id="system",
chat_id="direct",
content="/goal stale request",
metadata={"original_command": "/goal", "goal_requested": True},
),
)
user_msg = str(messages[-1]["content"])
assert "Write a durable objective" in user_msg
assert "complete `/goal <task>` command" in user_msg
guidance = user_msg[
user_msg.index(ContextBuilder._GOAL_RUNTIME_GUIDANCE_TAG) :
user_msg.index(ContextBuilder._GOAL_RUNTIME_GUIDANCE_END)
].lower()
assert "authorization" not in guidance
assert "host-issued" not in guidance
assert user_msg.index("/goal audit the repo") < user_msg.index(
ContextBuilder._GOAL_RUNTIME_GUIDANCE_TAG
)
assert user_msg.index(ContextBuilder._GOAL_RUNTIME_GUIDANCE_TAG) < user_msg.index(
ContextBuilder._RUNTIME_CONTEXT_TAG
)
assert normal_messages[0]["content"] == messages[0]["content"]
assert ContextBuilder._GOAL_RUNTIME_GUIDANCE_TAG not in str(
normal_messages[-1]["content"]
)
assert ContextBuilder._GOAL_RUNTIME_GUIDANCE_TAG not in str(
stale_messages[-1]["content"]
)
def test_goal_state_does_not_leak_without_session_metadata(self, tmp_path):
builder = _builder(tmp_path)
other_session_meta = {
GOAL_STATE_KEY: {"status": "active", "objective": "Other chat goal."},
}
with_goal = builder.build_messages(
[],
"hi",
channel="websocket",
chat_id="chat-a",
session_metadata=other_session_meta,
)
without_goal = builder.build_messages(
[],
"hi",
channel="websocket",
chat_id="chat-b",
session_metadata={},
)
assert "Other chat goal." in str(with_goal[-1]["content"])
assert "Other chat goal." not in str(without_goal[-1]["content"])
assert "Goal (active):" not in str(without_goal[-1]["content"])
def test_current_runtime_lines_are_injected(self, tmp_path):
def test_explicit_runtime_context_blocks_are_appended(self, tmp_path):
builder = _builder(tmp_path)
messages = builder.build_messages(
[],
"please use @zoom tonight",
current_runtime_lines=[
"CLI App Attachment: @zoom (installed; tool=run_cli_app; entry_point=cli-anything-zoom).",
runtime_context_blocks=[
RuntimeContextBlock(
source="cli_apps",
content="CLI App Attachment: @zoom (installed; tool=run_cli_app).",
),
],
)
user_msg = str(messages[-1]["content"])
assert "CLI App Attachment: @zoom" in user_msg
assert "tool=run_cli_app" in user_msg
assert "entry_point=cli-anything-zoom" in user_msg
assert user_msg.index("please use @zoom tonight") < user_msg.index(
"CLI App Attachment: @zoom"
)
assert messages[-1]["_meta"]["runtime_context"]["sources"] == ["cli_apps"]
def test_consecutive_same_role_merged(self, tmp_path):
builder = _builder(tmp_path)
tests/agent/test_context_prompt_cache.py
+13 -16
View File
@@ -9,6 +9,7 @@ from importlib.resources import files as pkg_files
from pathlib import Path
from nanobot.agent.context import ContextBuilder
from nanobot.runtime_context import RuntimeContextBlock
class _FakeDatetime(real_datetime):
@@ -61,8 +62,7 @@ def test_system_prompt_reflects_current_dream_memory_contract(tmp_path) -> None:
assert "write important facts here" not in prompt
def test_runtime_context_is_separate_untrusted_user_message(tmp_path) -> None:
"""Runtime metadata should be merged with the user message."""
def test_default_user_message_has_no_runtime_context(tmp_path) -> None:
workspace = _make_workspace(tmp_path)
builder = ContextBuilder(workspace)
@@ -76,19 +76,13 @@ def test_runtime_context_is_separate_untrusted_user_message(tmp_path) -> None:
assert messages[0]["role"] == "system"
assert "## Current Session" not in messages[0]["content"]
# Runtime context is now merged with user message into a single message
assert messages[-1]["role"] == "user"
user_content = messages[-1]["content"]
assert isinstance(user_content, str)
assert ContextBuilder._RUNTIME_CONTEXT_TAG in user_content
assert "Current Time:" in user_content
assert "Channel: cli" in user_content
assert "Chat ID: direct" in user_content
assert "Return exactly: OK" in user_content
assert user_content == "Return exactly: OK"
assert "_meta" not in messages[-1]
def test_runtime_context_appended_after_user_content(tmp_path) -> None:
"""User content must precede runtime context for prompt-cache prefix stability."""
def test_provider_context_appended_after_user_content(tmp_path) -> None:
workspace = _make_workspace(tmp_path)
builder = ContextBuilder(workspace)
@@ -97,16 +91,18 @@ def test_runtime_context_appended_after_user_content(tmp_path) -> None:
current_message="hello world",
channel="cli",
chat_id="direct",
runtime_context_blocks=[
RuntimeContextBlock(source="test", content="provider context"),
],
)
content = messages[-1]["content"]
user_pos = content.find("hello world")
tag_pos = content.find(ContextBuilder._RUNTIME_CONTEXT_TAG)
assert user_pos < tag_pos, "user content must precede runtime context for prefix stability"
context_pos = content.find("provider context")
assert user_pos < context_pos, "user content must precede provider context"
def test_runtime_context_includes_sender_id_when_provided(tmp_path) -> None:
"""Sender ID should be included in runtime context when provided."""
def test_sender_id_is_not_injected_without_provider(tmp_path) -> None:
workspace = _make_workspace(tmp_path)
builder = ContextBuilder(workspace)
@@ -120,7 +116,8 @@ def test_runtime_context_includes_sender_id_when_provided(tmp_path) -> None:
user_content = messages[-1]["content"]
assert isinstance(user_content, str)
assert "Sender ID: user-12345" in user_content
assert user_content == "Return exactly: OK"
assert "Sender ID:" not in user_content
def test_runtime_context_excludes_sender_id_when_not_provided(tmp_path) -> None:
tests/agent/test_loop_runner_integration.py
+109 -6
View File
@@ -7,15 +7,16 @@ from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from nanobot.agent.context import ContextBuilder
from nanobot.agent.goal_permission import goal_mutation_allowed, goal_mutation_permission
from nanobot.bus.outbound_events import StreamedResponseEvent
from nanobot.config.schema import AgentDefaults
from nanobot.providers.base import GenerationSettings, LLMResponse, ToolCallRequest
from nanobot.providers.base import GenerationSettings, LLMProvider, LLMResponse, ToolCallRequest
from nanobot.runtime_context import RuntimeContextBlock, public_history_message
from nanobot.session.goal_state import GOAL_STATE_KEY
from nanobot.utils.llm_runtime import LLMRuntime
_MAX_TOOL_RESULT_CHARS = AgentDefaults().max_tool_result_chars
_GOAL_RUNTIME_GUIDANCE_TAG = "[Goal Runtime Guidance — host instructions]"
def _make_loop(tmp_path):
@@ -124,15 +125,117 @@ async def test_goal_command_can_implement_plan_from_prior_discussion(tmp_path):
first_request = provider.chat_with_retry.await_args_list[0].kwargs["messages"]
assert "staged migration plan" in str(first_request)
assert "/goal implement the plan above" in str(first_request)
assert ContextBuilder._GOAL_RUNTIME_GUIDANCE_TAG in str(first_request)
assert _GOAL_RUNTIME_GUIDANCE_TAG in str(first_request)
final_request = provider.chat_with_retry.await_args_list[-1].kwargs["messages"]
assert "create_goal is unavailable for this turn" in str(final_request)
assert all(
ContextBuilder._GOAL_RUNTIME_GUIDANCE_TAG not in str(message.get("content") or "")
for message in session.messages
assert _GOAL_RUNTIME_GUIDANCE_TAG in str(session.messages[2]["content"])
assert _GOAL_RUNTIME_GUIDANCE_TAG not in str(
public_history_message(session.messages[2])["content"]
)
@pytest.mark.asyncio
async def test_runtime_context_is_persisted_as_next_turn_prompt_prefix(tmp_path):
from nanobot.agent.loop import AgentLoop
from nanobot.bus.events import InboundMessage
from nanobot.bus.queue import MessageBus
provider = MagicMock()
provider.get_default_model.return_value = "test-model"
provider.generation = GenerationSettings()
provider.chat_with_retry = AsyncMock(side_effect=[
LLMResponse(content="first answer", usage={}),
LLMResponse(content="second answer", usage={}),
])
loop = AgentLoop(bus=MessageBus(), provider=provider, workspace=tmp_path, model="test-model")
loop.consolidator.maybe_consolidate_by_tokens = AsyncMock(return_value=None)
session = loop.sessions.get_or_create("cli:direct")
provider_calls: list[str | None] = []
async def provide_context(request):
provider_calls.append(request.turn_id)
return RuntimeContextBlock(source="test", content="stable provider context")
loop.register_runtime_context_provider(provide_context)
loop.register_runtime_context_provider(provide_context)
await loop._process_message(InboundMessage(
channel="cli",
sender_id="user",
chat_id="direct",
content="first turn",
))
await loop._process_message(InboundMessage(
channel="cli",
sender_id="user",
chat_id="direct",
content="second turn",
))
first_request = provider.chat_with_retry.await_args_list[0].kwargs["messages"]
second_request = provider.chat_with_retry.await_args_list[1].kwargs["messages"]
first_wire = LLMProvider._sanitize_empty_content(first_request)
second_wire = LLMProvider._sanitize_empty_content(second_request)
assert second_wire[: len(first_wire)] == first_wire
assert first_wire[1] == second_wire[1]
assert second_wire[2]["role"] == "assistant"
assert second_wire[2]["content"] == "first answer"
assert second_wire[3]["content"].startswith("second turn")
assert "Current Time:" not in str(second_wire)
assert "Chat ID:" not in str(second_wire)
assert len(provider_calls) == 2
persisted_first_user = session.messages[0]
assert persisted_first_user["content"] == first_wire[1]["content"]
assert public_history_message(persisted_first_user)["content"] == "first turn"
@pytest.mark.asyncio
async def test_runtime_context_provider_runs_once_across_tool_iterations(tmp_path):
from nanobot.agent.loop import AgentLoop
from nanobot.bus.events import InboundMessage
from nanobot.bus.queue import MessageBus
(tmp_path / "note.txt").write_text("hello", encoding="utf-8")
provider = MagicMock()
provider.get_default_model.return_value = "test-model"
provider.generation = GenerationSettings()
provider.chat_with_retry = AsyncMock(side_effect=[
LLMResponse(
content="reading",
tool_calls=[ToolCallRequest(
id="call_read",
name="read_file",
arguments={"path": "note.txt"},
)],
usage={},
),
LLMResponse(content="done", usage={}),
])
loop = AgentLoop(bus=MessageBus(), provider=provider, workspace=tmp_path, model="test-model")
loop.consolidator.maybe_consolidate_by_tokens = AsyncMock(return_value=None)
provider_calls = 0
async def provide_context(_request):
nonlocal provider_calls
provider_calls += 1
return RuntimeContextBlock(source="test", content="frozen context")
loop.register_runtime_context_provider(provide_context)
await loop._process_message(InboundMessage(
channel="cli",
sender_id="user",
chat_id="direct",
content="read the note",
))
assert provider.chat_with_retry.await_count == 2
assert provider_calls == 1
for call in provider.chat_with_retry.await_args_list:
assert "frozen context" in str(call.kwargs["messages"])
@pytest.mark.asyncio
async def test_non_goal_direct_turn_cannot_reuse_prior_goal_command(tmp_path):
from nanobot.agent.loop import AgentLoop
tests/agent/test_loop_save_turn.py
+60 -57
View File
@@ -18,8 +18,15 @@ from nanobot.bus.outbound_events import (
)
from nanobot.bus.queue import MessageBus
from nanobot.cron.session_turns import CRON_HISTORY_META, CRON_TRIGGER_META
from nanobot.providers.base import LLMProvider, LLMResponse
from nanobot.providers.base import LLMResponse
from nanobot.providers.factory import ProviderSnapshot
from nanobot.runtime_context import (
RUNTIME_CONTEXT_HISTORY_META,
RUNTIME_CONTEXT_MESSAGE_META,
RuntimeContextBlock,
append_runtime_context,
public_history_message,
)
from nanobot.session.automation_turns import AUTOMATION_HISTORY_META
from nanobot.session.goal_state import GOAL_STATE_KEY
from nanobot.session.manager import Session, SessionManager
@@ -48,19 +55,13 @@ def _mk_loop() -> AgentLoop:
return loop
def _host_text_message(content: str, suffix: str) -> dict:
def _runtime_message(content, blocks: list[RuntimeContextBlock]) -> dict:
merged, marker = append_runtime_context(content, blocks)
assert marker is not None
return {
"role": "user",
"content": content,
"_meta": {ContextBuilder._HOST_TEXT_SUFFIX_META_KEY: suffix},
}
def _host_text_block(text: str) -> dict:
return {
"type": "text",
"text": text,
"_meta": {ContextBuilder._HOST_BLOCK_META_KEY: True},
"content": merged,
"_meta": {RUNTIME_CONTEXT_MESSAGE_META: marker},
}
@@ -357,74 +358,80 @@ def test_webui_title_update_uses_captured_llm_runtime(
assert captured["model"] == "turn-model"
def test_save_turn_skips_multimodal_user_when_only_runtime_context() -> None:
def test_save_turn_keeps_multimodal_runtime_context_for_model_replay() -> None:
loop = _mk_loop()
session = Session(key="test:runtime-only")
runtime = ContextBuilder._RUNTIME_CONTEXT_TAG + "\nCurrent Time: now (UTC)"
block = RuntimeContextBlock(source="test", content="provider context")
loop._save_turn(
session,
[{"role": "user", "content": [_host_text_block(runtime)]}],
[_runtime_message([], [block])],
skip=0,
)
assert session.messages == []
assert session.messages[0]["content"] == [
{"type": "text", "text": "provider context"}
]
assert public_history_message(session.messages[0])["content"] == []
def test_save_turn_keeps_image_placeholder_with_path_after_runtime_strip() -> None:
def test_save_turn_keeps_image_placeholder_and_runtime_context() -> None:
loop = _mk_loop()
session = Session(key="test:image")
runtime = ContextBuilder._RUNTIME_CONTEXT_TAG + "\nCurrent Time: now (UTC)"
block = RuntimeContextBlock(source="test", content="provider context")
loop._save_turn(
session,
[{
"role": "user",
"content": [
[_runtime_message(
[
{"type": "image_url", "image_url": {"url": "data:image/png;base64,abc"}, "_meta": {"path": "/media/feishu/photo.jpg"}},
_host_text_block(runtime),
],
}],
[block],
)],
skip=0,
)
assert session.messages[0]["content"] == [{"type": "text", "text": "[image: /media/feishu/photo.jpg]"}]
assert session.messages[0]["content"] == [
{"type": "text", "text": "[image: /media/feishu/photo.jpg]"},
{"type": "text", "text": "provider context"},
]
assert public_history_message(session.messages[0])["content"] == [
{"type": "text", "text": "[image: /media/feishu/photo.jpg]"}
]
def test_save_turn_keeps_image_placeholder_without_meta() -> None:
loop = _mk_loop()
session = Session(key="test:image-no-meta")
runtime = ContextBuilder._RUNTIME_CONTEXT_TAG + "\nCurrent Time: now (UTC)"
block = RuntimeContextBlock(source="test", content="provider context")
loop._save_turn(
session,
[{
"role": "user",
"content": [
[_runtime_message(
[
{"type": "image_url", "image_url": {"url": "data:image/png;base64,abc"}},
_host_text_block(runtime),
],
}],
[block],
)],
skip=0,
)
assert session.messages[0]["content"] == [{"type": "text", "text": "[image]"}]
assert session.messages[0]["content"] == [
{"type": "text", "text": "[image]"},
{"type": "text", "text": "provider context"},
]
def test_save_turn_strips_host_guidance_suffix_from_string() -> None:
def test_save_turn_persists_runtime_context_and_public_view_hides_it() -> None:
loop = _mk_loop()
session = Session(key="test:suffix-strip")
guidance = ContextBuilder._GOAL_RUNTIME_GUIDANCE_TAG + "\ninternal guidance"
runtime = (
ContextBuilder._RUNTIME_CONTEXT_TAG
+ "\nCurrent Time: now\n"
+ ContextBuilder._RUNTIME_CONTEXT_END
)
suffix = f"{guidance}\n\n{runtime}"
block = RuntimeContextBlock(source="goal", content="internal goal guidance")
loop._save_turn(
session,
[_host_text_message(f"hello world\n\n{suffix}", suffix)],
[_runtime_message("hello world", [block])],
skip=0,
)
assert session.messages[0]["content"] == "hello world"
assert session.messages[0]["content"] == "hello world\n\ninternal goal guidance"
assert session.messages[0][RUNTIME_CONTEXT_HISTORY_META]["sources"] == ["goal"]
assert public_history_message(session.messages[0])["content"] == "hello world"
def test_build_and_save_preserves_user_text_containing_goal_guidance_tag(tmp_path: Path) -> None:
@@ -432,7 +439,7 @@ def test_build_and_save_preserves_user_text_containing_goal_guidance_tag(tmp_pat
session = Session(key="test:user-guidance-literal")
user_text = (
"Keep this prefix\n"
f"{ContextBuilder._GOAL_RUNTIME_GUIDANCE_TAG}\n"
"[Goal Runtime Guidance — host instructions]\n"
"This label and everything after it are user-authored."
)
messages = ContextBuilder(tmp_path).build_messages(
@@ -440,10 +447,8 @@ def test_build_and_save_preserves_user_text_containing_goal_guidance_tag(tmp_pat
user_text,
channel="cli",
chat_id="direct",
goal_start_requested=True,
)
assert "_meta" in messages[-1]
assert "_meta" not in LLMProvider._sanitize_empty_content(messages)[-1]
assert "_meta" not in messages[-1]
loop._save_turn(session, messages, skip=1)
@@ -467,7 +472,6 @@ def test_build_and_save_preserves_multimodal_user_block_starting_with_runtime_ta
media=[str(image)],
channel="cli",
chat_id="direct",
goal_start_requested=True,
)
loop._save_turn(session, messages, skip=1)
@@ -475,21 +479,18 @@ def test_build_and_save_preserves_multimodal_user_block_starting_with_runtime_ta
assert {"type": "text", "text": user_text} in session.messages[0]["content"]
def test_save_turn_skips_string_user_when_only_runtime_context_suffix() -> None:
def test_save_turn_keeps_string_when_only_runtime_context() -> None:
loop = _mk_loop()
session = Session(key="test:suffix-only")
runtime = (
ContextBuilder._RUNTIME_CONTEXT_TAG
+ "\nCurrent Time: now\n"
+ ContextBuilder._RUNTIME_CONTEXT_END
)
block = RuntimeContextBlock(source="test", content="provider context")
loop._save_turn(
session,
[_host_text_message(runtime, runtime)],
[_runtime_message("", [block])],
skip=0,
)
assert session.messages == []
assert session.messages[0]["content"] == "provider context"
assert public_history_message(session.messages[0])["content"] == ""
def test_save_turn_keeps_tool_results_under_16k() -> None:
@@ -847,9 +848,10 @@ async def test_internal_continuation_queues_turn_without_fake_user_history(
assert "Finish the long goal." in queued.content
session = loop.sessions.get_or_create("feishu:c-auto")
assert "Finish the long goal." in str(session.messages[0]["content"])
assert [
{k: v for k, v in m.items() if k in {"role", "content"}}
for m in session.messages
for m in map(public_history_message, session.messages)
] == [{"role": "user", "content": "start the goal"}]
second = await loop._process_message(queued, pending_queue=asyncio.Queue())
@@ -859,7 +861,7 @@ async def test_internal_continuation_queues_turn_without_fake_user_history(
session = loop.sessions.get_or_create("feishu:c-auto")
assert [
{k: v for k, v in m.items() if k in {"role", "content"}}
for m in session.messages
for m in map(public_history_message, session.messages)
] == [
{"role": "user", "content": "start the goal"},
{"role": "assistant", "content": "done"},
@@ -1399,7 +1401,8 @@ async def test_system_subagent_followup_is_persisted_before_prompt_assembly(tmp_
assert "[Message Time:" not in non_system[0]["content"]
assert "[Message Time:" not in non_system[1]["content"]
assert non_system[2]["content"].count("subagent result") == 1
assert "Current Time:" in non_system[2]["content"]
assert "Current Time:" not in non_system[2]["content"]
assert non_system[2]["content"] == "subagent result"
loop.sessions.invalidate("cli:test")
persisted = loop.sessions.get_or_create("cli:test")
tests/agent/test_runtime_context.py
+118
View File
@@ -0,0 +1,118 @@
from __future__ import annotations
from types import SimpleNamespace
import pytest
from nanobot.agent.tools.context import RequestContext
from nanobot.runtime_context import (
RUNTIME_CONTEXT_HISTORY_META,
RuntimeContextBlock,
append_runtime_context,
public_history_message,
resolve_runtime_context,
)
from nanobot.sdk.types import snapshot_from_session
from nanobot.session.manager import Session, _message_preview_text
from nanobot.session.webui_turns import _title_inputs
from nanobot.webui.transcript import _session_user_event
@pytest.mark.asyncio
async def test_resolve_runtime_context_preserves_provider_order() -> None:
calls: list[str] = []
async def first(_request: RequestContext):
calls.append("first")
return RuntimeContextBlock(source="first", content="one")
async def second(_request: RequestContext):
calls.append("second")
return [RuntimeContextBlock(source="second", content="two")]
blocks = await resolve_runtime_context(
[first, second],
RequestContext(channel="cli", chat_id="direct"),
)
assert calls == ["first", "second"]
assert [(block.source, block.content) for block in blocks] == [
("first", "one"),
("second", "two"),
]
def test_public_history_removes_only_trusted_exact_suffix() -> None:
block = RuntimeContextBlock(source="goal", content="private goal context")
content, marker = append_runtime_context("visible user text", [block])
assert marker is not None
persisted = {
"role": "user",
"content": content,
RUNTIME_CONTEXT_HISTORY_META: marker,
}
assert public_history_message(persisted) == {
"role": "user",
"content": "visible user text",
}
user_authored = {
"role": "user",
"content": "visible user text\n\nprivate goal context",
}
assert public_history_message(user_authored) == user_authored
def test_public_history_keeps_content_when_marker_does_not_match() -> None:
message = {
"role": "user",
"content": "user-edited content",
RUNTIME_CONTEXT_HISTORY_META: {
"version": 1,
"sources": ["goal"],
"suffix": "different suffix",
},
}
assert public_history_message(message) == {
"role": "user",
"content": "user-edited content",
}
def test_sdk_snapshot_hides_runtime_context() -> None:
block = RuntimeContextBlock(source="goal", content="private goal context")
content, marker = append_runtime_context("visible user text", [block])
session = SimpleNamespace(
key="cli:direct",
created_at=SimpleNamespace(isoformat=lambda: "created"),
updated_at=SimpleNamespace(isoformat=lambda: "updated"),
metadata={},
messages=[{
"role": "user",
"content": content,
RUNTIME_CONTEXT_HISTORY_META: marker,
}],
)
snapshot = snapshot_from_session(session)
assert snapshot.messages == [{"role": "user", "content": "visible user text"}]
def test_webui_preview_title_and_backfill_hide_runtime_context() -> None:
block = RuntimeContextBlock(source="goal", content="private goal context")
content, marker = append_runtime_context("visible user text", [block])
persisted = {
"role": "user",
"content": content,
RUNTIME_CONTEXT_HISTORY_META: marker,
}
session = Session(key="websocket:chat", messages=[persisted])
assert _message_preview_text(persisted) == "visible user text"
assert _title_inputs(session) == ("visible user text", "")
event = _session_user_event("websocket:chat", persisted)
assert event is not None
assert event["text"] == "visible user text"
tests/agent/test_session_manager_history.py
+40
View File
@@ -1,3 +1,8 @@
from nanobot.runtime_context import (
RUNTIME_CONTEXT_HISTORY_META,
RuntimeContextBlock,
append_runtime_context,
)
from nanobot.session.manager import Session, SessionManager
@@ -425,6 +430,41 @@ def test_get_history_synthesizes_cli_app_attachment_breadcrumb():
}]
def test_get_history_does_not_duplicate_persisted_capability_runtime_context():
content, marker = append_runtime_context(
"please use @drawio",
[RuntimeContextBlock(
source="cli_apps",
content="[Runtime Context]\nCLI App Attachment: @drawio",
), RuntimeContextBlock(
source="mcp",
content="[Runtime Context]\nMCP Preset Attachment: @linear",
)],
)
session = Session(key="test:cli-app-persisted")
session.messages.append({
"role": "user",
"content": content,
"cli_apps": [{
"name": "drawio",
"entry_point": "cli-anything-drawio",
}],
"mcp_presets": [{"name": "linear", "transport": "stdio"}],
RUNTIME_CONTEXT_HISTORY_META: marker,
})
model_history = session.get_history(max_messages=500)
public_history = session.get_history(
max_messages=500,
include_runtime_context=False,
)
assert model_history == [{"role": "user", "content": content}]
assert model_history[0]["content"].count("CLI App Attachment: @drawio") == 1
assert model_history[0]["content"].count("MCP Preset Attachment: @linear") == 1
assert public_history == [{"role": "user", "content": "please use @drawio"}]
def test_fork_session_before_user_index_copies_only_prefix(tmp_path):
manager = SessionManager(tmp_path)
source = manager.get_or_create("websocket:source")
tests/channels/test_websocket_http_routes.py
+41
View File
@@ -21,6 +21,11 @@ from nanobot.channels.websocket import WebSocketChannel, WebSocketConfig
from nanobot.cron.service import CronService
from nanobot.cron.types import CronJob, CronPayload, CronSchedule
from nanobot.optional_features import InstallResult
from nanobot.runtime_context import (
RUNTIME_CONTEXT_HISTORY_META,
RuntimeContextBlock,
append_runtime_context,
)
from nanobot.session.keys import UNIFIED_SESSION_KEY
from nanobot.session.manager import Session, SessionManager
from nanobot.triggers.local_store import LocalTriggerStore
@@ -1739,6 +1744,42 @@ async def test_session_routes_accept_percent_encoded_websocket_keys(
await server_task
@pytest.mark.asyncio
async def test_session_messages_hide_persisted_runtime_context(
bus: MagicMock, tmp_path: Path
) -> None:
sm = SessionManager(tmp_path)
session = sm.get_or_create("websocket:runtime-context")
content, marker = append_runtime_context(
"visible user text",
[RuntimeContextBlock(source="goal", content="private goal context")],
)
session.add_message(
"user",
content,
**{RUNTIME_CONTEXT_HISTORY_META: marker},
)
sm.save(session)
channel = _ch(bus, session_manager=sm, port=29919)
server_task = asyncio.create_task(channel.start())
await asyncio.sleep(0.3)
try:
token = channel.gateway.tokens.issue_api_token(300)
response = await _http_get(
"http://127.0.0.1:29919/api/sessions/websocket:runtime-context/messages",
headers={"Authorization": f"Bearer {token}"},
)
assert response.status_code == 200
message = response.json()["messages"][0]
assert message["content"] == "visible user text"
assert RUNTIME_CONTEXT_HISTORY_META not in message
assert "private goal context" not in response.text
finally:
await channel.stop()
await server_task
@pytest.mark.asyncio
async def test_webui_thread_resigns_assistant_media_urls(
bus: MagicMock, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
tests/cli_apps/test_tool.py
+31
View File
@@ -7,6 +7,7 @@ import time
from pathlib import Path
from nanobot.agent.tools.cli_apps import CliAppsTool
from nanobot.agent.tools.context import RequestContext
from nanobot.apps.cli.service import CliAppManager, CliAppsRuntimeConfig
@@ -125,3 +126,33 @@ def test_run_cli_app_description_names_only_settings_installed_apps(tmp_path: Pa
assert "Settings CLI Apps: drawio" in tool.description
assert "ordinary system CLIs such as git, gh" in tool.description
def test_cli_app_tool_provides_context_only_for_attachment(tmp_path: Path) -> None:
tool = CliAppsTool(workspace=tmp_path)
provider = tool.runtime_context_provider()
assert provider is not None
empty = asyncio.run(provider(RequestContext(
channel="websocket",
chat_id="chat",
original_user_text="hello",
workspace=tmp_path,
)))
attached = asyncio.run(provider(RequestContext(
channel="websocket",
chat_id="chat",
original_user_text="use @drawio",
metadata={
"cli_apps": [{
"name": "drawio",
"entry_point": "cli-anything-drawio",
}],
},
workspace=tmp_path,
)))
assert empty is None
assert attached is not None
assert attached.source == "cli_apps"
assert "CLI App Attachment: @drawio" in attached.content