diff --git a/dify-agent/src/dify_agent/runtime/runner.py b/dify-agent/src/dify_agent/runtime/runner.py index 8d5527264c8..0ad20ca75f0 100644 --- a/dify-agent/src/dify_agent/runtime/runner.py +++ b/dify-agent/src/dify_agent/runtime/runner.py @@ -188,9 +188,9 @@ class AgentRunRunner: self._terminal_usage = None if self.is_cancelled(): return - _ = await emit_run_started(self.sink, run_id=self.run_id) try: + _ = await emit_run_started(self.sink, run_id=self.run_id) outcome = await self._run_agent() except Exception as exc: if self.is_cancelled(): 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 ba869daabe5..7c365ec4ac0 100644 --- a/dify-agent/tests/local/dify_agent/runtime/test_runner.py +++ b/dify-agent/tests/local/dify_agent/runtime/test_runner.py @@ -204,6 +204,46 @@ async def test_failure_saves_user_history_and_classified_event(monkeypatch): assert saved.messages[-1].state == "interrupted" +class FailingAppendSink(InMemoryRunEventSink): + """Double that fails non-terminal appends so run_started cannot be stored. + + Terminal finalization keeps the real in-memory path, so the test observes + whether the runner still finalizes the run as failed. + """ + + def __init__(self, message): + super().__init__() + self.message = message + + async def append_event(self, event): + raise RuntimeError(self.message) + + +@pytest.mark.anyio +async def test_run_started_store_failure_finalizes_run_as_failed(monkeypatch): + async def stream(messages, info): + pytest.fail("the model must not run when run_started cannot be stored") + yield "unreachable" + + async with httpx.AsyncClient() as client: + model = FunctionModel(stream_function=stream) + monkeypatch.setattr(llm_layer.Capability, "build_model", lambda self, deps: model) + sink = FailingAppendSink("store unavailable") + runner = AgentRunRunner( + sink=sink, + request=request(), + run_id="test", + services=Services(client, client), + run_timeout_seconds=10, + ) + with pytest.raises(RuntimeError, match="store unavailable"): + await runner.run() + assert sink.statuses["test"] == "failed" + terminal = sink.events["test"][-1] + assert isinstance(terminal, RunFailedEvent) + assert terminal.data.error == "store unavailable" + + @pytest.mark.anyio @pytest.mark.parametrize("invalid_first", [False, True]) async def test_native_structured_output_uses_output_module(monkeypatch, invalid_first):