mirror of
https://gh-proxy.org/https://github.com/klinsmaya/multica.git
synced 2026-11-08 17:36:45 +08:00
* chore: clean historical comments (MUL-8034) Co-authored-by: multica-agent <[email protected]> * docs: preserve pending Composio expiry contract Co-authored-by: multica-agent <[email protected]> --------- Co-authored-by: multica-agent <[email protected]>
2111 lines
80 KiB
Go
2111 lines
80 KiB
Go
package handler
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"log/slog"
|
|
"net/http"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/go-chi/chi/v5"
|
|
"github.com/jackc/pgx/v5"
|
|
"github.com/jackc/pgx/v5/pgtype"
|
|
"github.com/multica-ai/multica/server/internal/analytics"
|
|
"github.com/multica-ai/multica/server/internal/events"
|
|
"github.com/multica-ai/multica/server/internal/logger"
|
|
obsmetrics "github.com/multica-ai/multica/server/internal/metrics"
|
|
"github.com/multica-ai/multica/server/internal/middleware"
|
|
"github.com/multica-ai/multica/server/internal/service"
|
|
"github.com/multica-ai/multica/server/internal/util"
|
|
db "github.com/multica-ai/multica/server/pkg/db/generated"
|
|
"github.com/multica-ai/multica/server/pkg/dbid"
|
|
"github.com/multica-ai/multica/server/pkg/protocol"
|
|
)
|
|
|
|
// chatSessionTitleMaxLen caps the rename input. Long enough to fit a
|
|
// meaningful summary, short enough to keep the dropdown row scannable.
|
|
const chatSessionTitleMaxLen = 200
|
|
|
|
// Chat Sessions
|
|
|
|
type CreateChatSessionRequest struct {
|
|
AgentID string `json:"agent_id"`
|
|
Title string `json:"title"`
|
|
ProjectID *string `json:"project_id"`
|
|
}
|
|
|
|
func (h *Handler) CreateChatSession(w http.ResponseWriter, r *http.Request) {
|
|
userID, ok := requireUserID(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
workspaceID := ctxWorkspaceID(r.Context())
|
|
|
|
var req CreateChatSessionRequest
|
|
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid request body")
|
|
return
|
|
}
|
|
if req.AgentID == "" {
|
|
writeError(w, http.StatusBadRequest, "agent_id is required")
|
|
return
|
|
}
|
|
agentID, ok := parseUUIDOrBadRequest(w, req.AgentID, "agent_id")
|
|
if !ok {
|
|
return
|
|
}
|
|
workspaceUUID, ok := parseUUIDOrBadRequest(w, workspaceID, "workspace id")
|
|
if !ok {
|
|
return
|
|
}
|
|
projectID := pgtype.UUID{Valid: false}
|
|
if req.ProjectID != nil && strings.TrimSpace(*req.ProjectID) != "" {
|
|
projectID, ok = parseUUIDOrBadRequest(w, strings.TrimSpace(*req.ProjectID), "project_id")
|
|
if !ok {
|
|
return
|
|
}
|
|
}
|
|
|
|
// Verify agent exists in workspace.
|
|
agent, err := h.Queries.GetAgentInWorkspace(r.Context(), db.GetAgentInWorkspaceParams{
|
|
ID: agentID,
|
|
WorkspaceID: workspaceUUID,
|
|
})
|
|
if err != nil {
|
|
writeError(w, http.StatusNotFound, "agent not found")
|
|
return
|
|
}
|
|
if agent.ArchivedAt.Valid {
|
|
writeError(w, http.StatusBadRequest, "agent is archived")
|
|
return
|
|
}
|
|
// Invocation gate: starting a chat produces agent runs, so it uses the
|
|
// invoke permission (MUL-3963), not the softer view gate. Agent-to-agent
|
|
// chat sessions are judged by the top-of-chain originator.
|
|
actorType, actorID := h.resolveActor(r, userID, workspaceID)
|
|
if !h.canInvokeAgent(r.Context(), agent, actorType, actorID, h.invokeOriginatorFromRequest(r, actorType, actorID), workspaceID) {
|
|
writeError(w, http.StatusForbidden, "you do not have access to this agent")
|
|
return
|
|
}
|
|
|
|
// Create inside a tx that first takes a FOR KEY SHARE lock on the workspace
|
|
// row: it conflicts with DeleteWorkspace's FOR UPDATE, so a session cannot be
|
|
// created into a workspace whose delete is in progress and then be orphaned by
|
|
// the finalizer after the delete's sweep (#5219 create/delete protocol; see
|
|
// LockWorkspaceForChatSessionCreate).
|
|
tx, err := h.TxStarter.Begin(r.Context())
|
|
if err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to start transaction")
|
|
return
|
|
}
|
|
defer tx.Rollback(r.Context())
|
|
qtx := h.Queries.WithTx(tx)
|
|
|
|
if _, err := qtx.LockWorkspaceForChatSessionCreate(r.Context(), workspaceUUID); err != nil {
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
writeError(w, http.StatusNotFound, "workspace not found")
|
|
return
|
|
}
|
|
writeError(w, http.StatusInternalServerError, "failed to lock workspace")
|
|
return
|
|
}
|
|
if projectID.Valid {
|
|
if _, err := qtx.LockProjectForChatSessionCreate(r.Context(), db.LockProjectForChatSessionCreateParams{
|
|
ID: projectID,
|
|
WorkspaceID: workspaceUUID,
|
|
}); err != nil {
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
writeError(w, http.StatusNotFound, "project not found")
|
|
return
|
|
}
|
|
writeError(w, http.StatusInternalServerError, "failed to lock project")
|
|
return
|
|
}
|
|
}
|
|
|
|
session, err := qtx.CreateChatSession(r.Context(), db.CreateChatSessionParams{
|
|
ID: dbid.NewV7(),
|
|
WorkspaceID: workspaceUUID,
|
|
AgentID: agentID,
|
|
CreatorID: parseUUID(userID),
|
|
Title: req.Title,
|
|
ProjectID: projectID,
|
|
})
|
|
if err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to create chat session")
|
|
return
|
|
}
|
|
session, err = qtx.MarkChatSessionExplicitlyCreated(r.Context(), session.ID)
|
|
if err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to mark chat session explicit")
|
|
return
|
|
}
|
|
|
|
if err := tx.Commit(r.Context()); err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to commit chat session create")
|
|
return
|
|
}
|
|
|
|
writeJSON(w, http.StatusCreated, chatSessionToResponse(session))
|
|
}
|
|
|
|
func (h *Handler) ListChatSessions(w http.ResponseWriter, r *http.Request) {
|
|
userID, ok := requireUserID(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
workspaceID := ctxWorkspaceID(r.Context())
|
|
|
|
// Compute the accessible-agents set once and use it to drop sessions
|
|
// whose target agent the caller no longer has access to — without this,
|
|
// a member whose role was downgraded would still see the session list
|
|
// (and transcripts via ListChatMessages) for any private agent they
|
|
// previously had access to. Falls back to the user's role from the
|
|
// workspace member context.
|
|
member, ok := h.workspaceMember(w, r, workspaceID)
|
|
if !ok {
|
|
return
|
|
}
|
|
actorType, actorID := h.resolveActor(r, userID, workspaceID)
|
|
allowed, ok := h.accessibleAgentIDs(r.Context(), workspaceID, actorType, actorID, member.Role)
|
|
if !ok {
|
|
writeError(w, http.StatusInternalServerError, "failed to resolve agent access")
|
|
return
|
|
}
|
|
|
|
status := r.URL.Query().Get("status")
|
|
|
|
// Two call sites → two row types with identical shape. Collect into a
|
|
// common response slice via small per-branch loops.
|
|
var resp []ChatSessionResponse
|
|
if status == "all" {
|
|
rows, err := h.Queries.ListAllChatSessionsByCreator(r.Context(), db.ListAllChatSessionsByCreatorParams{
|
|
WorkspaceID: parseUUID(workspaceID),
|
|
CreatorID: parseUUID(userID),
|
|
})
|
|
if err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to list chat sessions")
|
|
return
|
|
}
|
|
resp = make([]ChatSessionResponse, 0, len(rows))
|
|
for _, s := range rows {
|
|
if _, ok := allowed[uuidToString(s.AgentID)]; !ok {
|
|
continue
|
|
}
|
|
resp = append(resp, ChatSessionResponse{
|
|
ID: uuidToString(s.ID),
|
|
WorkspaceID: uuidToString(s.WorkspaceID),
|
|
AgentID: uuidToString(s.AgentID),
|
|
CreatorID: uuidToString(s.CreatorID),
|
|
ProjectID: uuidToPtr(s.ProjectID),
|
|
Title: s.Title,
|
|
Status: s.Status,
|
|
HasUnread: s.UnreadCount > 0,
|
|
UnreadCount: int(s.UnreadCount),
|
|
LastMessage: buildChatLastMessage(s.LastMessageAt, s.LastMessageContent, s.LastMessageRole, s.LastMessageFailureReason, s.LastMessageKind),
|
|
Pinned: s.PinnedAt.Valid,
|
|
CreatedAt: timestampToString(s.CreatedAt),
|
|
UpdatedAt: timestampToString(s.UpdatedAt),
|
|
})
|
|
}
|
|
} else {
|
|
rows, err := h.Queries.ListChatSessionsByCreator(r.Context(), db.ListChatSessionsByCreatorParams{
|
|
WorkspaceID: parseUUID(workspaceID),
|
|
CreatorID: parseUUID(userID),
|
|
})
|
|
if err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to list chat sessions")
|
|
return
|
|
}
|
|
resp = make([]ChatSessionResponse, 0, len(rows))
|
|
for _, s := range rows {
|
|
if _, ok := allowed[uuidToString(s.AgentID)]; !ok {
|
|
continue
|
|
}
|
|
resp = append(resp, ChatSessionResponse{
|
|
ID: uuidToString(s.ID),
|
|
WorkspaceID: uuidToString(s.WorkspaceID),
|
|
AgentID: uuidToString(s.AgentID),
|
|
CreatorID: uuidToString(s.CreatorID),
|
|
ProjectID: uuidToPtr(s.ProjectID),
|
|
Title: s.Title,
|
|
Status: s.Status,
|
|
HasUnread: s.UnreadCount > 0,
|
|
UnreadCount: int(s.UnreadCount),
|
|
LastMessage: buildChatLastMessage(s.LastMessageAt, s.LastMessageContent, s.LastMessageRole, s.LastMessageFailureReason, s.LastMessageKind),
|
|
Pinned: s.PinnedAt.Valid,
|
|
CreatedAt: timestampToString(s.CreatedAt),
|
|
UpdatedAt: timestampToString(s.UpdatedAt),
|
|
})
|
|
}
|
|
}
|
|
if err := h.hydrateChatSessionChannelMetadata(r.Context(), resp); err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to load chat channel metadata")
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusOK, resp)
|
|
}
|
|
|
|
func (h *Handler) loadChatSessionForUser(w http.ResponseWriter, r *http.Request, userID, workspaceID, sessionID string) (db.ChatSession, bool) {
|
|
sessionUUID, ok := parseUUIDOrBadRequest(w, sessionID, "chat session id")
|
|
if !ok {
|
|
return db.ChatSession{}, false
|
|
}
|
|
workspaceUUID, ok := parseUUIDOrBadRequest(w, workspaceID, "workspace id")
|
|
if !ok {
|
|
return db.ChatSession{}, false
|
|
}
|
|
session, err := h.Queries.GetChatSessionInWorkspace(r.Context(), db.GetChatSessionInWorkspaceParams{
|
|
ID: sessionUUID,
|
|
WorkspaceID: workspaceUUID,
|
|
})
|
|
if err != nil {
|
|
writeError(w, http.StatusNotFound, "chat session not found")
|
|
return db.ChatSession{}, false
|
|
}
|
|
if uuidToString(session.CreatorID) != userID {
|
|
writeError(w, http.StatusForbidden, "not your chat session")
|
|
return db.ChatSession{}, false
|
|
}
|
|
return session, true
|
|
}
|
|
|
|
// gateChatSessionForUser combines the session ownership check with the
|
|
// private-agent access gate so a member who has lost access to the target
|
|
// agent (role downgrade, ownership transfer, agent flipped to private)
|
|
// cannot continue reading the chat transcript even though they remain the
|
|
// session creator. Returns ok=false after writing the error response.
|
|
func (h *Handler) gateChatSessionForUser(w http.ResponseWriter, r *http.Request, userID, workspaceID, sessionID string) (db.ChatSession, bool) {
|
|
session, ok := h.loadChatSessionForUser(w, r, userID, workspaceID, sessionID)
|
|
if !ok {
|
|
return db.ChatSession{}, false
|
|
}
|
|
agent, err := h.Queries.GetAgent(r.Context(), session.AgentID)
|
|
if err != nil {
|
|
writeError(w, http.StatusNotFound, "agent not found")
|
|
return db.ChatSession{}, false
|
|
}
|
|
actorType, actorID := h.resolveActor(r, userID, workspaceID)
|
|
if !h.canAccessPrivateAgent(r.Context(), agent, actorType, actorID, workspaceID) {
|
|
writeError(w, http.StatusForbidden, "you do not have access to this agent")
|
|
return db.ChatSession{}, false
|
|
}
|
|
return session, true
|
|
}
|
|
|
|
// gatePublicChatSessionForUser adds the member-visible projection boundary to
|
|
// the normal ownership and private-agent checks. Internal channel commands stay
|
|
// durable, but a session containing only those commands is not a public Chat.
|
|
// Cleanup and recovery operations keep using the lower-level ownership gates
|
|
// so an empty bound session can still be archived or deleted and a member's
|
|
// own cancelled draft is never stranded. Every ordinary Chat interaction uses
|
|
// this gate so a cached hidden session ID cannot resurrect a command-only Chat.
|
|
func (h *Handler) gatePublicChatSessionForUser(w http.ResponseWriter, r *http.Request, userID, workspaceID, sessionID string) (db.ChatSession, bool) {
|
|
session, ok := h.gateChatSessionForUser(w, r, userID, workspaceID, sessionID)
|
|
if !ok {
|
|
return db.ChatSession{}, false
|
|
}
|
|
if _, err := h.Queries.GetPublicChatSessionInWorkspace(r.Context(), db.GetPublicChatSessionInWorkspaceParams{
|
|
ID: session.ID,
|
|
WorkspaceID: session.WorkspaceID,
|
|
}); err != nil {
|
|
writeError(w, http.StatusNotFound, "chat session not found")
|
|
return db.ChatSession{}, false
|
|
}
|
|
return session, true
|
|
}
|
|
|
|
func (h *Handler) GetChatSession(w http.ResponseWriter, r *http.Request) {
|
|
userID, ok := requireUserID(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
workspaceID := ctxWorkspaceID(r.Context())
|
|
sessionID := chi.URLParam(r, "sessionId")
|
|
|
|
session, ok := h.gatePublicChatSessionForUser(w, r, userID, workspaceID, sessionID)
|
|
if !ok {
|
|
return
|
|
}
|
|
|
|
resp := chatSessionToResponse(session)
|
|
// hydrateChatSessionChannelMetadata mutates the slice element, so retain a
|
|
// concrete slice here rather than passing a temporary value to writeJSON.
|
|
responses := []ChatSessionResponse{resp}
|
|
if err := h.hydrateChatSessionChannelMetadata(r.Context(), responses); err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to load chat channel metadata")
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusOK, responses[0])
|
|
}
|
|
|
|
type UpdateChatSessionRequest struct {
|
|
Title *string `json:"title"`
|
|
ProjectID json.RawMessage `json:"project_id"`
|
|
}
|
|
|
|
// UpdateChatSession updates one user-editable field on a chat session. Title
|
|
// is surfaced by inline rename; project_id controls the project context used
|
|
// by subsequent turns. Status and pinned keep their dedicated endpoints,
|
|
// agent/creator/workspace are immutable, and the resume pointers
|
|
// (session_id / work_dir / runtime_id) remain daemon-owned.
|
|
func (h *Handler) UpdateChatSession(w http.ResponseWriter, r *http.Request) {
|
|
userID, ok := requireUserID(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
workspaceID := ctxWorkspaceID(r.Context())
|
|
sessionID := chi.URLParam(r, "sessionId")
|
|
|
|
var req UpdateChatSessionRequest
|
|
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid request body")
|
|
return
|
|
}
|
|
hasTitle := req.Title != nil
|
|
hasProjectID := req.ProjectID != nil
|
|
if hasTitle == hasProjectID {
|
|
writeError(w, http.StatusBadRequest, "exactly one of title or project_id is required")
|
|
return
|
|
}
|
|
|
|
session, ok := h.gatePublicChatSessionForUser(w, r, userID, workspaceID, sessionID)
|
|
if !ok {
|
|
return
|
|
}
|
|
|
|
var (
|
|
updated db.ChatSession
|
|
err error
|
|
)
|
|
var projectIDChanged bool
|
|
if hasTitle {
|
|
title := strings.TrimSpace(*req.Title)
|
|
if title == "" {
|
|
writeError(w, http.StatusBadRequest, "title is required")
|
|
return
|
|
}
|
|
if len([]rune(title)) > chatSessionTitleMaxLen {
|
|
writeError(w, http.StatusBadRequest, "title is too long")
|
|
return
|
|
}
|
|
updated, err = h.Queries.UpdateChatSessionTitle(r.Context(), db.UpdateChatSessionTitleParams{
|
|
ID: session.ID,
|
|
Title: title,
|
|
})
|
|
} else {
|
|
projectID := pgtype.UUID{Valid: false}
|
|
if string(req.ProjectID) != "null" {
|
|
var rawProjectID string
|
|
if err := json.Unmarshal(req.ProjectID, &rawProjectID); err != nil {
|
|
writeError(w, http.StatusBadRequest, "project_id must be a UUID or null")
|
|
return
|
|
}
|
|
rawProjectID = strings.TrimSpace(rawProjectID)
|
|
if rawProjectID == "" {
|
|
writeError(w, http.StatusBadRequest, "project_id must be a UUID or null")
|
|
return
|
|
}
|
|
projectID, ok = parseUUIDOrBadRequest(w, rawProjectID, "project_id")
|
|
if !ok {
|
|
return
|
|
}
|
|
}
|
|
|
|
tx, txErr := h.TxStarter.Begin(r.Context())
|
|
if txErr != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to start transaction")
|
|
return
|
|
}
|
|
defer tx.Rollback(r.Context())
|
|
qtx := h.Queries.WithTx(tx)
|
|
|
|
if projectID.Valid {
|
|
if _, lockErr := qtx.LockProjectForChatSessionCreate(r.Context(), db.LockProjectForChatSessionCreateParams{
|
|
ID: projectID,
|
|
WorkspaceID: session.WorkspaceID,
|
|
}); lockErr != nil {
|
|
if errors.Is(lockErr, pgx.ErrNoRows) {
|
|
writeError(w, http.StatusNotFound, "project not found")
|
|
return
|
|
}
|
|
writeError(w, http.StatusInternalServerError, "failed to lock project")
|
|
return
|
|
}
|
|
}
|
|
|
|
updated, err = qtx.UpdateChatSessionProject(r.Context(), db.UpdateChatSessionProjectParams{
|
|
ID: session.ID,
|
|
WorkspaceID: session.WorkspaceID,
|
|
ProjectID: projectID,
|
|
})
|
|
if err == nil {
|
|
err = tx.Commit(r.Context())
|
|
}
|
|
projectIDChanged = true
|
|
}
|
|
if err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to update chat session")
|
|
return
|
|
}
|
|
|
|
resolvedSessionID := uuidToString(updated.ID)
|
|
payload := protocol.ChatSessionUpdatedPayload{
|
|
ChatSessionID: resolvedSessionID,
|
|
Title: updated.Title,
|
|
UpdatedAt: timestampToString(updated.UpdatedAt),
|
|
}
|
|
if projectIDChanged {
|
|
projectID := uuidToPtr(updated.ProjectID)
|
|
payload.ProjectID = &projectID
|
|
}
|
|
h.publishChat(protocol.EventChatSessionUpdated, workspaceID, "member", userID, resolvedSessionID, payload)
|
|
|
|
writeJSON(w, http.StatusOK, chatSessionToResponse(updated))
|
|
}
|
|
|
|
type SetChatSessionPinnedRequest struct {
|
|
Pinned bool `json:"pinned"`
|
|
}
|
|
|
|
// SetChatSessionPinned pins or unpins a chat so it sticks to the top of the
|
|
// caller's conversation list. Pin state is per-session and, since sessions are
|
|
// per-creator, inherently per-user. It never bumps updated_at (see the SQL) so
|
|
// an unpinned chat does not jump the activity-sorted list.
|
|
func (h *Handler) SetChatSessionPinned(w http.ResponseWriter, r *http.Request) {
|
|
userID, ok := requireUserID(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
workspaceID := ctxWorkspaceID(r.Context())
|
|
sessionID := chi.URLParam(r, "sessionId")
|
|
|
|
var req SetChatSessionPinnedRequest
|
|
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid request body")
|
|
return
|
|
}
|
|
|
|
session, ok := h.gatePublicChatSessionForUser(w, r, userID, workspaceID, sessionID)
|
|
if !ok {
|
|
return
|
|
}
|
|
|
|
updated, err := h.Queries.SetChatSessionPinned(r.Context(), db.SetChatSessionPinnedParams{
|
|
ID: session.ID,
|
|
Pinned: req.Pinned,
|
|
})
|
|
if err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to update chat session")
|
|
return
|
|
}
|
|
|
|
resolvedSessionID := uuidToString(updated.ID)
|
|
pinned := updated.PinnedAt.Valid
|
|
h.publishChat(protocol.EventChatSessionUpdated, workspaceID, "member", userID, resolvedSessionID, protocol.ChatSessionUpdatedPayload{
|
|
ChatSessionID: resolvedSessionID,
|
|
Title: updated.Title,
|
|
Pinned: &pinned,
|
|
UpdatedAt: timestampToString(updated.UpdatedAt),
|
|
})
|
|
|
|
writeJSON(w, http.StatusOK, chatSessionToResponse(updated))
|
|
}
|
|
|
|
type SetChatSessionArchivedRequest struct {
|
|
Archived bool `json:"archived"`
|
|
}
|
|
|
|
// SetChatSessionArchived archives or unarchives a chat session owned by the
|
|
// caller. Archiving is the reversible sibling of delete: the row stays in the
|
|
// user's history (behind the "Archived" entry) and the conversation becomes
|
|
// read-only — SendChatMessage refuses status='archived'. Hard delete is only
|
|
// offered from the archived list, so nothing is destroyed in one hover click.
|
|
//
|
|
// Archiving also severs any external-channel binding. The web send path already
|
|
// treats status='archived' as read-only, but the channel engine (Feishu/Slack)
|
|
// resolves inbound traffic through channel_chat_session_binding without checking
|
|
// session status, so a bound session kept accumulating agent replies — and a
|
|
// stuck unread badge — after the user archived it (MUL-4372). Dropping the
|
|
// binding in the same tx makes the next inbound message for that external chat
|
|
// create a fresh chat_session under a new binding (see EnsureSession) instead of
|
|
// reviving this archived one. Unarchive deliberately does NOT recreate the
|
|
// binding: if later traffic already forked a new session, that session owns the
|
|
// channel now, and restoring the old binding would steal it back.
|
|
//
|
|
// Severing that binding is what makes archiving a bound session cancel its
|
|
// in-flight tasks: the adapter still holds the chat id, so a turn that survives
|
|
// the archive answers into a group room the session no longer belongs to. That
|
|
// cancel is therefore scoped to sessions that HAD a binding. Archiving a
|
|
// web-only chat leaves its tasks alone — it is the reversible organizing action
|
|
// the UI presents (no confirmation, unarchive undoes it), and the destructive
|
|
// counterpart is the Stop button, which cancels through CancelTaskByUser and
|
|
// gives the user their typed prompt back.
|
|
func (h *Handler) SetChatSessionArchived(w http.ResponseWriter, r *http.Request) {
|
|
userID, ok := requireUserID(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
workspaceID := ctxWorkspaceID(r.Context())
|
|
sessionID := chi.URLParam(r, "sessionId")
|
|
|
|
var req SetChatSessionArchivedRequest
|
|
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid request body")
|
|
return
|
|
}
|
|
|
|
session, ok := h.gateChatSessionForUser(w, r, userID, workspaceID, sessionID)
|
|
if !ok {
|
|
return
|
|
}
|
|
|
|
tx, err := h.TxStarter.Begin(r.Context())
|
|
if err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to start transaction")
|
|
return
|
|
}
|
|
defer tx.Rollback(r.Context())
|
|
qtx := h.Queries.WithTx(tx)
|
|
|
|
updated, err := qtx.SetChatSessionArchived(r.Context(), db.SetChatSessionArchivedParams{
|
|
ID: session.ID,
|
|
Archived: req.Archived,
|
|
})
|
|
if err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to update chat session")
|
|
return
|
|
}
|
|
|
|
// Rows cancelled inside the tx, broadcast after it commits. nil when the
|
|
// request unarchives and nil for a web-only chat; BroadcastCancelledTasks
|
|
// is a no-op on an empty slice.
|
|
var cancelled []db.AgentTaskQueue
|
|
var reactionTargets map[string]*events.ChannelReactionTarget
|
|
|
|
if req.Archived {
|
|
// Read the binding BEFORE the delete below, which is what erases the
|
|
// evidence: after it runs there is no way to tell a session that was
|
|
// on a channel from one that never was.
|
|
_, bindingErr := qtx.GetChannelChatSessionBindingBySessionAny(r.Context(), session.ID)
|
|
switch {
|
|
case bindingErr == nil:
|
|
// Bound, and the delete below is about to drop that binding.
|
|
// ClaimAgentTask does not read chat_session.status, so a task
|
|
// queued before the archive stays claimable after it: the daemon
|
|
// runs a turn on a conversation the user closed, spending runtime
|
|
// and quota and writing assistant messages into an archived chat.
|
|
// The binding is gone by then, so the brief describes the run as a
|
|
// private web chat — the misattribution the history note below
|
|
// exists to prevent — while the outbound senders, which resolve
|
|
// their destination through that same deleted row, have no route
|
|
// left to deliver on. Cancel in the same tx that drops the
|
|
// binding, as DeleteChatSession does.
|
|
//
|
|
// The query is not limited to never-started work: it also matches
|
|
// dispatched, running, waiting_local_directory and deferred, which
|
|
// is why the returned rows have to reach the post-commit broadcast
|
|
// below rather than being discarded.
|
|
cancelled, err = qtx.CancelAgentTasksByChatSession(r.Context(), session.ID)
|
|
if err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to cancel queued tasks for the archived session")
|
|
return
|
|
}
|
|
if err = service.SettleTerminalTaskState(r.Context(), qtx, cancelled...); err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to settle terminal task state")
|
|
return
|
|
}
|
|
case errors.Is(bindingErr, pgx.ErrNoRows):
|
|
// A web-only chat has no room, no adapter and nowhere for a late
|
|
// answer to land, so there is nothing here to protect anyone from
|
|
// and archiving stays what the UI already says it is: the
|
|
// reversible "put it away" step that unarchive undoes, offered
|
|
// with no confirmation, with hard delete available only after it.
|
|
// Cancelling would also destroy the user's typed prompt — the Stop
|
|
// button cancels through CancelTaskByUser, which hands the prompt
|
|
// back (cancelled_chat_message.restore_to_input, or a durable
|
|
// chat_draft_restore row), and CancelAgentTasksByChatSession does
|
|
// neither.
|
|
default:
|
|
// We cannot tell whether this session was bound. Carrying on would
|
|
// silently skip the cancel and reopen the bug above, so fail the
|
|
// archive rather than guess; the caller can retry.
|
|
slog.Warn("read chat session channel binding failed", "session_id", sessionID, "error", bindingErr)
|
|
writeError(w, http.StatusInternalServerError, "failed to read chat session channel binding")
|
|
return
|
|
}
|
|
reactionTargets, err = service.CaptureChannelReactionTargets(r.Context(), qtx, cancelled)
|
|
if err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to capture channel reaction targets")
|
|
return
|
|
}
|
|
if err := qtx.DeleteChannelChatSessionBindingBySession(r.Context(), session.ID); err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to clear chat session channel binding")
|
|
return
|
|
}
|
|
}
|
|
|
|
if err := tx.Commit(r.Context()); err != nil {
|
|
slog.Warn("commit chat session archive failed", "session_id", sessionID, "error", err)
|
|
writeError(w, http.StatusInternalServerError, "failed to commit chat session update")
|
|
return
|
|
}
|
|
|
|
// Post-commit broadcasts, same as DeleteChatSession: subscribers should
|
|
// never observe events for a tx that didn't actually persist. This is the
|
|
// call that captures the cancellation and revokes the tasks' mat_ tokens,
|
|
// reconciles each agent off 'working', emits task:cancelled so other
|
|
// clients drop the row instead of showing it queued until the next
|
|
// refresh, and wakes the runtime so a queued successor is claimed now
|
|
// rather than at the daemon's next poll.
|
|
h.TaskService.BroadcastCancelledTasks(r.Context(), workspaceID, cancelled, reactionTargets)
|
|
|
|
resolvedSessionID := uuidToString(updated.ID)
|
|
status := updated.Status
|
|
h.publishChat(protocol.EventChatSessionUpdated, workspaceID, "member", userID, resolvedSessionID, protocol.ChatSessionUpdatedPayload{
|
|
ChatSessionID: resolvedSessionID,
|
|
Title: updated.Title,
|
|
Status: &status,
|
|
UpdatedAt: timestampToString(updated.UpdatedAt),
|
|
})
|
|
|
|
writeJSON(w, http.StatusOK, chatSessionToResponse(updated))
|
|
}
|
|
|
|
// DeleteChatSession hard-deletes a chat session owned by the caller. The
|
|
// row lock + cancel + delete run inside a single tx so a concurrent
|
|
// SendChatMessage cannot enqueue a task that would later be orphaned by
|
|
// the FK ON DELETE SET NULL on agent_task_queue.chat_session_id. Cancel
|
|
// failure aborts the delete; events fire only after commit.
|
|
func (h *Handler) DeleteChatSession(w http.ResponseWriter, r *http.Request) {
|
|
userID, ok := requireUserID(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
workspaceID := ctxWorkspaceID(r.Context())
|
|
sessionID := chi.URLParam(r, "sessionId")
|
|
|
|
session, ok := h.loadChatSessionForUser(w, r, userID, workspaceID, sessionID)
|
|
if !ok {
|
|
return
|
|
}
|
|
|
|
tx, err := h.TxStarter.Begin(r.Context())
|
|
if err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to start transaction")
|
|
return
|
|
}
|
|
defer tx.Rollback(r.Context())
|
|
qtx := h.Queries.WithTx(tx)
|
|
|
|
// FOR UPDATE on the chat_session row blocks any concurrent INSERT into
|
|
// agent_task_queue that references it (the FK validation needs a
|
|
// KEY SHARE lock). After we commit the delete, the blocked INSERT
|
|
// fails its FK check, so it can't land an orphaned task.
|
|
if _, err := qtx.LockChatSessionForDelete(r.Context(), session.ID); err != nil {
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
// Already gone — treat as idempotent success.
|
|
w.WriteHeader(http.StatusNoContent)
|
|
return
|
|
}
|
|
writeError(w, http.StatusInternalServerError, "failed to lock chat session")
|
|
return
|
|
}
|
|
|
|
// Claim, clear, prioritize, and direct send all lock the agent before task
|
|
// rows. Keep delete on the same agent -> task suffix after its session lock,
|
|
// otherwise a builder-agent delete can deadlock with a concurrent claim.
|
|
if _, err := qtx.GetAgentForClaimUpdate(r.Context(), session.AgentID); err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to lock chat agent")
|
|
return
|
|
}
|
|
|
|
cancelled, err := qtx.CancelAgentTasksByChatSession(r.Context(), session.ID)
|
|
if err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to cancel chat session tasks")
|
|
return
|
|
}
|
|
if err := service.SettleTerminalTaskState(r.Context(), qtx, cancelled...); err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to settle terminal task state")
|
|
return
|
|
}
|
|
|
|
reactionTargets, err := service.CaptureChannelReactionTargets(r.Context(), qtx, cancelled)
|
|
if err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to capture channel reaction targets")
|
|
return
|
|
}
|
|
|
|
// channel_chat_session_binding used to carry a chat_session FK with
|
|
// ON DELETE CASCADE; MUL-3515 §4 dropped every channel_* foreign key, so
|
|
// prune the binding here in the same tx that deletes its chat_session.
|
|
if err := qtx.DeleteChannelChatSessionBindingBySession(r.Context(), session.ID); err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to delete chat session binding")
|
|
return
|
|
}
|
|
// channel_outbound_card_message is also keyed by chat_session_id with no FK
|
|
// and no reaper, so prune it in the same tx or a deleted chat session leaves
|
|
// permanent orphan card rows (#4810 follow-up).
|
|
if err := qtx.DeleteChannelOutboundCardMessagesBySession(r.Context(), session.ID); err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to delete chat session outbound cards")
|
|
return
|
|
}
|
|
// chat_draft_restore is keyed by chat_session_id with no FK either, so its
|
|
// pending restores are pruned here rather than by a DB cascade (#5219). The
|
|
// LockChatSessionForDelete above doubles as the deleter half of the
|
|
// draft-restore protocol: FinalizeDeferredCancelledChat takes the same
|
|
// chat_session lock before inserting a restore, so it cannot slip one past
|
|
// this sweep (see LockChatSessionForTask in chat.sql).
|
|
if err := qtx.DeleteChatDraftRestoresBySession(r.Context(), session.ID); err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to delete chat session draft restores")
|
|
return
|
|
}
|
|
|
|
// Same no-FK chore, for the agent configuration a builder conversation was
|
|
// editing. A no-op for ordinary chats, which never have one.
|
|
if err := qtx.DeleteAgentBuilderDraft(r.Context(), session.ID); err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to delete agent builder draft")
|
|
return
|
|
}
|
|
|
|
if err := qtx.DeleteChatSession(r.Context(), db.DeleteChatSessionParams{
|
|
ID: session.ID,
|
|
WorkspaceID: session.WorkspaceID,
|
|
}); err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to delete chat session")
|
|
return
|
|
}
|
|
if err := qtx.DeleteAgentLabelAssignmentsByAgent(r.Context(), session.AgentID); err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to remove chat session agent label assignments")
|
|
return
|
|
}
|
|
if err := qtx.DeleteSystemAgentByID(r.Context(), session.AgentID); err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to clean up chat session agent")
|
|
return
|
|
}
|
|
|
|
if err := tx.Commit(r.Context()); err != nil {
|
|
slog.Warn("commit chat session delete failed", "session_id", sessionID, "error", err)
|
|
writeError(w, http.StatusInternalServerError, "failed to commit chat session delete")
|
|
return
|
|
}
|
|
|
|
// Post-commit broadcasts. Subscribers should never observe events for a
|
|
// tx that didn't actually persist.
|
|
//
|
|
// The workspace has to come from the session we just deleted: the tasks were
|
|
// cancelled and returned before the delete, so they still carry its id, but
|
|
// the row they would be resolved through is gone by now.
|
|
h.TaskService.BroadcastCancelledTasks(r.Context(), workspaceID, cancelled, reactionTargets)
|
|
|
|
resolvedSessionID := uuidToString(session.ID)
|
|
h.publishChat(protocol.EventChatSessionDeleted, workspaceID, "member", userID, resolvedSessionID, protocol.ChatSessionDeletedPayload{
|
|
ChatSessionID: resolvedSessionID,
|
|
})
|
|
|
|
w.WriteHeader(http.StatusNoContent)
|
|
}
|
|
|
|
// Chat Messages
|
|
|
|
type SendChatMessageRequest struct {
|
|
Content string `json:"content"`
|
|
AttachmentIDs []string `json:"attachment_ids"`
|
|
}
|
|
|
|
type SendChatMessageResponse struct {
|
|
MessageID string `json:"message_id"`
|
|
TaskID string `json:"task_id"`
|
|
SupportsQueue bool `json:"supports_queue"`
|
|
Queued bool `json:"queued"`
|
|
// AttachmentIDs are the attachment rows actually bound to this message by
|
|
// the server. The client diffs these against the ids it requested so it
|
|
// can warn the user when an attachment silently failed to bind — no extra
|
|
// round-trip needed. No `omitempty`: a send that requested attachments but
|
|
// bound none must serialize `[]` (not be omitted), otherwise the client
|
|
// can't tell "all binds failed" from "older server without this field" and
|
|
// would silently skip the very warning this exists for. When no
|
|
// attachments were requested the value is nil → `null`, which the client's
|
|
// guard short-circuits on the requested-ids check.
|
|
AttachmentIDs []string `json:"attachment_ids"`
|
|
// CreatedAt anchors the chat StatusPill timer the instant the user
|
|
// hits send. Without it the front-end falls back to its local clock
|
|
// and the timer "snaps backwards" later when WS events deliver the
|
|
// real created_at. Returning it here means the pill renders 0s from
|
|
// the start with a stable anchor.
|
|
CreatedAt string `json:"created_at"`
|
|
}
|
|
|
|
func (h *Handler) SendChatMessage(w http.ResponseWriter, r *http.Request) {
|
|
userID, ok := requireUserID(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
workspaceID := ctxWorkspaceID(r.Context())
|
|
sessionID := chi.URLParam(r, "sessionId")
|
|
|
|
var req SendChatMessageRequest
|
|
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid request body")
|
|
return
|
|
}
|
|
if req.Content == "" {
|
|
writeError(w, http.StatusBadRequest, "content is required")
|
|
return
|
|
}
|
|
|
|
// Pre-validate attachment ids early so invalid input returns 400 before
|
|
// any state mutation. The actual link runs after CreateChatMessage so we
|
|
// have a message_id to back-fill into the attachment rows.
|
|
attachmentIDs, ok := parseUUIDSliceOrBadRequest(w, req.AttachmentIDs, "attachment_ids")
|
|
if !ok {
|
|
return
|
|
}
|
|
|
|
// Load chat session and re-check the private-agent gate on every send.
|
|
// The session's creator passed the gate at create time, but their
|
|
// workspace role (or the agent's owner) may have changed since — keep
|
|
// stale sessions from being a back-door into a private agent the user
|
|
// can no longer reach. Agent senders bypass to preserve A2A collaboration.
|
|
session, ok := h.gatePublicChatSessionForUser(w, r, userID, workspaceID, sessionID)
|
|
if !ok {
|
|
return
|
|
}
|
|
// Archived sessions are read-only: refuse to enqueue new agent work for
|
|
// them (see SetChatSessionArchived). The frontend disables the composer
|
|
// for status='archived' and only offers unarchive/delete there. Legacy
|
|
// soft-archived rows from before the feature are covered by the same check.
|
|
if session.Status != "active" {
|
|
writeError(w, http.StatusBadRequest, "chat session is archived")
|
|
return
|
|
}
|
|
|
|
// Preflight the agent's enqueue preconditions BEFORE persisting the user
|
|
// message. EnqueueChatTask (below) rejects an archived or runtime-less agent
|
|
// with an error; without this check a stale client (agent archived in
|
|
// another tab) would land the user message, then get a 500, leaving an
|
|
// orphan message with no task or reply. Fail fast with a 4xx and mutate
|
|
// nothing. See EnqueueChatTask's ErrChatTaskAgentArchived / NoRuntime.
|
|
agent, err := h.Queries.GetAgent(r.Context(), session.AgentID)
|
|
if err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to load chat agent")
|
|
return
|
|
}
|
|
if agent.ArchivedAt.Valid {
|
|
writeError(w, http.StatusConflict, "chat agent is archived")
|
|
return
|
|
}
|
|
// Shared verdict: an unbound agent and a machine whose CLI cannot run are
|
|
// both refusals here, with their own codes. A merely offline runtime is not
|
|
// checked at all — chat messages queue for it, as they always have.
|
|
if verdict, err := service.AgentReadiness(r.Context(), h.runtimeLookup(obsmetrics.RuntimeLookupSourceChat), agent); err == nil && verdict.Blocked() {
|
|
h.writeDispatchBlocked(w, http.StatusConflict, verdict.Reason)
|
|
return
|
|
}
|
|
|
|
// Re-run the INVOKE gate on every send, not just the softer view gate in
|
|
// gatePublicChatSessionForUser (MUL-4525). canAccessPrivateAgent lets a workspace
|
|
// admin keep reading a transcript, but sending a message enqueues a run and
|
|
// must satisfy canInvokeAgent — which has no admin bypass. A session created
|
|
// while the user could invoke the agent must stop enqueuing work the instant
|
|
// that permission is revoked (agent flipped private, ownership moved, target
|
|
// removed from the allow-list), and it must fail BEFORE we persist the user
|
|
// message / attachments / task. Blocked returns a structured, enumeration-safe
|
|
// reason so the composer can explain it without leaking private-agent details.
|
|
actorType, actorID := h.resolveActor(r, userID, workspaceID)
|
|
if !h.canInvokeAgent(r.Context(), agent, actorType, actorID, h.invokeOriginatorFromRequest(r, actorType, actorID), workspaceID) {
|
|
h.writeDispatchBlocked(w, http.StatusForbidden, ReasonInvocationNotAllowed)
|
|
return
|
|
}
|
|
|
|
// Detect whether this is the very first human message in the session,
|
|
// BEFORE we insert the new row. This scopes LLM auto-titling (MUL-4295) to
|
|
// the opening turn: we upgrade the default/original title exactly once, off
|
|
// the first user message, and never re-run it on every subsequent send. A
|
|
// query error here is treated as "not first" so we simply skip generation
|
|
// (best-effort — never block the send).
|
|
hadUserMessage := true
|
|
if existed, err := h.Queries.ChatSessionHasPublicUserMessage(r.Context(), session.ID); err == nil {
|
|
hadUserMessage = existed
|
|
}
|
|
channelBacked := false
|
|
channelSourceKnown := true
|
|
if !hadUserMessage {
|
|
if _, err := h.Queries.GetChannelChatSessionBindingBySessionAny(r.Context(), session.ID); err == nil {
|
|
channelBacked = true
|
|
} else if !errors.Is(err, pgx.ErrNoRows) {
|
|
// Title generation is optional. If source authority is unavailable,
|
|
// skip it rather than risk treating a channel manual rename as an
|
|
// auto-generated fallback and overwriting it.
|
|
channelSourceKnown = false
|
|
}
|
|
}
|
|
|
|
// Persist the whole turn atomically (MUL-4351): the owning task, the user
|
|
// message bound to that task (so it belongs to the task's immutable input
|
|
// batch the instant it exists), attachment bindings, and the session touch
|
|
// all commit together, and the daemon is only notified after the commit. For
|
|
// web chat the sender is the authenticated request user (sessions are
|
|
// creator-only), so they are the task initiator and the run's originator —
|
|
// surfaced to the agent under `## On Behalf Of`. actorType/actorID were
|
|
// resolved above for the invoke gate.
|
|
sent, err := h.TaskService.SendDirectChatMessage(r.Context(), session, agent, parseUUID(userID), req.Content, attachmentIDs, actorType, parseUUID(actorID))
|
|
if err != nil {
|
|
switch {
|
|
case errors.Is(err, service.ErrChatSessionArchived):
|
|
writeError(w, http.StatusConflict, "chat session is archived")
|
|
case errors.Is(err, service.ErrChatTaskAgentArchived):
|
|
writeError(w, http.StatusConflict, "chat agent is archived")
|
|
case errors.Is(err, service.ErrChatTaskAgentNoRuntime):
|
|
writeError(w, http.StatusConflict, "chat agent has no runtime")
|
|
default:
|
|
slog.Warn("send chat message failed", append(logger.RequestAttrs(r), "error", err)...)
|
|
writeError(w, http.StatusInternalServerError, "failed to send chat message")
|
|
}
|
|
return
|
|
}
|
|
msg := sent.Message
|
|
task := sent.Task
|
|
currentTitle := session.Title
|
|
if sent.InitialTitle != "" {
|
|
currentTitle = sent.InitialTitle
|
|
h.ChannelChatTitleInitialized(session.WorkspaceID, session.CreatorID, session.ID, sent.InitialTitle)
|
|
}
|
|
|
|
// AttachmentIDs actually bound by the server. Requested-but-unbound ids are
|
|
// surfaced to the client so it can warn the user (see SendChatMessageResponse).
|
|
var boundAttachmentIDs []string
|
|
if len(attachmentIDs) > 0 {
|
|
boundAttachmentIDs = make([]string, 0, len(sent.BoundAttachmentIDs))
|
|
for _, id := range sent.BoundAttachmentIDs {
|
|
boundAttachmentIDs = append(boundAttachmentIDs, uuidToString(id))
|
|
}
|
|
}
|
|
|
|
taskContext := h.TaskService.AnalyticsContextForTask(r.Context(), task)
|
|
platform, _, _ := middleware.ClientMetadataFromContext(r.Context())
|
|
obsmetrics.RecordEvent(h.Analytics, h.Metrics, analytics.ChatMessageSent(
|
|
userID,
|
|
workspaceID,
|
|
uuidToString(session.ID),
|
|
uuidToString(task.ID),
|
|
uuidToString(session.AgentID),
|
|
taskContext.RuntimeMode,
|
|
taskContext.Provider,
|
|
platform,
|
|
))
|
|
|
|
// Broadcast the user message.
|
|
resolvedSessionID := uuidToString(session.ID)
|
|
h.publishChat(protocol.EventChatMessage, workspaceID, "member", userID, resolvedSessionID, protocol.ChatMessagePayload{
|
|
ChatSessionID: resolvedSessionID,
|
|
MessageID: uuidToString(msg.ID),
|
|
Role: "user",
|
|
Content: req.Content,
|
|
TaskID: uuidToString(task.ID),
|
|
CreatedAt: timestampToString(msg.CreatedAt),
|
|
})
|
|
|
|
// First user message → kick off best-effort LLM auto-titling (MUL-4295).
|
|
// Fire-and-forget and non-blocking: the response below is written whether
|
|
// or not a title is ever generated, and a disabled/failing LLM layer
|
|
// silently keeps the original first-message-derived title. session.Title
|
|
// is the default/original title observed here and drives the CAS so a
|
|
// manual rename mid-generation is never clobbered.
|
|
shouldGenerateTitle := shouldGenerateFirstMessageTitle(hadUserMessage, currentTitle, sent.InitialTitle, channelBacked, channelSourceKnown)
|
|
if shouldGenerateTitle {
|
|
h.maybeGenerateChatTitleAsync(workspaceID, userID, session.ID, currentTitle, req.Content)
|
|
}
|
|
|
|
writeJSON(w, http.StatusCreated, SendChatMessageResponse{
|
|
MessageID: uuidToString(msg.ID),
|
|
TaskID: uuidToString(task.ID),
|
|
SupportsQueue: true,
|
|
Queued: sent.Queued,
|
|
CreatedAt: timestampToString(task.CreatedAt),
|
|
AttachmentIDs: boundAttachmentIDs,
|
|
})
|
|
}
|
|
|
|
func shouldGenerateFirstMessageTitle(hadUserMessage bool, currentTitle, initializedTitle string, channelBacked, channelSourceKnown bool) bool {
|
|
if hadUserMessage || currentTitle == "" || !channelSourceKnown {
|
|
return false
|
|
}
|
|
if channelBacked {
|
|
// For an explicitly empty /new Chat, a pre-existing non-empty title is
|
|
// a manual rename. Only a title initialized by this send is eligible for
|
|
// the best-effort LLM replacement; otherwise manual naming always wins.
|
|
return initializedTitle != ""
|
|
}
|
|
return true
|
|
}
|
|
|
|
type ChatMessagesCursorResponse struct {
|
|
CreatedAt string `json:"created_at"`
|
|
ID string `json:"id"`
|
|
}
|
|
|
|
type ChatMessagesPageResponse struct {
|
|
Messages []ChatMessageResponse `json:"messages"`
|
|
Limit int `json:"limit"`
|
|
HasMore bool `json:"has_more"`
|
|
NextCursor *ChatMessagesCursorResponse `json:"next_cursor,omitempty"`
|
|
}
|
|
|
|
func parseChatMessagesPageParams(r *http.Request) (int, pgtype.Timestamptz, pgtype.UUID, error) {
|
|
limit := 50
|
|
if raw := r.URL.Query().Get("limit"); raw != "" {
|
|
parsed, err := strconv.Atoi(raw)
|
|
if err != nil || parsed < 1 || parsed > 100 {
|
|
return 0, pgtype.Timestamptz{}, pgtype.UUID{}, errors.New("invalid limit")
|
|
}
|
|
limit = parsed
|
|
}
|
|
|
|
rawBeforeCreatedAt := r.URL.Query().Get("before_created_at")
|
|
rawBeforeID := r.URL.Query().Get("before_id")
|
|
if rawBeforeCreatedAt == "" && rawBeforeID == "" {
|
|
return limit, pgtype.Timestamptz{}, pgtype.UUID{}, nil
|
|
}
|
|
if rawBeforeCreatedAt == "" || rawBeforeID == "" {
|
|
return 0, pgtype.Timestamptz{}, pgtype.UUID{}, errors.New("invalid cursor")
|
|
}
|
|
beforeTime, err := time.Parse(time.RFC3339Nano, rawBeforeCreatedAt)
|
|
if err != nil {
|
|
return 0, pgtype.Timestamptz{}, pgtype.UUID{}, errors.New("invalid cursor")
|
|
}
|
|
beforeID, err := util.ParseUUID(rawBeforeID)
|
|
if err != nil {
|
|
return 0, pgtype.Timestamptz{}, pgtype.UUID{}, errors.New("invalid cursor")
|
|
}
|
|
return limit, pgtype.Timestamptz{Time: beforeTime, Valid: true}, beforeID, nil
|
|
}
|
|
|
|
// RegenerateChatQuickActionsResponse acknowledges an accepted refresh request.
|
|
// message_id is the assistant turn the refreshed pills will attach to — the
|
|
// client anchors its pending placeholder on it and resolves it when the
|
|
// chat:quick_actions supplement arrives.
|
|
// RegenerateChatQuickActionsRequest names the assistant turn the client is
|
|
// refreshing. The server confirms it is still the session's latest turn before
|
|
// enqueuing (409 otherwise), so the client's pending marker stays aligned with
|
|
// the turn chat:quick_actions will resolve — no ack reconciliation needed
|
|
// (MUL-5149).
|
|
type RegenerateChatQuickActionsRequest struct {
|
|
MessageID string `json:"message_id"`
|
|
}
|
|
|
|
type RegenerateChatQuickActionsResponse struct {
|
|
MessageID string `json:"message_id"`
|
|
}
|
|
|
|
// RegenerateChatQuickActions re-runs the suggestion pass for a session's latest
|
|
// assistant turn on explicit user request (the "refresh" button on the
|
|
// quick-actions row, MUL-5149). Generation is server-side, so this spawns no
|
|
// agent run; the refreshed pills arrive over the same chat:quick_actions
|
|
// realtime path as the automatic pass.
|
|
func (h *Handler) RegenerateChatQuickActions(w http.ResponseWriter, r *http.Request) {
|
|
userID, ok := requireUserID(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
workspaceID := ctxWorkspaceID(r.Context())
|
|
sessionID := chi.URLParam(r, "sessionId")
|
|
|
|
session, ok := h.gatePublicChatSessionForUser(w, r, userID, workspaceID, sessionID)
|
|
if !ok {
|
|
return
|
|
}
|
|
if session.Status != "active" {
|
|
writeError(w, http.StatusBadRequest, "chat session is archived")
|
|
return
|
|
}
|
|
agent, err := h.Queries.GetAgent(r.Context(), session.AgentID)
|
|
if err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to load chat agent")
|
|
return
|
|
}
|
|
if agent.ArchivedAt.Valid {
|
|
writeError(w, http.StatusConflict, "chat agent is archived")
|
|
return
|
|
}
|
|
// The refresh no longer runs the agent, but it is still a user-triggered
|
|
// spend against that agent's conversation, so it keeps clearing the same
|
|
// INVOKE gate as a send (MUL-4525) rather than the softer view gate in
|
|
// gatePublicChatSessionForUser. Deliberately NOT relaxed as a side effect of
|
|
// moving generation server-side.
|
|
actorType, actorID := h.resolveActor(r, userID, workspaceID)
|
|
if !h.canInvokeAgent(r.Context(), agent, actorType, actorID, h.invokeOriginatorFromRequest(r, actorType, actorID), workspaceID) {
|
|
h.writeDispatchBlocked(w, http.StatusForbidden, ReasonInvocationNotAllowed)
|
|
return
|
|
}
|
|
|
|
var req RegenerateChatQuickActionsRequest
|
|
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid request body")
|
|
return
|
|
}
|
|
expectedMessageID, ok := parseUUIDOrBadRequest(w, req.MessageID, "message_id")
|
|
if !ok {
|
|
return
|
|
}
|
|
|
|
messageID, targetTask, err := h.TaskService.RegenerateChatQuickActions(r.Context(), session, expectedMessageID)
|
|
if err != nil {
|
|
switch {
|
|
case errors.Is(err, service.ErrChatQuickActionsStale):
|
|
// The turn the client was refreshing is no longer the latest — its
|
|
// view is stale (a newer reply arrived). 409 → the client rolls back
|
|
// its optimistic marker and re-offers refresh on the new turn.
|
|
writeError(w, http.StatusConflict, "a newer reply arrived — refresh it instead")
|
|
case errors.Is(err, service.ErrChatQuickActionsBusy):
|
|
// A turn is already running for this session; it is about to replace
|
|
// the reply these pills would hang on. 409 → the client rolls back
|
|
// and can refresh once the session settles.
|
|
writeError(w, http.StatusConflict, "still working — try refreshing in a moment")
|
|
case errors.Is(err, service.ErrChatQuickActionsNoTurn):
|
|
writeError(w, http.StatusConflict, "no assistant reply to refresh yet")
|
|
case errors.Is(err, service.ErrChatQuickActionsUnavailable):
|
|
writeFeatureDisabled(w, "suggestions_not_available", "suggestions are not available on this deployment")
|
|
default:
|
|
writeError(w, http.StatusInternalServerError, "failed to regenerate quick actions")
|
|
}
|
|
return
|
|
}
|
|
// Detached like the automatic pass: answer 202 now, deliver the refreshed
|
|
// pills over chat:quick_actions.
|
|
h.TaskService.GenerateChatQuickActionsAsync(targetTask, service.ChatQuickActionsRefresh)
|
|
writeJSON(w, http.StatusAccepted, RegenerateChatQuickActionsResponse{
|
|
MessageID: uuidToString(messageID),
|
|
})
|
|
}
|
|
|
|
func (h *Handler) ListChatMessages(w http.ResponseWriter, r *http.Request) {
|
|
userID, ok := requireUserID(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
workspaceID := ctxWorkspaceID(r.Context())
|
|
sessionID := chi.URLParam(r, "sessionId")
|
|
|
|
session, ok := h.gatePublicChatSessionForUser(w, r, userID, workspaceID, sessionID)
|
|
if !ok {
|
|
return
|
|
}
|
|
|
|
messages, err := h.Queries.ListChatMessages(r.Context(), session.ID)
|
|
if err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to list chat messages")
|
|
return
|
|
}
|
|
messages = visibleChatMessages(messages)
|
|
|
|
messageIDs := make([]pgtype.UUID, len(messages))
|
|
for i, m := range messages {
|
|
messageIDs[i] = m.ID
|
|
}
|
|
groupedAtt := h.groupChatMessageAttachments(r.Context(), workspaceID, messageIDs)
|
|
|
|
resp := make([]ChatMessageResponse, len(messages))
|
|
for i, m := range messages {
|
|
resp[i] = chatMessageToResponse(m, groupedAtt[uuidToString(m.ID)])
|
|
}
|
|
writeJSON(w, http.StatusOK, resp)
|
|
}
|
|
|
|
func (h *Handler) ListChatMessagesPage(w http.ResponseWriter, r *http.Request) {
|
|
userID, ok := requireUserID(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
workspaceID := ctxWorkspaceID(r.Context())
|
|
sessionID := chi.URLParam(r, "sessionId")
|
|
|
|
session, ok := h.gatePublicChatSessionForUser(w, r, userID, workspaceID, sessionID)
|
|
if !ok {
|
|
return
|
|
}
|
|
|
|
limit, beforeCreatedAt, beforeID, err := parseChatMessagesPageParams(r)
|
|
if err != nil {
|
|
writeError(w, http.StatusBadRequest, err.Error())
|
|
return
|
|
}
|
|
|
|
// A session can contain one server-authored onboarding kickoff. Fetch one
|
|
// extra row beyond the normal lookahead so hiding it cannot make a visible
|
|
// page appear shorter or lose its next cursor.
|
|
messages, err := h.Queries.ListChatMessagesPage(r.Context(), db.ListChatMessagesPageParams{
|
|
ChatSessionID: session.ID,
|
|
Limit: int32(limit + 2),
|
|
BeforeCreatedAt: beforeCreatedAt,
|
|
BeforeID: beforeID,
|
|
})
|
|
if err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to list chat messages")
|
|
return
|
|
}
|
|
messages = visibleChatMessages(messages)
|
|
hasMore := len(messages) > limit
|
|
if hasMore {
|
|
messages = messages[:limit]
|
|
}
|
|
var nextCursor *ChatMessagesCursorResponse
|
|
if hasMore && len(messages) > 0 {
|
|
oldest := messages[len(messages)-1]
|
|
nextCursor = &ChatMessagesCursorResponse{
|
|
CreatedAt: oldest.CreatedAt.Time.Format(time.RFC3339Nano),
|
|
ID: uuidToString(oldest.ID),
|
|
}
|
|
}
|
|
// SQL fetches newest windows first so the empty cursor opens at the recent
|
|
// tail. Reverse each cursor page before serializing to keep message order
|
|
// chronological within the viewport.
|
|
for i, j := 0, len(messages)-1; i < j; i, j = i+1, j-1 {
|
|
messages[i], messages[j] = messages[j], messages[i]
|
|
}
|
|
|
|
messageIDs := make([]pgtype.UUID, len(messages))
|
|
for i, m := range messages {
|
|
messageIDs[i] = m.ID
|
|
}
|
|
groupedAtt := h.groupChatMessageAttachments(r.Context(), workspaceID, messageIDs)
|
|
|
|
resp := make([]ChatMessageResponse, len(messages))
|
|
for i, m := range messages {
|
|
resp[i] = chatMessageToResponse(m, groupedAtt[uuidToString(m.ID)])
|
|
}
|
|
writeJSON(w, http.StatusOK, ChatMessagesPageResponse{
|
|
Messages: resp,
|
|
Limit: limit,
|
|
HasMore: hasMore,
|
|
NextCursor: nextCursor,
|
|
})
|
|
}
|
|
|
|
// PendingChatTaskResponse is returned by GetPendingChatTask — either the
|
|
// current in-flight task's id/status, or an empty object when none is active.
|
|
// CreatedAt is the anchor the frontend uses to time the chat StatusPill
|
|
// (elapsed seconds = now - CreatedAt). It must come from the server because
|
|
// optimistic seeds don't have a real task created_at and the timer needs to
|
|
// survive refresh / reopen.
|
|
type PendingChatTaskResponse struct {
|
|
TaskID string `json:"task_id,omitempty"`
|
|
Status string `json:"status,omitempty"`
|
|
CreatedAt string `json:"created_at,omitempty"`
|
|
// WaitReason explains a waiting_local_directory hold: which directory the
|
|
// task is parked on and, when known, the short id of the task holding it.
|
|
// Emitted only for that status — on any other one the column still carries
|
|
// the last hold's text, which would read as a live explanation for a task
|
|
// that is running fine. Absent on older servers, which the client renders
|
|
// as today's bare "Waiting for local directory".
|
|
WaitReason string `json:"wait_reason,omitempty"`
|
|
SupportsQueue bool `json:"supports_queue"`
|
|
QueuedTasks []QueuedChatTaskResponse `json:"queued_tasks,omitempty"`
|
|
}
|
|
|
|
// waitReasonForStatus gates the stored hold text on the status it describes.
|
|
// wait_reason is never cleared when a task resumes — the daemon writes it once
|
|
// on the way into the hold — so returning it unconditionally would attach "held
|
|
// by task abc12345" to a task that has been running for ten minutes.
|
|
func waitReasonForStatus(status string, reason pgtype.Text) string {
|
|
if status != "waiting_local_directory" || !reason.Valid {
|
|
return ""
|
|
}
|
|
return strings.TrimSpace(reason.String)
|
|
}
|
|
|
|
type QueuedChatTaskResponse struct {
|
|
TaskID string `json:"task_id"`
|
|
Status string `json:"status"`
|
|
CreatedAt string `json:"created_at"`
|
|
MessageID string `json:"message_id,omitempty"`
|
|
Content string `json:"content,omitempty"`
|
|
}
|
|
|
|
type PrioritizeQueuedChatTaskResponse struct {
|
|
TaskID string `json:"task_id"`
|
|
ActiveTaskID string `json:"active_task_id,omitempty"`
|
|
}
|
|
|
|
// MarkChatSessionRead clears the session's unread_since (→ has_unread=false)
|
|
// and broadcasts chat:session_read so other devices of the same user drop
|
|
// their badges.
|
|
func (h *Handler) MarkChatSessionRead(w http.ResponseWriter, r *http.Request) {
|
|
userID, ok := requireUserID(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
workspaceID := ctxWorkspaceID(r.Context())
|
|
sessionID := chi.URLParam(r, "sessionId")
|
|
|
|
session, ok := h.gatePublicChatSessionForUser(w, r, userID, workspaceID, sessionID)
|
|
if !ok {
|
|
return
|
|
}
|
|
|
|
if err := h.Queries.MarkChatSessionRead(r.Context(), session.ID); err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to mark session read")
|
|
return
|
|
}
|
|
|
|
resolvedSessionID := uuidToString(session.ID)
|
|
h.publishChat(protocol.EventChatSessionRead, workspaceID, "member", userID, resolvedSessionID, protocol.ChatSessionReadPayload{
|
|
ChatSessionID: resolvedSessionID,
|
|
})
|
|
|
|
w.WriteHeader(http.StatusNoContent)
|
|
}
|
|
|
|
// Deferred-cancellation draft restores (#5219)
|
|
|
|
// ChatDraftRestoreResponse is one recoverable composer draft: a deferred
|
|
// cancellation settled as empty-transcript after the cancel HTTP response
|
|
// returned, so the deleted prompt is persisted server-side until the
|
|
// creator's client applies and consumes it. Attachments are resolved from
|
|
// the stored ids at read time so they carry the normal handler URL policy.
|
|
type ChatDraftRestoreResponse struct {
|
|
ID string `json:"id"`
|
|
ChatSessionID string `json:"chat_session_id"`
|
|
TaskID string `json:"task_id"`
|
|
Content string `json:"content"`
|
|
Attachments []AttachmentResponse `json:"attachments,omitempty"`
|
|
CreatedAt string `json:"created_at"`
|
|
}
|
|
|
|
type ChatDraftRestoresResponse struct {
|
|
Restores []ChatDraftRestoreResponse `json:"restores"`
|
|
}
|
|
|
|
// ListChatDraftRestores returns the session's pending draft restores.
|
|
// Creator-only via loadChatSessionForUser — deliberately not the
|
|
// private-agent gate: the content is the caller's own deleted prompt, and
|
|
// losing agent access must not strand it server-side forever.
|
|
func (h *Handler) ListChatDraftRestores(w http.ResponseWriter, r *http.Request) {
|
|
userID, ok := requireUserID(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
workspaceID := ctxWorkspaceID(r.Context())
|
|
sessionID := chi.URLParam(r, "sessionId")
|
|
|
|
session, ok := h.loadChatSessionForUser(w, r, userID, workspaceID, sessionID)
|
|
if !ok {
|
|
return
|
|
}
|
|
|
|
rows, err := h.Queries.ListChatDraftRestoresBySession(r.Context(), session.ID)
|
|
if err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to list draft restores")
|
|
return
|
|
}
|
|
|
|
var attachmentIDs []pgtype.UUID
|
|
for _, row := range rows {
|
|
attachmentIDs = append(attachmentIDs, row.AttachmentIds...)
|
|
}
|
|
attachmentsByID := map[string]AttachmentResponse{}
|
|
if len(attachmentIDs) > 0 {
|
|
attRows, err := h.Queries.ListAttachmentsByIDs(r.Context(), db.ListAttachmentsByIDsParams{
|
|
AttachmentIds: attachmentIDs,
|
|
WorkspaceID: session.WorkspaceID,
|
|
})
|
|
if err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to load draft restore attachments")
|
|
return
|
|
}
|
|
attachmentMode := attachmentURLModeFromRequest(r)
|
|
for _, a := range attRows {
|
|
attachmentsByID[uuidToString(a.ID)] = h.attachmentToResponse(a, attachmentMode)
|
|
}
|
|
}
|
|
|
|
resp := ChatDraftRestoresResponse{Restores: make([]ChatDraftRestoreResponse, 0, len(rows))}
|
|
for _, row := range rows {
|
|
item := ChatDraftRestoreResponse{
|
|
ID: uuidToString(row.ID),
|
|
ChatSessionID: uuidToString(row.ChatSessionID),
|
|
TaskID: uuidToString(row.TaskID),
|
|
Content: row.Content,
|
|
CreatedAt: timestampToString(row.CreatedAt),
|
|
}
|
|
for _, attID := range row.AttachmentIds {
|
|
if a, ok := attachmentsByID[uuidToString(attID)]; ok {
|
|
item.Attachments = append(item.Attachments, a)
|
|
}
|
|
}
|
|
resp.Restores = append(resp.Restores, item)
|
|
}
|
|
writeJSON(w, http.StatusOK, resp)
|
|
}
|
|
|
|
// ConsumeChatDraftRestore deletes one draft restore after the creator's
|
|
// client applied it. Idempotent: consuming an already-consumed (or never
|
|
// existing) restore still returns 204, so a retried consume after a lost
|
|
// response can't fail the client.
|
|
func (h *Handler) ConsumeChatDraftRestore(w http.ResponseWriter, r *http.Request) {
|
|
userID, ok := requireUserID(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
workspaceID := ctxWorkspaceID(r.Context())
|
|
sessionID := chi.URLParam(r, "sessionId")
|
|
|
|
session, ok := h.loadChatSessionForUser(w, r, userID, workspaceID, sessionID)
|
|
if !ok {
|
|
return
|
|
}
|
|
restoreUUID, ok := parseUUIDOrBadRequest(w, chi.URLParam(r, "restoreId"), "restore id")
|
|
if !ok {
|
|
return
|
|
}
|
|
|
|
if _, err := h.Queries.DeleteChatDraftRestore(r.Context(), db.DeleteChatDraftRestoreParams{
|
|
ID: restoreUUID,
|
|
ChatSessionID: session.ID,
|
|
}); err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to consume draft restore")
|
|
return
|
|
}
|
|
w.WriteHeader(http.StatusNoContent)
|
|
}
|
|
|
|
// PendingChatTasksResponse is the aggregate view consumed by the FAB.
|
|
type PendingChatTasksResponse struct {
|
|
Tasks []PendingChatTaskItem `json:"tasks"`
|
|
}
|
|
|
|
type PendingChatTaskItem struct {
|
|
TaskID string `json:"task_id"`
|
|
Status string `json:"status"`
|
|
ChatSessionID string `json:"chat_session_id"`
|
|
}
|
|
|
|
type CancelledChatMessageResponse struct {
|
|
ChatSessionID string `json:"chat_session_id"`
|
|
MessageID string `json:"message_id"`
|
|
Content string `json:"content"`
|
|
RestoreToInput bool `json:"restore_to_input"`
|
|
Attachments []AttachmentResponse `json:"attachments,omitempty"`
|
|
}
|
|
|
|
type CancelTaskByUserResponse struct {
|
|
AgentTaskResponse
|
|
CancelledChatMessage *CancelledChatMessageResponse `json:"cancelled_chat_message,omitempty"`
|
|
}
|
|
|
|
// ListPendingChatTasks returns every in-flight chat task owned by the current
|
|
// user in this workspace. Drives the FAB's "running" indicator when the chat
|
|
// window is closed (no per-session query is subscribed). Tasks belonging to
|
|
// private agents the caller has lost access to are dropped from the response.
|
|
func (h *Handler) ListPendingChatTasks(w http.ResponseWriter, r *http.Request) {
|
|
userID, ok := requireUserID(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
workspaceID := ctxWorkspaceID(r.Context())
|
|
|
|
member, ok := h.workspaceMember(w, r, workspaceID)
|
|
if !ok {
|
|
return
|
|
}
|
|
actorType, actorID := h.resolveActor(r, userID, workspaceID)
|
|
allowed, ok := h.accessibleAgentIDs(r.Context(), workspaceID, actorType, actorID, member.Role)
|
|
if !ok {
|
|
writeError(w, http.StatusInternalServerError, "failed to resolve agent access")
|
|
return
|
|
}
|
|
|
|
// No accessible agents → every row would be filtered out anyway. Skip the
|
|
// DB round-trip and return an empty list (mirrors HasPendingChatTasks).
|
|
if len(allowed) == 0 {
|
|
writeJSON(w, http.StatusOK, PendingChatTasksResponse{Tasks: []PendingChatTaskItem{}})
|
|
return
|
|
}
|
|
|
|
rows, err := h.Queries.ListPendingChatTasksByCreator(r.Context(), db.ListPendingChatTasksByCreatorParams{
|
|
WorkspaceID: parseUUID(workspaceID),
|
|
CreatorID: parseUUID(userID),
|
|
})
|
|
if err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to list pending chat tasks")
|
|
return
|
|
}
|
|
|
|
// The pending query now returns cs.agent_id per row, so we can filter
|
|
// out private agents the caller has lost access to directly against the
|
|
// already-loaded `allowed` set — no second ListAllChatSessionsByCreator
|
|
// scan on this hot path (MUL-4159).
|
|
items := make([]PendingChatTaskItem, 0, len(rows))
|
|
for _, row := range rows {
|
|
agentID := uuidToString(row.AgentID)
|
|
if _, ok := allowed[agentID]; !ok {
|
|
continue
|
|
}
|
|
items = append(items, PendingChatTaskItem{
|
|
TaskID: uuidToString(row.TaskID),
|
|
Status: row.Status,
|
|
ChatSessionID: uuidToString(row.ChatSessionID),
|
|
})
|
|
}
|
|
writeJSON(w, http.StatusOK, PendingChatTasksResponse{Tasks: items})
|
|
}
|
|
|
|
// HasPendingChatTasksResponse is the boolean fast-path payload consumed by the
|
|
// FAB, which only needs to know whether any in-flight chat task exists — not
|
|
// the full list.
|
|
type HasPendingChatTasksResponse struct {
|
|
HasPending bool `json:"has_pending"`
|
|
}
|
|
|
|
// HasPendingChatTasks answers "does the current user have any in-flight chat
|
|
// task in this workspace?" with a single EXISTS query, for the FAB's running
|
|
// indicator. It is the boolean sibling of ListPendingChatTasks: the detailed
|
|
// list is reserved for the ChatWindow history / stop-task flows.
|
|
//
|
|
// Permission filtering is preserved end-to-end: the set of agents the caller
|
|
// may currently see is resolved the same way as ListPendingChatTasks and
|
|
// pushed into the query as agent_ids, so a member who lost access to a private
|
|
// agent never sees a true from a task on that agent (MUL-4159). An empty
|
|
// accessible-agent set short-circuits to false without hitting the DB.
|
|
func (h *Handler) HasPendingChatTasks(w http.ResponseWriter, r *http.Request) {
|
|
userID, ok := requireUserID(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
workspaceID := ctxWorkspaceID(r.Context())
|
|
|
|
member, ok := h.workspaceMember(w, r, workspaceID)
|
|
if !ok {
|
|
return
|
|
}
|
|
actorType, actorID := h.resolveActor(r, userID, workspaceID)
|
|
allowed, ok := h.accessibleAgentIDs(r.Context(), workspaceID, actorType, actorID, member.Role)
|
|
if !ok {
|
|
writeError(w, http.StatusInternalServerError, "failed to resolve agent access")
|
|
return
|
|
}
|
|
|
|
// No accessible agents → nothing the caller may see can be pending.
|
|
// Skip the round-trip and return false.
|
|
if len(allowed) == 0 {
|
|
writeJSON(w, http.StatusOK, HasPendingChatTasksResponse{HasPending: false})
|
|
return
|
|
}
|
|
|
|
agentIDs := make([]pgtype.UUID, 0, len(allowed))
|
|
for id := range allowed {
|
|
agentIDs = append(agentIDs, parseUUID(id))
|
|
}
|
|
|
|
hasPending, err := h.Queries.HasPendingChatTasksByCreator(r.Context(), db.HasPendingChatTasksByCreatorParams{
|
|
WorkspaceID: parseUUID(workspaceID),
|
|
CreatorID: parseUUID(userID),
|
|
AgentIds: agentIDs,
|
|
})
|
|
if err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to check pending chat tasks")
|
|
return
|
|
}
|
|
|
|
writeJSON(w, http.StatusOK, HasPendingChatTasksResponse{HasPending: hasPending})
|
|
}
|
|
|
|
// GetPendingChatTask returns the current task plus its ordered follow-ups. The
|
|
// root fields retain the original response shape so older clients continue to
|
|
// recover a single pending task after refresh / reopen.
|
|
func (h *Handler) GetPendingChatTask(w http.ResponseWriter, r *http.Request) {
|
|
userID, ok := requireUserID(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
workspaceID := ctxWorkspaceID(r.Context())
|
|
sessionID := chi.URLParam(r, "sessionId")
|
|
|
|
session, ok := h.gatePublicChatSessionForUser(w, r, userID, workspaceID, sessionID)
|
|
if !ok {
|
|
return
|
|
}
|
|
|
|
tasks, err := h.Queries.ListPendingChatTasksForSession(r.Context(), session.ID)
|
|
if err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to list pending chat tasks")
|
|
return
|
|
}
|
|
if len(tasks) == 0 {
|
|
writeJSON(w, http.StatusOK, PendingChatTaskResponse{SupportsQueue: true})
|
|
return
|
|
}
|
|
|
|
head := tasks[0]
|
|
queued := make([]QueuedChatTaskResponse, 0, len(tasks)-1)
|
|
for _, task := range tasks[1:] {
|
|
if task.Status != "queued" {
|
|
continue
|
|
}
|
|
queued = append(queued, QueuedChatTaskResponse{
|
|
TaskID: uuidToString(task.ID),
|
|
Status: task.Status,
|
|
CreatedAt: timestampToString(task.CreatedAt),
|
|
MessageID: uuidToString(task.MessageID),
|
|
Content: task.Content,
|
|
})
|
|
}
|
|
|
|
writeJSON(w, http.StatusOK, PendingChatTaskResponse{
|
|
TaskID: uuidToString(head.ID),
|
|
Status: head.Status,
|
|
CreatedAt: timestampToString(head.CreatedAt),
|
|
WaitReason: waitReasonForStatus(head.Status, head.WaitReason),
|
|
SupportsQueue: true,
|
|
QueuedTasks: queued,
|
|
})
|
|
}
|
|
|
|
// PrioritizeQueuedChatTask moves one queued follow-up ahead of its FIFO peers.
|
|
// The client then cancels the current task through the existing cancellation
|
|
// endpoint, preserving that path's transcript and draft-restore semantics.
|
|
func (h *Handler) PrioritizeQueuedChatTask(w http.ResponseWriter, r *http.Request) {
|
|
userID, ok := requireUserID(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
workspaceID := ctxWorkspaceID(r.Context())
|
|
sessionID := chi.URLParam(r, "sessionId")
|
|
taskID, ok := parseUUIDOrBadRequest(w, chi.URLParam(r, "taskId"), "task id")
|
|
if !ok {
|
|
return
|
|
}
|
|
|
|
session, ok := h.gatePublicChatSessionForUser(w, r, userID, workspaceID, sessionID)
|
|
if !ok {
|
|
return
|
|
}
|
|
tx, err := h.TxStarter.Begin(r.Context())
|
|
if err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to start prioritize transaction")
|
|
return
|
|
}
|
|
defer tx.Rollback(r.Context())
|
|
qtx := h.Queries.WithTx(tx)
|
|
|
|
// ClaimTask takes this same row lock before selecting work. Holding it here
|
|
// makes "prioritize + report the active task" one server-authoritative
|
|
// decision instead of two client requests racing the daemon.
|
|
if _, err := qtx.GetAgentForClaimUpdate(r.Context(), session.AgentID); err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to lock chat agent")
|
|
return
|
|
}
|
|
prioritized, err := qtx.PrioritizeQueuedChatTask(
|
|
r.Context(),
|
|
db.PrioritizeQueuedChatTaskParams{ID: taskID, ChatSessionID: session.ID},
|
|
)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
// The compare-and-set rejects both a stale queue row and a still-valid
|
|
// follow-up that has no claimed reply to replace. Distinguish them for
|
|
// current clients while preserving the same 409 compatibility contract.
|
|
queuedTask, loadErr := qtx.GetAgentTask(r.Context(), taskID)
|
|
if loadErr != nil && !errors.Is(loadErr, pgx.ErrNoRows) {
|
|
writeError(w, http.StatusInternalServerError, "failed to load queued task")
|
|
return
|
|
}
|
|
if loadErr == nil &&
|
|
queuedTask.Status == "queued" &&
|
|
queuedTask.ChatSessionID.Valid &&
|
|
uuidToString(queuedTask.ChatSessionID) == uuidToString(session.ID) {
|
|
writeError(w, http.StatusConflict, "there is no active reply to replace")
|
|
return
|
|
}
|
|
writeError(w, http.StatusConflict, "task is no longer queued")
|
|
return
|
|
}
|
|
if err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to prioritize queued task")
|
|
return
|
|
}
|
|
queuedTask, err := qtx.GetAgentTask(r.Context(), taskID)
|
|
if err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to load prioritized task")
|
|
return
|
|
}
|
|
if err := tx.Commit(r.Context()); err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to commit queued task priority")
|
|
return
|
|
}
|
|
h.TaskService.BroadcastTaskQueued(r.Context(), queuedTask)
|
|
|
|
writeJSON(w, http.StatusOK, PrioritizeQueuedChatTaskResponse{
|
|
TaskID: uuidToString(prioritized.TaskID),
|
|
ActiveTaskID: uuidToString(prioritized.ActiveTaskID),
|
|
})
|
|
}
|
|
|
|
// ClearQueuedChatTasks cancels every queued follow-up without touching the
|
|
// session's current positional head, even when that head is not claimed yet.
|
|
func (h *Handler) ClearQueuedChatTasks(w http.ResponseWriter, r *http.Request) {
|
|
userID, ok := requireUserID(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
workspaceID := ctxWorkspaceID(r.Context())
|
|
sessionID := chi.URLParam(r, "sessionId")
|
|
session, ok := h.gatePublicChatSessionForUser(w, r, userID, workspaceID, sessionID)
|
|
if !ok {
|
|
return
|
|
}
|
|
if err := h.TaskService.CancelQueuedChatTasks(r.Context(), session.ID, session.AgentID); err != nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to clear queued tasks")
|
|
return
|
|
}
|
|
w.WriteHeader(http.StatusNoContent)
|
|
}
|
|
|
|
// Task cancellation (user-facing, with ownership check)
|
|
|
|
// CancelTaskByUser cancels a task the caller is allowed to act on within the
|
|
// current workspace.
|
|
//
|
|
// Tenancy is enforced uniformly through the task's owning agent: every
|
|
// agent_task_queue row carries a NOT NULL agent_id (ON DELETE CASCADE, so the
|
|
// agent always exists), and agents are workspace-scoped. GetAgentTaskInWorkspace
|
|
// is therefore the single tenant guard that works regardless of which optional
|
|
// source FK (issue / chat_session / autopilot_run) is set — which is what makes
|
|
// run_only autopilot tasks and quick_create tasks (whose issue does not exist
|
|
// yet) cancellable at all. Keying cancellation off issue_id / chat_session_id
|
|
// alone is exactly what 404'd these tasks before (MUL-2827).
|
|
//
|
|
// On top of tenancy, two privacy models layer on:
|
|
// - a chat task is private to the member who started the conversation, so
|
|
// only that creator may cancel it;
|
|
// - every other task surfaces on the agent Activity tab and the workspace
|
|
// task snapshot, both of which hide private agents from members without
|
|
// access. Cancellation mirrors that gate via canAccessPrivateAgent so the
|
|
// id-only endpoint is never more permissive than the surface that exposes
|
|
// the task.
|
|
func (h *Handler) CancelTaskByUser(w http.ResponseWriter, r *http.Request) {
|
|
userID, ok := requireUserID(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
workspaceID := ctxWorkspaceID(r.Context())
|
|
wsUUID, ok := parseUUIDOrBadRequest(w, workspaceID, "workspace id")
|
|
if !ok {
|
|
return
|
|
}
|
|
taskID := chi.URLParam(r, "taskId")
|
|
taskUUID, ok := parseUUIDOrBadRequest(w, taskID, "task id")
|
|
if !ok {
|
|
return
|
|
}
|
|
|
|
task, err := h.Queries.GetAgentTaskInWorkspace(r.Context(), db.GetAgentTaskInWorkspaceParams{
|
|
ID: taskUUID,
|
|
WorkspaceID: wsUUID,
|
|
})
|
|
if err != nil {
|
|
writeError(w, http.StatusNotFound, "task not found")
|
|
return
|
|
}
|
|
actorType, actorID := h.resolveActor(r, userID, workspaceID)
|
|
|
|
var (
|
|
queuedOnly bool
|
|
expectedSession pgtype.UUID
|
|
queueAction string
|
|
)
|
|
if expectedStatus := r.URL.Query().Get("expected_status"); expectedStatus != "" {
|
|
if expectedStatus != "queued" {
|
|
writeError(w, http.StatusBadRequest, "expected_status must be queued")
|
|
return
|
|
}
|
|
sessionID := r.URL.Query().Get("chat_session_id")
|
|
var ok bool
|
|
expectedSession, ok = parseUUIDOrBadRequest(w, sessionID, "chat_session_id")
|
|
if !ok {
|
|
return
|
|
}
|
|
if !task.ChatSessionID.Valid || task.ChatSessionID != expectedSession {
|
|
writeError(w, http.StatusConflict, "task does not belong to the expected chat session")
|
|
return
|
|
}
|
|
queueAction = r.URL.Query().Get("queue_action")
|
|
if queueAction != "edit" && queueAction != "remove" {
|
|
writeError(w, http.StatusBadRequest, "queue_action must be edit or remove")
|
|
return
|
|
}
|
|
queuedOnly = true
|
|
}
|
|
|
|
if task.ChatSessionID.Valid {
|
|
// Chat privacy: only the member who opened the conversation may
|
|
// cancel its task, even though the workspace is shared.
|
|
cs, err := h.Queries.GetChatSessionInWorkspace(r.Context(), db.GetChatSessionInWorkspaceParams{
|
|
ID: task.ChatSessionID,
|
|
WorkspaceID: wsUUID,
|
|
})
|
|
if err != nil {
|
|
writeError(w, http.StatusNotFound, "task not found")
|
|
return
|
|
}
|
|
if uuidToString(cs.CreatorID) != userID {
|
|
writeError(w, http.StatusForbidden, "not your task")
|
|
return
|
|
}
|
|
} else {
|
|
// Issue / autopilot / quick_create tasks are all visible on the
|
|
// agent Activity tab + workspace snapshot, which gate private
|
|
// agents. Mirror that gate here.
|
|
agent, err := h.Queries.GetAgentInWorkspace(r.Context(), db.GetAgentInWorkspaceParams{
|
|
ID: task.AgentID,
|
|
WorkspaceID: wsUUID,
|
|
})
|
|
if err != nil {
|
|
writeError(w, http.StatusNotFound, "task not found")
|
|
return
|
|
}
|
|
if !h.canAccessPrivateAgent(r.Context(), agent, actorType, actorID, workspaceID) {
|
|
writeError(w, http.StatusForbidden, "you do not have access to this agent")
|
|
return
|
|
}
|
|
}
|
|
|
|
cancelled, err := h.TaskService.CancelTaskWithResult(r.Context(), taskUUID, service.CancelTaskOptions{
|
|
ClientSupportsDraftRestore: requestHasClientCapability(r, protocol.AppCapabilityChatDraftRestoreV1),
|
|
QueuedOnly: queuedOnly,
|
|
ExpectedChatSession: expectedSession,
|
|
QueueAction: queueAction,
|
|
CancelledBy: h.taskCancellationActor(r.Context(), actorType, actorID),
|
|
UserInitiated: true,
|
|
})
|
|
if errors.Is(err, service.ErrTaskNoLongerQueued) {
|
|
writeError(w, http.StatusConflict, err.Error())
|
|
return
|
|
}
|
|
if err != nil {
|
|
writeError(w, http.StatusBadRequest, err.Error())
|
|
return
|
|
}
|
|
|
|
resp := CancelTaskByUserResponse{
|
|
AgentTaskResponse: taskToResponse(cancelled.Task, workspaceID),
|
|
}
|
|
h.hydrateTaskAttributions(r.Context(), []*TaskAttribution{resp.AgentTaskResponse.Attribution})
|
|
if cancelled.CancelledChatMessage != nil {
|
|
attachments := make([]AttachmentResponse, 0, len(cancelled.CancelledChatMessage.Attachments))
|
|
attachmentMode := attachmentURLModeFromRequest(r)
|
|
for _, a := range cancelled.CancelledChatMessage.Attachments {
|
|
attachments = append(attachments, h.attachmentToResponse(a, attachmentMode))
|
|
}
|
|
resp.CancelledChatMessage = &CancelledChatMessageResponse{
|
|
ChatSessionID: cancelled.CancelledChatMessage.ChatSessionID,
|
|
MessageID: cancelled.CancelledChatMessage.MessageID,
|
|
Content: cancelled.CancelledChatMessage.Content,
|
|
RestoreToInput: cancelled.CancelledChatMessage.RestoreToInput,
|
|
Attachments: attachments,
|
|
}
|
|
}
|
|
|
|
writeJSON(w, http.StatusOK, resp)
|
|
}
|
|
|
|
// Response types & helpers
|
|
|
|
type ChatSessionResponse struct {
|
|
ID string `json:"id"`
|
|
WorkspaceID string `json:"workspace_id"`
|
|
AgentID string `json:"agent_id"`
|
|
CreatorID string `json:"creator_id"`
|
|
ProjectID *string `json:"project_id"`
|
|
Title string `json:"title"`
|
|
Status string `json:"status"`
|
|
// Only populated by list endpoints — single-session fetches return 0/false/nil.
|
|
// HasUnread is kept as a convenience (== UnreadCount > 0) for existing consumers.
|
|
HasUnread bool `json:"has_unread"`
|
|
UnreadCount int `json:"unread_count"`
|
|
LastMessage *ChatLastMessage `json:"last_message"`
|
|
// Pinned marks a chat the user has stuck to the top of the list. Populated
|
|
// by list endpoints and by the pin/unpin + single-session responses.
|
|
Pinned bool `json:"pinned"`
|
|
// ChannelSource is present only for Chats created from an external channel.
|
|
// IsCurrentChannelRoute distinguishes the active route generation from an
|
|
// older Chat that remains readable and writable in Multica.
|
|
ChannelSource *ChatSessionChannelSourceResponse `json:"channel_source,omitempty"`
|
|
IsCurrentChannelRoute *bool `json:"is_current_channel_route,omitempty"`
|
|
CreatedAt string `json:"created_at"`
|
|
UpdatedAt string `json:"updated_at"`
|
|
}
|
|
|
|
type ChatSessionChannelSourceResponse struct {
|
|
ChannelType string `json:"channel_type"`
|
|
InstallationID string `json:"installation_id"`
|
|
RouteRevision int64 `json:"route_revision"`
|
|
}
|
|
|
|
func (h *Handler) hydrateChatSessionChannelMetadata(ctx context.Context, sessions []ChatSessionResponse) error {
|
|
if len(sessions) == 0 {
|
|
return nil
|
|
}
|
|
ids := make([]pgtype.UUID, 0, len(sessions))
|
|
for _, session := range sessions {
|
|
ids = append(ids, parseUUID(session.ID))
|
|
}
|
|
bindings, err := h.Queries.ListChannelChatSessionBindingsBySessions(ctx, ids)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
bySession := make(map[string]db.ChannelChatSessionBinding, len(bindings))
|
|
for _, binding := range bindings {
|
|
bySession[uuidToString(binding.ChatSessionID)] = binding
|
|
}
|
|
for i := range sessions {
|
|
binding, ok := bySession[sessions[i].ID]
|
|
if !ok {
|
|
continue
|
|
}
|
|
current := !binding.RetiredAt.Valid
|
|
sessions[i].ChannelSource = &ChatSessionChannelSourceResponse{
|
|
ChannelType: binding.ChannelType, InstallationID: uuidToString(binding.InstallationID), RouteRevision: binding.RouteRevision,
|
|
}
|
|
sessions[i].IsCurrentChannelRoute = ¤t
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// ChatLastMessage is a preview of a session's most recent message, used to
|
|
// render the IM-style conversation list (name + snippet + time). nil when the
|
|
// session has no messages yet.
|
|
type ChatLastMessage struct {
|
|
Content string `json:"content"`
|
|
Role string `json:"role"`
|
|
CreatedAt string `json:"created_at"`
|
|
FailureReason *string `json:"failure_reason"`
|
|
// MessageKind is 'message' (default) or 'no_response'. Hidden onboarding
|
|
// kickoff rows make buildChatLastMessage return nil and are never exposed.
|
|
MessageKind string `json:"message_kind"`
|
|
}
|
|
|
|
// buildChatLastMessage assembles the preview from list-row columns; returns nil
|
|
// when there is no last message (the LEFT JOIN produced a NULL timestamp).
|
|
func buildChatLastMessage(at pgtype.Timestamptz, content, role string, failure pgtype.Text, kind string) *ChatLastMessage {
|
|
if !at.Valid || kind == protocol.ChatMessageKindOnboardingKickoff {
|
|
return nil
|
|
}
|
|
return &ChatLastMessage{
|
|
Content: content,
|
|
Role: role,
|
|
CreatedAt: timestampToString(at),
|
|
FailureReason: textToPtr(failure),
|
|
MessageKind: normalizeMessageKind(kind),
|
|
}
|
|
}
|
|
|
|
type ChatMessageResponse struct {
|
|
ID string `json:"id"`
|
|
ChatSessionID string `json:"chat_session_id"`
|
|
Role string `json:"role"`
|
|
Content string `json:"content"`
|
|
TaskID *string `json:"task_id"`
|
|
CreatedAt string `json:"created_at"`
|
|
// FailureReason flags an assistant row synthesized by FailTask's chat
|
|
// fallback. Front-end uses it to switch to the destructive bubble.
|
|
FailureReason *string `json:"failure_reason"`
|
|
// ElapsedMs is the wall-clock duration from task creation to terminal
|
|
// state. Drives "Replied in 38s" / "Failed after 12s" captions.
|
|
ElapsedMs *int64 `json:"elapsed_ms"`
|
|
// MessageKind is additive. User-facing list handlers filter onboarding
|
|
// kickoff rows; clients still understand that kind as a compatibility guard.
|
|
MessageKind string `json:"message_kind"`
|
|
// QuickActions are sanitized follow-ups generated with this assistant turn.
|
|
// Always an empty array for legacy rows and user messages.
|
|
QuickActions []protocol.ChatQuickAction `json:"quick_actions"`
|
|
// Attachments linked to this message via chat_message_id. The chat
|
|
// bubble renders file cards from these, and the daemon claim path
|
|
// (daemon.go) pulls structured metadata from the same source so the
|
|
// agent can `multica attachment download <id>` rather than guessing
|
|
// from a markdown URL that may expire.
|
|
Attachments []AttachmentResponse `json:"attachments,omitempty"`
|
|
}
|
|
|
|
func chatSessionToResponse(s db.ChatSession) ChatSessionResponse {
|
|
return ChatSessionResponse{
|
|
ID: uuidToString(s.ID),
|
|
WorkspaceID: uuidToString(s.WorkspaceID),
|
|
AgentID: uuidToString(s.AgentID),
|
|
CreatorID: uuidToString(s.CreatorID),
|
|
ProjectID: uuidToPtr(s.ProjectID),
|
|
Title: s.Title,
|
|
Status: s.Status,
|
|
Pinned: s.PinnedAt.Valid,
|
|
CreatedAt: timestampToString(s.CreatedAt),
|
|
UpdatedAt: timestampToString(s.UpdatedAt),
|
|
}
|
|
}
|
|
|
|
func chatMessageToResponse(m db.ChatMessage, attachments []AttachmentResponse) ChatMessageResponse {
|
|
return ChatMessageResponse{
|
|
ID: uuidToString(m.ID),
|
|
ChatSessionID: uuidToString(m.ChatSessionID),
|
|
Role: m.Role,
|
|
Content: m.Content,
|
|
TaskID: uuidToPtr(m.TaskID),
|
|
CreatedAt: timestampToString(m.CreatedAt),
|
|
FailureReason: textToPtr(m.FailureReason),
|
|
ElapsedMs: int8ToPtr(m.ElapsedMs),
|
|
MessageKind: normalizeMessageKind(m.MessageKind),
|
|
QuickActions: decodeChatQuickActions(m.QuickActions),
|
|
Attachments: attachments,
|
|
}
|
|
}
|
|
|
|
// visibleChatMessages removes product-authored context that is sent to the
|
|
// agent runtime but is never part of the member-visible conversation. The
|
|
// daemon reads its task-scoped input through ListChatInputMessages, so this
|
|
// user-facing filter does not remove the kickoff from Mika's execution.
|
|
func visibleChatMessages(messages []db.ChatMessage) []db.ChatMessage {
|
|
visible := make([]db.ChatMessage, 0, len(messages))
|
|
for _, message := range messages {
|
|
if message.MessageKind == protocol.ChatMessageKindOnboardingKickoff {
|
|
continue
|
|
}
|
|
visible = append(visible, message)
|
|
}
|
|
return visible
|
|
}
|
|
|
|
func decodeChatQuickActions(raw []byte) []protocol.ChatQuickAction {
|
|
actions := make([]protocol.ChatQuickAction, 0)
|
|
if len(raw) == 0 {
|
|
return actions
|
|
}
|
|
if err := json.Unmarshal(raw, &actions); err != nil {
|
|
return []protocol.ChatQuickAction{}
|
|
}
|
|
if actions == nil {
|
|
return []protocol.ChatQuickAction{}
|
|
}
|
|
if len(actions) > chatQuickActionResponseLimit {
|
|
actions = actions[:chatQuickActionResponseLimit]
|
|
}
|
|
return actions
|
|
}
|
|
|
|
const chatQuickActionResponseLimit = 3
|
|
|
|
// normalizeMessageKind maps a stored chat_message.message_kind to the value the
|
|
// API exposes. Unknown / empty kinds degrade to 'message' so a future kind
|
|
// never surprises an older client into a broken render (MUL-4351).
|
|
func normalizeMessageKind(kind string) string {
|
|
switch kind {
|
|
case protocol.ChatMessageKindNoResponse:
|
|
return protocol.ChatMessageKindNoResponse
|
|
case protocol.ChatMessageKindOnboardingKickoff:
|
|
return protocol.ChatMessageKindOnboardingKickoff
|
|
case protocol.ChatMessageKindOnboardingOpening:
|
|
return protocol.ChatMessageKindOnboardingOpening
|
|
default:
|
|
return protocol.ChatMessageKindMessage
|
|
}
|
|
}
|