fix(agent): finalize run as failed when run_started cannot be appended (#43959)

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():

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