backend/internal/server/handlers_chat.go

package server

import (
	"encoding/json"
	"errors"
	"fmt"
	"net/http"

	"github.com/gin-gonic/gin"

	"code.littlelan.cn/CarrotAssistant/backend/internal/agent"
	"code.littlelan.cn/CarrotAssistant/backend/internal/llm"
	"code.littlelan.cn/CarrotAssistant/backend/internal/skill"
	"code.littlelan.cn/CarrotAssistant/backend/internal/store"
)

type chatReq struct {
	SessionID string `json:"session_id,omitempty"`
	Message   string `json:"message" binding:"required"`
}

// POST /api/v1/chat — opens an SSE stream that runs one assistant turn.
//
// Flow: authenticate app -> load/create session -> load published skill ->
// build history -> run agent, streaming events -> persist the assistant's
// final message + any tool messages. If a mutating tool needs confirmation,
// the pending call is stored on the session and a confirmation_required
// event ends this stream; the user then POSTs /chat/confirm to resume.
func (d Deps) handleChat(c *gin.Context) {
	app := appCtx(c)
	var req chatReq
	if !bind(c, &req) {
		return
	}

	// Resolve skill + session up front so failures return clean JSON, not a
	// half-started SSE stream.
	_, sk, err := d.loadPublishedSkill(app)
	if err != nil {
		c.JSON(http.StatusConflict, gin.H{"error": err.Error()})
		return
	}
	if _, err := d.resolveLLMConfig(app); err != nil {
		c.JSON(http.StatusConflict, gin.H{"error": err.Error()})
		return
	}
	var session *store.Session
	if req.SessionID != "" {
		session, err = d.loadSessionOwnedBy(req.SessionID, app.ID)
		if err != nil {
			c.JSON(http.StatusNotFound, gin.H{"error": "会话不存在"})
			return
		}
		if len(session.PendingConfirmation) > 0 {
			c.JSON(http.StatusConflict, gin.H{"error": "该会话有一个待确认的操作,请先确认或拒绝"})
			return
		}
	} else {
		title := truncate(req.Message, 40)
		session, err = d.createSession(app.ID, title)
		if err != nil {
			c.JSON(http.StatusInternalServerError, gin.H{"error": "创建会话失败"})
			return
		}
	}

	// Persist the user message before running, so it survives a mid-stream
	// disconnect.
	if err := d.persistMessage(session.ID, llm.Message{Role: llm.RoleUser, Content: req.Message}); err != nil {
		c.JSON(http.StatusInternalServerError, gin.H{"error": "保存消息失败"})
		return
	}

	history, err := d.sessionHistory(session.ID, sk)
	if err != nil {
		c.JSON(http.StatusInternalServerError, gin.H{"error": "加载历史失败"})
		return
	}

	// Begin SSE. From here on, all output goes through the stream.
	d.streamChat(c, app, session, sk, history, false)
}

// streamChat writes SSE events for one agent run. On confirmation pause it
// stores the pending call and emits confirmation_required; the caller's
// confirm handler resumes by calling this again with resume=true.
func (d Deps) streamChat(c *gin.Context, app *store.App, session *store.Session, sk *skill.Skill, history []llm.Message, resume bool) {
	flusher, ok := c.Writer.(interface {
		http.ResponseWriter
		http.Flusher
	})
	if !ok {
		c.JSON(http.StatusInternalServerError, gin.H{"error": "streaming not supported"})
		return
	}
	c.Writer.Header().Set("Content-Type", "text/event-stream")
	c.Writer.Header().Set("Cache-Control", "no-cache")
	c.Writer.Header().Set("Connection", "keep-alive")
	c.Writer.Header().Set("X-Accel-Buffering", "no") // disable nginx buffering

	send := func(event string, data any) {
		payload, _ := json.Marshal(data)
		fmt.Fprintf(c.Writer, "event: %s\ndata: %s\n\n", event, payload)
		flusher.Flush()
	}
	sendSession := func() {
		send("session", gin.H{"session_id": session.ID})
	}

	var sawDone bool
	onEvent := func(ev agent.Event) {
		switch ev.Kind {
		case agent.KindToken:
			send("token", gin.H{"delta": ev.Text})
		case agent.KindToolCall:
			send("tool_call", ev.ToolCall)
		case agent.KindToolResult:
			send("tool_result", ev.ToolResult)
		case agent.KindConfirmation:
			// Persist the pending call so /chat/confirm can resume. The
			// runner has already stashed its internal state; we also keep a
			// copy on the session for crash-safety across requests.
			pending := pendingBlob{
				CallID:    ev.Pending.CallID,
				Name:      ev.Pending.Name,
				Arguments: ev.Pending.Arguments,
				Detail:    ev.Pending.Detail,
			}
			d.DB.Model(&store.Session{}).Where("id = ?", session.ID).
				Update("pending_confirmation", jsonRaw(pending))
			send("confirmation_required", ev.Pending)
		case agent.KindDone:
			sawDone = true
		case agent.KindError:
			send("error", gin.H{"message": ev.Message})
		}
	}

	sendSession()
	runner := agent.NewRunner(llm.New(mustLLMConfig(d, app)), sk, history, onEvent)

	var runErr error
	if resume {
		// Confirm handler already validated approval; run the resume path.
		runErr = runner.ResumeAfterConfirmation(c.Request.Context(), true)
	} else {
		// The user message is already the last history entry.
		runErr = runner.Run(c.Request.Context(), "")
	}

	if errors.Is(runErr, agent.ErrPendingConfirmation) {
		// Stream ends here; client must POST /chat/confirm to continue.
		send("paused", gin.H{"reason": "confirmation_required"})
		return
	}
	if runErr != nil && !errors.Is(runErr, agent.ErrTooManyRounds) {
		// Non-fatal errors already surfaced via KindError; just close.
		send("done", gin.H{"ok": false})
		return
	}

	// Persist the assistant's final message (and any tool messages that the
	// runner accumulated). We re-derive them from the runner's history tail
	// since the runner owns the authoritative copy.
	d.persistRunnerTurn(session.ID, runner)

	if sawDone {
		send("done", gin.H{"ok": true})
	} else {
		send("done", gin.H{"ok": false})
	}
}

// persistRunnerTurn stores any new messages the runner added beyond the
// initial history length. The runner's History is the full conversation
// including the system prompt (index 0), which we skip when persisting.
func (d Deps) persistRunnerTurn(sessionID string, r *agent.Runner) {
	// Best-effort: failures here don't fail the stream, but the next turn
	// would lose context. Log via audit.
	// We rely on the runner exposing its history; agent.Runner.History is
	// public for this purpose.
	for _, m := range r.HistoryTail() {
		if m.Role == llm.RoleSystem {
			continue
		}
		if err := d.persistMessage(sessionID, m); err != nil {
			d.auditBy("system", nil, &sessionID, "session.persist_error",
				map[string]any{"role": m.Role, "err": err.Error()})
		}
	}
}

// mustLLMConfig resolves the LLM config, sending a JSON error and aborting if
// it fails. Used at the start of streamChat where we already began SSE setup
// in the caller — but resolution happens before SSE begins in practice (see
// handleChat). This helper is a safety net.
func mustLLMConfig(d Deps, app *store.App) llm.Config {
	cfg, err := d.resolveLLMConfig(app)
	if err != nil {
		// Returning a zero config would produce a clearer API error from the
		// SDK than a panic; the chat will fail loudly.
		return llm.Config{}
	}
	_ = err
	return cfg
}

// truncate clamps a string to n runes for session titles.
func truncate(s string, n int) string {
	r := []rune(s)
	if len(r) <= n {
		return s
	}
	return string(r[:n]) + "…"
}

// pendingBlob is the JSON shape stored on Session.PendingConfirmation.
type pendingBlob struct {
	CallID    string         `json:"call_id"`
	Name      string         `json:"name"`
	Arguments map[string]any `json:"arguments"`
	Detail    string         `json:"detail"`
}

// confirmReq is the body for POST /api/v1/chat/confirm.
type confirmReq struct {
	SessionID string `json:"session_id" binding:"required"`
	Approved  bool   `json:"approved"`
}

// POST /api/v1/chat/confirm — resolves a paused confirmation. On approve,
// opens a new SSE stream that resumes the agent loop. On deny, the pending
// call is cleared and a tool "user declined" message is appended.
func (d Deps) handleConfirm(c *gin.Context) {
	app := appCtx(c)
	var req confirmReq
	if !bind(c, &req) {
		return
	}
	session, err := d.loadSessionOwnedBy(req.SessionID, app.ID)
	if err != nil {
		c.JSON(http.StatusNotFound, gin.H{"error": "会话不存在"})
		return
	}
	if len(session.PendingConfirmation) == 0 {
		c.JSON(http.StatusConflict, gin.H{"error": "该会话没有待确认的操作"})
		return
	}
	_, sk, err := d.loadPublishedSkill(app)
	if err != nil {
		c.JSON(http.StatusConflict, gin.H{"error": err.Error()})
		return
	}

	if !req.Approved {
		// Deny: append a synthetic tool message so the model can answer
		// without the data, then clear the pending flag. We still stream so
		// the client gets the model's follow-up.
		var pb pendingBlob
		_ = jsonUnmarshal(session.PendingConfirmation, &pb)
		d.DB.Model(&store.Session{}).Where("id = ?", session.ID).
			Update("pending_confirmation", nil)
		d.persistMessage(session.ID, llm.Message{
			Role: llm.RoleTool, ToolCallID: pb.CallID, Name: pb.Name,
			Content: "[用户拒绝了该操作]",
		})
	}

	history, err := d.sessionHistory(session.ID, sk)
	if err != nil {
		c.JSON(http.StatusInternalServerError, gin.H{"error": "加载历史失败"})
		return
	}

	// Clear the pending flag now (on approve, resume handles the call).
	if req.Approved {
		d.DB.Model(&store.Session{}).Where("id = ?", session.ID).
			Update("pending_confirmation", nil)
	}

	// For "deny" we need a fresh runner to continue the loop with the new
	// tool message; for "approve" the resume path runs the pending call.
	// Both flow through streamChat: approve with resume=true, deny with a
	// normal Run (the deny message is already in history).
	d.streamChatResume(c, app, session, sk, history, req.Approved)
}

// streamChatResume mirrors streamChat but handles the deny case (a fresh
// Run) versus the approve case (ResumeAfterConfirmation). Kept separate so
// the main handleChat path stays readable.
func (d Deps) streamChatResume(c *gin.Context, app *store.App, session *store.Session, sk *skill.Skill, history []llm.Message, approve bool) {
	flusher, ok := c.Writer.(interface {
		http.ResponseWriter
		http.Flusher
	})
	if !ok {
		c.JSON(http.StatusInternalServerError, gin.H{"error": "streaming not supported"})
		return
	}
	c.Writer.Header().Set("Content-Type", "text/event-stream")
	c.Writer.Header().Set("Cache-Control", "no-cache")
	c.Writer.Header().Set("Connection", "keep-alive")
	c.Writer.Header().Set("X-Accel-Buffering", "no")

	send := func(event string, data any) {
		payload, _ := json.Marshal(data)
		fmt.Fprintf(c.Writer, "event: %s\ndata: %s\n\n", event, payload)
		flusher.Flush()
	}
	send("session", gin.H{"session_id": session.ID})

	var sawDone bool
	onEvent := func(ev agent.Event) {
		switch ev.Kind {
		case agent.KindToken:
			send("token", gin.H{"delta": ev.Text})
		case agent.KindToolCall:
			send("tool_call", ev.ToolCall)
		case agent.KindToolResult:
			send("tool_result", ev.ToolResult)
		case agent.KindConfirmation:
			pending := pendingBlob{
				CallID: ev.Pending.CallID, Name: ev.Pending.Name,
				Arguments: ev.Pending.Arguments, Detail: ev.Pending.Detail,
			}
			d.DB.Model(&store.Session{}).Where("id = ?", session.ID).
				Update("pending_confirmation", jsonRaw(pending))
			send("confirmation_required", ev.Pending)
		case agent.KindDone:
			sawDone = true
		case agent.KindError:
			send("error", gin.H{"message": ev.Message})
		}
	}

	cfg, err := d.resolveLLMConfig(app)
	if err != nil {
		send("error", gin.H{"message": err.Error()})
		send("done", gin.H{"ok": false})
		return
	}
	runner := agent.NewRunner(llm.New(cfg), sk, history, onEvent)

	var runErr error
	if approve {
		runErr = runner.ResumeAfterConfirmation(c.Request.Context(), true)
	} else {
		// The deny tool message is the last history entry; run with empty
		// user message to continue.
		runErr = runner.Run(c.Request.Context(), "")
	}

	if errors.Is(runErr, agent.ErrPendingConfirmation) {
		send("paused", gin.H{"reason": "confirmation_required"})
		return
	}
	d.persistRunnerTurn(session.ID, runner)
	if sawDone {
		send("done", gin.H{"ok": true})
	} else {
		send("done", gin.H{"ok": false})
	}
}