diff --git a/api/tests/unit_tests/core/app/apps/agent_app/test_app_runner.py b/api/tests/unit_tests/core/app/apps/agent_app/test_app_runner.py index 95505a9949f..63c198b2c43 100644 --- a/api/tests/unit_tests/core/app/apps/agent_app/test_app_runner.py +++ b/api/tests/unit_tests/core/app/apps/agent_app/test_app_runner.py @@ -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=[])