diff --git a/nanobot/agent/loop.py b/nanobot/agent/loop.py index 28b6131b..a5ba9250 100644 --- a/nanobot/agent/loop.py +++ b/nanobot/agent/loop.py @@ -647,13 +647,17 @@ class AgentLoop: session, pending = self.auto_compact.prepare_session(session, key) await self.consolidator.maybe_consolidate_by_tokens(session) + if msg.sender_id == "subagent" and self._persist_subagent_followup(session, msg): + self.sessions.save(session) self._set_tool_context(channel, chat_id, msg.metadata.get("message_id")) history = session.get_history(max_messages=0) current_role = "assistant" if msg.sender_id == "subagent" else "user" messages = self.context.build_messages( history=history, - current_message=msg.content, channel=channel, chat_id=chat_id, + current_message="" if msg.sender_id == "subagent" else msg.content, + channel=channel, + chat_id=chat_id, session_summary=pending, current_role=current_role, ) @@ -870,6 +874,23 @@ class AgentLoop: session.messages.append(entry) session.updated_at = datetime.now() + def _persist_subagent_followup(self, session: Session, msg: InboundMessage) -> bool: + """Persist subagent follow-ups before prompt assembly so history stays durable.""" + task_id = msg.metadata.get("subagent_task_id") if isinstance(msg.metadata, dict) else None + if task_id and any( + m.get("injected_event") == "subagent_result" and m.get("subagent_task_id") == task_id + for m in session.messages + ): + return False + session.add_message( + "assistant", + msg.content, + sender_id=msg.sender_id, + injected_event="subagent_result", + subagent_task_id=task_id, + ) + return True + 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 diff --git a/nanobot/agent/subagent.py b/nanobot/agent/subagent.py index 394292c8..388a634c 100644 --- a/nanobot/agent/subagent.py +++ b/nanobot/agent/subagent.py @@ -246,6 +246,10 @@ class SubagentManager: sender_id="subagent", chat_id=f"{origin['channel']}:{origin['chat_id']}", content=announce_content, + metadata={ + "injected_event": "subagent_result", + "subagent_task_id": task_id, + }, ) await self.bus.publish_inbound(msg) diff --git a/tests/agent/test_loop_save_turn.py b/tests/agent/test_loop_save_turn.py index 8885e0cc..62ba504f 100644 --- a/tests/agent/test_loop_save_turn.py +++ b/tests/agent/test_loop_save_turn.py @@ -417,3 +417,147 @@ async def test_stop_preserves_runtime_checkpoint_for_next_turn(tmp_path: Path) - ] assert AgentLoop._PENDING_USER_TURN_KEY not in session.metadata assert AgentLoop._RUNTIME_CHECKPOINT_KEY not in session.metadata + + +@pytest.mark.asyncio +async def test_system_subagent_followup_is_persisted_before_prompt_assembly(tmp_path: Path) -> None: + loop = _make_full_loop(tmp_path) + loop.consolidator.maybe_consolidate_by_tokens = AsyncMock(return_value=False) # type: ignore[method-assign] + + session = loop.sessions.get_or_create("cli:test") + session.add_message("user", "question") + session.add_message("assistant", "working") + loop.sessions.save(session) + + seen: dict[str, list[dict]] = {} + + async def fake_run_agent_loop(initial_messages, **_kwargs): + seen["initial_messages"] = initial_messages + return ( + "done", + [], + [*initial_messages, {"role": "assistant", "content": "done"}], + "stop", + False, + ) + + loop._run_agent_loop = fake_run_agent_loop # type: ignore[method-assign] + + await loop._process_message( + InboundMessage( + channel="system", + sender_id="subagent", + chat_id="cli:test", + content="subagent result", + metadata={"subagent_task_id": "sub-1"}, + ) + ) + + non_system = [m for m in seen["initial_messages"] if m.get("role") != "system"] + assert [m["content"] for m in non_system[:2]] == ["question", "working"] + assert non_system[2]["content"].count("subagent result") == 1 + assert "Current Time:" in non_system[2]["content"] + + loop.sessions.invalidate("cli:test") + persisted = loop.sessions.get_or_create("cli:test") + assert [ + {k: v for k, v in m.items() if k in {"role", "content", "injected_event", "subagent_task_id"}} + for m in persisted.messages + ] == [ + {"role": "user", "content": "question"}, + {"role": "assistant", "content": "working"}, + { + "role": "assistant", + "content": "subagent result", + "injected_event": "subagent_result", + "subagent_task_id": "sub-1", + }, + {"role": "assistant", "content": "done"}, + ] + + +@pytest.mark.asyncio +async def test_multiple_subagent_followups_all_persist_as_standalone_history(tmp_path: Path) -> None: + loop = _make_full_loop(tmp_path) + loop.consolidator.maybe_consolidate_by_tokens = AsyncMock(return_value=False) # type: ignore[method-assign] + + async def fake_run_agent_loop(initial_messages, **_kwargs): + return ( + "ack", + [], + [*initial_messages, {"role": "assistant", "content": "ack"}], + "stop", + False, + ) + + loop._run_agent_loop = fake_run_agent_loop # type: ignore[method-assign] + + for idx in range(3): + await loop._process_message( + InboundMessage( + channel="system", + sender_id="subagent", + chat_id="cli:multi", + content=f"subagent result {idx}", + metadata={"subagent_task_id": f"sub-{idx}"}, + ) + ) + + loop.sessions.invalidate("cli:multi") + persisted = loop.sessions.get_or_create("cli:multi") + followups = [m for m in persisted.messages if m.get("injected_event") == "subagent_result"] + assert [m["content"] for m in followups] == [ + "subagent result 0", + "subagent result 1", + "subagent result 2", + ] + + +def test_prompt_merge_does_not_replace_standalone_subagent_history_entry(tmp_path: Path) -> None: + loop = _mk_loop() + session = Session(key="cli:merge") + session.add_message("assistant", "previous assistant") + + inserted = loop._persist_subagent_followup( + session, + InboundMessage( + channel="system", + sender_id="subagent", + chat_id="cli:merge", + content="subagent result", + metadata={"subagent_task_id": "sub-1"}, + ), + ) + + assert inserted is True + + builder = ContextBuilder(tmp_path) + projected = builder.build_messages( + history=session.get_history(max_messages=0), + current_message="", + current_role="assistant", + channel="cli", + chat_id="merge", + ) + + non_system = [m for m in projected if m.get("role") != "system"] + assert len(non_system) == 2 + assert "subagent result" in non_system[-1]["content"] + assert session.messages[-1]["content"] == "subagent result" + assert session.messages[-1]["injected_event"] == "subagent_result" + + +def test_subagent_followup_dedupes_by_task_id() -> None: + loop = _mk_loop() + session = Session(key="cli:dedupe") + msg = InboundMessage( + channel="system", + sender_id="subagent", + chat_id="cli:dedupe", + content="subagent result", + metadata={"subagent_task_id": "sub-1"}, + ) + + assert loop._persist_subagent_followup(session, msg) is True + assert loop._persist_subagent_followup(session, msg) is False + assert len(session.messages) == 1