feat(events): publish runtime service events

Migrate hook observation to runtime events and update the process hook notification protocol. Add runtime event publication for message bus failures, channel lifecycle/outbound flow, gateway reloads, MCP server state, and MCP tool calls.

Validation: go test ./pkg/events/... ./pkg/bus ./pkg/agent ./pkg/channels ./pkg/mcp ./pkg/tools/integration ./pkg/gateway; make lint

pkg/bus/bus.go

@@ -6,6 +6,7 @@ import (
 	"sync"
 	"sync/atomic"
 
+	runtimeevents "github.com/sipeed/picoclaw/pkg/events"
 	"github.com/sipeed/picoclaw/pkg/logger"
 )
 
@@ -48,6 +49,12 @@ type MessageBus struct {
 	closed         atomic.Bool
 	wg             sync.WaitGroup
 	streamDelegate atomic.Value // stores StreamDelegate
+	eventPublisher atomic.Value // stores EventPublisher
+}
+
+// EventPublisher is the minimal runtime event publisher used by MessageBus.
+type EventPublisher interface {
+	Publish(ctx context.Context, evt runtimeevents.Event) runtimeevents.PublishResult
 }
 
 func NewMessageBus() *MessageBus {
@@ -92,9 +99,14 @@ func publish[T any](ctx context.Context, mb *MessageBus, ch chan T, msg T) error
 func (mb *MessageBus) PublishInbound(ctx context.Context, msg InboundMessage) error {
 	msg = NormalizeInboundMessage(msg)
 	if msg.Context.isZero() {
+		mb.publishFailure("inbound", runtimeScopeFromInboundContext(msg.Context), ErrMissingInboundContext)
 		return ErrMissingInboundContext
 	}
-	return publish(ctx, mb, mb.inbound, msg)
+	if err := publish(ctx, mb, mb.inbound, msg); err != nil {
+		mb.publishFailure("inbound", runtimeScopeFromInboundContext(msg.Context), err)
+		return err
+	}
+	return nil
 }
 
 func (mb *MessageBus) InboundChan() <-chan InboundMessage {
@@ -104,9 +116,14 @@ func (mb *MessageBus) InboundChan() <-chan InboundMessage {
 func (mb *MessageBus) PublishOutbound(ctx context.Context, msg OutboundMessage) error {
 	msg = NormalizeOutboundMessage(msg)
 	if msg.Context.isZero() {
+		mb.publishFailure("outbound", runtimeScopeFromInboundContext(msg.Context), ErrMissingOutboundContext)
 		return ErrMissingOutboundContext
 	}
-	return publish(ctx, mb, mb.outbound, msg)
+	if err := publish(ctx, mb, mb.outbound, msg); err != nil {
+		mb.publishFailure("outbound", runtimeScopeFromInboundContext(msg.Context), err)
+		return err
+	}
+	return nil
 }
 
 func (mb *MessageBus) OutboundChan() <-chan OutboundMessage {
@@ -116,9 +133,14 @@ func (mb *MessageBus) OutboundChan() <-chan OutboundMessage {
 func (mb *MessageBus) PublishOutboundMedia(ctx context.Context, msg OutboundMediaMessage) error {
 	msg = NormalizeOutboundMediaMessage(msg)
 	if msg.Context.isZero() {
+		mb.publishFailure("outbound_media", runtimeScopeFromInboundContext(msg.Context), ErrMissingOutboundMediaContext)
 		return ErrMissingOutboundMediaContext
 	}
-	return publish(ctx, mb, mb.outboundMedia, msg)
+	if err := publish(ctx, mb, mb.outboundMedia, msg); err != nil {
+		mb.publishFailure("outbound_media", runtimeScopeFromInboundContext(msg.Context), err)
+		return err
+	}
+	return nil
 }
 
 func (mb *MessageBus) OutboundMediaChan() <-chan OutboundMediaMessage {
@@ -126,7 +148,11 @@ func (mb *MessageBus) OutboundMediaChan() <-chan OutboundMediaMessage {
 }
 
 func (mb *MessageBus) PublishAudioChunk(ctx context.Context, chunk AudioChunk) error {
-	return publish(ctx, mb, mb.audioChunks, chunk)
+	if err := publish(ctx, mb, mb.audioChunks, chunk); err != nil {
+		mb.publishFailure("audio_chunk", runtimeScopeFromAudioChunk(chunk), err)
+		return err
+	}
+	return nil
 }
 
 func (mb *MessageBus) AudioChunksChan() <-chan AudioChunk {
@@ -134,7 +160,11 @@ func (mb *MessageBus) AudioChunksChan() <-chan AudioChunk {
 }
 
 func (mb *MessageBus) PublishVoiceControl(ctx context.Context, ctrl VoiceControl) error {
-	return publish(ctx, mb, mb.voiceControls, ctrl)
+	if err := publish(ctx, mb, mb.voiceControls, ctrl); err != nil {
+		mb.publishFailure("voice_control", runtimeScopeFromVoiceControl(ctrl), err)
+		return err
+	}
+	return nil
 }
 
 func (mb *MessageBus) VoiceControlsChan() <-chan VoiceControl {
@@ -146,6 +176,11 @@ func (mb *MessageBus) SetStreamDelegate(d StreamDelegate) {
 	mb.streamDelegate.Store(d)
 }
 
+// SetEventPublisher registers a runtime event publisher for bus errors and lifecycle events.
+func (mb *MessageBus) SetEventPublisher(p EventPublisher) {
+	mb.eventPublisher.Store(p)
+}
+
 // GetStreamer returns a Streamer for the given channel+chatID via the delegate.
 func (mb *MessageBus) GetStreamer(ctx context.Context, channel, chatID string) (Streamer, bool) {
 	if d, ok := mb.streamDelegate.Load().(StreamDelegate); ok && d != nil {
@@ -156,6 +191,7 @@ func (mb *MessageBus) GetStreamer(ctx context.Context, channel, chatID string) (
 
 func (mb *MessageBus) Close() {
 	mb.closeOnce.Do(func() {
+		mb.publishCloseEvent(runtimeevents.KindBusCloseStarted, 0)
 		// notify all blocked publishers to exit
 		close(mb.done)
 
@@ -195,6 +231,8 @@ func (mb *MessageBus) Close() {
 			logger.DebugCF("bus", "Drained buffered messages during close", map[string]any{
 				"count": drained,
 			})
+			mb.publishCloseEvent(runtimeevents.KindBusCloseDrained, drained)
 		}
+		mb.publishCloseEvent(runtimeevents.KindBusCloseCompleted, drained)
 	})
 }

pkg/bus/bus_test.go

@@ -5,6 +5,8 @@ import (
 	"sync"
 	"testing"
 	"time"
+
+	runtimeevents "github.com/sipeed/picoclaw/pkg/events"
 )
 
 func TestPublishConsume(t *testing.T) {
@@ -171,6 +173,76 @@ func TestPublishInbound_BackfillsContextFromLegacyFields(t *testing.T) {
 	}
 }
 
+func TestMessageBusPublishesRuntimeFailureAndCloseEvents(t *testing.T) {
+	eventBus := runtimeevents.NewBus()
+	defer func() {
+		if err := eventBus.Close(); err != nil {
+			t.Errorf("event bus close failed: %v", err)
+		}
+	}()
+
+	_, eventsCh, err := eventBus.Channel().OfKind(
+		runtimeevents.KindBusPublishFailed,
+		runtimeevents.KindBusCloseStarted,
+		runtimeevents.KindBusCloseDrained,
+		runtimeevents.KindBusCloseCompleted,
+	).SubscribeChan(t.Context(), runtimeevents.SubscribeOptions{Name: "bus-events", Buffer: 4})
+	if err != nil {
+		t.Fatalf("SubscribeChan failed: %v", err)
+	}
+
+	mb := NewMessageBus()
+	mb.SetEventPublisher(eventBus)
+
+	if err := mb.PublishInbound(context.Background(), InboundMessage{}); err == nil {
+		t.Fatal("expected PublishInbound to fail")
+	}
+	failed := receiveBusRuntimeEvent(t, eventsCh)
+	if failed.Kind != runtimeevents.KindBusPublishFailed ||
+		failed.Source.Name != "inbound" ||
+		failed.Severity != runtimeevents.SeverityError {
+		t.Fatalf("publish failed event = %+v", failed)
+	}
+
+	if err := mb.PublishOutbound(context.Background(), OutboundMessage{
+		Context: NewOutboundContext("telegram", "chat-1", ""),
+		Content: "queued",
+	}); err != nil {
+		t.Fatalf("PublishOutbound failed: %v", err)
+	}
+	mb.Close()
+
+	seen := map[runtimeevents.Kind]bool{}
+	for range 3 {
+		evt := receiveBusRuntimeEvent(t, eventsCh)
+		seen[evt.Kind] = true
+	}
+	for _, kind := range []runtimeevents.Kind{
+		runtimeevents.KindBusCloseStarted,
+		runtimeevents.KindBusCloseDrained,
+		runtimeevents.KindBusCloseCompleted,
+	} {
+		if !seen[kind] {
+			t.Fatalf("missing %s event, seen=%v", kind, seen)
+		}
+	}
+}
+
+func receiveBusRuntimeEvent(t *testing.T, ch <-chan runtimeevents.Event) runtimeevents.Event {
+	t.Helper()
+
+	select {
+	case evt, ok := <-ch:
+		if !ok {
+			t.Fatal("runtime event channel closed before expected event")
+		}
+		return evt
+	case <-time.After(time.Second):
+		t.Fatal("timed out waiting for runtime event")
+		return runtimeevents.Event{}
+	}
+}
+
 func TestPublishOutboundSubscribe(t *testing.T) {
 	mb := NewMessageBus()
 	defer mb.Close()

pkg/bus/events.go

@@ -0,0 +1,88 @@
+package bus
+
+import (
+	"context"
+	"time"
+
+	runtimeevents "github.com/sipeed/picoclaw/pkg/events"
+)
+
+const busEventPublishTimeout = 100 * time.Millisecond
+
+type busPublishFailedPayload struct {
+	Stream string `json:"stream"`
+	Error  string `json:"error"`
+}
+
+type busClosePayload struct {
+	Drained int `json:"drained,omitempty"`
+}
+
+func (mb *MessageBus) publishFailure(stream string, scope runtimeevents.Scope, err error) {
+	if mb == nil || err == nil {
+		return
+	}
+	publisher, ok := mb.eventPublisher.Load().(EventPublisher)
+	if !ok || publisher == nil {
+		return
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), busEventPublishTimeout)
+	defer cancel()
+	publisher.Publish(ctx, runtimeevents.Event{
+		Kind:     runtimeevents.KindBusPublishFailed,
+		Source:   runtimeevents.Source{Component: "bus", Name: stream},
+		Scope:    scope,
+		Severity: runtimeevents.SeverityError,
+		Payload: busPublishFailedPayload{
+			Stream: stream,
+			Error:  err.Error(),
+		},
+	})
+}
+
+func (mb *MessageBus) publishCloseEvent(kind runtimeevents.Kind, drained int) {
+	if mb == nil {
+		return
+	}
+	publisher, ok := mb.eventPublisher.Load().(EventPublisher)
+	if !ok || publisher == nil {
+		return
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), busEventPublishTimeout)
+	defer cancel()
+	publisher.Publish(ctx, runtimeevents.Event{
+		Kind:     kind,
+		Source:   runtimeevents.Source{Component: "bus"},
+		Severity: runtimeevents.SeverityInfo,
+		Payload:  busClosePayload{Drained: drained},
+	})
+}
+
+func runtimeScopeFromInboundContext(ctx InboundContext) runtimeevents.Scope {
+	return runtimeevents.Scope{
+		Channel:   ctx.Channel,
+		Account:   ctx.Account,
+		ChatID:    ctx.ChatID,
+		TopicID:   ctx.TopicID,
+		SpaceID:   ctx.SpaceID,
+		SpaceType: ctx.SpaceType,
+		ChatType:  ctx.ChatType,
+		SenderID:  ctx.SenderID,
+		MessageID: ctx.MessageID,
+	}
+}
+
+func runtimeScopeFromAudioChunk(chunk AudioChunk) runtimeevents.Scope {
+	return runtimeevents.Scope{
+		Channel: chunk.Channel,
+		ChatID:  chunk.ChatID,
+	}
+}
+
+func runtimeScopeFromVoiceControl(ctrl VoiceControl) runtimeevents.Scope {
+	return runtimeevents.Scope{
+		ChatID: ctrl.ChatID,
+	}
+}