HCGStudio · mahoshojoHCG · Aug 29, 2026 · Aug 30, 2026
diff --git a/Plugins/SecondDimensionWatcherReDive.AI/Engines/CodexAppServerEngine.cs b/Plugins/SecondDimensionWatcherReDive.AI/Engines/CodexAppServerEngine.cs
@@ -226,14 +226,14 @@ await rpc.SendRequestAsync(
             turnId = GetRequiredNestedString(turnResult, "turn", "id");
             state.SetTurnId(turnId);
 
-            while (state.Updates.TryDequeue(out var bufferedUpdate))
+            while (state.TryDequeueUpdate(out var bufferedUpdate))
                 yield return bufferedUpdate;
 
             while (!state.IsTerminal)
             {
                 var message = await rpc.ReceiveAsync(cancellationToken);
                 await ProcessTurnMessageAsync(rpc, state, message, cancellationToken);
-                while (state.Updates.TryDequeue(out var update))
+                while (state.TryDequeueUpdate(out var update))
                     yield return update;
             }
 
@@ -488,8 +488,8 @@ await rpc.SendErrorAsync(requestId, -32602,
         }
 
         var argumentsJson = arguments.GetRawText();
-        state.Updates.Enqueue(new ToolCallBegin(callId, toolName));
-        state.Updates.Enqueue(new ToolCallDelta(callId, argumentsJson));
+        state.TryEnqueueUpdate(new ToolCallBegin(callId, toolName));
+        state.TryEnqueueUpdate(new ToolCallDelta(callId, argumentsJson));
 
         IToolResult toolResult;
         var executionCanceled = false;
@@ -516,7 +516,7 @@ await rpc.SendErrorAsync(requestId, -32602,
 
         var serializedResult = JsonSerializer.SerializeToElement(
             toolResult, toolResult.GetType(), ToolJsonOptions.Options);
-        state.Updates.Enqueue(new ToolResultUpdate(callId, serializedResult));
+        state.TryEnqueueUpdate(new ToolResultUpdate(callId, serializedResult));
 
         if (executionCanceled || cancellationToken.IsCancellationRequested)
         {
@@ -827,17 +827,18 @@ private sealed class TurnState(
         IToolExecutor? toolExecutor,
         int maxDynamicToolCalls)
     {
+        private const int MaxBufferedUpdates = 1024;
         private readonly Dictionary<string, string> _finalAgentTexts =
             new(StringComparer.Ordinal);
         private readonly HashSet<string> _completedFinalAgentItems =
             new(StringComparer.Ordinal);
         private readonly HashSet<string> _toolCallIds = new(StringComparer.Ordinal);
+        private readonly Queue<IChatUpdate> _updates = new();
         private int _dynamicToolCallCount;
 
         public string ThreadId { get; } = threadId;
         public string? TurnId { get; private set; }
         public IToolExecutor? ToolExecutor { get; } = toolExecutor;
-        public Queue<IChatUpdate> Updates { get; } = new();
         public bool IsTerminal { get; private set; }
         public bool ServerTerminalReceived { get; private set; }
         public string? Status { get; private set; }
@@ -854,7 +855,7 @@ public void AppendFinalAgentDelta(string itemId, string delta)
 
             _finalAgentTexts[itemId] = accumulated + delta;
             if (delta.Length > 0)
-                Updates.Enqueue(new TextDelta(delta));
+                TryEnqueueUpdate(new TextDelta(delta));
         }
 
         public void CompleteFinalAgentItem(string itemId, string authoritativeText)
@@ -870,9 +871,24 @@ public void CompleteFinalAgentItem(string itemId, string authoritativeText)
             var suffix = authoritativeText[accumulated.Length..];
             _finalAgentTexts[itemId] = authoritativeText;
             if (suffix.Length > 0)
-                Updates.Enqueue(new TextDelta(suffix));
+                TryEnqueueUpdate(new TextDelta(suffix));
         }
 
+        public bool TryEnqueueUpdate(IChatUpdate update)
+        {
+            if (_updates.Count < MaxBufferedUpdates)
+            {
+                _updates.Enqueue(update);
+                return true;
+            }
+
+            Fail($"Codex app-server buffered update limit ({MaxBufferedUpdates}) was exceeded.");
+            return false;
+        }
+
+        public bool TryDequeueUpdate(out IChatUpdate update) =>
+            _updates.TryDequeue(out update!);
+
         public bool TryBeginToolCall(string callId, out string rejectionReason)
         {
             if (!_toolCallIds.Add(callId))

diff --git a/Plugins/SecondDimensionWatcherReDive.Chat/ChatController.cs b/Plugins/SecondDimensionWatcherReDive.Chat/ChatController.cs
@@ -179,7 +179,7 @@ public async Task<IResult> SendMessage(
                 cancellationToken));
     }
 
-    private async IAsyncEnumerable<SseItem<string>> StreamChatEvents(
+    internal async IAsyncEnumerable<SseItem<string>> StreamChatEvents(
         IAIEngine aiEngine,
         List<IMessage> messages,
         ChatOptions chatOptions,
@@ -190,15 +190,23 @@ private async IAsyncEnumerable<SseItem<string>> StreamChatEvents(
         string? model,
         [EnumeratorCancellation] CancellationToken cancellationToken)
     {
-        var channel = Channel.CreateUnbounded<SseItem<string>>();
+        var channel = Channel.CreateBounded<SseItem<string>>(
+            new BoundedChannelOptions(256)
+            {
+                SingleReader = true,
+                SingleWriter = true,
+                FullMode = BoundedChannelFullMode.Wait
+            });
+        using var producerCancellation = CancellationTokenSource.CreateLinkedTokenSource(
+            cancellationToken);
 
         // Producer: runs AI chat streaming in background, writes SSE items to channel.
         // Keep the task and await it during iterator disposal so a disconnected request cannot
         // release this controller's scoped repository before tool-call audit records are saved.
         var producer = ProduceChatEventsAsync(
             aiEngine, messages, chatOptions, conversationId, messageOrder,
             firstUserMessage, autoTitleEligible, model,
-            channel.Writer, cancellationToken);
+            channel.Writer, producerCancellation.Token);
 
         try
         {
@@ -208,13 +216,15 @@ private async IAsyncEnumerable<SseItem<string>> StreamChatEvents(
         }
         finally
         {
-            // Do not use the canceled request token here. The producer receives it directly,
-            // performs bounded engine cleanup, and persists accumulated messages before returning.
+            // The reader can be disposed independently of RequestAborted (for example when
+            // response-body I/O fails). Explicitly stop the producer, then let it persist
+            // accumulated messages before this request scope is released.
+            await producerCancellation.CancelAsync();
             await producer;
         }
     }
 
-    private async Task ProduceChatEventsAsync(
+    internal async Task ProduceChatEventsAsync(
         IAIEngine aiEngine,
         List<IMessage> messages,
         ChatOptions chatOptions,
@@ -328,12 +338,14 @@ await writer.WriteAsync(
         catch (Exception ex)
         {
             LogStreamingError(ex, conversationId);
-            await writer.WriteAsync(
+            // A disconnected/failed reader may leave this bounded channel full.
+            // Terminal error delivery is best-effort so persistence and request
+            // scope disposal can never wait forever for a reader that is gone.
+            writer.TryWrite(
                 new SseItem<string>(
                     JsonSerializer.Serialize(new SseError(ex.Message),
                         ChatJsonSerializerContext.Default.SseError),
-                    "error"),
-                CancellationToken.None);
+                    "error"));
         }
 
         // Save all accumulated segments to DB

diff --git a/Plugins/SecondDimensionWatcherReDive.Chat/Tools/ManageTasksTool.cs b/Plugins/SecondDimensionWatcherReDive.Chat/Tools/ManageTasksTool.cs
@@ -9,9 +9,10 @@ namespace SecondDimensionWatcherReDive.Chat.Tools;
     "manage_tasks",
     "Manage background scheduled tasks. List all task statuses or manually trigger a specific task to run.")]
 internal sealed partial class ManageTasksTool(
-    IEnumerable<IScheduledTask> scheduledTasks) : ITool
+    IEnumerable<IScheduledTask> scheduledTasks,
+    IScheduledTaskLeaseManager leaseManager) : ITool
 {
-    private Task<IToolResult> ExecuteCoreAsync(
+    private async Task<IToolResult> ExecuteCoreAsync(
         ManageTasksParams param, CancellationToken cancellationToken)
     {
         var taskList = scheduledTasks.ToList();
@@ -20,10 +21,24 @@ private Task<IToolResult> ExecuteCoreAsync(
         switch (param.Action)
         {
             case ManageTasksAction.List:
+            {
+                var statuses = await leaseManager.GetStatusesAsync(
+                    taskList.Select(task => task.Id).ToArray(),
+                    cancellationToken);
                 result = new ToolSuccessResult<TaskListResult>(new TaskListResult(
-                    taskList.Select(t => new TaskSummary(
-                        t.Id, t.Interval.ToString(), t.IsEnabled, t.LastRunAt, t.IsRunning))));
+                    taskList.Select(task =>
+                    {
+                        var status = statuses.GetValueOrDefault(task.Id)
+                                     ?? new ScheduledTaskStatus(null, false);
+                        return new TaskSummary(
+                            task.Id,
+                            task.Interval.ToString(),
+                            task.IsEnabled,
+                            status.LastRunAt,
+                            status.IsRunning);
+                    })));
                 break;
+            }
 
             case ManageTasksAction.Run:
             {
@@ -52,7 +67,7 @@ private Task<IToolResult> ExecuteCoreAsync(
                 break;
         }
 
-        return Task.FromResult(result);
+        return result;
     }
 }
 

diff --git a/README.md b/README.md
@@ -36,6 +36,8 @@
 - [x] 按动画分组的主页展示（卡片 + 剧集列表）
 - [x] 当季番组发现（mikanani.me 爬取）+ 一键订阅
 - [x] 后台任务仪表盘（查看状态、手动触发）
+- [x] PostgreSQL 持久任务 / Outbox（崩溃恢复、指数重试、死信处理、多实例租约）
+- [x] 存活与就绪探针、Prometheus 指标及 OpenTelemetry 链路
 - [x] 一次性数据迁移框架（`MigrationMarkers` 表幂等记录）
 - [x] 插件事件系统（下载前 / 下载完成后钩子）
 - [x] 多语言界面（简体中文 / English / 日本語）
@@ -105,11 +107,15 @@ bash <(curl -fsSL https://raw.githubusercontent.com/HCGStudio/SecondDimensionWat
 | `AI:CodexAppServer:Endpoint` / `BearerToken` / `Model` / `PermissionProfile` / `TimeoutSeconds` | Codex app-server WebSocket 端点；空模型使用服务端默认模型；权限配置默认 `:read-only`，也可填写管理员定义的 profile id |
 | `Inference:RateLimitDelayMs` | 推断 API 调用最小间隔（毫秒，默认 1000） |
 | `Valkey:ConnectionString` | Valkey / Redis 连接（可选；为空则使用内存缓存） |
+| `Health:ValkeyRequired` / `QbittorrentRequired` / `StorageRequired` / `AIRequired` | `/health/ready` 的依赖要求；PostgreSQL 始终必需，AI 默认不阻塞就绪 |
+| `OpenTelemetry:OtlpEndpoint` | 可选 OTLP Collector 地址；Prometheus `/metrics` 无需配置即启用 |
 
 > 使用现有媒体库导入前，必须至少配置一个 `MediaLibrary:AllowedRoots`。导入源必须位于白名单内，且不能与 `FileStore:Local` 管理的下载目录相同、互为父目录或以其他方式重叠。导入与后续对账只会修改数据库中的媒体记录和虚拟路径映射；系统绝不会移动、重命名或删除原文件。短暂缺失的条目会先撤下映射并保留观看/审核记录，超过 `MissingGracePeriod`（默认 24 小时）后才清理数据库记录。
 
 > 从 v2.2 之前升级：旧的 `Inference:ApiKey/Provider/Model` 已迁移到 `AI:` 前缀。运行 `deployments/migrate-config.sh` 自动迁移；包管理器安装时 `postinstall.sh` 会自动执行。
 
+运行探针、持久任务恢复、死信操作和遥测标签约束详见 [运行可靠性与可观测性](docs/runtime-reliability.md)。
+
 ### 网页运行时设置
 
 登录后打开「设置」，可修改 AI 执行模式与 Provider、AI/TMDB 密钥、qBittorrent、媒体库扫描、异常检测和 NFS。保存值存入 PostgreSQL，并覆盖部署文件或环境变量中的默认值；密钥和密码使用持久化 Data Protection 密钥环加密，API 不会回显明文。可对单个敏感项选择保留、替换、清除或恢复部署默认值。

diff --git a/SecondDimensionWatcherReDive.Client/mock-server.mjs b/SecondDimensionWatcherReDive.Client/mock-server.mjs
@@ -2589,10 +2589,53 @@ async function route(method, pathname, searchParams, req, res) {
     },
   ];
 
+  const mockDurableJobs =
+    globalThis._mockDurableJobs ??
+    (globalThis._mockDurableJobs = [
+      {
+        id: randomUUID(),
+        type: "downloadCompletion",
+        status: "deadLetter",
+        stage: "mapFiles",
+        attemptCount: 8,
+        createdAt: new Date(Date.now() - 3_600_000).toISOString(),
+        updatedAt: new Date(Date.now() - 60_000).toISOString(),
+        nextAttemptAt: new Date(Date.now() - 60_000).toISOString(),
+        lastAttemptAt: new Date(Date.now() - 60_000).toISOString(),
+        completedAt: null,
+        lastError: "InvalidOperationException: No file mapping could be produced.",
+      },
+    ]);
+
   if (method === "GET" && pathname === "/api/tasks") {
     return json(res, MOCK_TASKS);
   }
 
+  if (method === "GET" && pathname === "/api/jobs") {
+    const status = searchParams.get("status");
+    const items = status
+      ? mockDurableJobs.filter(
+          (job) => job.status.toLowerCase() === status.toLowerCase(),
+        )
+      : mockDurableJobs;
+    return json(res, { items, totalCount: items.length });
+  }
+
+  if (
+    method === "POST" &&
+    (pathname === "/api/jobs/retry" || pathname === "/api/jobs/resolve")
+  ) {
+    const body = await readBody(req);
+    const ids = new Set(Array.isArray(body.ids) ? body.ids : []);
+    let affectedCount = 0;
+    for (let index = mockDurableJobs.length - 1; index >= 0; index--) {
+      if (!ids.has(mockDurableJobs[index].id)) continue;
+      mockDurableJobs.splice(index, 1);
+      affectedCount++;
+    }
+    return json(res, { affectedCount });
+  }
+
   // POST /api/tasks/:id/run
   {
     const m = pathname.match(/^\/api\/tasks\/(.+)\/run$/);

diff --git a/SecondDimensionWatcherReDive.Client/src/i18n/locales/en/tasks.json b/SecondDimensionWatcherReDive.Client/src/i18n/locales/en/tasks.json
@@ -29,6 +29,37 @@
     "success": "Task \"{{name}}\" finished",
     "failure": "Task \"{{name}}\" failed"
   },
+  "deadLetters": {
+    "title": "Failed durable jobs",
+    "description": "Jobs that exhausted automatic retries. Retry them after fixing the cause, or mark them handled.",
+    "empty": "No durable jobs require attention.",
+    "loadFailed": "Could not load failed durable jobs.",
+    "retry": "Retry",
+    "resolve": "Mark handled",
+    "columns": {
+      "type": "Job",
+      "stage": "Failed stage",
+      "attempts": "Attempts",
+      "updated": "Last attempt",
+      "error": "Last error",
+      "actions": "Actions"
+    },
+    "types": {
+      "downloadCompletion": "Download completion"
+    },
+    "stages": {
+      "mapFiles": "Map files",
+      "notify": "Notify",
+      "invokePlugins": "Invoke plugins",
+      "done": "Finalize"
+    },
+    "toast": {
+      "retrySuccess": "Job queued for retry",
+      "retryFailure": "Could not retry the job",
+      "resolveSuccess": "Job marked handled",
+      "resolveFailure": "Could not mark the job handled"
+    }
+  },
   "metadata": {
     "SyncFeed": {
       "name": "SyncFeed",

diff --git a/SecondDimensionWatcherReDive.Client/src/i18n/locales/ja/tasks.json b/SecondDimensionWatcherReDive.Client/src/i18n/locales/ja/tasks.json
@@ -26,6 +26,37 @@
     "success": "タスク「{{name}}」を完了しました",
     "failure": "タスク「{{name}}」が失敗しました"
   },
+  "deadLetters": {
+    "title": "失敗した永続ジョブ",
+    "description": "自動再試行を使い切ったジョブです。原因を修正して再試行するか、処理済みにしてください。",
+    "empty": "対応が必要な永続ジョブはありません。",
+    "loadFailed": "失敗した永続ジョブを読み込めませんでした。",
+    "retry": "再試行",
+    "resolve": "処理済みにする",
+    "columns": {
+      "type": "ジョブ",
+      "stage": "失敗ステージ",
+      "attempts": "試行回数",
+      "updated": "最終試行",
+      "error": "最終エラー",
+      "actions": "操作"
+    },
+    "types": {
+      "downloadCompletion": "ダウンロード完了処理"
+    },
+    "stages": {
+      "mapFiles": "ファイルのマッピング",
+      "notify": "通知",
+      "invokePlugins": "プラグインの呼び出し",
+      "done": "完了"
+    },
+    "toast": {
+      "retrySuccess": "ジョブを再試行キューに追加しました",
+      "retryFailure": "ジョブを再試行できませんでした",
+      "resolveSuccess": "ジョブを処理済みにしました",
+      "resolveFailure": "ジョブを処理済みにできませんでした"
+    }
+  },
   "metadata": {
     "SyncFeed": {
       "name": "SyncFeed",