opensource/dify
2
0
Fork 0
mirror of https://github.com/langgenius/dify.git synced 2026-09-05 08:48:10 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(dify-agent): persist interrupted run history (#40972)

Browse Source
This commit is contained in:
盐粒 Yanli 2026-08-19 10:35:13 +00:00 committed by GitHub
parent 356e3a8ab6
commit 930f4e6c26
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 235 additions and 140 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
dify-agent/docs/dify-agent/guide/index.md
View File

@ -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

33
dify-agent/docs/dify-agent/user-manual/history-layer/index.md
View File

@ -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. |

9
dify-agent/docs/dify-agent/user-manual/plugin-llm-layer/index.md
View File

@ -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

12
dify-agent/src/dify_agent/runtime/history.py
View File

@ -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",
]

39
dify-agent/src/dify_agent/runtime/runner.py
View File

@ -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(

265
dify-agent/tests/local/dify_agent/runtime/test_runner.py
View File

@ -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:

6
dify-agent/tests/local/dify_agent/runtime/test_runtime_history.py
View File

@ -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