From cdfc216faee449de058be0400cd2f2f3fc20589d Mon Sep 17 00:00:00 2001
From: Delicious233 <101502465+DeliciousBuding@users.noreply.github.com>
Date: Thu, 13 Aug 2026 07:57:49 +0800
Subject: [PATCH] =?UTF-8?q?refactor(edge):=20=E6=8B=86=E8=A7=A3=E4=BA=8B?=
 =?UTF-8?q?=E4=BB=B6=E6=80=BB=E7=BA=BF=E4=B8=8E=E7=BB=93=E6=9E=9C=E8=81=9A?=
 =?UTF-8?q?=E5=90=88=E5=9F=9F=EF=BC=8C=E4=B8=8B=E6=B2=89=E6=9D=83=E9=99=90?=
 =?UTF-8?q?=E6=B3=A8=E5=86=8C=E8=A1=A8?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

三个零行为变化的架构拆解，降低大文件职责耦合、理顺依赖方向：

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>
---
 edge-server/internal/api/handlers.go          |   7 +-
 .../internal/api/handlers_approvals.go        |   7 +-
 edge-server/internal/api/handlers_test.go     |  12 +-
 edge-server/internal/events/bus.go            | 551 ------------------
 edge-server/internal/events/eventlog.go       | 396 +++++++++++++
 edge-server/internal/events/persist.go        | 120 ++++
 edge-server/internal/events/types.go          |  52 ++
 .../internal/lifecycle/result_aggregator.go   | 220 -------
 .../internal/lifecycle/subagent_collector.go  | 231 ++++++++
 .../lifecycle/subagent_collector_test.go      |  99 ++++
 edge-server/internal/mcp/server.go            |   6 +-
 edge-server/internal/mcp/server_test.go       |   6 +-
 .../permission.go}                            |  13 +-
 .../permission_test.go}                       |   2 +-
 14 files changed, 931 insertions(+), 791 deletions(-)
 create mode 100644 edge-server/internal/events/eventlog.go
 create mode 100644 edge-server/internal/events/persist.go
 create mode 100644 edge-server/internal/events/types.go
 create mode 100644 edge-server/internal/lifecycle/subagent_collector.go
 create mode 100644 edge-server/internal/lifecycle/subagent_collector_test.go
 rename edge-server/internal/{api/permission_registry.go => permission/permission.go} (85%)
 rename edge-server/internal/{api/permission_registry_test.go => permission/permission_test.go} (99%)

diff --git a/edge-server/internal/api/handlers.go b/edge-server/internal/api/handlers.go
index ca4929c58..3bbafacfa 100644
--- a/edge-server/internal/api/handlers.go
+++ b/edge-server/internal/api/handlers.go
@@ -19,6 +19,7 @@ import (
 	"github.com/agenthub/edge-server/internal/hub"
 	"github.com/agenthub/edge-server/internal/lifecycle"
 	"github.com/agenthub/edge-server/internal/metrics"
+	"github.com/agenthub/edge-server/internal/permission"
 	"github.com/agenthub/edge-server/internal/runners"
 	"github.com/agenthub/edge-server/internal/security"
 	"github.com/agenthub/edge-server/internal/skills"
@@ -55,7 +56,7 @@ type Handler struct {
 		DurableSnapshot(afterSeq uint64) ([]hub.DeliveryJournalEntry, error)
 	}
 
-	PermissionRegistry *PermissionRegistry
+	PermissionRegistry *permission.PermissionRegistry
 	PermissionBroker   *adapters.PermissionDecisionBroker
 
 	// PlanApprovalBroker manages pending orchestrator plans and connects
@@ -183,11 +184,11 @@ func ensurePreviewRunner(h *Handler, repository store.Repository) lifecycle.Prev
 	return h.PreviewRunner
 }
 
-func (h *Handler) ensurePermissionRegistry() *PermissionRegistry {
+func (h *Handler) ensurePermissionRegistry() *permission.PermissionRegistry {
 	h.permissionRegistryMu.Lock()
 	defer h.permissionRegistryMu.Unlock()
 	if h.PermissionRegistry == nil {
-		h.PermissionRegistry = NewPermissionRegistry(0)
+		h.PermissionRegistry = permission.NewPermissionRegistry(0)
 	}
 	if h.PermissionBroker == nil {
 		h.PermissionBroker = adapters.NewPermissionDecisionBroker()
diff --git a/edge-server/internal/api/handlers_approvals.go b/edge-server/internal/api/handlers_approvals.go
index ed600daf7..0645efc87 100644
--- a/edge-server/internal/api/handlers_approvals.go
+++ b/edge-server/internal/api/handlers_approvals.go
@@ -7,6 +7,7 @@ import (
 
 	"github.com/agenthub/edge-server/internal/adapters"
 	"github.com/agenthub/edge-server/internal/errcode"
+	"github.com/agenthub/edge-server/internal/permission"
 )
 
 // Handler holds dependencies for HTTP and WebSocket handlers.
@@ -74,15 +75,15 @@ func (h *Handler) PostPermissionDecide(w http.ResponseWriter, r *http.Request) {
 	writeSuccess(w, http.StatusOK, map[string]any{"status": "ok"})
 }
 
-func pendingPermissionFromBroker(broker *adapters.PermissionDecisionBroker, runID, requestID, decision, reason string) (PendingPermission, bool) {
+func pendingPermissionFromBroker(broker *adapters.PermissionDecisionBroker, runID, requestID, decision, reason string) (permission.PendingPermission, bool) {
 	pending, ok := broker.Decide(runID, requestID, adapters.PermissionDecision{
 		Behavior: decision,
 		Message:  reason,
 	})
 	if !ok {
-		return PendingPermission{}, false
+		return permission.PendingPermission{}, false
 	}
-	return PendingPermission{
+	return permission.PendingPermission{
 		ProjectID: pending.ProjectID,
 		ThreadID:  pending.ThreadID,
 		RunID:     pending.RunID,
diff --git a/edge-server/internal/api/handlers_test.go b/edge-server/internal/api/handlers_test.go
index 5abd8d465..8f5969ffa 100644
--- a/edge-server/internal/api/handlers_test.go
+++ b/edge-server/internal/api/handlers_test.go
@@ -22,6 +22,7 @@ import (
 	"github.com/agenthub/edge-server/internal/hub"
 	"github.com/agenthub/edge-server/internal/jwtutil"
 	"github.com/agenthub/edge-server/internal/lifecycle"
+	"github.com/agenthub/edge-server/internal/permission"
 	"github.com/agenthub/edge-server/internal/runners"
 	"github.com/agenthub/edge-server/internal/store"
 	"github.com/golang-jwt/jwt/v5"
@@ -1985,7 +1986,7 @@ func TestPostPermissionDecideRejectsUnknownRequest(t *testing.T) {
 
 func TestPostPermissionDecideRejectsWrongRun(t *testing.T) {
 	h := newTestHandler()
-	h.ensurePermissionRegistry().Register(PendingPermission{
+	h.ensurePermissionRegistry().Register(permission.PendingPermission{
 		RunID:     "run_real",
 		RequestID: "req_1",
 	})
@@ -2005,7 +2006,7 @@ func TestPostPermissionDecideRejectsWrongRun(t *testing.T) {
 
 func TestPostPermissionDecideConsumesPendingRequestAndPublishesEvent(t *testing.T) {
 	h := newTestHandler()
-	h.ensurePermissionRegistry().Register(PendingPermission{
+	h.ensurePermissionRegistry().Register(permission.PendingPermission{
 		ProjectID: "proj_1",
 		ThreadID:  "thread_1",
 		RunID:     "run_1",
@@ -2165,7 +2166,7 @@ func TestRegisterRoutesInstallsPermissionBrokerOnClaudeAdapter(t *testing.T) {
 
 func TestPostPermissionDecideRejectsSecondDecision(t *testing.T) {
 	h := newTestHandler()
-	h.ensurePermissionRegistry().Register(PendingPermission{
+	h.ensurePermissionRegistry().Register(permission.PendingPermission{
 		RunID:     "run_1",
 		RequestID: "req_1",
 	})
@@ -2189,10 +2190,9 @@ func TestPostPermissionDecideRejectsSecondDecision(t *testing.T) {
 func TestPostPermissionDecideRejectsExpiredRequestWithoutPublishing(t *testing.T) {
 	h := newTestHandler()
 	now := time.Date(2026, 5, 29, 8, 0, 0, 0, time.UTC)
-	registry := NewPermissionRegistry(time.Minute)
-	registry.now = func() time.Time { return now }
+	registry := permission.NewPermissionRegistryWithClock(time.Minute, func() time.Time { return now })
 	h.PermissionRegistry = registry
-	h.PermissionRegistry.Register(PendingPermission{
+	h.PermissionRegistry.Register(permission.PendingPermission{
 		ProjectID: "proj_1",
 		ThreadID:  "thread_1",
 		RunID:     "run_1",
diff --git a/edge-server/internal/events/bus.go b/edge-server/internal/events/bus.go
index 3d177d90e..0dcf7c4c6 100644
--- a/edge-server/internal/events/bus.go
+++ b/edge-server/internal/events/bus.go
@@ -1,11 +1,8 @@
 package events
 
 import (
-	"encoding/json"
-	"io"
 	"log/slog"
 	"os"
-	"path/filepath"
 	"sort"
 	"strconv"
 	"sync"
@@ -13,340 +10,6 @@ import (
 	"time"
 )
 
-const (
-	defaultMaxHistory           = 10000
-	subscriberChannelBufferSize = 256
-	defaultWorkerCount          = 4
-	observerJobBufferSize       = 1024
-
-	// persistDefaultMaxRetries bounds how many times persistWithRetry retries a
-	// persistFn call that returned an error before declaring the event lost.
-	// 1 = one original attempt + one retry; 3 = original + three retries.
-	persistDefaultMaxRetries = 3
-	// persistRetryBaseDelay is the exponential backoff base between persist
-	// retry attempts. Kept short so the synchronous retry path does not
-	// stall Publish under normal transient failures.
-	persistRetryBaseDelay = 2 * time.Millisecond
-)
-
-// GapEventType is the event type for a gap-detection control message sent to a
-// subscriber when one or more events were dropped because the subscriber channel
-// was full. The payload is a *GapPayload.
-const GapEventType = "system.gap"
-
-// GapPayload describes a range of dropped events for a subscriber.
-type GapPayload struct {
-	FirstDroppedSeq int64 `json:"firstDroppedSeq"`
-	LastDroppedSeq  int64 `json:"lastDroppedSeq"`
-	DroppedCount    int64 `json:"droppedCount"`
-}
-
-// observerJob is a unit of work dispatched to the observer worker pool.
-type observerJob struct {
-	fn  func(EventEnvelope)
-	evt EventEnvelope
-}
-
-// EventEnvelope is the standard event wrapper for all WebSocket events.
-type EventEnvelope struct {
-	Version string         `json:"version"`
-	ID      string         `json:"id"`
-	Seq     int64          `json:"seq"`
-	Type    string         `json:"type"`
-	Scope   map[string]any `json:"scope"`
-	TraceID string         `json:"traceId"`
-	SentAt  string         `json:"sentAt"`
-	Payload any            `json:"payload"`
-}
-
-// subscriber receives events on its channel.
-type subscriber struct {
-	id          int64
-	ch          chan EventEnvelope
-	gapDetected bool  // true when events were dropped since last successful send
-	firstGapSeq int64 // seq of first dropped event in the gap
-	lastGapSeq  int64 // seq of last dropped event in the gap
-}
-
-type observer struct {
-	id int64
-	fn func(EventEnvelope)
-}
-
-// EventLog is an append-only JSON-lines event log backed by a file on disk.
-// Each event is serialised as a single JSON line. Writes are safe for
-// concurrent use.
-//
-// The log also maintains a seq→offset index so that a Bus restarting with an
-// empty in-memory history can still replay events to a cursor-bearing
-// subscriber (crash recovery / replay). The index is rebuilt on open and after
-// every truncation; Subscribe also detects external file-size changes and
-// rebuilds lazily so a truncate that happened between ticks stays safe.
-type EventLog struct {
-	mu          sync.Mutex
-	f           *os.File
-	path        string
-	maxSize     int64           // max file size in bytes before truncation; 0 = unlimited
-	index       map[int64]int64 // seq → byte offset of that line (exclusive of the seq→offset map)
-	orderedSeq  []int64         // sorted seqs parallel to index, for binary search
-	indexedSize int64           // file size at last index build, to detect external changes
-
-	// truncations counts truncateLocked invocations (a truncation was
-	// attempted because the log exceeded maxSize). Exposed via
-	// Bus.EventLogTruncations for the edge_event_log_truncations_total metric.
-	truncations atomic.Int64
-	// truncateFailures counts truncateLocked invocations that hit an error
-	// branch (seek/read/truncate/rewrite failure) and previously returned
-	// silently. Exposed via Bus.EventLogTruncateFailures for
-	// edge_event_log_truncate_failures_total so an operator whose log was
-	// growing unbounded or losing replay data finally gets a signal.
-	truncateFailures atomic.Int64
-	// gaps counts ReadFrom calls that detected a cursor predating the oldest
-	// surviving log event (replay would lose events). Exposed via
-	// Bus.EventLogGaps for the edge_event_log_gaps_total metric.
-	gaps atomic.Int64
-}
-
-const defaultEventLogMaxSize = 50 * 1024 * 1024 // 50 MiB
-
-// NewEventLog opens or creates the append-only event log at the given path.
-// The parent directory is created if it does not exist. The file is opened
-// read+write so the log can serve replay reads in addition to appends. The
-// seq→offset index is built by scanning existing lines so a restarted Bus can
-// replay history from disk before any new Publish lands.
-func NewEventLog(path string) (*EventLog, error) {
-	if path == "" {
-		return nil, nil
-	}
-	dir := filepath.Dir(path)
-	if err := os.MkdirAll(dir, 0o750); err != nil {
-		return nil, err
-	}
-	// #nosec G304 -- event log path comes from operator config (WithEventLogPath)
-	// O_RDWR so the log is readable for replay; O_APPEND so writes always go to
-	// the end regardless of the read seek position.
-	f, err := os.OpenFile(path, os.O_APPEND|os.O_CREATE|os.O_RDWR, 0o600)
-	if err != nil {
-		return nil, err
-	}
-	l := &EventLog{f: f, path: path, maxSize: defaultEventLogMaxSize, index: make(map[int64]int64)}
-	if err := l.rebuildIndexLocked(); err != nil {
-		// A corrupt or unreadable log is not fatal: the Bus keeps working with
-		// an empty index (replay falls back to in-memory history only). Log the
-		// error so operators can repair the file.
-		slog.Error("event log index build failed; replay will be incomplete until the file is repaired",
-			"path", path, "error", err)
-		l.index = make(map[int64]int64)
-		l.orderedSeq = nil
-	}
-	return l, nil
-}
-
-// rebuildIndexLocked scans the entire log file line by line, recording each
-// event's seq and its byte offset. Must be called with l.mu held.
-func (l *EventLog) rebuildIndexLocked() error {
-	if l == nil || l.f == nil {
-		return nil
-	}
-	if _, err := l.f.Seek(0, 0); err != nil {
-		return err
-	}
-	index := make(map[int64]int64)
-	var ordered []int64
-	// Track the byte offset of each line start. json.Decoder does not expose
-	// offsets, so we count bytes consumed by reading the raw file in a single
-	// pass and splitting on newlines.
-	raw, err := io.ReadAll(l.f)
-	if err != nil {
-		return err
-	}
-	offset := int64(0)
-	for len(raw) > 0 {
-		nl := indexByte(raw, '\n')
-		var line []byte
-		var lineLen int
-		if nl < 0 {
-			line = raw
-			lineLen = len(raw)
-			raw = nil
-		} else {
-			line = raw[:nl]
-			lineLen = nl + 1
-			raw = raw[lineLen:]
-		}
-		if len(line) == 0 {
-			offset += int64(lineLen)
-			continue
-		}
-		var env EventEnvelope
-		if json.Unmarshal(line, &env) == nil && env.Seq > 0 {
-			if _, exists := index[env.Seq]; !exists {
-				ordered = append(ordered, env.Seq)
-			}
-			index[env.Seq] = offset
-		}
-		offset += int64(lineLen)
-	}
-	l.index = index
-	l.orderedSeq = ordered
-	if fi, statErr := l.f.Stat(); statErr == nil {
-		l.indexedSize = fi.Size()
-	}
-	// Restore the write cursor to the end so the next Append writes at EOF.
-	_, _ = l.f.Seek(0, 2)
-	return nil
-}
-
-// 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
-}
-
-// Append writes an event to the log as a single JSON line followed by a newline.
-// After the write, the event's seq→offset is added to the index so replay can
-// find it without rescanning the file. When the file exceeds maxSize the log
-// is truncated and the index rebuilt.
-func (l *EventLog) Append(evt EventEnvelope) error {
-	if l == nil {
-		return nil
-	}
-	l.mu.Lock()
-	defer l.mu.Unlock()
-
-	// Record the write offset before writing so the index points at the line
-	// start. With O_APPEND the current seek position is irrelevant for writes,
-	// so we stat to get the current end-of-file size.
-	writeOffset := int64(0)
-	if fi, statErr := l.f.Stat(); statErr == nil {
-		writeOffset = fi.Size()
-	}
-
-	data, err := json.Marshal(evt)
-	if err != nil {
-		return err
-	}
-	data = append(data, '\n')
-	_, err = l.f.Write(data)
-	if err == nil {
-		// Extend the live index so the just-appended event is immediately
-		// replayable without a full rescan.
-		if evt.Seq > 0 {
-			if _, exists := l.index[evt.Seq]; !exists {
-				l.orderedSeq = appendSorted(l.orderedSeq, evt.Seq)
-			}
-			l.index[evt.Seq] = writeOffset
-		}
-		if l.maxSize > 0 {
-			if fi, statErr := l.f.Stat(); statErr == nil && fi.Size() > l.maxSize {
-				l.truncateLocked()
-			}
-		}
-	}
-	return err
-}
-
-// appendSorted inserts seq into the sorted slice maintaining order. Used by
-// Append to extend the ordered index without a full sort.
-func appendSorted(sorted []int64, seq int64) []int64 {
-	idx := sort.Search(len(sorted), func(i int) bool { return sorted[i] >= seq })
-	if idx < len(sorted) && sorted[idx] == seq {
-		return sorted // already present
-	}
-	sorted = append(sorted, 0)
-	copy(sorted[idx+1:], sorted[idx:])
-	sorted[idx] = seq
-	return sorted
-}
-
-// truncateLocked rewrites the log file keeping only the trailing portion.
-// Must be called with l.mu held. The index is rebuilt after the rewrite so
-// replay offsets stay accurate after truncation. Every failure branch that
-// previously returned silently now increments truncateFailures and emits a
-// slog.Error so an operator whose log is growing unbounded or losing replay
-// offsets finally gets a signal (edge_event_log_truncate_failures_total).
-func (l *EventLog) truncateLocked() {
-	l.truncations.Add(1)
-	keepBytes := l.maxSize * 3 / 4 // keep last 75%
-	if _, seekErr := l.f.Seek(-keepBytes, 2); seekErr != nil {
-		// File too small or seek failed; skip truncation but surface it.
-		l.truncateFailures.Add(1)
-		slog.Error("event log truncate seek failed",
-			"path", l.path, "keepBytes", keepBytes, "error", seekErr)
-		return
-	}
-	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
-	}
-	// Skip to next newline so we don't keep a partial line.
-	start := 0
-	for i := 0; i < n; i++ {
-		if buf[i] == '\n' {
-			start = i + 1
-			break
-		}
-	}
-	// Truncate and rewrite.
-	if truncErr := l.f.Truncate(0); truncErr != nil {
-		l.truncateFailures.Add(1)
-		slog.Error("event log truncate Truncate(0) failed",
-			"path", l.path, "error", truncErr)
-		return
-	}
-	if _, seekErr := l.f.Seek(0, 0); seekErr != nil {
-		l.truncateFailures.Add(1)
-		slog.Error("event log truncate seek-to-start failed",
-			"path", l.path, "error", seekErr)
-		return
-	}
-	if start < n {
-		// Best-effort rewrite: the log truncation path degrades silently on
-		// write failure rather than failing the enclosing Append call.
-		if written, writeErr := l.f.Write(buf[start:n]); writeErr != nil || written != n-start {
-			l.truncateFailures.Add(1)
-			slog.Error("event log truncate rewrite failed",
-				"path", l.path, "written", written, "want", n-start, "error", writeErr)
-		}
-	}
-	// Rebuild the index so replay offsets reflect the truncated file. A
-	// rebuild failure also counts as a truncate failure (the log is now in a
-	// partially-rewritten state and replay offsets are unreliable).
-	if err := l.rebuildIndexLocked(); err != nil {
-		l.truncateFailures.Add(1)
-		slog.Error("event log truncate index rebuild failed",
-			"path", l.path, "error", err)
-	}
-}
-
-// EventLogTruncations returns the total number of truncateLocked invocations
-// (truncations attempted because the log exceeded maxSize). Exposed for the
-// edge_event_log_truncations_total Prometheus metric.
-func (l *EventLog) EventLogTruncations() int64 {
-	if l == nil {
-		return 0
-	}
-	return l.truncations.Load()
-}
-
-// EventLogTruncateFailures returns the total number of truncateLocked
-// invocations that hit an error branch. Exposed for the
-// edge_event_log_truncate_failures_total Prometheus metric.
-func (l *EventLog) EventLogTruncateFailures() int64 {
-	if l == nil {
-		return 0
-	}
-	return l.truncateFailures.Load()
-}
-
 // EventLogGaps returns the total number of replay/ fanout gaps detected
 // (events lost to truncation or pre-dating the log, or subscriber-channel-full
 // drops). Exposed for the edge_event_log_gaps_total Prometheus metric.
@@ -382,220 +45,6 @@ func (b *Bus) EventLogTruncateFailures() int64 {
 	return b.eventLog.EventLogTruncateFailures()
 }
 
-// ReadFrom returns all events with seq >= cursor from the on-disk log. It is
-// used by Bus.Subscribe when the in-memory history does not cover the cursor
-// (e.g. after a process restart). The returned hasGap flag is true when cursor
-// is non-zero and below the first seq in the log, indicating events were lost
-// to truncation or predate the log. The caller injects a GapPayload in that
-// case so the subscriber knows it must resync.
-//
-// Safe to call concurrently with Append; both serialize on l.mu. The caller
-// (Subscribe) holds the Bus lock, but Append runs outside it, so the EventLog
-// mutex is the serialization point.
-func (l *EventLog) ReadFrom(cursor int64) (events []EventEnvelope, hasGap bool) {
-	if l == nil || l.f == nil {
-		return nil, false
-	}
-	l.mu.Lock()
-	defer l.mu.Unlock()
-
-	// Detect external file-size changes (e.g. an operator truncating the log
-	// between ticks) and rebuild the index so offsets stay accurate.
-	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 len(l.orderedSeq) == 0 {
-		// No events in the log. A non-zero cursor means the caller expects
-		// events that predate the (empty) log → gap.
-		return nil, cursor > 0
-	}
-	firstSeq := l.orderedSeq[0]
-	if cursor > 0 && cursor < firstSeq {
-		// The cursor predates the oldest surviving log event: events between
-		// cursor and firstSeq were lost (truncated or predate the log).
-		hasGap = true
-		// Surface the data loss so edge_event_log_gaps_total shows replay
-		// gaps instead of letting them stay silent.
-		l.gaps.Add(1)
-	}
-
-	// Binary search for the first seq >= cursor (or first overall when cursor
-	// is 0 / below firstSeq).
-	startSeqIdx := 0
-	if cursor > firstSeq {
-		startSeqIdx = sort.Search(len(l.orderedSeq), func(i int) bool {
-			return l.orderedSeq[i] >= cursor
-		})
-	}
-	if startSeqIdx >= len(l.orderedSeq) {
-		// cursor is at or past the last logged seq: no log events to replay.
-		return nil, hasGap
-	}
-	startOffset := l.index[l.orderedSeq[startSeqIdx]]
-
-	// Seek to the start offset and read from there to EOF.
-	if _, err := l.f.Seek(startOffset, 0); err != nil {
-		slog.Warn("event log replay seek failed", "path", l.path, "offset", startOffset, "error", err)
-		return nil, hasGap
-	}
-	raw, err := io.ReadAll(l.f)
-	if err != nil {
-		slog.Warn("event log replay read failed", "path", l.path, "error", err)
-		return nil, hasGap
-	}
-	// Restore the write cursor to EOF for the next Append.
-	_, _ = l.f.Seek(0, 2)
-
-	for len(raw) > 0 {
-		nl := indexByte(raw, '\n')
-		var line []byte
-		if nl < 0 {
-			line = raw
-			raw = nil
-		} else {
-			line = raw[:nl]
-			raw = raw[nl+1:]
-		}
-		if len(line) == 0 {
-			continue
-		}
-		var env EventEnvelope
-		if json.Unmarshal(line, &env) == nil && env.Seq >= cursor {
-			events = append(events, env)
-		}
-	}
-	// Ensure the replay slice is sorted by seq (the file is append-order which
-	// should already be seq-ordered, but truncation can leave partial overlap).
-	sort.Slice(events, func(i, j int) bool { return events[i].Seq < events[j].Seq })
-	return events, hasGap
-}
-
-// Close flushes and closes the underlying file.
-func (l *EventLog) Close() error {
-	if l == nil {
-		return nil
-	}
-	l.mu.Lock()
-	defer l.mu.Unlock()
-	return l.f.Close()
-}
-
-// Path returns the file path of the event log, or empty if nil.
-func (l *EventLog) Path() string {
-	if l == nil {
-		return ""
-	}
-	return l.path
-}
-
-// PersistFn is called before an event is broadcast to subscribers.
-// If it returns an error, the event is NOT appended to history and NOT broadcast.
-type PersistFn func(EventEnvelope) error
-
-// BusOption configures a Bus.
-type BusOption func(*Bus)
-
-// WithPersister sets the persistence hook called before every event broadcast.
-// If the hook returns an error, Publish() does NOT fan out the event.
-func WithPersister(fn PersistFn) BusOption {
-	return func(b *Bus) { b.persistFn = fn }
-}
-
-// WithPersistOutputBatch controls whether run.output.batch events are
-// persisted before broadcast. Defaults to true (persist) for crash safety.
-// Set to false to accept a tradeoff: output batch events exist only in the
-// in-memory ring buffer and may be lost on crash, trading durability for
-// throughput on high-frequency stdout events.
-func WithPersistOutputBatch(persist bool) BusOption {
-	return func(b *Bus) { b.persistOutputBatch = persist }
-}
-
-// WithPersistMaxRetries overrides the number of synchronous retry attempts a
-// Publish call makes when persistFn returns an error before declaring the
-// event lost. n must be >= 0; 0 disables retries (original-attempt-only),
-// negative values are ignored (default applies). This is primarily a test
-// seam for forcing fast failure in tests that assert the persist-failure path,
-// but also lets operators tune the retry budget.
-func WithPersistMaxRetries(n int) BusOption {
-	return func(b *Bus) {
-		if n >= 0 {
-			b.persistMaxRetries = n
-		}
-	}
-}
-
-// maxPersistRetries returns the effective persist retry count for the bus,
-// falling back to persistDefaultMaxRetries when no override is set (the -1
-// sentinel left by the zero value / unset state).
-func (b *Bus) maxPersistRetries() int {
-	if b.persistMaxRetries >= 0 {
-		return b.persistMaxRetries
-	}
-	return persistDefaultMaxRetries
-}
-
-// persistWithRetry calls persistFn for evt, retrying up to maxRetries times
-// with exponential backoff on error. It returns the last error if every
-// attempt failed, or nil if any attempt succeeded. The retry loop is
-// synchronous so the Publish contract (persist-before-broadcast) is
-// preserved: an event either lands in the durable store before it is seen by
-// subscribers, or it is dropped with persistFailures incremented.
-func (b *Bus) persistWithRetry(evt EventEnvelope) error {
-	maxRetries := b.maxPersistRetries()
-	var lastErr error
-	for attempt := 0; attempt <= maxRetries; attempt++ {
-		err := b.persistFn(evt)
-		if err == nil {
-			return nil
-		}
-		lastErr = err
-		if attempt < maxRetries {
-			// Exponential backoff: 2ms, 4ms, 8ms, … capped at 50ms so the
-			// synchronous retry path cannot stall Publish for too long.
-			delay := persistRetryBaseDelay << attempt
-			if delay > 50*time.Millisecond {
-				delay = 50 * time.Millisecond
-			}
-			time.Sleep(delay)
-		}
-	}
-	return lastErr
-}
-
-// PersistFailures returns the total number of events that exhausted all
-// persist retry attempts and were dropped. Exposed for the
-// edge_event_persist_failures_total Prometheus metric.
-func (b *Bus) PersistFailures() int64 {
-	if b == nil {
-		return 0
-	}
-	return b.persistFailures.Load()
-}
-
-// WithEventLogPath configures an append-only JSON-lines event log at the
-// given path. Events are written to durable storage before being broadcast
-// to subscribers (persist-before-broadcast). This enables crash recovery
-// and replay for reconnecting clients. The event log is closed when
-// Bus.Close() is called.
-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)
-		}
-	}
-}
-
 // Bus is an in-memory event bus with monotonic sequence numbers and
 // support for cursor-based replay.
 type Bus struct {
diff --git a/edge-server/internal/events/eventlog.go b/edge-server/internal/events/eventlog.go
new file mode 100644
index 000000000..aad447cfa
--- /dev/null
+++ b/edge-server/internal/events/eventlog.go
@@ -0,0 +1,396 @@
+package events
+
+import (
+	"encoding/json"
+	"io"
+	"log/slog"
+	"os"
+	"path/filepath"
+	"sort"
+	"sync"
+	"sync/atomic"
+)
+
+// EventLog is an append-only JSON-lines event log backed by a file on disk.
+// Each event is serialised as a single JSON line. Writes are safe for
+// concurrent use.
+//
+// The log also maintains a seq→offset index so that a Bus restarting with an
+// empty in-memory history can still replay events to a cursor-bearing
+// subscriber (crash recovery / replay). The index is rebuilt on open and after
+// every truncation; Subscribe also detects external file-size changes and
+// rebuilds lazily so a truncate that happened between ticks stays safe.
+type EventLog struct {
+	mu          sync.Mutex
+	f           *os.File
+	path        string
+	maxSize     int64           // max file size in bytes before truncation; 0 = unlimited
+	index       map[int64]int64 // seq → byte offset of that line (exclusive of the seq→offset map)
+	orderedSeq  []int64         // sorted seqs parallel to index, for binary search
+	indexedSize int64           // file size at last index build, to detect external changes
+
+	// truncations counts truncateLocked invocations (a truncation was
+	// attempted because the log exceeded maxSize). Exposed via
+	// Bus.EventLogTruncations for the edge_event_log_truncations_total metric.
+	truncations atomic.Int64
+	// truncateFailures counts truncateLocked invocations that hit an error
+	// branch (seek/read/truncate/rewrite failure) and previously returned
+	// silently. Exposed via Bus.EventLogTruncateFailures for
+	// edge_event_log_truncate_failures_total so an operator whose log was
+	// growing unbounded or losing replay data finally gets a signal.
+	truncateFailures atomic.Int64
+	// gaps counts ReadFrom calls that detected a cursor predating the oldest
+	// surviving log event (replay would lose events). Exposed via
+	// Bus.EventLogGaps for the edge_event_log_gaps_total metric.
+	gaps atomic.Int64
+}
+
+const defaultEventLogMaxSize = 50 * 1024 * 1024 // 50 MiB
+
+// NewEventLog opens or creates the append-only event log at the given path.
+// The parent directory is created if it does not exist. The file is opened
+// read+write so the log can serve replay reads in addition to appends. The
+// seq→offset index is built by scanning existing lines so a restarted Bus can
+// replay history from disk before any new Publish lands.
+func NewEventLog(path string) (*EventLog, error) {
+	if path == "" {
+		return nil, nil
+	}
+	dir := filepath.Dir(path)
+	if err := os.MkdirAll(dir, 0o750); err != nil {
+		return nil, err
+	}
+	// #nosec G304 -- event log path comes from operator config (WithEventLogPath)
+	// O_RDWR so the log is readable for replay; O_APPEND so writes always go to
+	// the end regardless of the read seek position.
+	f, err := os.OpenFile(path, os.O_APPEND|os.O_CREATE|os.O_RDWR, 0o600)
+	if err != nil {
+		return nil, err
+	}
+	l := &EventLog{f: f, path: path, maxSize: defaultEventLogMaxSize, index: make(map[int64]int64)}
+	if err := l.rebuildIndexLocked(); err != nil {
+		// A corrupt or unreadable log is not fatal: the Bus keeps working with
+		// an empty index (replay falls back to in-memory history only). Log the
+		// error so operators can repair the file.
+		slog.Error("event log index build failed; replay will be incomplete until the file is repaired",
+			"path", path, "error", err)
+		l.index = make(map[int64]int64)
+		l.orderedSeq = nil
+	}
+	return l, nil
+}
+
+// rebuildIndexLocked scans the entire log file line by line, recording each
+// event's seq and its byte offset. Must be called with l.mu held.
+func (l *EventLog) rebuildIndexLocked() error {
+	if l == nil || l.f == nil {
+		return nil
+	}
+	if _, err := l.f.Seek(0, 0); err != nil {
+		return err
+	}
+	index := make(map[int64]int64)
+	var ordered []int64
+	// Track the byte offset of each line start. json.Decoder does not expose
+	// offsets, so we count bytes consumed by reading the raw file in a single
+	// pass and splitting on newlines.
+	raw, err := io.ReadAll(l.f)
+	if err != nil {
+		return err
+	}
+	offset := int64(0)
+	for len(raw) > 0 {
+		nl := indexByte(raw, '\n')
+		var line []byte
+		var lineLen int
+		if nl < 0 {
+			line = raw
+			lineLen = len(raw)
+			raw = nil
+		} else {
+			line = raw[:nl]
+			lineLen = nl + 1
+			raw = raw[lineLen:]
+		}
+		if len(line) == 0 {
+			offset += int64(lineLen)
+			continue
+		}
+		var env EventEnvelope
+		if json.Unmarshal(line, &env) == nil && env.Seq > 0 {
+			if _, exists := index[env.Seq]; !exists {
+				ordered = append(ordered, env.Seq)
+			}
+			index[env.Seq] = offset
+		}
+		offset += int64(lineLen)
+	}
+	l.index = index
+	l.orderedSeq = ordered
+	if fi, statErr := l.f.Stat(); statErr == nil {
+		l.indexedSize = fi.Size()
+	}
+	// Restore the write cursor to the end so the next Append writes at EOF.
+	_, _ = l.f.Seek(0, 2)
+	return nil
+}
+
+// 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
+}
+
+// Append writes an event to the log as a single JSON line followed by a newline.
+// After the write, the event's seq→offset is added to the index so replay can
+// find it without rescanning the file. When the file exceeds maxSize the log
+// is truncated and the index rebuilt.
+func (l *EventLog) Append(evt EventEnvelope) error {
+	if l == nil {
+		return nil
+	}
+	l.mu.Lock()
+	defer l.mu.Unlock()
+
+	// Record the write offset before writing so the index points at the line
+	// start. With O_APPEND the current seek position is irrelevant for writes,
+	// so we stat to get the current end-of-file size.
+	writeOffset := int64(0)
+	if fi, statErr := l.f.Stat(); statErr == nil {
+		writeOffset = fi.Size()
+	}
+
+	data, err := json.Marshal(evt)
+	if err != nil {
+		return err
+	}
+	data = append(data, '\n')
+	_, err = l.f.Write(data)
+	if err == nil {
+		// Extend the live index so the just-appended event is immediately
+		// replayable without a full rescan.
+		if evt.Seq > 0 {
+			if _, exists := l.index[evt.Seq]; !exists {
+				l.orderedSeq = appendSorted(l.orderedSeq, evt.Seq)
+			}
+			l.index[evt.Seq] = writeOffset
+		}
+		if l.maxSize > 0 {
+			if fi, statErr := l.f.Stat(); statErr == nil && fi.Size() > l.maxSize {
+				l.truncateLocked()
+			}
+		}
+	}
+	return err
+}
+
+// appendSorted inserts seq into the sorted slice maintaining order. Used by
+// Append to extend the ordered index without a full sort.
+func appendSorted(sorted []int64, seq int64) []int64 {
+	idx := sort.Search(len(sorted), func(i int) bool { return sorted[i] >= seq })
+	if idx < len(sorted) && sorted[idx] == seq {
+		return sorted // already present
+	}
+	sorted = append(sorted, 0)
+	copy(sorted[idx+1:], sorted[idx:])
+	sorted[idx] = seq
+	return sorted
+}
+
+// truncateLocked rewrites the log file keeping only the trailing portion.
+// Must be called with l.mu held. The index is rebuilt after the rewrite so
+// replay offsets stay accurate after truncation. Every failure branch that
+// previously returned silently now increments truncateFailures and emits a
+// slog.Error so an operator whose log is growing unbounded or losing replay
+// offsets finally gets a signal (edge_event_log_truncate_failures_total).
+func (l *EventLog) truncateLocked() {
+	l.truncations.Add(1)
+	keepBytes := l.maxSize * 3 / 4 // keep last 75%
+	if _, seekErr := l.f.Seek(-keepBytes, 2); seekErr != nil {
+		// File too small or seek failed; skip truncation but surface it.
+		l.truncateFailures.Add(1)
+		slog.Error("event log truncate seek failed",
+			"path", l.path, "keepBytes", keepBytes, "error", seekErr)
+		return
+	}
+	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
+	}
+	// Skip to next newline so we don't keep a partial line.
+	start := 0
+	for i := 0; i < n; i++ {
+		if buf[i] == '\n' {
+			start = i + 1
+			break
+		}
+	}
+	// Truncate and rewrite.
+	if truncErr := l.f.Truncate(0); truncErr != nil {
+		l.truncateFailures.Add(1)
+		slog.Error("event log truncate Truncate(0) failed",
+			"path", l.path, "error", truncErr)
+		return
+	}
+	if _, seekErr := l.f.Seek(0, 0); seekErr != nil {
+		l.truncateFailures.Add(1)
+		slog.Error("event log truncate seek-to-start failed",
+			"path", l.path, "error", seekErr)
+		return
+	}
+	if start < n {
+		// Best-effort rewrite: the log truncation path degrades silently on
+		// write failure rather than failing the enclosing Append call.
+		if written, writeErr := l.f.Write(buf[start:n]); writeErr != nil || written != n-start {
+			l.truncateFailures.Add(1)
+			slog.Error("event log truncate rewrite failed",
+				"path", l.path, "written", written, "want", n-start, "error", writeErr)
+		}
+	}
+	// Rebuild the index so replay offsets reflect the truncated file. A
+	// rebuild failure also counts as a truncate failure (the log is now in a
+	// partially-rewritten state and replay offsets are unreliable).
+	if err := l.rebuildIndexLocked(); err != nil {
+		l.truncateFailures.Add(1)
+		slog.Error("event log truncate index rebuild failed",
+			"path", l.path, "error", err)
+	}
+}
+
+// EventLogTruncations returns the total number of truncateLocked invocations
+// (truncations attempted because the log exceeded maxSize). Exposed for the
+// edge_event_log_truncations_total Prometheus metric.
+func (l *EventLog) EventLogTruncations() int64 {
+	if l == nil {
+		return 0
+	}
+	return l.truncations.Load()
+}
+
+// EventLogTruncateFailures returns the total number of truncateLocked
+// invocations that hit an error branch. Exposed for the
+// edge_event_log_truncate_failures_total Prometheus metric.
+func (l *EventLog) EventLogTruncateFailures() int64 {
+	if l == nil {
+		return 0
+	}
+	return l.truncateFailures.Load()
+}
+
+// ReadFrom returns all events with seq >= cursor from the on-disk log. It is
+// used by Bus.Subscribe when the in-memory history does not cover the cursor
+// (e.g. after a process restart). The returned hasGap flag is true when cursor
+// is non-zero and below the first seq in the log, indicating events were lost
+// to truncation or predate the log. The caller injects a GapPayload in that
+// case so the subscriber knows it must resync.
+//
+// Safe to call concurrently with Append; both serialize on l.mu. The caller
+// (Subscribe) holds the Bus lock, but Append runs outside it, so the EventLog
+// mutex is the serialization point.
+func (l *EventLog) ReadFrom(cursor int64) (events []EventEnvelope, hasGap bool) {
+	if l == nil || l.f == nil {
+		return nil, false
+	}
+	l.mu.Lock()
+	defer l.mu.Unlock()
+
+	// Detect external file-size changes (e.g. an operator truncating the log
+	// between ticks) and rebuild the index so offsets stay accurate.
+	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 len(l.orderedSeq) == 0 {
+		// No events in the log. A non-zero cursor means the caller expects
+		// events that predate the (empty) log → gap.
+		return nil, cursor > 0
+	}
+	firstSeq := l.orderedSeq[0]
+	if cursor > 0 && cursor < firstSeq {
+		// The cursor predates the oldest surviving log event: events between
+		// cursor and firstSeq were lost (truncated or predate the log).
+		hasGap = true
+		// Surface the data loss so edge_event_log_gaps_total shows replay
+		// gaps instead of letting them stay silent.
+		l.gaps.Add(1)
+	}
+
+	// Binary search for the first seq >= cursor (or first overall when cursor
+	// is 0 / below firstSeq).
+	startSeqIdx := 0
+	if cursor > firstSeq {
+		startSeqIdx = sort.Search(len(l.orderedSeq), func(i int) bool {
+			return l.orderedSeq[i] >= cursor
+		})
+	}
+	if startSeqIdx >= len(l.orderedSeq) {
+		// cursor is at or past the last logged seq: no log events to replay.
+		return nil, hasGap
+	}
+	startOffset := l.index[l.orderedSeq[startSeqIdx]]
+
+	// Seek to the start offset and read from there to EOF.
+	if _, err := l.f.Seek(startOffset, 0); err != nil {
+		slog.Warn("event log replay seek failed", "path", l.path, "offset", startOffset, "error", err)
+		return nil, hasGap
+	}
+	raw, err := io.ReadAll(l.f)
+	if err != nil {
+		slog.Warn("event log replay read failed", "path", l.path, "error", err)
+		return nil, hasGap
+	}
+	// Restore the write cursor to EOF for the next Append.
+	_, _ = l.f.Seek(0, 2)
+
+	for len(raw) > 0 {
+		nl := indexByte(raw, '\n')
+		var line []byte
+		if nl < 0 {
+			line = raw
+			raw = nil
+		} else {
+			line = raw[:nl]
+			raw = raw[nl+1:]
+		}
+		if len(line) == 0 {
+			continue
+		}
+		var env EventEnvelope
+		if json.Unmarshal(line, &env) == nil && env.Seq >= cursor {
+			events = append(events, env)
+		}
+	}
+	// Ensure the replay slice is sorted by seq (the file is append-order which
+	// should already be seq-ordered, but truncation can leave partial overlap).
+	sort.Slice(events, func(i, j int) bool { return events[i].Seq < events[j].Seq })
+	return events, hasGap
+}
+
+// Close flushes and closes the underlying file.
+func (l *EventLog) Close() error {
+	if l == nil {
+		return nil
+	}
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	return l.f.Close()
+}
+
+// Path returns the file path of the event log, or empty if nil.
+func (l *EventLog) Path() string {
+	if l == nil {
+		return ""
+	}
+	return l.path
+}
diff --git a/edge-server/internal/events/persist.go b/edge-server/internal/events/persist.go
new file mode 100644
index 000000000..5a6ba76c5
--- /dev/null
+++ b/edge-server/internal/events/persist.go
@@ -0,0 +1,120 @@
+package events
+
+import (
+	"log/slog"
+	"time"
+)
+
+// persistDefaultMaxRetries bounds how many times persistWithRetry retries a
+// persistFn call that returned an error before declaring the event lost.
+// 1 = one original attempt + one retry; 3 = original + three retries.
+const persistDefaultMaxRetries = 3
+
+// persistRetryBaseDelay is the exponential backoff base between persist
+// retry attempts. Kept short so the synchronous retry path does not
+// stall Publish under normal transient failures.
+const persistRetryBaseDelay = 2 * time.Millisecond
+
+// PersistFn is called before an event is broadcast to subscribers.
+// If it returns an error, the event is NOT appended to history and NOT broadcast.
+type PersistFn func(EventEnvelope) error
+
+// BusOption configures a Bus.
+type BusOption func(*Bus)
+
+// WithPersister sets the persistence hook called before every event broadcast.
+// If the hook returns an error, Publish() does NOT fan out the event.
+func WithPersister(fn PersistFn) BusOption {
+	return func(b *Bus) { b.persistFn = fn }
+}
+
+// WithPersistOutputBatch controls whether run.output.batch events are
+// persisted before broadcast. Defaults to true (persist) for crash safety.
+// Set to false to accept a tradeoff: output batch events exist only in the
+// in-memory ring buffer and may be lost on crash, trading durability for
+// throughput on high-frequency stdout events.
+func WithPersistOutputBatch(persist bool) BusOption {
+	return func(b *Bus) { b.persistOutputBatch = persist }
+}
+
+// WithPersistMaxRetries overrides the number of synchronous retry attempts a
+// Publish call makes when persistFn returns an error before declaring the
+// event lost. n must be >= 0; 0 disables retries (original-attempt-only),
+// negative values are ignored (default applies). This is primarily a test
+// seam for forcing fast failure in tests that assert the persist-failure path,
+// but also lets operators tune the retry budget.
+func WithPersistMaxRetries(n int) BusOption {
+	return func(b *Bus) {
+		if n >= 0 {
+			b.persistMaxRetries = n
+		}
+	}
+}
+
+// maxPersistRetries returns the effective persist retry count for the bus,
+// falling back to persistDefaultMaxRetries when no override is set (the -1
+// sentinel left by the zero value / unset state).
+func (b *Bus) maxPersistRetries() int {
+	if b.persistMaxRetries >= 0 {
+		return b.persistMaxRetries
+	}
+	return persistDefaultMaxRetries
+}
+
+// persistWithRetry calls persistFn for evt, retrying up to maxRetries times
+// with exponential backoff on error. It returns the last error if every
+// attempt failed, or nil if any attempt succeeded. The retry loop is
+// synchronous so the Publish contract (persist-before-broadcast) is
+// preserved: an event either lands in the durable store before it is seen by
+// subscribers, or it is dropped with persistFailures incremented.
+func (b *Bus) persistWithRetry(evt EventEnvelope) error {
+	maxRetries := b.maxPersistRetries()
+	var lastErr error
+	for attempt := 0; attempt <= maxRetries; attempt++ {
+		err := b.persistFn(evt)
+		if err == nil {
+			return nil
+		}
+		lastErr = err
+		if attempt < maxRetries {
+			// Exponential backoff: 2ms, 4ms, 8ms, … capped at 50ms so the
+			// synchronous retry path cannot stall Publish for too long.
+			delay := persistRetryBaseDelay << attempt
+			if delay > 50*time.Millisecond {
+				delay = 50 * time.Millisecond
+			}
+			time.Sleep(delay)
+		}
+	}
+	return lastErr
+}
+
+// PersistFailures returns the total number of events that exhausted all
+// persist retry attempts and were dropped. Exposed for the
+// edge_event_persist_failures_total Prometheus metric.
+func (b *Bus) PersistFailures() int64 {
+	if b == nil {
+		return 0
+	}
+	return b.persistFailures.Load()
+}
+
+// WithEventLogPath configures an append-only JSON-lines event log at the
+// given path. Events are written to durable storage before being broadcast
+// to subscribers (persist-before-broadcast). This enables crash recovery
+// and replay for reconnecting clients. The event log is closed when
+// Bus.Close() is called.
+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)
+		}
+	}
+}
diff --git a/edge-server/internal/events/types.go b/edge-server/internal/events/types.go
new file mode 100644
index 000000000..2925acf41
--- /dev/null
+++ b/edge-server/internal/events/types.go
@@ -0,0 +1,52 @@
+package events
+
+const (
+	defaultMaxHistory           = 10000
+	subscriberChannelBufferSize = 256
+	defaultWorkerCount          = 4
+	observerJobBufferSize       = 1024
+)
+
+// GapEventType is the event type for a gap-detection control message sent to a
+// subscriber when one or more events were dropped because the subscriber channel
+// was full. The payload is a *GapPayload.
+const GapEventType = "system.gap"
+
+// GapPayload describes a range of dropped events for a subscriber.
+type GapPayload struct {
+	FirstDroppedSeq int64 `json:"firstDroppedSeq"`
+	LastDroppedSeq  int64 `json:"lastDroppedSeq"`
+	DroppedCount    int64 `json:"droppedCount"`
+}
+
+// observerJob is a unit of work dispatched to the observer worker pool.
+type observerJob struct {
+	fn  func(EventEnvelope)
+	evt EventEnvelope
+}
+
+// EventEnvelope is the standard event wrapper for all WebSocket events.
+type EventEnvelope struct {
+	Version string         `json:"version"`
+	ID      string         `json:"id"`
+	Seq     int64          `json:"seq"`
+	Type    string         `json:"type"`
+	Scope   map[string]any `json:"scope"`
+	TraceID string         `json:"traceId"`
+	SentAt  string         `json:"sentAt"`
+	Payload any            `json:"payload"`
+}
+
+// subscriber receives events on its channel.
+type subscriber struct {
+	id          int64
+	ch          chan EventEnvelope
+	gapDetected bool  // true when events were dropped since last successful send
+	firstGapSeq int64 // seq of first dropped event in the gap
+	lastGapSeq  int64 // seq of last dropped event in the gap
+}
+
+type observer struct {
+	id int64
+	fn func(EventEnvelope)
+}
diff --git a/edge-server/internal/lifecycle/result_aggregator.go b/edge-server/internal/lifecycle/result_aggregator.go
index 5dc388032..34a04b374 100644
--- a/edge-server/internal/lifecycle/result_aggregator.go
+++ b/edge-server/internal/lifecycle/result_aggregator.go
@@ -1,11 +1,7 @@
-// Package lifecycle provides result aggregation for orchestrator sub-agent runs.
 package lifecycle
 
 import (
-	"fmt"
 	"log/slog"
-	"sort"
-	"strings"
 	"sync"
 	"time"
 
@@ -14,222 +10,6 @@ import (
 	"github.com/agenthub/edge-server/internal/events"
 )
 
-// ── SubAgent Result Types ──────────────────────────────────────────────────
-
-// SubAgentResult holds the structured output of a single sub-agent run.
-// It packages status, output, artifacts, and token usage for downstream
-// synthesis by the orchestrator LLM.
-//
-// Reference: AionUi Team Mode Mailbox — persisted sub-agent results queryable by Leader.
-// Reference: LibreChat — structured subagent result return with output and metadata.
-type SubAgentResult struct {
-	AgentID     string        `json:"agentId"`
-	AgentName   string        `json:"agentName"`
-	RunID       string        `json:"runId"`
-	Status      string        `json:"status"` // "finished", "failed", "cancelled"
-	Output      any           `json:"output,omitempty"`
-	Error       string        `json:"error,omitempty"`
-	TokenUsage  *TokenUsage   `json:"tokenUsage,omitempty"`
-	Artifacts   []ArtifactRef `json:"artifacts,omitempty"`
-	CompletedAt time.Time     `json:"completedAt"`
-}
-
-// TokenUsage tracks token consumption for a sub-agent run.
-type TokenUsage struct {
-	InputTokens  int64 `json:"inputTokens"`
-	OutputTokens int64 `json:"outputTokens"`
-	TotalTokens  int64 `json:"totalTokens"`
-}
-
-// ArtifactRef references a generated artifact from a sub-agent run.
-type ArtifactRef struct {
-	ID       string `json:"id"`
-	Type     string `json:"type"`
-	Filename string `json:"filename,omitempty"`
-	URL      string `json:"url,omitempty"`
-}
-
-// SubAgentAggregatedResult is the synthesized result payload emitted
-// in the run.agent.sub_agents_complete event after all children finish
-// (or the timeout fallback triggers).
-type SubAgentAggregatedResult struct {
-	ParentID      string           `json:"parentId"`
-	TotalChildren int              `json:"totalChildren"`
-	Succeeded     int              `json:"succeeded"`
-	Failed        int              `json:"failed"`
-	Cancelled     int              `json:"cancelled"`
-	Pending       int              `json:"pending"` // children that never completed (timeout)
-	Results       []SubAgentResult `json:"results"`
-	Partial       bool             `json:"partial"`           // true if timeout fallback triggered
-	Summary       string           `json:"summary,omitempty"` // human-readable synthesis
-}
-
-// ── SubAgentResultCollector ───────────────────────────────────────────────
-
-// DefaultSubAgentTimeout is the default timeout for waiting on all sub-agent
-// children to complete before emitting partial results.
-const DefaultSubAgentTimeout = 5 * time.Minute
-
-// SubAgentResultCollectorTimeoutCheckInterval is how often the timeout
-// goroutine checks for expired parents.
-const SubAgentResultCollectorTimeoutCheckInterval = 30 * time.Second
-
-// SubAgentResultCollector stores structured results from sub-agent runs,
-// providing aggregation and synthesis capabilities when all children complete.
-//
-// References:
-//   - AionUi Team Mode Mailbox: persisted sub-agent results queryable by Leader
-//   - LibreChat: structured subagent result return with output and metadata
-type SubAgentResultCollector struct {
-	mu         sync.RWMutex
-	results    map[string][]SubAgentResult // parentID -> results list (appended as children complete)
-	firstSpawn map[string]time.Time        // parentID -> time first child was spawned
-	exhausted  map[string]bool             // parentID -> true once results emitted (full or partial)
-	timeout    time.Duration
-}
-
-// NewSubAgentResultCollector creates a collector with the given timeout.
-// A value <= 0 uses DefaultSubAgentTimeout.
-func NewSubAgentResultCollector(timeout time.Duration) *SubAgentResultCollector {
-	if timeout <= 0 {
-		timeout = DefaultSubAgentTimeout
-	}
-	return &SubAgentResultCollector{
-		results:    make(map[string][]SubAgentResult),
-		firstSpawn: make(map[string]time.Time),
-		exhausted:  make(map[string]bool),
-		timeout:    timeout,
-	}
-}
-
-// RecordSpawn records that a child was spawned for the given parent.
-// This enables timeout tracking.
-func (c *SubAgentResultCollector) RecordSpawn(parentID string) {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-	if _, ok := c.firstSpawn[parentID]; !ok {
-		c.firstSpawn[parentID] = time.Now()
-	}
-}
-
-// Store adds a sub-agent result to the collector.
-func (c *SubAgentResultCollector) Store(parentID string, result SubAgentResult) {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-	c.results[parentID] = append(c.results[parentID], result)
-}
-
-// 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]
-}
-
-// HasTimedOut returns true if the parent's first spawn was more than
-// the configured timeout ago and results have not already been exhausted.
-func (c *SubAgentResultCollector) HasTimedOut(parentID string) bool {
-	c.mu.RLock()
-	defer c.mu.RUnlock()
-	if c.exhausted[parentID] {
-		return false
-	}
-	spawnTime, ok := c.firstSpawn[parentID]
-	if !ok {
-		return false
-	}
-	return time.Since(spawnTime) > c.timeout
-}
-
-// ExpiredParents returns parent IDs whose timeout has elapsed and results
-// have not yet been exhausted. Callers should emit partial results for these.
-func (c *SubAgentResultCollector) ExpiredParents() []string {
-	c.mu.RLock()
-	defer c.mu.RUnlock()
-	var expired []string
-	for parentID, spawnTime := range c.firstSpawn {
-		if c.exhausted[parentID] {
-			continue
-		}
-		if time.Since(spawnTime) > c.timeout {
-			expired = append(expired, parentID)
-		}
-	}
-	return expired
-}
-
-// Aggregate builds a SubAgentAggregatedResult for the given parent from
-// stored results. The partial flag indicates whether the aggregation includes
-// all expected children (false) or is a timeout-induced partial result (true).
-func (c *SubAgentResultCollector) Aggregate(parentID string, partial bool) *SubAgentAggregatedResult {
-	c.mu.RLock()
-	defer c.mu.RUnlock()
-
-	stored := c.results[parentID]
-	agg := &SubAgentAggregatedResult{
-		ParentID:      parentID,
-		TotalChildren: len(stored),
-		Results:       make([]SubAgentResult, len(stored)),
-		Partial:       partial,
-	}
-	copy(agg.Results, stored)
-
-	// Sort by completion time for deterministic output.
-	sort.Slice(agg.Results, func(i, j int) bool {
-		return agg.Results[i].CompletedAt.Before(agg.Results[j].CompletedAt)
-	})
-
-	for _, r := range agg.Results {
-		switch r.Status {
-		case "finished":
-			agg.Succeeded++
-		case "failed":
-			agg.Failed++
-		case "cancelled":
-			agg.Cancelled++
-		default:
-			agg.Pending++
-		}
-	}
-
-	agg.Summary = buildAggregateSummary(agg)
-	return agg
-}
-
-// buildAggregateSummary produces a human-readable summary of aggregated results.
-func buildAggregateSummary(agg *SubAgentAggregatedResult) string {
-	var parts []string
-	if agg.Succeeded > 0 {
-		parts = append(parts, fmt.Sprintf("%d succeeded", agg.Succeeded))
-	}
-	if agg.Failed > 0 {
-		parts = append(parts, fmt.Sprintf("%d failed", agg.Failed))
-	}
-	if agg.Cancelled > 0 {
-		parts = append(parts, fmt.Sprintf("%d cancelled", agg.Cancelled))
-	}
-	if agg.Pending > 0 {
-		parts = append(parts, fmt.Sprintf("%d pending (timed out)", agg.Pending))
-	}
-
-	base := fmt.Sprintf("Sub-agents complete: %d total", agg.TotalChildren)
-	if len(parts) > 0 {
-		base += " (" + strings.Join(parts, ", ") + ")"
-	}
-	if agg.Partial {
-		base += " [partial — timeout fallback]"
-	}
-	return base
-}
-
 // ── ResultAggregator ──────────────────────────────────────────────────────
 
 // ResultAggregator listens for sub-agent run completion events on the event bus,
diff --git a/edge-server/internal/lifecycle/subagent_collector.go b/edge-server/internal/lifecycle/subagent_collector.go
new file mode 100644
index 000000000..c22752e2a
--- /dev/null
+++ b/edge-server/internal/lifecycle/subagent_collector.go
@@ -0,0 +1,231 @@
+// SubAgentResultCollector aggregates structured results from orchestrator
+// sub-agent runs. This file is the pure in-memory state machine half of the
+// result-aggregation domain; ResultAggregator in result_aggregator.go wires it
+// to the event bus and the agent registry.
+package lifecycle
+
+import (
+	"fmt"
+	"sort"
+	"strings"
+	"sync"
+	"time"
+)
+
+// ── SubAgent Result Types ──────────────────────────────────────────────────
+
+// SubAgentResult holds the structured output of a single sub-agent run.
+// It packages status, output, artifacts, and token usage for downstream
+// synthesis by the orchestrator LLM.
+//
+// Reference: AionUi Team Mode Mailbox — persisted sub-agent results queryable by Leader.
+// Reference: LibreChat — structured subagent result return with output and metadata.
+type SubAgentResult struct {
+	AgentID     string        `json:"agentId"`
+	AgentName   string        `json:"agentName"`
+	RunID       string        `json:"runId"`
+	Status      string        `json:"status"` // "finished", "failed", "cancelled"
+	Output      any           `json:"output,omitempty"`
+	Error       string        `json:"error,omitempty"`
+	TokenUsage  *TokenUsage   `json:"tokenUsage,omitempty"`
+	Artifacts   []ArtifactRef `json:"artifacts,omitempty"`
+	CompletedAt time.Time     `json:"completedAt"`
+}
+
+// TokenUsage tracks token consumption for a sub-agent run.
+type TokenUsage struct {
+	InputTokens  int64 `json:"inputTokens"`
+	OutputTokens int64 `json:"outputTokens"`
+	TotalTokens  int64 `json:"totalTokens"`
+}
+
+// ArtifactRef references a generated artifact from a sub-agent run.
+type ArtifactRef struct {
+	ID       string `json:"id"`
+	Type     string `json:"type"`
+	Filename string `json:"filename,omitempty"`
+	URL      string `json:"url,omitempty"`
+}
+
+// SubAgentAggregatedResult is the synthesized result payload emitted
+// in the run.agent.sub_agents_complete event after all children finish
+// (or the timeout fallback triggers).
+type SubAgentAggregatedResult struct {
+	ParentID      string           `json:"parentId"`
+	TotalChildren int              `json:"totalChildren"`
+	Succeeded     int              `json:"succeeded"`
+	Failed        int              `json:"failed"`
+	Cancelled     int              `json:"cancelled"`
+	Pending       int              `json:"pending"` // children that never completed (timeout)
+	Results       []SubAgentResult `json:"results"`
+	Partial       bool             `json:"partial"`           // true if timeout fallback triggered
+	Summary       string           `json:"summary,omitempty"` // human-readable synthesis
+}
+
+// ── SubAgentResultCollector ───────────────────────────────────────────────
+
+// DefaultSubAgentTimeout is the default timeout for waiting on all sub-agent
+// children to complete before emitting partial results.
+const DefaultSubAgentTimeout = 5 * time.Minute
+
+// SubAgentResultCollectorTimeoutCheckInterval is how often the timeout
+// goroutine checks for expired parents.
+const SubAgentResultCollectorTimeoutCheckInterval = 30 * time.Second
+
+// SubAgentResultCollector stores structured results from sub-agent runs,
+// providing aggregation and synthesis capabilities when all children complete.
+//
+// References:
+//   - AionUi Team Mode Mailbox: persisted sub-agent results queryable by Leader
+//   - LibreChat: structured subagent result return with output and metadata
+type SubAgentResultCollector struct {
+	mu         sync.RWMutex
+	results    map[string][]SubAgentResult // parentID -> results list (appended as children complete)
+	firstSpawn map[string]time.Time        // parentID -> time first child was spawned
+	exhausted  map[string]bool             // parentID -> true once results emitted (full or partial)
+	timeout    time.Duration
+	now        func() time.Time // injectable clock for deterministic tests
+}
+
+// NewSubAgentResultCollector creates a collector with the given timeout.
+// A value <= 0 uses DefaultSubAgentTimeout.
+func NewSubAgentResultCollector(timeout time.Duration) *SubAgentResultCollector {
+	if timeout <= 0 {
+		timeout = DefaultSubAgentTimeout
+	}
+	return &SubAgentResultCollector{
+		results:    make(map[string][]SubAgentResult),
+		firstSpawn: make(map[string]time.Time),
+		exhausted:  make(map[string]bool),
+		timeout:    timeout,
+		now:        time.Now,
+	}
+}
+
+// RecordSpawn records that a child was spawned for the given parent.
+// This enables timeout tracking.
+func (c *SubAgentResultCollector) RecordSpawn(parentID string) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if _, ok := c.firstSpawn[parentID]; !ok {
+		c.firstSpawn[parentID] = c.now()
+	}
+}
+
+// Store adds a sub-agent result to the collector.
+func (c *SubAgentResultCollector) Store(parentID string, result SubAgentResult) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.results[parentID] = append(c.results[parentID], result)
+}
+
+// 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]
+}
+
+// HasTimedOut returns true if the parent's first spawn was more than
+// the configured timeout ago and results have not already been exhausted.
+func (c *SubAgentResultCollector) HasTimedOut(parentID string) bool {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	if c.exhausted[parentID] {
+		return false
+	}
+	spawnTime, ok := c.firstSpawn[parentID]
+	if !ok {
+		return false
+	}
+	return c.now().Sub(spawnTime) > c.timeout
+}
+
+// ExpiredParents returns parent IDs whose timeout has elapsed and results
+// have not yet been exhausted. Callers should emit partial results for these.
+func (c *SubAgentResultCollector) ExpiredParents() []string {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	var expired []string
+	for parentID, spawnTime := range c.firstSpawn {
+		if c.exhausted[parentID] {
+			continue
+		}
+		if c.now().Sub(spawnTime) > c.timeout {
+			expired = append(expired, parentID)
+		}
+	}
+	return expired
+}
+
+// Aggregate builds a SubAgentAggregatedResult for the given parent from
+// stored results. The partial flag indicates whether the aggregation includes
+// all expected children (false) or is a timeout-induced partial result (true).
+func (c *SubAgentResultCollector) Aggregate(parentID string, partial bool) *SubAgentAggregatedResult {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+
+	stored := c.results[parentID]
+	agg := &SubAgentAggregatedResult{
+		ParentID:      parentID,
+		TotalChildren: len(stored),
+		Results:       make([]SubAgentResult, len(stored)),
+		Partial:       partial,
+	}
+	copy(agg.Results, stored)
+
+	// Sort by completion time for deterministic output.
+	sort.Slice(agg.Results, func(i, j int) bool {
+		return agg.Results[i].CompletedAt.Before(agg.Results[j].CompletedAt)
+	})
+
+	for _, r := range agg.Results {
+		switch r.Status {
+		case "finished":
+			agg.Succeeded++
+		case "failed":
+			agg.Failed++
+		case "cancelled":
+			agg.Cancelled++
+		default:
+			agg.Pending++
+		}
+	}
+
+	agg.Summary = buildAggregateSummary(agg)
+	return agg
+}
+
+// buildAggregateSummary produces a human-readable summary of aggregated results.
+func buildAggregateSummary(agg *SubAgentAggregatedResult) string {
+	var parts []string
+	if agg.Succeeded > 0 {
+		parts = append(parts, fmt.Sprintf("%d succeeded", agg.Succeeded))
+	}
+	if agg.Failed > 0 {
+		parts = append(parts, fmt.Sprintf("%d failed", agg.Failed))
+	}
+	if agg.Cancelled > 0 {
+		parts = append(parts, fmt.Sprintf("%d cancelled", agg.Cancelled))
+	}
+	if agg.Pending > 0 {
+		parts = append(parts, fmt.Sprintf("%d pending (timed out)", agg.Pending))
+	}
+
+	base := fmt.Sprintf("Sub-agents complete: %d total", agg.TotalChildren)
+	if len(parts) > 0 {
+		base += " (" + strings.Join(parts, ", ") + ")"
+	}
+	if agg.Partial {
+		base += " [partial — timeout fallback]"
+	}
+	return base
+}
diff --git a/edge-server/internal/lifecycle/subagent_collector_test.go b/edge-server/internal/lifecycle/subagent_collector_test.go
new file mode 100644
index 000000000..ab495be2a
--- /dev/null
+++ b/edge-server/internal/lifecycle/subagent_collector_test.go
@@ -0,0 +1,99 @@
+package lifecycle
+
+import (
+	"testing"
+	"time"
+)
+
+func TestSubAgentResultCollectorAggregate(t *testing.T) {
+	collector := NewSubAgentResultCollector(time.Minute)
+
+	base := time.Date(2026, 8, 13, 12, 0, 0, 0, time.UTC)
+	collector.RecordSpawn("parent_1")
+	collector.Store("parent_1", SubAgentResult{
+		AgentID:     "a1",
+		RunID:       "run_1",
+		Status:      "finished",
+		CompletedAt: base.Add(30 * time.Second),
+	})
+	collector.Store("parent_1", SubAgentResult{
+		AgentID:     "a2",
+		RunID:       "run_2",
+		Status:      "failed",
+		CompletedAt: base,
+	})
+	collector.Store("parent_1", SubAgentResult{
+		AgentID:     "a3",
+		RunID:       "run_3",
+		Status:      "cancelled",
+		CompletedAt: base.Add(time.Minute),
+	})
+
+	agg := collector.Aggregate("parent_1", false)
+	if agg == nil {
+		t.Fatal("Aggregate returned nil")
+	}
+	if agg.TotalChildren != 3 || agg.Succeeded != 1 || agg.Failed != 1 || agg.Cancelled != 1 || agg.Pending != 0 {
+		t.Fatalf("counts = %#v, want 3 total, 1/1/1/0", agg)
+	}
+	if agg.Partial {
+		t.Fatal("Partial = true, want false for full aggregation")
+	}
+	if agg.Summary == "" {
+		t.Fatal("Summary is empty")
+	}
+	// Results are sorted by completion time: a2 (base), a1 (+30s), a3 (+1m).
+	if len(agg.Results) != 3 || agg.Results[0].AgentID != "a2" || agg.Results[1].AgentID != "a1" || agg.Results[2].AgentID != "a3" {
+		t.Fatalf("result order = %#v, want [a2 a1 a3]", agg.Results)
+	}
+}
+
+func TestSubAgentResultCollectorExhaust(t *testing.T) {
+	collector := NewSubAgentResultCollector(time.Minute)
+	collector.RecordSpawn("parent_1")
+
+	if collector.IsExhausted("parent_1") {
+		t.Fatal("IsExhausted = true before Exhaust")
+	}
+	collector.Exhaust("parent_1")
+	if !collector.IsExhausted("parent_1") {
+		t.Fatal("IsExhausted = false after Exhaust")
+	}
+}
+
+func TestSubAgentResultCollectorTimeout(t *testing.T) {
+	collector := NewSubAgentResultCollector(time.Minute)
+	now := time.Date(2026, 8, 13, 12, 0, 0, 0, time.UTC)
+	collector.now = func() time.Time { return now }
+	collector.RecordSpawn("parent_1")
+
+	if collector.HasTimedOut("parent_1") {
+		t.Fatal("HasTimedOut = true immediately after spawn")
+	}
+	now = now.Add(time.Minute + time.Second)
+	if !collector.HasTimedOut("parent_1") {
+		t.Fatal("HasTimedOut = false after timeout elapsed")
+	}
+	if expired := collector.ExpiredParents(); len(expired) != 1 || expired[0] != "parent_1" {
+		t.Fatalf("ExpiredParents = %#v, want [parent_1]", expired)
+	}
+
+	// Exhaust suppresses the timeout fallback for that parent.
+	collector.Exhaust("parent_1")
+	if collector.HasTimedOut("parent_1") {
+		t.Fatal("HasTimedOut = true after Exhaust")
+	}
+	if expired := collector.ExpiredParents(); len(expired) != 0 {
+		t.Fatalf("ExpiredParents = %#v after Exhaust, want empty", expired)
+	}
+}
+
+func TestSubAgentResultCollectorUnknownParent(t *testing.T) {
+	collector := NewSubAgentResultCollector(time.Minute)
+	if collector.HasTimedOut("missing") {
+		t.Fatal("HasTimedOut = true for unknown parent")
+	}
+	if agg := collector.Aggregate("missing", false); agg == nil || agg.TotalChildren != 0 {
+		t.Fatalf("Aggregate(missing) = %#v, want empty aggregate", agg)
+	}
+}
diff --git a/edge-server/internal/mcp/server.go b/edge-server/internal/mcp/server.go
index 560e08884..6a45c77b1 100644
--- a/edge-server/internal/mcp/server.go
+++ b/edge-server/internal/mcp/server.go
@@ -36,9 +36,9 @@ import (
 	"net/http"
 	"strings"
 
-	"github.com/agenthub/edge-server/internal/api"
 	"github.com/agenthub/edge-server/internal/events"
 	"github.com/agenthub/edge-server/internal/lifecycle"
+	"github.com/agenthub/edge-server/internal/permission"
 	"github.com/agenthub/edge-server/internal/store"
 )
 
@@ -109,7 +109,7 @@ type Server struct {
 	store              store.Repository
 	executor           lifecycle.RunExecutor
 	bus                *events.Bus
-	permissionRegistry *api.PermissionRegistry
+	permissionRegistry *permission.PermissionRegistry
 	workspaceAllowlist []string
 
 	// authToken, if non-empty, is required as Bearer token on every MCP request.
@@ -124,7 +124,7 @@ func NewServer(
 	repository store.Repository,
 	executor lifecycle.RunExecutor,
 	bus *events.Bus,
-	permissionRegistry *api.PermissionRegistry,
+	permissionRegistry *permission.PermissionRegistry,
 ) *Server {
 	return &Server{
 		store:              repository,
diff --git a/edge-server/internal/mcp/server_test.go b/edge-server/internal/mcp/server_test.go
index 4b098f6ff..efd2923e6 100644
--- a/edge-server/internal/mcp/server_test.go
+++ b/edge-server/internal/mcp/server_test.go
@@ -12,7 +12,7 @@ import (
 	"testing"
 
 	"github.com/agenthub/edge-server/internal/adapters"
-	"github.com/agenthub/edge-server/internal/api"
+	"github.com/agenthub/edge-server/internal/permission"
 	"github.com/agenthub/edge-server/internal/errcode"
 	"github.com/agenthub/edge-server/internal/events"
 	"github.com/agenthub/edge-server/internal/lifecycle"
@@ -30,7 +30,7 @@ func newTestServer(t *testing.T) (*Server, *store.Store) {
 	_, _ = s.CreateThread("thread_test", "proj_test", "Test Thread", "", "", "")
 
 	bus := events.NewBus(100)
-	permReg := api.NewPermissionRegistry(0)
+	permReg := permission.NewPermissionRegistry(0)
 
 	srv := NewServer(s, nil, bus, permReg)
 	srv.SetWorkspaceAllowlist([]string{t.TempDir()})
@@ -1099,7 +1099,7 @@ func TestApproveActionRequiresPermission(t *testing.T) {
 
 func TestToolApproveActionSuccessPublishesDecision(t *testing.T) {
 	srv, _ := newTestServer(t)
-	if !srv.permissionRegistry.Register(api.PendingPermission{
+	if !srv.permissionRegistry.Register(permission.PendingPermission{
 		ProjectID: "proj_test",
 		ThreadID:  "thread_test",
 		RunID:     "run_test",
diff --git a/edge-server/internal/api/permission_registry.go b/edge-server/internal/permission/permission.go
similarity index 85%
rename from edge-server/internal/api/permission_registry.go
rename to edge-server/internal/permission/permission.go
index 9a1ce74f8..8185d1fd2 100644
--- a/edge-server/internal/api/permission_registry.go
+++ b/edge-server/internal/permission/permission.go
@@ -1,4 +1,7 @@
-package api
+// Package permission tracks pending tool-permission requests for live agent
+// runs. The registry is consumed by the REST approval API and the MCP tool
+// layer without depending on either transport package.
+package permission
 
 import (
 	"strings"
@@ -47,6 +50,14 @@ func NewPermissionRegistry(ttl time.Duration) *PermissionRegistry {
 	}
 }
 
+// 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
+}
+
 func (r *PermissionRegistry) Register(permission PendingPermission) bool {
 	permission.RunID = strings.TrimSpace(permission.RunID)
 	permission.RequestID = strings.TrimSpace(permission.RequestID)
diff --git a/edge-server/internal/api/permission_registry_test.go b/edge-server/internal/permission/permission_test.go
similarity index 99%
rename from edge-server/internal/api/permission_registry_test.go
rename to edge-server/internal/permission/permission_test.go
index cd9d9470d..38e594d80 100644
--- a/edge-server/internal/api/permission_registry_test.go
+++ b/edge-server/internal/permission/permission_test.go
@@ -1,4 +1,4 @@
-package api
+package permission
 
 import (
 	"testing"