fix(agent): preserve pending runtime context
This commit is contained in:
+42
-8
@@ -745,14 +745,23 @@ class AgentLoop:
|
||||
self,
|
||||
ctx: TurnContext,
|
||||
) -> list[RuntimeContextBlock]:
|
||||
tools = ctx.tools or self.tools
|
||||
assert ctx.request_context is not None
|
||||
return await self._resolve_runtime_context_for_request(
|
||||
ctx.request_context,
|
||||
ctx.tools or self.tools,
|
||||
)
|
||||
|
||||
async def _resolve_runtime_context_for_request(
|
||||
self,
|
||||
request: RequestContext,
|
||||
tools: ToolRegistry,
|
||||
) -> list[RuntimeContextBlock]:
|
||||
providers = [
|
||||
*tools.get_runtime_context_providers(),
|
||||
*self._runtime_context_providers,
|
||||
]
|
||||
assert ctx.request_context is not None
|
||||
blocks = runtime_context_blocks_from_metadata(ctx.request_context.metadata)
|
||||
blocks.extend(await resolve_runtime_context(providers, ctx.request_context))
|
||||
blocks = runtime_context_blocks_from_metadata(request.metadata)
|
||||
blocks.extend(await resolve_runtime_context(providers, request))
|
||||
return blocks
|
||||
|
||||
async def _dispatch_command_inline(
|
||||
@@ -855,7 +864,7 @@ class AgentLoop:
|
||||
if pending_queue is None:
|
||||
return []
|
||||
|
||||
def _to_user_message(pending_msg: InboundMessage) -> dict[str, Any]:
|
||||
async def _to_user_message(pending_msg: InboundMessage) -> dict[str, Any]:
|
||||
content = pending_msg.content
|
||||
media = pending_msg.media if pending_msg.media else None
|
||||
if media:
|
||||
@@ -864,6 +873,31 @@ class AgentLoop:
|
||||
user_content = self.context._build_user_content(content, media)
|
||||
row: dict[str, Any] = {"role": "user", "content": user_content}
|
||||
metadata = pending_msg.metadata if isinstance(pending_msg.metadata, dict) else {}
|
||||
if pending_msg.channel != "system":
|
||||
scope = self.workspace_scopes.for_turn(
|
||||
channel=pending_msg.channel,
|
||||
message_metadata=metadata,
|
||||
session_metadata=session.metadata if session is not None else None,
|
||||
)
|
||||
pending_request = RequestContext(
|
||||
channel=pending_msg.channel,
|
||||
chat_id=pending_msg.chat_id,
|
||||
message_id=metadata.get("message_id"),
|
||||
session_key=active_session_key,
|
||||
original_user_text=pending_msg.content,
|
||||
runtime=runtime,
|
||||
metadata=dict(metadata),
|
||||
sender_id=pending_msg.sender_id,
|
||||
turn_id=request_ctx.turn_id,
|
||||
workspace=scope.project_path,
|
||||
)
|
||||
blocks = await self._resolve_runtime_context_for_request(
|
||||
pending_request,
|
||||
effective_tools,
|
||||
)
|
||||
row["content"], marker = append_runtime_context(user_content, blocks)
|
||||
if marker is not None:
|
||||
row["_meta"] = {RUNTIME_CONTEXT_MESSAGE_META: marker}
|
||||
if (
|
||||
pending_msg.sender_id == "subagent"
|
||||
and metadata.get("injected_event") == "subagent_result"
|
||||
@@ -880,7 +914,7 @@ class AgentLoop:
|
||||
items: list[dict[str, Any]] = []
|
||||
while len(items) < limit:
|
||||
try:
|
||||
items.append(_to_user_message(pending_queue.get_nowait()))
|
||||
items.append(await _to_user_message(pending_queue.get_nowait()))
|
||||
except asyncio.QueueEmpty:
|
||||
break
|
||||
|
||||
@@ -898,10 +932,10 @@ class AgentLoop:
|
||||
session.key,
|
||||
)
|
||||
return items
|
||||
items.append(_to_user_message(msg))
|
||||
items.append(await _to_user_message(msg))
|
||||
while len(items) < limit:
|
||||
try:
|
||||
items.append(_to_user_message(pending_queue.get_nowait()))
|
||||
items.append(await _to_user_message(pending_queue.get_nowait()))
|
||||
except asyncio.QueueEmpty:
|
||||
break
|
||||
|
||||
|
||||
@@ -468,6 +468,100 @@ async def test_loop_injected_followup_preserves_image_media(tmp_path):
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_pending_injection_resolves_its_own_runtime_context(tmp_path):
|
||||
from nanobot.agent.loop import AgentLoop
|
||||
from nanobot.bus.events import InboundMessage
|
||||
from nanobot.bus.queue import MessageBus
|
||||
from nanobot.runtime_context import (
|
||||
RUNTIME_CONTEXT_MESSAGE_META,
|
||||
RuntimeContextBlock,
|
||||
public_history_message,
|
||||
wrap_runtime_context_lines,
|
||||
)
|
||||
|
||||
provider = MagicMock()
|
||||
provider.get_default_model.return_value = "test-model"
|
||||
provider.chat_with_retry = AsyncMock(side_effect=[
|
||||
LLMResponse(content="first answer", tool_calls=[], usage={}),
|
||||
LLMResponse(content="second answer", tool_calls=[], usage={}),
|
||||
])
|
||||
loop = AgentLoop(
|
||||
bus=MessageBus(),
|
||||
provider=provider,
|
||||
workspace=tmp_path,
|
||||
model="test-model",
|
||||
)
|
||||
loop.tools.get_definitions = MagicMock(return_value=[])
|
||||
seen_contexts = []
|
||||
|
||||
async def provide_identity(request):
|
||||
seen_contexts.append((
|
||||
request.channel,
|
||||
request.chat_id,
|
||||
request.sender_id,
|
||||
request.message_id,
|
||||
request.session_key,
|
||||
request.original_user_text,
|
||||
request.metadata["sender_name"],
|
||||
request.metadata["thread_id"],
|
||||
))
|
||||
return RuntimeContextBlock(
|
||||
source="identity",
|
||||
content=wrap_runtime_context_lines([
|
||||
" | ".join(str(value) for value in seen_contexts[-1]),
|
||||
]),
|
||||
)
|
||||
|
||||
loop.register_runtime_context_provider(provide_identity)
|
||||
session = loop.sessions.get_or_create("telegram:group-1")
|
||||
pending_queue = asyncio.Queue()
|
||||
await pending_queue.put(InboundMessage(
|
||||
channel="telegram",
|
||||
sender_id="user-b",
|
||||
chat_id="group-1",
|
||||
content="follow-up from the second speaker",
|
||||
metadata={
|
||||
"message_id": "message-2",
|
||||
"sender_name": "Bob",
|
||||
"thread_id": "topic-7",
|
||||
},
|
||||
))
|
||||
|
||||
_, _, all_messages, _, _ = await loop._run_agent_loop(
|
||||
[{"role": "user", "content": "initial message from user A"}],
|
||||
runtime=loop.llm_runtime(),
|
||||
session=session,
|
||||
channel="telegram",
|
||||
chat_id="group-1",
|
||||
session_key=session.key,
|
||||
pending_queue=pending_queue,
|
||||
)
|
||||
|
||||
assert seen_contexts == [(
|
||||
"telegram",
|
||||
"group-1",
|
||||
"user-b",
|
||||
"message-2",
|
||||
session.key,
|
||||
"follow-up from the second speaker",
|
||||
"Bob",
|
||||
"topic-7",
|
||||
)]
|
||||
|
||||
injected = [message for message in all_messages if message.get("role") == "user"][-1]
|
||||
assert "follow-up from the second speaker" in str(injected["content"])
|
||||
model_messages = provider.chat_with_retry.await_args_list[-1].kwargs["messages"]
|
||||
assert "telegram | group-1 | user-b | message-2" in str(model_messages)
|
||||
assert "Bob | topic-7" in str(model_messages)
|
||||
assert injected["_meta"][RUNTIME_CONTEXT_MESSAGE_META]["sources"] == ["identity"]
|
||||
|
||||
loop._save_turn(session, all_messages, skip=1)
|
||||
persisted = [message for message in session.messages if message.get("role") == "user"][-1]
|
||||
assert "telegram | group-1 | user-b | message-2" in str(persisted["content"])
|
||||
assert public_history_message(persisted)["content"] == "follow-up from the second speaker"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_subagent_pending_injection_is_hidden_history_and_not_merged(tmp_path):
|
||||
from nanobot.agent.loop import AgentLoop
|
||||
|
||||
Reference in New Issue
Block a user