diff --git a/cmd/backend/main.go b/cmd/backend/main.go index 4b2abf4a..e3d97e80 100644 --- a/cmd/backend/main.go +++ b/cmd/backend/main.go @@ -23,6 +23,7 @@ import ( "github.com/warmbly/warmbly/internal/app/admin" "github.com/warmbly/warmbly/internal/app/adminoutreach" "github.com/warmbly/warmbly/internal/app/advanced" + "github.com/warmbly/warmbly/internal/app/aiagent" "github.com/warmbly/warmbly/internal/app/aitools" "github.com/warmbly/warmbly/internal/app/analytics" "github.com/warmbly/warmbly/internal/app/apikey" @@ -146,6 +147,7 @@ func main() { var aiProvider generation.Provider var aiSearch generation.SearchClient var aiToolRegistry *aitools.Registry + var aiAgentService aiagent.Service var emailVerifyService emailverifyapp.Service var placementRepository repository.PlacementRepository var placementService placement.Service @@ -994,6 +996,15 @@ func main() { FeatureGate: featureGateService, AppBaseURL: cfg.GetStringOptional(ctx, "APP_BASE_URL", "app_base_url", ""), }) + + // Dashboard AI agent: sessions + streamed, approval-gated, credit-charged + // runs over the tool registry. Only constructed when a provider is set. + if aiProvider != nil { + aiAgentService = aiagent.NewService( + repository.NewAgentRepository(primaryDB), + aiToolRegistry, aiProvider, creditService, featureGateService, auditService, + ) + } advancedService = advanced.NewService( advancedRepository, campaignRepostory, @@ -1268,6 +1279,7 @@ func main() { AIProvider: aiProvider, AISearch: aiSearch, AITools: aiToolRegistry, + AIAgentService: aiAgentService, // Pre-send email verification EmailVerifyService: emailVerifyService, diff --git a/docs/content/docs/api/endpoints.mdx b/docs/content/docs/api/endpoints.mdx index 832907d3..d144cf57 100644 --- a/docs/content/docs/api/endpoints.mdx +++ b/docs/content/docs/api/endpoints.mdx @@ -209,6 +209,17 @@ Credit balance, top-up purchases, and the transaction log are part of `/subscrip | GET | `/subscription/credits/transactions` | `manage_billing` | | POST | `/subscription/credits/checkout` | `manage_billing` | +### AI assistant + +The dashboard AI assistant is JWT only: sessions are private to the member who started them. The message and approval runs stream over Server-Sent Events. Each tool the assistant runs is gated by the member's own organization permission bits, so the assistant can never do more than the member could by hand. See the [AI assistant](/guides/ai-assistant/) guide. (API-key callers reach the same tools through the [MCP server](/api/mcp/), gated by the `AI_AGENT` scope.) + +| Method | Path | JWT permission | +|--------|------|----------------| +| POST | `/ai/sessions` | organization member | +| GET | `/ai/sessions` | organization member | +| POST | `/ai/sessions/:id/messages` | organization member (SSE) | +| POST | `/ai/sessions/:id/approve` | organization member (SSE) | + ## Public - `GET /health` diff --git a/docs/content/docs/api/error-codes.mdx b/docs/content/docs/api/error-codes.mdx index 7825fe0e..02823437 100644 --- a/docs/content/docs/api/error-codes.mdx +++ b/docs/content/docs/api/error-codes.mdx @@ -29,11 +29,12 @@ All errors follow this structure: |------|-------|-------------| | 400 | Bad Request | Invalid request syntax or parameters | | 401 | Unauthorized | Missing or invalid authentication | +| 402 | Payment Required | Out of AI credits (`insufficient_credits`) | | 403 | Forbidden | Authenticated but lacks permission | | 404 | Not Found | Resource doesn't exist | | 409 | Conflict | Resource already exists | | 422 | Unprocessable | Validation failed | -| 429 | Too Many Requests | Rate limit exceeded | +| 429 | Too Many Requests | Rate limit or AI usage cap exceeded (`rate_limit_exceeded`, `usage_cap_exceeded`) | ### Server errors (5xx) @@ -98,6 +99,23 @@ Returned when authentication fails. - Check if your key has expired or been revoked - Generate a new key if necessary +### 402 Payment Required + +Returned when an AI action is requested but the organization is out of credits. The response carries the stable code `insufficient_credits`. + +```json +{ + "error": "Payment Required", + "message": "You're out of AI credits. Add more to keep using the assistant.", + "code": "insufficient_credits", + "request_id": "req_..." +} +``` + +**How to fix:** +- Wait for the monthly allowance to reset, or buy a top-up pack (see [AI credits](/guides/ai-credits/)) +- Related: a `429` with code `usage_cap_exceeded` means a short-term AI usage cap was hit; retry later + ### 403 Forbidden Returned when authenticated but lacking necessary permissions. diff --git a/docs/content/docs/api/permissions.mdx b/docs/content/docs/api/permissions.mdx index 174964a0..c4272966 100644 --- a/docs/content/docs/api/permissions.mdx +++ b/docs/content/docs/api/permissions.mdx @@ -34,6 +34,7 @@ These same permissions are the [OAuth](/api/oauth/) scopes, lowercased: `READ_EM | `READ_AUDIT_LOGS` | 19 | 524288 | read | View organization audit logs | | `INTEGRATIONS` | 20 | 1048576 | special | Connect and manage third-party integrations | | `WARMUP_ROUTING` | 21 | 2097152 | special | Manage warmup routing rules | +| `AI_AGENT` | 22 | 4194304 | special | Run the AI assistant and MCP tools | `SEND_CAMPAIGNS` is intentionally separate from `WRITE_CAMPAIGNS`: editing a campaign draft and starting one that actually transmits mail are different blast radii, so a key can be granted the first without the second. @@ -53,7 +54,7 @@ High-volume operations. These can touch large numbers of rows in a single reques ### Special -Realtime subscriptions, webhooks, integrations, warmup routing, and self-service key management. Grant individually. +Realtime subscriptions, webhooks, integrations, warmup routing, AI assistant / MCP access, and self-service key management. Grant individually. ## Preset combinations @@ -70,12 +71,12 @@ READ_EMAILS | READ_CAMPAIGNS | READ_CONTACTS | READ_UNIBOX | READ_ANALYTICS = 688159 ``` -### Full access (4194303) +### Full access (8388607) -All 22 permissions: +All 23 permissions: ``` -(1 << 22) - 1 = 4194303 +(1 << 23) - 1 = 8388607 ``` ## Working with bitmasks @@ -178,6 +179,7 @@ export const Permissions = { READ_AUDIT_LOGS: 524288, INTEGRATIONS: 1048576, WARMUP_ROUTING: 2097152, + AI_AGENT: 4194304, } as const; ``` @@ -209,6 +211,7 @@ class Permissions: READ_AUDIT_LOGS = 524288 INTEGRATIONS = 1048576 WARMUP_ROUTING = 2097152 + AI_AGENT = 4194304 ``` @@ -239,6 +242,7 @@ const ( APIPermReadAuditLogs // 524288 APIPermIntegrations // 1048576 APIPermWarmupRouting // 2097152 + APIPermAIAgent // 4194304 ) ``` diff --git a/docs/content/docs/guides/ai-assistant.mdx b/docs/content/docs/guides/ai-assistant.mdx new file mode 100644 index 00000000..3052dce7 --- /dev/null +++ b/docs/content/docs/guides/ai-assistant.mdx @@ -0,0 +1,55 @@ +--- +title: AI assistant +description: The in-product AI assistant that searches your data, drafts replies, and sets up draft campaigns and automations, asking before it changes anything. +icon: Sparkles +--- + +The AI assistant is a chat panel built into the dashboard. It can look things up across your contacts, campaigns, unified inbox, and CRM, and it can take actions on your behalf, but it always asks before changing anything, and it never sends email for you. + +Open it from the sparkle button in the top bar, or press `Cmd/Ctrl + I`. The panel stays with you as you move between pages, so you can keep a conversation going while you work. + + +The assistant acts as you. It can only see and do what your own role allows, so it can never reach data or actions you could not reach by hand. It drafts replies but never sends them, and it never starts a campaign on its own. + + +## What you can ask + +The assistant has a set of tools it uses to answer you: + +- Find and read contacts, and update contact fields or tags +- List campaigns and read a campaign's send, open, click, and reply stats +- Read your unified inbox threads and draft a reply for you to review +- Create a CRM task or deal +- Create a draft campaign or a disabled automation for you to finish and turn on +- Search the web and read a public page + +Because it knows what page you are on, you can say things like "summarize this campaign" or "draft a reply here" and it understands what "this" and "here" mean. + +## How actions are approved + +The assistant splits what it does into two kinds: + +- **Reading** runs automatically. Looking up a contact or checking campaign stats needs no confirmation. +- **Changing** always asks first. When the assistant wants to update a contact, create a task, or set up a draft campaign, it pauses and shows you an approval card with a plain-language summary of exactly what it will do. + +On an approval card you can: + +- **Approve** to run just that action +- **Skip** to decline it and let the assistant carry on differently +- **Always allow** to let that kind of action run automatically from now on, for your whole workspace + +Sending is different. If an action would send mail it is never auto-approved, and the card shows the full message with an explicit send step. In this version the assistant only ever drafts replies; you send them yourself from the composer. + +## Draft campaigns and automations + +When you ask the assistant to build a campaign or an automation, it creates it as a **draft** (a campaign is never started, an automation is left disabled) and gives you a button to open it in the real editor. You review, adjust, and turn it on yourself. Nothing the assistant builds goes live on its own. + +## Credits and limits + +Each step the assistant takes costs one AI credit, and a single request is capped at a set number of steps (shown as the run progresses) so a question can never run away. The credits remaining are shown at the bottom of the panel. If you run out, the assistant stops and tells you; see [AI credits](/guides/ai-credits/) for how allowances and top-ups work. + +You can stop a run at any time with the stop button, and start a fresh conversation with **New**. + +## Privacy + +Conversations are private to you. Other members of your workspace do not see your assistant sessions. When the assistant changes something (creates a task, updates a contact), that change shows up in the shared activity feed like any other edit, so your teammates' views stay current. diff --git a/docs/content/docs/guides/meta.json b/docs/content/docs/guides/meta.json index 40fe9c13..bfcedacc 100644 --- a/docs/content/docs/guides/meta.json +++ b/docs/content/docs/guides/meta.json @@ -24,6 +24,7 @@ "team-roles", "collaboration", "referral-program", - "ai-credits" + "ai-credits", + "ai-assistant" ] } diff --git a/internal/api/handler/ai_agent.go b/internal/api/handler/ai_agent.go new file mode 100644 index 00000000..114c0115 --- /dev/null +++ b/internal/api/handler/ai_agent.go @@ -0,0 +1,224 @@ +// Dashboard AI agent endpoints. Sessions and their message runs are per-user; +// message and approval runs stream over SSE (text deltas, tool step events, +// approval_required, done with credits_remaining). The run executes in the +// request context, so a client that aborts the fetch cancels the run (the stop +// mechanism). Tools execute AS the invoking member with their org permission +// bits enforced by the registry. +package handler + +import ( + "encoding/json" + "fmt" + "net/http" + + "github.com/gin-gonic/gin" + "github.com/google/uuid" + + "github.com/warmbly/warmbly/internal/api/middleware" + "github.com/warmbly/warmbly/internal/app/aiagent" + "github.com/warmbly/warmbly/internal/app/aitools" + "github.com/warmbly/warmbly/internal/errx" + "github.com/warmbly/warmbly/internal/utils/paging" +) + +// jwtInvocation builds a tool invocation for a JWT (dashboard) caller: it runs +// as the member with their org permission bits, never an API key. +func (h *Handler) jwtInvocation(c *gin.Context) (aitools.Invocation, *errx.Error) { + orgID := middleware.GetOrganizationID(c) + if orgID == nil { + return aitools.Invocation{}, errx.New(errx.BadRequest, "no organization selected") + } + userID, err := middleware.GetUserUUID(c) + if err != nil { + return aitools.Invocation{}, errx.New(errx.Unauthorized, "invalid user") + } + member, xerr := h.OrganizationService.GetMembership(c.Request.Context(), *orgID, userID) + if xerr != nil || member == nil { + return aitools.Invocation{}, errx.New(errx.Forbidden, "not a member of this organization") + } + return aitools.Invocation{ + OrgID: *orgID, + UserID: userID, + OrgPerms: member.Permissions, + IsAPIKey: false, + IP: c.ClientIP(), + UserAgent: c.Request.UserAgent(), + }, nil +} + +// CreateAgentSession — POST /ai/sessions +func (h *Handler) CreateAgentSession(c *gin.Context) { + if h.AIAgentService == nil { + errx.JSON(c, errx.New(errx.ServiceUnavailable, "the AI assistant is not configured")) + return + } + inv, xerr := h.jwtInvocation(c) + if xerr != nil { + errx.JSON(c, xerr) + return + } + var req struct { + Page string `json:"page"` + Resource string `json:"resource"` + } + _ = c.ShouldBindJSON(&req) + + sess, serr := h.AIAgentService.CreateSession(c.Request.Context(), inv.OrgID, inv.UserID, req.Page, req.Resource) + if serr != nil { + errx.JSON(c, serr) + return + } + c.JSON(http.StatusOK, sess) +} + +// ListAgentSessions — GET /ai/sessions (cursor paginated, newest first) +func (h *Handler) ListAgentSessions(c *gin.Context) { + if h.AIAgentService == nil { + errx.JSON(c, errx.New(errx.ServiceUnavailable, "the AI assistant is not configured")) + return + } + inv, xerr := h.jwtInvocation(c) + if xerr != nil { + errx.JSON(c, xerr) + return + } + limit, xerr := parseCreditLimit(c.Query("limit"), 25, 100) + if xerr != nil { + errx.JSON(c, xerr) + return + } + beforeAt, beforeID, xerr := paging.DecodeTimeCursor(c.Query("cursor")) + if xerr != nil { + errx.JSON(c, xerr) + return + } + sessions, err := h.AIAgentService.ListSessions(c.Request.Context(), inv.OrgID, inv.UserID, limit+1, beforeAt, beforeID) + if err != nil { + errx.JSON(c, errx.New(errx.Internal, "failed to list sessions")) + return + } + var nextCursor *string + if len(sessions) > limit { + last := sessions[limit-1] + nextCursor = paging.EncodeTime(last.CreatedAt, last.ID) + sessions = sessions[:limit] + } + c.JSON(http.StatusOK, gin.H{ + "data": sessions, + "pagination": gin.H{"next_cursor": nextCursor, "has_more": nextCursor != nil}, + }) +} + +// AgentMessage — POST /ai/sessions/:id/messages (SSE) +func (h *Handler) AgentMessage(c *gin.Context) { + if h.AIAgentService == nil { + errx.JSON(c, errx.New(errx.ServiceUnavailable, "the AI assistant is not configured")) + return + } + inv, xerr := h.jwtInvocation(c) + if xerr != nil { + errx.JSON(c, xerr) + return + } + sessionID, err := uuid.Parse(c.Param("id")) + if err != nil { + errx.JSON(c, errx.New(errx.BadRequest, "invalid session id")) + return + } + var req struct { + MessageID string `json:"message_id"` + Text string `json:"text"` + Page string `json:"page"` + Resource string `json:"resource"` + } + if err := c.ShouldBindJSON(&req); err != nil { + errx.JSON(c, errx.New(errx.BadRequest, "invalid request body")) + return + } + if req.MessageID == "" { + req.MessageID = uuid.NewString() + } + + emit := sseEmitter(c) + if serr := h.AIAgentService.RunMessage(c.Request.Context(), inv, sessionID, req.MessageID, req.Text, req.Page, req.Resource, emit); serr != nil { + emit(aiagent.StreamEvent{Type: "error", Code: string(codeIdentifier(serr)), Message: serr.Message}) + } +} + +// AgentApprove — POST /ai/sessions/:id/approve (SSE) resumes a paused run. +func (h *Handler) AgentApprove(c *gin.Context) { + if h.AIAgentService == nil { + errx.JSON(c, errx.New(errx.ServiceUnavailable, "the AI assistant is not configured")) + return + } + inv, xerr := h.jwtInvocation(c) + if xerr != nil { + errx.JSON(c, xerr) + return + } + sessionID, err := uuid.Parse(c.Param("id")) + if err != nil { + errx.JSON(c, errx.New(errx.BadRequest, "invalid session id")) + return + } + var req struct { + Decision string `json:"decision"` // approve | deny | always_allow + } + if err := c.ShouldBindJSON(&req); err != nil { + errx.JSON(c, errx.New(errx.BadRequest, "invalid request body")) + return + } + switch req.Decision { + case "approve", "deny", "always_allow": + default: + errx.JSON(c, errx.New(errx.BadRequest, "decision must be approve, deny, or always_allow")) + return + } + + emit := sseEmitter(c) + if serr := h.AIAgentService.Resume(c.Request.Context(), inv, sessionID, req.Decision, emit); serr != nil { + emit(aiagent.StreamEvent{Type: "error", Code: string(codeIdentifier(serr)), Message: serr.Message}) + } +} + +// sseEmitter prepares the response for Server-Sent Events and returns a +// flush-per-event emitter. Safe to call once per request. +func sseEmitter(c *gin.Context) func(aiagent.StreamEvent) { + h := c.Writer.Header() + h.Set("Content-Type", "text/event-stream") + h.Set("Cache-Control", "no-cache") + h.Set("Connection", "keep-alive") + h.Set("X-Accel-Buffering", "no") // disable proxy buffering so deltas flush + c.Writer.WriteHeader(http.StatusOK) + flusher, _ := c.Writer.(http.Flusher) + if flusher != nil { + flusher.Flush() + } + return func(ev aiagent.StreamEvent) { + b, err := json.Marshal(ev) + if err != nil { + return + } + _, _ = fmt.Fprintf(c.Writer, "data: %s\n\n", b) + if flusher != nil { + flusher.Flush() + } + } +} + +// codeIdentifier gives a short machine code for an errx to surface in the SSE +// error event. +func codeIdentifier(e *errx.Error) string { + switch e.Code { + case errx.NotFound: + return "not_found" + case errx.BadRequest: + return "bad_request" + case errx.Forbidden: + return "forbidden" + case errx.ServiceUnavailable: + return "service_unavailable" + default: + return "error" + } +} diff --git a/internal/api/handler/handler.go b/internal/api/handler/handler.go index 7ade4052..c2203110 100644 --- a/internal/api/handler/handler.go +++ b/internal/api/handler/handler.go @@ -4,6 +4,7 @@ import ( "github.com/warmbly/warmbly/internal/app/admin" "github.com/warmbly/warmbly/internal/app/adminoutreach" "github.com/warmbly/warmbly/internal/app/advanced" + "github.com/warmbly/warmbly/internal/app/aiagent" "github.com/warmbly/warmbly/internal/app/aitools" "github.com/warmbly/warmbly/internal/app/analytics" "github.com/warmbly/warmbly/internal/app/apikey" @@ -158,6 +159,10 @@ type Handler struct { // run on. Handlers bound to the invoking user's permissions. AITools *aitools.Registry + // AIAgentService orchestrates the dashboard agent (sessions, SSE runs, + // approvals, per-iteration credits). Nil when no LLM provider is configured. + AIAgentService aiagent.Service + // Seed inbox-placement testing. PlacementRepo repository.PlacementRepository PlacementService placement.Service diff --git a/internal/api/routes.go b/internal/api/routes.go index 99b2e761..a87ce412 100644 --- a/internal/api/routes.go +++ b/internal/api/routes.go @@ -834,6 +834,19 @@ func Run( // Websocket bootstrap. The token returned here is single-session. jwtOnly.POST("/getaway", h.GenerateWebsocket) + // Dashboard AI agent (JWT-only; per-user sessions). Each tool the + // agent runs is gated by the invoking member's org permission bits in + // the registry, so no extra per-route permission is needed beyond + // membership. Message/approval runs stream over SSE. + ai := jwtOnly.Group("/ai") + ai.Use(m.RequireOrganization()) + { + ai.POST("/sessions", h.CreateAgentSession) + ai.GET("/sessions", h.ListAgentSessions) + ai.POST("/sessions/:id/messages", h.AgentMessage) + ai.POST("/sessions/:id/approve", h.AgentApprove) + } + subscriptions := jwtOnly.Group("/subscription") subscriptions.Use(m.RateLimitMiddleware(models.RateLimitWrite)) { diff --git a/internal/app/aiagent/service.go b/internal/app/aiagent/service.go new file mode 100644 index 00000000..a2ceaa55 --- /dev/null +++ b/internal/app/aiagent/service.go @@ -0,0 +1,457 @@ +// Package aiagent orchestrates the dashboard AI agent: it runs the M2 tool +// registry through the M1 provider loop, streams step events to the client over +// SSE, pauses write/send tools for human approval, charges one credit per loop +// iteration (bounded by a per-run budget), and persists a resumable transcript. +package aiagent + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strconv" + "strings" + "time" + + "github.com/google/uuid" + + "github.com/warmbly/warmbly/internal/app/aitools" + "github.com/warmbly/warmbly/internal/app/credits" + "github.com/warmbly/warmbly/internal/errx" + "github.com/warmbly/warmbly/internal/models" + "github.com/warmbly/warmbly/internal/pkg/generation" + "github.com/warmbly/warmbly/internal/repository" +) + +// DefaultRunBudget bounds one run's loop iterations (and therefore its credit +// cost). Surfaced to the client so the meter shows the ceiling. +const DefaultRunBudget = 20 + +// FeatureGate is the slice of the feature service used to route the model tier. +type FeatureGate interface { + IsPaidOrganization(ctx context.Context, orgID uuid.UUID) (bool, *errx.Error) +} + +// AuditLogger fires the ai_session audit so the create shows in the spine. +type AuditLogger interface { + LogAction(ctx context.Context, orgID, actorID uuid.UUID, action models.AuditAction, entityType models.AuditEntityType, entityID *uuid.UUID, ip, userAgent string, changes, metadata map[string]string) +} + +// StreamEvent is one SSE step emitted to the client. +type StreamEvent struct { + Type string `json:"type"` + Text string `json:"text,omitempty"` + Tool string `json:"tool,omitempty"` + Risk string `json:"risk,omitempty"` + ArgsSummary string `json:"args_summary,omitempty"` + ToolCallID string `json:"tool_call_id,omitempty"` + Result string `json:"result,omitempty"` + Iteration int `json:"iteration,omitempty"` + CreditsRemaining int `json:"credits_remaining,omitempty"` + Budget int `json:"budget,omitempty"` + Code string `json:"code,omitempty"` + Message string `json:"message,omitempty"` + // Draft artifact (create_campaign_draft / create_automation_draft): the + // client renders a card that deep-links into the real editor. + EntityType string `json:"entity_type,omitempty"` + EntityID string `json:"entity_id,omitempty"` + OpenURL string `json:"open_url,omitempty"` +} + +// toolResultEvent builds the tool_result SSE step, extracting a draft artifact +// (id + open url) so the client can render a deep-link card. +func toolResultEvent(tool, result string) StreamEvent { + ev := StreamEvent{Type: evToolDone, Tool: tool, Result: summarize(tool, result)} + var m map[string]any + if err := json.Unmarshal([]byte(result), &m); err == nil { + if id, ok := m["campaign_id"].(string); ok { + ev.EntityType, ev.EntityID = "campaign", id + } else if id, ok := m["automation_id"].(string); ok { + ev.EntityType, ev.EntityID = "automation", id + } + if url, ok := m["open_url"].(string); ok { + ev.OpenURL = url + } + } + return ev +} + +const ( + evText = "text" + evTool = "tool_start" + evToolDone = "tool_result" + evApproval = "approval_required" + evError = "error" + evDone = "done" +) + +// Service is the dashboard-agent application API. +type Service interface { + CreateSession(ctx context.Context, orgID, userID uuid.UUID, page, resource string) (*models.AgentSession, *errx.Error) + ListSessions(ctx context.Context, orgID, userID uuid.UUID, limit int, beforeCreatedAt time.Time, beforeID uuid.UUID) ([]models.AgentSession, error) + GetSession(ctx context.Context, orgID, userID, sessionID uuid.UUID) (*models.AgentSession, error) + + // RunMessage streams a new user message's run. inv carries the caller's + // identity + org permission bits. emit is called for each SSE step. + RunMessage(ctx context.Context, inv aitools.Invocation, sessionID uuid.UUID, messageID, text, page, resource string, emit func(StreamEvent)) *errx.Error + + // Resume continues a paused run after the user's decision + // (approve | deny | always_allow). + Resume(ctx context.Context, inv aitools.Invocation, sessionID uuid.UUID, decision string, emit func(StreamEvent)) *errx.Error +} + +type service struct { + repo repository.AgentRepository + registry *aitools.Registry + provider generation.Provider + credits credits.CreditService + feature FeatureGate + audit AuditLogger +} + +func NewService(repo repository.AgentRepository, registry *aitools.Registry, provider generation.Provider, creditSvc credits.CreditService, feature FeatureGate, audit AuditLogger) Service { + return &service{repo: repo, registry: registry, provider: provider, credits: creditSvc, feature: feature, audit: audit} +} + +func (s *service) CreateSession(ctx context.Context, orgID, userID uuid.UUID, page, resource string) (*models.AgentSession, *errx.Error) { + sess, err := s.repo.CreateSession(ctx, orgID, userID, "", models.AgentSessionContext{Page: page, Resource: resource}) + if err != nil { + return nil, errx.New(errx.Internal, "failed to create session") + } + if s.audit != nil { + s.audit.LogAction(ctx, orgID, userID, models.AuditActionCreate, models.AuditEntityAISession, &sess.ID, "", "", nil, nil) + } + return sess, nil +} + +func (s *service) ListSessions(ctx context.Context, orgID, userID uuid.UUID, limit int, beforeCreatedAt time.Time, beforeID uuid.UUID) ([]models.AgentSession, error) { + return s.repo.ListSessions(ctx, orgID, userID, limit, beforeCreatedAt, beforeID) +} + +func (s *service) GetSession(ctx context.Context, orgID, userID, sessionID uuid.UUID) (*models.AgentSession, error) { + return s.repo.GetSession(ctx, orgID, userID, sessionID) +} + +// errOutOfCredits / errCapExceeded are recorded by the credit PreIteration hook +// so the run can stop cleanly and the reason surfaced to the client. +var ( + errOutOfCredits = errors.New("insufficient credits") + errCapExceeded = errors.New("usage cap exceeded") + errStopped = errors.New("stopped") +) + +func (s *service) RunMessage(ctx context.Context, inv aitools.Invocation, sessionID uuid.UUID, messageID, text, page, resource string, emit func(StreamEvent)) *errx.Error { + if s.provider == nil { + return errx.New(errx.ServiceUnavailable, "the AI assistant is not configured") + } + sess, err := s.repo.GetSession(ctx, inv.OrgID, inv.UserID, sessionID) + if err != nil || sess == nil { + return errx.New(errx.NotFound, "session not found") + } + text = strings.TrimSpace(text) + if text == "" { + return errx.New(errx.BadRequest, "message is required") + } + + // Load prior transcript, append the user message, persist it. + genMsgs, xerr := s.loadTranscript(ctx, inv.OrgID, inv.UserID, sessionID) + if xerr != nil { + return xerr + } + userMsg := generation.AgentMessage{Role: "user", Content: text} + if perr := s.persist(ctx, inv.OrgID, inv.UserID, sessionID, []generation.AgentMessage{userMsg}, 0); perr != nil { + return errx.New(errx.Internal, "failed to save message") + } + genMsgs = append(genMsgs, userMsg) + + // Update session context (page/resource) and set a title from the first + // message. + sess.Context.Page = page + sess.Context.Resource = resource + sess.Context.Pending = nil + _ = s.repo.UpdateSessionContext(ctx, inv.OrgID, inv.UserID, sessionID, sess.Context) + _ = s.repo.UpdateSessionTitle(ctx, inv.OrgID, inv.UserID, sessionID, deriveTitle(text)) + + return s.runLoop(ctx, inv, sess, genMsgs, len(genMsgs), messageID, emit) +} + +func (s *service) Resume(ctx context.Context, inv aitools.Invocation, sessionID uuid.UUID, decision string, emit func(StreamEvent)) *errx.Error { + sess, err := s.repo.GetSession(ctx, inv.OrgID, inv.UserID, sessionID) + if err != nil || sess == nil { + return errx.New(errx.NotFound, "session not found") + } + pending := sess.Context.Pending + if pending == nil { + return errx.New(errx.BadRequest, "no tool awaiting approval") + } + + genMsgs, xerr := s.loadTranscript(ctx, inv.OrgID, inv.UserID, sessionID) + if xerr != nil { + return xerr + } + baseline := len(genMsgs) + + // Always-allow persists an org policy for write tools (never send). + if decision == "always_allow" && pending.Risk == string(generation.RiskWrite) { + _ = s.repo.SetToolPolicy(ctx, inv.OrgID, pending.ToolName, "always_allow", inv.UserID) + } + + assistant := generation.AgentMessage{Role: "assistant", ToolCalls: []generation.ToolCall{{ + ID: pending.ToolCallID, Name: pending.ToolName, Args: pending.Args, + }}} + + var toolResult string + if decision == "deny" { + toolResult = `{"status":"denied","note":"The user declined to run this action."}` + } else { + emit(StreamEvent{Type: evTool, Tool: pending.ToolName, Risk: pending.Risk, ArgsSummary: pending.ArgsSummary, ToolCallID: pending.ToolCallID}) + out, cerr := s.registry.Call(ctx, inv, pending.ToolName, pending.Args) + if cerr != nil { + b, _ := json.Marshal(map[string]string{"error": cerr.Error()}) + out = string(b) + } + toolResult = out + emit(toolResultEvent(pending.ToolName, out)) + } + + genMsgs = append(genMsgs, + assistant, + generation.AgentMessage{Role: "tool", ToolCallID: pending.ToolCallID, Content: toolResult}, + ) + + // Clear the pending marker before continuing. + sess.Context.Pending = nil + _ = s.repo.UpdateSessionContext(ctx, inv.OrgID, inv.UserID, sessionID, sess.Context) + + // Give the resumed segment its own credit-idempotency namespace. Reusing the + // original message id would collide with the pre-pause iterations (which + // already consumed messageID:1..N), so the resumed loop's iterations 1..N + // would replay for free. Pending is cleared above, so a resume is single- + // shot and a fresh id is safe. + return s.runLoop(ctx, inv, sess, genMsgs, baseline, "resume:"+uuid.NewString(), emit) +} + +// runLoop drives one RunAgent segment: credit-charged per iteration, approval- +// gated, streamed, and persisted. baseline is the count of genMsgs already +// persisted (new tail beyond it is written on completion). +func (s *service) runLoop(ctx context.Context, inv aitools.Invocation, sess *models.AgentSession, genMsgs []generation.AgentMessage, baseline int, messageID string, emit func(StreamEvent)) *errx.Error { + paid, _ := s.feature.IsPaidOrganization(ctx, inv.OrgID) + model := s.provider.ModelForTier(paid) + sess.Context.Model = model + + policies, _ := s.repo.GetToolPolicies(ctx, inv.OrgID) + + var ( + stopReason string + lastRemaining int + ) + + req := generation.AgentRequest{ + System: s.systemPrompt(sess), + Messages: genMsgs, + Tools: s.registry.ToolDefs(inv), + Model: model, + MaxIterations: DefaultRunBudget, + OnEvent: func(ev generation.AgentEvent) { + switch ev.Type { + case generation.EventText: + emit(StreamEvent{Type: evText, Text: ev.Text}) + case generation.EventToolStart: + emit(StreamEvent{Type: evTool, Tool: ev.ToolName, ArgsSummary: summarizeArgs(ev.ToolArgs)}) + case generation.EventToolResult: + emit(toolResultEvent(ev.ToolName, ev.ToolResult)) + } + }, + PreIteration: func(ctx context.Context, iter int) error { + key := messageID + ":" + strconv.Itoa(iter) + remaining, cerr := s.credits.Consume(ctx, inv.OrgID, credits.CostAgentIteration, "agent_iteration", model, 0, key) + if cerr != nil { + switch { + case errors.Is(cerr, credits.ErrInsufficientCredits): + stopReason = "out_of_credits" + return errOutOfCredits + case errors.Is(cerr, credits.ErrCapExceeded): + stopReason = "usage_cap" + return errCapExceeded + default: + stopReason = "error" + return errStopped + } + } + lastRemaining = remaining + emit(StreamEvent{Type: "iteration", Iteration: iter, CreditsRemaining: remaining, Budget: DefaultRunBudget}) + return nil + }, + Approve: func(ctx context.Context, tool generation.ToolDef, call generation.ToolCall) error { + // Send-class is always per-action (never auto-allowed). Write-class + // auto-runs only when an org policy says always_allow. + if tool.Risk == generation.RiskWrite && policies[tool.Name] == "always_allow" { + return nil + } + sess.Context.Pending = &models.PendingAgentTool{ + MessageID: messageID, + ToolCallID: call.ID, + ToolName: call.Name, + Risk: string(tool.Risk), + Args: call.Args, + ArgsSummary: summarizeArgs(call.Args), + } + emit(StreamEvent{ + Type: evApproval, Tool: call.Name, Risk: string(tool.Risk), + ArgsSummary: summarizeArgs(call.Args), ToolCallID: call.ID, + }) + return generation.ErrApprovalRequired + }, + } + + result, rerr := s.provider.RunAgent(ctx, req) + if rerr != nil { + // The credit for the failed iteration was charged before the model call + // (PreIteration); refund it so the user is not billed for output they + // never received. + if bal, gerr := s.credits.Grant(ctx, inv.OrgID, credits.CostAgentIteration, "agent_iteration_refund"); gerr == nil { + lastRemaining = bal + } + emit(StreamEvent{Type: evError, Code: "provider_error", Message: "The assistant hit an error. Please try again.", CreditsRemaining: lastRemaining}) + return nil + } + + // Persist the new transcript tail. + if len(result.Messages) > baseline { + _ = s.persist(ctx, sess.OrgID, sess.UserID, sess.ID, result.Messages[baseline:], result.TokensUsed) + } + + switch result.StopReason { + case "approval_required": + _ = s.repo.UpdateSessionContext(ctx, sess.OrgID, sess.UserID, sess.ID, sess.Context) + emit(StreamEvent{Type: evDone, CreditsRemaining: lastRemaining, Message: "awaiting_approval"}) + case "stopped": + s.emitStop(emit, stopReason, lastRemaining) + default: // "stop" or "max_iterations" + sess.Context.Pending = nil + _ = s.repo.UpdateSessionContext(ctx, sess.OrgID, sess.UserID, sess.ID, sess.Context) + emit(StreamEvent{Type: evDone, CreditsRemaining: lastRemaining}) + } + return nil +} + +func (s *service) emitStop(emit func(StreamEvent), reason string, remaining int) { + switch reason { + case "out_of_credits": + emit(StreamEvent{Type: evError, Code: "insufficient_credits", Message: "You're out of AI credits. Add more to keep using the assistant.", CreditsRemaining: remaining}) + case "usage_cap": + emit(StreamEvent{Type: evError, Code: "usage_cap_exceeded", Message: "AI usage limit reached, please try again later.", CreditsRemaining: remaining}) + default: + emit(StreamEvent{Type: evError, Code: "stopped", Message: "The run was stopped.", CreditsRemaining: remaining}) + } +} + +// --- helpers --- + +func (s *service) loadTranscript(ctx context.Context, orgID, userID, sessionID uuid.UUID) ([]generation.AgentMessage, *errx.Error) { + rows, err := s.repo.LoadTranscript(ctx, orgID, userID, sessionID) + if err != nil { + return nil, errx.New(errx.Internal, "failed to load conversation") + } + out := make([]generation.AgentMessage, 0, len(rows)) + for _, r := range rows { + var m generation.AgentMessage + if uerr := json.Unmarshal(r.Content, &m); uerr != nil { + continue + } + out = append(out, m) + } + return out, nil +} + +func (s *service) persist(ctx context.Context, orgID, userID, sessionID uuid.UUID, msgs []generation.AgentMessage, tokens int) error { + rows := make([]models.AgentMessageRow, 0, len(msgs)) + for i, m := range msgs { + b, err := json.Marshal(m) + if err != nil { + return err + } + t := 0 + if i == len(msgs)-1 { + t = tokens // attribute the run's tokens to its last message + } + rows = append(rows, models.AgentMessageRow{Role: m.Role, Content: b, Tokens: t}) + } + return s.repo.AppendMessages(ctx, orgID, userID, sessionID, rows) +} + +func (s *service) systemPrompt(sess *models.AgentSession) string { + var b strings.Builder + b.WriteString(`You are Warmbly's in-product AI assistant. You help the user manage their cold email outreach: contacts, campaigns, the unified inbox, CRM, and automations. Use the available tools to look things up and take actions. Be concise and specific. + +Rules: +- Read tools run automatically. Write actions (creating or changing data) require the user's approval, which the product handles for you; just call the tool and it will be gated. +- Never claim you sent an email. You can draft replies, but the user always sends. +- When you create a draft campaign or automation, tell the user it is a draft and give them the link to open it. +- If a tool returns an error, explain it plainly and suggest a next step.`) + if sess.Context.Page != "" || sess.Context.Resource != "" { + fmt.Fprintf(&b, "\n\nThe user is currently on page %q", sess.Context.Page) + if sess.Context.Resource != "" { + fmt.Fprintf(&b, " looking at %q", sess.Context.Resource) + } + b.WriteString(". Use this as context for what they mean by \"this\" or \"here\".") + } + return b.String() +} + +// deriveTitle makes a short session title from the first user message. +func deriveTitle(text string) string { + text = strings.TrimSpace(strings.ReplaceAll(text, "\n", " ")) + if len(text) > 60 { + return truncateRunesTitle(text, 60) + } + return text +} + +func truncateRunesTitle(s string, n int) string { + r := []rune(s) + if len(r) <= n { + return s + } + return strings.TrimSpace(string(r[:n])) + "…" +} + +// summarizeArgs renders a short one-line summary of tool arguments for the +// approval card / step row. +func summarizeArgs(args json.RawMessage) string { + if len(args) == 0 { + return "" + } + var m map[string]any + if err := json.Unmarshal(args, &m); err != nil { + return "" + } + parts := make([]string, 0, len(m)) + for k, v := range m { + parts = append(parts, fmt.Sprintf("%s=%v", k, v)) + if len(parts) >= 4 { + break + } + } + return truncateRunesTitle(strings.Join(parts, ", "), 160) +} + +// summarize renders a short human line for a tool result step row. +func summarize(tool, result string) string { + var m map[string]any + if err := json.Unmarshal([]byte(result), &m); err == nil { + if c, ok := m["count"]; ok { + return fmt.Sprintf("%v result(s)", c) + } + if e, ok := m["error"]; ok { + return fmt.Sprintf("error: %v", e) + } + if id, ok := m["campaign_id"]; ok { + return fmt.Sprintf("created campaign %v", id) + } + if id, ok := m["automation_id"]; ok { + return fmt.Sprintf("created automation %v", id) + } + } + return truncateRunesTitle(result, 120) +} diff --git a/internal/infrastructure/db/migrations/000058_ai_agent.down.sql b/internal/infrastructure/db/migrations/000058_ai_agent.down.sql new file mode 100644 index 00000000..fa78a9cf --- /dev/null +++ b/internal/infrastructure/db/migrations/000058_ai_agent.down.sql @@ -0,0 +1,3 @@ +DROP TABLE IF EXISTS ai_tool_policies; +DROP TABLE IF EXISTS agent_messages; +DROP TABLE IF EXISTS agent_sessions; diff --git a/internal/infrastructure/db/migrations/000058_ai_agent.up.sql b/internal/infrastructure/db/migrations/000058_ai_agent.up.sql new file mode 100644 index 00000000..aa516ee1 --- /dev/null +++ b/internal/infrastructure/db/migrations/000058_ai_agent.up.sql @@ -0,0 +1,50 @@ +-- Dashboard AI agent: chat sessions, their message transcript, and per-org +-- tool approval policies. +-- +-- agent_sessions is one conversation. context is a read-then-execute jsonb blob +-- (the client's {page, resource} awareness plus any pending tool call awaiting +-- approval); it is validated at the app boundary by a Go struct, not filtered +-- in SQL, so jsonb is the right representation. +-- +-- agent_messages is the append-only transcript. content is jsonb holding the +-- provider-agnostic message (role, text, tool_calls, tool results) so a run can +-- be resumed after an approval pause. + +CREATE TABLE IF NOT EXISTS agent_sessions ( + id uuid PRIMARY KEY DEFAULT gen_random_uuid(), + org_id uuid NOT NULL REFERENCES organizations (id) ON DELETE CASCADE, + user_id uuid NOT NULL REFERENCES users (id) ON DELETE CASCADE, + title text NOT NULL DEFAULT '', + context jsonb NOT NULL DEFAULT '{}'::jsonb, + created_at timestamptz NOT NULL DEFAULT now(), + updated_at timestamptz NOT NULL DEFAULT now() +); + +-- Sessions are per-user; list newest-first for the owning user within an org. +CREATE INDEX IF NOT EXISTS idx_agent_sessions_user + ON agent_sessions (org_id, user_id, created_at DESC); + +CREATE TABLE IF NOT EXISTS agent_messages ( + id uuid PRIMARY KEY DEFAULT gen_random_uuid(), + session_id uuid NOT NULL REFERENCES agent_sessions (id) ON DELETE CASCADE, + role text NOT NULL CHECK (role IN ('user', 'assistant', 'tool')), + content jsonb NOT NULL DEFAULT '{}'::jsonb, + tokens integer NOT NULL DEFAULT 0, + created_at timestamptz NOT NULL DEFAULT now() +); + +CREATE INDEX IF NOT EXISTS idx_agent_messages_session + ON agent_messages (session_id, created_at ASC, id ASC); + +-- Per-org tool approval policy. A row means "this org has decided how to handle +-- this tool by default". decision is 'always_allow' (auto-run write tools) — the +-- only persistable policy; send-class tools are never auto-allowed and so never +-- get a row. created_by records who set it. +CREATE TABLE IF NOT EXISTS ai_tool_policies ( + org_id uuid NOT NULL REFERENCES organizations (id) ON DELETE CASCADE, + tool_name text NOT NULL, + decision text NOT NULL DEFAULT 'always_allow' CHECK (decision IN ('always_allow')), + created_by uuid REFERENCES users (id) ON DELETE SET NULL, + created_at timestamptz NOT NULL DEFAULT now(), + PRIMARY KEY (org_id, tool_name) +); diff --git a/internal/models/agent.go b/internal/models/agent.go new file mode 100644 index 00000000..0fbdb11e --- /dev/null +++ b/internal/models/agent.go @@ -0,0 +1,68 @@ +package models + +import ( + "encoding/json" + "time" + + "github.com/google/uuid" +) + +// AgentSession is one dashboard-agent conversation. Sessions are per-user +// (private to the member who started them); the org scope is for tenancy only. +type AgentSession struct { + ID uuid.UUID `json:"id"` + OrgID uuid.UUID `json:"org_id"` + UserID uuid.UUID `json:"user_id"` + Title string `json:"title"` + Context AgentSessionContext `json:"context"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` +} + +// AgentSessionContext is the read-then-execute jsonb blob on a session: the +// client's page/resource awareness, the model chosen for the run, and any tool +// call paused awaiting approval. Validated at the app boundary (this struct), +// never filtered in SQL. +type AgentSessionContext struct { + // Page / Resource mirror the presence shape the client already pushes + // ({page, resource}); injected into the system prompt as context. + Page string `json:"page,omitempty"` + Resource string `json:"resource,omitempty"` + // Model is the provider model id resolved for this session's tier. + Model string `json:"model,omitempty"` + // Pending is the tool call awaiting the user's approve/deny when a run is + // paused; nil when the session is idle or running. + Pending *PendingAgentTool `json:"pending,omitempty"` +} + +// PendingAgentTool is a write/send tool call paused for human approval. +type PendingAgentTool struct { + MessageID string `json:"message_id"` + ToolCallID string `json:"tool_call_id"` + ToolName string `json:"tool_name"` + Risk string `json:"risk"` + Args json.RawMessage `json:"args"` + ArgsSummary string `json:"args_summary,omitempty"` +} + +// AgentMessageRow is one persisted transcript turn. Content is the serialized +// provider-agnostic message (role, text, tool_calls, tool result) so a run +// resumes losslessly after an approval pause. +type AgentMessageRow struct { + ID uuid.UUID `json:"id"` + SessionID uuid.UUID `json:"session_id"` + Role string `json:"role"` + Content json.RawMessage `json:"content"` + Tokens int `json:"tokens"` + CreatedAt time.Time `json:"created_at"` +} + +// AIToolPolicy is a per-org "always allow this tool" decision. Only write-class +// tools can have a policy; send-class tools are never auto-allowed. +type AIToolPolicy struct { + OrgID uuid.UUID `json:"org_id"` + ToolName string `json:"tool_name"` + Decision string `json:"decision"` + CreatedBy *uuid.UUID `json:"created_by,omitempty"` + CreatedAt time.Time `json:"created_at"` +} diff --git a/internal/models/api_permission.go b/internal/models/api_permission.go index 2e06320f..c17c0b60 100644 --- a/internal/models/api_permission.go +++ b/internal/models/api_permission.go @@ -47,6 +47,9 @@ const ( // Organization operations APIPermIntegrations // Connect and manage third-party integrations APIPermWarmupRouting // Manage warmup routing rules + + // AI + APIPermAIAgent // Run the AI assistant / agent (dashboard agent + MCP) ) // AllAPIPermissionsMask is the OR of every defined permission bit. @@ -63,7 +66,8 @@ const AllAPIPermissionsMask uint64 = APIPermReadEmails | APIPermReadCampaigns | APIPermReadTemplates | APIPermWriteTemplates | APIPermReadCRM | APIPermWriteCRM | APIPermReadAuditLogs | - APIPermIntegrations | APIPermWarmupRouting + APIPermIntegrations | APIPermWarmupRouting | + APIPermAIAgent // Preset permission sets surfaced via GET /api-keys/permissions so a // caller can grant a sane default without picking bits by hand. @@ -79,7 +83,8 @@ var ( APIPermSendCampaigns | APIPermWriteTemplates | APIPermWriteCRM | APIPermRealtimeSubscribe | APIPermWebhooks | APIPermAPIKeys | - APIPermIntegrations | APIPermWarmupRouting + APIPermIntegrations | APIPermWarmupRouting | + APIPermAIAgent ) type APIPermission struct { @@ -112,6 +117,7 @@ var AllAPIPermissions = []APIPermission{ {"API_KEYS", APIPermAPIKeys, "Create and manage API keys", "special"}, {"INTEGRATIONS", APIPermIntegrations, "Connect and manage third-party integrations", "special"}, {"WARMUP_ROUTING", APIPermWarmupRouting, "Manage warmup routing rules", "special"}, + {"AI_AGENT", APIPermAIAgent, "Run the AI assistant and MCP tools", "special"}, } // HasAPIPermission reports whether the bitmask grants every bit in `required`. diff --git a/internal/models/audit.go b/internal/models/audit.go index 53a438e2..0f87d46a 100644 --- a/internal/models/audit.go +++ b/internal/models/audit.go @@ -102,6 +102,9 @@ const ( // billing/credits view, so the spine refreshes teammates on either. AuditEntityCreditPurchase AuditEntityType = "credit_purchase" AuditEntityCreditGrant AuditEntityType = "credit_grant" + + // AI assistant session (per-user dashboard agent conversation). + AuditEntityAISession AuditEntityType = "ai_session" ) // AuditActor is the minimal identity of the member who performed an action, diff --git a/internal/pkg/generation/anthropic_provider.go b/internal/pkg/generation/anthropic_provider.go index e863156a..55dc867a 100644 --- a/internal/pkg/generation/anthropic_provider.go +++ b/internal/pkg/generation/anthropic_provider.go @@ -203,6 +203,13 @@ func (p *anthropicProvider) RunAgent(ctx context.Context, req AgentRequest) (*Ag result := &AgentResult{Model: model} for iter := 0; iter < maxIter; iter++ { + if req.PreIteration != nil { + if err := req.PreIteration(ctx, iter+1); err != nil { + result.Messages = messages + result.StopReason = "stopped" + return result, nil + } + } result.Iterations++ if req.OnEvent != nil { req.OnEvent(AgentEvent{Type: EventIteration, Iteration: result.Iterations}) diff --git a/internal/pkg/generation/openai_provider.go b/internal/pkg/generation/openai_provider.go index c668a2f4..381bcb74 100644 --- a/internal/pkg/generation/openai_provider.go +++ b/internal/pkg/generation/openai_provider.go @@ -234,6 +234,13 @@ func (p *openAIProvider) RunAgent(ctx context.Context, req AgentRequest) (*Agent result := &AgentResult{Model: model} for iter := 0; iter < maxIter; iter++ { + if req.PreIteration != nil { + if err := req.PreIteration(ctx, iter+1); err != nil { + result.Messages = messages + result.StopReason = "stopped" + return result, nil + } + } result.Iterations++ if req.OnEvent != nil { req.OnEvent(AgentEvent{Type: EventIteration, Iteration: result.Iterations}) diff --git a/internal/pkg/generation/provider.go b/internal/pkg/generation/provider.go index 9eed8409..280e5b1f 100644 --- a/internal/pkg/generation/provider.go +++ b/internal/pkg/generation/provider.go @@ -105,6 +105,12 @@ type AgentRequest struct { // holds the tool call awaiting approval, so the caller can persist state // and resume. Returning any other error aborts the run. Approve func(ctx context.Context, tool ToolDef, call ToolCall) error + // PreIteration, if set, is called at the start of each loop iteration + // (before the model call). Returning an error stops the loop cleanly with + // StopReason "stopped" and returns the transcript so far. Used to charge + // per-iteration credits and enforce a per-run budget; the caller records + // its own reason (e.g. out-of-credits) before returning. + PreIteration func(ctx context.Context, iteration int) error } // PendingToolCall is the tool awaiting approval when a run stops with diff --git a/internal/repository/pg_agent.go b/internal/repository/pg_agent.go new file mode 100644 index 00000000..c3fc255d --- /dev/null +++ b/internal/repository/pg_agent.go @@ -0,0 +1,217 @@ +package repository + +import ( + "context" + "encoding/json" + "errors" + "time" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5" + "github.com/warmbly/warmbly/internal/infrastructure/db" + "github.com/warmbly/warmbly/internal/models" +) + +// AgentRepository persists dashboard-agent sessions, their transcript, and the +// per-org tool approval policies. Sessions are per-user; every read is scoped by +// (org_id, user_id) so one member can never see another's conversations. +type AgentRepository interface { + CreateSession(ctx context.Context, orgID, userID uuid.UUID, title string, sctx models.AgentSessionContext) (*models.AgentSession, error) + GetSession(ctx context.Context, orgID, userID, sessionID uuid.UUID) (*models.AgentSession, error) + ListSessions(ctx context.Context, orgID, userID uuid.UUID, limit int, beforeCreatedAt time.Time, beforeID uuid.UUID) ([]models.AgentSession, error) + // The transcript/context mutators are scoped by (org_id, user_id) at the + // SQL layer too, not only via the caller's prior GetSession, so a mis-wired + // future caller can never touch another member's session by raw id. + UpdateSessionContext(ctx context.Context, orgID, userID, sessionID uuid.UUID, sctx models.AgentSessionContext) error + UpdateSessionTitle(ctx context.Context, orgID, userID, sessionID uuid.UUID, title string) error + + AppendMessages(ctx context.Context, orgID, userID, sessionID uuid.UUID, msgs []models.AgentMessageRow) error + LoadTranscript(ctx context.Context, orgID, userID, sessionID uuid.UUID) ([]models.AgentMessageRow, error) + + GetToolPolicies(ctx context.Context, orgID uuid.UUID) (map[string]string, error) + SetToolPolicy(ctx context.Context, orgID uuid.UUID, toolName, decision string, createdBy uuid.UUID) error +} + +type agentRepository struct { + DB *db.DB +} + +func NewAgentRepository(database *db.DB) AgentRepository { + return &agentRepository{DB: database} +} + +const agentSessionCols = `id, org_id, user_id, title, context, created_at, updated_at` + +func scanSession(row pgx.Row, s *models.AgentSession) error { + var ctxRaw []byte + if err := row.Scan(&s.ID, &s.OrgID, &s.UserID, &s.Title, &ctxRaw, &s.CreatedAt, &s.UpdatedAt); err != nil { + return err + } + if len(ctxRaw) > 0 { + if err := json.Unmarshal(ctxRaw, &s.Context); err != nil { + return err + } + } + return nil +} + +func (r *agentRepository) CreateSession(ctx context.Context, orgID, userID uuid.UUID, title string, sctx models.AgentSessionContext) (*models.AgentSession, error) { + ctxRaw, err := json.Marshal(sctx) + if err != nil { + return nil, err + } + s := &models.AgentSession{} + err = scanSession(r.DB.QueryRow(ctx, ` + INSERT INTO agent_sessions (org_id, user_id, title, context) + VALUES ($1, $2, $3, $4) + RETURNING `+agentSessionCols, orgID, userID, title, ctxRaw), s) + if err != nil { + return nil, err + } + return s, nil +} + +func (r *agentRepository) GetSession(ctx context.Context, orgID, userID, sessionID uuid.UUID) (*models.AgentSession, error) { + s := &models.AgentSession{} + err := scanSession(r.DB.QueryRow(ctx, + `SELECT `+agentSessionCols+` FROM agent_sessions WHERE id = $1 AND org_id = $2 AND user_id = $3`, + sessionID, orgID, userID), s) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return nil, nil + } + return nil, err + } + return s, nil +} + +func (r *agentRepository) ListSessions(ctx context.Context, orgID, userID uuid.UUID, limit int, beforeCreatedAt time.Time, beforeID uuid.UUID) ([]models.AgentSession, error) { + if limit <= 0 || limit > 100 { + limit = 25 + } + query := ` + SELECT ` + agentSessionCols + ` + FROM agent_sessions + WHERE org_id = $1 AND user_id = $2 + ORDER BY created_at DESC, id DESC + LIMIT $3` + args := []any{orgID, userID, limit} + if !beforeCreatedAt.IsZero() { + query = ` + SELECT ` + agentSessionCols + ` + FROM agent_sessions + WHERE org_id = $1 AND user_id = $2 AND (created_at, id) < ($4, $5) + ORDER BY created_at DESC, id DESC + LIMIT $3` + args = append(args, beforeCreatedAt, beforeID) + } + rows, err := r.DB.Query(ctx, query, args...) + if err != nil { + return nil, err + } + defer rows.Close() + out := make([]models.AgentSession, 0) + for rows.Next() { + var s models.AgentSession + if err := scanSession(rows, &s); err != nil { + return nil, err + } + out = append(out, s) + } + return out, rows.Err() +} + +func (r *agentRepository) UpdateSessionContext(ctx context.Context, orgID, userID, sessionID uuid.UUID, sctx models.AgentSessionContext) error { + ctxRaw, err := json.Marshal(sctx) + if err != nil { + return err + } + _, err = r.DB.Exec(ctx, `UPDATE agent_sessions SET context = $2, updated_at = now() WHERE id = $1 AND org_id = $3 AND user_id = $4`, sessionID, ctxRaw, orgID, userID) + return err +} + +func (r *agentRepository) UpdateSessionTitle(ctx context.Context, orgID, userID, sessionID uuid.UUID, title string) error { + _, err := r.DB.Exec(ctx, `UPDATE agent_sessions SET title = $2, updated_at = now() WHERE id = $1 AND title = '' AND org_id = $3 AND user_id = $4`, sessionID, title, orgID, userID) + return err +} + +func (r *agentRepository) AppendMessages(ctx context.Context, orgID, userID, sessionID uuid.UUID, msgs []models.AgentMessageRow) error { + if len(msgs) == 0 { + return nil + } + tx, err := r.DB.Begin(ctx) + if err != nil { + return err + } + defer tx.Rollback(ctx) + // Verify the session belongs to this member before appending, so a raw + // session id from a mis-wired caller can't write into another org's log. + var ok bool + if err := tx.QueryRow(ctx, `SELECT EXISTS (SELECT 1 FROM agent_sessions WHERE id = $1 AND org_id = $2 AND user_id = $3)`, sessionID, orgID, userID).Scan(&ok); err != nil { + return err + } + if !ok { + return errors.New("session not found for this member") + } + for _, m := range msgs { + if _, err := tx.Exec(ctx, ` + INSERT INTO agent_messages (session_id, role, content, tokens) + VALUES ($1, $2, $3, $4)`, sessionID, m.Role, []byte(m.Content), m.Tokens); err != nil { + return err + } + } + if _, err := tx.Exec(ctx, `UPDATE agent_sessions SET updated_at = now() WHERE id = $1`, sessionID); err != nil { + return err + } + return tx.Commit(ctx) +} + +func (r *agentRepository) LoadTranscript(ctx context.Context, orgID, userID, sessionID uuid.UUID) ([]models.AgentMessageRow, error) { + rows, err := r.DB.Query(ctx, ` + SELECT id, session_id, role, content, tokens, created_at + FROM agent_messages + WHERE session_id = $1 + AND EXISTS (SELECT 1 FROM agent_sessions WHERE id = $1 AND org_id = $2 AND user_id = $3) + ORDER BY created_at ASC, id ASC`, sessionID, orgID, userID) + if err != nil { + return nil, err + } + defer rows.Close() + out := make([]models.AgentMessageRow, 0) + for rows.Next() { + var m models.AgentMessageRow + var content []byte + if err := rows.Scan(&m.ID, &m.SessionID, &m.Role, &content, &m.Tokens, &m.CreatedAt); err != nil { + return nil, err + } + m.Content = content + out = append(out, m) + } + return out, rows.Err() +} + +func (r *agentRepository) GetToolPolicies(ctx context.Context, orgID uuid.UUID) (map[string]string, error) { + rows, err := r.DB.Query(ctx, `SELECT tool_name, decision FROM ai_tool_policies WHERE org_id = $1`, orgID) + if err != nil { + return nil, err + } + defer rows.Close() + out := make(map[string]string) + for rows.Next() { + var name, decision string + if err := rows.Scan(&name, &decision); err != nil { + return nil, err + } + out[name] = decision + } + return out, rows.Err() +} + +func (r *agentRepository) SetToolPolicy(ctx context.Context, orgID uuid.UUID, toolName, decision string, createdBy uuid.UUID) error { + _, err := r.DB.Exec(ctx, ` + INSERT INTO ai_tool_policies (org_id, tool_name, decision, created_by) + VALUES ($1, $2, $3, $4) + ON CONFLICT (org_id, tool_name) DO UPDATE SET decision = EXCLUDED.decision`, + orgID, toolName, decision, createdBy) + return err +} diff --git a/web/src/components/app/agent/AgentPanel.tsx b/web/src/components/app/agent/AgentPanel.tsx new file mode 100644 index 00000000..ebbd656e --- /dev/null +++ b/web/src/components/app/agent/AgentPanel.tsx @@ -0,0 +1,508 @@ +// Right-side AI assistant panel. Persistent across routes, opened from the +// header sparkle button or Cmd+I. Streams a tool-using agent over SSE: text +// deltas, collapsible tool steps, inline approval cards for write/send actions, +// draft-artifact deep links, a stop button, and a credits meter in the footer. +// +// Follows the InboxDetails drawer visual language (fixed right drawer, slate/sky +// theme). Sends never happen from AI: draft artifacts open the real editors. + +import React from "react"; +import { AnimatePresence, motion } from "framer-motion"; +import { useLocation, useNavigate } from "react-router-dom"; +import { + SparklesIcon, + XIcon, + ArrowUpIcon, + SquareIcon, + PlusIcon, + CheckIcon, + Loader2Icon, + WrenchIcon, + ExternalLinkIcon, + AlertTriangleIcon, + ShieldQuestionIcon, +} from "lucide-react"; +import { useAppStore } from "@/stores"; +import createAgentSession from "@/lib/api/client/app/agent/createAgentSession"; +import streamAgentRun from "@/lib/api/client/app/agent/streamAgentRun"; +import type { AgentStreamEvent } from "@/lib/api/models/app/agent/Agent"; + +type ToolStep = { + id: string; + tool: string; + argsSummary?: string; + result?: string; + done: boolean; + entityType?: string; + entityId?: string; + openURL?: string; +}; + +type Pending = { + toolCallId: string; + tool: string; + risk: string; + argsSummary?: string; +}; + +type Block = + | { kind: "text"; text: string } + | { kind: "tool"; step: ToolStep } + | { kind: "error"; code?: string; message: string }; + +type Turn = { + id: string; + role: "user" | "assistant"; + blocks: Block[]; +}; + +let mid = 0; +const nextId = () => `m${++mid}`; + +export default function AgentPanel() { + const open = useAppStore((s) => s.aiAssistantOpen); + const setOpen = useAppStore((s) => s.setAIAssistantOpen); + const navigate = useNavigate(); + const location = useLocation(); + + const [sessionId, setSessionId] = React.useState(null); + const [turns, setTurns] = React.useState([]); + const [input, setInput] = React.useState(""); + const [running, setRunning] = React.useState(false); + const [pending, setPending] = React.useState(null); + const [credits, setCredits] = React.useState(null); + const [budget, setBudget] = React.useState(20); + const [iteration, setIteration] = React.useState(0); + const abortRef = React.useRef(null); + const scrollRef = React.useRef(null); + + React.useEffect(() => { + if (scrollRef.current) { + scrollRef.current.scrollTop = scrollRef.current.scrollHeight; + } + }, [turns, pending]); + + // Resource string mirrors the presence shape so the agent knows what the + // user is looking at ("this campaign", "here"). + const resource = React.useMemo(() => resourceFromPath(location.pathname), [location.pathname]); + + function resetChat() { + abortRef.current?.abort(); + setSessionId(null); + setTurns([]); + setPending(null); + setRunning(false); + } + + function appendAssistantEvent(ev: AgentStreamEvent) { + setTurns((prev) => { + const next = [...prev]; + let cur = next[next.length - 1]; + if (!cur || cur.role !== "assistant") { + cur = { id: nextId(), role: "assistant", blocks: [] }; + next.push(cur); + } else { + cur = { ...cur, blocks: [...cur.blocks] }; + next[next.length - 1] = cur; + } + applyEvent(cur, ev); + return next; + }); + } + + async function runStream(path: string, body: Record) { + setRunning(true); + setPending(null); + setIteration(0); + const ac = new AbortController(); + abortRef.current = ac; + await streamAgentRun( + path, + body, + (ev) => { + if (ev.type === "iteration") { + if (typeof ev.credits_remaining === "number") setCredits(ev.credits_remaining); + if (typeof ev.budget === "number") setBudget(ev.budget); + if (typeof ev.iteration === "number") setIteration(ev.iteration); + return; + } + if (ev.type === "approval_required") { + setPending({ + toolCallId: ev.tool_call_id || "", + tool: ev.tool || "", + risk: ev.risk || "write", + argsSummary: ev.args_summary, + }); + return; + } + if (ev.type === "done") { + if (typeof ev.credits_remaining === "number") setCredits(ev.credits_remaining); + return; + } + appendAssistantEvent(ev); + if (ev.type === "error" && typeof ev.credits_remaining === "number") { + setCredits(ev.credits_remaining); + } + }, + ac.signal, + ); + setRunning(false); + abortRef.current = null; + } + + async function send() { + const text = input.trim(); + if (!text || running) return; + setInput(""); + setTurns((prev) => [...prev, { id: nextId(), role: "user", blocks: [{ kind: "text", text }] }]); + + let sid = sessionId; + if (!sid) { + try { + const sess = await createAgentSession({ page: location.pathname, resource }); + sid = sess.id; + setSessionId(sid); + } catch { + appendAssistantEvent({ type: "error", message: "Could not start a session." }); + return; + } + } + await runStream(`/ai/sessions/${sid}/messages`, { + message_id: nextId() + ":" + Date.now(), + text, + page: location.pathname, + resource, + }); + } + + async function decide(decision: "approve" | "deny" | "always_allow") { + if (!sessionId || !pending) return; + setPending(null); + await runStream(`/ai/sessions/${sessionId}/approve`, { decision }); + } + + function stop() { + abortRef.current?.abort(); + setRunning(false); + } + + return ( + + {open && ( + <> + setOpen(false)} + className="fixed inset-0 z-40 bg-slate-900/20 md:hidden" + /> + + {/* Header */} +
+
+ +
+
+
Assistant
+
+ + +
+ + {/* Messages */} +
+ {turns.length === 0 && !running && ( + + )} + {turns.map((t) => ( + navigate(stripOrigin(u))} /> + ))} + {pending && ( + + )} + {running && !pending && ( +
+ + Working… + {iteration > 0 && ( + + step {iteration}/{budget} + + )} +
+ )} +
+ + {/* Composer */} +
+
+