RFC: Event-Driven Control Plane and Unified Policy Decision Architecture

[hufeide]

RFC: 事件驱动控制面与统一策略决策架构

• 状态： 草案 / 开放评论
• 日期： 2026-08-13
• 范围： loopx/loopx/control_plane
• 作者： LoopX contributors
• 目标： 控制面架构

---

1. 摘要

本 RFC 提议围绕五项架构原则，对 LoopX 控制面进行增量演进：

1. 策略（Policy）拥有单一决策契约。
2. 事件是决策与执行历史的事实来源（Source of Truth）。
3. 用量与成本是对既有事件的只读投影（Read Projection）。
4. Checkpoint 与重放（Replay）提供 Task 级恢复，且不引入第二事实源。
5. 执行最终走向事件驱动，heartbeat 仅作为触发源之一，而非主控制机制。

本提案刻意保持增量。

前四项变更可以在不改变现有执行路径的前提下引入。事件驱动调度被单独视为行为变更阶段，需要专门评审。

首要目标不是替换现有控制面，而是引入更清晰的架构边界，并提供一条通向更简事件驱动模型的迁移路径。

---

2. 动机

当前 control_plane 已经具备大部分所需能力：

• 配额与结算（quota and accounting）
• 能力检查（capability checks）
• Agent 范围检查（agent scope checks）
• 调度执行上下文（scheduler execution context）
• 发布事件（rollout events）
• 事件账本（event ledger）
• 运行历史（run history）
• 任务状态迁移（task state transitions）
• 心跳调度（heartbeat scheduling）
• 恢复相关兼容机制（recovery-related compatibility mechanisms）

问题主要不是功能缺失。

问题是：若干架构职责目前分散在不同模块中，且经常由 heartbeat 路径组装。

例如，判定「Agent 是否应该运行」可能需要理解：

heartbeat
  ↓
quota/should_run
  ↓
capability_gate
  ↓
agent_scope
  ↓
scheduler/execution_context

类似地，执行事实以事件形式存在，但决策历史、成本归因、恢复 Checkpoint 并未通过一致的投影暴露。

这带来若干问题：

• 调用方需要理解多套决策机制；
• 决策语义没有用单一契约表达；
• 「这个任务为什么没运行？」难以从持久化状态回答；
• 用量事实需要调用方自行聚合；
• 被中断的 Task 缺乏标准化的恢复锚点；
• heartbeat 把「触发」与「业务决策」两类职责混在一起；
• 事件存储与事件驱动执行尚未清晰分离。

---

3. 目标

本 RFC 的目标如下：

G1. 引入统一的策略决策契约

调用方应能直接询问：

decision = policy.decide(...)

而无需手动组装 quota、capability 与 scope 检查。

G2. 保留既有策略逻辑

新的 Policy 层应组装并归一化现有规则，而非重新实现。

G3. 让决策可审计

系统应能回答：

这个 Task 为什么运行、等待或被拒绝？

并且基于持久化事件回答。

G4. 提供可复用的用量/成本投影

现有 spend 事件应可按以下维度查询：

• Goal
• Task
• Agent
• 时间

且不复制结算事实。

G5. 提供 Task 级恢复

长任务应能从 checkpoint 续跑，且不引入第二个权威状态存储。

G6. 建立事件驱动执行路径

状态变化应最终能唤醒调度器，而不依赖周期性的 heartbeat 轮询。

---

4. 非目标

本 RFC 不提议：

• 重写现有 quota 实现；
• 重写现有 capability 系统；
• 替换现有事件存储；
• 引入新数据库；
• 在第一阶段替换全部 heartbeat 行为；
• 立即替换全部 legacy/migration 机制；
• 不做迁移地变更公开 API；
• 默认把每一次策略决策都持久化为事件。

---

5. 设计原则

5.1 复用，不重写

现有实现仍然是各自领域的权威。

新层次起初只是门面（facade）、适配器（adapter）与投影（projection）。

---

5.2 单一事实源

事件仍是「发生了什么」的权威记录。

投影是派生的视图。

Checkpoint 是恢复快照，不是第二事实源。

---

5.3 策略做决策；调度器管时机；Worker 执行

职责应收敛为：

Policy（策略）
  → 这次执行是否被允许？

Scheduler（调度器）
  → 它应该在何时执行？

Queue（队列）
  → 什么已经可以执行？

Worker（工作者）
  → 执行工作。

Agent Runtime（Agent 运行时）
  → 执行 Agent 级别的推理与工具调用。

---

5.4 事件存储（Event Store）与事件总线（Event Bus）是不同关注点

事件存储回答：

发生了什么？

事件总线 / 队列回答：

现在谁应该响应？

现有 rollout 事件日志不应被自动当作低延迟事件总线，除非其排序、投递、重试、消费偏移与扇出语义被明确验证充分。

---

6. 目标架构

目标架构为：

                 ┌──────────────────┐
                 │    Event Store   │
                 │  事实来源        │
                 └────────▲─────────┘
                          │
             ┌────────────┼────────────┐
             │            │            │
        Projection      Replay     Audit/Query
         （投影）      （重放）    （审计/查询）
             │            │
       ┌─────┼─────┐      │
       │     │     │      │
      Cost Decision State Checkpoint
    成本投影 决策投影 状态投影
             │
             │

触发源（Trigger Sources）
────────────────────────────────────
heartbeat / timer / cron / webhook / event
                    │
                    ▼
                Scheduler（调度器）
                    │
                    ▼
              Policy Engine（策略引擎）
                    │
                    ▼
                 Decision（决策）
                    │
              ┌─────┴─────┐
              │           │
             run       wait / deny
              │
              ▼
             Queue（队列）
              │
              ▼
            Worker（工作者）
              │
              ▼
        Agent Runtime（Agent 运行时）
              │
              ▼
            Events（事件）
              │
              └──────────────→ Event Store

这是目标方向，而非要求一次性实现所有组件。

---

7. C1 —— 统一策略决策契约

7.1 当前问题

当前执行决策分散在：

• quota/should_run.py
• quota/should_run_packet.py
• heartbeat_receipt.py
• agents/capability_gate.py
• agents/agent_scope.py
• scheduler/execution_context.py

不同层次使用不同的表示，例如：

decision（决策）
action（动作）
scope status（范围状态）

对于「这个 Task 是否应该运行？」这一问题，没有单一稳定契约。

---

7.2 提议的变更

新增：

loopx/control_plane/policy/
├── __init__.py
├── decision.py
└── engine.py

公开接口变为：

decision = PolicyEngine.decide(
    agent=...,
    task=...,
    goal=...,
    context=...,
)

初始决策契约为：

Decision（决策）
├── outcome（结果）
│   ├── run（运行）
│   ├── wait（等待）
│   └── deny（拒绝）
├── reason（原因）
├── source（来源）
├── detail（细节）
└── retry metadata（可选，重试元数据）

示例：

{
  "outcome": "wait",
  "reason": "quota_backoff",
  "source": "quota",
  "retry_at": "2026-08-13T12:30:00Z"
}

关键区分在于：

三个 outcome 是归一化的控制语义；reason/source 字段保留领域特定信息。

这避免了把 backoff、recovery、repair_bridge、ask_owner 压扁成无法区分的状态。

---

7.3 实现规则

PolicyEngine 不得重复现有领域规则。

它应组装现有纯函数：

PolicyEngine
    ├── quota（配额）
    ├── capability（能力）
    ├── scope（范围）
    └── execution-context validation（执行上下文校验）

策略引擎拥有的是组装与归一化，而非底层业务规则。

---

8. C2 —— 策略决策事件

8.1 当前问题

策略决策在调用方消费完返回值后即消失。

因此系统能记录「一次 Run 发生了」，却无法一致地回答：

这次 Run 为什么被拒绝或延迟？

---

8.2 提议的变更

新增：

policy/decision_events.py

带独立的记录器：

record_policy_decision(...)

架构应为：

PolicyEngine
      │
      ▼
   Decision（决策）
      │
      ├────────→ caller（调用方）
      │
      └────────→ DecisionRecorder（决策记录器）
                         │
                         ▼
                    Event Store（事件存储）

策略引擎本身应保持无持久化副作用。

---

8.3 决策事件

决策事件应包含稳定标识符，例如：

goal_id
todo_id
agent_id
run_id
decision/outcome（决策/结果）
reason（原因）
source（来源）
timestamp（时间戳）
decision_fingerprint（决策指纹）

敏感的任务内容与凭据不得持久化。

---

8.4 去重

重复轮询不得产生无界的相同决策事件。

初始实现应支持以下任一方式：

• 仅记录状态迁移（transition-only recording）；或
• 确定性决策指纹（deterministic decision fingerprinting）。

例如：

wait/quota
wait/quota
wait/quota
run

通常应可表示为：

wait/quota
run

而不是四条相同事件。

---

9. C3 —— 用量与成本投影

9.1 当前问题

原始 spend 事实已存在，但调用方必须手动聚合。

现有结算层负责校验并产生 spend 事实，不负责回答分析型查询。

---

9.2 提议的变更

新增只读投影：

quota/cost_projection.py

初始 API 可包括：

goal_cost_summary(goal_id)
task_cost(goal_id, todo_id)

投影可提供：

total_usage（总用量）
by_agent（按 Agent）
by_task（按 Task）
by_day（按天）

---

9.3 术语

compute_units 不得自动被当作货币成本。

第一版应区分：

usage_units（用量单位）

与：

monetary_cost（货币成本）

若供应商定价可用，货币成本可在之后从用量推导。

因此架构应为：

Spend Events（消费事件）
     ↓
Usage Projection（用量投影）
     ↓
Cost Projection（成本投影）

而非假设：

compute_units == cost

---

9.4 扩展

第一版实现可直接从既有事件聚合。

若事件量变大，同一投影契约之后可物化（materialize）：

Event Store（事件存储）
     ↓
Projection Worker（投影工作者）
     ↓
Materialized Cost Projection（物化成本投影）

本 RFC 不要求引入新的存储系统。

---

10. C4 —— Task Checkpoint 与重放

10.1 当前问题

运行历史与事件历史提供历史视图，但没有标准化的 Task 级恢复锚点。

长任务在中断后可能不得不从初始状态重新开始。

---

10.2 提议的模型

新增：

runtime/checkpoint.py
runtime/replay.py

Checkpoint 是恢复优化，不是权威状态存储。

模型为：

Event Store（事件存储）
     │
     ├── E1
     ├── E2
     ├── E3
     ├── E4
     ├── E5  ← checkpoint（检查点）
     ├── E6
     ├── E7
     └── E8

恢复变为：

Checkpoint Snapshot（检查点快照）
       +
Events after checkpoint（检查点之后的事件）
       ↓
Current Task State（当前 Task 状态）

---

10.3 Checkpoint 契约

一个 checkpoint 至少应包含：

checkpoint_id（检查点 ID）
goal_id
todo_id
run_id
last_event_id（最后事件 ID）
schema_version（模式版本）
state_snapshot（状态快照）
state_hash（状态哈希）
created_at（创建时间）

其中 last_event_id 尤其重要，它确定了重放应从何处续起。

---

10.4 重放要求

重放必须是：

• 确定性的（deterministic）；
• 幂等的（idempotent）；
• 无副作用的（side-effect free）；
• 感知 schema 版本的（schema-version aware）。

重放应重建状态，而非执行外部副作用。

例如：

Event（事件）
  ↓
State Transition（状态迁移）
  ↓
New State（新状态）

不得在重放过程中意外触发：

API call（API 调用）
email（邮件）
LLM invocation（LLM 调用）
file mutation（文件修改）

---

10.5 兼容性

Checkpoint/重放最初应与现有 migration 与兼容机制并存。

本 RFC 不要求立即移除 shadow、migration 或 legacy 路径。

一旦重放证明了等价恢复语义，可在单独的变更中评估这些机制的缩减或移除。

---

11. C5 —— 事件驱动调度

11.1 当前问题

Heartbeat 当前承担两个职责：

产生周期性唤醒
+
执行控制面决策编排

这导致即使状态毫无变化，周期性轮询仍反复检查状态。

---

11.2 提议的方向

把 heartbeat 视为若干触发源之一：

heartbeat（心跳）
timer（定时器）
cron
webhook
manual（手动）
domain event（领域事件）

所有触发源汇聚到调度器：

Trigger（触发）
   ↓
Scheduler（调度器）
   ↓
Policy（策略）
   ↓
Queue（队列）
   ↓
Worker（工作者）

因此 heartbeat 变成触发机制，而不再是控制面决策的拥有者。

---

11.3 事件驱动示例

一个依赖完成：

Task T1 Completed（T1 完成）
       ↓
Task dependency event（任务依赖事件）
       ↓
Scheduler（调度器）
       ↓
T2 becomes READY（T2 变为就绪）
       ↓
Queue（队列）
       ↓
Worker（工作者）

系统无需等待下一个 heartbeat 间隔。

---

11.4 重要约束

现有 rollout 事件日志不应自动成为执行事件总线。

若事件驱动调度需要：

• 低延迟（low latency）；
• 消费偏移（consumer offsets）；
• 重试（retries）；
• 排序（ordering）；
• 投递保证（delivery guarantees）；
• 扇出（fan-out）；
• 背压（backpressure）；

则可能需要显式的队列/总线抽象。

事件存储仍负责持久事实与重放。

---

12. 迁移计划

迁移应分阶段进行。

阶段 1 —— 策略门面（Policy Facade）

新增：

policy/

不改动任何现有执行路径。

新增单元测试与契约测试。

---

阶段 2 —— 决策投影（Decision Projection）

新增：

policy/decision_events.py

记录保持可选（opt-in）。

验证：

• 事件模式（event schema）；
• 去重（deduplication）；
• 公开安全字段（public-safe fields）；
• 审计查询（audit queries）。

---

阶段 3 —— 用量投影（Usage Projection）

新增：

quota/cost_projection.py

对照现有 spend 事实验证聚合。

结算行为不变。

---

阶段 4 —— Checkpoint / 重放

新增：

runtime/checkpoint.py
runtime/replay.py

验证：

• 确定性重放（deterministic replay）；
• checkpoint 完整性（checkpoint integrity）；
• schema 兼容性（schema compatibility）；
• 中断 Task 恢复（interrupted Task recovery）。

---

阶段 5 —— 策略集成（Policy Integration）

评估用以下方式替换 heartbeat 决策组装：

PolicyEngine.decide(...)

这是现有控制路径可能发生行为变更的第一个阶段。

---

阶段 6 —— 事件驱动调度试点（Event-Driven Scheduling Pilot）

从一条狭窄的 Task 生命周期开始：

TaskCompleted（任务完成）
    ↓
dependency satisfied（依赖满足）
    ↓
TaskReady（任务就绪）
    ↓
Queue（队列）
    ↓
Worker（工作者）

度量：

• 调度延迟（scheduling latency）；
• CPU 利用率（CPU utilization）；
• 重复执行（duplicate execution）；
• 漏事件（missed events）；
• 恢复行为（recovery behavior）。

只有试点成功之后，才应迁移更广泛的 heartbeat 行为。

---

13. 向后兼容

初始阶段应保留现有行为。

兼容规则为：

旧路径（Old path）
   ↓
继续工作

新策略门面（New Policy facade）
   ↓
使用旧的策略函数

新架构最初应表现为现有实现之上的适配器，而非替换它。

任何行为变更型迁移都必须单独评审。

---

14. 测试策略

每个阶段都应包含契约测试。

策略（Policy）

测试每一条现有决策映射：

run
skip
backoff
recovery
repair_bridge
deny
missing_capability
ask_owner
...

并验证归一化后的：

outcome + reason + source

契约。

决策事件（Decision Events）

测试：

• 确定性事件身份（deterministic event identity）；
• 重复抑制（duplicate suppression）；
• 公开安全序列化（public-safe serialization）；
• 查询正确性（query correctness）。

用量投影（Usage Projection）

将投影结果与独立聚合的 spend 事件对照。

重放（Replay）

验证：

replay(events) == replay(events)

以及：

checkpoint + events_after_checkpoint（checkpoint + 检查点之后的事件）
==
full_replay(events)（全量重放）

事件驱动调度（Event-Driven Scheduling）

测试：

• 至少一次投递（at-least-once delivery）；
• 重复事件（duplicate events）；
• 漏事件（missed events）；
• 重试（retry）；
• 队列背压（queue backpressure）；
• Worker 失败（worker failure）。

---

15. 风险

R1. 决策语义丢失

缓解：

使用：

outcome + reason + source + metadata

而非仅三个字符串。

---

R2. 决策事件爆炸

缓解：

使用状态迁移检测或确定性指纹。

---

R3. 把事件存储误当事件总线

缓解：

将持久事件存储与低延迟事件投递保持为独立抽象。

---

R4. 重放不兼容

缓解：

对 checkpoint 与状态 schema 做版本管理。

---

R5. 多事实源

缓解：

事件保持权威。

投影与 checkpoint 是派生产物。

---

R6. 迁移复杂度

缓解：

先用增量变更，推迟行为变更型重新接线，直到新组件有了独立的测试。

---

16. 曾考虑的备选方案

备选方案 A —— 继续扩展 heartbeat

作为长期方向被否决，因为 heartbeat 会继续累积：

• 策略（policy）；
• 配额（quota）；
• 恢复（recovery）；
• 调度（scheduling）；
• 状态检查（state inspection）；
• 事件协调（event coordination）。

---

备选方案 B —— 重写控制面

被否决，因为现有实现已包含大量领域逻辑与兼容行为。

重写会增加迁移风险，且不一定改善概念模型。

---

备选方案 C —— 引入全新的状态存储

暂被否决。

现有事件基础设施已提供所需的事实源语义。

首要目标应是在现有事实上暴露更好的投影与恢复机制。

---

备选方案 D —— 把事件存储当作执行队列

被否决，除非现有事件存储能提供显式队列语义。

持久历史与执行投递具有不同的需求。

---

17. 开放问题

以下问题应在 RFC 评审期间解决，而非藏在实施计划里。

Q1. 决策契约

wait 是否应包含显式重试语义？

例如：

retry_at（重试时间）
retry_after（重试间隔）
manual_approval_required（是否需要人工批准）

---

Q2. 决策事件频率

应记录每一次决策，还是仅记录决策迁移？

---

Q3. 事件总线

LoopX 是否已有合适的队列/事件投递机制，还是应从事件存储中独立引入一个？

---

Q4. Checkpoint 存储

checkpoint 应放在：

现有事件基础设施内部

还是作为被事件引用的独立快照产物？

---

Q5. 重放边界

哪些 Task 状态迁移被保证可重放？

---

Q6. 成本语义

第一版投影应只暴露：

usage_units（用量单位）

还是同时引入货币定价？

---

Q7. Heartbeat 迁移

在不改变全局行为的前提下，可迁移到事件驱动调度的最小 Task 生命周期是什么？

---

18. 提议的验收标准

当满足以下条件时，本 RFC 应视为成功验证：

1. 调用方无需了解底层 quota/capability/scope 模块即可获得统一的 Decision。
2. 现有策略规则仍是领域决策的来源。
3. 策略决策可（可选地）通过持久化事件审计。
4. 重复的相同决策不会造成失控的事件增长。
5. 用量可按 Goal、Task、Agent 与时间段查询。
6. 一个 Task 能通过「checkpoint + 后续事件」恢复。
7. 重放是确定性的且无副作用。
8. 事件存储与事件投递被清晰分离。
9. 一条狭窄的事件驱动调度路径可以脱离 heartbeat 轮询运行。
10. 在显式评审的迁移阶段之前，现有执行行为保持不变。

---

19. 结论

本提案架构是增量过渡而非重写。

关键的架构转变是：

当前（Current）

Heartbeat
   ↓
分散的控制逻辑（distributed control logic）
   ↓
scheduler（调度器）
   ↓
agent

转向：

Trigger（触发）
   ↓
Scheduler（调度器）
   ↓
Policy（策略）
   ↓
Queue（队列）
   ↓
Worker（工作者）
   ↓
Agent
   ↓
Events（事件）
   ↓
Event Store（事件存储）
   ↓
Projections / Replay（投影 / 重放）

最重要的边界不是任何一个新模块。

而是以下各层的分离：

Decision（决策）
Scheduling（调度）
Execution（执行）
Facts（事实）
Projection（投影）
Recovery（恢复）

一旦这些边界显式化，现有 LoopX 实现即可增量演进，而无需控制面重写。

因此推荐的实施顺序为：

Policy Contract（策略契约）
      ↓
Decision Events（决策事件）
      ↓
Usage Projection（用量投影）
      ↓
Checkpoint / Replay（检查点 / 重放）
      ↓
Policy Integration（策略集成）
      ↓
Event-Driven Scheduling Pilot（事件驱动调度试点）

前四个阶段主要是增量性的。

最后的调度迁移被刻意留作独立的行为变更提案，且只有在既有契约与测试稳定之后才应推进。