288 lines
9.9 KiB
Go
288 lines
9.9 KiB
Go
// Package handlers (chat.go) replaces internal/handlers/chat.go's HTTP
|
|
// surface. All of the orchestration that file used to do directly
|
|
// (quota check, persisting messages, running the agent loop) now
|
|
// happens inside the agent service; this file's job is just to call
|
|
// StartTurn, translate Agent's gRPC server-stream into the exact same
|
|
// SSE wire format the front end (web/static/js/app.js, chat_response/
|
|
// assistant_stream in the templates) already expects, and do the one
|
|
// thing that must stay here: render HTML (Markdown, tool-call cards,
|
|
// deliverable cards) from the plain text/structured data the agent
|
|
// service streams back. See core.go's package doc for why rendering
|
|
// lives only in this service.
|
|
package handlers
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"html/template"
|
|
"io"
|
|
"net/http"
|
|
"strconv"
|
|
"strings"
|
|
|
|
"google.golang.org/grpc/metadata"
|
|
|
|
agentv1 "enkly/gen/agent/v1"
|
|
"enkly/pkg/i18n"
|
|
"enkly/pkg/renderkit"
|
|
)
|
|
|
|
// langMetadataKey is the gRPC metadata key the gateway uses to tell
|
|
// the agent service which locale this visitor has chosen, so the
|
|
// model can be told that as a standing preference (see
|
|
// agentrun.WithPreferredLanguage) rather than only inferring it
|
|
// message-by-message. A metadata header, not a StartTurnRequest
|
|
// field: adding a proto field means regenerating every language's
|
|
// generated client/server code, and this repo's gen/ directory is
|
|
// hand-maintained without network access to protoc in some
|
|
// environments — a header costs nothing to add or remove later and
|
|
// needs no regeneration on either side. See ARCHITECTURE.md's
|
|
// "Language preference storage" note.
|
|
const langMetadataKey = "x-enkly-lang"
|
|
|
|
// withLangMetadata attaches the request's resolved locale (see
|
|
// localeFrom) to r's context as outgoing gRPC metadata, for a call to
|
|
// the agent service that should know the visitor's chosen UI
|
|
// language — currently just StartTurn, which is the one call that
|
|
// kicks off a run whose system prompt gets built.
|
|
func withLangMetadata(r *http.Request) context.Context {
|
|
return metadata.AppendToOutgoingContext(r.Context(), langMetadataKey, string(localeFrom(r)))
|
|
}
|
|
|
|
type chatErrorData struct{ Message string }
|
|
|
|
type chatResponseData struct {
|
|
UserPromptHTML template.HTML
|
|
StreamID string
|
|
CSRFToken string
|
|
}
|
|
|
|
// Chat replaces Handlers.Chat. Attachments (see attachments.go) ride
|
|
// along as repeated "attachments" form values — one per file the
|
|
// composer already uploaded in the background before this submit —
|
|
// and get folded into the exact prompt string sent to StartTurn (and
|
|
// therefore persisted as the message's content), so the agent, the
|
|
// just-sent bubble, and every future reload of this conversation all
|
|
// see the same thing.
|
|
func (h *Handlers) Chat(w http.ResponseWriter, r *http.Request) {
|
|
user := userFrom(r)
|
|
convID, _ := strconv.ParseInt(r.PostFormValue("conversation_id"), 10, 64)
|
|
prompt := strings.TrimSpace(r.PostFormValue("prompt"))
|
|
attachments := cleanAttachmentPaths(r.PostForm[attachmentFormField])
|
|
if prompt == "" && len(attachments) > 0 {
|
|
prompt = defaultAttachmentPrompt(attachments)
|
|
}
|
|
fullPrompt := appendAttachmentsBlock(prompt, attachments)
|
|
|
|
resp, err := h.Agent.StartTurn(withLangMetadata(r), &agentv1.StartTurnRequest{
|
|
UserId: user.GetId(), ConversationId: convID, Prompt: fullPrompt,
|
|
})
|
|
if err != nil {
|
|
h.Logger.Error("start turn", "err", err)
|
|
h.render(w, http.StatusInternalServerError, "chat_error", chatErrorData{Message: i18n.T(localeFrom(r), "chat.send_failed")})
|
|
return
|
|
}
|
|
if resp.GetOutcome() != agentv1.StartTurnOutcome_START_TURN_OUTCOME_OK {
|
|
status := http.StatusBadRequest
|
|
if resp.GetOutcome() == agentv1.StartTurnOutcome_START_TURN_OUTCOME_QUOTA_EXCEEDED {
|
|
status = http.StatusPaymentRequired
|
|
}
|
|
h.render(w, status, "chat_error", chatErrorData{Message: resp.GetMessage()})
|
|
return
|
|
}
|
|
|
|
h.render(w, http.StatusOK, "chat_response", chatResponseData{
|
|
UserPromptHTML: h.renderUserMessage(fullPrompt),
|
|
StreamID: resp.GetStreamId(),
|
|
CSRFToken: h.csrfToken(w, r),
|
|
})
|
|
}
|
|
|
|
// StreamChat replaces Handlers.StreamChat: opens Agent's server-
|
|
// streaming StreamTurn RPC and re-emits every event as an SSE frame.
|
|
func (h *Handlers) StreamChat(w http.ResponseWriter, r *http.Request) {
|
|
user := userFrom(r)
|
|
streamID := r.PathValue("id")
|
|
|
|
resumeFrom := 0
|
|
if id := r.Header.Get("Last-Event-ID"); id != "" {
|
|
resumeFrom, _ = strconv.Atoi(id)
|
|
}
|
|
|
|
grpcStream, err := h.Agent.StreamTurn(r.Context(), &agentv1.StreamTurnRequest{
|
|
UserId: user.GetId(), StreamId: streamID, ResumeFromToolIndex: int32(resumeFrom),
|
|
})
|
|
if err != nil {
|
|
http.NotFound(w, r)
|
|
return
|
|
}
|
|
|
|
w.Header().Set("Content-Type", "text/event-stream")
|
|
w.Header().Set("Cache-Control", "no-cache")
|
|
w.Header().Set("Connection", "keep-alive")
|
|
w.Header().Set("X-Accel-Buffering", "no")
|
|
w.WriteHeader(http.StatusOK)
|
|
flusher, _ := w.(http.Flusher)
|
|
|
|
toolIndex := resumeFrom
|
|
var finalToolCalls []*agentv1.ToolCall
|
|
// shownDeliverables tracks (by cleanWorkspacePath) every file
|
|
// already streamed live via a "deliverable" event below, so the
|
|
// "done" event's deliverablesHTML doesn't render a duplicate card
|
|
// for the same file — see liveDeliverable/remainingToolCalls.
|
|
shownDeliverables := map[string]bool{}
|
|
|
|
for {
|
|
ev, err := grpcStream.Recv()
|
|
if err == io.EOF || err != nil {
|
|
return
|
|
}
|
|
switch e := ev.GetEvent().(type) {
|
|
case *agentv1.TurnEvent_ThinkingDelta:
|
|
writeSSE(w, "thinking", renderkit.PlainStream(e.ThinkingDelta))
|
|
case *agentv1.TurnEvent_TokenDelta:
|
|
writeSSE(w, "token", renderkit.PlainStream(e.TokenDelta))
|
|
case *agentv1.TurnEvent_ToolCall:
|
|
toolIndex++
|
|
writeSSEID(w, "tool", toolIndex, h.renderToolCallOOB(streamID, e.ToolCall))
|
|
// As soon as this tool call has created or changed a
|
|
// document (create_document, edit_spreadsheet, ...),
|
|
// show its Download button right away — don't make the
|
|
// user wait for the whole turn to finish, which can come
|
|
// much later if the agent keeps working after this.
|
|
if frag, key, ok := h.liveDeliverable(r.Context(), streamID, e.ToolCall, shownDeliverables); ok {
|
|
shownDeliverables[key] = true
|
|
writeSSE(w, "deliverable", frag)
|
|
}
|
|
case *agentv1.TurnEvent_ThinkingRound:
|
|
writeSSE(w, "thinking_round", h.renderThinkingRoundOOB(streamID, e.ThinkingRound))
|
|
case *agentv1.TurnEvent_Done:
|
|
finalToolCalls = e.Done.GetToolCalls()
|
|
html := renderkit.Markdown(e.Done.GetFinalText())
|
|
html += h.deliverablesHTML(r.Context(), remainingToolCalls(finalToolCalls, shownDeliverables))
|
|
writeSSE(w, "done", html)
|
|
if flusher != nil {
|
|
flusher.Flush()
|
|
}
|
|
return
|
|
}
|
|
if flusher != nil {
|
|
flusher.Flush()
|
|
}
|
|
}
|
|
}
|
|
|
|
// StopStream replaces Handlers.StopStream.
|
|
func (h *Handlers) StopStream(w http.ResponseWriter, r *http.Request) {
|
|
user := userFrom(r)
|
|
_, err := h.Agent.StopTurn(r.Context(), &agentv1.StopTurnRequest{UserId: user.GetId(), StreamId: r.PathValue("id")})
|
|
if err != nil {
|
|
http.NotFound(w, r)
|
|
return
|
|
}
|
|
w.WriteHeader(http.StatusNoContent)
|
|
}
|
|
|
|
// ---------------------------------------------------------------------
|
|
// SSE framing + tool-card rendering
|
|
// ---------------------------------------------------------------------
|
|
|
|
func writeSSE(w http.ResponseWriter, event, data string) {
|
|
fmt.Fprintf(w, "event: %s\n", event)
|
|
for _, line := range splitLines(data) {
|
|
fmt.Fprintf(w, "data: %s\n", line)
|
|
}
|
|
fmt.Fprint(w, "\n")
|
|
}
|
|
|
|
func writeSSEID(w http.ResponseWriter, event string, id int, data string) {
|
|
fmt.Fprintf(w, "id: %d\n", id)
|
|
writeSSE(w, event, data)
|
|
}
|
|
|
|
func splitLines(s string) []string {
|
|
var out []string
|
|
start := 0
|
|
for i := 0; i < len(s); i++ {
|
|
if s[i] == '\n' {
|
|
out = append(out, s[start:i])
|
|
start = i + 1
|
|
}
|
|
}
|
|
out = append(out, s[start:])
|
|
return out
|
|
}
|
|
|
|
type toolCallView struct {
|
|
StreamID string
|
|
ToolName string
|
|
Summary string
|
|
Body template.HTML
|
|
IsError bool
|
|
}
|
|
|
|
// renderToolCallOOB replaces h.renderToolOOB: builds the
|
|
// hx-swap-oob-wrapped tool card for one completed tool call.
|
|
func (h *Handlers) renderToolCallOOB(streamID string, tc *agentv1.ToolCall) string {
|
|
return h.renderString("tool_call_oob", toolCallView{
|
|
StreamID: streamID,
|
|
ToolName: tc.GetName(),
|
|
Summary: summarizeToolCall(tc.GetName(), tc.GetArgumentsJson()),
|
|
Body: toolBodyHTML(tc.GetResult()),
|
|
IsError: tc.GetIsError(),
|
|
})
|
|
}
|
|
|
|
type thinkingRoundView struct {
|
|
StreamID string
|
|
Round int32
|
|
Thinking string
|
|
}
|
|
|
|
// renderThinkingRoundOOB builds the hx-swap-oob-wrapped card for one
|
|
// round's standalone reasoning trace — the TurnEvent_ThinkingRound
|
|
// counterpart to renderToolCallOOB. Unlike tool cards, there's no
|
|
// resume cursor for these (see server.go's StreamTurn comment on
|
|
// sentThinkRounds): a client that reconnects mid-turn will see
|
|
// already-shown round cards inserted again. Acceptable for now since
|
|
// a run has at most MaxRounds of these, each a short card, and the
|
|
// existing cumulative "thinking" delta already re-sends in full on
|
|
// every reconnect anyway.
|
|
func (h *Handlers) renderThinkingRoundOOB(streamID string, tr *agentv1.ThinkingRound) string {
|
|
return h.renderString("thinking_round_oob", thinkingRoundView{
|
|
StreamID: streamID,
|
|
Round: tr.GetRound(),
|
|
Thinking: tr.GetThinking(),
|
|
})
|
|
}
|
|
|
|
// summarizeToolCall replaces handlers.summarizeCall — a short,
|
|
// human-readable one-liner shown in the card's collapsed summary row
|
|
// (e.g. "internal/db/users.go" for a read_file call). Tries the
|
|
// arguments' most identifying field first (path, pattern, command,
|
|
// query); falls back to the raw (truncated) arguments JSON.
|
|
func summarizeToolCall(name, argsJSON string) string {
|
|
if argsJSON == "" || argsJSON == "{}" {
|
|
return ""
|
|
}
|
|
var args map[string]any
|
|
if err := json.Unmarshal([]byte(argsJSON), &args); err == nil {
|
|
for _, key := range []string{"path", "pattern", "command", "query", "task"} {
|
|
if v, ok := args[key].(string); ok && v != "" {
|
|
return truncateSummary(v)
|
|
}
|
|
}
|
|
}
|
|
return truncateSummary(argsJSON)
|
|
}
|
|
|
|
const maxSummaryChars = 80
|
|
|
|
func truncateSummary(s string) string {
|
|
if len(s) > maxSummaryChars {
|
|
return s[:maxSummaryChars] + "…"
|
|
}
|
|
return s
|
|
}
|