diff --git a/README.md b/README.md index 59f942d878..555fb2075e 100644 --- a/README.md +++ b/README.md @@ -1858,6 +1858,8 @@ request the binary capability retain the legacy JSON/base64 frame protocol. **Visible Tool-Run Completion**: For interactive turns, DeerFlow retries an empty post-tool final response once, then surfaces a visible error instead of reporting a silent successful run. +**Run History**: Image-only user input is recorded once per run, even when the agent makes multiple model calls. Its media content remains in history without requiring a text summary. + ### Reading a Referenced Conversation Gateway API callers can opt into `read_conversation` and submit a diff --git a/backend/packages/harness/deerflow/runtime/AGENTS.md b/backend/packages/harness/deerflow/runtime/AGENTS.md index 80b5734b5f..edc8fa794c 100644 --- a/backend/packages/harness/deerflow/runtime/AGENTS.md +++ b/backend/packages/harness/deerflow/runtime/AGENTS.md @@ -26,6 +26,11 @@ Checkpointer storage runs in one of two channel modes, selected by `checkpoint_c **Message sequence placement:** Keep backend and frontend message identity rules aligned. Details: `backend/docs/runtime-guidance-details.md`. +**Human-input capture** (`runtime/journal.py`): track capture separately from +the optional display summary. Image-only input has no text but must still stop +the batch scan and later model calls from appending another human-input event. +`tests/test_run_journal.py` covers callback and full/delta graph paths. + **LLM response callback coalescing** (`runtime/journal.py`): a provider may fire `on_llm_end` twice for one LangChain run id, first without usage (or with all token counts zero) and immediately again with usage populated. The first callback's generation diff --git a/backend/packages/harness/deerflow/runtime/journal.py b/backend/packages/harness/deerflow/runtime/journal.py index fb49a5ffdd..4fb4dfaba5 100644 --- a/backend/packages/harness/deerflow/runtime/journal.py +++ b/backend/packages/harness/deerflow/runtime/journal.py @@ -299,6 +299,7 @@ class RunJournal(BaseCallbackHandler): # Convenience fields self._last_ai_msg: str | None = None self._first_human_msg: str | None = None + self._first_human_message_captured = False self._msg_count = 0 self._had_llm_error_fallback = False self._llm_error_fallback_message: str | None = None @@ -437,7 +438,7 @@ class RunJournal(BaseCallbackHandler): if caller == "lead_agent": for batch in messages: self._reconcile_consumed_tool_messages(batch) - if caller == "lead_agent" and not self._first_human_msg and messages: + if caller == "lead_agent" and not self._first_human_message_captured and messages: for batch in reversed(messages): for m in reversed(batch): if _should_persist_human_input_message(m): @@ -451,7 +452,7 @@ class RunJournal(BaseCallbackHandler): ) self._record_message_summary(persisted_message, caller=caller) break - if self._first_human_msg: + if self._first_human_message_captured: break def on_llm_start(self, serialized: dict, prompts: list[str], *, run_id: UUID, parent_run_id: UUID | None = None, tags: list[str] | None = None, metadata: dict[str, Any] | None = None, **kwargs: Any) -> None: @@ -1065,6 +1066,8 @@ class RunJournal(BaseCallbackHandler): def set_first_human_message(self, content: str) -> None: """Record the first human message for convenience fields.""" + # Media-only input is still captured even when it has no display text. + self._first_human_message_captured = True self._first_human_msg = content[:2000] if content else None def record_middleware(self, tag: str, *, name: str, hook: str, action: str, changes: dict) -> None: diff --git a/backend/tests/test_run_journal.py b/backend/tests/test_run_journal.py index 91ba063d5e..c708bc0350 100644 --- a/backend/tests/test_run_journal.py +++ b/backend/tests/test_run_journal.py @@ -2217,6 +2217,47 @@ class TestChatModelStartHumanMessage: human_events = [e for e in events if e["event_type"] == "llm.human.input"] assert len(human_events) == 1 + @pytest.mark.anyio + @pytest.mark.parametrize( + "content", + [ + pytest.param("", id="empty-text"), + pytest.param([{"type": "image_url", "image_url": {"url": "data:image/png;base64,aW1hZ2U="}}], id="image-only"), + ], + ) + async def test_textless_human_input_is_captured_once(self, journal_setup, content): + """A missing display summary must not duplicate input on later model calls.""" + j, store = journal_setup + message = HumanMessage(content=content, id="user-image") + for _ in range(2): + j.on_chat_model_start({}, [[message]], run_id=uuid4(), tags=["lead_agent"]) + await j.flush() + + events = await store.list_events("t1", "r1") + human_events = [event for event in events if event["event_type"] == "llm.human.input"] + assert len(human_events) == 1 + assert human_events[0]["content"]["content"] == content + assert j.get_completion_data()["message_count"] == 1 + assert j.get_completion_data()["first_human_message"] is None + + @pytest.mark.anyio + async def test_textless_input_stops_capture_before_older_batches(self, journal_setup): + j, store = journal_setup + image = HumanMessage(content=[{"type": "image_url", "image_url": {"url": "data:image/png;base64,aW1hZ2U="}}], id="latest-input") + j.on_chat_model_start( + {}, + [[HumanMessage(content="Older question", id="old-input")], [image]], + run_id=uuid4(), + tags=["lead_agent"], + ) + await j.flush() + + events = await store.list_events("t1", "r1") + human_events = [event for event in events if event["event_type"] == "llm.human.input"] + assert [event["content"]["id"] for event in human_events] == ["latest-input"] + assert j.get_completion_data()["first_human_message"] is None + assert j.get_completion_data()["message_count"] == 1 + @pytest.mark.anyio async def test_empty_messages_no_crash(self, journal_setup): """on_chat_model_start with empty messages does not crash.""" @@ -2226,6 +2267,46 @@ class TestChatModelStartHumanMessage: assert j._first_human_msg is None +@pytest.mark.anyio +@pytest.mark.parametrize("mode", ["full", "delta"]) +async def test_image_input_is_journaled_once_across_graph_model_calls(mode): + from langchain_core.language_models.fake_chat_models import FakeMessagesListChatModel + from langgraph.checkpoint.memory import InMemorySaver + from langgraph.graph import StateGraph + + from deerflow.agents.thread_state import get_thread_state_schema + from deerflow.runtime.checkpoint_state import CheckpointStateAccessor + + store = MemoryRunEventStore() + journal = RunJournal("image-run", "image-thread", store) + model = FakeMessagesListChatModel(responses=[AIMessage(content="First response", id="a1"), AIMessage(content="Final response", id="a2")]) + + async def answer(state): + return {"messages": [await model.ainvoke(state["messages"])]} + + builder = StateGraph(get_thread_state_schema(mode)) + builder.add_node("first", answer) + builder.add_node("second", answer) + builder.set_entry_point("first") + builder.add_edge("first", "second") + builder.set_finish_point("second") + saver = InMemorySaver() + graph = builder.compile(checkpointer=saver) + accessor = CheckpointStateAccessor.bind(graph, saver, mode=mode) + image_content = [{"type": "image_url", "image_url": {"url": "data:image/png;base64,aW1hZ2U="}}] + config = {"configurable": {"thread_id": "image-thread"}, "callbacks": [journal]} + + await graph.ainvoke({"messages": [HumanMessage(content=image_content, id="h1")]}, config) + await journal.flush() + events = await store.list_messages("image-thread") + assert [event["content"]["id"] for event in events] == ["h1", "a1", "a2"] + assert events[0]["content"]["content"] == image_content + assert journal.get_completion_data()["message_count"] == 3 + snapshot = await accessor.aget(config) + assert [message.id for message in snapshot.values["messages"]] == ["h1", "a1", "a2"] + await journal.close() + + class TestDeliveryTracking: """Slice 1 (#4272): journal records artifact production for run.delivery."""