refactor(edge): 拆解事件总线与结果聚合域，下沉权限注册表 - #1668
Conversation
三个零行为变化的架构拆解，降低大文件职责耦合、理顺依赖方向： 1. events/bus.go（994 行）按域拆为 types.go / eventlog.go / persist.go / bus.go：事件日志磁盘引擎、持久化策略与总线主体各自独立，纯文件切分。 2. PermissionRegistry 从 internal/api 下沉到 internal/permission，解除 mcp → api 的反向依赖（协议层不再依赖 HTTP 层）；api 与 mcp 各自注入 同一注册表。新增 NewPermissionRegistryWithClock 测试时钟注入点。 3. lifecycle/result_aggregator.go（505 行）拆出纯内存状态机 subagent_collector.go，ResultAggregator 只保留事件总线粘合。collector 增加可注入时钟并补 4 个直接单测，填补原先只有间接覆盖的空白。 验证：go build/vet/staticcheck 全绿；go test ./... -short -race 通过； verify-orchestrator-deps、verify-test-sleep-ratchet、git diff --check 通过； lifecycle 覆盖率 93.1%（CI 最低 60%）。 Co-authored-by: Cursor <cursor@vectorcontrol.tech>
📝 WalkthroughWalkthroughThe change moves permission types into ChangesPermission package migration
Event persistence and replay
Sub-agent result collection
Estimated code review effort: 4 (Complex) | ~45 minutes Mergeability Score: 🟠 High · up to This refactor can duplicate sub-agent completion events, fail to complete runs when spawned results are missing, and silently discard replayable event data during log truncation. It also has smaller persistence, observability, and injected-clock safety issues, so the PR is not safe to merge until the major correctness risks are fixed or explicitly accepted. Sequence Diagram(s)sequenceDiagram
participant Bus
participant PersistFn
participant EventLog
participant Subscriber
Bus->>PersistFn: Persist EventEnvelope
PersistFn->>EventLog: Append event
EventLog-->>PersistFn: Return append result
PersistFn-->>Bus: Return persistence result
Subscriber->>EventLog: ReadFrom cursor
EventLog-->>Subscriber: Return sorted events and gap status
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches 💡 1📝 Generate docstrings 💡
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 6
🧹 Nitpick comments (2)
edge-server/internal/events/eventlog.go (2)
139-146: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueReplace
indexBytewithbytes.IndexByte.The standard library provides the same function with an optimized implementation.
♻️ Suggested change
-// indexByte returns the index of the first occurrence of b in s, or -1. -func indexByte(s []byte, b byte) int { - for i, c := range s { - if c == b { - return i - } - } - return -1 -}Then import
bytesand callbytes.IndexByte(raw, '\n')inrebuildIndexLockedandReadFrom.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@edge-server/internal/events/eventlog.go` around lines 139 - 146, Remove the custom indexByte helper and import the standard bytes package. Update rebuildIndexLocked and ReadFrom to use bytes.IndexByte for newline searches, preserving the existing behavior.
152-189: 🗄️ Data Integrity & Integration | 🔵 Trivial | ⚡ Quick winAppend does not fsync, so the durability claim is weaker than documented.
The package documents persist-before-broadcast for crash recovery.
Writeonly reaches the OS page cache. A host crash or power loss can lose the last events even though subscribers already received them.Add an optional
Sync()after the write so operators can choose durability over throughput.♻️ Suggested change
_, err = l.f.Write(data) if err == nil { + if l.syncOnAppend { + if syncErr := l.f.Sync(); syncErr != nil { + slog.Warn("event log fsync failed", "path", l.path, "error", syncErr) + } + } // Extend the live index so the just-appended event is immediately🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@edge-server/internal/events/eventlog.go` around lines 152 - 189, Update EventLog.Append to support optional durability by invoking l.f.Sync() after a successful write when the configured durability option is enabled; return any sync error and only update the live index or perform truncation after both write and sync succeed. Add or reuse the package’s existing configuration symbol for selecting this behavior, preserving the current throughput-oriented default.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@edge-server/internal/events/eventlog.go`:
- Around line 307-312: In the rebuildIndexLocked failure branch within the
event-log read flow, increment l.gaps before returning the gap result so the
metric matches the gap signal sent to subscribers. Preserve the existing cursor
and l.orderedSeq-based gap calculation and failure logging.
- Around line 220-227: Update truncateLocked to preserve the entire retained
tail when Read returns a short count: continue reading until keepBytes is filled
or a terminal read condition occurs before rewriting the buffer. Replace the
readErr.Error() text comparison with errors.Is checks for io.EOF and
io.ErrUnexpectedEOF, adding the required standard-library imports and retaining
failure handling for other errors.
In `@edge-server/internal/events/persist.go`:
- Around line 107-120: Prevent empty event-log paths from being treated as
successful persistence: in edge-server/internal/events/persist.go:107-120,
update WithEventLogPath to detect a nil log, emit a warning that persistence
remains disabled, and return before assigning eventLog or persistFn; in
edge-server/internal/events/eventlog.go:55-58, make NewEventLog return an
explicit error for an empty path instead of (nil, nil).
In `@edge-server/internal/lifecycle/subagent_collector.go`:
- Around line 122-135: Replace the separate IsExhausted/Exhaust check-and-set in
emitAggregatedResult with a new locked TryExhaust(parentID string) bool method
on SubAgentResultCollector that returns false when already exhausted and
otherwise marks the parent exhausted and returns true. Update
emitAggregatedResult to return immediately when TryExhaust returns false,
ensuring only the successful caller publishes run.agent.sub_agents_complete.
- Around line 107-113: Update SubAgentResultCollector.RecordSpawn and the
associated per-parent state to count every spawned child, not only the first
spawn timestamp. Have Aggregate derive TotalChildren and timeout Pending from
that spawn count, including parents with no stored results, while preserving
existing timestamp behavior. Add coverage for one missing result and multiple
spawns with zero stored results so timeout completion is emitted.
In `@edge-server/internal/permission/permission.go`:
- Around line 53-60: Update NewPermissionRegistryWithClock to handle a nil now
function at construction by defaulting it to time.Now (or explicitly rejecting
nil), ensuring registry operations never invoke a nil clock. Add a test covering
the selected nil-clock contract.
---
Nitpick comments:
In `@edge-server/internal/events/eventlog.go`:
- Around line 139-146: Remove the custom indexByte helper and import the
standard bytes package. Update rebuildIndexLocked and ReadFrom to use
bytes.IndexByte for newline searches, preserving the existing behavior.
- Around line 152-189: Update EventLog.Append to support optional durability by
invoking l.f.Sync() after a successful write when the configured durability
option is enabled; return any sync error and only update the live index or
perform truncation after both write and sync succeed. Add or reuse the package’s
existing configuration symbol for selecting this behavior, preserving the
current throughput-oriented default.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: cf30a031-cd20-4064-9714-f05d6a68be6b
📒 Files selected for processing (14)
edge-server/internal/api/handlers.goedge-server/internal/api/handlers_approvals.goedge-server/internal/api/handlers_test.goedge-server/internal/events/bus.goedge-server/internal/events/eventlog.goedge-server/internal/events/persist.goedge-server/internal/events/types.goedge-server/internal/lifecycle/result_aggregator.goedge-server/internal/lifecycle/subagent_collector.goedge-server/internal/lifecycle/subagent_collector_test.goedge-server/internal/mcp/server.goedge-server/internal/mcp/server_test.goedge-server/internal/permission/permission.goedge-server/internal/permission/permission_test.go
💤 Files with no reviewable changes (2)
- edge-server/internal/lifecycle/result_aggregator.go
- edge-server/internal/events/bus.go
| buf := make([]byte, keepBytes) | ||
| n, readErr := l.f.Read(buf) | ||
| if readErr != nil && readErr.Error() != "EOF" { | ||
| l.truncateFailures.Add(1) | ||
| slog.Error("event log truncate read failed", | ||
| "path", l.path, "keepBytes", keepBytes, "error", readErr) | ||
| return | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Fix the short-read and the error-text comparison in truncateLocked.
os.File.Read can return fewer bytes than len(buf) without an error. The code then rewrites only buf[start:n] and discards the remaining retained tail. Every truncation can silently drop replayable events.
readErr.Error() != "EOF" also depends on the error message text. Use errors.Is with io.EOF and io.ErrUnexpectedEOF.
🐛 Proposed fix
buf := make([]byte, keepBytes)
- n, readErr := l.f.Read(buf)
- if readErr != nil && readErr.Error() != "EOF" {
+ n, readErr := io.ReadFull(l.f, buf)
+ if readErr != nil && !errors.Is(readErr, io.EOF) && !errors.Is(readErr, io.ErrUnexpectedEOF) {
l.truncateFailures.Add(1)
slog.Error("event log truncate read failed",
"path", l.path, "keepBytes", keepBytes, "error", readErr)
return
}Add the import:
import (
+ "errors"
"encoding/json"📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| buf := make([]byte, keepBytes) | |
| n, readErr := l.f.Read(buf) | |
| if readErr != nil && readErr.Error() != "EOF" { | |
| l.truncateFailures.Add(1) | |
| slog.Error("event log truncate read failed", | |
| "path", l.path, "keepBytes", keepBytes, "error", readErr) | |
| return | |
| } | |
| buf := make([]byte, keepBytes) | |
| n, readErr := io.ReadFull(l.f, buf) | |
| if readErr != nil && !errors.Is(readErr, io.EOF) && !errors.Is(readErr, io.ErrUnexpectedEOF) { | |
| l.truncateFailures.Add(1) | |
| slog.Error("event log truncate read failed", | |
| "path", l.path, "keepBytes", keepBytes, "error", readErr) | |
| return | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@edge-server/internal/events/eventlog.go` around lines 220 - 227, Update
truncateLocked to preserve the entire retained tail when Read returns a short
count: continue reading until keepBytes is filled or a terminal read condition
occurs before rewriting the buffer. Replace the readErr.Error() text comparison
with errors.Is checks for io.EOF and io.ErrUnexpectedEOF, adding the required
standard-library imports and retaining failure handling for other errors.
| if fi, statErr := l.f.Stat(); statErr == nil && fi.Size() != l.indexedSize { | ||
| if err := l.rebuildIndexLocked(); err != nil { | ||
| slog.Warn("event log index rebuild on size change failed", "path", l.path, "error", err) | ||
| return nil, cursor > 0 && len(l.orderedSeq) > 0 && cursor < l.orderedSeq[0] | ||
| } | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
The rebuild-failure path reports a gap without counting it.
If rebuildIndexLocked fails, line 310 computes hasGap from the stale l.orderedSeq and returns without l.gaps.Add(1). The subscriber then receives a gap signal that edge_event_log_gaps_total never records. Increment gaps on that branch so the metric matches the signal sent to subscribers.
🐛 Proposed fix
if err := l.rebuildIndexLocked(); err != nil {
slog.Warn("event log index rebuild on size change failed", "path", l.path, "error", err)
- return nil, cursor > 0 && len(l.orderedSeq) > 0 && cursor < l.orderedSeq[0]
+ staleGap := cursor > 0 && (len(l.orderedSeq) == 0 || cursor < l.orderedSeq[0])
+ if staleGap {
+ l.gaps.Add(1)
+ }
+ return nil, staleGap
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| if fi, statErr := l.f.Stat(); statErr == nil && fi.Size() != l.indexedSize { | |
| if err := l.rebuildIndexLocked(); err != nil { | |
| slog.Warn("event log index rebuild on size change failed", "path", l.path, "error", err) | |
| return nil, cursor > 0 && len(l.orderedSeq) > 0 && cursor < l.orderedSeq[0] | |
| } | |
| } | |
| if fi, statErr := l.f.Stat(); statErr == nil && fi.Size() != l.indexedSize { | |
| if err := l.rebuildIndexLocked(); err != nil { | |
| slog.Warn("event log index rebuild on size change failed", "path", l.path, "error", err) | |
| staleGap := cursor > 0 && (len(l.orderedSeq) == 0 || cursor < l.orderedSeq[0]) | |
| if staleGap { | |
| l.gaps.Add(1) | |
| } | |
| return nil, staleGap | |
| } | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@edge-server/internal/events/eventlog.go` around lines 307 - 312, In the
rebuildIndexLocked failure branch within the event-log read flow, increment
l.gaps before returning the gap result so the metric matches the gap signal sent
to subscribers. Preserve the existing cursor and l.orderedSeq-based gap
calculation and failure logging.
| func WithEventLogPath(path string) BusOption { | ||
| return func(b *Bus) { | ||
| log, err := NewEventLog(path) | ||
| if err != nil { | ||
| slog.Error("failed to open event log, events will not be persisted to disk", | ||
| "path", path, "error", err) | ||
| return | ||
| } | ||
| b.eventLog = log | ||
| b.persistFn = func(evt EventEnvelope) error { | ||
| return log.Append(evt) | ||
| } | ||
| } | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
An empty event-log path yields a nil EventLog that is installed as a working persister. NewEventLog returns (nil, nil) when path is empty, so WithEventLogPath sees err == nil, sets b.eventLog = nil, and installs a persistFn whose nil-receiver Append always returns nil. Publish then reports durable persistence while nothing reaches disk, and no log line records the misconfiguration.
edge-server/internal/events/persist.go#L107-L120: return early whenlog == nil, and log a warning that persistence stays disabled, sopersistFnis not replaced by a no-op.edge-server/internal/events/eventlog.go#L55-L58: return an explicit error for an emptypathinstead of(nil, nil), so callers cannot mistake the empty case for a successful open.
🐛 Proposed fix in persist.go
log, err := NewEventLog(path)
if err != nil {
slog.Error("failed to open event log, events will not be persisted to disk",
"path", path, "error", err)
return
}
+ if log == nil {
+ slog.Warn("empty event log path, disk persistence disabled", "path", path)
+ return
+ }
b.eventLog = log📍 Affects 2 files
edge-server/internal/events/persist.go#L107-L120(this comment)edge-server/internal/events/eventlog.go#L55-L58
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@edge-server/internal/events/persist.go` around lines 107 - 120, Prevent empty
event-log paths from being treated as successful persistence: in
edge-server/internal/events/persist.go:107-120, update WithEventLogPath to
detect a nil log, emit a warning that persistence remains disabled, and return
before assigning eventLog or persistFn; in
edge-server/internal/events/eventlog.go:55-58, make NewEventLog return an
explicit error for an empty path instead of (nil, nil).
| func (c *SubAgentResultCollector) RecordSpawn(parentID string) { | ||
| c.mu.Lock() | ||
| defer c.mu.Unlock() | ||
| if _, ok := c.firstSpawn[parentID]; !ok { | ||
| c.firstSpawn[parentID] = c.now() | ||
| } | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Track each spawned child before building a partial aggregate.
RecordSpawn stores only the first timestamp. It discards the number of children spawned for parentID.
Aggregate sets TotalChildren from len(stored). It cannot count children that never store a result. A partial aggregate can therefore report one completed child as one total child with zero pending children, even when another spawned child timed out.
If no child stores a result, Aggregate returns zero total children. edge-server/internal/lifecycle/result_aggregator.go Line 247 through Line 259 then skips the timeout completion event indefinitely.
Track a per-parent spawn count. Derive TotalChildren and timeout Pending from that count. Add tests for one missing result and zero stored results after multiple spawns.
Based on the provided downstream flow, edge-server/internal/lifecycle/result_aggregator.go Line 247 through Line 259 suppresses zero-result timeout aggregates.
Also applies to: 176-201
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@edge-server/internal/lifecycle/subagent_collector.go` around lines 107 - 113,
Update SubAgentResultCollector.RecordSpawn and the associated per-parent state
to count every spawned child, not only the first spawn timestamp. Have Aggregate
derive TotalChildren and timeout Pending from that spawn count, including
parents with no stored results, while preserving existing timestamp behavior.
Add coverage for one missing result and multiple spawns with zero stored results
so timeout completion is emitted.
| // Exhaust marks a parent as exhausted (results fully emitted). This prevents | ||
| // the timeout fallback from re-emitting for this parent. | ||
| func (c *SubAgentResultCollector) Exhaust(parentID string) { | ||
| c.mu.Lock() | ||
| defer c.mu.Unlock() | ||
| c.exhausted[parentID] = true | ||
| } | ||
|
|
||
| // IsExhausted returns true if the parent's results have already been emitted. | ||
| func (c *SubAgentResultCollector) IsExhausted(parentID string) bool { | ||
| c.mu.RLock() | ||
| defer c.mu.RUnlock() | ||
| return c.exhausted[parentID] | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Make exhaustion check-and-set atomic.
IsExhausted and Exhaust use separate lock operations. Two concurrent callers can both observe false, both mark the parent exhausted, and both publish run.agent.sub_agents_complete.
Add a locked TryExhaust(parentID) bool method. Update emitAggregatedResult to return when TryExhaust returns false.
Based on the provided downstream flow, edge-server/internal/lifecycle/result_aggregator.go Line 196 through Line 203 performs the non-atomic check-and-set sequence.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@edge-server/internal/lifecycle/subagent_collector.go` around lines 122 - 135,
Replace the separate IsExhausted/Exhaust check-and-set in emitAggregatedResult
with a new locked TryExhaust(parentID string) bool method on
SubAgentResultCollector that returns false when already exhausted and otherwise
marks the parent exhausted and returns true. Update emitAggregatedResult to
return immediately when TryExhaust returns false, ensuring only the successful
caller publishes run.agent.sub_agents_complete.
| // NewPermissionRegistryWithClock is NewPermissionRegistry with an explicit | ||
| // clock, used by deterministic expiry tests and future time-source injection. | ||
| func NewPermissionRegistryWithClock(ttl time.Duration, now func() time.Time) *PermissionRegistry { | ||
| registry := NewPermissionRegistry(ttl) | ||
| registry.now = now | ||
| return registry | ||
| } | ||
|
|
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
Handle a nil clock at construction.
If now is nil, the constructor stores a nil function. A later registry operation invokes it and panics. Default to time.Now or reject nil immediately, and add a test for the chosen contract.
Proposed defensive default
func NewPermissionRegistryWithClock(ttl time.Duration, now func() time.Time) *PermissionRegistry {
+ if now == nil {
+ now = time.Now
+ }
registry := NewPermissionRegistry(ttl)
registry.now = now
return registry
}The failure follows from the supplied Register implementation invoking the injected clock during registry operations.
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| // NewPermissionRegistryWithClock is NewPermissionRegistry with an explicit | |
| // clock, used by deterministic expiry tests and future time-source injection. | |
| func NewPermissionRegistryWithClock(ttl time.Duration, now func() time.Time) *PermissionRegistry { | |
| registry := NewPermissionRegistry(ttl) | |
| registry.now = now | |
| return registry | |
| } | |
| // NewPermissionRegistryWithClock is NewPermissionRegistry with an explicit | |
| // clock, used by deterministic expiry tests and future time-source injection. | |
| func NewPermissionRegistryWithClock(ttl time.Duration, now func() time.Time) *PermissionRegistry { | |
| if now == nil { | |
| now = time.Now | |
| } | |
| registry := NewPermissionRegistry(ttl) | |
| registry.now = now | |
| return registry | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@edge-server/internal/permission/permission.go` around lines 53 - 60, Update
NewPermissionRegistryWithClock to handle a nil now function at construction by
defaulting it to time.Now (or explicitly rejecting nil), ensuring registry
operations never invoke a nil clock. Add a test covering the selected nil-clock
contract.
Summary
edge-server 内部架构拆解：三个零行为变化的分解，降低大文件职责耦合、理顺依赖方向。
types.go（信封/订阅者/常量）、eventlog.go（磁盘事件日志引擎：索引/截断/回放）、persist.go（持久化钩子/重试策略）、bus.go（总线主体）。纯文件切分，无行为变化。internal/permission：解除mcp → api反向依赖（协议层不再依赖 HTTP handler 层）；api 与 mcp 各自注入同一注册表。新增NewPermissionRegistryWithClock时钟注入点（仓库已有 WithPersistMaxRetries 同类 test seam 先例）。SubAgentResultCollector（含结果类型、聚合、超时判定）独立为subagent_collector.go，ResultAggregator只保留事件订阅与完成判定。collector 增加可注入时钟（now func() time.Time），补 4 个直接单测（此前该状态机只有间接覆盖，见架构探查）。架构依据
源自对 edge-server 全包依赖扫描的结构化分析：
events/bus.go是最大单体（994 行六种职责）；mcp→api是唯一协议层→HTTP 层反向依赖；SubAgentResultCollector是天然可测的纯状态机却无直接测试。Test plan
go build ./.../go vet ./.../ staticcheck 全绿go test ./... -count=1 -short -race -coverprofile=coverage.out -covermode=atomic通过（CI 同款命令）verify-orchestrator-deps.py、verify-test-sleep-ratchet.py通过；git diff --check干净后续队列（另开 PR）
Summary by CodeRabbit
New Features
Bug Fixes