Files
multica/server/internal/handler/runtime.go
Bohan Jiang f628e48775 refactor(server): error-returning ParseUUID to prevent silent data loss
* refactor(server): make ParseUUID error-returning to prevent silent data loss (MUL-1410)

util.ParseUUID previously swallowed errors and returned a zero pgtype.UUID
on invalid input. When this zero UUID reached a write query (DELETE/UPDATE),
the SQL matched zero rows and the handler returned 2xx success — producing
silent data corruption. #1661 (DeleteIssue with identifier-style ID) was the
visible symptom; PR #1680 patched that one site, this commit closes the
class of bug.

Changes:

- util.ParseUUID now returns (pgtype.UUID, error). Add util.MustParseUUID
  for trusted round-trips that should panic on invalid input.
- handler/handler.go: parseUUID wrapper now calls MustParseUUID — any
  unguarded user-input string reaching it surfaces as a recovered panic
  (chi middleware.Recoverer → 500) instead of silently corrupting data.
  Add parseUUIDOrBadRequest(w, s, fieldName) for handler entry points.
- Convert every Queries.Delete*/Update* call site reachable from raw user
  input (autopilot, comment, project, skill, skill_file, label, pin,
  attachment, feedback, issue assignee, daemon runtime, workspace) to
  validate UUIDs explicitly with parseUUIDOrBadRequest, returning 400 on
  invalid input. Where a resolved entity.ID is already in scope, write
  queries now use it directly instead of re-parsing the URL string.
- Update getWorkspaceMember + loadIssueForUser to handle invalid UUIDs
  gracefully (404/400 instead of panic).
- Update util/middleware/cmd-level callers (subscriber_listeners,
  notification_listeners, activity_listeners, scope_authorizer,
  middleware/workspace) to use the error-returning API.
- Add server/internal/util/pgx_test.go covering valid/invalid input and
  the MustParseUUID panic contract.
- Add TestDeleteIssueByIdentifier + TestDeleteIssueRejectsInvalidUUID
  regression tests in handler_test.go (the original #1661 bug + the
  invalid-input case).
- Document the handler UUID parsing convention in CLAUDE.md so the rule
  is enforceable in future PR review.

* fix(server): address GPT-Boy review of #1748

P1 fixes from PR #1748 review:

1. Migrate remaining request-boundary UUIDs to parseUUIDOrBadRequest so
   malformed input returns 400 instead of panic/500. Was missing on:
   - issue.go: workspace_id in CreateIssue/ChildIssueProgress/ListIssues/
     SearchIssues/BatchUpdateIssues/BatchDeleteIssues; project_id /
     parent_issue_id / lead_id / assignee_id / assignee_ids / creator_id
     filters; batch issue_ids and assignee/parent/project fields in
     BatchUpdateIssues (skip on bad input via util.ParseUUID, matching
     the existing per-row continue semantics).
   - project.go: project id + workspace_id in GetProject/UpdateProject/
     DeleteProject; lead_id in CreateProject/UpdateProject;
     workspace_id in ListProjects + SearchProjects.
   - handler.go: resolveActor now uses util.ParseUUID for X-Agent-ID /
     X-Task-ID headers; invalid UUID falls back to "member" (matches
     pre-existing semantics) instead of panicking.
   - issue.go: validateAssigneePair returns 400 on invalid workspace_id
     instead of panicking.

2. Fix issue:deleted WS event payloads to emit uuidToString(issue.ID)
   instead of the raw URL string. After an identifier-path delete
   ("MUL-7"), the previous payload would have leaked the identifier to
   subscribers, leaving stale entries in frontend caches that key by
   UUID. Updated DeleteIssue (issue.go:1341) and BatchDeleteIssues
   (issue.go:1641). The slog "issue deleted" log line also now records
   the resolved UUID so logs match the WS payload.

3. Extend TestDeleteIssueByIdentifier to subscribe to the bus and
   assert issue:deleted.payload.issue_id is the resolved UUID, not
   the identifier.

* fix(server): validate remaining reviewed UUID inputs

* fix(server): validate remaining handler UUID inputs

* fix(server): finish request boundary UUID audit

* fix(server): validate remaining request body UUIDs

* fix(server): validate runtime path UUIDs

* fix(server): validate remaining audit UUID inputs

---------

Co-authored-by: Eve <eve@multica.ai>
2026-04-28 14:50:28 +08:00

345 lines
11 KiB
Go

package handler
import (
"encoding/json"
"log/slog"
"net/http"
"strconv"
"time"
"github.com/go-chi/chi/v5"
"github.com/jackc/pgx/v5/pgtype"
"github.com/multica-ai/multica/server/pkg/agent"
db "github.com/multica-ai/multica/server/pkg/db/generated"
"github.com/multica-ai/multica/server/pkg/protocol"
)
type AgentRuntimeResponse struct {
ID string `json:"id"`
WorkspaceID string `json:"workspace_id"`
DaemonID *string `json:"daemon_id"`
Name string `json:"name"`
RuntimeMode string `json:"runtime_mode"`
Provider string `json:"provider"`
LaunchHeader string `json:"launch_header"`
Status string `json:"status"`
DeviceInfo string `json:"device_info"`
Metadata any `json:"metadata"`
OwnerID *string `json:"owner_id"`
LastSeenAt *string `json:"last_seen_at"`
CreatedAt string `json:"created_at"`
UpdatedAt string `json:"updated_at"`
}
func runtimeToResponse(rt db.AgentRuntime) AgentRuntimeResponse {
var metadata any
if rt.Metadata != nil {
json.Unmarshal(rt.Metadata, &metadata)
}
if metadata == nil {
metadata = map[string]any{}
}
return AgentRuntimeResponse{
ID: uuidToString(rt.ID),
WorkspaceID: uuidToString(rt.WorkspaceID),
DaemonID: textToPtr(rt.DaemonID),
Name: rt.Name,
RuntimeMode: rt.RuntimeMode,
Provider: rt.Provider,
LaunchHeader: agent.LaunchHeader(rt.Provider),
Status: rt.Status,
DeviceInfo: rt.DeviceInfo,
Metadata: metadata,
OwnerID: uuidToPtr(rt.OwnerID),
LastSeenAt: timestampToPtr(rt.LastSeenAt),
CreatedAt: timestampToString(rt.CreatedAt),
UpdatedAt: timestampToString(rt.UpdatedAt),
}
}
// ---------------------------------------------------------------------------
// Runtime Usage
// ---------------------------------------------------------------------------
type RuntimeUsageResponse struct {
RuntimeID string `json:"runtime_id"`
Date string `json:"date"`
Provider string `json:"provider"`
Model string `json:"model"`
InputTokens int64 `json:"input_tokens"`
OutputTokens int64 `json:"output_tokens"`
CacheReadTokens int64 `json:"cache_read_tokens"`
CacheWriteTokens int64 `json:"cache_write_tokens"`
}
// GetRuntimeUsage returns daily token usage for a runtime, aggregated from
// per-task usage records captured by the daemon. This is scoped to
// Daemon-executed tasks only (i.e. excludes users' local CLI usage of the
// same tool).
func (h *Handler) GetRuntimeUsage(w http.ResponseWriter, r *http.Request) {
runtimeID := chi.URLParam(r, "runtimeId")
runtimeUUID, ok := parseUUIDOrBadRequest(w, runtimeID, "runtime_id")
if !ok {
return
}
rt, err := h.Queries.GetAgentRuntime(r.Context(), runtimeUUID)
if err != nil {
writeError(w, http.StatusNotFound, "runtime not found")
return
}
if _, ok := h.requireWorkspaceMember(w, r, uuidToString(rt.WorkspaceID), "runtime not found"); !ok {
return
}
since := parseSinceParam(r, 90)
rows, err := h.Queries.ListRuntimeUsage(r.Context(), db.ListRuntimeUsageParams{
RuntimeID: rt.ID,
Since: since,
})
if err != nil {
writeError(w, http.StatusInternalServerError, "failed to list usage")
return
}
resp := make([]RuntimeUsageResponse, len(rows))
resolvedRuntimeID := uuidToString(rt.ID)
for i, row := range rows {
resp[i] = RuntimeUsageResponse{
RuntimeID: resolvedRuntimeID,
Date: row.Date.Time.Format("2006-01-02"),
Provider: row.Provider,
Model: row.Model,
InputTokens: row.InputTokens,
OutputTokens: row.OutputTokens,
CacheReadTokens: row.CacheReadTokens,
CacheWriteTokens: row.CacheWriteTokens,
}
}
writeJSON(w, http.StatusOK, resp)
}
// GetRuntimeTaskActivity returns hourly task activity distribution for a runtime.
func (h *Handler) GetRuntimeTaskActivity(w http.ResponseWriter, r *http.Request) {
runtimeID := chi.URLParam(r, "runtimeId")
runtimeUUID, ok := parseUUIDOrBadRequest(w, runtimeID, "runtime_id")
if !ok {
return
}
rt, err := h.Queries.GetAgentRuntime(r.Context(), runtimeUUID)
if err != nil {
writeError(w, http.StatusNotFound, "runtime not found")
return
}
if _, ok := h.requireWorkspaceMember(w, r, uuidToString(rt.WorkspaceID), "runtime not found"); !ok {
return
}
rows, err := h.Queries.GetRuntimeTaskHourlyActivity(r.Context(), rt.ID)
if err != nil {
writeError(w, http.StatusInternalServerError, "failed to get task activity")
return
}
type HourlyActivity struct {
Hour int `json:"hour"`
Count int `json:"count"`
}
resp := make([]HourlyActivity, len(rows))
for i, row := range rows {
resp[i] = HourlyActivity{Hour: int(row.Hour), Count: int(row.Count)}
}
writeJSON(w, http.StatusOK, resp)
}
// GetWorkspaceUsageByDay returns daily token usage aggregated by model for the workspace.
func (h *Handler) GetWorkspaceUsageByDay(w http.ResponseWriter, r *http.Request) {
workspaceID := h.resolveWorkspaceID(r)
since := parseSinceParam(r, 30)
rows, err := h.Queries.GetWorkspaceUsageByDay(r.Context(), db.GetWorkspaceUsageByDayParams{
WorkspaceID: parseUUID(workspaceID),
Since: since,
})
if err != nil {
writeError(w, http.StatusInternalServerError, "failed to get usage")
return
}
type DailyUsageRow struct {
Date string `json:"date"`
Model string `json:"model"`
TotalInputTokens int64 `json:"total_input_tokens"`
TotalOutputTokens int64 `json:"total_output_tokens"`
TotalCacheReadTokens int64 `json:"total_cache_read_tokens"`
TotalCacheWriteTokens int64 `json:"total_cache_write_tokens"`
TaskCount int32 `json:"task_count"`
}
resp := make([]DailyUsageRow, len(rows))
for i, row := range rows {
resp[i] = DailyUsageRow{
Date: row.Date.Time.Format("2006-01-02"),
Model: row.Model,
TotalInputTokens: row.TotalInputTokens,
TotalOutputTokens: row.TotalOutputTokens,
TotalCacheReadTokens: row.TotalCacheReadTokens,
TotalCacheWriteTokens: row.TotalCacheWriteTokens,
TaskCount: row.TaskCount,
}
}
writeJSON(w, http.StatusOK, resp)
}
// GetWorkspaceUsageSummary returns total token usage aggregated by model for the workspace.
func (h *Handler) GetWorkspaceUsageSummary(w http.ResponseWriter, r *http.Request) {
workspaceID := h.resolveWorkspaceID(r)
since := parseSinceParam(r, 30)
rows, err := h.Queries.GetWorkspaceUsageSummary(r.Context(), db.GetWorkspaceUsageSummaryParams{
WorkspaceID: parseUUID(workspaceID),
Since: since,
})
if err != nil {
writeError(w, http.StatusInternalServerError, "failed to get usage summary")
return
}
type UsageSummaryRow struct {
Model string `json:"model"`
TotalInputTokens int64 `json:"total_input_tokens"`
TotalOutputTokens int64 `json:"total_output_tokens"`
TotalCacheReadTokens int64 `json:"total_cache_read_tokens"`
TotalCacheWriteTokens int64 `json:"total_cache_write_tokens"`
TaskCount int32 `json:"task_count"`
}
resp := make([]UsageSummaryRow, len(rows))
for i, row := range rows {
resp[i] = UsageSummaryRow{
Model: row.Model,
TotalInputTokens: row.TotalInputTokens,
TotalOutputTokens: row.TotalOutputTokens,
TotalCacheReadTokens: row.TotalCacheReadTokens,
TotalCacheWriteTokens: row.TotalCacheWriteTokens,
TaskCount: row.TaskCount,
}
}
writeJSON(w, http.StatusOK, resp)
}
// parseSinceParam parses the "days" query parameter and returns a timestamptz.
func parseSinceParam(r *http.Request, defaultDays int) pgtype.Timestamptz {
days := defaultDays
if d := r.URL.Query().Get("days"); d != "" {
if parsed, err := strconv.Atoi(d); err == nil && parsed > 0 && parsed <= 365 {
days = parsed
}
}
t := time.Now().AddDate(0, 0, -days)
return pgtype.Timestamptz{Time: t, Valid: true}
}
func (h *Handler) ListAgentRuntimes(w http.ResponseWriter, r *http.Request) {
workspaceID := h.resolveWorkspaceID(r)
var runtimes []db.AgentRuntime
var err error
if ownerFilter := r.URL.Query().Get("owner"); ownerFilter == "me" {
userID, ok := requireUserID(w, r)
if !ok {
return
}
runtimes, err = h.Queries.ListAgentRuntimesByOwner(r.Context(), db.ListAgentRuntimesByOwnerParams{
WorkspaceID: parseUUID(workspaceID),
OwnerID: parseUUID(userID),
})
} else {
runtimes, err = h.Queries.ListAgentRuntimes(r.Context(), parseUUID(workspaceID))
}
if err != nil {
writeError(w, http.StatusInternalServerError, "failed to list runtimes")
return
}
resp := make([]AgentRuntimeResponse, len(runtimes))
for i, rt := range runtimes {
resp[i] = runtimeToResponse(rt)
}
writeJSON(w, http.StatusOK, resp)
}
// DeleteAgentRuntime deletes a runtime after permission and dependency checks.
func (h *Handler) DeleteAgentRuntime(w http.ResponseWriter, r *http.Request) {
runtimeID := chi.URLParam(r, "runtimeId")
runtimeUUID, ok := parseUUIDOrBadRequest(w, runtimeID, "runtime_id")
if !ok {
return
}
rt, err := h.Queries.GetAgentRuntime(r.Context(), runtimeUUID)
if err != nil {
writeError(w, http.StatusNotFound, "runtime not found")
return
}
wsID := uuidToString(rt.WorkspaceID)
member, ok := h.requireWorkspaceMember(w, r, wsID, "runtime not found")
if !ok {
return
}
// Permission: owner/admin can delete any runtime; members can only delete their own.
userID := uuidToString(member.UserID)
isAdmin := roleAllowed(member.Role, "owner", "admin")
isOwner := rt.OwnerID.Valid && uuidToString(rt.OwnerID) == userID
if !isAdmin && !isOwner {
writeError(w, http.StatusForbidden, "you can only delete your own runtimes")
return
}
// Check if any active (non-archived) agents are bound to this runtime.
activeCount, err := h.Queries.CountActiveAgentsByRuntime(r.Context(), rt.ID)
if err != nil {
writeError(w, http.StatusInternalServerError, "failed to check runtime dependencies")
return
}
if activeCount > 0 {
writeError(w, http.StatusConflict, "cannot delete runtime: it has active agents bound to it. Archive or reassign the agents first.")
return
}
// Remove archived agents so the FK constraint (ON DELETE RESTRICT) won't block deletion.
if err := h.Queries.DeleteArchivedAgentsByRuntime(r.Context(), rt.ID); err != nil {
writeError(w, http.StatusInternalServerError, "failed to clean up archived agents")
return
}
if err := h.Queries.DeleteAgentRuntime(r.Context(), rt.ID); err != nil {
writeError(w, http.StatusInternalServerError, "failed to delete runtime")
return
}
slog.Info("runtime deleted", "runtime_id", uuidToString(rt.ID), "deleted_by", userID)
// Notify frontend to refresh runtime list.
h.publish(protocol.EventDaemonRegister, wsID, "member", userID, map[string]any{
"action": "delete",
})
writeJSON(w, http.StatusOK, map[string]string{"status": "ok"})
}