Cyrene_v2/internal/services/gateway/ws_hub_turn.go

package gateway

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"encoding/json"
	"strings"
	"time"

	"git.yeij.top/AskaEth/Cyrene_v2/internal/bus"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
)

// relay is the bus handler for the ws.> pattern and the producer bridge
// patterns. The ws.chat subject is a streaming trigger (run the loop and emit
// chunks); a ws.* subject becomes an outbound event of its design type; a
// bridged producer subject (device_update, task output/done, automation result,
// reminder, subagent progress) is normalized into the matching ws.* event type.
// Events with no WS-visible mapping are dropped so internal agent-to-agent
// traffic never leaks to clients.
func (h *hub) relay(ctx context.Context, e bus.Event) error {
	if e.Subject == wsChatSubject {
		go h.streamChat(ctx, e.Payload)
		return nil
	}
	if e.Subject == wsStreamSubject {
		// OnStream→bus→relay path: the loop already published a structured
		// model.StreamEvent; map it to the typed downstream event.
		h.broadcastStream(e.Payload)
		return nil
	}
	ev, ok := normalizeEvent(e)
	if !ok {
		return nil
	}
	h.broadcast(ev)
	return nil
}

// streamChat is the ws.chat trigger entry point. It decodes the request, and
// routes the turn through the session orchestrator when one is wired (the
// non-blocking busy/pending/resume path of design/15); otherwise it falls back to
// the legacy direct run so a hub built without an orchestrator keeps the exact
// stream_chunk/stream_end contract. It runs on its own goroutine so a slow model
// call does not stall the bus subscription worker.
func (h *hub) streamChat(ctx context.Context, payload any) {
	var req chatStreamRequest
	if b, err := json.Marshal(payload); err == nil {
		_ = json.Unmarshal(b, &req)
	}
	if h.runner == nil || (strings.TrimSpace(req.Text) == "" && strings.TrimSpace(req.Image) == "") {
		h.broadcast(wsEvent{Type: "stream_end", TurnID: newTurnID(), Seq: int(h.streamSeq.Add(1)), SessionID: req.SessionID, Data: map[string]any{"stopReason": "idle"}})
		return
	}
	if req.SessionID == "" {
		req.SessionID = "web"
	}
	// The Web main session is the owner's private conversation (JWT = owner), so
	// it binds the full-access env: memory read+write, unchanged behavior.
	msgs := userChatMessages(req.Text, req.Image)
	// Thread the client's deviceID and behavior mode into the orchestrator via ctx
	// so both the persisted user turn and the assistant reply are tagged
	// (design/16 §4.5, §5.1).
	ctx = withMode(withDeviceID(ctx, req.DeviceID), req.Mode)
	if h.orchestrator != nil {
		_ = h.orchestrator.Submit(ctx, req.SessionID, msgs, loop.OwnerEnv())
		return
	}
	_, _ = h.runTurn(ctx, req.SessionID, msgs, loop.OwnerEnv())
}

// runTurn executes one agent turn and broadcasts its output stream. It is the
// unit both the orchestrator (via its injected TurnRunner) and the legacy
// fallback path share, so a turn always emits the same stream_chunk/stream_end
// events regardless of who drives it. When the loop is streaming (OnStream wired)
// the structured deltas already fan out as message.*/thinking.*/tool_call.*
// events through the OnStream→bus→relay path, so only the non-streaming path
// emits a single whole-text chunk to hold the stream contract.
func (h *hub) runTurn(ctx context.Context, sessionID string, msgs []model.Message, env loop.SessionEnv) (*loop.Result, error) {
	turnID := newTurnID()
	// Thread the turn identity into the run context so the loop stamps it on every
	// streamed event it emits (message.*/thinking.*/tool_call.*); the hub then puts
	// it on the outbound frame. Every turn takes a fresh id — a session has at most
	// one in-flight turn — so a client can correlate one burst of deltas to one
	// assistant turn (design/16 §4.2).
	ctx = loop.WithTurnID(ctx, turnID)
	if h.runner == nil || len(msgs) == 0 {
		h.broadcast(wsEvent{Type: "stream_end", TurnID: turnID, Seq: int(h.streamSeq.Add(1)), SessionID: sessionID, Data: map[string]any{"stopReason": "idle"}})
		return &loop.Result{StopReason: "idle"}, nil
	}
	res, err := h.runner.Run(loop.WithSessionEnv(ctx, env), sessionID, msgs)
	if err != nil {
		h.log.Error("chat run", "err", err, "session", sessionID)
		h.broadcast(wsEvent{Type: "stream_end", TurnID: turnID, Seq: int(h.streamSeq.Add(1)), SessionID: sessionID, Data: map[string]any{"stopReason": "error", "error": err.Error()}})
		return nil, err
	}
	if !h.streaming {
		h.broadcast(wsEvent{Type: "stream_chunk", TurnID: turnID, Seq: int(h.streamSeq.Add(1)), SessionID: sessionID, Data: map[string]any{"delta": res.Text}})
	}
	h.broadcast(wsEvent{Type: "stream_end", TurnID: turnID, Seq: int(h.streamSeq.Add(1)), SessionID: sessionID, Data: map[string]any{"stopReason": res.StopReason, "steps": res.Steps}})
	return res, nil
}

// handleInterrupt routes a client "interrupt" frame (design/15 §3) to the
// orchestrator, canceling the in-flight turn and clearing the pending queue for
// the requested session. It is a no-op when no orchestrator is wired.
func (h *hub) handleInterrupt(payload json.RawMessage) {
	if h.orchestrator == nil {
		return
	}
	var req struct {
		SessionID string `json:"sessionId"`
	}
	_ = json.Unmarshal(payload, &req)
	if req.SessionID == "" {
		return
	}
	h.orchestrator.Interrupt(req.SessionID)
}

// broadcastStream maps a streamed model.StreamEvent to its downstream WS event
// type and fans it out to every client. It runs on the bus subscription worker;
// seq is assigned atomically so a burst stays ordered even if the loop emits
// events from more than one goroutine. SessionID/TurnID ride on the event itself
// (the loop stamps them at the turn boundary from the run context), so the frame
// carries the scope a client needs to route it to the active session and to one
// assistant turn.
func (h *hub) broadcastStream(payload any) {
	ev, ok := payload.(model.StreamEvent)
	if !ok {
		return
	}
	typ, data := streamEventData(ev)
	h.broadcast(wsEvent{Type: typ, SessionID: ev.SessionID, TurnID: ev.TurnID, Seq: int(h.streamSeq.Add(1)), Data: data})
}

// chatStreamRequest is the decoded ws.chat payload.
type chatStreamRequest struct {
	SessionID string `json:"sessionId"`
	Text      string `json:"text"`
	Image     string `json:"image,omitempty"`
	// DeviceID is the client-produced identifier of the sending browser/device
	// (design/16 §4.5), e.g. "web:chrome:a1b2c3"; it is optional and persisted
	// alongside the message so the agent knows which end the owner is chatting from.
	DeviceID string `json:"deviceId,omitempty"`
	// Mode is the behavior mode the client sent the turn in (design/01 §3,
	// design/16 §5.1); it is optional and persisted per message, falling back to
	// "chat" when omitted.
	Mode string `json:"mode,omitempty"`
}

// newTurnID returns a fresh random hex identifier used to correlate a burst of
// streamed WS events to one assistant turn. It burns the crypto/rand source and
// falls back to a nanosecond timestamp if the source is unavailable, so a id is
// always produced (a blank turnId would break client correlation silently).
func newTurnID() string {
	var b [8]byte
	if _, err := rand.Read(b[:]); err == nil {
		return "turn-" + hex.EncodeToString(b[:])
	}
	return "turn-" + time.Now().UTC().Format("20060102150405.000000000")
}

// publishChatStream publishes a ws.chat streaming trigger (REST stream mode).
// deviceID is the sending browser/device id and mode the behavior mode the client
// sent the turn in, both carried on the payload so the ws.chat consumer can tag
// the persisted turn (design/16 §4.5, §5.1).
func (s *Server) publishChatStream(ctx context.Context, sessionID, text, image, deviceID, mode string) {
	if s.bus == nil {
		return
	}
	_ = s.bus.Publish(ctx, wsChatSubject, map[string]any{"sessionId": sessionID, "text": text, "image": image, "deviceId": deviceID, "mode": mode})
}