fix(loop): persist subagent follow-up events in history

2026-04-18 13:30:22 +08:00
parent c27b4d07c4
commit 1c939e8a5f
3 changed files with 170 additions and 1 deletions
@@ -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
@@ -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)
@@ -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