fix(agent): seal event stream after terminal run

[qdivan]

Summary

• atomically check the run status, append non-terminal events, and refresh TTLs in Redis
• reject cross-writer progress writes after a terminal state, without extending retention
• align the in-memory sink and stop runners when the event stream was already sealed

Test plan

• uv run pytest tests/local/dify_agent/storage/test_redis_run_store.py tests/local/dify_agent/runtime/test_runner.py -q
• make check

Closes #40765

[dosubot]

📄 Knowledge review

✏️ Suggested updates

1 page suggestion needs review.

Page	Library	Status
Dify Agent Server 模块分析 (Commit 55f95dbc)	dify	🟡 Review

📝 Dify Agent Server 模块分析 (Commit 55f95db)

@@ -159,8 +159,8 @@
 负责所有持久化操作：
 
 - **Run 记录**：以 JSON 字符串存储在 Redis Key 中，记录 `run_id`、`status`、`created_at`、`updated_at`、`error`
-- **事件流**：每个 run 对应一条 Redis Stream，事件以 `xadd` 追加写入
-- **TTL 刷新**：每次写入状态或事件时，同步刷新 run 记录和事件流的过期时间（默认保留 3 天）
+- **事件流**：每个 run 对应一条 Redis Stream，事件通过原子性 Lua 脚本追加写入（PR #40911）
+- **TTL 刷新**：仅在事件成功追加时（状态为 `"running"`）刷新 run 记录和事件流的过期时间（默认保留 3 天）。被拒绝的写入（密封后）不延长保留时间
 - **取消观察**：`wait_for_cancellation(run_id)` 方法阻塞等待取消或另一个终端状态赢得竞争。使用 Redis Stream 阻塞读取观察终端事件
 - **事件迭代**：`iter_events(run_id, after)` 先回放已有事件，再以阻塞 `XREAD` 读取实时事件。**迭代在产出第一个终端事件（run_succeeded、run_failed 或 run_cancelled）后立即返回**，无论该事件来自回放还是实时推送。方法不再在终端事件后继续阻塞循环
 
@@ -823,11 +823,13 @@
    - 确保 runner_task 被取消（若仍在运行）
 ```
 
-`AgentRunRunner.run()` 执行流程不变 [[39]](https://github.com/langgenius/dify/blob/7210f856c9c07ae03d7c2e5def96c949efee6397/dify-agent/src/dify_agent/runtime/runner.py#L77-L96)：
+`AgentRunRunner.run()` 执行流程（PR #40911 更新）[[39]](https://github.com/langgenius/dify/blob/7210f856c9c07ae03d7c2e5def96c949efee6397/dify-agent/src/dify_agent/runtime/runner.py#L77-L96)：
 
 ```
 1. emit_run_started → 写入 run_started 事件到 Redis Stream
+   （若抛出 RunSealedError，runner 立即早停并返回）
 2. _run_agent() 核心执行
+   （若抛出 RunSealedError，runner 立即早停并返回）
 3. 成功路径：finalize_run(SUCCEEDED)
 4. 失败路径：finalize_run(FAILED)
 5. 取消路径：通过 is_cancelled() 回调检测，调用 finalize_run(CANCELLED)
@@ -852,6 +854,29 @@
 ## 事件模型
 
 `dify-agent` 使用**追加写入的事件日志**作为运行观察的唯一接口。所有事件通过 Redis Streams 持久化，客户端通过轮询或 SSE 消费 [[42]](https://github.com/langgenius/dify/blob/7210f856c9c07ae03d7c2e5def96c949efee6397/dify-agent/src/dify_agent/protocol/schemas.py#L1-L52)。
+
+### 事件流密封与写入拒绝
+
+PR #40911 引入了**事件流密封机制**：非终端事件在运行处于 `running` 状态时保持追加写入语义，但一旦终端事件（`run_succeeded`、`run_failed` 或 `run_cancelled`）成功持久化，后续的非终端事件写入将被拒绝。
+
+**`RunSealedError` 异常**：当非终端事件失败于终端转换时抛出，包含密封运行的终端状态字段（`status: RunStatus`）。
+
+**InMemoryRunEventSink 行为变更**：`append_event()` 方法现在在追加事件前检查运行状态是否为 `"running"`，若运行已处于终端状态则抛出 `RunSealedError`。
+
+**RedisRunStore 原子性检查**：`append_event()` 实现现在使用 Lua 脚本 `_APPEND_EVENT_SCRIPT` 原子性地检查运行状态后再追加事件。脚本返回三种结果：
+- `-1`：run 记录不存在，抛出 `RunNotFoundError`
+- `0`：run 已密封（终端状态），抛出 `RunSealedError`
+- `1`：成功追加事件，返回事件 ID
+
+**TTL 刷新策略调整**：TTL 仅在事件成功追加时刷新（状态为 `"running"`），被拒绝的写入（密封后）不再延长保留时间。
+
+**Runner 早停行为**：`AgentRunRunner` 现在在两处捕获 `RunSealedError`：
+1. 发出 `run_started` 事件时
+2. Agent 执行期间（`_run_agent()` 调用时）
+
+当捕获到 `RunSealedError` 时，runner 早停并返回（不进行进一步处理），因为事件流已被另一个 writer 密封。
+
+这些变更实现了跨写入者的进度写入拒绝，改善了一致性并防止了多个 writer 在运行完成后追加事件的竞态条件。
 
 ### 事件信封结构
 

Accept · Edit · Decline

---