LobeHub Agent Signal 架构解析：从 source 到 action 的信号驱动运行时

LobeHub 定位为「Chief Agent Operator」，让 Agent 以 7×24 方式持续运转；要支撑这一形态，就必须把记忆写入、技能管理、夜间复盘等「后台工作」从前台聊天请求中解耦出来。Agent Signal 正是承担这一解耦的信号驱动运行时：本文将完整解读其三层包边界（共享语义核心 / 服务端实现层 / OTEL 归属层）、source → signal → action 的队列式调度模型、scopeKey 去重与静默入队机制，以及 analyzeIntent 参考链的端到端流程，帮助读者理解并扩展 LobeHub 中任意事件驱动的 Agent 后台能力。

一、整体 Pipeline：先建立心智模型

Agent Signal 的运行时形态只有一条固定链路。理解下面的三个流水线图，是阅读后续所有章节的前提。

1. 主链路：生产者 → 结果信号 → 可观测投影

producer
  -> emitAgentSignalSourceEvent(...) or enqueueAgentSignalSourceEvent(...)
    -> emitSourceEvent(...)
      -> dedupe + scope lock + source normalization
        -> runtime.emitNormalized(source)
          -> source handlers
            -> signal handlers
              -> action handlers
                -> built-in result signals
                  -> observability projection + persistence

即：生产者（producer）负责发出一个「来源事件」（source event），入口先做去重（dedupe）与 scope 锁，再完成来源归一化（normalization），随后进入运行时逐层分发：来源处理器解释事实 → 信号处理器产生语义 → 动作处理器执行副作用 → 内置结果信号上报，最终被可观测层投影并持久化。

2. 异步自迭代分支：记忆 / 技能 / 复盘写入

对于持久记忆（memory）、技能（skill）与自我复盘（self-review）这类重写入，action handler 不做同步落库，而是入队一次异步的 execAgent 运行：

action.user-memory.handle | action.skill-management.handle | self-iteration source
  -> stamp AgentSignal operation marker
  -> enqueue execAgent
    -> agent.execution.completed with selfIteration finalState
      -> completion policy
        -> buildSelfIterationReceipts(...)
          -> receipt store

关键纪律在于：即时的运行时链路只为「入队结果」发出 signal.action.applied | skipped | failed；用户可见的记忆与技能回执（receipt）是从 agent.execution.completed 完成事件投影出来的，而不是从入队动作本身投影出来的。这一设计避免了「入队成功 ≠ 写入成功」的语义混乱。

3. 调度器是队列驱动的，不为某一种策略硬编码

source node
  -> matching source handlers
    -> dispatch signals/actions
      -> matching signal handlers
        -> dispatch more signals/actions
          -> matching action handlers
            -> ExecutorResult
              -> signal.action.applied | signal.action.skipped | signal.action.failed

从源码结构看，调度器位于 AgentSignalScheduler.ts，运行时主体在 AgentSignalRuntime.ts：处理器返回 RuntimeProcessorResult（见 base/types.ts），其中 dispatch 状态可以携带新的 signals 与 actions 继续派发，这就是图中「dispatch more signals/actions」的实现基础——调度器本身是通用的，行为完全由注册的 policy 决定。

核心入口代码：

• apps/server/src/services/agentSignal/index.ts
• apps/server/src/services/agentSignal/sources/index.ts
• apps/server/src/services/agentSignal/runtime/AgentSignalScheduler.ts

二、包边界：三层职责严格分离

2.1 packages/agent-signal：共享语义核心

把它当作「共享语义核心」：只定义类型、builder 与契约，不含任何服务端策略。它提供：

• 基础节点类型：source、signal、action；
• 构建器：createSource、createSignal、createAction；
• 共享的 source 类型与 payload 目录；
• source-event 信封与 scope 辅助；
• 内置结果信号类型；
• 运行时结果契约，如 RuntimeProcessorResult 与 ExecutorResult。

从源码可以确认这些声明：base/builders.ts 中 createSource / createSignal / createAction 会生成带因果链元数据的规范节点——chain 字段（rootSourceId、parentNodeId、parentSignalId 等）贯穿三层节点，保证任意 action 都能回溯到根 source。

执行结果契约在 base/types.ts 中定义得很严格：

• ExecutorResult 是判别联合：成功或跳过为 status: 'applied' | 'skipped'（不允许携带 error），失败为 status: 'failed' 且必须携带结构化 ExecutorError（含 code、message、retriable、retryAfterMs）；
• RuntimeProcessorResult 是四态联合：wait / dispatch / schedule / conclude，分别表达「宿主等待」「继续派发新节点」「调度下一跳」「结束链路」；
• 去重结果 EmitSourceEventResult 同样为判别联合：被去重时返回 { deduped: true, reason: 'duplicate' | 'scope_locked' }。

推荐阅读顺序：

• packages/agent-signal/src/base/types.ts
• packages/agent-signal/src/base/builders.ts
• packages/agent-signal/src/source/sourceTypes.ts
• packages/agent-signal/src/source/sourceEvent.ts
• packages/agent-signal/src/source/scopeKey.ts
• packages/agent-signal/src/types/events.ts
• packages/agent-signal/src/types/builtin.ts

2.2 apps/server/src/services/agentSignal：服务端实现层

这是「服务端自有实现层」，拥有：

• source 归一化、hydration 与 renderers；
• 策略专属的 signal 与 action 目录；
• 中间件注册；
• 运行时调度与 guard 后端；
• 基于 Redis 的去重、waypoint 与策略状态；
• 同步 / 异步执行的服务入口；
• 完成态自迭代运行的回执投影与持久化。

从目录结构看，这些职责对应清晰子目录：sources/（含 buildSource.ts、renderers/、hydration/）、runtime/（含 backend/redisGuard.ts 与 memoryGuard.ts 两种 guard 后端）、policies/、services/（含 selfIteration/completion/）、observability/、store/ 等。

2.3 packages/observability-otel/src/modules/agent-signal：OTEL 归属

该模块是 Agent Signal 指标与 tracer 实例的共享 OTEL 归属层，对应 packages/observability-otel/src/modules/agent-signal/index.ts，保证指标/追踪在包之间的一致性。

三、核心词汇：Source / Signal / Action / Policy / Procedure

3.1 Source：被归一化的外部事实

Source 是启动整条链路的「被归一化的外部事实」。文档示例包括 agent.user.message、runtime.before_step、runtime.after_step、client.runtime.start、bot.message.merged。

实际 source 目录比示例更完整。source/sourceTypes.ts 中的 AGENT_SIGNAL_SOURCE_TYPES 定义了 17 种 source 类型：

Source 类型	典型触发场景
agent.user.message	用户消息进入，反馈分类主入口
agent.execution.completed	一次 agent 运行完成（携带 selfIteration 终态侧载）
agent.execution.failed	agent 运行失败
agent.nightly_review.requested	夜间复盘请求
agent.self_reflection.requested	自我反思请求
agent.self_feedback_intent.declared	Agent 主动声明反馈意图（memory/skill/gap）
bot.message.merged	外部 bot 平台消息合并
runtime.before_step / runtime.after_step	运行时步骤前后钩子
client.runtime.start / client.runtime.complete	客户端运行时生命周期
client.gateway.stream_start / step_complete / runtime_end / error	客户端网关事件
tool.outcome.completed / tool.outcome.failed	工具结果

每种类型都有强类型 payload（AgentSignalSourcePayloadMap），并提供 isAgentUserMessageSource、isClientRuntimeStartSource、isToolOutcomeSource 等类型收窄函数；另外 AGENT_SIGNAL_CLIENT_SOURCE_TYPES 明确界定了哪些 client.* 类型可以经认证边缘由浏览器侧生产者发出。

定义与构建位置：

• 类型与 payload：packages/agent-signal/src/source/sourceTypes.ts
• 归一化信封：packages/agent-signal/src/source/sourceEvent.ts
• 服务端构建：apps/server/src/services/agentSignal/sources/buildSource.ts、apps/server/src/services/agentSignal/sources/renderers/
• builder：packages/agent-signal/src/base/builders.ts

3.2 Signal：可复用的语义解释

Signal 是「语义解释」，应当可复用、面向含义而非面向事件。analyzeIntent 策略中的例子：

• signal.feedback.satisfaction（反馈满意度）
• signal.feedback.domain.memory / signal.feedback.domain.prompt / signal.feedback.domain.skill（反馈领域分类）

服务端自有的 signal 类型定义在 apps/server/src/services/agentSignal/policies/types.ts。

3.3 Action：具体的副作用

Action 是运行时应当执行的「具体副作用」，例如 action.user-memory.handle。Action handler 通常做三件事：检查幂等、调用工具/模型/服务、返回 ExecutorResult。

3.4 Policy：可安装的处理程序束

Policy 是一个「可安装的 handler bundle」，是把通用运行时组装成功能单元的组合单位，例如 createAnalyzeIntentPolicy(...)。

3.5 Procedure：不是运行时类型

「Procedure」在此运行时中不是一等类型。这个词只用于描述一个端到端用例：

1. 定义入口 source；
2. emit 或 enqueue 该 source；
3. 将 source 解释为 signals；
4. 从 signals 规划 actions；
5. 执行 actions；
6. 持久化 trace 与 metrics。

当有人问「这个流程（procedure）是什么」时，应指向上面的链路，并给出具体的 producer、handlers 与执行入口。

四、Scope、去重与静默后台工作

4.1 scopeKey：相关工作的串行化边界

scopeKey 是相关工作的串行化边界，用于四处：

• source 去重窗口；
• source 生成期间的 scope 锁；
• 运行时 guard 状态；
• 队列化处理的 waypoint 持久化。

推导规则在 source/scopeKey.ts 中非常明确，AgentSignalScopeKey 提供五类前缀：

topic:<topicId>          # 主题级，最高优先级
task:<taskId>            # 异步任务级
bot:<platform>:<applicationId>:<platformThreadId>
agent:<agentId>:user:<userId>
user:<userId>            # 最宽泛的认证范围

fromProducerInput 的推导优先级为：topicId > taskId > bot 三元组 > agentId+userId > userId，全部缺失时返回 fallback:global。服务端在 sources/index.ts 复用同一套键，常量见 constants.ts，运行时上下文见 runtime/context.ts。

4.2 enqueueAgentSignalSourceEvent：让 UI 立即返回

当工作应当「安静地」在带外执行时，使用 enqueueAgentSignalSourceEvent(...)。该路径分四步：

1. 归一化 source 信封；
2. 推导或复用 scopeKey；
3. 触发 AgentSignalWorkflow；
4. 之后由 runAgentSignalWorkflow 执行。

从 emitter.ts 的源码可以看到这条链路的真实实现：

• 先通过 assertAgentUsableBy 做安全门：防止工作区成员对他人私有 agent 入队事件；
• 再经 isAgentSignalEnabledForUser 特性门（feature gate），未开启的用户直接返回 { accepted: false } 与推导出的 scopeKey；
• 经 createSourceEvent(input) 生成规范信封；
• 最后 AgentSignalWorkflow.triggerRun({ sourceEvent, agentId, userId, workspaceId }) 入队，返回 { accepted: true, scopeKey, workflowRunId }。

对比之下，同步入口 emitAgentSignalSourceEvent 会在检查特性门后动态 import('./orchestrator')——源码注释解释了原因：orchestrator 会拖入 agent-execution / model-runtime 核心子系统（其模块初始化即触碰服务端环境变量），而 emitter 位于轻量请求路径上，静态导入会污染所有导入方。

4.3 投递模式由 AGENT_RUNTIME_MODE 决定

• queue 模式：使用 Upstash Workflow 做持久化执行，提供重试与流控；
• local 模式：用 setTimeout 延迟一次进程内运行，返回合成的 workflowRunId，刻意不提供持久化与重试。local 模式使用进程级 source-event store 与运行时 guard 来维持去重、scope 锁、窗口与 action 幂等性——即在没有 Redis 的情况下仍然「可用」，且这些保证在单个服务端进程生命周期内有效。

workflow 入口：apps/server/src/workflows/agentSignal/index.ts 与 apps/server/src/workflows/agentSignal/run.ts。

选择准则很直接：当 UI 请求应当立即结束、而策略可以在后台跑时，这是首选路径。

五、参考实例：analyzeIntent 完整链路

以 analyzeIntent 作为参考链，它示范了 source → signal → action → 结果信号的完整形态：

agent.user.message
  -> feedback satisfaction source handler
    -> signal.feedback.satisfaction
      -> feedback domain signal handler
        -> signal.feedback.domain.*
          -> feedback action planner
            -> action.user-memory.handle | action.skill-management.handle
              -> signal.action.applied | skipped | failed

当前策略还包含：工具结果投影（tool-outcome projection）、技能管理（skill management）、延迟完成态技能合成（deferred completion skill synthesis）、夜间复盘（nightly review）与完成态扇出（completion fan-out）：

action.user-memory.handle | action.skill-management.handle
  -> enqueue execAgent with AgentSignal marker
    -> agent.execution.completed
      -> completion policy
        -> buildSelfIterationReceipts(...)
          -> memory | skill | review receipts

一个值得注意的细节：技能合成支持「停车—恢复」——入站用户消息候选可以在前台链路中被暂存（parked），等到 agent.execution.completed 之后再恢复处理，此时完整轨迹与工具结果都可用，合成质量显著高于只看到单条消息时。

该策略的源码分布（均位于 apps/server/src/services/agentSignal/ 下）：

• 策略入口：policies/analyzeIntent/index.ts
• 满意度源处理：policies/analyzeIntent/feedbackSatisfaction.ts
• 领域分类：policies/analyzeIntent/feedbackDomain.ts
• 动作规划：policies/analyzeIntent/feedbackAction.ts
• 动作实现：policies/analyzeIntent/actions/userMemory.ts、policies/analyzeIntent/actions/skillManagement.ts
• 完成态技能合成：policies/analyzeIntent/completionSkillSynthesis.ts
• 完成策略：policies/completionPolicy.ts
• 回执构建与投影：services/selfIteration/completion/buildSelfIterationReceipts.ts、services/selfIteration/completion/selfIterationCompletionHandler.ts

六、入口点选择与实现清单

6.1 四个执行入口各司其职

结合 SKILL.md 的约定：

• emitAgentSignalSourceEvent(...)：服务端生产者需要立即执行 pipeline 时使用；
• executeAgentSignalSourceEvent(...)：worker 或受控后端路径已拥有执行时机、可能需要注入自定义 runtime guard 后端时使用；
• enqueueAgentSignalSourceEvent(...)：调用方需要快速返回、把事件交给 Upstash Workflow 带外处理时使用；
• emitAgentSignalSourceEventWithStore(...)：隔离测试或 eval 中需要绕开环境级 Redis 状态时使用。

6.2 新增一个流程的实现路径

按架构文档与 SKILL 给出的顺序推进：

1. 判断用例是同步还是静默后台工作；
2. 在 source/sourceTypes.ts 定义或复用 source 类型；
3. 在 policies/types.ts 定义或复用 signal / action 类型；
4. 用 defineSourceHandler、defineSignalHandler、defineActionHandler 实现处理器；
5. 如生产者 payload 需要整形，在 apps/server/src/services/agentSignal/sources/** 增加归一化或 hydration；
6. 用 defineAgentSignalHandlers(...) 打包处理器；
7. 在 apps/server/src/services/agentSignal/policies/index.ts 注册 policy，并在需要时传入运行时工厂；
8. 增加或更新 emit / enqueue 的入口代码；
9. 异步自迭代写入时：打上 Agent Signal operation marker，并从完成路径（createSelfIterationCompletionHandler(...) + agent.execution.completed 的 selfIteration finalState）投影用户可见回执；
10. 补齐可观测性与测试后再视为完成。

实现守则（来自技能规范与源码契约的共同约束）：

• 先复用已有 source / signal / action 类型，再考虑新增；
• source handler 只负责解释与扇出，不放重副作用；action handler 负责副作用、幂等与 executor 风格的结果上报；
• 同一 source 可能多次到达时，使用稳定的 id 与幂等键；
• 保持 scope 纪律：运行时用 scopeKey 串行化相关后台工作；
• 不要从入队动作投影 memory / skill 回执——必须走完成事件；
• 在触达的 runtime / policy / store 模块附近补测试，apps/server/src/services/agentSignal/**/__tests__ 下现有用例（如 scopeKey.test.ts、triggerSourceEvent.test.ts、index.integration.test.ts）是参照模式。

七、延伸阅读地图

围绕本文主题，仓库中值得继续深入的位置：

• 语义核心导出：packages/agent-signal/src/index.ts
• 运行时与中间件：runtime/AgentSignalRuntime.ts、runtime/middleware.ts
• guard 后端：runtime/backend/redisGuard.ts、runtime/backend/memoryGuard.ts
• 可观测投影：observability/projector.ts、observability/traceEvents.ts
• 配套技能文档：handlers 写法参考、observability 参考、架构原文

小结：Agent Signal 用一个「source → signal → action → 结果信号」的固定形状，把 LobeHub 中所有事件驱动的 Agent 后台工作统一为可组合、可去重、可幂等、可观测的队列式管道；scopeKey 保证相关工作的串行化纪律，AGENT_RUNTIME_MODE 决定持久化与重试的级别，而 memory/skill 回执「只从完成事件投影」的规则则保证了用户可见状态的最终一致性。掌握这套词汇与边界，即可在 LobeHub 中安全地扩展任何新的 Agent 后台能力。