feat(plugin): support observing upstream websocket response events

- Introduce `WebSocketResponseObserver` capability and bump plugin ABI schema version to 4.
- Forward upstream WebSocket response frames from Codex and xAI executors to configured observers.
- Wire `WebSocketResponseObserver` across API handlers and plugin host dispatchers.

Closes: router-for-me#5248

Author: luispater

internal/pluginhost/adapters_interceptors.go

@@ -72,6 +72,20 @@ func (h *Host) callStreamChunkInterceptor(ctx context.Context, record capability
 	return resp, true
 }
 
+func (h *Host) callWebSocketResponseObserver(ctx context.Context, record capabilityRecord, observer pluginapi.WebSocketResponseObserver, event pluginapi.WebSocketResponseEvent) {
+	if h == nil || observer == nil || h.isPluginFused(record.id) || !h.recordCurrent(record) {
+		return
+	}
+	defer func() {
+		if recovered := recover(); recovered != nil {
+			h.fusePlugin(record.id, "WebSocketResponseObserver.ObserveWebSocketResponseEvent", recovered)
+		}
+	}()
+	if errObserve := observer.ObserveWebSocketResponseEvent(ctx, event); errObserve != nil {
+		log.Warnf("pluginhost: websocket response observer %s failed: %v", record.id, errObserve)
+	}
+}
+
 func (h *Host) InterceptRequestBeforeAuth(ctx context.Context, req pluginapi.RequestInterceptRequest) pluginapi.RequestInterceptResponse {
 	return h.InterceptRequestBeforeAuthExcept(ctx, req, "")
 }
@@ -131,6 +145,47 @@ func (h *Host) CompleteRequest(ctx context.Context, completion pluginapi.Request
 	h.CompleteRequestExcept(ctx, completion, "")
 }
 
+func (h *Host) HasWebSocketResponseObservers() bool {
+	if h == nil {
+		return false
+	}
+	for _, record := range h.activeRecords() {
+		if h.isPluginFused(record.id) {
+			continue
+		}
+		if record.plugin.Capabilities.WebSocketResponseObserver != nil {
+			return true
+		}
+	}
+	return false
+}
+
+func (h *Host) ObserveWebSocketResponseEvent(ctx context.Context, event pluginapi.WebSocketResponseEvent) {
+	h.ObserveWebSocketResponseEventExcept(ctx, event, "")
+}
+
+func (h *Host) ObserveWebSocketResponseEventExcept(ctx context.Context, event pluginapi.WebSocketResponseEvent, skipPluginID string) {
+	if h == nil {
+		return
+	}
+	if ctx == nil {
+		ctx = context.Background()
+	} else {
+		ctx = context.WithoutCancel(ctx)
+	}
+	skipPluginID = strings.TrimSpace(skipPluginID)
+	for _, record := range h.activeRecords() {
+		observer := record.plugin.Capabilities.WebSocketResponseObserver
+		if h.isPluginFused(record.id) || observer == nil || record.id == skipPluginID || !h.recordCurrent(record) {
+			continue
+		}
+		next := event
+		next.Payload = bytes.Clone(event.Payload)
+		next.Metadata = cloneInterceptorMetadata(event.Metadata)
+		h.callWebSocketResponseObserver(ctx, record, observer, next)
+	}
+}
+
 // CompleteRequestExcept notifies lifecycle plugins except the plugin that initiated a nested host execution.
 func (h *Host) CompleteRequestExcept(ctx context.Context, completion pluginapi.RequestCompletion, skipPluginID string) {
 	if h == nil {

internal/pluginhost/host.go

@@ -1052,6 +1052,7 @@ func validPlugin(plugin pluginapi.Plugin) bool {
 		caps.ResponseAfterTranslator != nil ||
 		caps.ResponseInterceptor != nil ||
 		caps.StreamChunkInterceptor != nil ||
+		caps.WebSocketResponseObserver != nil ||
 		caps.ThinkingApplier != nil ||
 		caps.UsagePlugin != nil ||
 		caps.CommandLinePlugin != nil ||

internal/pluginhost/rpc_client.go

@@ -133,6 +133,9 @@ func registerRPCPlugin(ctx context.Context, host *Host, id string, client plugin
 	if resp.Capabilities.StreamChunkInterceptor {
 		plugin.Capabilities.StreamChunkInterceptor = adapter
 	}
+	if resp.Capabilities.WebSocketResponseObserver {
+		plugin.Capabilities.WebSocketResponseObserver = adapter
+	}
 	if resp.Capabilities.ThinkingApplier {
 		plugin.Capabilities.ThinkingApplier = rpcThinkingApplier{rpcPluginAdapter: adapter}
 	}
@@ -211,6 +214,9 @@ func sanitizePluginRequest(request any) any {
 	case pluginapi.StreamChunkInterceptRequest:
 		req.Metadata = sanitizePluginMetadata(req.Metadata)
 		return req
+	case pluginapi.WebSocketResponseEvent:
+		req.Metadata = sanitizePluginMetadata(req.Metadata)
+		return req
 	case rpcRequestInterceptRequest:
 		req.Metadata = sanitizePluginMetadata(req.Metadata)
 		return req
@@ -226,6 +232,9 @@ func sanitizePluginRequest(request any) any {
 	case rpcStreamChunkInterceptRequest:
 		req.Metadata = sanitizePluginMetadata(req.Metadata)
 		return req
+	case rpcWebSocketResponseEvent:
+		req.Metadata = sanitizePluginMetadata(req.Metadata)
+		return req
 	case pluginapi.ExecutorHTTPRequest:
 		req.HTTPClient = nil
 		return req
@@ -530,6 +539,16 @@ func (a *rpcPluginAdapter) InterceptStreamChunk(ctx context.Context, req plugina
 	})
 }
 
+func (a *rpcPluginAdapter) ObserveWebSocketResponseEvent(ctx context.Context, event pluginapi.WebSocketResponseEvent) error {
+	callbackID, closeCallback := a.openHostCallbackContext(ctx)
+	defer closeCallback()
+	_, errCall := callPlugin[rpcEmptyResponse](ctx, a.client, pluginabi.MethodWebSocketResponseEvent, rpcWebSocketResponseEvent{
+		WebSocketResponseEvent: event,
+		HostCallbackID:         callbackID,
+	})
+	return errCall
+}
+
 func (a rpcThinkingApplier) ApplyThinking(ctx context.Context, req pluginapi.ThinkingApplyRequest) (pluginapi.PayloadResponse, error) {
 	callbackID, closeCallback := a.openHostCallbackContext(ctx)
 	defer closeCallback()

internal/pluginhost/rpc_schema.go

@@ -39,6 +39,7 @@ type rpcCapabilities struct {
 	ResponseAfterTranslator       bool                         `json:"response_after_translator"`
 	ResponseInterceptor           bool                         `json:"response_interceptor"`
 	StreamChunkInterceptor        bool                         `json:"response_stream_interceptor"`
+	WebSocketResponseObserver     bool                         `json:"websocket_response_observer"`
 	ThinkingApplier               bool                         `json:"thinking_applier"`
 	UsagePlugin                   bool                         `json:"usage_plugin"`
 	CommandLinePlugin             bool                         `json:"command_line_plugin"`
@@ -110,6 +111,11 @@ type rpcStreamChunkInterceptRequest struct {
 	HostCallbackID string `json:"host_callback_id,omitempty"`
 }
 
+type rpcWebSocketResponseEvent struct {
+	pluginapi.WebSocketResponseEvent
+	HostCallbackID string `json:"host_callback_id,omitempty"`
+}
+
 type rpcThinkingApplyRequest struct {
 	pluginapi.ThinkingApplyRequest
 	HostCallbackID string `json:"host_callback_id,omitempty"`
@@ -150,6 +156,7 @@ func rpcCapabilitiesFromPlugin(plugin pluginapi.Plugin) rpcCapabilities {
 		ResponseAfterTranslator:       caps.ResponseAfterTranslator != nil,
 		ResponseInterceptor:           caps.ResponseInterceptor != nil,
 		StreamChunkInterceptor:        caps.StreamChunkInterceptor != nil,
+		WebSocketResponseObserver:     caps.WebSocketResponseObserver != nil,
 		ThinkingApplier:               caps.ThinkingApplier != nil,
 		UsagePlugin:                   caps.UsagePlugin != nil,
 		CommandLinePlugin:             caps.CommandLinePlugin != nil,

internal/pluginhost/websocket_observer_test.go

@@ -0,0 +1,247 @@
+package pluginhost
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"testing"
+
+	"github.com/router-for-me/CLIProxyAPI/v7/sdk/pluginabi"
+	"github.com/router-for-me/CLIProxyAPI/v7/sdk/pluginapi"
+)
+
+type testWebSocketObserverFunc func(context.Context, pluginapi.WebSocketResponseEvent) error
+
+func (f testWebSocketObserverFunc) ObserveWebSocketResponseEvent(ctx context.Context, event pluginapi.WebSocketResponseEvent) error {
+	return f(ctx, event)
+}
+
+func TestObserveWebSocketResponseEventInvokesPlugin(t *testing.T) {
+	var got pluginapi.WebSocketResponseEvent
+	called := false
+	host := newHostWithRecords(capabilityRecord{
+		id: "quota-tracker",
+		plugin: pluginapi.Plugin{
+			Capabilities: pluginapi.Capabilities{
+				WebSocketResponseObserver: testWebSocketObserverFunc(func(_ context.Context, event pluginapi.WebSocketResponseEvent) error {
+					got = event
+					called = true
+					return nil
+				}),
+			},
+		},
+	})
+
+	rawPayload := []byte(`{"type":"codex.rate_limits","rate_limits":{"primary":{"used_percent":42}}}`)
+	host.ObserveWebSocketResponseEvent(context.Background(), pluginapi.WebSocketResponseEvent{
+		RequestID:      "req-123",
+		SourceFormat:   "openai",
+		Model:          "gpt-5.3-codex",
+		RequestedModel: "gpt-5.3-codex",
+		Provider:       "codex",
+		AuthID:         "auth-abc",
+		AuthLabel:      "test-auth",
+		AuthType:       "oauth",
+		EventType:      "codex.rate_limits",
+		Payload:        rawPayload,
+	})
+
+	if !called {
+		t.Fatal("observer callback was not invoked")
+	}
+	if got.RequestID != "req-123" {
+		t.Fatalf("RequestID = %q, want req-123", got.RequestID)
+	}
+	if got.Provider != "codex" {
+		t.Fatalf("Provider = %q, want codex", got.Provider)
+	}
+	if got.AuthID != "auth-abc" || got.AuthLabel != "test-auth" {
+		t.Fatalf("Auth = (%q, %q), want (auth-abc, test-auth)", got.AuthID, got.AuthLabel)
+	}
+	if got.EventType != "codex.rate_limits" {
+		t.Fatalf("EventType = %q, want codex.rate_limits", got.EventType)
+	}
+	if !bytes.Equal(got.Payload, rawPayload) {
+		t.Fatalf("Payload = %s, want %s", got.Payload, rawPayload)
+	}
+}
+
+func TestObserveWebSocketResponseEventClonesPayloadAndMetadata(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	originalPayload := []byte(`{"type":"codex.rate_limits","rate_limits":{"primary":{"used_percent":1}}}`)
+	originalMetadata := map[string]any{"key": "value"}
+	called := false
+
+	host := newHostWithRecords(capabilityRecord{
+		id: "observer-clone",
+		plugin: pluginapi.Plugin{
+			Capabilities: pluginapi.Capabilities{
+				WebSocketResponseObserver: testWebSocketObserverFunc(func(_ context.Context, event pluginapi.WebSocketResponseEvent) error {
+					event.Payload[0] = 'X'
+					event.Metadata["key"] = "mutated"
+					called = true
+					return nil
+				}),
+			},
+		},
+	})
+
+	host.ObserveWebSocketResponseEvent(ctx, pluginapi.WebSocketResponseEvent{
+		RequestID: "req-clone",
+		Payload:   originalPayload,
+		Metadata:  originalMetadata,
+	})
+
+	if !called {
+		t.Fatal("observer callback was not invoked")
+	}
+	if originalPayload[0] != '{' {
+		t.Fatalf("original payload was mutated: %s", originalPayload)
+	}
+	if originalMetadata["key"] != "value" {
+		t.Fatalf("original metadata was mutated: %#v", originalMetadata)
+	}
+}
+
+func TestObserveWebSocketResponseEventFusesOnPanic(t *testing.T) {
+	host := newHostWithRecords(capabilityRecord{
+		id: "panicking-observer",
+		plugin: pluginapi.Plugin{
+			Capabilities: pluginapi.Capabilities{
+				WebSocketResponseObserver: testWebSocketObserverFunc(func(context.Context, pluginapi.WebSocketResponseEvent) error {
+					panic("observer panic")
+				}),
+			},
+		},
+	})
+
+	if !host.HasWebSocketResponseObservers() {
+		t.Fatal("HasWebSocketResponseObservers() = false, want true")
+	}
+
+	host.ObserveWebSocketResponseEvent(context.Background(), pluginapi.WebSocketResponseEvent{
+		RequestID: "req-panic",
+		Payload:   []byte(`{"type":"test"}`),
+	})
+
+	if !host.isPluginFused("panicking-observer") {
+		t.Fatal("isPluginFused(panicking-observer) = false, want true")
+	}
+	if host.HasWebSocketResponseObservers() {
+		t.Fatal("HasWebSocketResponseObservers() = true after fusing, want false")
+	}
+}
+
+func TestObserveWebSocketResponseEventSkipPlugin(t *testing.T) {
+	calls := 0
+	host := newHostWithRecords(capabilityRecord{
+		id: "skipped-plugin",
+		plugin: pluginapi.Plugin{
+			Capabilities: pluginapi.Capabilities{
+				WebSocketResponseObserver: testWebSocketObserverFunc(func(context.Context, pluginapi.WebSocketResponseEvent) error {
+					calls++
+					return nil
+				}),
+			},
+		},
+	})
+
+	host.ObserveWebSocketResponseEventExcept(context.Background(), pluginapi.WebSocketResponseEvent{
+		RequestID: "req-skip",
+	}, "skipped-plugin")
+
+	if calls != 0 {
+		t.Fatalf("observer calls = %d, want 0 when skipped", calls)
+	}
+}
+
+func TestRegisterRPCPluginRegistersWebSocketResponseObserver(t *testing.T) {
+	lookup := newTestSymbolLookup(&testPlugin{
+		registerResult: pluginapi.Plugin{
+			Capabilities: pluginapi.Capabilities{
+				WebSocketResponseObserver: testWebSocketObserverFunc(func(context.Context, pluginapi.WebSocketResponseEvent) error {
+					return nil
+				}),
+			},
+		},
+	})
+
+	registered, errRegister := registerRPCPlugin(context.Background(), nil, "rpc-observer", lookup, pluginabi.MethodPluginRegister, nil)
+	if errRegister != nil {
+		t.Fatalf("registerRPCPlugin() error = %v", errRegister)
+	}
+	if registered.Capabilities.WebSocketResponseObserver == nil {
+		t.Fatal("WebSocketResponseObserver = nil, want RPC adapter")
+	}
+}
+
+type rpcObserverRecordingClient struct {
+	lastMethod  string
+	lastRequest []byte
+}
+
+func (c *rpcObserverRecordingClient) Call(_ context.Context, method string, request []byte) ([]byte, error) {
+	c.lastMethod = method
+	c.lastRequest = bytes.Clone(request)
+	return json.Marshal(pluginabi.Envelope{OK: true, Result: json.RawMessage(`{}`)})
+}
+
+func (c *rpcObserverRecordingClient) Shutdown() {}
+
+func TestObserveWebSocketResponseEventRPCSanitizesMetadata(t *testing.T) {
+	client := &rpcObserverRecordingClient{}
+	adapter := &rpcPluginAdapter{
+		id:     "rpc-observer-sanitize",
+		client: client,
+	}
+
+	rawPayload := []byte(`{"type":"codex.rate_limits","rate_limits":{"primary":{"used_percent":99}}}`)
+	unserializableMetadata := map[string]any{
+		"safe_key":   "safe_value",
+		"func_field": func() {},
+		"chan_field": make(chan int),
+	}
+
+	err := adapter.ObserveWebSocketResponseEvent(context.Background(), pluginapi.WebSocketResponseEvent{
+		RequestID:      "req-rpc-sanitize",
+		SourceFormat:   "openai",
+		Model:          "gpt-5.3-codex",
+		RequestedModel: "gpt-5.3-codex",
+		Provider:       "codex",
+		AuthID:         "auth-xyz",
+		AuthLabel:      "test-auth-rpc",
+		AuthType:       "oauth",
+		EventType:      "codex.rate_limits",
+		Payload:        rawPayload,
+		Metadata:       unserializableMetadata,
+	})
+	if err != nil {
+		t.Fatalf("ObserveWebSocketResponseEvent() error = %v", err)
+	}
+
+	if client.lastMethod != pluginabi.MethodWebSocketResponseEvent {
+		t.Fatalf("lastMethod = %q, want %q", client.lastMethod, pluginabi.MethodWebSocketResponseEvent)
+	}
+
+	var decoded rpcWebSocketResponseEvent
+	if errUnmarshal := json.Unmarshal(client.lastRequest, &decoded); errUnmarshal != nil {
+		t.Fatalf("unmarshal rpc request: %v", errUnmarshal)
+	}
+
+	if decoded.RequestID != "req-rpc-sanitize" {
+		t.Fatalf("decoded RequestID = %q, want req-rpc-sanitize", decoded.RequestID)
+	}
+	if decoded.EventType != "codex.rate_limits" {
+		t.Fatalf("decoded EventType = %q, want codex.rate_limits", decoded.EventType)
+	}
+	if decoded.Metadata["safe_key"] != "safe_value" {
+		t.Fatalf("decoded safe_key = %v, want safe_value", decoded.Metadata["safe_key"])
+	}
+	if _, exists := decoded.Metadata["func_field"]; exists {
+		t.Fatalf("func_field was not sanitized: %#v", decoded.Metadata)
+	}
+	if _, exists := decoded.Metadata["chan_field"]; exists {
+		t.Fatalf("chan_field was not sanitized: %#v", decoded.Metadata)
+	}
+}