diff --git a/dify-agent/docs/dify-agent/guide/index.md b/dify-agent/docs/dify-agent/guide/index.md index 68a69772984..3090dd6e0cb 100644 --- a/dify-agent/docs/dify-agent/guide/index.md +++ b/dify-agent/docs/dify-agent/guide/index.md @@ -341,11 +341,12 @@ whose Agenton layers provide user input. With the MVP provider set, use effective prompts are rejected during create-run validation before the run is persisted or scheduled. -There is no Pydantic AI history layer. To resume Agenton layer state, pass the -`session_snapshot` from a previous terminal event together with a composition -that has the same layer names and order. Success always contains a snapshot. -Failure and cancellation contain one only when compositor entry succeeded and -layer exit completed; otherwise callers should retain their previous snapshot. +The optional Pydantic AI history layer uses the reserved name `history` and +persists captured messages in session snapshots for later resume. Resume from a +terminal event's `session_snapshot` using the same layer composition, names, and +order. Success always contains a snapshot. Failure and cancellation contain one +only when compositor entry succeeded and layer exit completed; otherwise callers +should retain their previous snapshot. ## Observing runs diff --git a/dify-agent/docs/dify-agent/user-manual/history-layer/index.md b/dify-agent/docs/dify-agent/user-manual/history-layer/index.md index 9bda3401d58..4a8550c8636 100644 --- a/dify-agent/docs/dify-agent/user-manual/history-layer/index.md +++ b/dify-agent/docs/dify-agent/user-manual/history-layer/index.md @@ -47,15 +47,22 @@ tool-call/result pairs and their inputs. If the history is still over target, th same current model incrementally summarizes older messages while retaining the latest twenty messages and the first user message. -With a history layer, a successful run replaces its stored messages with the -rewritten complete history in the returned session snapshot. Without this layer, -compaction affects only the current run. Failed runs do not write a resumable -success snapshot, so their history rewrites do not persist across runs. +With a history layer, once pydantic-ai binds and builds messages in the run +capture, the captured, possibly rewritten history replaces the stored messages +in the terminal session snapshot. This applies to successful, failed, and +cancelled runs. A failure or cancellation before the capture contains any +messages preserves the previously restored history. An interrupted capture can +include a partial response or tool-return request marked `state="interrupted"`; +pydantic-ai repairs that state when the snapshot is used by a later independent +run. Without this layer, compaction and interrupted messages affect only the +current run. ## Resume a conversation Successful runs return a terminal event with both final output and a resumable -session snapshot: +session snapshot. Failed and cancelled terminal events can also carry a session +snapshot that checkpoints current history, but they do not change the interrupted +run's terminal status into success. ```python {test="skip" lint="skip"} accepted = await client.create_run(request) @@ -87,10 +94,16 @@ Dify Agent handles memory conservatively: 2. Stored history is sent to the model before the current user prompt. 3. When the LLM layer includes `context_window_tokens`, Harness may rewrite over-target history immediately before a model request as described above. -4. After a successful run, the complete possibly compacted history is written - back to the layer. -5. Run-level system instructions are removed before history is persisted. -6. Failed runs emit `run_failed` and do not return a success snapshot to resume. +4. Once pydantic-ai binds and builds messages in the run capture, the complete + captured and possibly compacted history is written back to the layer on + success, failure, timeout, or cancellation. +5. If failure or cancellation occurs before the capture contains any messages, + the previously restored history remains unchanged. +6. Run-level system instructions are removed before history is persisted. +7. Interrupted partial messages retain pydantic-ai's `state="interrupted"` marker + so a later independent run can repair and continue from the checkpoint. +8. Failed and cancelled runs keep their terminal status; their snapshot is a + checkpoint, not a successful continuation of the interrupted run. ## Persist snapshots outside the client process @@ -118,5 +131,5 @@ Always restore snapshots with the same layer names and order that produced them. | --- | --- | | `must use reserved layer name 'history'` | Rename the layer to `history`. | | `does not support dependencies` | Remove `deps` from the history layer. | -| Resume fails with snapshot lifecycle errors | Use the success snapshot from `run_succeeded` and keep layer names/order unchanged. | +| Resume fails with snapshot lifecycle errors | Use a terminal snapshot whose layers were suspended, and keep layer names/order unchanged. | | System prompts appear missing from saved memory | This is expected; current system prompts are temporary and are not persisted. | diff --git a/dify-agent/docs/dify-agent/user-manual/plugin-llm-layer/index.md b/dify-agent/docs/dify-agent/user-manual/plugin-llm-layer/index.md index 4fe7f4c0849..8b8c970e9e8 100644 --- a/dify-agent/docs/dify-agent/user-manual/plugin-llm-layer/index.md +++ b/dify-agent/docs/dify-agent/user-manual/plugin-llm-layer/index.md @@ -72,8 +72,13 @@ model incrementally summarizes older history while retaining the latest twenty messages and the first user message. Compaction affects later runs only when the composition has a -[history layer](../history-layer/index.md) and a successful run writes the -rewritten history into its session snapshot. +[history layer](../history-layer/index.md). Once pydantic-ai binds and builds +messages in the run capture, successful, failed, timed-out, and cancelled runs +write the captured rewritten history into their terminal session snapshot. A +failure or cancellation before the capture contains any messages preserves the +previously restored history. Interrupted partial messages may be included and +repaired when that checkpoint is used by a later independent run; the interrupted +run's terminal status remains unchanged. ## Complete minimal model composition diff --git a/dify-agent/src/dify_agent/runtime/history.py b/dify-agent/src/dify_agent/runtime/history.py index 8b2d4d11e73..872f76a348f 100644 --- a/dify-agent/src/dify_agent/runtime/history.py +++ b/dify-agent/src/dify_agent/runtime/history.py @@ -2,8 +2,10 @@ Dify Agent keeps pydantic-ai conversation history as an optional Agenton layer named ``history``. Current system instructions belong to each run and are never -stored; successful runs replace the layer with Pydantic AI's complete, possibly -compacted history. +stored. Once Pydantic AI binds and builds messages in the run capture, its +complete captured history replaces the layer for every terminal outcome, +including interrupted runs. A failure or cancellation before the capture +contains messages preserves the previously restored history. """ from __future__ import annotations @@ -63,11 +65,11 @@ def get_history_layer(run: SupportsHistoryLayerLookup) -> PydanticAIHistoryLayer return None -def replace_successful_run_history( +def replace_run_history( history_layer: PydanticAIHistoryLayer | None, messages: Sequence[ModelMessage], ) -> None: - """Persist a successful run's complete history without transient instructions.""" + """Persist a run's captured history without transient instructions.""" if history_layer is None: return persistent_messages = [ @@ -79,6 +81,6 @@ def replace_successful_run_history( __all__ = [ "SupportsHistoryLayerLookup", "get_history_layer", - "replace_successful_run_history", + "replace_run_history", "validate_history_layer_composition", ] diff --git a/dify-agent/src/dify_agent/runtime/runner.py b/dify-agent/src/dify_agent/runtime/runner.py index c295c6607ff..b7dcd127df1 100644 --- a/dify-agent/src/dify_agent/runtime/runner.py +++ b/dify-agent/src/dify_agent/runtime/runner.py @@ -11,10 +11,12 @@ policy is validated: request-level ``on_exit`` signals, and publish a terminal success or failure event; The Pydantic AI model is resolved from the active Agenton layer named by ``DIFY_AGENT_MODEL_LAYER_ID``. An optional history layer contributes stored -message history only through session state; successful model runs replace that -state with ``result.all_messages()`` after transient instructions are cleared so -compaction rewrites persist without saving current system prompts. An optional -structured output layer named by +message history only through session state. Once pydantic-ai binds and builds +messages in the run capture, every terminal outcome replaces that state with the +captured messages after transient instructions are cleared; a failure or +cancellation before the capture contains messages preserves the restored state. +This preserves compaction rewrites and interrupted partial messages without +saving current system prompts. An optional structured output layer named by ``DIFY_AGENT_OUTPUT_LAYER_ID`` is read after entry and resolved into an output contract whose type both exposes the output schema to the model and performs runtime JSON Schema validation through custom Pydantic hooks. When the ask-human @@ -37,6 +39,7 @@ from typing import Any, Literal, Protocol, cast, runtime_checkable import httpx from graphon.model_runtime.entities.llm_entities import LLMUsage from pydantic import JsonValue, TypeAdapter +from pydantic_ai import capture_run_messages from pydantic_ai.exceptions import ModelHTTPError, UsageLimitExceeded from pydantic_ai.messages import AgentStreamEvent, PartDeltaEvent, PartStartEvent, TextPart, TextPartDelta from pydantic_ai.output import OutputSpec @@ -73,7 +76,7 @@ from dify_agent.runtime.event_sink import ( ) from dify_agent.runtime.history import ( get_history_layer, - replace_successful_run_history, + replace_run_history, validate_history_layer_composition, ) from dify_agent.runtime.layer_exit_signals import apply_layer_exit_signals, validate_layer_exit_signals @@ -362,16 +365,21 @@ class AgentRunRunner: ) run_timeout = asyncio.timeout(self.run_timeout_seconds) try: - async with run_timeout: - result = await agent.run( - None if deferred_tool_results is not None else normalize_user_input(user_prompts), - message_history=message_history, - deferred_tool_results=deferred_tool_results, - event_stream_handler=handle_events, - instructions=run.prompts or None, - capabilities=[compaction] if compaction is not None else None, - usage_limits=UsageLimits(request_limit=_MAX_AGENT_STEPS_PER_RUN), - ) + with capture_run_messages() as captured_messages: + try: + async with run_timeout: + result = await agent.run( + None if deferred_tool_results is not None else normalize_user_input(user_prompts), + message_history=message_history, + deferred_tool_results=deferred_tool_results, + event_stream_handler=handle_events, + instructions=run.prompts or None, + capabilities=[compaction] if compaction is not None else None, + usage_limits=UsageLimits(request_limit=_MAX_AGENT_STEPS_PER_RUN), + ) + finally: + if captured_messages: + replace_run_history(history_layer, captured_messages) except TimeoutError as exc: if not run_timeout.expired(): raise @@ -381,7 +389,6 @@ class AgentRunRunner: complete_usage = model.accumulated_usage if isinstance(model, _HasAccumulatedUsage) else None usage = _serialize_agent_usage(complete_usage if complete_usage is not None else _result_usage(result)) self._terminal_usage = usage - replace_successful_run_history(history_layer, result.all_messages()) if isinstance(result.output, DeferredToolRequests): if ask_human_layer is None: raise AgentRunValidationError( diff --git a/dify-agent/tests/local/dify_agent/runtime/test_runner.py b/dify-agent/tests/local/dify_agent/runtime/test_runner.py index edd5adcc74a..86f44a9b648 100644 --- a/dify-agent/tests/local/dify_agent/runtime/test_runner.py +++ b/dify-agent/tests/local/dify_agent/runtime/test_runner.py @@ -1,5 +1,6 @@ import asyncio -from collections.abc import Iterable, Mapping +from collections.abc import AsyncIterator, Generator, Iterable, Mapping +from contextlib import contextmanager from decimal import Decimal from typing import Any, ClassVar, cast @@ -20,6 +21,7 @@ from pydantic_ai.messages import ( UserPromptPart, ) from pydantic_ai.models import ModelRequestParameters +from pydantic_ai.models.function import FunctionModel from pydantic_ai.models.test import TestModel from pydantic_ai.tools import DeferredToolRequests, DeferredToolResults from pydantic_ai.usage import UsageLimits @@ -488,16 +490,45 @@ def _flatten_message_parts(messages: list[ModelMessage]) -> list[object]: return [part for message in messages for part in message.parts] +def _assert_interrupted_history( + snapshot: CompositorSessionSnapshot, + stored_history: list[ModelMessage], +) -> None: + saved_history = _history_messages_from_snapshot(snapshot) + + assert saved_history[: len(stored_history)] == stored_history + assert len(saved_history) == len(stored_history) + 2 + current_request = saved_history[-2] + assert isinstance(current_request, ModelRequest) + assert current_request.instructions is None + assert len(current_request.parts) == 1 + assert isinstance(current_request.parts[0], UserPromptPart) + assert current_request.parts[0].content == "current user" + partial_response = saved_history[-1] + assert isinstance(partial_response, ModelResponse) + assert partial_response.state == "interrupted" + assert len(partial_response.parts) == 1 + assert isinstance(partial_response.parts[0], TextPart) + assert partial_response.parts[0].content == "partial" + + +def _install_fake_message_capture(monkeypatch: pytest.MonkeyPatch) -> list[ModelMessage]: + captured_messages: list[ModelMessage] = [] + + @contextmanager + def fake_capture_run_messages() -> Generator[list[ModelMessage]]: + captured_messages.clear() + yield captured_messages + + monkeypatch.setattr("dify_agent.runtime.runner.capture_run_messages", fake_capture_run_messages) + return captured_messages + + class FakeAgentRunResult: output: object - _all_messages: list[ModelMessage] - def __init__(self, output: object, all_messages: list[ModelMessage]) -> None: + def __init__(self, output: object) -> None: self.output = output - self._all_messages = all_messages - - def all_messages(self) -> list[ModelMessage]: - return list(self._all_messages) def test_runner_emits_terminal_success_and_snapshot(monkeypatch: pytest.MonkeyPatch) -> None: @@ -616,7 +647,7 @@ def test_runner_preserves_explicit_json_null_output(monkeypatch: pytest.MonkeyPa class FakeAgent: async def run(self, *_args: object, **_kwargs: object) -> FakeAgentRunResult: - return FakeAgentRunResult(None, []) + return FakeAgentRunResult(None) monkeypatch.setattr(DifyPluginLLMLayer, "get_model", fake_get_model) monkeypatch.setattr("dify_agent.runtime.runner.create_agent", lambda *_args, **_kwargs: FakeAgent()) @@ -651,7 +682,7 @@ def test_runner_passes_explicit_step_limit_to_agent(monkeypatch: pytest.MonkeyPa async def run(self, *_args: object, **kwargs: object) -> FakeAgentRunResult: usage_limits = cast(UsageLimits, kwargs["usage_limits"]) assert usage_limits.request_limit == 500 - return FakeAgentRunResult("done", []) + return FakeAgentRunResult("done") monkeypatch.setattr(DifyPluginLLMLayer, "get_model", fake_get_model) monkeypatch.setattr("dify_agent.runtime.runner.create_agent", lambda *_args, **_kwargs: FakeAgent()) @@ -684,7 +715,7 @@ def test_runner_passes_context_compaction(monkeypatch: pytest.MonkeyPatch) -> No capability = capabilities[0] assert isinstance(capability, TieredCompaction) assert capability.target_tokens == 7_000 - return FakeAgentRunResult("done", []) + return FakeAgentRunResult("done") monkeypatch.setattr(DifyPluginLLMLayer, "get_model", fake_get_model) monkeypatch.setattr("dify_agent.runtime.runner.create_agent", lambda *_args, **_kwargs: FakeAgent()) @@ -724,7 +755,7 @@ def test_runner_rejects_compaction_budget_before_model_resolution_or_invocation( async def run(self, *_args: object, **_kwargs: object) -> FakeAgentRunResult: nonlocal model_invocation_called model_invocation_called = True - return FakeAgentRunResult("unused", []) + return FakeAgentRunResult("unused") def fake_create_agent(*_args: object, **_kwargs: object) -> FakeAgent: nonlocal agent_creation_called @@ -787,7 +818,7 @@ def test_runner_timeout_excludes_tool_preparation_and_runtime_cleanup(monkeypatc class ImmediateAgent: async def run(self, *_args: object, **_kwargs: object) -> FakeAgentRunResult: - return FakeAgentRunResult("done", []) + return FakeAgentRunResult("done") async def slow_resolve_run_tools( _run: object, @@ -935,32 +966,37 @@ def test_runner_does_not_classify_nested_timeout_as_agent_limit(monkeypatch: pyt assert sink.statuses["run-provider-timeout"] == "failed" -def test_runner_captures_post_exit_snapshot_when_task_is_cancelled(monkeypatch: pytest.MonkeyPatch) -> None: - started = asyncio.Event() +def test_runner_captures_interrupted_history_when_task_is_cancelled(monkeypatch: pytest.MonkeyPatch) -> None: + partial_streamed = asyncio.Event() + stored_history = [ + ModelRequest(parts=[UserPromptPart(content="old user")]), + ModelResponse(parts=[TextPart(content="old assistant")]), + ] + + async def stream_response(_messages: list[ModelMessage], _info: object) -> AsyncIterator[str]: + yield "partial" + _ = partial_streamed.set() + await asyncio.Event().wait() def fake_get_model(_self: DifyPluginLLMLayer, *, http_client: httpx.AsyncClient, agent_run_id: str): - return TestModel(custom_output_text="unused") # pyright: ignore[reportReturnType] - - class FakeAgent: - async def run(self, *_args: object, **_kwargs: object) -> None: - started.set() - await asyncio.Event().wait() + return FunctionModel(stream_function=stream_response) monkeypatch.setattr(DifyPluginLLMLayer, "get_model", fake_get_model) - monkeypatch.setattr("dify_agent.runtime.runner.create_agent", lambda *_args, **_kwargs: FakeAgent()) + request = _request("current user", include_history=True) + request.session_snapshot = _history_session_snapshot(stored_history) sink = InMemoryRunEventSink() async def scenario() -> AgentRunRunner: async with httpx.AsyncClient() as client: runner = AgentRunRunner( sink=sink, - request=_request(), - run_id="run-cancel-snapshot", + request=request, + run_id="run-cancel-history", plugin_daemon_http_client=client, dify_api_http_client=client, ) task = asyncio.create_task(runner.run()) - await asyncio.wait_for(started.wait(), timeout=1) + await asyncio.wait_for(partial_streamed.wait(), timeout=1) task.cancel() with pytest.raises(asyncio.CancelledError): await task @@ -970,12 +1006,13 @@ def test_runner_captures_post_exit_snapshot_when_task_is_cancelled(monkeypatch: assert runner.terminal_session_snapshot is not None assert all(layer.lifecycle_state is LifecycleState.SUSPENDED for layer in runner.terminal_session_snapshot.layers) - assert [event.type for event in sink.events["run-cancel-snapshot"]] == ["run_started"] + _assert_interrupted_history(runner.terminal_session_snapshot, stored_history) def test_runner_emits_deferred_tool_call_and_persists_pending_history(monkeypatch: pytest.MonkeyPatch) -> None: captured_output_types: list[object] = [] captured_user_prompts: list[object] = [] + captured_messages = _install_fake_message_capture(monkeypatch) pending_tool_call = ToolCallPart( tool_name="ask_human", args={ @@ -993,13 +1030,12 @@ def test_runner_emits_deferred_tool_call_and_persists_pending_history(monkeypatc async def run(self, user_prompt: object, **kwargs: object) -> FakeAgentRunResult: captured_user_prompts.append(user_prompt) assert kwargs["deferred_tool_results"] is None - return FakeAgentRunResult( - DeferredToolRequests(calls=[pending_tool_call]), - [ - ModelRequest(parts=[UserPromptPart(content="current user")]), - ModelResponse(parts=[pending_tool_call]), - ], - ) + messages: list[ModelMessage] = [ + ModelRequest(parts=[UserPromptPart(content="current user")]), + ModelResponse(parts=[pending_tool_call]), + ] + captured_messages.extend(messages) + return FakeAgentRunResult(DeferredToolRequests(calls=[pending_tool_call])) def fake_create_agent(model: object, *, tools: list[Tool[object]], output_type: object) -> FakeAgent: del model, tools @@ -1056,6 +1092,7 @@ def test_runner_emits_deferred_tool_call_and_persists_pending_history(monkeypatc def test_runner_resumes_with_deferred_tool_results_and_no_user_prompt(monkeypatch: pytest.MonkeyPatch) -> None: seen_user_prompts: list[object] = [] seen_deferred_results: list[object] = [] + captured_messages = _install_fake_message_capture(monkeypatch) pending_tool_call = ToolCallPart( tool_name="ask_human", args={"question": "Need approval"}, @@ -1071,35 +1108,33 @@ def test_runner_resumes_with_deferred_tool_results_and_no_user_prompt(monkeypatc seen_user_prompts.append(user_prompt) seen_deferred_results.append(kwargs.get("deferred_tool_results")) if kwargs.get("deferred_tool_results") is None: - return FakeAgentRunResult( - DeferredToolRequests(calls=[pending_tool_call]), - [ - ModelRequest(parts=[UserPromptPart(content="current user")]), - ModelResponse(parts=[pending_tool_call]), - ], - ) + messages: list[ModelMessage] = [ + ModelRequest(parts=[UserPromptPart(content="current user")]), + ModelResponse(parts=[pending_tool_call]), + ] + captured_messages.extend(messages) + return FakeAgentRunResult(DeferredToolRequests(calls=[pending_tool_call])) deferred_tool_results = cast(DeferredToolResults, kwargs["deferred_tool_results"]) assert deferred_tool_results is not None submitted_result = cast(dict[str, object], deferred_tool_results.calls["tool-call-1"]) assert submitted_result["status"] == "submitted" message_history = cast(list[ModelMessage], kwargs["message_history"]) - return FakeAgentRunResult( - "done after human", - [ - *message_history, - ModelRequest( - parts=[ - ToolReturnPart( - tool_name="ask_human", - content={"status": "submitted", "values": {"comment": "Ship it"}}, - tool_call_id="tool-call-1", - ) - ] - ), - ModelResponse(parts=[TextPart(content="done after human")]), - ], - ) + messages = [ + *message_history, + ModelRequest( + parts=[ + ToolReturnPart( + tool_name="ask_human", + content={"status": "submitted", "values": {"comment": "Ship it"}}, + tool_call_id="tool-call-1", + ) + ] + ), + ModelResponse(parts=[TextPart(content="done after human")]), + ] + captured_messages.extend(messages) + return FakeAgentRunResult("done after human") def fake_create_agent(model: object, *, tools: list[Tool[object]], output_type: object) -> FakeAgent: del model, tools, output_type @@ -1158,6 +1193,7 @@ def test_runner_resumes_with_deferred_tool_results_and_no_user_prompt(monkeypatc def test_runner_can_emit_second_deferred_tool_call_after_resume(monkeypatch: pytest.MonkeyPatch) -> None: seen_user_prompts: list[object] = [] + captured_messages = _install_fake_message_capture(monkeypatch) first_pending_tool_call = ToolCallPart( tool_name="ask_human", args={"question": "Need deployment owner"}, @@ -1178,31 +1214,29 @@ def test_runner_can_emit_second_deferred_tool_call_after_resume(monkeypatch: pyt seen_user_prompts.append(user_prompt) deferred_tool_results = kwargs.get("deferred_tool_results") if deferred_tool_results is None: - return FakeAgentRunResult( - DeferredToolRequests(calls=[first_pending_tool_call]), - [ - ModelRequest(parts=[UserPromptPart(content="current user")]), - ModelResponse(parts=[first_pending_tool_call]), - ], - ) + messages: list[ModelMessage] = [ + ModelRequest(parts=[UserPromptPart(content="current user")]), + ModelResponse(parts=[first_pending_tool_call]), + ] + captured_messages.extend(messages) + return FakeAgentRunResult(DeferredToolRequests(calls=[first_pending_tool_call])) message_history = cast(list[ModelMessage], kwargs["message_history"]) - return FakeAgentRunResult( - DeferredToolRequests(calls=[second_pending_tool_call]), - [ - *message_history, - ModelRequest( - parts=[ - ToolReturnPart( - tool_name="ask_human", - content={"status": "submitted", "values": {"owner": "ops"}}, - tool_call_id="tool-call-1", - ) - ] - ), - ModelResponse(parts=[second_pending_tool_call]), - ], - ) + messages = [ + *message_history, + ModelRequest( + parts=[ + ToolReturnPart( + tool_name="ask_human", + content={"status": "submitted", "values": {"owner": "ops"}}, + tool_call_id="tool-call-1", + ) + ] + ), + ModelResponse(parts=[second_pending_tool_call]), + ] + captured_messages.extend(messages) + return FakeAgentRunResult(DeferredToolRequests(calls=[second_pending_tool_call])) def fake_create_agent(model: object, *, tools: list[Tool[object]], output_type: object) -> FakeAgent: del model, tools, output_type @@ -1281,8 +1315,7 @@ def test_runner_rejects_deferred_tool_call_without_history_layer(monkeypatch: py calls=[ ToolCallPart(tool_name="ask_human", args={"question": "Need owner"}, tool_call_id="tool-call-1") ] - ), - [], + ) ) monkeypatch.setattr(DifyPluginLLMLayer, "get_model", fake_get_model) @@ -1322,7 +1355,7 @@ def test_runner_rejects_resume_with_deferred_tool_results_without_history_layer( async def run(self, *_args: object, **_kwargs: object) -> FakeAgentRunResult: nonlocal agent_run_called agent_run_called = True - return FakeAgentRunResult("unexpected", []) + return FakeAgentRunResult("unexpected") monkeypatch.setattr(DifyPluginLLMLayer, "get_model", fake_get_model) monkeypatch.setattr("dify_agent.runtime.runner.create_agent", lambda *args, **kwargs: FakeAgent()) @@ -1373,8 +1406,7 @@ def test_runner_rejects_multiple_deferred_tool_calls(monkeypatch: pytest.MonkeyP ToolCallPart(tool_name="ask_human", args={"question": "One"}, tool_call_id="tool-call-1"), ToolCallPart(tool_name="ask_human", args={"question": "Two"}, tool_call_id="tool-call-2"), ] - ), - [], + ) ) monkeypatch.setattr(DifyPluginLLMLayer, "get_model", fake_get_model) @@ -1412,8 +1444,7 @@ def test_runner_rejects_deferred_approval_requests(monkeypatch: pytest.MonkeyPat tool_name="ask_human", args={"question": "Need approval"}, tool_call_id="tool-call-1" ) ] - ), - [], + ) ) monkeypatch.setattr(DifyPluginLLMLayer, "get_model", fake_get_model) @@ -1461,9 +1492,6 @@ def test_runner_passes_dynamic_dify_plugin_tools_to_agent(monkeypatch: pytest.Mo class FakeResult: output: str = "done" - def all_messages(self) -> list[ModelMessage]: - return [] - class FakeAgent: async def run(self, *_args: object, **_kwargs: object) -> FakeResult: return FakeResult() @@ -1563,9 +1591,6 @@ def test_runner_passes_dynamic_dify_knowledge_tools_to_agent(monkeypatch: pytest class FakeResult: output: str = "done" - def all_messages(self) -> list[ModelMessage]: - return [] - class FakeAgent: async def run(self, *_args: object, **_kwargs: object) -> FakeResult: return FakeResult() @@ -1669,9 +1694,6 @@ def test_runner_passes_dynamic_dify_core_tools_to_agent(monkeypatch: pytest.Monk class FakeResult: output: str = "done" - def all_messages(self) -> list[ModelMessage]: - return [] - class FakeAgent: async def run(self, *_args: object, **_kwargs: object) -> FakeResult: return FakeResult() @@ -2255,18 +2277,21 @@ def test_runner_with_empty_history_layer_uses_instructions_and_saves_full_histor assert all(not isinstance(message, ModelRequest) or message.instructions is None for message in saved_history) -def test_runner_failure_with_history_layer_emits_post_exit_snapshot_without_new_history( +def test_runner_failure_with_history_layer_captures_interrupted_history( monkeypatch: pytest.MonkeyPatch, ) -> None: - model = RecordingTestModel(failure=RuntimeError("boom")) stored_history = [ ModelRequest(parts=[UserPromptPart(content="old user")]), ModelResponse(parts=[TextPart(content="old assistant")]), ] + async def stream_response(_messages: list[ModelMessage], _info: object) -> AsyncIterator[str]: + yield "partial" + raise RuntimeError("boom") + def fake_get_model(_self: DifyPluginLLMLayer, *, http_client: httpx.AsyncClient, agent_run_id: str): assert http_client.is_closed is False - return model # pyright: ignore[reportReturnType] + return FunctionModel(stream_function=stream_response) monkeypatch.setattr(DifyPluginLLMLayer, "get_model", fake_get_model) request = _request("current user", include_history=True) @@ -2286,16 +2311,58 @@ def test_runner_failure_with_history_layer_emits_post_exit_snapshot_without_new_ asyncio.run(scenario()) - assert [event.type for event in sink.events["run-history-failure"]] == ["run_started", "run_failed"] + event_types = [event.type for event in sink.events["run-history-failure"]] + assert event_types[0] == "run_started" assert sink.statuses["run-history-failure"] == "failed" terminal = sink.events["run-history-failure"][-1] assert isinstance(terminal, RunFailedEvent) assert terminal.data.session_snapshot is not None - assert _history_messages_from_snapshot(terminal.data.session_snapshot) == stored_history + _assert_interrupted_history(terminal.data.session_snapshot, stored_history) assert request.session_snapshot is not None assert _history_messages_from_snapshot(request.session_snapshot) == stored_history +def test_runner_preserves_history_when_agent_fails_before_capture_is_bound( + monkeypatch: pytest.MonkeyPatch, +) -> None: + stored_history = [ + ModelRequest(parts=[UserPromptPart(content="old user")]), + ModelResponse(parts=[TextPart(content="old assistant")]), + ] + + def fake_get_model(_self: DifyPluginLLMLayer, *, http_client: httpx.AsyncClient, agent_run_id: str): + assert http_client.is_closed is False + return TestModel(custom_output_text="unused") # pyright: ignore[reportReturnType] + + class FakeAgent: + async def run(self, *_args: object, **_kwargs: object) -> None: + raise RuntimeError("boom before capture") + + monkeypatch.setattr(DifyPluginLLMLayer, "get_model", fake_get_model) + monkeypatch.setattr("dify_agent.runtime.runner.create_agent", lambda *_args, **_kwargs: FakeAgent()) + request = _request("current user", include_history=True) + request.session_snapshot = _history_session_snapshot(stored_history) + sink = InMemoryRunEventSink() + + async def scenario() -> None: + async with httpx.AsyncClient() as client: + with pytest.raises(RuntimeError, match="boom before capture"): + await AgentRunRunner( + sink=sink, + request=request, + run_id="run-history-empty-capture", + plugin_daemon_http_client=client, + dify_api_http_client=client, + ).run() + + asyncio.run(scenario()) + + terminal = sink.events["run-history-empty-capture"][-1] + assert isinstance(terminal, RunFailedEvent) + assert terminal.data.session_snapshot is not None + assert _history_messages_from_snapshot(terminal.data.session_snapshot) == stored_history + + def test_runner_persists_usage_limit_failure_type_in_event_and_status( monkeypatch: pytest.MonkeyPatch, ) -> None: diff --git a/dify-agent/tests/local/dify_agent/runtime/test_runtime_history.py b/dify-agent/tests/local/dify_agent/runtime/test_runtime_history.py index 3072a0c6a1f..be77124bfee 100644 --- a/dify-agent/tests/local/dify_agent/runtime/test_runtime_history.py +++ b/dify-agent/tests/local/dify_agent/runtime/test_runtime_history.py @@ -13,7 +13,7 @@ from dify_agent.protocol.schemas import RunComposition, RunLayerSpec from dify_agent.runtime.compositor_factory import create_default_layer_providers from dify_agent.runtime.history import ( get_history_layer, - replace_successful_run_history, + replace_run_history, validate_history_layer_composition, ) @@ -88,7 +88,7 @@ def test_get_history_layer_returns_optional_active_history_layer() -> None: asyncio.run(scenario()) -def test_replace_successful_run_history_persists_full_history_without_instructions() -> None: +def test_replace_run_history_persists_full_history_without_instructions() -> None: history_layer = PydanticAIHistoryLayer() history_layer.replace_messages([ModelRequest(parts=[UserPromptPart(content="stale")])]) messages = [ @@ -100,7 +100,7 @@ def test_replace_successful_run_history_persists_full_history_without_instructio ModelResponse(parts=[TextPart(content="new assistant")]), ] - replace_successful_run_history(history_layer, messages) + replace_run_history(history_layer, messages) persisted = history_layer.message_history assert len(persisted) == 3