fix(dify-agent): atomically finalize run terminals (#39929)

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>

dify-agent/docs/dify-agent/guide/index.md

@@ -238,13 +238,29 @@ rejected synchronously once the request DTO itself is accepted.
 
 During FastAPI shutdown the scheduler rejects new runs, waits up to
 `DIFY_AGENT_SHUTDOWN_GRACE_SECONDS` for active tasks, then cancels remaining tasks
-and best-effort appends a `run_failed` event plus failed status. A hard process
-crash can still leave active runs stuck as `running`; there is no in-service
-recovery or worker handoff.
+and attempts to finalize them as failed. Success, failure, cancellation, and this
+shutdown path all use one atomic Redis transition: only the first transition from
+`running` appends a terminal event and updates the run record. A later terminal
+attempt leaves both the record and event stream unchanged. A hard process crash
+can still leave active runs stuck as `running`; there is no in-service recovery
+or worker handoff.
 
 Horizontal scaling is possible by running multiple API processes against the same
 Redis prefix, but each process executes only the runs it accepted. Redis provides
-shared status/event visibility, not load balancing or queued-job recovery.
+shared status/event visibility, not load balancing or queued-job recovery. The
+cancel endpoint can only signal a task owned by the process that receives the
+request; a request routed to a different process returns `409` while the run is
+still running. Retrying a cancellation after the run is already `cancelled` is
+idempotent.
+
+Atomic terminal finalization currently assumes the configured Redis URL targets
+one Redis deployment that can execute both run keys in a Lua script. The existing
+record and event key names are unchanged and do not contain a shared Redis
+Cluster hash tag, so Redis Cluster is not supported for this transition. During
+a rolling upgrade, older processes can still use the former split event/status
+writes; treat the single-terminal invariant as active only after those processes
+have exited. Operators should then alert on more than one terminal event per run
+and on disagreement between the run record status and terminal event type.
 
 ## Run inputs and session snapshots
 
@@ -284,7 +300,10 @@ Use the HTTP status endpoint for coarse state and the event endpoints for detail
 progress:
 
 - `POST /runs` creates a running run and schedules it locally.
-- `GET /runs/{run_id}` returns `running`, `succeeded`, or `failed`.
+- `GET /runs/{run_id}` returns `running`, `succeeded`, `failed`, or `cancelled`.
+- `POST /runs/{run_id}/cancel` atomically accepts cancellation for a locally
+  owned running task and emits `run_cancelled`; it returns `409` for a non-local
+  running task or a run whose success/failure terminal already won.
 - `GET /runs/{run_id}/events` polls the Redis Stream event log with `after` and
   `next_cursor` cursors.
 - `GET /runs/{run_id}/events/sse` replays and streams events over SSE. The SSE
@@ -292,8 +311,10 @@ progress:
   `Last-Event-ID` headers.
 
 Successful runs emit `run_started`, zero or more `pydantic_ai_event`, and
-`run_succeeded`. Failed runs end with `run_failed`. Event envelopes retain `id`,
-`run_id`, `type`, `data`, and `created_at`; `data` is typed per event type,
+`run_succeeded`. Failed runs end with `run_failed`, and accepted cancellations
+end with `run_cancelled`. Each run can append at most one of these terminal
+events. Event envelopes retain `id`, `run_id`, `type`, `data`, and `created_at`;
+`data` is typed per event type,
 including Pydantic AI's `AgentStreamEvent` payload for `pydantic_ai_event` and a
 terminal `run_succeeded.data` object containing a `CompositorSessionSnapshot` for
 resumption. A successful run has exactly one active result branch: JSON-safe

dify-agent/src/dify_agent/protocol/schemas.py

@@ -362,7 +362,7 @@ class RunSucceededEvent(BaseRunEvent):
 
 
 class RunFailedEvent(BaseRunEvent):
-    """Terminal failure event emitted before the run status becomes failed."""
+    """Terminal failure event atomically committed with the failed run status."""
 
     type: Literal["run_failed"] = "run_failed"
     data: RunFailedEventData

dify-agent/src/dify_agent/runtime/event_sink.py

@@ -1,16 +1,15 @@
 """Event sink contracts used by the runner and storage adapters.
 
-The runner only needs append-only event writes and status transitions, so tests
-can use ``InMemoryRunEventSink`` without Redis. Production storage implements the
-same protocol with Redis streams in ``dify_agent.storage.redis_run_store``. The
-terminal success helper writes either the final JSON-safe output or one deferred
-tool request together with the resumable session snapshot in a single event so
-consumers can stop at ``run_succeeded`` without correlating separate payload
-events.
+Non-terminal events remain append-only. Terminal events use ``finalize_run`` so
+the event and matching run status are committed as one compare-and-set
+transition. Tests can use ``InMemoryRunEventSink`` without Redis; production
+storage implements the same contract with Redis streams in
+``dify_agent.storage.redis_run_store``.
 """
 
 from collections import defaultdict
-from typing import Protocol, cast
+from dataclasses import dataclass
+from typing import Protocol, TypeAlias, cast
 
 from pydantic import JsonValue
 from pydantic_ai.messages import AgentStreamEvent
@@ -35,17 +34,28 @@ from dify_agent.protocol.schemas import (
 
 
 _UNSET = object()
+TerminalRunEvent: TypeAlias = RunSucceededEvent | RunFailedEvent | RunCancelledEvent
+NonTerminalRunEvent: TypeAlias = RunStartedEvent | PydanticAIStreamRunEvent
+
+
+@dataclass(frozen=True, slots=True)
+class RunFinalizationResult:
+    """Outcome of attempting the only terminal transition for one run."""
+
+    applied: bool
+    status: RunStatus
+    event_id: str | None = None
 
 
 class RunEventSink(Protocol):
     """Boundary used by runtime code to publish observable run progress."""
 
-    async def append_event(self, event: RunEvent) -> str:
-        """Persist ``event`` and return its cursor id."""
+    async def append_event(self, event: NonTerminalRunEvent) -> str:
+        """Persist a non-terminal event and return its cursor id."""
         ...
 
-    async def update_status(self, run_id: str, status: RunStatus, error: str | None = None) -> None:
-        """Persist the current run status."""
+    async def finalize_run(self, event: TerminalRunEvent) -> RunFinalizationResult:
+        """Atomically persist the first terminal event and matching status."""
         ...
 
 
@@ -61,25 +71,44 @@ class InMemoryRunEventSink:
         self.statuses = {}
         self.errors = {}
 
-    async def append_event(self, event: RunEvent) -> str:
-        """Store an event and assign a monotonic per-run cursor."""
+    async def append_event(self, event: NonTerminalRunEvent) -> str:
+        """Store a non-terminal event and assign a monotonic per-run cursor."""
         event_id = str(len(self.events[event.run_id]) + 1)
         stored = event.model_copy(update={"id": event_id})
         self.events[event.run_id].append(stored)
         return event_id
 
-    async def update_status(self, run_id: str, status: RunStatus, error: str | None = None) -> None:
-        """Record the latest status; timestamps are owned by run stores."""
-        self.statuses[run_id] = status
-        self.errors[run_id] = error
+    async def finalize_run(self, event: TerminalRunEvent) -> RunFinalizationResult:
+        """Store only the first terminal event and its derived status."""
+        current_status = self.statuses.get(event.run_id, "running")
+        if current_status != "running":
+            return RunFinalizationResult(applied=False, status=current_status)
+
+        status, error = terminal_event_status_and_error(event)
+        event_id = str(len(self.events[event.run_id]) + 1)
+        self.events[event.run_id].append(event.model_copy(update={"id": event_id}))
+        self.statuses[event.run_id] = status
+        self.errors[event.run_id] = error
+        return RunFinalizationResult(applied=True, status=status, event_id=event_id)
+
+
+def terminal_event_status_and_error(event: TerminalRunEvent) -> tuple[RunStatus, str | None]:
+    """Derive the persisted terminal status fields from one typed event."""
+    match event:
+        case RunSucceededEvent():
+            return "succeeded", None
+        case RunFailedEvent():
+            return "failed", event.data.error
+        case RunCancelledEvent():
+            return "cancelled", event.data.message or event.data.reason
 
 
 async def emit_run_event(
     sink: RunEventSink,
     *,
-    event: RunEvent,
+    event: NonTerminalRunEvent,
 ) -> str:
-    """Append an already typed public run event."""
+    """Append an already typed non-terminal public run event."""
     return await sink.append_event(event)
 
 
@@ -118,8 +147,8 @@ async def emit_run_succeeded(
     deferred_tool_call: DeferredToolCallPayload | object = _UNSET,
     session_snapshot: CompositorSessionSnapshot,
     usage: AgentRunUsage | None = None,
-) -> str:
-    """Emit the terminal success event with output or deferred continuation.
+) -> RunFinalizationResult:
+    """Finalize a run as succeeded with output or deferred continuation.
 
     Callers must activate exactly one result branch. ``_UNSET`` is used instead
     of ``None`` to preserve the distinction between an omitted inactive branch
@@ -137,9 +166,8 @@ async def emit_run_succeeded(
     if usage is not None:
         data["usage"] = usage
 
-    return await emit_run_event(
-        sink,
-        event=RunSucceededEvent(
+    return await sink.finalize_run(
+        RunSucceededEvent(
             run_id=run_id,
             data=RunSucceededEventData.model_validate(data),
             created_at=utc_now(),
@@ -153,11 +181,10 @@ async def emit_run_failed(
     run_id: str,
     error: str,
     reason: str | None = None,
-) -> str:
-    """Emit the terminal failure lifecycle event."""
-    return await emit_run_event(
-        sink,
-        event=RunFailedEvent(run_id=run_id, data=RunFailedEventData(error=error, reason=reason), created_at=utc_now()),
+) -> RunFinalizationResult:
+    """Finalize a run with a failed terminal event."""
+    return await sink.finalize_run(
+        RunFailedEvent(run_id=run_id, data=RunFailedEventData(error=error, reason=reason), created_at=utc_now()),
     )
 
 
@@ -167,11 +194,10 @@ async def emit_run_cancelled(
     run_id: str,
     reason: str | None = None,
     message: str | None = None,
-) -> str:
-    """Emit the terminal cancellation lifecycle event."""
-    return await emit_run_event(
-        sink,
-        event=RunCancelledEvent(
+) -> RunFinalizationResult:
+    """Finalize a run with a cancelled terminal event."""
+    return await sink.finalize_run(
+        RunCancelledEvent(
             run_id=run_id,
             data=RunCancelledEventData(reason=reason, message=message),
             created_at=utc_now(),
@@ -181,11 +207,15 @@ async def emit_run_cancelled(
 
 __all__ = [
     "InMemoryRunEventSink",
+    "NonTerminalRunEvent",
     "RunEventSink",
+    "RunFinalizationResult",
+    "TerminalRunEvent",
     "emit_pydantic_ai_event",
     "emit_run_cancelled",
     "emit_run_event",
     "emit_run_failed",
     "emit_run_started",
     "emit_run_succeeded",
+    "terminal_event_status_and_error",
 ]

dify-agent/src/dify_agent/runtime/run_scheduler.py

@@ -131,15 +131,18 @@ class RunScheduler:
             task = self.active_tasks.get(run_id)
             if task is None:
                 raise RunCancellationConflictError("run is not active in this scheduler process")
-            self.cancelled_run_ids.add(run_id)
-            _ = task.cancel(request.message or request.reason)
-            _ = await emit_run_cancelled(
+            finalization = await emit_run_cancelled(
                 self.store,
                 run_id=run_id,
                 reason=request.reason,
                 message=request.message,
             )
-            await self.store.update_status(run_id, "cancelled", request.message or request.reason)
+            if not finalization.applied:
+                if finalization.status == "cancelled":
+                    return CancelRunResponse(run_id=run_id, status="cancelled")
+                raise RunCancellationConflictError(f"run already finished with status {finalization.status!r}")
+            self.cancelled_run_ids.add(run_id)
+            _ = task.cancel(request.message or request.reason)
 
         # Some model/tool stacks can consume one CancelledError. Re-inject it
         # after the terminal state is durable without making the HTTP request
@@ -206,7 +209,6 @@ class RunScheduler:
         message = "run cancelled during server shutdown"
         try:
             _ = await emit_run_failed(self.store, run_id=run_id, error=message, reason="shutdown")
-            await self.store.update_status(run_id, "failed", message)
         except Exception:
             logger.exception("failed to mark cancelled run failed", extra={"run_id": run_id})
 

dify-agent/src/dify_agent/runtime/runner.py

@@ -197,9 +197,6 @@ class AgentRunRunner:
 
     async def run(self) -> None:
         """Execute the run and emit the documented event sequence."""
-        if self.is_cancelled():
-            return
-        await self.sink.update_status(self.run_id, "running")
         if self.is_cancelled():
             return
         _ = await emit_run_started(self.sink, run_id=self.run_id)
@@ -210,9 +207,10 @@ class AgentRunRunner:
             if self.is_cancelled():
                 return
             message, reason = _run_failed_error_payload(exc)
-            _ = await emit_run_failed(self.sink, run_id=self.run_id, error=message, reason=reason)
-            await self.sink.update_status(self.run_id, "failed", message)
-            raise
+            finalization = await emit_run_failed(self.sink, run_id=self.run_id, error=message, reason=reason)
+            if finalization.applied:
+                raise
+            return
 
         if self.is_cancelled():
             return
@@ -227,7 +225,6 @@ class AgentRunRunner:
             session_snapshot=outcome.session_snapshot,
             usage=outcome.usage,
         )
-        await self.sink.update_status(self.run_id, "succeeded")
 
     async def _run_agent(self) -> RunSuccessOutcome:
         """Run the normalized request through the model path.

dify-agent/src/dify_agent/storage/redis_run_store.py

@@ -9,13 +9,19 @@ create-run payloads are never persisted because layer config may include model
 credentials.
 """
 
-from collections.abc import AsyncIterator
+from collections.abc import AsyncIterator, Awaitable
 from typing import cast
 
 from redis.asyncio import Redis
 
-from dify_agent.protocol.schemas import RUN_EVENT_ADAPTER, RunEvent, RunEventsResponse, RunStatus, utc_now
-from dify_agent.runtime.event_sink import RunEventSink
+from dify_agent.protocol.schemas import RUN_EVENT_ADAPTER, RunEvent, RunEventsResponse, RunStatus
+from dify_agent.runtime.event_sink import (
+    NonTerminalRunEvent,
+    RunEventSink,
+    RunFinalizationResult,
+    TerminalRunEvent,
+    terminal_event_status_and_error,
+)
 from dify_agent.server.schemas import RunRecord, new_run_id
 from dify_agent.server.settings import DEFAULT_RUN_RETENTION_SECONDS
 from dify_agent.storage.redis_keys import run_events_key, run_record_key
@@ -25,6 +31,34 @@ class RunNotFoundError(LookupError):
     """Raised when a requested run record does not exist."""
 
 
+_FINALIZE_RUN_SCRIPT = """
+local record_json = redis.call("GET", KEYS[1])
+if not record_json then
+    return {-1, "", ""}
+end
+
+local record = cjson.decode(record_json)
+if record.status ~= "running" then
+    return {0, tostring(record.status), ""}
+end
+
+record.status = ARGV[1]
+record.updated_at = ARGV[2]
+if ARGV[3] == "1" then
+    record.error = ARGV[4]
+else
+    record.error = cjson.null
+end
+
+local ttl = tonumber(ARGV[6])
+local updated_record_json = cjson.encode(record)
+local event_id = redis.call("XADD", KEYS[2], "*", "payload", ARGV[5])
+redis.call("EXPIRE", KEYS[2], ttl)
+redis.call("SET", KEYS[1], updated_record_json, "EX", ttl)
+return {1, ARGV[1], event_id}
+"""
+
+
 class RedisRunStore(RunEventSink):
     """Async Redis implementation for run records and event logs.
 
@@ -73,18 +107,8 @@ class RedisRunStore(RunEventSink):
             value = value.decode()
         return RunRecord.model_validate_json(value)
 
-    async def update_status(self, run_id: str, status: RunStatus, error: str | None = None) -> None:
-        """Update the status fields of an existing run record."""
-        record = await self.get_run(run_id)
-        updated = record.model_copy(update={"status": status, "updated_at": utc_now(), "error": error})
-        await self.redis.set(
-            run_record_key(self.prefix, run_id),
-            updated.model_dump_json(),
-            ex=self.run_retention_seconds,
-        )
-
-    async def append_event(self, event: RunEvent) -> str:
-        """Append an event JSON payload to the run's Redis stream with TTLs."""
+    async def append_event(self, event: NonTerminalRunEvent) -> str:
+        """Append a non-terminal event JSON payload with refreshed TTLs."""
         events_key = run_events_key(self.prefix, event.run_id)
         payload = RUN_EVENT_ADAPTER.dump_json(event, exclude={"id"}).decode()
         async with self.redis.pipeline(transaction=True) as pipeline:
@@ -98,6 +122,39 @@ class RedisRunStore(RunEventSink):
         event_id = results[0]
         return event_id.decode() if isinstance(event_id, bytes) else str(event_id)
 
+    async def finalize_run(self, event: TerminalRunEvent) -> RunFinalizationResult:
+        """Atomically append the first terminal event and update its run record."""
+        status, error = terminal_event_status_and_error(event)
+        payload = RUN_EVENT_ADAPTER.dump_json(event, exclude={"id"}).decode()
+        evaluation = cast(
+            Awaitable[object],
+            self.redis.eval(
+                _FINALIZE_RUN_SCRIPT,
+                2,
+                run_record_key(self.prefix, event.run_id),
+                run_events_key(self.prefix, event.run_id),
+                status,
+                event.created_at.isoformat(),
+                "1" if error is not None else "0",
+                error or "",
+                payload,
+                str(self.run_retention_seconds),
+            ),
+        )
+        raw_result = await evaluation
+        result = cast(list[object], raw_result)
+        applied = int(cast(int | bytes | str, result[0]))
+        if applied == -1:
+            raise RunNotFoundError(event.run_id)
+
+        persisted_status = cast(RunStatus, _decode_redis_text(result[1]))
+        event_id = _decode_redis_text(result[2]) or None
+        return RunFinalizationResult(
+            applied=applied == 1,
+            status=persisted_status,
+            event_id=event_id,
+        )
+
     async def get_events(self, run_id: str, *, after: str = "0-0", limit: int = 100) -> RunEventsResponse:
         """Read a bounded page of events after ``after`` cursor."""
         await self.get_run(run_id)
@@ -140,4 +197,8 @@ class RedisRunStore(RunEventSink):
         return event.model_copy(update={"id": event_id, "run_id": run_id})
 
 
+def _decode_redis_text(value: object) -> str:
+    return value.decode() if isinstance(value, bytes) else str(value)
+
+
 __all__ = ["DEFAULT_RUN_RETENTION_SECONDS", "RedisRunStore", "RunNotFoundError"]

dify-agent/tests/integration/dify_agent/storage/test_terminal_finalization.py

@@ -0,0 +1,133 @@
+"""Real-Redis contracts for atomic terminal run finalization."""
+
+import asyncio
+from collections.abc import Iterator
+import shutil
+import socket
+import subprocess
+import time
+from uuid import uuid4
+
+import pytest
+from redis.asyncio import Redis
+
+from agenton.compositor import CompositorSessionSnapshot
+from dify_agent.protocol.schemas import (
+    RunCancelledEvent,
+    RunCancelledEventData,
+    RunSucceededEvent,
+    RunSucceededEventData,
+)
+from dify_agent.runtime.event_sink import TerminalRunEvent, terminal_event_status_and_error
+from dify_agent.storage.redis_keys import run_events_key, run_record_key
+from dify_agent.storage.redis_run_store import RedisRunStore
+
+
+pytestmark = pytest.mark.integration
+
+
+@pytest.fixture
+def redis_url() -> Iterator[str]:
+    """Start an isolated Redis when the binary is available locally."""
+    redis_server = shutil.which("redis-server")
+    if redis_server is None:
+        pytest.skip("redis-server is required for the atomic finalization integration test")
+
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as probe:
+        probe.bind(("127.0.0.1", 0))
+        port = probe.getsockname()[1]
+
+    process = subprocess.Popen(  # noqa: S603
+        [
+            redis_server,
+            "--bind",
+            "127.0.0.1",
+            "--port",
+            str(port),
+            "--save",
+            "",
+            "--appendonly",
+            "no",
+        ],
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.STDOUT,
+    )
+    try:
+        deadline = time.monotonic() + 5
+        while time.monotonic() < deadline:
+            if process.poll() is not None:
+                pytest.fail(f"redis-server exited during startup with code {process.returncode}")
+            try:
+                with socket.create_connection(("127.0.0.1", port), timeout=0.1):
+                    break
+            except OSError:
+                time.sleep(0.05)
+        else:
+            pytest.fail("redis-server did not accept connections within 5 seconds")
+        yield f"redis://127.0.0.1:{port}/0"
+    finally:
+        process.terminate()
+        try:
+            process.wait(timeout=5)
+        except subprocess.TimeoutExpired:
+            process.kill()
+            process.wait(timeout=5)
+
+
+def test_two_redis_clients_commit_exactly_one_matching_terminal(redis_url: str) -> None:
+    async def scenario() -> None:
+        first_client = Redis.from_url(redis_url)
+        second_client = Redis.from_url(redis_url)
+        prefix = f"terminal-finalization-{uuid4().hex}"
+        retention_seconds = 60
+        first_store = RedisRunStore(first_client, prefix=prefix, run_retention_seconds=retention_seconds)
+        second_store = RedisRunStore(second_client, prefix=prefix, run_retention_seconds=retention_seconds)
+        try:
+            record = await first_store.create_run()
+            terminal_events: tuple[TerminalRunEvent, TerminalRunEvent] = (
+                RunSucceededEvent(
+                    run_id=record.run_id,
+                    data=RunSucceededEventData(
+                        output="done",
+                        session_snapshot=CompositorSessionSnapshot(layers=[]),
+                    ),
+                ),
+                RunCancelledEvent(
+                    run_id=record.run_id,
+                    data=RunCancelledEventData(
+                        reason="concurrent_cancel",
+                        message="cancel accepted",
+                    ),
+                ),
+            )
+
+            results = await asyncio.gather(
+                first_store.finalize_run(terminal_events[0]),
+                second_store.finalize_run(terminal_events[1]),
+            )
+
+            assert sum(result.applied for result in results) == 1
+            winner_index = next(index for index, result in enumerate(results) if result.applied)
+            winner_event = terminal_events[winner_index]
+            winner_result = results[winner_index]
+            expected_status, expected_error = terminal_event_status_and_error(winner_event)
+
+            persisted = await first_store.get_run(record.run_id)
+            page = await second_store.get_events(record.run_id)
+            assert persisted.status == expected_status
+            assert persisted.error == expected_error
+            assert persisted.updated_at == winner_event.created_at
+            assert len(page.events) == 1
+            assert page.events[0].type == winner_event.type
+            assert page.events[0].created_at == winner_event.created_at
+            assert page.events[0].id == winner_result.event_id
+
+            record_ttl = await first_client.ttl(run_record_key(prefix, record.run_id))
+            events_ttl = await second_client.ttl(run_events_key(prefix, record.run_id))
+            assert 0 < record_ttl <= retention_seconds
+            assert 0 < events_ttl <= retention_seconds
+        finally:
+            await first_client.aclose()
+            await second_client.aclose()
+
+    asyncio.run(scenario())

dify-agent/tests/local/dify_agent/runtime/test_run_scheduler.py

@@ -20,6 +20,13 @@ from dify_agent.protocol.schemas import (
     RunLayerSpec,
     RunStatus,
 )
+from dify_agent.runtime.event_sink import (
+    NonTerminalRunEvent,
+    RunFinalizationResult,
+    TerminalRunEvent,
+    emit_run_succeeded,
+    terminal_event_status_and_error,
+)
 from dify_agent.runtime.run_scheduler import RunCancellationConflictError, RunScheduler, SchedulerStoppingError
 from dify_agent.server.schemas import RunRecord
 
@@ -99,7 +106,7 @@ class FakeStore:
         self.statuses[run_id] = "running"
         return record
 
-    async def append_event(self, event: RunEvent) -> str:
+    async def append_event(self, event: NonTerminalRunEvent) -> str:
         event_id = str(len(self.events[event.run_id]) + 1)
         self.events[event.run_id].append(event.model_copy(update={"id": event_id}))
         return event_id
@@ -109,9 +116,17 @@ class FakeStore:
             update={"status": self.statuses[run_id], "error": self.errors.get(run_id)},
         )
 
-    async def update_status(self, run_id: str, status: RunStatus, error: str | None = None) -> None:
-        self.statuses[run_id] = status
-        self.errors[run_id] = error
+    async def finalize_run(self, event: TerminalRunEvent) -> RunFinalizationResult:
+        current_status = self.statuses[event.run_id]
+        if current_status != "running":
+            return RunFinalizationResult(applied=False, status=current_status)
+
+        status, error = terminal_event_status_and_error(event)
+        event_id = str(len(self.events[event.run_id]) + 1)
+        self.events[event.run_id].append(event.model_copy(update={"id": event_id}))
+        self.statuses[event.run_id] = status
+        self.errors[event.run_id] = error
+        return RunFinalizationResult(applied=True, status=status, event_id=event_id)
 
 
 class SlowCreateStore(FakeStore):
@@ -159,6 +174,76 @@ class SwallowOneCancellationRunner:
             await asyncio.Event().wait()
 
 
+class SuccessThenWaitRunner:
+    def __init__(
+        self,
+        *,
+        store: FakeStore,
+        run_id: str,
+        finalized: asyncio.Event,
+        release: asyncio.Event,
+    ) -> None:
+        self.store = store
+        self.run_id = run_id
+        self.finalized = finalized
+        self.release = release
+
+    async def run(self) -> None:
+        result = await emit_run_succeeded(
+            self.store,
+            run_id=self.run_id,
+            output="done",
+            session_snapshot=CompositorSessionSnapshot(layers=[]),
+        )
+        assert result.applied is True
+        self.finalized.set()
+        await self.release.wait()
+
+
+class IgnoreCancellationThenSucceedRunner:
+    def __init__(self, *, store: FakeStore, run_id: str, started: asyncio.Event, release: asyncio.Event) -> None:
+        self.store = store
+        self.run_id = run_id
+        self.started = started
+        self.release = release
+
+    async def run(self) -> None:
+        self.started.set()
+        while not self.release.is_set():
+            try:
+                await self.release.wait()
+            except asyncio.CancelledError:
+                continue
+        result = await emit_run_succeeded(
+            self.store,
+            run_id=self.run_id,
+            output="late success",
+            session_snapshot=CompositorSessionSnapshot(layers=[]),
+        )
+        assert result.applied is False
+        assert result.status == "cancelled"
+
+
+class FinalizeSuccessOnCancellationRunner:
+    def __init__(self, *, store: FakeStore, run_id: str, started: asyncio.Event) -> None:
+        self.store = store
+        self.run_id = run_id
+        self.started = started
+
+    async def run(self) -> None:
+        self.started.set()
+        try:
+            await asyncio.Event().wait()
+        except asyncio.CancelledError:
+            result = await emit_run_succeeded(
+                self.store,
+                run_id=self.run_id,
+                output="completed during shutdown",
+                session_snapshot=CompositorSessionSnapshot(layers=[]),
+            )
+            assert result.applied is True
+
+
 def test_create_run_starts_background_task_and_returns_running() -> None:
     async def scenario() -> None:
         store = FakeStore()
@@ -276,13 +361,107 @@ def test_cancel_run_reinjects_cancellation_without_waiting_for_runner_cleanup()
     asyncio.run(scenario())
 
 
+def test_cancel_run_does_not_override_successful_terminal() -> None:
+    async def scenario() -> None:
+        store = FakeStore()
+        finalized = asyncio.Event()
+        release = asyncio.Event()
+        async with httpx.AsyncClient() as client:
+            scheduler = RunScheduler(
+                store=store,
+                plugin_daemon_http_client=client,
+                dify_api_http_client=client,
+                runner_factory=lambda record, _request: SuccessThenWaitRunner(
+                    store=store,
+                    run_id=record.run_id,
+                    finalized=finalized,
+                    release=release,
+                ),
+            )
+            record = await scheduler.create_run(_request())
+            await asyncio.wait_for(finalized.wait(), timeout=1)
+            task = scheduler.active_tasks[record.run_id]
+
+            with pytest.raises(RunCancellationConflictError, match="already finished with status 'succeeded'"):
+                await scheduler.cancel_run(record.run_id, CancelRunRequest(reason="late_cancel"))
+
+            assert task.done() is False
+            assert store.statuses[record.run_id] == "succeeded"
+            assert [event.type for event in store.events[record.run_id]] == ["run_succeeded"]
+            release.set()
+            await asyncio.wait_for(task, timeout=1)
+
+    asyncio.run(scenario())
+
+
+def test_cancel_run_wins_before_a_runner_that_consumes_cancellation_finishes() -> None:
+    async def scenario() -> None:
+        store = FakeStore()
+        started = asyncio.Event()
+        release = asyncio.Event()
+        async with httpx.AsyncClient() as client:
+            scheduler = RunScheduler(
+                store=store,
+                plugin_daemon_http_client=client,
+                dify_api_http_client=client,
+                runner_factory=lambda record, _request: IgnoreCancellationThenSucceedRunner(
+                    store=store,
+                    run_id=record.run_id,
+                    started=started,
+                    release=release,
+                ),
+            )
+            record = await scheduler.create_run(_request())
+            await asyncio.wait_for(started.wait(), timeout=1)
+
+            response = await scheduler.cancel_run(record.run_id, CancelRunRequest(reason="workflow_aborted"))
+
+            assert response.status == "cancelled"
+            assert store.statuses[record.run_id] == "cancelled"
+            assert [event.type for event in store.events[record.run_id]] == ["run_cancelled"]
+            task = scheduler.active_tasks[record.run_id]
+            release.set()
+            await asyncio.wait_for(task, timeout=1)
+            await asyncio.sleep(0)
+            assert [event.type for event in store.events[record.run_id]] == ["run_cancelled"]
+
+    asyncio.run(scenario())
+
+
+def test_shutdown_does_not_append_failed_after_success_wins() -> None:
+    async def scenario() -> None:
+        store = FakeStore()
+        started = asyncio.Event()
+        async with httpx.AsyncClient() as client:
+            scheduler = RunScheduler(
+                store=store,
+                plugin_daemon_http_client=client,
+                dify_api_http_client=client,
+                shutdown_grace_seconds=0,
+                runner_factory=lambda record, _request: FinalizeSuccessOnCancellationRunner(
+                    store=store,
+                    run_id=record.run_id,
+                    started=started,
+                ),
+            )
+            record = await scheduler.create_run(_request())
+            await asyncio.wait_for(started.wait(), timeout=1)
+
+            await scheduler.shutdown()
+
+            assert store.statuses[record.run_id] == "succeeded"
+            assert [event.type for event in store.events[record.run_id]] == ["run_succeeded"]
+
+    asyncio.run(scenario())
+
+
 def test_cancel_run_rejects_finished_run() -> None:
     async def scenario() -> None:
         store = FakeStore()
         async with httpx.AsyncClient() as client:
             scheduler = RunScheduler(store=store, plugin_daemon_http_client=client, dify_api_http_client=client)
             record = await store.create_run()
-            await store.update_status(record.run_id, "succeeded")
+            store.statuses[record.run_id] = "succeeded"
 
             with pytest.raises(RunCancellationConflictError, match="already finished"):
                 await scheduler.cancel_run(record.run_id, CancelRunRequest())

dify-agent/tests/local/dify_agent/runtime/test_runner.py

@@ -63,7 +63,7 @@ from dify_agent.protocol.schemas import (
     RunLayerSpec,
     RunSucceededEvent,
 )
-from dify_agent.runtime.event_sink import InMemoryRunEventSink
+from dify_agent.runtime.event_sink import InMemoryRunEventSink, emit_run_cancelled
 from dify_agent.runtime.compositor_factory import create_default_layer_providers
 from dify_agent.runtime.runner import (
     AgentRunRunner,
@@ -226,7 +226,12 @@ def test_cancelled_runner_does_not_overwrite_cancelled_status_with_late_failure(
             async def fail_after_cancel() -> RunSuccessOutcome:
                 nonlocal cancelled
                 cancelled = True
-                await sink.update_status("run-cancelled", "cancelled", "workflow stopped")
+                _ = await emit_run_cancelled(
+                    sink,
+                    run_id="run-cancelled",
+                    reason="workflow_aborted",
+                    message="workflow stopped",
+                )
                 raise RuntimeError("late model failure")
 
             monkeypatch.setattr(runner, "_run_agent", fail_after_cancel)
@@ -234,7 +239,7 @@ def test_cancelled_runner_does_not_overwrite_cancelled_status_with_late_failure(
 
         assert sink.statuses["run-cancelled"] == "cancelled"
         assert sink.errors["run-cancelled"] == "workflow stopped"
-        assert [event.type for event in sink.events["run-cancelled"]] == ["run_started"]
+        assert [event.type for event in sink.events["run-cancelled"]] == ["run_started", "run_cancelled"]
 
     asyncio.run(scenario())
 

dify-agent/tests/local/dify_agent/storage/test_redis_run_store.py

@@ -1,13 +1,25 @@
 import asyncio
 from collections.abc import Mapping
+import json
 from typing import cast
 
+import pytest
 from pydantic import JsonValue
 
 from agenton.compositor import CompositorSessionSnapshot, LayerSessionSnapshot
 from agenton.layers import LifecycleState
-from dify_agent.protocol.schemas import RunStartedEvent, RunSucceededEvent, RunSucceededEventData
-from dify_agent.storage.redis_run_store import DEFAULT_RUN_RETENTION_SECONDS, RedisRunStore
+from dify_agent.protocol.schemas import (
+    RunCancelledEvent,
+    RunCancelledEventData,
+    RunFailedEvent,
+    RunFailedEventData,
+    RunStartedEvent,
+    RunStatus,
+    RunSucceededEvent,
+    RunSucceededEventData,
+)
+from dify_agent.runtime.event_sink import RunFinalizationResult
+from dify_agent.storage.redis_run_store import DEFAULT_RUN_RETENTION_SECONDS, RedisRunStore, RunNotFoundError
 
 
 class FakeRedis:
@@ -55,6 +67,30 @@ class FakeRedis:
         self.commands.append(("expire", key, seconds))
         return True
 
+    async def eval(self, script: str, numkeys: int, *keys_and_args: object) -> list[object]:
+        self.commands.append(("eval", script, numkeys, *keys_and_args))
+        assert numkeys == 2
+        record_key = str(keys_and_args[0])
+        events_key = str(keys_and_args[1])
+        status = str(keys_and_args[2])
+        updated_at = str(keys_and_args[3])
+        has_error = str(keys_and_args[4]) == "1"
+        error = str(keys_and_args[5]) if has_error else None
+        payload = str(keys_and_args[6])
+        record_json = self.values.get(record_key)
+        if record_json is None:
+            return [-1, "", ""]
+        if isinstance(record_json, bytes):
+            record_json = record_json.decode()
+        record = json.loads(cast(str, record_json))
+        if record["status"] != "running":
+            return [0, record["status"], ""]
+
+        record.update({"status": status, "updated_at": updated_at, "error": error})
+        event_id = self._append_stream_entry(events_key, {"payload": payload})
+        self.values[record_key] = json.dumps(record, separators=(",", ":"))
+        return [1, status, event_id]
+
     @staticmethod
     def _is_after_min(event_id: str, min_id: str) -> bool:
         if min_id == "-":
@@ -113,17 +149,166 @@ def test_create_run_writes_running_record_without_job_queue_and_with_retention()
     assert "request" not in str(redis.commands[0][2])
 
 
-def test_update_status_refreshes_record_retention() -> None:
+def test_finalize_run_atomically_writes_terminal_event_and_status() -> None:
     redis = FakeRedis()
     store = RedisRunStore(redis, prefix="test", run_retention_seconds=60)  # pyright: ignore[reportArgumentType]
     record = asyncio.run(store.create_run())
     redis.commands.clear()
+    event = RunCancelledEvent(
+        run_id=record.run_id,
+        data=RunCancelledEventData(reason="workflow_aborted", message="workflow stopped"),
+    )
 
-    asyncio.run(store.update_status(record.run_id, "succeeded"))
+    result = asyncio.run(store.finalize_run(event))
+    updated = asyncio.run(store.get_run(record.run_id))
 
-    assert [command[0] for command in redis.commands] == ["get", "set"]
-    assert redis.commands[1][1] == f"test:runs:{record.run_id}:record"
-    assert redis.commands[1][3] == 60
+    assert result.applied is True
+    assert result.status == "cancelled"
+    assert result.event_id == "1-0"
+    assert updated.status == "cancelled"
+    assert updated.error == "workflow stopped"
+    assert updated.updated_at == event.created_at
+    stream_entry_id, stream_fields = redis.streams[f"test:runs:{record.run_id}:events"][0]
+    assert stream_entry_id == result.event_id
+    payload = json.loads(cast(str, stream_fields["payload"]))
+    assert "id" not in payload
+    assert payload["type"] == "run_cancelled"
+    assert payload["data"] == {"reason": "workflow_aborted", "message": "workflow stopped"}
+    assert payload["created_at"] == event.created_at.isoformat().replace("+00:00", "Z")
+    eval_command = redis.commands[0]
+    assert eval_command[0] == "eval"
+    assert eval_command[2] == 2
+    assert eval_command[-1] == "60"
+
+
+def test_finalize_run_rejects_a_second_terminal_without_appending_event() -> None:
+    redis = FakeRedis()
+    store = RedisRunStore(redis, prefix="test", run_retention_seconds=60)  # pyright: ignore[reportArgumentType]
+    record = asyncio.run(store.create_run())
+    snapshot = CompositorSessionSnapshot(layers=[])
+
+    first = asyncio.run(
+        store.finalize_run(
+            RunSucceededEvent(
+                run_id=record.run_id,
+                data=RunSucceededEventData(output="done", session_snapshot=snapshot),
+            )
+        )
+    )
+    second = asyncio.run(
+        store.finalize_run(
+            RunCancelledEvent(
+                run_id=record.run_id,
+                data=RunCancelledEventData(reason="late_cancel"),
+            )
+        )
+    )
+
+    assert first.applied is True
+    assert second.applied is False
+    assert second.status == "succeeded"
+    assert second.event_id is None
+    assert len(redis.streams[f"test:runs:{record.run_id}:events"]) == 1
+
+
+def test_finalize_failed_run_derives_error_and_timestamp_from_event() -> None:
+    redis = FakeRedis()
+    store = RedisRunStore(redis, prefix="test")  # pyright: ignore[reportArgumentType]
+    record = asyncio.run(store.create_run())
+    event = RunFailedEvent(
+        run_id=record.run_id,
+        data=RunFailedEventData(error="model failed", reason="model_error"),
+    )
+
+    result = asyncio.run(store.finalize_run(event))
+    updated = asyncio.run(store.get_run(record.run_id))
+
+    assert result.applied is True
+    assert result.status == "failed"
+    assert updated.status == "failed"
+    assert updated.error == "model failed"
+    assert updated.updated_at == event.created_at
+
+
+def test_two_store_instances_choose_exactly_one_terminal_winner() -> None:
+    redis = FakeRedis()
+    first_store = RedisRunStore(redis, prefix="test")  # pyright: ignore[reportArgumentType]
+    second_store = RedisRunStore(redis, prefix="test")  # pyright: ignore[reportArgumentType]
+
+    async def scenario() -> tuple[list[RunFinalizationResult], RunStatus, list[str]]:
+        record = await first_store.create_run()
+        snapshot = CompositorSessionSnapshot(layers=[])
+        results = await asyncio.gather(
+            first_store.finalize_run(
+                RunSucceededEvent(
+                    run_id=record.run_id,
+                    data=RunSucceededEventData(output="done", session_snapshot=snapshot),
+                )
+            ),
+            second_store.finalize_run(
+                RunCancelledEvent(
+                    run_id=record.run_id,
+                    data=RunCancelledEventData(reason="concurrent_cancel"),
+                )
+            ),
+        )
+        persisted = await first_store.get_run(record.run_id)
+        page = await second_store.get_events(record.run_id)
+        return list(results), persisted.status, [event.type for event in page.events]
+
+    results, status, event_types = asyncio.run(scenario())
+
+    assert sum(result.applied for result in results) == 1
+    assert len(event_types) == 1
+    assert (status, event_types[0]) in {
+        ("succeeded", "run_succeeded"),
+        ("cancelled", "run_cancelled"),
+    }
+
+
+def test_failure_and_cancellation_compete_for_one_terminal() -> None:
+    redis = FakeRedis()
+    failure_store = RedisRunStore(redis, prefix="test")  # pyright: ignore[reportArgumentType]
+    cancellation_store = RedisRunStore(redis, prefix="test")  # pyright: ignore[reportArgumentType]
+
+    async def scenario() -> tuple[list[RunFinalizationResult], RunStatus, list[str]]:
+        record = await failure_store.create_run()
+        results = await asyncio.gather(
+            failure_store.finalize_run(
+                RunFailedEvent(
+                    run_id=record.run_id,
+                    data=RunFailedEventData(error="model failed", reason="model_error"),
+                )
+            ),
+            cancellation_store.finalize_run(
+                RunCancelledEvent(
+                    run_id=record.run_id,
+                    data=RunCancelledEventData(reason="concurrent_cancel"),
+                )
+            ),
+        )
+        persisted = await failure_store.get_run(record.run_id)
+        page = await cancellation_store.get_events(record.run_id)
+        return list(results), persisted.status, [event.type for event in page.events]
+
+    results, status, event_types = asyncio.run(scenario())
+
+    assert sum(result.applied for result in results) == 1
+    assert len(event_types) == 1
+    assert (status, event_types[0]) in {
+        ("failed", "run_failed"),
+        ("cancelled", "run_cancelled"),
+    }
+
+
+def test_finalize_run_raises_when_record_is_missing() -> None:
+    redis = FakeRedis()
+    store = RedisRunStore(redis, prefix="test")  # pyright: ignore[reportArgumentType]
+
+    with pytest.raises(RunNotFoundError):
+        asyncio.run(
+            store.finalize_run(RunCancelledEvent(run_id="missing", data=RunCancelledEventData(reason="cancelled")))
+        )
 
 
 def test_append_event_serializes_typed_event_without_id_and_expires_run_keys() -> None:
@@ -166,18 +351,19 @@ def test_get_events_round_trips_run_succeeded_output_and_session_snapshot() -> N
 
     async def scenario() -> tuple[str, RunSucceededEvent]:
         record = await store.create_run()
-        event_id = await store.append_event(
+        result = await store.finalize_run(
             RunSucceededEvent(
                 id="local-only",
                 run_id=record.run_id,
                 data=RunSucceededEventData(output=output, session_snapshot=session_snapshot),
             )
         )
+        assert result.event_id is not None
         page = await store.get_events(record.run_id, after="0-0", limit=10)
         decoded = page.events[0]
         assert isinstance(decoded, RunSucceededEvent)
-        assert page.next_cursor == event_id
-        return event_id, decoded
+        assert page.next_cursor == result.event_id
+        return result.event_id, decoded
 
     event_id, decoded = asyncio.run(scenario())