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 更正）

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 推荐目标态

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 用）

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

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

统一排序规则：

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 业务任务状态机

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 起）

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 跳过：

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 函数：

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）：

-- 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；服务端比对：

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 抢占流程

用户点击 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，建议约定路由键：

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 链路（已落地）：

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 落地）

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

主选分片键（唯一）：

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 对应：

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

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"
}

返回建议：

{
  "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

{
  "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

{
  "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 能力：

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

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

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

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

13.5 立即采集请求查询

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 或同一把互斥锁内，不能分散在不同线程各自修改。

本地排序建议：

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 按该字段路由：

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. 若必须强制回迁，执行脚本：

-- 强制回迁仅在紧急故障场景使用
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）

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

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

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	新增	新增	修订对照表