geo/docs/ai-brand-monitoring-priority-dispatch-design-v1.1.md

# AI 品牌监测优先级调度与立即采集抢占方案 V1.1

## 1. 文档信息

| 项目 | 内容 |
| --- | --- |
| 文档名称 | AI 品牌监测优先级调度与立即采集抢占方案 V1.1 |
| 文档版本 | V1.1 |
| 文档状态 | 修订完成，待二次评审 |
| 创建日期 | 2026-04-20 |
| 修订日期 | 2026-04-20 |
| 上一版本 | `docs/ai-brand-monitoring-priority-dispatch-design-v1.md`（V1.0） |
| 适用范围 | `monitoring` 采集链路、`scheduler`、`tenant-api`、`desktop client` |
| 关联文档 | `docs/ai-brand-monitoring-tech-design-v5.md` |
| 关联文档 | `docs/geo-platform-ops-admin-tech-architecture-v1.md` |
| 当前基线 | `desktop client (pull-lease) + monitoring_collect_tasks + RabbitMQ result ingest + projection rebuild` |

## 2. 修订记录（V1.0 → V1.1）

本版本针对 V1.0 架构评审中的 5 项 Critical 与 7 项 Warning 做全面修订：

| 编号 | 原问题 | 修订章节 | 修订摘要 |
| --- | --- | --- | --- |
| C1 | `status` 与 `execution_status` 二选一决策留白 | §7.1 | 明确不新增 `execution_status`；Phase 1 继续复用现有 `status`，Phase 2 执行权威迁至 `desktop_tasks` |
| C2 | `interrupt_generation` 语义未定义，存在抢占竞态 | §10.2 | 新增代际规范：归属、递增规则、事件携带、回写比对 |
| C3 | Phase 1→2 迁移期双写一致性空白 | §16.2 | 引入 `execution_owner` 路由字段 + 灰度 + 回滚 + reconcile |
| C4 | 抢占 15s/60s 硬编码无经验基线 | §10.5 | 改为 per-platform 可配置参数，附测量落地流程 |
| C5 | dispatch 预算四维耦合关系未定义 | §9.2 | 新增决策树、拒绝处置矩阵、high lane 绕过规则 |
| W1 | `target_client_id` 设备漂移场景未覆盖 | §10.6 | 新增绑定时机、选择规则、漂移 4 场景处置、TTL |
| W2 | aging 简单加分会集体失效 | §9.4 | 改为百分位触发 + 对数衰减 |
| W3 | publish/monitor 共表 priority 值域未规范 | §7.3 | 新增跨 kind priority 值域表 + lane_weight 排序 |
| W4 | 伪代码 16.1 单大事务 + 外部副作用 | §17.1 | 改为 outbox 模式，短事务 + 独立 pump |
| W5 | Phase 1 抢占语义与双层模型矛盾 | §16.1 | 收敛 Phase 1 范围：只抢占 queued/leased，不抢 running |
| W6 | desktop pull→push 过渡期通道空窗 | §14.3 §16.1 | **Phase 1 即启用 WebSocket 通道，SSE `/api/desktop/events` Phase 1 完成后立即下线**（修订 2：用户明确要求早期拿掉 SSE） |
| W7 | 幂等键与现有表约束未对齐 | §12.3 | 对齐 `question_monitor_runs`、`projection_rebuild_queue` 现有键 |
| I1 | §6.1 当前态描述偏旧 | §6.1 | 更新 `monitor.projection.rebuild` 已存在 |
| I2 | `ready` 与 `pending` 混用 | 全文 | 统一为 `pending` |
| I3 | `collection_mode` 字段语义在插件下线后歧义 | §13.1 | 保留用于日志审计，文档明确说明 |
| I4 | 观测指标不足 | §15.1 | 补充 `preempt_induced_retry`、`safe_point_interval`、`dispatch_reject_by_reason` 等 |

## 3. 背景与问题

当前监测采集已经具备以下基础能力：

1. `collect-now` 已存在入口，能够把当天任务重置回可执行状态。
2. desktop client 已具备任务租约、续租、回写结果、使用 WS 推送等基础设施（monitor 仍以 pull-lease 为主，其实是无法正常使用的，不用太在意；publish 已使用 WS push（可用，体验非常好）；旧 SSE `/api/desktop/events` **Phase 1 即下线**，所有事件统一走 WebSocket）。
3. 结果上报已走 `API -> RabbitMQ -> ingest worker -> Monitoring PG` 的异步写链路，`monitor.projection.rebuild` worker 与 `MonitoringProjectionRebuildQueue` 已存在并完成聚合投影合并。

但当前实现仍存在几个关键缺口：

1. `collect-now` 只能重置 `pending / expired / failed / completed / skipped`，不会抢占已经 `leased / running` 的定时任务。
2. monitoring 任务当前仍以"按 `planned_at` 顺序从 DB 拉取"作为主调度信号，缺少显式优先级和抢占语义。
3. monitoring 尚未像 publish 一样迁移到 `RabbitMQ -> tenant-api -> desktop dispatch websocket` 的低延迟推送链路。
4. 定时任务如果按分钟大批量扫到期后直接下发，容易把 desktop client、结果回写链路和数据库同时打满。
5. 结果入库虽然已异步化，但"原始事实写入"的分片串行尚未落地，高峰期仍有打挂库风险。

本方案要解决的核心问题是：

1. 每日定时采集如何稳态运行，不形成瞬时巨量计算洪峰。
2. `collect-now` 如何保证明显高于定时任务的体验优先级。
3. 当定时任务正在执行时，如何安全让路给 `collect-now`。
4. 结果如何在不牺牲吞吐的前提下做到局部有序、幂等落库。

## 4. 目标与非目标

### 4.1 目标

1. `collect-now` 从点击到 desktop client 开始执行的链路尽量走低延迟通道。
2. `collect-now` 对同一 client 上的定时监测任务具备抢占权。
3. 定时调度采用"分批 claim + 按预算 dispatch"，而不是"扫到即全量下发"。
4. 数据库只承担真相存储与有界写入，不承担大规模实时排队压力。
5. 结果入库按业务分片串行，局部有序，不要求全局单线程。
6. 全链路具备幂等、回放、超时回收、死信与可观测性。

### 4.2 非目标

1. 不要求一次 `collect-now` 抢占整个租户所有后台任务；默认只抢占目标 client 上的定时监测任务。
2. 不要求依赖 RabbitMQ 单队列优先级特性解决全部排序问题（没顺序要求）。
3. 不要求 desktop client 直接连接 RabbitMQ（不可以直接连接 RabbitMQ）。
4. 不在本方案内重做全部监测业务口径，仅聚焦任务控制面与执行链路。

## 5. 核心结论

### 5.1 任务真相源

推荐采用"双层任务模型"（Phase 2 全量引入）：

1. `monitoring_collect_tasks` 继续作为"业务任务真相源"，表示某品牌、某问题、某平台、某业务日是否需要采集。
2. `desktop_tasks(kind=monitor)` 作为"执行任务控制面"，承接优先级、抢占、租约、取消、推送与恢复。

换句话说：

- `monitoring_collect_tasks` 负责"要不要做"
- `desktop_tasks(kind=monitor)` 负责"怎么派发给哪台 desktop 做"

这样可以复用现有 publish 的成熟能力，而不是让 monitoring 永久维持一套独立调度机制。

**过渡期说明**：Phase 1 不引入 `desktop_tasks(kind=monitor)`，执行态从 `monitoring_collect_tasks` 的 `lease_expires_at / leased_to_executor` 推导；Phase 2 按 tenant 灰度迁移，详见 §16.2。

### 5.2 RabbitMQ 的角色

RabbitMQ 只负责运输、削峰、解耦和异步化，不作为唯一真相源。

推荐边界：

1. 下行：`DB claim -> RabbitMQ dispatch event -> tenant-api dispatch hub -> desktop websocket`
2. 上行：`desktop callback -> tenant-api receive -> RabbitMQ result ingest -> worker -> Monitoring PG`

不推荐 desktop client 直连 RabbitMQ，原因：

1. 证书与凭证分发复杂。
2. 网络拓扑和私有化部署适配更难。
3. 权限收敛、审计、灰度、风控和回放都不如经由 API 网关统一处理稳定。

### 5.3 优先级策略

推荐采用"多队列 + 服务端调度预算"：

1. `high`：`collect-now`
2. `normal`：每日定时
3. `retry`：失败重试、过期回收

不要依赖 RabbitMQ 单队列 `x-max-priority` 解决一切问题。大厂常见做法是多队列分层，再由业务侧决定是否暂停 `normal` lane。

### 5.4 抢占策略

抢占采用"协作式抢占"，不做粗暴 kill：

1. `queued` 的执行任务：直接取消或降级回 `normal` 队列。
2. `leased` 但尚未开始浏览器交互：不在服务端盲目清租约，而是下发 `task_control`；desktop 只在本地单线程调度循环中、且 `executionPhase in (PREPARING, NAVIGATING)` 时才主动 cancel lease 并回队（见 §14.2 / §14.4）。
3. `running` 的执行任务（**Phase 2 起支持**）：下发 `interrupt_requested`，desktop client 在安全点主动中断并回写 `aborted(preempted)`。
4. 如果 `interrupt_soft_timeout` 内无法安全中断，则禁止新的 `normal` 任务继续进入该 client，待当前任务结束后立即执行 `collect-now`。

本文中“尚未真正开始执行”严格等价于 desktop 本地 `executionPhase in (PREPARING, NAVIGATING)`；自 `AUTHENTICATING` 起一律视为“已开始”，Phase 1 不再直接 cancel，Phase 2 改为在安全点处理中断。

### 5.5 数据库保护策略

数据库保护依赖四层手段：

1. 调度限流：到期任务先 claim，不立刻全量 dispatch。
2. 写链路拆分：原始事实写入与聚合重算分开（已落地）。
3. 分片串行：按 `(tenant_id, brand_id, business_date)` 做结果分片。
4. 局部重算：只重算受影响的品牌/日期快照，不做全局全表聚合（已落地）。

## 6. 当前态与目标态

### 6.1 当前态（V1.1 更正）

```text
schedule/collect-now API (status reset only)
  -> monitoring_collect_tasks (status=pending)
  -> desktop client 定时 pull lease
  -> desktop 执行
  -> API callback
  -> RabbitMQ monitor.result.ingest (已存在)
  -> ingest worker (已存在)
  -> Monitoring PG raw facts
  -> RabbitMQ monitor.projection.rebuild (已存在)
  -> projection rebuild worker (已存在，带 coalesce)
  -> monitoring_brand_daily / monitoring_brand_platform_daily
```

特点（实际代码核实）：

1. 上行结果链路合理，projection rebuild 已带 coalesce。
2. 下行仍偏 DB 轮询，无 push 推送。
3. `collect-now` 目前只做"重置可执行状态"，不做优先级提升也不做抢占（§`server/internal/tenant/app/monitoring_service.go` 中 `CollectNow` 方法）。
4. SSE `/api/desktop/events` 存在且仍在使用，**Phase 1 即下线**，所有事件统一走 WebSocket `desktop.task.dispatch`。
5. `desktop_tasks` 表已存在并支持 `kind=publish \| monitor`，但 monitor kind 目前**未被任何调度逻辑使用**。

### 6.2 推荐目标态

```text
schedule scanner / collect-now API
  -> monitoring_collect_tasks (业务真相，带 lane/priority/target_client_id)
  -> monitor dispatch coordinator (按预算 claim execution slots)
  -> desktop_tasks(kind=monitor) (执行控制面)
  -> RabbitMQ desktop.task.dispatch (复用 publish 通道)
  -> tenant-api dispatch hub
  -> desktop websocket
  -> desktop lease / execute / extend / cancel / complete
  -> monitor result receive API
  -> RabbitMQ monitor.result.ingest (分片串行消费)
  -> ingest worker
  -> Monitoring PG raw facts
  -> RabbitMQ monitor.projection.rebuild (已有 coalesce)
  -> projection worker
  -> monitoring_brand_daily / monitoring_brand_platform_daily
```

### 6.3 渐进落地原则

推荐分三阶段，详见 §16：

1. Phase 1：业务任务表加优先级字段 + collect-now 高优提升 + queued/leased 抢占（不抢 running）+ 直接复用 WebSocket 下发 `task_available / task_control`，SSE 同步下线。
2. Phase 2：引入 `desktop_tasks(kind=monitor)` + 收敛执行控制面 + 支持 running 任务抢占。
3. Phase 3：ingest 分片串行 + 多租户公平调度 + aging + 自动降速。

## 7. 任务模型设计

### 7.1 业务任务：`monitoring_collect_tasks`（Phase 1 新增字段）

**架构决策（C1）**：**不新增** `execution_status` 字段，但必须明确“谁是唯一执行真相源”。Phase 1 沿用现有 `status` 承载 legacy 执行流转；Phase 2 当 `execution_owner='desktop_tasks'` 时，`desktop_tasks.status` 成为**唯一执行真相**，`monitoring_collect_tasks.status` 只保留业务终态，不再镜像中间执行态，也不再作为新路径的租约判断依据。

`status` 兼容值保持与当前实现一致：

| 值 | 语义 |
| --- | --- |
| `pending` | 待执行或待再次派发 |
| `leased` | 已被 desktop 领取租约（legacy 路径） |
| `received` | desktop 已回调结果，等待 ingest worker 最终落库 |
| `completed` | 今天已完成 |
| `failed` | 今天尝试但失败（达到最大重试） |
| `skipped` | 今天跳过（配额/关闭/排除） |
| `expired` | 今天到期未完成 |

约束说明：

1. `execution_owner='legacy'` 时，`LeaseTasks / ResumeTasks / expire / result callback` 继续依赖本表 `status`，保持与现网代码一致。
2. `execution_owner='desktop_tasks'` 时，执行归属、抢占、续租、取消与中间执行态一律以 `desktop_tasks.status` 为准；`monitoring_collect_tasks.status` 只允许保留 `pending / completed / failed / skipped / expired` 五类业务态，**禁止**写入 `leased / received`。
3. Phase 2 迁移期不得同时让两张表都成为“执行真相源”，否则会出现双状态竞争。
4. Phase 2 灰度仅影响新生成任务；已 in-flight 的 `legacy` 存量任务继续沿用旧写法直至自然结束，不做半程切换。

**Phase 2 状态写入矩阵（强制约束）**：

| 场景 | `execution_owner='legacy'` | `execution_owner='desktop_tasks'` |
| --- | --- | --- |
| dispatch / lease 中 | `monitoring_collect_tasks.status` 可写 `leased` | `monitoring_collect_tasks.status` 保持 `pending`；执行细节写 `desktop_tasks.status` |
| desktop 回调结果已接收 | `monitoring_collect_tasks.status` 可写 `received` | `monitoring_collect_tasks.status` 仍保持 `pending`；原始结果照常入 MQ，等待最终终态 |
| 最终成功 | `monitoring_collect_tasks.status = completed` | `monitoring_collect_tasks.status = completed` |
| 最终失败 / 跳过 / 过期 | `failed / skipped / expired` | `failed / skipped / expired` |

**下游兼容策略**：

| 下游 | 读取策略 |
| --- | --- |
| UI | `monitoring_collect_tasks.status` 仅用于“是否已完成/是否终态”判断；当前执行细节读取 active `desktop_tasks.status` |
| 报表 | 以 `monitoring_collect_tasks` 为业务主表；若需要执行维度（排队、执行中、重试中），显式 join `desktop_tasks` |
| 审计 | 业务审计读 `monitoring_collect_tasks`；执行与 attempt 细节读 `desktop_tasks` / `desktop_task_attempts` |
| reconcile | `execution_owner='desktop_tasks'` 时只对账“是否存在 active desktop_task / 是否收敛为正确终态”，不再比较 `leased / received` 之类中间态 |

**术语约定**：

1. 业务任务层使用 `completed`，表示“该业务日已完成采集”。
2. 执行任务层与 `monitoring_collect_requests` 使用 `succeeded`，表示“本次执行 / 本次请求成功结束”。
3. 两套命名是刻意区分业务语义与执行语义，不是 typo。

新增字段：

| 字段 | 类型 | 说明 |
| --- | --- | --- |
| `trigger_source` | varchar(20) | `automatic` / `manual`（已存在，保留） |
| `dispatch_priority` | int | 服务端调度排序值，见 §7.3 值域规范 |
| `dispatch_lane` | varchar(20) | `high` / `normal` / `normal_boosted` / `retry` |
| `target_client_id` | uuid nullable | 绑定的 desktop client，见 §10.6 |
| `dispatch_after` | timestamptz | 不早于该时间派发 |
| `interrupt_generation` | int | 抢占代际，详见 §10.2 |
| `superseded_by_request_id` | uuid nullable | 被哪次手动请求覆盖 |
| `last_dispatched_at` | timestamptz nullable | 最近一次派发时间 |
| `dispatch_attempts` | int | 派发次数 |
| `ingest_shard_key` | varchar(120) | 结果入库分片键，见 §12.2 |
| `execution_owner` | varchar(20) | `legacy` / `desktop_tasks`，Phase 1→2 过渡路由字段，见 §16.2 |

说明：

1. `dispatch_priority` 不直接等同于 RabbitMQ 优先级，而是服务端调度排序依据。
2. `target_client_id` 在 `collect-now` 场景必须明确绑定到当前用户在线 client；普通定时任务可为 NULL。
3. `execution_owner` 默认 `legacy`，Phase 2 灰度切换为 `desktop_tasks`。

### 7.2 执行任务：`desktop_tasks(kind=monitor)`（Phase 2 新增字段）

建议扩展已有表，增加以下字段：

| 字段 | 类型 | 说明 |
| --- | --- | --- |
| `priority` | int | 执行优先级，见 §7.3 值域 |
| `lane` | varchar(20) | `high` / `normal` / `normal_boosted` / `retry` |
| `lane_weight` | int | 由 lane 衍生，排序用（见 §7.3） |
| `source` | varchar(20) | `collect_now` / `daily_schedule` / `retry_recovery` |
| `scheduler_group_key` | varchar(120) | 问题并发互斥键 |
| `monitor_task_id` | bigint nullable | 关联 `monitoring_collect_tasks.id` |
| `supersedes_task_id` | uuid nullable | 抢占时被替代的旧执行任务 |
| `control_flags` | jsonb | 如 `{"interrupt_requested": true, "interrupt_generation": 7}` |
| `interrupt_generation` | int | 与业务任务同步，权威副本 |
| `started_at` | timestamptz nullable | 真正开始执行时间 |
| `interrupted_at` | timestamptz nullable | 被请求中断时间 |
| `interrupt_reason` | varchar(64) | `collect_now_preempt` 等 |
| `enqueued_at` | timestamptz | 入队时间（aging 用） |

**新增唯一约束**（防止双写）：

```sql
CREATE UNIQUE INDEX uq_desktop_tasks_monitor_active
  ON desktop_tasks (monitor_task_id)
  WHERE kind = 'monitor' AND status IN ('queued', 'in_progress', 'unknown');
```

补充说明：

1. `unknown` 视为“未决活跃态”，在 reconcile 前不得为同一 `monitor_task_id` 再创建新的 active monitor 执行任务。
2. 只有当 `unknown` 被人工或自动对账收敛为 `retry / failed / succeeded / aborted` 后，才允许重新派发。

### 7.3 priority 值域规范（W3）

为避免 monitor 与 publish 共表时优先级错配，严格划分值域：

| kind | lane | priority 范围 | lane_weight | 典型值 |
| --- | --- | --- | --- | --- |
| publish | urgent | 20000–29999 | 100 | 25000 |
| publish | high | 15000–19999 | 90 | 18000 |
| publish | normal | 10000–14999 | 80 | 12000 |
| monitor | high（collect-now） | 5000–5999 | 70 | 5000 |
| monitor | normal_boosted（aged） | 1000–4999 | 50 | 1000~4999 |
| monitor | normal | 100–999 | 40 | 100 |
| monitor | retry | 10–99 | 20 | 50 |

**统一排序规则**：

```sql
ORDER BY lane_weight DESC, priority DESC, enqueued_at ASC
```

保证"优先级跨 lane 不能越级"——monitor 的 high (5000) 永远不会压过 publish 的 normal (10000)，publish 始终拥有最高话语权。

### 7.4 手动请求表 `monitoring_collect_requests`

| 字段 | 类型 | 说明 |
| --- | --- | --- |
| `id` | uuid | 请求 ID |
| `tenant_id` | bigint | 租户 |
| `workspace_id` | bigint | 工作空间 |
| `brand_id` | bigint | 品牌 |
| `keyword_id` | bigint nullable | 可选关键词 |
| `requested_by_user_id` | bigint | 发起人 |
| `target_client_id` | uuid nullable | 本次绑定 client（可能解绑） |
| `scope_hash` | varchar(64) | 幂等用，见 §10.6 |
| `request_scope` | jsonb | 选择了哪些问题/平台 |
| `status` | varchar(30) | `accepted` / `dispatching` / `running` / `partial` / `succeeded` / `failed` / `timed_out` / `superseded` |
| `requested_task_count` | int | 目标任务数 |
| `dispatched_task_count` | int | 已派发任务数 |
| `completed_task_count` | int | 已完成任务数 |
| `first_dispatched_at` | timestamptz nullable | 第一条派发时间 |
| `first_result_at` | timestamptz nullable | 第一条结果时间 |
| `ttl_expires_at` | timestamptz | TTL，默认创建后 5 分钟 |

作用：

1. 给 UI 返回"这次立即采集是否真的在推进"。
2. 支撑重复点击幂等。
3. 支撑超时告警与链路追踪。

## 8. 状态机设计

### 8.1 业务任务状态机

```text
pending
  -> (dispatch) -> (desktop lease+run+callback) -> completed

pending
  -> (dispatch+lease timeout) -> expired -> retry -> pending

pending (被抢占场景，Phase 2)
  -> (promote to high lane) -> (interrupt old execution) -> (new execution completed) -> completed

pending
  -> failed (超过 max_retry)

pending
  -> skipped (配额/规则排除)
```

约束：

1. 同一业务日同一问题同一平台只保留一条业务任务，不复制第二套任务。
2. `collect-now` 只提升优先级、刷新调度窗口、触发抢占，不再复制业务任务。
3. 最终结果事实仍归并到同一业务任务。
4. 当 `execution_owner='desktop_tasks'` 时，业务任务在执行全过程保持 `pending`，直到最终终态才跳到 `completed / failed / skipped / expired`；`leased / received` 不再写入业务表。

### 8.2 执行任务状态机（Phase 2 起）

```text
queued
  -> in_progress
  -> succeeded

queued
  -> aborted (Phase 1/2 均支持：直接取消)

in_progress
  -> (task_control: interrupt_requested with gen) -> (desktop safe point) -> aborted (Phase 2)

in_progress
  -> failed | unknown | succeeded
```

补充约束：

1. `unknown` 在 reconcile 前视为活跃执行态，与 `queued / in_progress` 一样阻止同一 `monitor_task_id` 生成新的 active monitor 执行任务。
2. `unknown` 不是可长期堆积的常态，必须通过 `ReconcileDesktopTask` 收敛回 `queued / failed / succeeded / aborted`。

## 9. 调度与优先级设计

### 9.1 定时调度策略

`scheduler` 不直接把所有到期任务一股脑发出，而是分成三步：

1. 扫描：找出 `next_run_at <= now()` 的计划任务。
2. claim：把到期任务转成 `monitoring_collect_tasks`（设置 `dispatch_lane='normal'`、`dispatch_priority=100`、`execution_owner` 按灰度规则）。
3. dispatch：由 `monitor dispatch coordinator` 按预算从 `monitoring_collect_tasks` 中选择一小批执行任务下发。

### 9.2 dispatch 预算决策树（C5）

当 dispatch coordinator 决定是否派发一条任务时，按下列顺序串行检查，**任一失败则当前 task 跳过**：

```text
dispatch_allowed(task) iff
  1. cluster.minute_budget.remaining > 0
  2. tenant(task.tenant_id).minute_budget.remaining > 0
  3. client(task.target_client_id).concurrent_monitor < client.max_concurrent
  4. client(task.target_client_id).queue_depth < client.max_queue_depth
  5. platform(task.ai_platform_id).concurrent < platform.max_concurrent
```

**拒绝处置矩阵**：

| 拒绝维度 | 处置 |
| --- | --- |
| budget 类（1、2） | 暂留 ready 队列，下一分钟预算窗口重试 |
| concurrent 类（3、5） | 暂留 ready 队列，等待 `completion` 事件驱动重试 |
| queue depth 类（4） | 选择次优 client（`target_client_id=NULL` 的任务）；若指定 client 饱和且 `target_client_id` 非空，暂留 |

**预算值建议**（可配置，热更新）：

| 维度 | 初始值 | 熔断阈值 | 说明 |
| --- | --- | --- | --- |
| 集群每分钟派发上限 | 300 | 500 | 保护 ingest 链路 |
| 单租户每分钟上限 | 30 | 60 | 公平性 |
| 单 client monitor 并发 | 2 | 4 | 浏览器资源 |
| 单 client 待执行队列深度 | 20 | 40 | 避免堆积失真 |
| 单平台并发（跨 client 聚合） | 1 | 2 | 登录态/风控保护 |

**high lane 绕过规则**：

- `dispatch_lane='high'` 的任务**绕过** 2、4 两项（租户预算、队列深度）
- 仍受 1、3、5 约束（集群预算、client 浏览器并发、平台风控）
- 保障 collect-now 体验同时不压垮基础设施

**耦合说明**：

- 集群预算 > 所有租户预算之和，预防极端峰值
- 单 client 并发仅对 **monitor kind** 生效，不含 publish（publish 有独立通道）
- 单平台并发跨 client 聚合计数（例如 OpenAI 在 client A 并发 1 + client B 并发 0 = 平台并发 1）

### 9.3 lane 调度规则

推荐调度顺序：

1. `high` 非空时，优先耗尽 `high`
2. `high` 非空时，暂停向**同一 client** 派发新的 `normal`（其他 client 上的 normal 不受影响）
3. 全局仍保留 10%~20% 的 `normal` 背景吞吐，防止定时任务永久饿死
4. `retry` 只在 `high` 与 `normal` 都低水位时进入

### 9.4 aging 提升（百分位 + 对数衰减，W2）

简单阶梯加分在批量积压时会**集体失效**（所有 normal 都 +50 等于没有 aging）。改为百分位触发式 aging：

**触发条件**：每 30s 扫描一次 `normal` lane ready 列表：

1. 按 `enqueued_at` 升序排序
2. 选出等待时间最长的 **top 10%** 任务
3. 对这批任务执行 `priority += agingBoost(wait_time)`，`dispatch_lane = 'normal_boosted'`

**boost 函数**：

```go
func agingBoost(waitMinutes float64) int {
    // 对数衰减：10min +50, 30min +85, 60min +115, 120min +141
    return int(50 * math.Log2(1 + waitMinutes/5))
}
```

**上限保护**：

- 单任务 aging 累计不超过 `+900`（保证 `normal_boosted` priority 永远 `< 5000`，不跨入 high 区）
- 若 aging 后 priority 达到 `normal_boosted` 上限，进入 `normal_boosted` lane 并停止继续 aging
- 观测指标 `monitor_normal_starvation_age_p99` 暴露 normal 最长等待时间

## 10. collect-now 抢占设计

### 10.1 抢占范围

默认只抢占以下任务：

1. 同一 `target_client_id`
2. `kind=monitor`
3. `source=daily_schedule`
4. `status in (queued, in_progress)`（Phase 1 仅 `queued`；Phase 2 起含 `in_progress`）

不抢占：

1. publish 任务
2. 其他用户或其他 client 的 monitor 任务
3. 已接近完成且无法安全中断的任务

### 10.2 interrupt_generation 规范（C2）

**定义**：`interrupt_generation` 归属 `desktop_tasks`（Phase 2）或 `monitoring_collect_tasks`（Phase 1），每次 collect-now 针对该 client 原子递增。

**递增操作**（单次 UPDATE）：

```sql
-- Phase 2 示例
UPDATE desktop_tasks
SET interrupt_generation = interrupt_generation + 1
WHERE target_client_id = $1
  AND kind = 'monitor'
  AND status IN ('queued', 'in_progress')
RETURNING id, interrupt_generation;
```

**下发约束**：

1. `task_control` 事件必须携带 `generation`
2. `task_available`（high lane）事件必须携带 `generation`
3. dispatch coordinator 发出 `abort` 指令时必须带当次 generation

**desktop 执行规则**：

1. 本地维护 per-task `last_seen_generation`
2. 仅执行 `event.generation >= local.last_seen_generation` 的指令
3. 收到指令后更新 `last_seen_generation`

**回写约束**：

desktop 回写 `aborted(preempted)` 时必须带产生 abort 的 generation；服务端比对：

```go
if callback.generation < currentTask.interrupt_generation {
    // 旧 abort 指令迟到，忽略本次回写并下发新的 abort
    log.Warn("stale abort callback", callback.generation, currentTask.interrupt_generation)
    publishInterruptSignal(currentTask, currentTask.interrupt_generation)
    return
}
```

**双击场景保护**：

用户双击 collect-now 产生两次请求时：

- 第一次递增 gen 从 N → N+1
- 第二次递增 gen 从 N+1 → N+2
- desktop 只会执行 gen=N+2 的指令（覆盖 N+1），第一次 abort 自然失效
- 两次 collect_request 通过 `superseded_by_request_id` 形成引用链，第一次 request 标记为 `superseded`

### 10.3 抢占流程

```text
用户点击 collect-now
  -> 幂等检查（scope_hash + 5min TTL）
  -> 短事务：create collect_request, increment gen, promote business tasks to high, write outbox
  -> 提交事务
  -> outbox pump 投递 dispatch signal (monitor.high.{client_id}) + interrupt signal (task_control)
  -> tenant-api dispatch hub 收到后 WebSocket 推送给 desktop
  -> desktop 在安全点中断 running 任务（Phase 2）或在下次 pull 时发现租约失效（Phase 1）
  -> desktop 回写 aborted(preempted) with generation
  -> desktop 立即领取 high lane monitor 任务
```

### 10.4 安全点（Phase 2 起）

desktop client 必须只在安全点响应抢占：

1. 打开平台页面前
2. 输入问题前
3. 提交请求前
4. 收到原始回答后、结构化解析前
5. 单次问题执行结束后

不要在以下阶段强杀：

1. 平台登录态关键跳转中
2. 人机验证流程中
3. 页面提交后等待平台响应但浏览器状态不确定时

**测量**：每个安全点埋点 `safe_point_interval_ms`，用于 §10.5 动态调参。

### 10.5 安全中断超时策略（参数化，C4）

**参数化原则**：硬编码的 15s/60s 无法适应不同平台的响应差异，改为 per-platform 配置：

| 平台 | 安全点 P95 | `interrupt_soft_timeout` | `interrupt_hard_timeout` |
| --- | --- | --- | --- |
| OpenAI 快答 | 8s | 15s | 45s |
| OpenAI 深度推理 | 40s | 60s | 180s |
| Gemini | 20s | 30s | 90s |
| Perplexity | 30s | 45s | 120s |
| 其他 | 动态测量 | max(P95 × 1.5, 15s) | soft × 3 |

**分级策略**：

1. `soft_timeout` 内未中断：禁止向该 client 派发新 normal，high 任务仍然等待
2. `soft_timeout` 到 `hard_timeout`：允许当前任务自然结束，结束后立即领 high
3. 超过 `hard_timeout`：触发告警 + 自动 reconcile（强制 cancel 租约、high 任务转派其他 client 或降级 web 端提示）

**测量落地**：

- 每个安全点埋点 `safe_point_interval_ms`
- 每周基于 P95 自动更新各平台 `interrupt_soft_timeout` 参数
- 参数热更新，无需重启服务

### 10.6 target_client_id 绑定策略（W1）

**绑定时机**：collect-now 接受请求时立刻选择 target client。

**选择规则**（按优先级）：

1. 请求体指定 `target_client_id` 且该 client **在线 + 健康** → 使用指定 client
2. 未指定或指定不可用 → 在当前用户 `tenant_id / workspace_id` 下的在线 client 中按下列权重选：
   - 最近心跳 < 30s（必要条件）
   - monitor 并发余量最大
   - 最近 1 分钟 abort 率最低
   - 任一平台登录态健康

**漂移处理矩阵**：

| 场景 | 策略 |
| --- | --- |
| 用户多设备同时在线 | 按上述权重选一个，不广播；`collect_request.target_client_id` 记录选择结果 |
| 绑定后 client 掉线（30s 无心跳） | 解绑：业务任务 `target_client_id = NULL`，`lane='high'` 不变，任一在线 client 可领 |
| 用户换设备登录 | 收到新 client 心跳后，正在执行的 collect_request 不迁移；新提交的 collect-now 绑定新 client |
| 所有 client 掉线 | collect_request 保持 `accepted`，TTL 5 分钟；5 分钟内有 client 上线则重新 dispatch；超时转 `timed_out` |

**失败边界**：

- collect_request `ttl_expires_at` = 创建时间 + 5 分钟
- `accepted` 状态下无任何 dispatch 动作超过 30s 触发预警
- 同一 `(user_id, brand_id, scope_hash)` 5 分钟内重复 collect-now 返回上一次的 `request_id`（幂等复用）

## 11. RabbitMQ 拓扑设计

### 11.1 下行 dispatch

复用现有 `desktop.task.dispatch` topic exchange，建议约定路由键：

```text
monitor.high.{client_id}
monitor.normal.{client_id}
monitor.retry.{client_id}
publish.{client_id}  (已存在)
```

说明：

1. tenant-api 实例通过临时队列消费所有 `monitor.*` 与 `publish.*`（现有绑定 `publish.*`，Phase 2 新增 `monitor.*`）
2. dispatch hub 再按 `target_client_id` 过滤并推给对应 websocket
3. desktop client 不需要感知 RabbitMQ，只消费 websocket 事件即可

### 11.2 上行结果

保留现有结果 ingest 链路（已落地）：

```text
monitor.result.ingest
monitor.result.ingest.dlq
monitor.projection.rebuild (已存在并带 coalesce)
monitor.projection.rebuild.dlq
```

Phase 3 增强：

1. `monitor.result.ingest` 按 `ingest_shard_key` 分片串行消费（见 §12.2）
2. projection worker 已支持 coalesce，同一个 `(tenant_id, brand_id, business_date)` 只保留一条重算请求

### 11.3 多队列而非单队列 priority

推荐：

1. `monitor.dispatch.high`
2. `monitor.dispatch.normal`
3. `monitor.dispatch.retry`

不推荐仅依赖单队列 `x-max-priority`。原因：

1. 单队列 priority 在消息堆积很深时观测和治理不清晰
2. 多队列更容易做 pause、drain、限流和运维看板
3. 更符合 lane 化流量治理方式

## 12. 结果回写与有序落库

### 12.1 写链路拆分

写链路分成两层（现状已落地）：

1. 原始事实层：`question_monitor_runs`、`question_monitor_parse_results`、`monitoring_citation_facts`
2. 聚合投影层：`monitoring_brand_daily`、`monitoring_brand_platform_daily`

### 12.2 分片串行（Phase 3 落地）

不建议全局单线程顺序落库。

**主选分片键**（唯一）：

```text
shard_key = hash(tenant_id, brand_id, business_date) % N
```

选择理由：与现有 `MonitoringProjectionRebuildQueue` coalesce 键对齐，避免同一投影维度上出现 ingest 与 projection 错位。**去掉** V1 中的备选 `(tenant_id, question_id, business_date)`，统一一个维度。

规则：

1. 同 shard 内串行消费，保证局部有序
2. 不同 shard 并行消费，保证总体吞吐
3. 原始事实 upsert 幂等即可，不追求全局顺序
4. 聚合快照按 `(tenant_id, brand_id, business_date)` 做局部重算与覆盖写

### 12.3 幂等键体系（与现有表对齐，W7）

| 幂等维度 | 键 | 对应现有约束 |
| --- | --- | --- |
| 业务任务幂等 | `(tenant_id, brand_id, question_id, ai_platform_id, collector_type, run_mode, business_date)` | `monitoring_collect_tasks` 现有唯一索引 |
| 执行任务幂等 | `(monitor_task_id)` WHERE status IN (queued,in_progress,unknown) | `desktop_tasks` Phase 2 新增唯一索引（见 §7.2） |
| 原始 run 入库幂等 | `(tenant_id, brand_id, question_id, ai_platform_id, collector_type, run_mode, business_date)` | `question_monitor_runs.uk_monitor_run_idempotent` |
| parse 结果入库幂等 | `(run_id, citation_fingerprint)` | `question_monitor_parse_results` / `monitoring_citation_facts` 现有约束 |
| projection 重算幂等 | `(tenant_id, brand_id, business_date)` | `MonitoringProjectionRebuildQueue` 现有 coalesce 键（未含 collector_type） |

**注意**：

1. projection 重算 coalesce 键统一为 `(tenant_id, brand_id, business_date)`，V1 文档中提及的 `collector_type` 维度**不纳入**，与现有实现对齐。
2. `question_monitor_runs` 当前没有 `task_id / attempt_id` 字段；attempt 级诊断信息仍在 `desktop_task_attempts`。如果后续要做 attempt 级 monitoring run 回放，需要单独迁移 `question_monitor_runs` 结构，不能假设现状已支持。
3. `question_monitor_runs` 与业务任务共用同一业务键，语义是“同一业务日同一问题同一平台只保留一条最新有效 run”，而不是保留 attempt 级历史。
4. `question_monitor_runs` 的 upsert 语义必须固定为“成功结果优先、最后一次有效结果生效”。列名以下仅为示意，实际实现按真实 schema 对应：

```sql
INSERT INTO question_monitor_runs (...)
VALUES (...)
ON CONFLICT (tenant_id, brand_id, question_id, ai_platform_id, collector_type, run_mode, business_date)
DO UPDATE SET
    result = EXCLUDED.result,
    status = EXCLUDED.status,
    updated_at = now()
WHERE question_monitor_runs.status = 'failed'
   OR EXCLUDED.status = 'succeeded';
```

5. 上述语义等价于：
   - 新 `succeeded` 可覆盖旧 `failed`，也可覆盖旧 `succeeded`（最后一次成功结果生效）
   - 新 `failed` 只能覆盖旧 `failed`，不得覆盖已存在的 `succeeded`
   - 失败 attempt 的诊断链路保留在 `desktop_task_attempts`，不在 `question_monitor_runs` 中展开

### 12.4 落库保护

当出现以下任一情况时，自动触发写链路降速：

1. ingest queue lag 超过阈值（默认 30s）
2. projection queue lag 超过阈值（默认 60s）
3. Monitoring PG 写入延迟明显升高（P95 > 500ms）
4. 事务冲突率升高（> 5%）

降速策略：

1. 暂停 `normal` dispatch
2. 保留 `high` lane
3. 将 projection rebuild 改为批次合并

## 13. API 与协议设计

### 13.1 collect-now

```http
POST /api/tenant/monitoring/brands/{brand_id}/collect-now
Content-Type: application/json

{
  "keyword_id": 123,
  "preempt": true,
  "wait_for_first_dispatch": false,
  "target_client_id": "optional-uuid"
}
```

返回建议：

```json
{
  "request_id": "uuid",
  "collection_mode": "desktop",
  "target_client_id": "uuid",
  "affected_task_count": 12,
  "promoted_task_count": 12,
  "aborted_queued_count": 4,
  "interrupt_requested_count": 2,
  "interrupt_generation": 7,
  "ttl_expires_at": "2026-04-20T10:05:00Z",
  "message": "已提升为高优先级并请求抢占当前定时采集任务"
}
```

**`collection_mode` 字段说明**：浏览器插件硬切换下线后，该字段恒为 `"desktop"`，保留仅用于日志审计与未来多端扩展。前端可忽略此字段的分支逻辑。

### 13.2 desktop dispatch event

```json
{
  "type": "task_available",
  "kind": "monitor",
  "task_id": "uuid",
  "target_client_id": "uuid",
  "priority": 5000,
  "lane": "high",
  "status": "queued",
  "scheduler_group_key": "qid:123",
  "business_date": "2026-04-20",
  "interrupt_generation": 7,
  "updated_at": "2026-04-20T10:00:00Z"
}
```

### 13.3 desktop control event

```json
{
  "type": "task_control",
  "kind": "monitor",
  "task_id": "uuid",
  "target_client_id": "uuid",
  "control": "interrupt_requested",
  "reason": "collect_now_preempt",
  "interrupt_generation": 7,
  "replacement_task_id": "uuid",
  "updated_at": "2026-04-20T10:00:05Z"
}
```

### 13.4 desktop 中断确认

优先复用现有 cancel 能力：

```http
POST /api/desktop/tasks/{task_id}/cancel
Content-Type: application/json

{
  "lease_token": "xxx",
  "reason": "collect_now_preempt",
  "interrupt_generation": 7
}
```

若后续希望保留更细致的中断原因，可新增：

```http
POST /api/desktop/tasks/{task_id}/interrupt-ack
```

### 13.5 立即采集请求查询

```http
GET /api/tenant/monitoring/collect-requests/{request_id}
```

用于前端轮询：

1. 是否已经抢占成功
2. 是否已开始执行
3. 是否已出现第一条结果
4. 当前状态与倒计时 TTL

## 14. desktop client 设计要求

### 14.1 调度优先级

desktop client 本地调度规则：

1. `publish` 继续保持最高优先级，不被 monitoring 抢占。
2. `monitor.high` 高于 `monitor.normal_boosted` 高于 `monitor.normal` 高于 `monitor.retry`
3. 收到 `interrupt_requested` 时，正在运行的 `monitor.normal` 必须让路给 `monitor.high`（Phase 2 起）
4. Phase 1 desktop 仍以 pull 为主；Phase 2 迁至 push + lease

### 14.2 本地队列

本地 monitor scheduler 需要新增：

1. `lane`
2. `priority`
3. `interruptRequested`
4. `interruptDeadline`
5. `interruptGeneration`（与服务端同步的代际号）
6. `executionPhase`（精确定义“是否已开始执行”，见下表）

`executionPhase` 建议统一枚举：

| 值 | 说明 | Phase 1 active lease 行为 | Phase 2 行为 |
| --- | --- | --- | --- |
| `PREPARING` | 浏览器 context / page 尚未初始化完成 | 可在下一次串行检查时 cancel lease | 可立即进入中断处理 |
| `NAVIGATING` | 正在打开 AI 平台页面 | 可在下一次串行检查时 cancel lease | 可立即进入中断处理 |
| `AUTHENTICATING` | 登录态校验 / 验证流程 | 视为已开始，不再直接 cancel | 等待下一个安全点 |
| `INPUTTING` | 输入问题 | 视为已开始，不再直接 cancel | 等待下一个安全点 |
| `SUBMITTING` | 提交请求 | 视为已开始，不再直接 cancel | 等待下一个安全点 |
| `WAITING` | 等待 AI 回答 | 视为已开始，不再直接 cancel | 等待下一个安全点 |
| `PARSING` | 结构化解析 | 视为已开始，不再直接 cancel | 等待下一个安全点 |
| `POSTING` | 回写结果 | 视为已开始，不再直接 cancel | 等待下一个安全点 |

本文中“尚未真正开始执行”严格定义为 `PREPARING | NAVIGATING`；自 `AUTHENTICATING` 起统一视为“已开始”。

状态推进与 `interruptRequested / interruptGeneration` 的更新，必须运行在同一个 desktop scheduler 单线程 loop 或同一把互斥锁内，不能分散在不同线程各自修改。

本地排序建议：

```text
lane_weight DESC
priority DESC
availableAt ASC
enqueuedAt ASC
updatedAt ASC
```

### 14.3 Phase 1 / Phase 2 通道策略（SSE 早期下线）

| 阶段 | `task_available` 下发通道 | `task_control` 下发通道 | SSE 状态 |
| --- | --- | --- | --- |
| Phase 1 | **WebSocket** (`desktop.task.dispatch`) 复用 publish 通道 | **WebSocket** 同通道 | **Phase 1 完成即下线** |
| Phase 2 | WebSocket（新增 running 抢占事件） | WebSocket | 已移除 |
| Phase 3 | WebSocket | WebSocket | 已移除 |

**Phase 1 直接启用 WebSocket 的前提**（代码核实）：

1. `desktop.task.dispatch` exchange 已存在并在 publish 场景生产使用
2. `PublishDesktopDispatch` 发布端已存在，Phase 1 扩展支持 `monitor.*` 路由键
3. tenant-api dispatch hub 已支持 WebSocket push，Phase 1 新增订阅 `monitor.*` binding key
4. desktop client WebSocket 连接已存在（publish 场景），Phase 1 新增 `kind=monitor` 的事件 handler

**Phase 1 WebSocket 事件扩展**：

1. `task_available` 事件新增 `kind=monitor` 分支（见 §13.2）
2. 新增 `task_control` 事件（见 §13.3），承载抢占指令
3. Phase 1 desktop 在 `task_available` 到来时仍以 pull-lease 确认任务归属（WebSocket 仅作为 dispatch 信号通知，lease 协议保持 pull 兼容），Phase 2 起改为 push-lease

**SSE 下线收尾**（Phase 1 上线前置工作）：

1. 代码层面：`server/internal/tenant/transport/desktop_events_handler.go` 移除路由注册
2. 代码层面：desktop-client 移除 `SseClient` 类、`api-client.ts` 相关 `sseConnected` 状态、`runtime-controller.ts` 的 SSE 启动调用
3. 路由层面：`/api/desktop/events` 返回 410 Gone 一个版本周期后彻底移除
4. 观测层面：SSE 连接数指标下线，改看 WebSocket 连接数

### 14.4 中断实现与串行化规则（Phase 1 active lease / Phase 2 running 通用）

desktop client 需要：

1. 监听 `task_control` 后，必须把处理投递到与主执行循环同一个 scheduler loop（或持有同一把互斥锁）的串行上下文中。
2. 在该串行上下文内先比对 `interrupt_generation`，再原子写入 `interruptRequested=true / interruptGeneration / interruptReason`。
3. `task_control` 处理线程**禁止**直接调用 cancel API；真正的 cancel / abort 只能由主执行循环在下一个检查点统一发出。
4. Phase 1 中，若串行检查时 `executionPhase in (PREPARING, NAVIGATING)`，则发送 `cancel lease`；若已进入 `AUTHENTICATING` 及之后阶段，则忽略本次中断并等待自然完成。
5. Phase 2 中，若任务已经开始执行，则在下一个安全点处理 `interruptRequested`，回写 `aborted(preempted)` 并携带 `interrupt_generation`。
6. 中断完成后立刻触发下一轮 `pumpExecutionLoop()`，优先消费 `monitor.high`。

## 15. 可观测性与风控

### 15.1 关键指标（含新增）

至少监控以下指标：

**现有指标**：

1. `collect_now_accept_count`
2. `collect_now_first_dispatch_latency_ms`
3. `collect_now_first_result_latency_ms`
4. `monitor_dispatch_high_backlog`
5. `monitor_dispatch_normal_backlog`
6. `monitor_result_ingest_lag_ms`
7. `monitor_projection_rebuild_lag_ms`
8. `monitor_preempt_request_count`
9. `monitor_preempt_success_count`
10. `monitor_preempt_timeout_count`
11. `desktop_client_monitor_queue_depth`
12. `monitoring_pg_write_qps`

**V1.1 新增指标**：

1. `collect_now_preempt_induced_retry_count` — 抢占导致定时任务额外重试次数
2. `monitor_normal_starvation_age_p99` — normal lane 最长等待时间
3. `safe_point_interval_ms_p95` — 各平台安全点间隔（per platform 维度）
4. `monitor_dispatch_reject_by_reason_count` — 按拒绝维度分桶（budget/concurrent/queue_depth）
5. `interrupt_generation_mismatch_count` — 旧 generation 被忽略次数（异常监控）
6. `collect_request_ttl_expired_count` — TTL 内未完成派发的请求数
7. `execution_owner_legacy_count` / `execution_owner_desktop_tasks_count` — Phase 2 灰度进度
8. `phase2_reconcile_fixed_count` — reconcile 任务修复数
9. `monitor_dispatch_high_oldest_age_ms` — `high` lane 中最老未派发任务年龄
10. `tenant_monitor_normal_backlog` — 按 tenant 分桶的 `normal` backlog 深度

### 15.2 告警

| 告警 | 指标 | 聚合方式 | 窗口 | 阈值 | 持续时间 |
| --- | --- | --- | --- | --- | --- |
| `high` lane 积压 | `monitor_dispatch_high_oldest_age_ms` | `max` | 1 分钟 | `> 60000ms` | 连续 2 个窗口 |
| `collect-now` 首条结果过慢 | `collect_now_first_result_latency_ms` | `P99` | 5 分钟 | `> 180000ms` | 连续 2 个窗口 |
| projection lag 过高 | `monitor_projection_rebuild_lag_ms` | `max` | 5 分钟 | `> 300000ms` | 连续 2 个窗口 |
| 单 client 抢占超时过多 | `increase(monitor_preempt_timeout_count{client_id})` | `sum by client_id` | 1 小时 | `>= 3` | 立即告警 |
| 单 tenant 的 `normal` backlog 无下降 | `tenant_monitor_normal_backlog{tenant_id}` | `min > 100` 且 `last-first >= 0` | 30 分钟 | 同时满足两条件 | 连续 1 个窗口 |
| generation mismatch 突增 | `increase(interrupt_generation_mismatch_count)` | `sum` | 10 分钟 | `>= 10` | 连续 2 个窗口 |
| Phase 2 reconcile 持续不一致 | `phase2_reconcile_fixed_count` | 每次 reconcile run | 15 分钟 | 连续 3 次 run `> 0` | 立即告警 |

说明：

1. 本节中的 queue lag / backlog age 一律按“最老未处理项的入队年龄”定义，而不是平均值。
2. `collect_request_ttl_expired_count` 继续作为单请求超时兜底告警，不与 P99 聚合告警互相替代。

### 15.3 风控与公平性

1. 单用户 `collect-now` 基于 `(user_id, brand_id, scope_hash)` 的 5 分钟幂等窗口（取代 V1 的 10 秒简单去重）
2. 单租户同时最多 1~2 个活跃 `collect-now request`
3. 对异常频繁的手动采集做限流（每用户每小时 20 次）
4. 同平台登录态异常时暂停该平台 lane

## 16. 分阶段落地方案

### 16.1 Phase 1：最小改造 + WebSocket 直上 + SSE 下线（W5 + W6）

**范围收敛**：Phase 1 **不做 running 任务的运行时抢占**（运行时抢占放 Phase 2），但通道层面**一步到位直上 WebSocket**，不保留 SSE 作为过渡。

**Phase 1 实施项清单**：

1. **DB migration**：`monitoring_collect_tasks` 加字段（`dispatch_priority / dispatch_lane / target_client_id / interrupt_generation / execution_owner / superseded_by_request_id / dispatch_after / last_dispatched_at / dispatch_attempts / ingest_shard_key`）；新增 `monitoring_collect_requests` 表
2. **lease 排序改造**：monitoring lease 查询从 `ORDER BY planned_at` 改为 `ORDER BY CASE dispatch_lane WHEN 'high' THEN 70 WHEN 'normal_boosted' THEN 50 WHEN 'normal' THEN 40 WHEN 'retry' THEN 20 ELSE 0 END DESC, dispatch_priority DESC, dispatch_after ASC NULLS FIRST, planned_at ASC`。
3. **collect-now handler**：幂等检查 + gen 递增 + 任务提升 + 取消 queued normal / 对 active lease 写入抢占控制信号 + 创建 collect_request（见 §17.1 outbox 模式）
4. **RabbitMQ 生产端扩展**：`PublishDesktopDispatch` 支持 `monitor.high.{client_id} / monitor.normal.{client_id} / monitor.retry.{client_id}` 路由键
5. **tenant-api dispatch hub 订阅扩展**：binding key 新增 `monitor.*`，与现有 `publish.*` 并列
6. **WebSocket push 下发**：dispatch hub 把 `task_available(kind=monitor)` 与 `task_control` 事件推给匹配 `target_client_id` 的 desktop WebSocket
7. **desktop client 改造**：
   - WebSocket handler 增加 `kind=monitor` 分支
   - 收到 `task_available(monitor)` 时触发一次 pull-lease（Phase 1 保持 lease 协议不变，仅把"定时 pull"改为"信号驱动 pull"）
   - 收到 `task_control(interrupt_requested)` 时：先在本地 scheduler 串行上下文内仅标记 `interruptRequested=true`；若目标任务处于本地队列（queued）则丢弃；若在下一次串行检查时 `executionPhase in (PREPARING, NAVIGATING)` 则主动 `cancel lease`；自 `AUTHENTICATING` 起视为已开始，Phase 1 忽略中断并等待自然完成
   - 比对 `interrupt_generation`，忽略过期指令
8. **SSE 下线**（Phase 1 一并完成）：
   - 移除 `server/internal/tenant/transport/desktop_events_handler.go` 路由注册
   - 移除 desktop-client 的 `SseClient`、`api-client.ts` 中 `sseConnected` 状态、`runtime-controller.ts` 的 SSE 启动
   - 旧端点返回 410 Gone 一个发布周期后删除
9. **指标埋点**：collect_now 全链路延迟、gen_mismatch、dispatch_reject、WebSocket 连接数

**Phase 1 的 active lease 处置**：

legacy 路径下，服务端无法仅凭 `monitoring_collect_tasks` 精确区分“已 lease 但尚未真正开始执行”和“已在浏览器内运行”。因此 Phase 1 统一采用保守策略：

1. 服务端只直接取消 queued 的 normal 任务。
2. 对已有 active lease 的 normal 任务，不在服务端盲目清空 `lease_token_hash`，只下发 `task_control` 提示 desktop 自判。
3. desktop 先在本地调度锁内原子标记 `interruptRequested`；仅当下一次串行检查仍处于 `PREPARING / NAVIGATING` 时，才主动 `cancel lease` 并让路给 high。
4. 若串行检查时已进入 `AUTHENTICATING` 及之后阶段，则 Phase 1 忽略本次中断请求，待自然完成；期间 coordinator 暂停向该 client 派发新的 normal。

**Phase 1 的 collect-now 优先级保障**：

1. 提升目标业务任务为 high lane 后，立即取消 queued 的 normal 任务，并向 active lease 的 normal 任务发送 `task_control`
2. 通过 WebSocket 立刻推 `task_available(high)` 给目标 client
3. desktop 收到信号后立即 pull-lease high 任务；若当前无 active lease，通常可直接开始执行
4. 无 active lease 的典型场景：WebSocket push 延迟 < 200ms，client pull-lease < 500ms，总体 collect-now → 开始执行 < 2s；若遇到已 running 任务，则首响时间取决于当前任务尾延迟

**Phase 1 核心诉求与边界**：

1. 解决 95% 场景：用户点击 collect-now 后 2s 内开始执行
2. 不解决 5% 场景：client 已在 running 且 AI 平台响应很慢（30s+），此场景 UI 提示"正在等上一条任务完成"

**优点**：

1. 通道层面一步到位，无 SSE/WS 双通道并存的复杂度
2. 解决"立即采集让路"核心诉求，延迟进入 2s 量级
3. 不引入双层模型，无大规模迁移成本

**缺点**：

1. monitoring 与 publish 仍是两套执行控制面（Phase 2 收敛）
2. 无法抢占 running 任务（业务方可接受，Phase 2 补齐）
3. Phase 1 工作量略增（多了 WebSocket 通道对接和 SSE 下线），估时 1.5~2 周

### 16.2 Phase 2：执行控制面收敛 + 迁移期一致性（C3）

**新增内容**（通道层面在 Phase 1 已就位，Phase 2 只做控制面收敛）：

1. 引入 `desktop_tasks(kind=monitor)` 全字段（见 §7.2），作为执行态权威表
2. `monitoring_collect_tasks.execution_owner` 字段落地，按 tenant 灰度
3. Phase 1 的"信号驱动 pull-lease"升级为"push-lease"：dispatch hub push `task_available` 时直接携带 lease_token
4. 真正的 `running` 任务抢占能力（基于 `desktop_tasks.control_flags` + 安全点协议）
5. desktop client 监听 `task_control` 事件，安全点响应中断并回写 `aborted(preempted)`

#### 16.2.1 执行源路由字段

`monitoring_collect_tasks.execution_owner` varchar(20)：

- `legacy` — 仍在业务任务表上做 lease（Phase 1 路径）
- `desktop_tasks` — 已迁移到执行任务表（Phase 2 路径）

dispatch coordinator 按该字段路由：

```go
switch task.ExecutionOwner {
case "legacy":
    leaseOnMonitoringCollectTasks(task)
case "desktop_tasks":
    dispatchViaDesktopTasks(task)
}
```

#### 16.2.2 灰度策略

按 `tenant_id % 100 < rollout_percentage` 灰度。Phase 2 上线时：

| 日期（D 表示 Phase 2 上线日） | `rollout_percentage` | 预期观测 |
| --- | --- | --- |
| D1 | 5% | 抢占成功率、ingest lag、reconcile 产出 |
| D3 | 25% | 扩大样本，观察多租户并发表现 |
| D7 | 50% | 压测高峰是否平稳 |
| D14 | 100% | 全量 |

灰度切换时：

- 仅对**新生成**的业务任务按 tenant 落 `execution_owner`
- **不迁移**已 in-flight 的 legacy 任务，等自然结束
- dispatch coordinator 同时消费两条路径，直到 legacy in-flight 清零

#### 16.2.3 回滚预案

若 Phase 2 发现严重问题：

1. 立即将 rollout 回调至 0%（新任务全部走 legacy）
2. 已迁移到 `desktop_tasks` 的 in-flight 任务**不强制回迁**，等自然结束（或 `lease_expires_at` 回收）
3. 若必须强制回迁，执行脚本：

```sql
-- 强制回迁仅在紧急故障场景使用
BEGIN;
-- Step 1: cancel 所有 monitor kind 的 in-flight desktop_tasks
UPDATE desktop_tasks
SET status = 'aborted', interrupt_reason = 'phase2_emergency_rollback'
WHERE kind = 'monitor' AND status IN ('queued', 'in_progress', 'unknown');

-- Step 2: 业务任务回流 legacy
UPDATE monitoring_collect_tasks
SET execution_owner = 'legacy',
    status = 'pending',
    lease_token_hash = NULL,
    leased_to_executor = NULL,
    leased_at = NULL,
    lease_expires_at = NULL
WHERE execution_owner = 'desktop_tasks'
  AND status IN ('pending');
COMMIT;
```

#### 16.2.4 双写一致性保护

**绝对禁止**：

- 同一 `monitoring_collect_tasks.id` 同时存在两条 active `desktop_tasks(kind=monitor)` → 通过 §7.2 的 UNIQUE 约束保证
- `legacy` 路径在业务任务标记 `execution_owner='desktop_tasks'` 后继续 lease → 通过 coordinator 启动前检查 + 定期 reconcile 保证

**Reconcile 任务**（每 5 分钟运行）：

1. 扫描 `execution_owner='legacy' AND leased_to_executor IS NOT NULL AND lease_expires_at < now()` 的僵尸租约 → 清理
2. 扫描 `execution_owner='desktop_tasks' AND status='pending'` 且无对应 active desktop_task 的业务任务 → 重新 dispatch
3. 扫描 `execution_owner='desktop_tasks'` 但 `desktop_tasks` 表无对应记录的异常业务任务 → 记录告警、人工介入

### 16.3 Phase 3：治理增强

目标：面向规模化与稳定性。

实施项：

1. result ingest 按 `ingest_shard_key` 分片串行消费（见 §12.2）
2. projection rebuild coalesce 已有，Phase 3 补充按租户分片
3. per-tenant fair scheduling
4. aging 提升按 §9.4 完全落地
5. 运维看板与自动降速

（SSE 相关移除已在 Phase 1 完成，无需 Phase 3 处理）

## 17. 推荐伪代码

### 17.1 collect-now（outbox 模式，W4）

```go
func CollectNow(ctx context.Context, req CollectNowRequest) (*CollectNowResponse, error) {
    // 阶段 1：幂等复用（无事务）
    scopeHash := hashScope(req)
    if existing := findActiveRequest(ctx, req.UserID, req.BrandID, scopeHash); existing != nil {
        return buildResponseFromExisting(existing), nil
    }

    // 阶段 2：选择 target_client（见 §10.6）
    targetClientID, err := selectTargetClient(ctx, req)
    if err != nil {
        return nil, err
    }

    requestID := uuid.New()
    var effects *CollectNowEffects

    // 阶段 3：短事务，仅写 DB 和 outbox，不做外部副作用
    err = withTx(ctx, func(tx Tx) error {
        createCollectRequest(tx, CollectRequest{
            ID:             requestID,
            UserID:         req.UserID,
            BrandID:        req.BrandID,
            ScopeHash:      scopeHash,
            TargetClientID: targetClientID,
            Status:         "accepted",
            TTLExpiresAt:   time.Now().Add(5 * time.Minute),
        })

        // 原子递增 generation
        gen := incrementInterruptGeneration(tx, targetClientID)

        // 批量提升业务任务为 high lane（单条 UPDATE，WHERE 带 scope 条件）
        effects = promoteToHighLane(tx, req.Scope, targetClientID, requestID, gen)

        // 仅统计本次真正从 queued normal -> canceled 的任务数；不含已被旧请求取消或非 normal 任务
        effects.AbortedQueuedCount = abortQueuedNormal(tx, targetClientID, gen)
        // 仅统计本次 newly-marked interrupt_requested 的 active lease 数；不含已存在更高/相同 generation 的任务
        effects.InterruptRequestedCount = markInterruptRequested(tx, targetClientID, gen, requestID)

        // 写 outbox
        writeOutbox(tx, OutboxEvent{
            Kind:      "monitor.dispatch.high",
            Payload:   DispatchPayload{ClientID: targetClientID, Generation: gen},
            CreatedAt: time.Now(),
        })
        writeOutbox(tx, OutboxEvent{
            Kind:      "monitor.interrupt",
            Payload:   InterruptPayload{ClientID: targetClientID, TaskIDs: effects.RunningTaskIDs, Generation: gen, RequestID: requestID},
            CreatedAt: time.Now(),
        })
        return nil
    })
    if err != nil {
        return nil, err
    }

    // 阶段 4：外部副作用由独立 outbox pump goroutine 处理（非阻塞）
    return &CollectNowResponse{
        RequestID:               requestID,
        CollectionMode:          "desktop",
        TargetClientID:          targetClientID,
        AffectedTaskCount:       effects.TotalCount,
        PromotedTaskCount:       effects.PromotedCount,
        AbortedQueuedCount:      effects.AbortedQueuedCount,
        InterruptRequestedCount: effects.InterruptRequestedCount,
        InterruptGeneration:     effects.Generation,
        TTLExpiresAt:            time.Now().Add(5 * time.Minute),
    }, nil
}

// Outbox pump（独立 goroutine）
func OutboxPump(ctx context.Context) {
    ticker := time.NewTicker(200 * time.Millisecond)
    defer ticker.Stop()
    for {
        select {
        case <-ctx.Done():
            return
        case <-ticker.C:
            events := fetchUndeliveredOutbox(ctx, 100)
            for _, e := range events {
                switch e.Kind {
                case "monitor.dispatch.high":
                    publishDispatchSignal(e.Payload)
                case "monitor.interrupt":
                    publishInterruptSignal(e.Payload)
                }
                markOutboxDelivered(ctx, e.ID)
            }
        }
    }
}
```

**CollectNowResponse 字段语义**：

1. `aborted_queued_count` = 本次 `collect-now` 实际取消的 queued normal 任务数，不含此前已取消或已失活任务。
2. `interrupt_requested_count` = 本次新写入 `task_control / interruptRequested` 的 active lease 任务数，不含已被更高或相同 generation 覆盖的任务。

### 17.2 dispatch coordinator

```go
func DispatchLoop(ctx context.Context) {
    for {
        // 先 drain high lane
        if highLaneBacklog() > 0 {
            drainHighLane()
            keepNormalBudget(0.2)  // 保留 20% normal 背景吞吐
            continue
        }

        // 检查降速信号
        if dbLagTooHigh() || ingestLagTooHigh() {
            pauseNormalLane()
            sleep(5 * time.Second)
            continue
        }

        // 按预算 dispatch normal
        dispatchByBudget("normal")
        sleep(1 * time.Second)
    }
}

func dispatchByBudget(lane string) {
    candidates := fetchReadyTasks(lane, 100)
    for _, task := range candidates {
        reason, ok := checkDispatchBudget(task)
        if !ok {
            metrics.Inc("monitor_dispatch_reject_by_reason_count", reason)
            continue
        }
        doDispatch(task)
    }
}
```

### 17.3 desktop client 抢占串行化（Phase 1 active lease / Phase 2 running，TypeScript）

文件：`apps/desktop-client/src/scheduler/monitor-scheduler.ts`

```ts
function onTaskControl(event: TaskControlEvent) {
  scheduler.enqueue(() => {
    const active = state.activeExecutions.get(event.task_id)
    if (!active) return
    if (active.kind !== "monitor") return

    // 代际比对必须和状态写入在同一个串行上下文内
    if (event.interrupt_generation < active.lastSeenGeneration) {
      metrics.inc("interrupt_generation_mismatch_count")
      return
    }

    active.lastSeenGeneration = event.interrupt_generation
    active.interruptRequested = true
    active.interruptReason = event.reason
  })
}

function canPhase1Cancel(phase: ExecutionPhase) {
  return phase === "PREPARING" || phase === "NAVIGATING"
}

async function maybeHandleInterrupt(taskId: string) {
  const active = state.activeExecutions.get(taskId)
  if (!active?.interruptRequested) return

  // Phase 1 不允许 control handler 直接 cancel，只能在主执行循环串行检查后决定
  if (runtime.phase === 1 && !canPhase1Cancel(active.executionPhase)) {
    return
  }

  await cancelDesktopTask(taskId, {
    lease_token: currentLeaseToken(taskId),
    reason: "collect_now_preempt",
    interrupt_generation: active.lastSeenGeneration,
  })

  // 记录安全点间隔（用于动态调参）
  metrics.observe("safe_point_interval_ms", Date.now() - active.lastSafePointAt)

  pumpExecutionLoop()
}
```

## 18. 最终建议

推荐决策如下：

1. 短期（Phase 1，1.5~2 周）：业务任务表加字段 + collect-now 高优提升 + queued/leased 抢占 + **WebSocket 通道直上（monitor.\* 路由键 + dispatch hub 订阅扩展）+ SSE `/api/desktop/events` 同步下线**。不抢 running，收敛业务范围。
2. 中期（Phase 2，4 周）：引入 `desktop_tasks(kind=monitor)` + `execution_owner` 灰度迁移 + pull-lease 升级为 push-lease + 支持 running 抢占（安全点协议）。严格遵守灰度节奏（5% → 25% → 50% → 100%）。
3. 长期（Phase 3，持续）：ingest 分片串行 + per-tenant fair scheduling + aging + 自动降速。
4. 结果回写坚持 `API ingress -> MQ -> shard serial ingest -> projection rebuild`，不要回退到 desktop 直写 DB。

如果只保留一句话概括本方案，就是：

**业务任务存在 `monitoring_collect_tasks`（Phase 1 兼容承载现有状态流转，Phase 2 仅保留业务终态与调度属性），执行任务逐步收敛到 `desktop_tasks(kind=monitor)`（承载唯一执行态），`collect-now` 通过 high lane + 代际化协作式抢占 + outbox 模式优先插队，结果再经 MQ 分片有序入库；迁移期用 `execution_owner` 字段做双路径路由，灰度上线、reconcile 兜底、随时可回滚。**

## 附录 A. V1.1 修订项与原文档对照速查

| V1.1 章节 | V1 章节 | 修订类型 | 修订要点 |
| --- | --- | --- | --- |
| §2 修订记录 | 新增 | 新增 | V1→V1.1 变更全景 |
| §6.1 当前态 | §5.1 | 更正 | `monitor.projection.rebuild` 实际已存在 |
| §7.1 | §6.1 | 重写 | 不新增 `execution_status`；Phase 1 兼容复用现有 `status`，Phase 2 明确 `desktop_tasks` 为唯一执行真相，并补齐状态写入矩阵 |
| §7.2 | §6.2 | 补充 | 新增 UNIQUE 约束防止双写 |
| §7.3 | 新增 | 新增 | priority 值域规范 + lane_weight |
| §9.2 | §8.2 | 重写 | dispatch 预算决策树 + 拒绝矩阵 |
| §9.4 | §8.4 | 重写 | aging 百分位触发 + 对数衰减 |
| §10.2 | 新增 | 新增 | interrupt_generation 规范 |
| §10.5 | §9.4 | 重写 | 参数化超时，per-platform 配置 |
| §10.6 | 新增 | 新增 | target_client_id 绑定与漂移 |
| §12.2 | §11.2 | 精简 | 去掉 shard_key 备选维度，统一 |
| §12.3 | §11.3 | 对齐 | 与现有表约束映射 + `question_monitor_runs` upsert 语义 |
| §13.1 | §12.1 | 补充 | collection_mode 字段语义 |
| §14.3 | 新增 | 新增 | Phase 1/2/3 通道策略（**修订 2：SSE 在 Phase 1 即下线**） |
| §16.1 Phase 1 通道部分 | §15.1 | 重写 | WebSocket 通道 Phase 1 直上，SSE 同步下线（**修订 2**） |
| §14.2 / §14.4 | 新增 | 新增 | `executionPhase` 定义 + desktop 中断串行化规则 |
| §15.1 | §14.1 | 扩充 | 新增 10 项指标 |
| §15.2 | §14.2 | 重写 | 告警五元组（指标/聚合/窗口/阈值/持续时间） |
| §16.1 | §15.1 | 重写 | Phase 1 范围收敛 + WebSocket 直上 + SSE 下线 |
| §16.2 | §15.2 | 重写 | 迁移期一致性（execution_owner + 灰度 + 回滚 + reconcile） |
| §17.1 | §16.1 | 重写 | outbox 模式 |
| 附录 A | 新增 | 新增 | 修订对照表 |