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

test: use SQLite sessions in core app apps (#39101)

Browse Source
This commit is contained in:
Asuka Minato 2026-07-30 12:18:09 +09:00 committed by GitHub
parent 7c9a0ce93b
commit 3f1036e55a
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 128 additions and 109 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

237
api/tests/unit_tests/core/app/apps/agent_app/test_app_runner.py
View File

@ -7,7 +7,6 @@ from __future__ import annotations
from collections.abc import Callable, Iterator
from datetime import UTC, datetime
from decimal import Decimal
from types import SimpleNamespace
from typing import Any, override
from unittest.mock import MagicMock
@ -33,6 +32,8 @@ from pydantic_ai.messages import (
ToolCallPart,
ToolReturnPart,
)
from sqlalchemy import event, select
from sqlalchemy.orm import Session
from clients.agent_backend import (
AgentBackendRunEventAdapter,
@ -47,7 +48,7 @@ from core.app.apps.agent_app.app_runner import AgentAppRunner
from core.app.apps.agent_app.runtime_request_builder import AgentAppRuntimeRequestBuilder
from core.app.apps.agent_app.session_store import AgentAppSessionScope, StoredAgentAppSession
from core.app.apps.exc import GenerateTaskStoppedError
from core.app.entities.app_invoke_entities import InvokeFrom, UserFrom
from core.app.entities.app_invoke_entities import DifyRunContext, InvokeFrom, UserFrom
from core.app.entities.queue_entities import (
QueueAgentMessageEvent,
QueueAgentThoughtEvent,
@ -55,20 +56,33 @@ from core.app.entities.queue_entities import (
QueueMessageEndEvent,
)
from core.workflow.nodes.agent_v2.ask_human_resume import AskHumanResumeOutcome
from core.workflow.nodes.agent_v2.dify_tools_builder import WorkflowAgentToolLayers
from graphon.model_runtime.entities.llm_entities import LLMResult
from graphon.model_runtime.errors.invoke import InvokeRateLimitError
from models.agent_config_entities import AgentSoulConfig
from models.model import MessageAgentThought
@pytest.fixture(autouse=True)
def bind_agent_db(monkeypatch: pytest.MonkeyPatch, sqlite_session: Session) -> None:
"""Bind the runner's ORM writes to the shared SQLite session."""
monkeypatch.setattr(app_runner_module.db, "session", sqlite_session)
def _thought_rows(session: Session) -> list[MessageAgentThought]:
session.expire_all()
return list(session.scalars(select(MessageAgentThought).order_by(MessageAgentThought.position)).all())
class _FakeCredentialsProvider:
def fetch(self, provider_name: str, model_name: str) -> dict[str, Any]:
return {"openai_api_key": "sk-test"}
class _NoToolsBuilder:
def build_layers(self, **kwargs):
def build_layers(self, **kwargs: Any) -> WorkflowAgentToolLayers:
del kwargs
return SimpleNamespace(plugin_tools=None, core_tools=None, exposed_tool_names=lambda: [])
return WorkflowAgentToolLayers()
class _FakeQueueManager:
@ -354,27 +368,6 @@ class _ProcessStreamingFakeAgentBackendRunClient(FakeAgentBackendRunClient):
)
class _FakeDbSession:
def __init__(self) -> None:
self.rows: dict[str, MessageAgentThought] = {}
self.rollback_count = 0
def add(self, row: MessageAgentThought) -> None:
self.rows[str(row.id)] = row
def commit(self) -> None:
pass
def get(self, _model: type[MessageAgentThought], row_id: str) -> MessageAgentThought | None:
return self.rows.get(row_id)
def delete(self, row: MessageAgentThought) -> None:
self.rows.pop(str(row.id), None)
def rollback(self) -> None:
self.rollback_count += 1
class _FakeSessionStore:
def __init__(
self,
@ -450,8 +443,8 @@ def _soul() -> AgentSoulConfig:
)
def _dify_ctx() -> Any:
return SimpleNamespace(
def _dify_ctx() -> DifyRunContext:
return DifyRunContext(
tenant_id="tenant-1",
app_id="app-1",
user_id="user-1",
@ -497,9 +490,14 @@ def _message_end(qm: _FakeQueueManager) -> QueueMessageEndEvent:
return next(e for e in qm.events if isinstance(e, QueueMessageEndEvent))
def _saved_user_query(qm: _FakeQueueManager) -> str:
def _llm_result(qm: _FakeQueueManager) -> LLMResult:
llm_result = _message_end(qm).llm_result
assert llm_result is not None
return llm_result
def _saved_user_query(qm: _FakeQueueManager) -> str:
llm_result = _llm_result(qm)
prompt_messages = llm_result.prompt_messages
assert len(prompt_messages) == 1
content = prompt_messages[0].content
@ -507,7 +505,7 @@ def _saved_user_query(qm: _FakeQueueManager) -> str:
return content
def test_successful_turn_publishes_chunk_and_message_end_and_saves_session():
def test_successful_turn_publishes_chunk_and_message_end_and_saves_session() -> None:
client = FakeAgentBackendRunClient() # SUCCESS: output {"text": "hello agent"}
store = _FakeSessionStore()
qm = _FakeQueueManager()
@ -522,8 +520,8 @@ def test_successful_turn_publishes_chunk_and_message_end_and_saves_session():
assert len(chunk_events) == 1
assert len(end_events) == 1
assert chunk_events[0].chunk.delta.message.content == "hello agent"
assert end_events[0].llm_result.message.content == "hello agent"
assert end_events[0].llm_result.model == "gpt-4o-mini"
assert _llm_result(qm).message.content == "hello agent"
assert _llm_result(qm).model == "gpt-4o-mini"
assert _saved_user_query(qm) == "hello"
# The conversation session snapshot is persisted for multi-turn continuity.
assert store.saved
@ -550,9 +548,9 @@ def test_turn_uses_resolved_backend_binding_before_backend_invocation() -> None:
assert len(store.resolved_scopes) == 1
def test_successful_turn_routes_stream_text_to_agent_message_and_uses_terminal_output(monkeypatch):
fake_session = _FakeDbSession()
monkeypatch.setattr(app_runner_module.db, "session", fake_session)
def test_successful_turn_routes_stream_text_to_agent_message_and_uses_terminal_output(
sqlite_session: Session,
) -> None:
client = _StreamingFakeAgentBackendRunClient()
store = _FakeSessionStore()
qm = _FakeQueueManager()
@ -565,22 +563,21 @@ def test_successful_turn_routes_stream_text_to_agent_message_and_uses_terminal_o
assert [event.chunk.delta.message.content for event in chunk_events] == ["hello agent"]
assert [event.chunk.delta.message.content for event in agent_message_events] == ["hello ", "agent"]
assert len(end_events) == 1
assert end_events[0].llm_result.message.content == "hello agent"
assert end_events[0].llm_result.usage.prompt_tokens == 3
assert end_events[0].llm_result.usage.completion_tokens == 5
assert end_events[0].llm_result.usage.total_tokens == 8
assert end_events[0].llm_result.usage.prompt_price == Decimal("0.000015")
assert end_events[0].llm_result.usage.completion_price == Decimal("0.000150")
assert end_events[0].llm_result.usage.total_price == Decimal("0.000165")
assert end_events[0].llm_result.usage.currency == "USD"
rows = sorted(fake_session.rows.values(), key=lambda row: row.position)
llm_result = _llm_result(qm)
assert llm_result.message.content == "hello agent"
assert llm_result.usage.prompt_tokens == 3
assert llm_result.usage.completion_tokens == 5
assert llm_result.usage.total_tokens == 8
assert llm_result.usage.prompt_price == Decimal("0.000015")
assert llm_result.usage.completion_price == Decimal("0.000150")
assert llm_result.usage.total_price == Decimal("0.000165")
assert llm_result.usage.currency == "USD"
rows = _thought_rows(sqlite_session)
assert rows == []
assert store.saved
def test_successful_turn_routes_single_agent_message_delta(monkeypatch):
fake_session = _FakeDbSession()
monkeypatch.setattr(app_runner_module.db, "session", fake_session)
def test_successful_turn_routes_single_agent_message_delta(sqlite_session: Session) -> None:
client = _StreamingSingleAgentMessageDeltaFakeAgentBackendRunClient()
store = _FakeSessionStore()
qm = _FakeQueueManager()
@ -593,12 +590,42 @@ def test_successful_turn_routes_single_agent_message_delta(monkeypatch):
assert [event.chunk.delta.message.content for event in chunk_events] == ["hello agent"]
assert [event.chunk.delta.message.content for event in agent_message_events] == ["hello"]
assert len(end_events) == 1
assert end_events[0].llm_result.message.content == "hello agent"
rows = sorted(fake_session.rows.values(), key=lambda row: row.position)
assert _llm_result(qm).message.content == "hello agent"
rows = _thought_rows(sqlite_session)
assert rows == []
def test_successful_turn_with_null_terminal_output_publishes_empty_answer_not_literal_null():
def test_thought_commit_failure_rolls_back_and_turn_continues(sqlite_session: Session) -> None:
rollback_events: list[Session] = []
should_fail = True
def fail_first_commit(_session: Session) -> None:
nonlocal should_fail
if should_fail:
should_fail = False
raise RuntimeError("forced thought commit failure")
def record_rollback(session: Session) -> None:
rollback_events.append(session)
event.listen(sqlite_session, "before_commit", fail_first_commit)
event.listen(sqlite_session, "after_rollback", record_rollback)
try:
client = _StreamingSingleAgentMessageDeltaFakeAgentBackendRunClient()
store = _FakeSessionStore()
qm = _FakeQueueManager()
_run(_runner(client, store), qm)
finally:
event.remove(sqlite_session, "before_commit", fail_first_commit)
event.remove(sqlite_session, "after_rollback", record_rollback)
assert rollback_events == [sqlite_session]
assert _thought_rows(sqlite_session) == []
assert _llm_result(qm).message.content == "hello agent"
def test_successful_turn_with_null_terminal_output_publishes_empty_answer_not_literal_null() -> None:
client = _NullOutputFakeAgentBackendRunClient()
store = _FakeSessionStore()
qm = _FakeQueueManager()
@ -611,12 +638,12 @@ def test_successful_turn_with_null_terminal_output_publishes_empty_answer_not_li
assert chunk_events == []
assert agent_message_events == []
assert len(end_events) == 1
assert end_events[0].llm_result.message.content == ""
assert _llm_result(qm).message.content == ""
def test_successful_turn_with_stream_text_and_null_terminal_output_keeps_empty_message(monkeypatch):
fake_session = _FakeDbSession()
monkeypatch.setattr(app_runner_module.db, "session", fake_session)
def test_successful_turn_with_stream_text_and_null_terminal_output_keeps_empty_message(
sqlite_session: Session,
) -> None:
client = _StreamingTextNullOutputFakeAgentBackendRunClient()
store = _FakeSessionStore()
qm = _FakeQueueManager()
@ -629,15 +656,13 @@ def test_successful_turn_with_stream_text_and_null_terminal_output_keeps_empty_m
assert chunk_events == []
assert [event.chunk.delta.message.content for event in agent_message_events] == ["streamed answer"]
assert len(end_events) == 1
assert end_events[0].llm_result.message.content == ""
rows = sorted(fake_session.rows.values(), key=lambda row: row.position)
assert _llm_result(qm).message.content == ""
rows = _thought_rows(sqlite_session)
assert len(rows) == 1
assert rows[0].answer == "streamed answer"
def test_successful_turn_routes_agent_answer_to_agent_message(monkeypatch):
fake_session = _FakeDbSession()
monkeypatch.setattr(app_runner_module.db, "session", fake_session)
def test_successful_turn_routes_agent_answer_to_agent_message(sqlite_session: Session) -> None:
client = _AgentAnswerStreamingFakeAgentBackendRunClient()
store = _FakeSessionStore()
qm = _FakeQueueManager()
@ -650,21 +675,21 @@ def test_successful_turn_routes_agent_answer_to_agent_message(monkeypatch):
assert [event.chunk.delta.message.content for event in agent_message_events] == ["hello ", "agent"]
end_events = [e for e in qm.events if isinstance(e, QueueMessageEndEvent)]
assert len(end_events) == 1
assert end_events[0].llm_result.message.content == "final answer"
assert _llm_result(qm).message.content == "final answer"
thought_events = [e for e in qm.events if isinstance(e, QueueAgentThoughtEvent)]
assert len(thought_events) == 2
rows = sorted(fake_session.rows.values(), key=lambda row: row.position)
rows = _thought_rows(sqlite_session)
assert len(rows) == 1
assert rows[0].answer == "hello agent"
assert rows[0].thought == ""
assert rows[0].tool == ""
def test_agent_message_deltas_are_debounced_to_agent_message(monkeypatch):
def test_agent_message_deltas_are_debounced_to_agent_message(
monkeypatch: pytest.MonkeyPatch, sqlite_session: Session
) -> None:
monkeypatch.setattr(app_runner_module.time, "monotonic", _MonotonicClock(0.0, 0.2))
fake_session = _FakeDbSession()
monkeypatch.setattr(app_runner_module.db, "session", fake_session)
client = _StreamingFakeAgentBackendRunClient()
store = _FakeSessionStore()
qm = _FakeQueueManager()
@ -675,13 +700,13 @@ def test_agent_message_deltas_are_debounced_to_agent_message(monkeypatch):
agent_message_events = [e for e in qm.events if isinstance(e, QueueAgentMessageEvent)]
assert [event.chunk.delta.message.content for event in chunk_events] == ["hello agent"]
assert [event.chunk.delta.message.content for event in agent_message_events] == ["hello agent"]
rows = sorted(fake_session.rows.values(), key=lambda row: row.position)
rows = _thought_rows(sqlite_session)
assert rows == []
def test_successful_turn_persists_thinking_and_tool_process_events(monkeypatch):
fake_session = _FakeDbSession()
monkeypatch.setattr(app_runner_module.db, "session", fake_session)
def test_successful_turn_persists_thinking_and_tool_process_events(
sqlite_session: Session,
) -> None:
client = _ProcessStreamingFakeAgentBackendRunClient()
store = _FakeSessionStore()
qm = _FakeQueueManager()
@ -695,7 +720,7 @@ def test_successful_turn_persists_thinking_and_tool_process_events(monkeypatch):
thought_events = [e for e in qm.events if isinstance(e, QueueAgentThoughtEvent)]
assert len(thought_events) >= 3
rows = sorted(fake_session.rows.values(), key=lambda row: row.position)
rows = _thought_rows(sqlite_session)
assert rows[0].thought == "I need to inspect the file."
assert rows[0].tool == ""
assert rows[1].tool == "bash"
@ -704,9 +729,9 @@ def test_successful_turn_persists_thinking_and_tool_process_events(monkeypatch):
assert len(rows) == 2
def test_streaming_turn_cancels_after_persisting_seen_agent_answer(monkeypatch):
fake_session = _FakeDbSession()
monkeypatch.setattr(app_runner_module.db, "session", fake_session)
def test_streaming_turn_cancels_after_persisting_seen_agent_answer(
sqlite_session: Session,
) -> None:
store = _FakeSessionStore()
qm = _FakeQueueManager()
client = _StreamingStopAfterFirstDeltaFakeAgentBackendRunClient(queue_manager=qm)
@ -718,15 +743,15 @@ def test_streaming_turn_cancels_after_persisting_seen_agent_answer(monkeypatch):
agent_message_events = [e for e in qm.events if isinstance(e, QueueAgentMessageEvent)]
assert chunk_events == []
assert [event.chunk.delta.message.content for event in agent_message_events] == ["hello "]
rows = sorted(fake_session.rows.values(), key=lambda row: row.position)
rows = _thought_rows(sqlite_session)
assert len(rows) == 1
assert rows[0].answer == "hello "
assert client.cancelled_run_ids == ["fake-run-1"]
def test_tool_result_without_identity_does_not_attach_to_previous_tool(monkeypatch):
fake_session = _FakeDbSession()
monkeypatch.setattr(app_runner_module.db, "session", fake_session)
def test_tool_result_without_identity_does_not_attach_to_previous_tool(
sqlite_session: Session,
) -> None:
qm = _FakeQueueManager()
recorder = app_runner_module._AgentProcessRecorder(
dify_context=_dify_ctx(),
@ -758,7 +783,7 @@ def test_tool_result_without_identity_does_not_attach_to_previous_tool(monkeypat
)
)
rows = sorted(fake_session.rows.values(), key=lambda row: row.position)
rows = _thought_rows(sqlite_session)
assert len(rows) == 2
assert rows[0].tool == "shell_run"
assert rows[0].tool_input == '{"script": "npx skills find browser"}'
@ -768,9 +793,7 @@ def test_tool_result_without_identity_does_not_attach_to_previous_tool(monkeypat
assert rows[1].observation == "Knowledge base search results: browser skill"
def test_answer_suffix_trim_keeps_non_terminal_prefix(monkeypatch):
fake_session = _FakeDbSession()
monkeypatch.setattr(app_runner_module.db, "session", fake_session)
def test_answer_suffix_trim_keeps_non_terminal_prefix(sqlite_session: Session) -> None:
qm = _FakeQueueManager()
recorder = app_runner_module._AgentProcessRecorder(
dify_context=_dify_ctx(),
@ -781,14 +804,12 @@ def test_answer_suffix_trim_keeps_non_terminal_prefix(monkeypatch):
recorder.append_answer_text("intermediate final answer")
recorder.trim_answer_suffix("final answer")
rows = sorted(fake_session.rows.values(), key=lambda row: row.position)
rows = _thought_rows(sqlite_session)
assert len(rows) == 1
assert rows[0].answer == "intermediate "
def test_tool_call_part_binds_late_call_id_to_delta_row(monkeypatch):
fake_session = _FakeDbSession()
monkeypatch.setattr(app_runner_module.db, "session", fake_session)
def test_tool_call_part_binds_late_call_id_to_delta_row(sqlite_session: Session) -> None:
qm = _FakeQueueManager()
recorder = app_runner_module._AgentProcessRecorder(
dify_context=_dify_ctx(),
@ -840,16 +861,14 @@ def test_tool_call_part_binds_late_call_id_to_delta_row(monkeypatch):
)
)
rows = sorted(fake_session.rows.values(), key=lambda row: row.position)
rows = _thought_rows(sqlite_session)
assert len(rows) == 1
assert rows[0].tool == "knowledge_base_search"
assert rows[0].tool_input == '{"query": "browser"}'
assert rows[0].observation == "Knowledge base search results: browser skill"
def test_thinking_after_tool_starts_new_snapshot_row(monkeypatch):
fake_session = _FakeDbSession()
monkeypatch.setattr(app_runner_module.db, "session", fake_session)
def test_thinking_after_tool_starts_new_snapshot_row(sqlite_session: Session) -> None:
qm = _FakeQueueManager()
recorder = app_runner_module._AgentProcessRecorder(
dify_context=_dify_ctx(),
@ -898,16 +917,16 @@ def test_thinking_after_tool_starts_new_snapshot_row(monkeypatch):
)
)
rows = sorted(fake_session.rows.values(), key=lambda row: row.position)
rows = _thought_rows(sqlite_session)
assert [row.thought for row in rows] == ["The first thought.", "", "The next thought."]
assert rows[0].id != rows[2].id
assert rows[1].tool == "shell_run"
assert rows[1].tool_input == '{"cmd": "date"}'
def test_tool_result_without_call_id_matches_unique_open_tool_name(monkeypatch):
fake_session = _FakeDbSession()
monkeypatch.setattr(app_runner_module.db, "session", fake_session)
def test_tool_result_without_call_id_matches_unique_open_tool_name(
sqlite_session: Session,
) -> None:
qm = _FakeQueueManager()
recorder = app_runner_module._AgentProcessRecorder(
dify_context=_dify_ctx(),
@ -942,16 +961,16 @@ def test_tool_result_without_call_id_matches_unique_open_tool_name(monkeypatch):
)
)
rows = sorted(fake_session.rows.values(), key=lambda row: row.position)
rows = _thought_rows(sqlite_session)
assert len(rows) == 1
assert rows[0].tool == "knowledge_base_search"
assert rows[0].tool_input == '{"query": "browser"}'
assert rows[0].observation == "Knowledge base search results: browser skill"
def test_repeated_tool_calls_without_call_id_or_index_create_distinct_rows(monkeypatch):
fake_session = _FakeDbSession()
monkeypatch.setattr(app_runner_module.db, "session", fake_session)
def test_repeated_tool_calls_without_call_id_or_index_create_distinct_rows(
sqlite_session: Session,
) -> None:
qm = _FakeQueueManager()
recorder = app_runner_module._AgentProcessRecorder(
dify_context=_dify_ctx(),
@ -1012,7 +1031,7 @@ def test_repeated_tool_calls_without_call_id_or_index_create_distinct_rows(monke
)
)
rows = sorted(fake_session.rows.values(), key=lambda row: row.position)
rows = _thought_rows(sqlite_session)
assert len(rows) == 2
assert rows[0].tool == "shell_run"
assert rows[0].tool_input == '{"script": "lookup find"}'
@ -1022,9 +1041,9 @@ def test_repeated_tool_calls_without_call_id_or_index_create_distinct_rows(monke
assert rows[1].observation == "out output"
def test_repeated_tool_calls_with_placeholder_call_id_and_reused_index_create_distinct_rows(monkeypatch):
fake_session = _FakeDbSession()
monkeypatch.setattr(app_runner_module.db, "session", fake_session)
def test_repeated_tool_calls_with_placeholder_call_id_and_reused_index_create_distinct_rows(
sqlite_session: Session,
) -> None:
qm = _FakeQueueManager()
recorder = app_runner_module._AgentProcessRecorder(
dify_context=_dify_ctx(),
@ -1063,7 +1082,7 @@ def test_repeated_tool_calls_with_placeholder_call_id_and_reused_index_create_di
)
)
rows = sorted(fake_session.rows.values(), key=lambda row: row.position)
rows = _thought_rows(sqlite_session)
assert len(rows) == 2
assert rows[0].tool == "shell_run"
assert rows[0].tool_input == '{"script": "lookup find"}'
@ -1073,7 +1092,7 @@ def test_repeated_tool_calls_with_placeholder_call_id_and_reused_index_create_di
assert rows[1].observation == "out output"
def test_prior_session_snapshot_is_threaded_into_request():
def test_prior_session_snapshot_is_threaded_into_request() -> None:
prior = CompositorSessionSnapshot(layers=[])
client = FakeAgentBackendRunClient()
store = _FakeSessionStore(loaded=prior)
@ -1085,7 +1104,7 @@ def test_prior_session_snapshot_is_threaded_into_request():
assert client.request.session_snapshot is prior
def test_debug_session_scope_can_reuse_conversation_across_config_snapshots():
def test_debug_session_scope_can_reuse_conversation_across_config_snapshots() -> None:
prior = CompositorSessionSnapshot(layers=[])
client = FakeAgentBackendRunClient()
store = _FakeSessionStore(loaded=prior)
@ -1111,7 +1130,7 @@ def test_debug_session_scope_can_reuse_conversation_across_config_snapshots():
assert store.saved[0][0].agent_config_snapshot_id == "snap-new"
def test_failed_run_raises_agent_backend_error():
def test_failed_run_raises_agent_backend_error() -> None:
client = FakeAgentBackendRunClient(scenario=FakeAgentBackendScenario.FAILED)
store = _FakeSessionStore()
qm = _FakeQueueManager()
@ -1123,7 +1142,7 @@ def test_failed_run_raises_agent_backend_error():
assert store.saved == []
def test_agent_backend_failure_to_exception_maps_rate_limit_reason():
def test_agent_backend_failure_to_exception_maps_rate_limit_reason() -> None:
err = app_runner_module._agent_backend_failure_to_exception(
AgentBackendRunFailedInternalEvent(
run_id="run-1",
@ -1136,7 +1155,7 @@ def test_agent_backend_failure_to_exception_maps_rate_limit_reason():
assert str(err) == "quota exceeded"
def test_agent_backend_failure_to_exception_preserves_unknown_reason_context():
def test_agent_backend_failure_to_exception_preserves_unknown_reason_context() -> None:
err = app_runner_module._agent_backend_failure_to_exception(
AgentBackendRunFailedInternalEvent(
run_id="run-1",
@ -1158,7 +1177,7 @@ def test_agent_backend_failure_to_exception_preserves_unknown_reason_context():
assert str(err) == "Knowledge retrieval failed (agent_run_id=run-1)"
def test_stopped_task_cancels_agent_backend_run_and_skips_session_save():
def test_stopped_task_cancels_agent_backend_run_and_skips_session_save() -> None:
client = _RecordingFakeAgentBackendRunClient()
store = _FakeSessionStore()
qm = _StoppedQueueManager()
@ -1170,14 +1189,14 @@ def test_stopped_task_cancels_agent_backend_run_and_skips_session_save():
assert store.saved == []
def test_terminal_output_to_answer_handles_plain_string_and_dict():
def test_terminal_output_to_answer_handles_plain_string_and_dict() -> None:
assert AgentAppRunner._terminal_output_to_answer(None) == ""
assert AgentAppRunner._terminal_output_to_answer("plain text") == "plain text"
assert AgentAppRunner._terminal_output_to_answer({"text": "hi"}) == "hi"
assert AgentAppRunner._terminal_output_to_answer({"a": 1}) == '{"a": 1}'
def test_ask_human_pauses_turn_creates_form_and_persists_correlation():
def test_ask_human_pauses_turn_creates_form_and_persists_correlation() -> None:
# ENG-635/637: the PAUSED scenario emits a dify.ask_human deferred call, so
# the chat turn ends by creating a conversation-owned HITL form + saving the
# pause correlation, instead of crashing. Stub the form repo (DB-free).
@ -1205,7 +1224,7 @@ def test_ask_human_pauses_turn_creates_form_and_persists_correlation():
assert store.saved[0][4] == "fake-ask-human-1"
def test_submitted_form_resumes_turn_with_deferred_tool_results(monkeypatch):
def test_submitted_form_resumes_turn_with_deferred_tool_results(monkeypatch: pytest.MonkeyPatch) -> None:
# ENG-638: a turn that runs while a pending form is answered threads the
# human's reply into the request as deferred_tool_results.
snapshot = CompositorSessionSnapshot(layers=[])