eventbus

package
v0.31.0 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Nov 11, 2026 License: MIT Imports: 5 Imported by: 0

Documentation

Overview

Package eventbus 事件单源双写(PHASE1_PLAN §2 决策 7): 引擎只向本总线 Publish 一次,订阅者分别落 bq_event(repo)与 SSE(handler)。

Index

Constants

View Source
const (
	EventSessionCreated      = "session_created"
	EventIntentRouted        = "intent_routed"
	EventDispatchDenied      = "dispatch_denied"
	EventStepStarted         = "step_started"
	EventStepFinished        = "step_finished"
	EventReportValidated     = "report_validated"
	EventVerdictJudged       = "verdict_judged" // 结论合议裁决留痕(spec 2026-09-15-verdict-judge-design §8)
	EventReportViolation     = "report_violation"
	EventApprovalRequested   = "approval_requested"
	EventApprovalDecided     = "approval_decided"
	EventApprovalExpired     = "approval_expired"
	EventExecutionDryrun     = "execution_dryrun"
	EventExecutionDegraded   = "execution_degraded"   // 执行面缺失(MCP 池未装配),已批准方案降级 DryRun 的显式留痕
	EventRemediationExecuted = "remediation_executed" // 受审变更执行回执(批次九十四):批准→执行完成后的汇总留痕(executor/change_ref/host/步骤数)
	EventSessionFinished     = "session_finished"
	// EventDecisionAnswered 判定面结果留痕(来源/置信语义/判定值/是否采纳)。
	// 载荷键面见 protocol.DecisionEventPayload(扁平键,跨宿主聚合的共同底座)。
	EventDecisionAnswered = "decision_answered"
	// EventDecisionEscalated 判定面升级/回落留痕(判定面 → 生成面 / 人工的移交原因)。
	EventDecisionEscalated = "decision_escalated"
	// 轮次生命周期事件对(批次二百八十四,对标 DeepSeek harness turn 括号语义+「锁最后
	// 释放」纪律):turn_started 在会话认领成功后立即落库,turn_ended 在收尾漏斗出口
	// 落库——「有 start 无 end」的残留=中断轮的可检测孤儿态(发版重启/进程死亡不再依赖
	// 事件静默启发式定性,时间线与诊断大脑可直接读出轮边界与中断点)。停机路径不落
	// turn_ended(轮括号留给恢复循环/启动对账以 closed_by=recovery|boot-reconcile 关闭,
	// 谎报「正常收尾」比留开放括号更伤审计)。
	EventTurnStarted = "turn_started"
	EventTurnEnded   = "turn_ended"
	// EventReportSuperseded 终报覆写版本尾迹(批次一百三十七):成功收尾覆写 final_report
	// 前旧报告非空且不同时发布——bq_event 持久化留痕,控制台回放可拼历史版本链(载荷只
	// 带头尾摘录防大字段;旧报告全文仍以折叠进压缩面的路径为准)。
	EventReportSuperseded = "report_superseded"

	// 过程可观测事件(此前是调度器里的字符串字面量,收编进词表同源管理):
	// reasoning/text 为 agent 过程思考与中间答复;tool_call/tool_result 经 call_id 配对;
	// llm_usage 为单次 LLM 调用真实用量(DeepSeek harness 式 token 面板的数据源)。
	// reasoning_delta/text_delta 为流式增量分片(agentkit v0.10.32 默认开流):生成过程
	// 中按 ~400ms/200 字符节流发布(seq 单调、text=分片非快照),同轮权威 reasoning/
	// text 快照到达即吸收——真流式渲染的数据源,回放/对账以权威快照为准。
	EventReasoning      = "reasoning"
	EventAgentText      = "text"
	EventToolCall       = "tool_call"
	EventToolResult     = "tool_result"
	EventLLMUsage       = "llm_usage"
	EventReasoningDelta = "reasoning_delta"
	EventTextDelta      = "text_delta"

	// 报告增量流(SSE 真打字机):终态成功收尾时把定稿报告按 markdown 行边界分片、
	// 节奏化发布(seq/total/text)。经 s.pub 落库保 bq_event 游标不变量;回放重放
	// 无害(同 seq 同文,前端按 seq 归并、终态权威报告覆盖)。
	EventReportDelta = "report_delta"

	// 用户输入事件(多轮对话的用户侧留痕):创建/注入时由 SessionSvc 经调度器出口发布,
	// 时间线可见,同时是下一轮对话历史(ListEventsByType)的数据源。
	EventUserInput = "user_input"

	// 上下文构成事件(批次八十一,对标 Octop ContextWindowRing):每环节装配完成后
	// 由 runner OnContext 回调发布,payload.segs=段名→字符数(soul/prompt/defense/
	// skills/skills_list/memory/env_notes/todo/query/tools)——前端构成条的数据源。
	// 批次二百八十七扩压力读数四字段(pressure_tokens/pressure_baseline/window_tokens/
	// window_pct,dsh token-meter 对标:usage 锚定与启发式取大)。
	EventContextComposition = "context_composition"

	// 上下文压力高水位告警(批次二百八十七,pressureHighWater=0.85):请求压力达窗口
	// 高水位时 warn 级显形(压力是事实不是错误,不阻断);同时是压缩面预折叠闸的
	// 触发信号(compact.go pressureHot)。
	EventContextPressure = "context_pressure"

	// 协议级重试留痕(批次二百九十五,dsh assistant/attempt 等价物):首跑失败决定
	// 回喂重试时 warn 级显形(payload.reason=失败原因摘要)——时间线可解释「两段近似
	// 重复环节」(此前无标记,排障误读为模型重复劳动)。agentkit 内部 LLM 调用级重试
	// 不在本面(实现侧在 agentkit,挂账)。
	EventAttemptRetried = "attempt_retried"
	// EventPtcRun 程序化工具编排明细(批三百零五):一次 run_program 执行的账目面——
	// bindings 调用清单/各预算余量/中断原因(程序把大输出「洗」成小结果时账目不缺)。
	EventPtcRun = "ptc_run"

	// 任务清单事件(执行进程浮层,对标 ZCode 进程面板):todo_plan 全量替换成功后由
	// runner OnTodos 回调发布,payload.todos=完整清单([{content,status}])、done/total
	// =完成数/总数——控制台会话页右上角进度浮层数据源;经 s.pub 落库,回放自水合。
	EventTodoUpdated = "todo_updated"

	// 记忆召回事件(批次一百三十五,「诊断大脑」观测台一期缺口补齐):分层记忆召回
	// 装配非空时由 runner OnMemoryRecall 回调发布,payload.chars=召回块字符数、
	// preview=首 240 字——此前召回只有 DB Hits 滞后计数无事件,脑图海马区活跃度数据源。
	EventMemoryRecalled = "memory_recalled"

	// 排队输入生命周期事件(真队列化 P1-1/P1-2/P2-3):abandoned=失败/预算禁续等收尾
	// 未消费的明示留痕;removed=队列管理面单条移除(气泡「已移除」判定依据)。
	EventQueueAbandoned = "queue_abandoned"
	EventQueueRemoved   = "queue_removed"

	// 选模悬空回落事件(批次二百五十二,spec dialog-zcode-alignment §3):会话设了
	// model_profile 但档案链已不在已装配快照(档案被删/停用/密钥摘除/mock 形态)——
	// 本轮按平台角色路由继续(回落是设计行为,悬空不得炸会话),事件显形「你选的模型
	// 已不可用」防静默换脑。payload=profile/returned_model。每轮一条(轮级判据)。
	EventModelFallback = "model_fallback"

	// 目标驻留事件(P3-1/P3-2):set=目标绑定;verified=judge 完成验证器裁决留痕
	// (passed/next_action/iteration 随 payload);completed=目标达成收敛。
	EventGoalSet       = "goal_set"
	EventGoalVerified  = "goal_verified"
	EventGoalCompleted = "goal_completed"

	// 跨轮历史压缩事件(批次三十六,对标 ZCode CompactBoundary):【历史压缩】环节把
	// 旧摘要+近轮结论折叠为结构化摘要后发布——before/after_tokens 为估算口径
	// (EstimateTokens),summary_chars/pending_count 供时间线卡与排障。
	EventHistoryCompacted = "history_compacted"

	// 跨副本取消控制事件(P2-1):Cancel API 在本副本信号未命中时发布,各副本监视器
	// 轮询 bq_event 命中本副本取消句柄即协作中止(append-only 幂等:非执行副本 no-op)。
	EventSessionCancelRequested = "session_cancel_requested"

	// 澄清反问观测事件(P3-5):反问轮此前仅 intent_routed 的 question 字段可辨,
	// 无独立可观测事件——补一条显式留痕(keyword/question 随 payload)。
	EventClarifyAsked = "clarify_asked"

	// Spawn 原语生命周期事件(批次二百四十一 R4 笼子四补位):dispatch_agent 单派与
	// dispatch_explore 各路勘探子任务的起止结构化留痕——spawn_id 配对两事件,
	// caller/target/mode/status/latency_ms 随 payload。此前 spawn 子任务只能靠父环节
	// tool_call/tool_result 的文本痕迹观察,无结构化父子归因;本事件是前端「进行中
	// spawn」进行式卡与审计面的数据源(StepID=父环节,落库/广播经调度器 pub 单源)。
	EventSpawnStarted  = "spawn_started"
	EventSpawnFinished = "spawn_finished"

	// 会话产物登记(批次八十六,spec COMPETENCE_UPLIFT_SPEC §5):版本化交付物
	// 落库即发布(artifact_id/kind/title/seq/mime/size 随 payload)——时间线产物卡
	// 与 SSE 实时刷新同一事件源。
	EventArtifactRegistered = "artifact_registered"

	// 平台化插拔审计事件(spec §4 操作审计):无会话归属的系统级事件——启停/重载
	// 的操作者/时间/slug 落台账,与决策树留痕同一套台账哲学。
	EventExpertToggled   = "expert_toggled"
	EventExpertsReloaded = "experts_reloaded"
	EventMcpToggled      = "mcp_toggled"    // MCP server 运行时启停(页面管理)
	EventMcpGovernance   = "mcp_governance" // MCP 治理变更(批次一百八十七:包声明激活位/凭证绑定覆盖,热生效)
	EventMCPPoolError    = "mcp_pool_error" // MCP 池建连/列举失败留痕(此前完全静默,agent 拿空工具表无从排查)

	// 凭证平面审计事件(spec 2026-09-11 §9.1):CRUD/启停/测试为操作留痕;
	// staged 为 MCP spawn 装载留痕(只记名字与键名,不记值)。
	EventSkillUpdated = "skill_updated" // 技能编辑留痕(mode/正文写回;批次十补审计缺口)

	EventCredentialUsed    = "credential_used" // 会话级凭证使用归因(host 命中凭证实例名时随会话发布)
	EventCredentialCreated = "credential_created"
	EventCredentialUpdated = "credential_updated"
	EventCredentialDeleted = "credential_deleted"
	EventCredentialToggled = "credential_toggled"
	EventCredentialTested  = "credential_tested"
	EventCredentialOAuth   = "credential_oauth" // OAuth2 授权流留痕(批次一百九十七 R8:authorized|failed|refreshed|refresh_failed)
	EventCredentialsStaged = "credentials_staged"

	// IM 通道审计事件(批次九十三 DB 化):通道 CRUD/启停/测试留痕(bq_event,
	// 经 /api/v1/audit?type=im_channel_ 前缀过滤可见;credentials 同款纪律)。
	EventImChannelCreated = "im_channel_created"
	EventImChannelUpdated = "im_channel_updated"
	EventImChannelDeleted = "im_channel_deleted"
	EventImChannelToggled = "im_channel_toggled"
	EventImChannelTested  = "im_channel_tested"

	// 自动化审计事件(spec 2026-09-11-automation):规则 CRUD/启停/触发留痕(bq_event,
	// 经 /api/v1/audit?type=automation_ 前缀过滤可见)。
	EventAutomationCreated   = "automation_created"
	EventAutomationUpdated   = "automation_updated"
	EventAutomationDeleted   = "automation_deleted"
	EventAutomationToggled   = "automation_toggled"
	EventAutomationTriggered = "automation_triggered"
	EventAutomationFired     = "automation_fired"     // 排期拍到点实跑(与 TriggerNow 的 manual 区分)
	EventAutomationSuggested = "automation_suggested" // FR-11:巡检成功后建议定时化(进会话时间线,前端渲染建议卡)

	// 链模板审计事件(spec 2026-09-18-chain-v2-workflow §10):模板 CRUD/启停/触发留痕
	//(bq_event,经 /api/v1/audit?type=chain_ 前缀过滤可见);节点级事件在 scheduler 侧。
	EventChainCreated        = "chain_created"
	EventChainUpdated        = "chain_updated"
	EventChainDeleted        = "chain_deleted"
	EventChainToggled        = "chain_toggled"
	EventChainTriggered      = "chain_triggered"
	EventChainCondition      = "chain_condition"       // 链内 condition 节点判定留痕(field/op/value/命中分支)
	EventChainGateRequested  = "chain_gate_requested"  // 链内 approval 节点提请人工闸门
	EventChainNotifySkipped  = "notify_skipped"        // notify 节点降级(未配置/失败,观测有损不阻塞)
	EventChainExecuteBlocked = "chain_execute_blocked" // execute 节点被拦(平台总闸关/黑名单/主机缺失;降级不阻塞链)
	EventChainParallel       = "chain_parallel"        // 并行子图起跑标记(branches 名单随 payload)
	EventChainNode           = "chain_node"            // 链节点执行留痕(path/节点类型/状态;path 与控制台画布节点 id 同构,运行时拓扑据此高亮)

	// 操作审计门事件(spec 2026-09-14-policy-modes-audit-gate-design):方案提审裁决与
	// 工具调用面逐次审计留痕——verdict/decider/rule_id 随 payload,审计可答「谁裁的、依据什么」。
	EventOperationAudited = "operation_audited"

	// 告警驱动面事件(spec 2026-09-14-aiops-uplift-design §2):外部告警接入即留痕
	// (fingerprint/severity/deduped/merged 随 payload),会话时间线渲染「来源告警」卡。
	EventAlertReceived = "alert_received"

	// 告警路由规则面事件(批次一百一十一 DB 化管理面):CRUD/启停审计留痕。
	EventAlertRouteCreated = "alert_route_created"
	EventAlertRouteUpdated = "alert_route_updated"
	EventAlertRouteToggled = "alert_route_toggled"
	EventAlertRouteDeleted = "alert_route_deleted"

	// 工具身份守卫事件(spec 2026-09-15-tool-identity-otel-guard-design P4):
	// violation 为热路径守卫实锤(requested/resolved/checksum/violation 随 payload);
	// pairing_leak 为被动 watcher 配对不变量告警(reason=declared_not_executed|duplicate_result)。
	EventToolIdentityViolation = "tool_identity_violation"
	EventToolPairingLeak       = "tool_pairing_leak"
	// 身份违规 IM 推送已处理标记(identity-watch 通知扫描兜底的幂等锚点;payload {"ref":<源事件ID>})。
	EventIdentityViolationNotified = "identity_violation_notified"
)

事件类型词表(bq_event.type 同源)。

Variables

This section is empty.

Functions

This section is empty.

Types

type Bus

type Bus struct {
	// contains filtered or unexported fields
}

Bus 进程内事件总线(单实例;cluster 多副本 Phase③ 换 NATS 扇出)。

func New

func New() *Bus

New 构造(序号从 0 起;仅测试或全新库使用)。

func (*Bus) Broadcast

func (b *Bus) Broadcast(e Event)

Broadcast 广播一个已构建事件(发布路径先 repo.InsertEvent 落库拿 bq_event 自增 ID,再由此广播)。 SSE after 游标从此只有 bq_event ID 一个单调空间——跨重启、跨副本(PG 序列全局唯一)都成立, 取代旧的 NewWithSeq 启动播种方案(bus 内部序号与回放 ID 是两个空间,多副本下必然错位)。

func (*Bus) Dropped

func (b *Bus) Dropped() int64

Dropped 慢订阅丢弃计数(可观测)。

func (*Bus) Publish

func (b *Bus) Publish(sessionID, caseID string, stepID *uint64, typ, level string, payload map[string]any) Event

Publish 赋 ID/时间戳后广播(未落库的临时序号;生产路径改用 InsertEvent+Broadcast)。

func (*Bus) Subscribe

func (b *Bus) Subscribe(ctx context.Context) (<-chan Event, func())

Subscribe 订阅事件流;cancel 释放订阅(SSE 断开/bq_event writer 停止时调用)。

契约:退订(cancel 或 ctx 取消)后 channel 会被**关闭**

消费方必须写 `v, ok := <-ch` 并处理 ok==false。这不是可选风格:关闭的 channel 在 select 里**恒可就绪**并反复给出零值,于是 `case v := <-ch:` 不查 ok 的循环 会空转——CPU 打满、事件是假的、而且不报错。

本包曾按上游旧契约(runtime<v0.10.0 的「channel 永不关闭」,oagentkit/runtime#20) 承诺「不关」,并以此为由禁止消费方 range。上游 v0.10.0 推翻了那条理由 (runtime#32):Publish 与 unsubscribe 同锁互斥,不存在 send-on-closed 窗口, 故退订即 close。本仓随之上车——关闭换来的是 `for range ch` 能正常退出、 且不会把已缓冲事件挂在无人消费的 channel 上(旧形态每退订一次就漏一条 channel)。

注意 Hub 的 Subscribe **不**关 channel(见 hub.go 的说明):两个投递面在这里 有意分叉,判据由 hub_test.go 与 eventbus_test.go 各钉一侧。

type Event

type Event struct {
	ID         uint64         `json:"id"`
	SessionID  string         `json:"session_id"`
	CaseID     string         `json:"case_id,omitempty"`
	StepID     *uint64        `json:"step_id,omitempty"`
	Type       string         `json:"type"`
	Level      string         `json:"level"` // info|warn|error
	Payload    map[string]any `json:"payload,omitempty"`
	OccurredAt time.Time      `json:"occurred_at"`
}

Event 引擎事件(比 model.Event 多运行时字段;落库时转换)。

type Hub

type Hub struct {
	// contains filtered or unexported fields
}

Hub 按会话键控的事件分发器(SSE 单路桥接,2026-10-03 会话平面审计): 订阅时绑定 session_id,发布只投递该会话的订阅者——SSE 连接不再为无关会话的事件 做逐条过滤(旧形态每连接收全实例事件再自行丢弃,成本 = 连接数 × 全局事件数), 分发成本随活跃会话数而非连接总数增长。

投递语义的**缓冲面**与 progress.Bus 一致:订阅 channel 带缓冲,满即丢帧计数 (不阻塞发布方——编排热路径不可被慢连接拖住);丢帧由 SSE 的 DB 回放游标 + 全局 tail 兜底补达,不丢语义。零订阅者的发布成本 = 一次 map 读 + 空 slice 遍历。

与 eventbus.Bus 的**有意分叉**:退订时不关闭 channel

上游 runtime v0.10.0 把 progress.Bus 改成了「退订即 close」(runtime#32), eventbus.Bus 随之关闭;Hub **不**跟:

  • Hub 的消费方是宿主的 SSE 循环,内核看不到那些循环有没有写 `v, ok := <-ch`;
  • 关闭一个「按旧契约写」的 channel,后果是循环**空转**——CPU 打满、拿到的 全是零值事件、且一声不响。这是个失败形态**变绿**的改动,不该在看不见 消费方的那一层单方面做。

所以关闭的时机留给能看见宿主代码的那一层决定;本包把「不关」写成声明而不是 遗漏,由 TestHubSubscribeDoesNotCloseChannel 钉住。

消费方约定:用 `select` + 自己的 done 信号退出(**禁用 `for range ch`**, 它在本包永不返回)。

func NewHub

func NewHub() *Hub

NewHub 构造。

func (*Hub) Dropped

func (h *Hub) Dropped() int64

Dropped 慢订阅丢帧累计(可观测)。

func (*Hub) Publish

func (h *Hub) Publish(ev Event)

Publish 向会话订阅者投递(非阻塞;满缓冲丢帧计数)。SessionID 为空的系统级事件 (expert_toggled 等)无会话受众,直接忽略——系统事件经 /audit 面,不走 SSE。

func (*Hub) Subscribe

func (h *Hub) Subscribe(sessionID string) (<-chan Event, func())

Subscribe 订阅指定会话的事件流(返回取消订阅函数,幂等安全)。

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL