esengine--deepseek-reasonix/desktop/bot_event_sink.go

package main

import (
	"context"
	"log"
	"strings"
	"sync"
	"time"

	"reasonix/internal/bot"
	"reasonix/internal/event"
)

const botForwardSendTimeout = 30 * time.Second
const botForwardQueueSize = 64

// ── Forward target ──────────────────────────────────────────────────────────

// botForwardTarget identifies one remote chat to send forwarded events to.
type botForwardTarget struct {
	ConnID   string
	Domain   string
	ChatID   string
	ChatType bot.ChatType
}

// ── Event forwarder ─────────────────────────────────────────────────────────

// botEventForwarder implements event.Sink and forwards relevant events to
// connected bot channels through the desktopBotRuntime. It is attached to a
// tabEventSink when a heartbeat task should push AI output to IM channels.
//
// It accumulates Text events and sends them as complete messages on TurnDone
// (and occasionally during generation when the buffer grows large enough), so
// the remote side sees progressive streaming output rather than one big blob.
type botEventForwarder struct {
	runtime *desktopBotRuntime
	targets []botForwardTarget

	mu        sync.Mutex
	buf       strings.Builder
	queueMu   sync.Mutex
	queue     chan string
	closed    bool
	closeOnce sync.Once
}

// newBotEventForwarder creates a forwarder that sends to all given targets.
// runtime may be nil — Emit calls are then no-ops.
func newBotEventForwarder(runtime *desktopBotRuntime, targets []botForwardTarget) *botEventForwarder {
	f := &botEventForwarder{
		runtime: runtime,
		targets: targets,
		queue:   make(chan string, botForwardQueueSize),
	}
	go f.run()
	return f
}

// Emit implements event.Sink. It forwards text and lifecycle events to the
// connected bot channels; reasoning, tool dispatch, and other internal events
// are dropped to avoid noisy IM output.
func (f *botEventForwarder) Emit(e event.Event) {
	if f.runtime == nil || len(f.targets) == 0 {
		return
	}
	switch e.Kind {
	case event.TurnStarted:
		f.mu.Lock()
		f.buf.Reset()
		f.mu.Unlock()

	case event.Text:
		f.mu.Lock()
		f.buf.WriteString(e.Text)
		size := f.buf.Len()
		f.mu.Unlock()
		// Flush opportunistically when the buffer crosses a threshold, so long
		// streams (e.g. "tell me three jokes") produce multiple messages.
		if size >= 400 {
			f.flush()
		}

	case event.TurnDone:
		f.flush()
		f.Close()

	case event.ApprovalRequest:
		// The heartbeat turn belongs to the desktop tab controller, not the bot
		// gateway session, so remote /approve replies cannot satisfy this ID.
		text := "⚠️ 需要在 Reasonix 桌面端批准操作: " + e.Approval.Tool + " — " + e.Approval.Subject
		text += "\n请回到桌面窗口处理。"
		f.sendToAll(text)

	case event.AskRequest:
		var qb strings.Builder
		qb.WriteString("❓ 需要在 Reasonix 桌面端回答问题:\n")
		for i, q := range e.Ask.Questions {
			if i > 0 {
				qb.WriteString("\n")
			}
			qb.WriteString(q.Prompt)
		}
		qb.WriteString("\n请回到桌面窗口处理。")
		f.sendToAll(qb.String())

	case event.Notice:
		if e.Level == event.LevelWarn {
			f.sendToAll("⚠️ " + e.Text)
		}

	case event.CompactionStarted:
		f.sendToAll("🔄 正在压缩上下文...")
	}
}

// flush sends the accumulated buffer as one message per target channel.
func (f *botEventForwarder) flush() {
	f.mu.Lock()
	text := strings.TrimSpace(f.buf.String())
	if text == "" {
		f.mu.Unlock()
		return
	}
	f.buf.Reset()
	f.mu.Unlock()

	f.sendToAll(text)
}

// sendToAll dispatches text to every target channel. Errors are logged and
// non-fatal; a failed target does not block other targets.
func (f *botEventForwarder) sendToAll(text string) {
	text = strings.TrimSpace(text)
	if f.runtime == nil || len(f.targets) == 0 || text == "" {
		return
	}
	f.queueMu.Lock()
	defer f.queueMu.Unlock()
	if f.closed {
		return
	}
	select {
	case f.queue <- text:
	default:
		log.Printf("[bot-forward] send queue full; dropping message for %d target(s)", len(f.targets))
	}
}

func (f *botEventForwarder) run() {
	for text := range f.queue {
		f.sendToAllNow(text)
	}
}

func (f *botEventForwarder) sendToAllNow(text string) {
	for _, tgt := range f.targets {
		ctx, cancel := context.WithTimeout(context.Background(), botForwardSendTimeout)
		_, err := f.runtime.SendToAdapter(ctx, tgt.ConnID, tgt.Domain, bot.OutboundMessage{
			ChatID:   tgt.ChatID,
			ChatType: tgt.ChatType,
			Text:     text,
		})
		cancel()
		if err != nil {
			log.Printf("[bot-forward] send to %s/%s failed: %v", tgt.ConnID, tgt.ChatType, err)
		}
	}
}

func (f *botEventForwarder) Close() {
	f.closeOnce.Do(func() {
		f.flush()
		f.queueMu.Lock()
		f.closed = true
		close(f.queue)
		f.queueMu.Unlock()
	})
}