mirror of
https://gh-proxy.org/https://github.com/klinsmaya/multica.git
synced 2026-11-08 17:36:45 +08:00
* MUL-7267: fix(autopilot): backfill legacy trigger creators from the autopilot creator Since MUL-6951 a schedule/webhook dispatch acts as the trigger's creator and fails closed without one. Migration 449 left created_by NULL for triggers that predate published_by (migration 189) and were never edited, so every run of those triggers has been recorded as skipped since the upgrade: webhook deliveries still read "dispatched" and schedules still advance, but nothing executes (#8284). Migration 467 fills those rows from the autopilot's creator, the principal automatic dispatch was admitted as before MUL-6951, when that creator is still a member of the workspace. Existing member creators are never rewritten, and dispatch still re-validates membership and invoke access on every run. Co-authored-by: multica-agent <[email protected]> * MUL-7267: docs(autopilot): state the trigger principal backfill as an accepted widening Review follow-up. The 467 comment claimed no trigger gains an authorization it did not pass before the upgrade; that holds for admission only. A pre-MUL-6951 automatic run carried no originator, while a filled trigger now runs, and delegates, as the autopilot's creator. Say so explicitly, as 449 does. created_by is also no longer "written once at creation" for every row: for a legacy trigger it is a principal inferred by backfill (449, then 467), not proof of who created it. Override the created_by_id column comment in 467 (the down migration restores both of 449's comments), regenerate models.go, and align the ResolveAutopilotTriggerPrincipal and dispatch attribution comments. No behavior change. Co-authored-by: multica-agent <[email protected]> * MUL-7267: docs(autopilot): name the trigger principal consistently at call sites Second review follow-up. The admission, dispatch and attribution call sites still called the automation principal "the trigger's creator" and spoke of a "recoverable" or "provable" creator. For a legacy trigger, created_by is a principal inferred by backfill, not proof of who created the trigger. Refer to it as the trigger's created_by principal and point at ResolveAutopilotTriggerPrincipal for the legacy semantics instead. No behavior change. Co-authored-by: multica-agent <[email protected]> --------- Co-authored-by: J <[email protected]> Co-authored-by: multica-agent <[email protected]>
2470 lines
95 KiB
Go
2470 lines
95 KiB
Go
package handler
|
||
|
||
import (
|
||
"bytes"
|
||
"context"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"io"
|
||
"log/slog"
|
||
"net/http"
|
||
"sort"
|
||
"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"
|
||
obsmetrics "github.com/multica-ai/multica/server/internal/metrics"
|
||
"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/protocol"
|
||
)
|
||
|
||
// computeNextRun delegates to the shared cron helper in the service package.
|
||
func computeNextRun(cronExpr, timezone string) (time.Time, error) {
|
||
return service.ComputeNextRun(cronExpr, timezone)
|
||
}
|
||
|
||
// ── Response types ──────────────────────────────────────────────────────────
|
||
|
||
type AutopilotResponse struct {
|
||
ID string `json:"id"`
|
||
WorkspaceID string `json:"workspace_id"`
|
||
Title string `json:"title"`
|
||
Description *string `json:"description"`
|
||
ProjectID *string `json:"project_id"`
|
||
// AssigneeType is "agent" or "squad". Path A from MUL-2429: when set
|
||
// to "squad", AssigneeID points at squad(id) rather than agent(id) and
|
||
// dispatch resolves to squad.leader_id at run time.
|
||
AssigneeType string `json:"assignee_type"`
|
||
AssigneeID string `json:"assignee_id"`
|
||
Status string `json:"status"`
|
||
PauseReason *string `json:"pause_reason"`
|
||
ExecutionMode string `json:"execution_mode"`
|
||
IssueTitleTemplate *string `json:"issue_title_template"`
|
||
CreatedByType string `json:"created_by_type"`
|
||
CreatedByID string `json:"created_by_id"`
|
||
LastRunAt *string `json:"last_run_at"`
|
||
CreatedAt string `json:"created_at"`
|
||
UpdatedAt string `json:"updated_at"`
|
||
|
||
// List-endpoint-only derived fields (absent on the detail/create/update
|
||
// responses and on older servers — clients must treat them as optional).
|
||
// Enabled triggers only; last_run_status is the most recent run's status.
|
||
TriggerKinds []string `json:"trigger_kinds,omitempty"`
|
||
NextRunAt *string `json:"next_run_at,omitempty"`
|
||
LastRunStatus *string `json:"last_run_status,omitempty"`
|
||
|
||
// Always non-nil (empty slice when no subscribers configured) so
|
||
// frontend optional-chain rules can treat the field as authoritative.
|
||
Subscribers []AutopilotSubscriberEntry `json:"subscribers"`
|
||
|
||
// CanWrite reports whether the requesting caller may perform write/execute
|
||
// operations on this autopilot — editing, deleting, triggering, and
|
||
// managing triggers/webhook secrets (creator, workspace owner/admin, or an
|
||
// explicit collaborator). Nil on responses built without a caller in
|
||
// context (older servers omit it; clients must treat absence as "unknown"
|
||
// and fall back to attempting the action). See MUL-3807.
|
||
CanWrite *bool `json:"can_write,omitempty"`
|
||
|
||
// CanManageAccess reports whether the caller may manage the collaborator
|
||
// (access) list — a narrower right held only by the creator and workspace
|
||
// owners/admins, NOT by granted collaborators (who can write but cannot
|
||
// re-grant). Nil when built without a caller in context. See MUL-3807.
|
||
CanManageAccess *bool `json:"can_manage_access,omitempty"`
|
||
}
|
||
|
||
type AutopilotQuotaUsageResponse struct {
|
||
Action string `json:"action"`
|
||
Used *int64 `json:"used"`
|
||
Reserved *int64 `json:"reserved"`
|
||
Total *int64 `json:"total"`
|
||
Limit *int64 `json:"limit"`
|
||
Reached *bool `json:"reached"`
|
||
PeriodStart *string `json:"period_start"`
|
||
PeriodEnd *string `json:"period_end"`
|
||
ResetAt *string `json:"reset_at"`
|
||
BlockedCounts map[string]int64 `json:"blocked_counts"`
|
||
}
|
||
|
||
// AutopilotCollaboratorEntry is a member explicitly granted write access to an
|
||
// autopilot, surfaced on the detail response and the collaborator endpoints.
|
||
type AutopilotCollaboratorEntry struct {
|
||
UserType string `json:"user_type"`
|
||
UserID string `json:"user_id"`
|
||
GrantedBy string `json:"granted_by"`
|
||
CreatedAt string `json:"created_at"`
|
||
}
|
||
|
||
func collaboratorToEntry(c db.AutopilotCollaborator) AutopilotCollaboratorEntry {
|
||
return AutopilotCollaboratorEntry{
|
||
UserType: c.UserType,
|
||
UserID: uuidToString(c.UserID),
|
||
GrantedBy: uuidToString(c.GrantedBy),
|
||
CreatedAt: timestampToString(c.CreatedAt),
|
||
}
|
||
}
|
||
|
||
// user_type is restricted to "member" at the DB layer; the field is kept on
|
||
// the wire so a future expansion to agents/squads is additive, not breaking.
|
||
type AutopilotSubscriberEntry struct {
|
||
UserType string `json:"user_type"`
|
||
UserID string `json:"user_id"`
|
||
CreatedAt string `json:"created_at"`
|
||
}
|
||
|
||
type AutopilotTriggerResponse struct {
|
||
ID string `json:"id"`
|
||
AutopilotID string `json:"autopilot_id"`
|
||
Kind string `json:"kind"`
|
||
Enabled bool `json:"enabled"`
|
||
CronExpression *string `json:"cron_expression"`
|
||
Timezone *string `json:"timezone"`
|
||
NextRunAt *string `json:"next_run_at"`
|
||
WebhookToken *string `json:"webhook_token"`
|
||
// WebhookPath is computed from webhook_token. Always present for webhook
|
||
// triggers; nil for schedule/api. Not stored — see triggerToResponse.
|
||
WebhookPath *string `json:"webhook_path"`
|
||
// WebhookURL is the absolute URL composed from the server's
|
||
// MULTICA_PUBLIC_URL setting. Nil when the server has no public URL
|
||
// configured; clients then build the URL themselves from webhook_path
|
||
// plus their API base / current origin.
|
||
WebhookURL *string `json:"webhook_url"`
|
||
// Provider names the per-endpoint signing/dedupe convention. For now:
|
||
// "generic" (bearer URL only, Idempotency-Key for dedupe) or "github"
|
||
// (X-Hub-Signature-256 + X-GitHub-Delivery). Omitted for non-webhook
|
||
// triggers.
|
||
Provider *string `json:"provider"`
|
||
// HasSigningSecret indicates whether a signing secret is configured on
|
||
// the trigger. The secret itself is never returned — it is set via a
|
||
// dedicated write-only endpoint. Always false for non-webhook triggers.
|
||
HasSigningSecret bool `json:"has_signing_secret"`
|
||
// SigningSecretHint is the last 4 characters of the configured secret,
|
||
// surfaced to help operators tell two secrets apart in the UI. Nil when
|
||
// no secret is configured.
|
||
SigningSecretHint *string `json:"signing_secret_hint"`
|
||
Label *string `json:"label"`
|
||
LastFiredAt *string `json:"last_fired_at"`
|
||
CreatedAt string `json:"created_at"`
|
||
UpdatedAt string `json:"updated_at"`
|
||
// EventFilters is the declared event scope. Only present for webhook
|
||
// triggers; omitted when the trigger accepts all events. Serializes as
|
||
// a JSON array of {event, actions?} objects — never as a base64 string
|
||
// (which is what []byte would produce through encoding/json).
|
||
EventFilters []WebhookEventFilter `json:"event_filters,omitempty"`
|
||
}
|
||
|
||
type AutopilotRunResponse struct {
|
||
ID string `json:"id"`
|
||
AutopilotID string `json:"autopilot_id"`
|
||
TriggerID *string `json:"trigger_id"`
|
||
Source string `json:"source"`
|
||
Status string `json:"status"`
|
||
IssueID *string `json:"issue_id"`
|
||
TaskID *string `json:"task_id"`
|
||
TriggeredAt string `json:"triggered_at"`
|
||
CompletedAt *string `json:"completed_at"`
|
||
FailureReason *string `json:"failure_reason"`
|
||
// ReasonCode is a stable, localizable, enumeration-safe classification of a
|
||
// non-success run (skipped/failed), persisted at the decision source. The UI
|
||
// localizes it instead of echoing the raw English reason (which may name a
|
||
// private assignee agent). Additive: nil for legacy/success-path runs.
|
||
ReasonCode *string `json:"reason_code,omitempty"`
|
||
TriggerPayload any `json:"trigger_payload"`
|
||
Result any `json:"result"`
|
||
CreatedAt string `json:"created_at"`
|
||
}
|
||
|
||
// ── Converters ──────────────────────────────────────────────────────────────
|
||
|
||
func autopilotToResponse(a db.Autopilot, subscribers []db.AutopilotSubscriber) AutopilotResponse {
|
||
assigneeType := a.AssigneeType
|
||
if assigneeType == "" {
|
||
// Older rows pre-MUL-2429 may surface as "" against an out-of-date
|
||
// schema view; default to "agent" so the API contract stays
|
||
// non-null.
|
||
assigneeType = "agent"
|
||
}
|
||
subResp := make([]AutopilotSubscriberEntry, len(subscribers))
|
||
for i, s := range subscribers {
|
||
subResp[i] = AutopilotSubscriberEntry{
|
||
UserType: s.UserType,
|
||
UserID: uuidToString(s.UserID),
|
||
CreatedAt: timestampToString(s.CreatedAt),
|
||
}
|
||
}
|
||
return AutopilotResponse{
|
||
ID: uuidToString(a.ID),
|
||
WorkspaceID: uuidToString(a.WorkspaceID),
|
||
Title: a.Title,
|
||
Description: textToPtr(a.Description),
|
||
ProjectID: uuidToPtr(a.ProjectID),
|
||
AssigneeType: assigneeType,
|
||
AssigneeID: uuidToString(a.AssigneeID),
|
||
Status: a.Status,
|
||
PauseReason: textToPtr(a.PauseReason),
|
||
ExecutionMode: a.ExecutionMode,
|
||
IssueTitleTemplate: textToPtr(a.IssueTitleTemplate),
|
||
CreatedByType: a.CreatedByType,
|
||
CreatedByID: uuidToString(a.CreatedByID),
|
||
LastRunAt: timestampToPtr(a.LastRunAt),
|
||
CreatedAt: timestampToString(a.CreatedAt),
|
||
UpdatedAt: timestampToString(a.UpdatedAt),
|
||
Subscribers: subResp,
|
||
}
|
||
}
|
||
|
||
func (h *Handler) triggerToResponse(t db.AutopilotTrigger) AutopilotTriggerResponse {
|
||
resp := AutopilotTriggerResponse{
|
||
ID: uuidToString(t.ID),
|
||
AutopilotID: uuidToString(t.AutopilotID),
|
||
Kind: t.Kind,
|
||
Enabled: t.Enabled,
|
||
CronExpression: textToPtr(t.CronExpression),
|
||
Timezone: textToPtr(t.Timezone),
|
||
NextRunAt: timestampToPtr(t.NextRunAt),
|
||
WebhookToken: textToPtr(t.WebhookToken),
|
||
Label: textToPtr(t.Label),
|
||
LastFiredAt: timestampToPtr(t.LastFiredAt),
|
||
CreatedAt: timestampToString(t.CreatedAt),
|
||
UpdatedAt: timestampToString(t.UpdatedAt),
|
||
}
|
||
if t.Kind == "webhook" && t.WebhookToken.Valid && t.WebhookToken.String != "" {
|
||
path := webhookPathForToken(t.WebhookToken.String)
|
||
resp.WebhookPath = &path
|
||
if h.cfg.PublicURL != "" {
|
||
full := h.cfg.PublicURL + path
|
||
resp.WebhookURL = &full
|
||
}
|
||
provider := t.Provider
|
||
if provider == "" {
|
||
provider = "generic"
|
||
}
|
||
resp.Provider = &provider
|
||
if t.SigningSecret.Valid && t.SigningSecret.String != "" {
|
||
resp.HasSigningSecret = true
|
||
hint := signingSecretHint(t.SigningSecret.String)
|
||
resp.SigningSecretHint = &hint
|
||
}
|
||
if len(t.EventFilters) > 0 {
|
||
var filters []WebhookEventFilter
|
||
if err := json.Unmarshal(t.EventFilters, &filters); err == nil {
|
||
resp.EventFilters = filters
|
||
}
|
||
// On unmarshal error we deliberately drop the field instead of
|
||
// surfacing raw bytes or 500ing — strict write-time validation
|
||
// is supposed to make this branch unreachable, and the matcher
|
||
// fails closed if a corrupt row ever slips through.
|
||
}
|
||
}
|
||
return resp
|
||
}
|
||
|
||
// signingSecretHint returns the last 4 characters of the signing secret so a
|
||
// configured-vs-rotated state is visible in the UI without exposing the
|
||
// secret itself. Truncating below 4 chars (which the validator already
|
||
// rejects) just returns an empty string.
|
||
func signingSecretHint(secret string) string {
|
||
if len(secret) < 4 {
|
||
return ""
|
||
}
|
||
return secret[len(secret)-4:]
|
||
}
|
||
|
||
// redactWebhookSecrets removes the webhook credential from a trigger response:
|
||
// the bearer token, and the path and URL that embed it. Holding any of the
|
||
// three is equivalent to being able to fire the autopilot from outside the
|
||
// permission system, so they travel together.
|
||
//
|
||
// One definition, shared by the read path's non-writer projection and the
|
||
// broadcast copy (broadcastAutopilotTriggerResponse), so a field added to one
|
||
// cannot be forgotten by the other.
|
||
func redactWebhookSecrets(resp *AutopilotTriggerResponse) {
|
||
resp.WebhookToken = nil
|
||
resp.WebhookPath = nil
|
||
resp.WebhookURL = nil
|
||
}
|
||
|
||
// broadcastAutopilotTriggerResponse strips the webhook credential from a
|
||
// trigger before it goes onto the WebSocket bus. Mutation handlers call it when
|
||
// fanning out autopilot:updated, following the same rule as
|
||
// broadcastAgentResponse: autopilot events reach the WHOLE workspace room —
|
||
// every member regardless of grant, agent processes on their own task tokens
|
||
// included — so a non-redacted broadcast hands them the token GetAutopilot just
|
||
// refused them, through a push that never passed a write gate (MUL-7108).
|
||
// The caller still receives the live value in the HTTP response; only the
|
||
// broadcast copy is redacted.
|
||
//
|
||
// Nothing downstream loses anything: clients treat these events as "refetch
|
||
// this autopilot" (packages/core/realtime/use-realtime-sync.ts), and a writer
|
||
// re-reads the token from the authenticated detail endpoint.
|
||
func broadcastAutopilotTriggerResponse(resp AutopilotTriggerResponse) AutopilotTriggerResponse {
|
||
out := resp
|
||
redactWebhookSecrets(&out)
|
||
return out
|
||
}
|
||
|
||
// webhookPathForToken composes the path used by the public ingress route.
|
||
// Kept as a free function (no Handler receiver) so test code that builds
|
||
// expected URLs without instantiating a Handler can call it.
|
||
func webhookPathForToken(token string) string {
|
||
return "/api/webhooks/autopilots/" + token
|
||
}
|
||
|
||
func runToResponse(r db.AutopilotRun) AutopilotRunResponse {
|
||
var payload any
|
||
if r.TriggerPayload != nil {
|
||
json.Unmarshal(r.TriggerPayload, &payload)
|
||
}
|
||
var result any
|
||
if r.Result != nil {
|
||
json.Unmarshal(r.Result, &result)
|
||
}
|
||
return AutopilotRunResponse{
|
||
ID: uuidToString(r.ID),
|
||
AutopilotID: uuidToString(r.AutopilotID),
|
||
TriggerID: uuidToPtr(r.TriggerID),
|
||
Source: r.Source,
|
||
Status: r.Status,
|
||
IssueID: uuidToPtr(r.IssueID),
|
||
TaskID: uuidToPtr(r.TaskID),
|
||
TriggeredAt: timestampToString(r.TriggeredAt),
|
||
CompletedAt: timestampToPtr(r.CompletedAt),
|
||
FailureReason: textToPtr(r.FailureReason),
|
||
ReasonCode: textToPtr(r.ReasonCode),
|
||
TriggerPayload: payload,
|
||
Result: result,
|
||
CreatedAt: timestampToString(r.CreatedAt),
|
||
}
|
||
}
|
||
|
||
// runToResponseSlim mirrors runToResponse but omits TriggerPayload, intended
|
||
// for list endpoints where echoing the full webhook envelope (up to
|
||
// 256 KiB × N rows) would dominate response size. Clients fetch the full
|
||
// payload via GET /api/autopilots/{id}/runs/{runId} when the user opens
|
||
// the run detail dialog.
|
||
func runToResponseSlim(r db.AutopilotRun) AutopilotRunResponse {
|
||
resp := runToResponse(r)
|
||
resp.TriggerPayload = nil
|
||
return resp
|
||
}
|
||
|
||
// ── Request types ───────────────────────────────────────────────────────────
|
||
|
||
type CreateAutopilotRequest struct {
|
||
Title string `json:"title"`
|
||
Description *string `json:"description"`
|
||
ProjectID *string `json:"project_id"`
|
||
// AssigneeType is optional and defaults to "agent" — preserves backward
|
||
// compatibility with desktop clients shipped before MUL-2429.
|
||
AssigneeType *string `json:"assignee_type"`
|
||
AssigneeID string `json:"assignee_id"`
|
||
ExecutionMode string `json:"execution_mode"`
|
||
IssueTitleTemplate *string `json:"issue_title_template"`
|
||
Subscribers []SubscriberInput `json:"subscribers"`
|
||
}
|
||
|
||
type UpdateAutopilotRequest struct {
|
||
Title *string `json:"title"`
|
||
Description *string `json:"description"`
|
||
ProjectID *string `json:"project_id"`
|
||
AssigneeType *string `json:"assignee_type"`
|
||
AssigneeID *string `json:"assignee_id"`
|
||
Status *string `json:"status"`
|
||
ExecutionMode *string `json:"execution_mode"`
|
||
IssueTitleTemplate *string `json:"issue_title_template"`
|
||
// Wholesale replacement when present; omit to leave subscribers untouched.
|
||
Subscribers []SubscriberInput `json:"subscribers"`
|
||
}
|
||
|
||
type SubscriberInput struct {
|
||
UserType string `json:"user_type"`
|
||
UserID string `json:"user_id"`
|
||
}
|
||
|
||
type CreateAutopilotTriggerRequest struct {
|
||
Kind string `json:"kind"`
|
||
CronExpression *string `json:"cron_expression"`
|
||
Timezone *string `json:"timezone"`
|
||
Label *string `json:"label"`
|
||
// Provider is currently only meaningful for kind=webhook. Allowed
|
||
// values: "generic" (default) or "github". Unset → "generic".
|
||
Provider *string `json:"provider"`
|
||
// EventFilters is an optional list of {event, actions?} scopes. Only
|
||
// meaningful for webhook triggers. nil/empty means "accept all events".
|
||
EventFilters []WebhookEventFilter `json:"event_filters,omitempty"`
|
||
}
|
||
|
||
// SetSigningSecretRequest is the body shape for PUT
|
||
// /api/autopilots/{id}/triggers/{triggerId}/signing-secret. Lives in its own
|
||
// type so the secret never appears alongside other fields on the trigger
|
||
// update path — handlers that log request bodies for debugging cannot pick it
|
||
// up by accident.
|
||
type SetSigningSecretRequest struct {
|
||
// SigningSecret is the new HMAC key. Sending an empty string explicitly
|
||
// clears the secret (disables signature verification). Pass any
|
||
// reasonably entropic value — GitHub's docs recommend at least 32 random
|
||
// characters; we enforce a 16-char minimum on non-empty input.
|
||
SigningSecret string `json:"signing_secret"`
|
||
}
|
||
|
||
type UpdateAutopilotTriggerRequest struct {
|
||
Enabled *bool `json:"enabled"`
|
||
CronExpression *string `json:"cron_expression"`
|
||
Timezone *string `json:"timezone"`
|
||
Label *string `json:"label"`
|
||
// EventFilters is the desired event-filter set with tri-state PATCH
|
||
// semantics:
|
||
//
|
||
// - omitted / explicit null (nil pointer) → leave the existing value
|
||
// untouched.
|
||
// - explicit [] (non-nil, length 0) → clear filters (the trigger
|
||
// reverts to "accept all events").
|
||
// - explicit [...] → replace with the supplied
|
||
// list.
|
||
//
|
||
// This is why the pointer matters: with a plain []WebhookEventFilter
|
||
// there is no way to tell "field absent from the PATCH body" from "field
|
||
// present but empty", and the user can never clear filters once set.
|
||
EventFilters *[]WebhookEventFilter `json:"event_filters,omitempty"`
|
||
}
|
||
|
||
// ── Handlers ────────────────────────────────────────────────────────────────
|
||
|
||
func (h *Handler) ListAutopilots(w http.ResponseWriter, r *http.Request) {
|
||
workspaceID := h.resolveWorkspaceID(r)
|
||
|
||
var statusFilter pgtype.Text
|
||
if s := r.URL.Query().Get("status"); s != "" {
|
||
statusFilter = pgtype.Text{String: s, Valid: true}
|
||
}
|
||
|
||
autopilots, err := h.Queries.ListAutopilots(r.Context(), db.ListAutopilotsParams{
|
||
WorkspaceID: parseUUID(workspaceID),
|
||
Status: statusFilter,
|
||
})
|
||
if err != nil {
|
||
writeError(w, http.StatusInternalServerError, "failed to list autopilots")
|
||
return
|
||
}
|
||
|
||
// Resolve the acting caller's write access for per-row can_write. The
|
||
// collaborator grants are fetched once as a set (keyed by autopilot id) so the
|
||
// flag costs no per-row query. A missing member — an agent run with no
|
||
// originator, or (behind the workspace-member middleware, shouldn't happen) a
|
||
// non-member — just yields can_write=false everywhere.
|
||
caller, callerErr := h.getWorkspaceMember(r.Context(), h.autopilotActingUserID(r, workspaceID), workspaceID)
|
||
collabSet := map[string]struct{}{}
|
||
if callerErr == nil {
|
||
if ids, err := h.Queries.ListAutopilotIDsForCollaborator(r.Context(), caller.UserID); err == nil {
|
||
for _, id := range ids {
|
||
collabSet[uuidToString(id)] = struct{}{}
|
||
}
|
||
}
|
||
}
|
||
|
||
// Subscribers are fetched for the whole page in one batched query. An
|
||
// earlier version passed nil here to dodge an N+1, but the response type
|
||
// serializes a nil slice as [] rather than omitting it, so every listed
|
||
// autopilot claimed to have no subscribers while the detail endpoint
|
||
// reported the real ones — a silently wrong value is worse than a missing
|
||
// one (MUL-6680). The batch keys off the primary key's leading column, so
|
||
// this costs one indexed query per page, not one per row.
|
||
subsByAutopilot := map[string][]db.AutopilotSubscriber{}
|
||
autopilotIDs := make([]pgtype.UUID, 0, len(autopilots))
|
||
for _, row := range autopilots {
|
||
autopilotIDs = append(autopilotIDs, row.Autopilot.ID)
|
||
}
|
||
if len(autopilotIDs) > 0 {
|
||
subs, err := h.Queries.ListAutopilotSubscribersForAutopilots(r.Context(), autopilotIDs)
|
||
if err != nil {
|
||
// Fail closed. Degrading to an empty set here would reintroduce
|
||
// exactly the bug this endpoint was fixed for: subscribers is a
|
||
// non-omitempty field documented as authoritative, so an empty
|
||
// value on a failed read is indistinguishable from "none
|
||
// configured" — and a caller acting on it can overwrite a real
|
||
// subscriber list. An error the caller can see and retry is the
|
||
// only honest answer.
|
||
writeError(w, http.StatusInternalServerError, "failed to list autopilot subscribers")
|
||
return
|
||
}
|
||
for _, s := range subs {
|
||
id := uuidToString(s.AutopilotID)
|
||
subsByAutopilot[id] = append(subsByAutopilot[id], s)
|
||
}
|
||
}
|
||
|
||
resp := make([]AutopilotResponse, len(autopilots))
|
||
for i, row := range autopilots {
|
||
r := autopilotToResponse(row.Autopilot, subsByAutopilot[uuidToString(row.Autopilot.ID)])
|
||
r.TriggerKinds = row.TriggerKinds
|
||
if row.NextRunAt.Valid {
|
||
r.NextRunAt = timestampToPtr(row.NextRunAt)
|
||
}
|
||
if row.LastRunStatus != "" {
|
||
s := row.LastRunStatus
|
||
r.LastRunStatus = &s
|
||
}
|
||
if callerErr == nil {
|
||
_, isCollaborator := collabSet[uuidToString(row.Autopilot.ID)]
|
||
cw := autopilotWriteByOwnership(row.Autopilot, caller) || isCollaborator
|
||
r.CanWrite = &cw
|
||
}
|
||
resp[i] = r
|
||
}
|
||
writeJSON(w, http.StatusOK, map[string]any{"autopilots": resp, "total": len(resp)})
|
||
}
|
||
|
||
func (h *Handler) GetAutopilot(w http.ResponseWriter, r *http.Request) {
|
||
id := chi.URLParam(r, "id")
|
||
workspaceID := h.resolveWorkspaceID(r)
|
||
|
||
autopilot, ok := h.loadAutopilotInWorkspace(w, r, id, workspaceID)
|
||
if !ok {
|
||
return
|
||
}
|
||
|
||
subs, err := h.Queries.ListAutopilotSubscribers(r.Context(), autopilot.ID)
|
||
if err != nil {
|
||
// Fail closed for the same reason the list endpoint does: an empty
|
||
// subscribers array is a claim, not an absence, and this response is
|
||
// what clients round-trip back into a full-replace PATCH.
|
||
writeError(w, http.StatusInternalServerError, "failed to list autopilot subscribers")
|
||
return
|
||
}
|
||
resp := autopilotToResponse(autopilot, subs)
|
||
|
||
// Resolve the acting caller's write access once: it both stamps can_write and
|
||
// gates webhook-secret exposure. Webhook tokens are trigger-granting
|
||
// secrets (anyone who reads the token can fire the autopilot from outside
|
||
// the permission system), so only writers — the creator, a workspace
|
||
// owner/admin, or a granted collaborator — get the live token/URL; every
|
||
// other member sees the trigger metadata with the secret fields stripped
|
||
// (MUL-3807).
|
||
//
|
||
// "Acting" is what makes the write gates worth having: judging the
|
||
// authenticated user would hand an agent its runtime owner's tokens, and a
|
||
// token read here is a trigger the write gates never see (MUL-7108).
|
||
canWrite := false
|
||
canManageAccess := false
|
||
if member, err := h.getWorkspaceMember(r.Context(), h.autopilotActingUserID(r, workspaceID), workspaceID); err == nil {
|
||
canWrite = h.memberCanWriteAutopilot(r.Context(), autopilot, member)
|
||
// Managing the access list is narrower than write: collaborators can
|
||
// write but cannot re-grant (MUL-3807).
|
||
canManageAccess = autopilotWriteByOwnership(autopilot, member)
|
||
}
|
||
resp.CanWrite = &canWrite
|
||
resp.CanManageAccess = &canManageAccess
|
||
|
||
// Include triggers.
|
||
triggers, err := h.Queries.ListAutopilotTriggers(r.Context(), autopilot.ID)
|
||
if err != nil {
|
||
triggers = nil
|
||
}
|
||
triggerResp := make([]AutopilotTriggerResponse, len(triggers))
|
||
for i, t := range triggers {
|
||
tr := h.triggerToResponse(t)
|
||
if !canWrite {
|
||
redactWebhookSecrets(&tr)
|
||
}
|
||
triggerResp[i] = tr
|
||
}
|
||
|
||
// Include the explicit collaborator grants so the "manage access" UI can
|
||
// render the current list without a second round-trip.
|
||
collaborators, err := h.Queries.ListAutopilotCollaborators(r.Context(), autopilot.ID)
|
||
if err != nil {
|
||
collaborators = nil
|
||
}
|
||
collabResp := make([]AutopilotCollaboratorEntry, len(collaborators))
|
||
for i, c := range collaborators {
|
||
collabResp[i] = collaboratorToEntry(c)
|
||
}
|
||
|
||
writeJSON(w, http.StatusOK, map[string]any{
|
||
"autopilot": resp,
|
||
"triggers": triggerResp,
|
||
"collaborators": collabResp,
|
||
})
|
||
}
|
||
|
||
func (h *Handler) loadAutopilotInWorkspace(w http.ResponseWriter, r *http.Request, autopilotID, workspaceID string) (db.Autopilot, bool) {
|
||
autopilotUUID, ok := parseUUIDOrBadRequest(w, autopilotID, "autopilot id")
|
||
if !ok {
|
||
return db.Autopilot{}, false
|
||
}
|
||
wsUUID, ok := parseUUIDOrBadRequest(w, workspaceID, "workspace id")
|
||
if !ok {
|
||
return db.Autopilot{}, false
|
||
}
|
||
|
||
autopilot, err := h.Queries.GetAutopilotInWorkspace(r.Context(), db.GetAutopilotInWorkspaceParams{
|
||
ID: autopilotUUID,
|
||
WorkspaceID: wsUUID,
|
||
})
|
||
if err != nil {
|
||
writeError(w, http.StatusNotFound, "autopilot not found")
|
||
return db.Autopilot{}, false
|
||
}
|
||
return autopilot, true
|
||
}
|
||
|
||
// autopilotWriteByOwnership is the implicit, query-free part of the write
|
||
// predicate: the autopilot's creator and workspace owners/admins always have
|
||
// write access. Explicit collaborator grants (memberCanWriteAutopilot) layer
|
||
// on top of this (MUL-3807).
|
||
func autopilotWriteByOwnership(ap db.Autopilot, member db.Member) bool {
|
||
if roleAllowed(member.Role, "owner", "admin") {
|
||
return true
|
||
}
|
||
return ap.CreatedByType == "member" && uuidToString(ap.CreatedByID) == uuidToString(member.UserID)
|
||
}
|
||
|
||
// memberCanWriteAutopilot reports whether the given member may perform write or
|
||
// execute operations on the autopilot — editing it, deleting it, triggering
|
||
// runs, replaying deliveries, and managing its triggers, webhook secrets, and
|
||
// access list. Write access is held by the autopilot's creator, by workspace
|
||
// owners/admins, and by members explicitly granted as collaborators. The same
|
||
// predicate also gates whether webhook secrets are exposed on the read path,
|
||
// since seeing a webhook token is equivalent to being able to trigger.
|
||
func (h *Handler) memberCanWriteAutopilot(ctx context.Context, ap db.Autopilot, member db.Member) bool {
|
||
if autopilotWriteByOwnership(ap, member) {
|
||
return true
|
||
}
|
||
granted, err := h.Queries.IsAutopilotCollaborator(ctx, db.IsAutopilotCollaboratorParams{
|
||
AutopilotID: ap.ID,
|
||
UserID: member.UserID,
|
||
})
|
||
return err == nil && granted
|
||
}
|
||
|
||
// Stable, machine-readable refusal codes for the autopilot write surface. The
|
||
// CLI keys its actionable output on these rather than on the English sentence,
|
||
// which is what lets a refusal survive translation and copy edits.
|
||
const (
|
||
autopilotNoOriginatorCode = "autopilot_no_originator"
|
||
autopilotForbiddenCode = "autopilot_forbidden"
|
||
autopilotActorNotMemberCode = "autopilot_actor_not_member"
|
||
autopilotTriggerNoOriginatorCode = "autopilot_trigger_no_originator"
|
||
autopilotTriggerForbiddenCode = "autopilot_trigger_forbidden"
|
||
)
|
||
|
||
// autopilotRefusal is the code + sentence a gate answers with when no human
|
||
// backs the request, and when the human it acts for holds no grant. Each gate
|
||
// carries its own pair so a refusal names the operation actually attempted; the
|
||
// resolution below it is shared.
|
||
type autopilotRefusal struct {
|
||
noOriginatorCode string
|
||
noOriginatorMsg string
|
||
forbiddenCode string
|
||
forbiddenMsg string
|
||
}
|
||
|
||
var (
|
||
autopilotWriteRefusal = autopilotRefusal{
|
||
noOriginatorCode: autopilotNoOriginatorCode,
|
||
noOriginatorMsg: "no human authorized this change: the calling run records no originator",
|
||
forbiddenCode: autopilotForbiddenCode,
|
||
forbiddenMsg: "only the autopilot creator, a workspace admin, or a granted collaborator can manage this autopilot",
|
||
}
|
||
// Create has no autopilot to hold a grant on, so its only "forbidden" is
|
||
// an acting human who is not in this workspace. That is a different fact
|
||
// from "lacks access to this autopilot" and gets its own code, so the CLI
|
||
// does not tell someone to ask for a collaborator grant they could not
|
||
// hold (MUL-7108).
|
||
autopilotCreateRefusal = autopilotRefusal{
|
||
noOriginatorCode: autopilotNoOriginatorCode,
|
||
noOriginatorMsg: "no human authorized this autopilot: the calling run records no originator",
|
||
forbiddenCode: autopilotActorNotMemberCode,
|
||
forbiddenMsg: "the person this run acts for is not a member of this workspace",
|
||
}
|
||
autopilotAccessRefusal = autopilotRefusal{
|
||
noOriginatorCode: autopilotNoOriginatorCode,
|
||
noOriginatorMsg: "no human authorized this change: the calling run records no originator",
|
||
forbiddenCode: autopilotForbiddenCode,
|
||
forbiddenMsg: "only the autopilot creator or a workspace admin can manage access",
|
||
}
|
||
autopilotTriggerRefusal = autopilotRefusal{
|
||
noOriginatorCode: autopilotTriggerNoOriginatorCode,
|
||
noOriginatorMsg: "no human authorized this trigger: the calling run records no originator",
|
||
forbiddenCode: autopilotTriggerForbiddenCode,
|
||
forbiddenMsg: "only the autopilot creator, a workspace admin, or a granted collaborator can trigger this autopilot",
|
||
}
|
||
)
|
||
|
||
// autopilotActingUserID resolves the human whose authority a request on this
|
||
// surface spends, without judging or refusing: a member acts for themselves, an
|
||
// agent acts for its run's ORIGINATOR. Returns "" when no human can be
|
||
// attributed. Read paths use it directly — they must not refuse an
|
||
// unattributed caller, only decline to hand it a writer's secrets.
|
||
func (h *Handler) autopilotActingUserID(r *http.Request, workspaceID string) string {
|
||
actorType, actorID := h.resolveActor(r, requestUserID(r), workspaceID)
|
||
return h.invokeOriginatorFromRequest(r, actorType, actorID)
|
||
}
|
||
|
||
// requireAutopilotActingMember resolves the workspace member whose authority an
|
||
// autopilot WRITE spends. Every write gate on this surface, and every identity
|
||
// it stamps into a row, goes through this one seam so the rule cannot drift
|
||
// between endpoints (MUL-7108).
|
||
//
|
||
// A member acts for themselves. An agent — the CLI running inside a task, or an
|
||
// A2A call — acts for its run's ORIGINATOR, the top-of-chain human. That is how
|
||
// every other invoke surface in the codebase judges an agent actor (MUL-3963 /
|
||
// canInvokeAgent), and an autopilot write is an invoke with a delay on it:
|
||
// editing the rule, adding a trigger, rotating a webhook token or granting a
|
||
// collaborator all decide what an agent will be told to do later.
|
||
//
|
||
// Judging the request's authenticated user instead resolves the RUNTIME OWNER
|
||
// under a task token (middleware/auth.go stamps the token's bound user, minted
|
||
// from rt.OwnerID), which is wrong in both directions. Too broad: it makes every
|
||
// agent a standing proxy for its machine's owner, so anyone who can talk to that
|
||
// agent inherits the owner's autopilot rights without holding any themselves —
|
||
// the escalation MUL-7108 reports. Too narrow, if kept as an ADDITIONAL
|
||
// condition: it would refuse a member who may edit this autopilot themselves
|
||
// merely because the agent they asked happens to run on a third party's machine,
|
||
// which is the shape #8099 rejected for the trigger endpoint. Workspace tenancy
|
||
// for the authenticated caller is already enforced by the router's
|
||
// RequireWorkspaceMember middleware, so judging the acting human alone loses no
|
||
// isolation.
|
||
//
|
||
// No resolvable human → refuse. A run carrying no originator is one this
|
||
// codebase has already decided authorizes nothing (rule_owner is documented
|
||
// "deliberately powerless"; a terminal task lends nothing), so it cannot hold
|
||
// the pen here either.
|
||
//
|
||
// On refusal the response is written and ok is false; the caller must return.
|
||
func (h *Handler) requireAutopilotActingMember(w http.ResponseWriter, r *http.Request, workspaceID string, refusal autopilotRefusal) (db.Member, bool) {
|
||
userID, ok := requireUserID(w, r)
|
||
if !ok {
|
||
return db.Member{}, false
|
||
}
|
||
actorType, actorID := h.resolveActor(r, userID, workspaceID)
|
||
if actorType != "agent" {
|
||
// A member acts for themselves: unchanged path, keeping the context
|
||
// member and the 404 requireWorkspaceMember writes for a non-member —
|
||
// a write refusal must not confirm that this workspace exists.
|
||
return h.workspaceMember(w, r, workspaceID)
|
||
}
|
||
originator := h.invokeOriginatorFromRequest(r, actorType, actorID)
|
||
if originator == "" {
|
||
// Distinct from the permission refusal below: nothing was denied, there
|
||
// was simply nobody to judge. It carries its own code because it is the
|
||
// one refusal here a workspace can act on.
|
||
writeErrorCode(w, http.StatusForbidden, refusal.noOriginatorCode, refusal.noOriginatorMsg)
|
||
return db.Member{}, false
|
||
}
|
||
member, err := h.getWorkspaceMember(r.Context(), originator, workspaceID)
|
||
if err != nil {
|
||
writeErrorCode(w, http.StatusForbidden, refusal.forbiddenCode, refusal.forbiddenMsg)
|
||
return db.Member{}, false
|
||
}
|
||
return member, true
|
||
}
|
||
|
||
// requireAutopilotWrite enforces memberCanWriteAutopilot for a mutating/
|
||
// executing request, judging the member the request acts for
|
||
// (requireAutopilotActingMember). On failure it writes the response (404 when a
|
||
// member caller is not in the workspace, 403 otherwise) and returns ok=false;
|
||
// the caller must return early. On success it returns that member — the identity
|
||
// the handler must also stamp into whatever it writes, so the gate and the row
|
||
// can never name two different people (MUL-7108).
|
||
func (h *Handler) requireAutopilotWrite(w http.ResponseWriter, r *http.Request, ap db.Autopilot, workspaceID string) (db.Member, bool) {
|
||
member, ok := h.requireAutopilotActingMember(w, r, workspaceID, autopilotWriteRefusal)
|
||
if !ok {
|
||
return db.Member{}, false
|
||
}
|
||
if !h.memberCanWriteAutopilot(r.Context(), ap, member) {
|
||
writeErrorCode(w, http.StatusForbidden, autopilotWriteRefusal.forbiddenCode, autopilotWriteRefusal.forbiddenMsg)
|
||
return db.Member{}, false
|
||
}
|
||
return member, true
|
||
}
|
||
|
||
// requireAutopilotAccessManagement enforces the narrower predicate used by the
|
||
// collaborator (access list) endpoints: only the autopilot's creator or a
|
||
// workspace owner/admin may grant or revoke access. A granted collaborator
|
||
// keeps its own write/execute rights (edit, trigger, manage triggers/secrets)
|
||
// but cannot manage the access list — this stops a collaborator from
|
||
// re-granting access to others or revoking peers (privilege escalation).
|
||
// See MUL-3807. Like every gate here it judges the acting member, since a grant
|
||
// is the most durable write on this surface: it outlives the run that made it.
|
||
func (h *Handler) requireAutopilotAccessManagement(w http.ResponseWriter, r *http.Request, ap db.Autopilot, workspaceID string) (db.Member, bool) {
|
||
member, ok := h.requireAutopilotActingMember(w, r, workspaceID, autopilotAccessRefusal)
|
||
if !ok {
|
||
return db.Member{}, false
|
||
}
|
||
if !autopilotWriteByOwnership(ap, member) {
|
||
writeErrorCode(w, http.StatusForbidden, autopilotAccessRefusal.forbiddenCode, autopilotAccessRefusal.forbiddenMsg)
|
||
return db.Member{}, false
|
||
}
|
||
return member, true
|
||
}
|
||
|
||
func (h *Handler) CreateAutopilot(w http.ResponseWriter, r *http.Request) {
|
||
var req CreateAutopilotRequest
|
||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||
writeError(w, http.StatusBadRequest, "invalid request body")
|
||
return
|
||
}
|
||
if req.Title == "" {
|
||
writeError(w, http.StatusBadRequest, "title is required")
|
||
return
|
||
}
|
||
if req.AssigneeID == "" {
|
||
writeError(w, http.StatusBadRequest, "assignee_id is required")
|
||
return
|
||
}
|
||
if req.ExecutionMode == "" {
|
||
writeError(w, http.StatusBadRequest, "execution_mode is required")
|
||
return
|
||
}
|
||
if req.ExecutionMode != "create_issue" && req.ExecutionMode != "run_only" {
|
||
writeError(w, http.StatusBadRequest, "execution_mode must be create_issue or run_only")
|
||
return
|
||
}
|
||
if req.IssueTitleTemplate != nil {
|
||
if err := service.ValidateIssueTitleTemplate(*req.IssueTitleTemplate); err != nil {
|
||
writeError(w, http.StatusBadRequest, err.Error())
|
||
return
|
||
}
|
||
}
|
||
|
||
workspaceID := h.resolveWorkspaceID(r)
|
||
// Everything this handler stamps — created_by, the v1 rule version, the
|
||
// realtime event, the analytics actor — is the human the create acts for,
|
||
// never the runtime owner whose token authenticated it. created_by is the
|
||
// autopilot's own write grant, so stamping the wrong human here would hand
|
||
// that grant to a machine's owner and lock the requesting member out of the
|
||
// autopilot they just asked for (MUL-7108).
|
||
actor, ok := h.requireAutopilotActingMember(w, r, workspaceID, autopilotCreateRefusal)
|
||
if !ok {
|
||
return
|
||
}
|
||
userID := uuidToString(actor.UserID)
|
||
|
||
assigneeUUID, ok := parseUUIDOrBadRequest(w, req.AssigneeID, "assignee_id")
|
||
if !ok {
|
||
return
|
||
}
|
||
wsUUID, ok := parseUUIDOrBadRequest(w, workspaceID, "workspace id")
|
||
if !ok {
|
||
return
|
||
}
|
||
|
||
assigneeType := "agent"
|
||
if req.AssigneeType != nil && *req.AssigneeType != "" {
|
||
assigneeType = *req.AssigneeType
|
||
}
|
||
if !isValidAutopilotAssigneeType(assigneeType) {
|
||
writeError(w, http.StatusBadRequest, "assignee_type must be agent or squad")
|
||
return
|
||
}
|
||
projectID, ok := h.parseAutopilotProjectID(w, r, req.ProjectID, wsUUID)
|
||
if !ok {
|
||
return
|
||
}
|
||
|
||
// Parse before insert so a malformed payload doesn't open a transaction.
|
||
subscribers, ok := parseAutopilotSubscribers(w, req.Subscribers)
|
||
if !ok {
|
||
return
|
||
}
|
||
|
||
tx, err := h.TxStarter.Begin(r.Context())
|
||
if err != nil {
|
||
writeError(w, http.StatusInternalServerError, "failed to create autopilot")
|
||
return
|
||
}
|
||
defer tx.Rollback(r.Context())
|
||
qtx := h.Queries.WithTx(tx)
|
||
|
||
// This must be the first lock family in the transaction. Member revocation
|
||
// takes the same per-(workspace, user) locks before pruning templates and
|
||
// deleting member rows. Re-checking membership after those locks means
|
||
// either this save commits first and revocation prunes it, or revocation
|
||
// commits first and this save rejects the departed subscriber.
|
||
if !h.lockAndValidateAutopilotSubscribers(w, r, qtx, subscribers, wsUUID) {
|
||
return
|
||
}
|
||
|
||
// Keep save-time readiness validation in the same transaction as the
|
||
// insert. The assignment lock serializes this path with Runtime teardown,
|
||
// so an active Autopilot cannot slip in after teardown's pause sweep.
|
||
if !h.validateAutopilotAssigneeForSave(w, r, qtx, assigneeType, assigneeUUID, wsUUID, true) {
|
||
return
|
||
}
|
||
|
||
autopilot, err := qtx.CreateAutopilot(r.Context(), db.CreateAutopilotParams{
|
||
WorkspaceID: wsUUID,
|
||
Title: req.Title,
|
||
AssigneeType: assigneeType,
|
||
AssigneeID: assigneeUUID,
|
||
Status: "active",
|
||
ExecutionMode: req.ExecutionMode,
|
||
CreatedByType: "member",
|
||
CreatedByID: actor.UserID,
|
||
Description: ptrToText(req.Description),
|
||
IssueTitleTemplate: ptrToText(req.IssueTitleTemplate),
|
||
ProjectID: projectID,
|
||
})
|
||
if err != nil {
|
||
writeError(w, http.StatusInternalServerError, "failed to create autopilot")
|
||
return
|
||
}
|
||
|
||
// Creating an autopilot IS a substantive publish: append rule-version v1 with
|
||
// the creating member as publisher, so every autopilot has an accountable
|
||
// human at dispatch time (MUL-4302 §3.4).
|
||
if err := h.recordAutopilotRuleVersion(r.Context(), qtx, autopilot, "member", actor.UserID); err != nil {
|
||
writeError(w, http.StatusInternalServerError, "failed to create autopilot")
|
||
return
|
||
}
|
||
|
||
for _, subscriber := range subscribers {
|
||
if err := qtx.AddAutopilotSubscriber(r.Context(), db.AddAutopilotSubscriberParams{
|
||
AutopilotID: autopilot.ID,
|
||
UserType: "member",
|
||
UserID: subscriber.UserID,
|
||
}); err != nil {
|
||
writeError(w, http.StatusInternalServerError, "failed to add autopilot subscriber")
|
||
return
|
||
}
|
||
}
|
||
if err := tx.Commit(r.Context()); err != nil {
|
||
writeError(w, http.StatusInternalServerError, "failed to create autopilot")
|
||
return
|
||
}
|
||
subs, err := h.Queries.ListAutopilotSubscribers(r.Context(), autopilot.ID)
|
||
if err != nil {
|
||
subs = nil
|
||
}
|
||
|
||
resp := autopilotToResponse(autopilot, subs)
|
||
h.publish(protocol.EventAutopilotCreated, workspaceID, "member", userID, map[string]any{"autopilot": resp})
|
||
obsmetrics.RecordEvent(h.Analytics, h.Metrics, analytics.AutopilotCreated(
|
||
userID,
|
||
workspaceID,
|
||
uuidToString(autopilot.ID),
|
||
"manual",
|
||
"manual",
|
||
))
|
||
writeJSON(w, http.StatusCreated, resp)
|
||
}
|
||
|
||
type autopilotSubscriberCandidate struct {
|
||
UserID pgtype.UUID
|
||
InputIndex int
|
||
}
|
||
|
||
// parseAutopilotSubscribers validates the wire shape without reading mutable
|
||
// membership state. Membership is checked only after the write transaction
|
||
// owns the same serialization locks as member revocation.
|
||
func parseAutopilotSubscribers(w http.ResponseWriter, raw []SubscriberInput) ([]autopilotSubscriberCandidate, bool) {
|
||
if len(raw) == 0 {
|
||
return nil, true
|
||
}
|
||
out := make([]autopilotSubscriberCandidate, 0, len(raw))
|
||
seen := make(map[string]bool, len(raw))
|
||
for i, entry := range raw {
|
||
if entry.UserType != "member" {
|
||
writeError(w, http.StatusBadRequest, fmt.Sprintf("subscribers[%d].user_type must be 'member'", i))
|
||
return nil, false
|
||
}
|
||
if entry.UserID == "" {
|
||
writeError(w, http.StatusBadRequest, fmt.Sprintf("subscribers[%d].user_id is required", i))
|
||
return nil, false
|
||
}
|
||
uid, ok := parseUUIDOrBadRequest(w, entry.UserID, fmt.Sprintf("subscribers[%d].user_id", i))
|
||
if !ok {
|
||
return nil, false
|
||
}
|
||
canonicalID := uuidToString(uid)
|
||
if seen[canonicalID] {
|
||
continue
|
||
}
|
||
seen[canonicalID] = true
|
||
out = append(out, autopilotSubscriberCandidate{UserID: uid, InputIndex: i})
|
||
}
|
||
return out, true
|
||
}
|
||
|
||
// lockAndValidateAutopilotSubscribers serializes subscriber-template writes
|
||
// with member revocation. Locks and row checks both use canonical UUID order,
|
||
// so two saves containing the same members in different request orders cannot
|
||
// deadlock each other.
|
||
func (h *Handler) lockAndValidateAutopilotSubscribers(
|
||
w http.ResponseWriter,
|
||
r *http.Request,
|
||
qtx *db.Queries,
|
||
subscribers []autopilotSubscriberCandidate,
|
||
workspaceID pgtype.UUID,
|
||
) bool {
|
||
ordered := append([]autopilotSubscriberCandidate(nil), subscribers...)
|
||
sort.Slice(ordered, func(i, j int) bool {
|
||
return uuidToString(ordered[i].UserID) < uuidToString(ordered[j].UserID)
|
||
})
|
||
|
||
for _, subscriber := range ordered {
|
||
if err := qtx.LockSubscriberWrites(r.Context(), db.LockSubscriberWritesParams{
|
||
WorkspaceID: workspaceID,
|
||
UserID: subscriber.UserID,
|
||
}); err != nil {
|
||
writeError(w, http.StatusInternalServerError, "failed to validate autopilot subscribers")
|
||
return false
|
||
}
|
||
}
|
||
for _, subscriber := range ordered {
|
||
if _, err := qtx.LockActiveMember(r.Context(), db.LockActiveMemberParams{
|
||
UserID: subscriber.UserID,
|
||
WorkspaceID: workspaceID,
|
||
}); err != nil {
|
||
if errors.Is(err, pgx.ErrNoRows) {
|
||
writeError(w, http.StatusBadRequest, fmt.Sprintf(
|
||
"subscribers[%d] is not a member of this workspace", subscriber.InputIndex,
|
||
))
|
||
return false
|
||
}
|
||
writeError(w, http.StatusInternalServerError, "failed to validate autopilot subscribers")
|
||
return false
|
||
}
|
||
}
|
||
return true
|
||
}
|
||
|
||
func (h *Handler) UpdateAutopilot(w http.ResponseWriter, r *http.Request) {
|
||
id := chi.URLParam(r, "id")
|
||
workspaceID := h.resolveWorkspaceID(r)
|
||
|
||
prev, ok := h.loadAutopilotInWorkspace(w, r, id, workspaceID)
|
||
if !ok {
|
||
return
|
||
}
|
||
actor, ok := h.requireAutopilotWrite(w, r, prev, workspaceID)
|
||
if !ok {
|
||
return
|
||
}
|
||
// The publisher recorded below is the member the gate just judged, so the
|
||
// rule version names whoever actually authorized the edit (MUL-7108).
|
||
userID := uuidToString(actor.UserID)
|
||
|
||
bodyBytes, err := io.ReadAll(r.Body)
|
||
if err != nil {
|
||
writeError(w, http.StatusBadRequest, "failed to read request body")
|
||
return
|
||
}
|
||
var req UpdateAutopilotRequest
|
||
if err := json.Unmarshal(bodyBytes, &req); err != nil {
|
||
writeError(w, http.StatusBadRequest, "invalid request body")
|
||
return
|
||
}
|
||
var rawFields map[string]json.RawMessage
|
||
json.Unmarshal(bodyBytes, &rawFields)
|
||
|
||
params := db.UpdateAutopilotParams{
|
||
ID: prev.ID,
|
||
Description: prev.Description,
|
||
AssigneeID: prev.AssigneeID,
|
||
IssueTitleTemplate: prev.IssueTitleTemplate,
|
||
ProjectID: prev.ProjectID,
|
||
}
|
||
if req.Title != nil {
|
||
params.Title = pgtype.Text{String: *req.Title, Valid: true}
|
||
}
|
||
if req.Status != nil {
|
||
params.Status = pgtype.Text{String: *req.Status, Valid: true}
|
||
}
|
||
if req.ExecutionMode != nil {
|
||
params.ExecutionMode = pgtype.Text{String: *req.ExecutionMode, Valid: true}
|
||
}
|
||
if _, ok := rawFields["description"]; ok {
|
||
params.Description = ptrToText(req.Description)
|
||
}
|
||
if _, ok := rawFields["issue_title_template"]; ok {
|
||
if req.IssueTitleTemplate != nil {
|
||
if err := service.ValidateIssueTitleTemplate(*req.IssueTitleTemplate); err != nil {
|
||
writeError(w, http.StatusBadRequest, err.Error())
|
||
return
|
||
}
|
||
}
|
||
params.IssueTitleTemplate = ptrToText(req.IssueTitleTemplate)
|
||
}
|
||
if _, ok := rawFields["project_id"]; ok {
|
||
projectID, ok := h.parseAutopilotProjectID(w, r, req.ProjectID, prev.WorkspaceID)
|
||
if !ok {
|
||
return
|
||
}
|
||
params.ProjectID = projectID
|
||
}
|
||
// assignee_type and assignee_id are validated as a pair: switching
|
||
// between agent and squad without supplying a new id would leave the
|
||
// row pointing at the wrong table. The client is expected to send both
|
||
// fields on any change; partial updates that change only one are
|
||
// rejected.
|
||
_, typeSent := rawFields["assignee_type"]
|
||
_, idSent := rawFields["assignee_id"]
|
||
nextType := prev.AssigneeType
|
||
nextID := prev.AssigneeID
|
||
if typeSent || idSent {
|
||
if typeSent && req.AssigneeType != nil && *req.AssigneeType != "" {
|
||
nextType = *req.AssigneeType
|
||
}
|
||
if !isValidAutopilotAssigneeType(nextType) {
|
||
writeError(w, http.StatusBadRequest, "assignee_type must be agent or squad")
|
||
return
|
||
}
|
||
if idSent {
|
||
if req.AssigneeID == nil {
|
||
writeError(w, http.StatusBadRequest, "assignee_id cannot be null")
|
||
return
|
||
}
|
||
parsed, ok := parseUUIDOrBadRequest(w, *req.AssigneeID, "assignee_id")
|
||
if !ok {
|
||
return
|
||
}
|
||
nextID = parsed
|
||
}
|
||
// Reject the agent↔squad switch without a paired id, otherwise the
|
||
// row would address agent(id) under assignee_type='squad' or vice
|
||
// versa.
|
||
if typeSent && !idSent && nextType != prev.AssigneeType {
|
||
writeError(w, http.StatusBadRequest, "assignee_id is required when changing assignee_type")
|
||
return
|
||
}
|
||
if typeSent {
|
||
params.AssigneeType = pgtype.Text{String: nextType, Valid: true}
|
||
}
|
||
if idSent {
|
||
params.AssigneeID = nextID
|
||
}
|
||
}
|
||
|
||
// Parse subscriber wire values up-front. Mutable membership is revalidated
|
||
// under the revocation serialization locks after the transaction starts.
|
||
var (
|
||
subscribers []autopilotSubscriberCandidate
|
||
replaceSubscribers bool
|
||
)
|
||
if _, sent := rawFields["subscribers"]; sent {
|
||
replaceSubscribers = true
|
||
validated, vok := parseAutopilotSubscribers(w, req.Subscribers)
|
||
if !vok {
|
||
return
|
||
}
|
||
subscribers = validated
|
||
}
|
||
|
||
tx, err := h.TxStarter.Begin(r.Context())
|
||
if err != nil {
|
||
writeError(w, http.StatusInternalServerError, "failed to update autopilot")
|
||
return
|
||
}
|
||
defer tx.Rollback(r.Context())
|
||
qtx := h.Queries.WithTx(tx)
|
||
|
||
// Lock subscriber identities before assignment and autopilot rows. This is
|
||
// the same global order used by CreateAutopilot and member revocation.
|
||
if !h.lockAndValidateAutopilotSubscribers(w, r, qtx, subscribers, prev.WorkspaceID) {
|
||
return
|
||
}
|
||
|
||
// Retargeting must validate the polymorphic reference; resuming must also
|
||
// validate Runtime readiness. Keep both in this transaction so Runtime
|
||
// teardown either pauses an active row after it commits or wins first and
|
||
// makes activation fail with a useful recovery message.
|
||
nextStatus := prev.Status
|
||
if req.Status != nil {
|
||
nextStatus = *req.Status
|
||
}
|
||
validateAssignee := typeSent || idSent || (req.Status != nil && *req.Status == "active")
|
||
if validateAssignee && !h.validateAutopilotAssigneeForSave(
|
||
w, r, qtx, nextType, nextID, prev.WorkspaceID, nextStatus == "active",
|
||
) {
|
||
return
|
||
}
|
||
|
||
// Assignment locks come first. Runtime teardown and squad leader changes
|
||
// also lock Agent/Squad before they update matching Autopilot rows; keeping
|
||
// that global order prevents an Agent↔Autopilot deadlock.
|
||
lockedPrev, err := qtx.LockAutopilotForUpdate(r.Context(), db.LockAutopilotForUpdateParams{
|
||
ID: prev.ID,
|
||
WorkspaceID: prev.WorkspaceID,
|
||
})
|
||
if errors.Is(err, pgx.ErrNoRows) {
|
||
writeError(w, http.StatusNotFound, "autopilot not found")
|
||
return
|
||
}
|
||
if err != nil {
|
||
writeError(w, http.StatusInternalServerError, "failed to update autopilot")
|
||
return
|
||
}
|
||
if lockedPrev.UpdatedAt.Valid != prev.UpdatedAt.Valid ||
|
||
(lockedPrev.UpdatedAt.Valid && !lockedPrev.UpdatedAt.Time.Equal(prev.UpdatedAt.Time)) {
|
||
writeJSON(w, http.StatusConflict, map[string]any{
|
||
"error": "the autopilot changed while it was being edited; reload and try again.",
|
||
"code": "autopilot_update_conflict",
|
||
})
|
||
return
|
||
}
|
||
|
||
autopilot, err := qtx.UpdateAutopilot(r.Context(), params)
|
||
if err != nil {
|
||
writeError(w, http.StatusInternalServerError, "failed to update autopilot")
|
||
return
|
||
}
|
||
|
||
// A substantive change (target / enabled-state / execution mode) republishes the
|
||
// rule: append a new version with THIS member as publisher, recording who last
|
||
// changed what the rule does. Cosmetic edits (title / description / template)
|
||
// write no version (MUL-4302 §3.4). Since MUL-6951 the rule publisher is an
|
||
// AUDIT value only — it is the coarse fallback a run degrades to when its
|
||
// trigger has no created_by principal, and such a run carries no authorization.
|
||
if autopilotRuleSubstantiveChange(prev, autopilot) {
|
||
if err := h.recordAutopilotRuleVersion(r.Context(), qtx, autopilot, "member", actor.UserID); err != nil {
|
||
writeError(w, http.StatusInternalServerError, "failed to update autopilot")
|
||
return
|
||
}
|
||
// An autopilot-level substantive edit governs every trigger, so CONFIG
|
||
// responsibility for each one transfers to this editor. A trigger-scoped edit
|
||
// re-stamps only its own row (see UpdateAutopilotTrigger). Since MUL-6951 this
|
||
// moves published_by alone: the runs each trigger fires keep acting as, and
|
||
// stay accountable to, that trigger's created_by, which no edit rewrites.
|
||
if err := qtx.SetAutopilotTriggerPublishersByAutopilot(r.Context(), db.SetAutopilotTriggerPublishersByAutopilotParams{
|
||
AutopilotID: autopilot.ID,
|
||
PublishedByType: pgtype.Text{String: "member", Valid: true},
|
||
PublishedByID: actor.UserID,
|
||
}); err != nil {
|
||
writeError(w, http.StatusInternalServerError, "failed to update autopilot")
|
||
return
|
||
}
|
||
}
|
||
|
||
if replaceSubscribers {
|
||
if err := qtx.DeleteAutopilotSubscribersForAutopilot(r.Context(), autopilot.ID); err != nil {
|
||
writeError(w, http.StatusInternalServerError, "failed to update subscribers")
|
||
return
|
||
}
|
||
for _, subscriber := range subscribers {
|
||
if err := qtx.AddAutopilotSubscriber(r.Context(), db.AddAutopilotSubscriberParams{
|
||
AutopilotID: autopilot.ID,
|
||
UserType: "member",
|
||
UserID: subscriber.UserID,
|
||
}); err != nil {
|
||
writeError(w, http.StatusInternalServerError, "failed to add autopilot subscriber")
|
||
return
|
||
}
|
||
}
|
||
}
|
||
|
||
if err := tx.Commit(r.Context()); err != nil {
|
||
writeError(w, http.StatusInternalServerError, "failed to update autopilot")
|
||
return
|
||
}
|
||
|
||
subs, err := h.Queries.ListAutopilotSubscribers(r.Context(), autopilot.ID)
|
||
if err != nil {
|
||
subs = nil
|
||
}
|
||
resp := autopilotToResponse(autopilot, subs)
|
||
h.publish(protocol.EventAutopilotUpdated, workspaceID, "member", userID, map[string]any{"autopilot": resp})
|
||
writeJSON(w, http.StatusOK, resp)
|
||
}
|
||
|
||
// autopilotRuleSubstantiveChange reports whether a substantive (publish-worthy)
|
||
// field of the autopilot ROW changed between prev and next — a change that alters
|
||
// WHAT the automation instructs the agent to do, or WHO / WHETHER it runs, and so
|
||
// transfers accountability to the editor (MUL-4302 §3.4; boundary pinned with Elon):
|
||
//
|
||
// - assignee_type / assignee_id — who (agent / squad leader) executes;
|
||
// - status — enabled state (active / paused / archived);
|
||
// - execution_mode — run_only vs create_issue;
|
||
// - description — the product surfaces this as the run PROMPT, i.e. the task
|
||
// instruction itself, so editing it must transfer responsibility (the gap Elon
|
||
// flagged: a fresh publisher of the instructions is the accountable human);
|
||
// - issue_title_template — templates the created issue in create_issue mode; part
|
||
// of the instruction / output spec the run produces.
|
||
//
|
||
// Deliberately NOT substantive (cosmetic / routing — they change neither the
|
||
// instruction nor the executor): title (display label) and project_id (which project
|
||
// created issues are filed under). The comparison is faithful because UpdateAutopilot
|
||
// seeds every param from prev, so an omitted field round-trips unchanged.
|
||
//
|
||
// Trigger-table edits (cron / timezone / enabled / event_filters) are substantive PER
|
||
// TRIGGER and handled in UpdateAutopilotTrigger; archive and system-pause republish in
|
||
// their own paths.
|
||
func autopilotRuleSubstantiveChange(prev, next db.Autopilot) bool {
|
||
return prev.AssigneeType != next.AssigneeType ||
|
||
prev.AssigneeID != next.AssigneeID ||
|
||
prev.Status != next.Status ||
|
||
prev.ExecutionMode != next.ExecutionMode ||
|
||
prev.Description != next.Description ||
|
||
prev.IssueTitleTemplate != next.IssueTitleTemplate
|
||
}
|
||
|
||
// recordAutopilotRuleVersion appends one rule-version snapshot for a substantive
|
||
// publish (MUL-4302 §3.4). Thin handler wrapper over service.RecordAutopilotRuleVersion
|
||
// (shared with the failure monitor); callers pass their tx-scoped Queries so the
|
||
// version is atomic with the autopilot write.
|
||
func (h *Handler) recordAutopilotRuleVersion(ctx context.Context, q *db.Queries, ap db.Autopilot, publishedByType string, publishedByID pgtype.UUID) error {
|
||
return service.RecordAutopilotRuleVersion(ctx, q, ap, publishedByType, publishedByID)
|
||
}
|
||
|
||
func (h *Handler) parseAutopilotProjectID(
|
||
w http.ResponseWriter,
|
||
r *http.Request,
|
||
raw *string,
|
||
workspaceID pgtype.UUID,
|
||
) (pgtype.UUID, bool) {
|
||
if raw == nil || *raw == "" {
|
||
return pgtype.UUID{}, true
|
||
}
|
||
projectID, ok := parseUUIDOrBadRequest(w, *raw, "project_id")
|
||
if !ok {
|
||
return pgtype.UUID{}, false
|
||
}
|
||
if _, err := h.Queries.GetProjectInWorkspace(r.Context(), db.GetProjectInWorkspaceParams{
|
||
ID: projectID,
|
||
WorkspaceID: workspaceID,
|
||
}); err != nil {
|
||
writeError(w, http.StatusBadRequest, "project_id must reference a project in this workspace")
|
||
return pgtype.UUID{}, false
|
||
}
|
||
return projectID, true
|
||
}
|
||
|
||
func (h *Handler) DeleteAutopilot(w http.ResponseWriter, r *http.Request) {
|
||
id := chi.URLParam(r, "id")
|
||
workspaceID := h.resolveWorkspaceID(r)
|
||
|
||
idUUID, ok := parseUUIDOrBadRequest(w, id, "autopilot id")
|
||
if !ok {
|
||
return
|
||
}
|
||
wsUUID, ok := parseUUIDOrBadRequest(w, workspaceID, "workspace id")
|
||
if !ok {
|
||
return
|
||
}
|
||
|
||
ap, err := h.Queries.GetAutopilotInWorkspace(r.Context(), db.GetAutopilotInWorkspaceParams{
|
||
ID: idUUID,
|
||
WorkspaceID: wsUUID,
|
||
})
|
||
if err != nil {
|
||
writeError(w, http.StatusNotFound, "autopilot not found")
|
||
return
|
||
}
|
||
actor, ok := h.requireAutopilotWrite(w, r, ap, workspaceID)
|
||
if !ok {
|
||
return
|
||
}
|
||
userID := uuidToString(actor.UserID)
|
||
|
||
// Product "delete" is archival: stop future triggers and hide the
|
||
// autopilot from default lists while preserving runs, tasks, webhook
|
||
// deliveries, subscribers, and collaborators as execution history.
|
||
// Archiving is a substantive status change (MUL-4302 §3.4), so republish the
|
||
// rule version with this member as publisher, atomically with the archive.
|
||
tx, err := h.TxStarter.Begin(r.Context())
|
||
if err != nil {
|
||
writeError(w, http.StatusInternalServerError, "failed to delete autopilot")
|
||
return
|
||
}
|
||
defer tx.Rollback(r.Context())
|
||
qtx := h.Queries.WithTx(tx)
|
||
|
||
if err := qtx.ArchiveAutopilot(r.Context(), idUUID); err != nil {
|
||
writeError(w, http.StatusInternalServerError, "failed to delete autopilot")
|
||
return
|
||
}
|
||
ap.Status = "archived" // reflect the post-archive state in the version snapshot
|
||
if err := h.recordAutopilotRuleVersion(r.Context(), qtx, ap, "member", actor.UserID); err != nil {
|
||
writeError(w, http.StatusInternalServerError, "failed to delete autopilot")
|
||
return
|
||
}
|
||
if err := tx.Commit(r.Context()); err != nil {
|
||
writeError(w, http.StatusInternalServerError, "failed to delete autopilot")
|
||
return
|
||
}
|
||
|
||
h.publish(protocol.EventAutopilotDeleted, workspaceID, "member", userID, map[string]any{"autopilot_id": uuidToString(idUUID)})
|
||
w.WriteHeader(http.StatusNoContent)
|
||
}
|
||
|
||
// ── Collaborator (access grant) management ───────────────────────────────────
|
||
|
||
type AutopilotCollaboratorRequest struct {
|
||
UserID string `json:"user_id"`
|
||
}
|
||
|
||
func (h *Handler) writeAutopilotCollaborators(w http.ResponseWriter, r *http.Request, autopilotID pgtype.UUID, status int) {
|
||
collaborators, err := h.Queries.ListAutopilotCollaborators(r.Context(), autopilotID)
|
||
if err != nil {
|
||
writeError(w, http.StatusInternalServerError, "failed to load collaborators")
|
||
return
|
||
}
|
||
resp := make([]AutopilotCollaboratorEntry, len(collaborators))
|
||
for i, c := range collaborators {
|
||
resp[i] = collaboratorToEntry(c)
|
||
}
|
||
writeJSON(w, status, map[string]any{"collaborators": resp})
|
||
}
|
||
|
||
// AddAutopilotCollaborator grants a workspace member explicit write access to
|
||
// the autopilot. Only the autopilot's creator or a workspace owner/admin can
|
||
// manage the access list; a granted collaborator cannot re-grant to others
|
||
// (privilege escalation). See MUL-3807.
|
||
func (h *Handler) AddAutopilotCollaborator(w http.ResponseWriter, r *http.Request) {
|
||
id := chi.URLParam(r, "id")
|
||
workspaceID := h.resolveWorkspaceID(r)
|
||
|
||
ap, ok := h.loadAutopilotInWorkspace(w, r, id, workspaceID)
|
||
if !ok {
|
||
return
|
||
}
|
||
actor, ok := h.requireAutopilotAccessManagement(w, r, ap, workspaceID)
|
||
if !ok {
|
||
return
|
||
}
|
||
|
||
var req AutopilotCollaboratorRequest
|
||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||
writeError(w, http.StatusBadRequest, "invalid request body")
|
||
return
|
||
}
|
||
if req.UserID == "" {
|
||
writeError(w, http.StatusBadRequest, "user_id is required")
|
||
return
|
||
}
|
||
targetUUID, ok := parseUUIDOrBadRequest(w, req.UserID, "user_id")
|
||
if !ok {
|
||
return
|
||
}
|
||
// Only workspace members can be granted access — agents already reach
|
||
// autopilots through their own dispatch path, not this grant list.
|
||
if !h.isWorkspaceEntity(r.Context(), "member", req.UserID, workspaceID) {
|
||
writeError(w, http.StatusBadRequest, "user_id must be a member of this workspace")
|
||
return
|
||
}
|
||
|
||
// granted_by is the audit record of who widened access. A grant outlives the
|
||
// run that made it, so it must name the human who authorized it rather than
|
||
// the owner of the machine the request came from (MUL-7108).
|
||
grantedByUUID := actor.UserID
|
||
grantedBy := uuidToString(grantedByUUID)
|
||
|
||
if _, err := h.Queries.AddAutopilotCollaborator(r.Context(), db.AddAutopilotCollaboratorParams{
|
||
AutopilotID: ap.ID,
|
||
UserType: "member",
|
||
UserID: targetUUID,
|
||
GrantedBy: grantedByUUID,
|
||
}); err != nil {
|
||
writeError(w, http.StatusInternalServerError, "failed to grant access")
|
||
return
|
||
}
|
||
|
||
h.publish(protocol.EventAutopilotUpdated, workspaceID, "member", grantedBy, map[string]any{
|
||
"autopilot_id": uuidToString(ap.ID),
|
||
})
|
||
h.writeAutopilotCollaborators(w, r, ap.ID, http.StatusCreated)
|
||
}
|
||
|
||
// RemoveAutopilotCollaborator revokes a member's explicit write grant. Only the
|
||
// autopilot's creator or a workspace owner/admin can manage the access list; a
|
||
// collaborator cannot revoke peers. Implicit writers (creator / owner / admin)
|
||
// are unaffected — there is no row to remove.
|
||
func (h *Handler) RemoveAutopilotCollaborator(w http.ResponseWriter, r *http.Request) {
|
||
id := chi.URLParam(r, "id")
|
||
userID := chi.URLParam(r, "userId")
|
||
workspaceID := h.resolveWorkspaceID(r)
|
||
|
||
ap, ok := h.loadAutopilotInWorkspace(w, r, id, workspaceID)
|
||
if !ok {
|
||
return
|
||
}
|
||
actor, ok := h.requireAutopilotAccessManagement(w, r, ap, workspaceID)
|
||
if !ok {
|
||
return
|
||
}
|
||
targetUUID, ok := parseUUIDOrBadRequest(w, userID, "user id")
|
||
if !ok {
|
||
return
|
||
}
|
||
|
||
if err := h.Queries.DeleteAutopilotCollaborator(r.Context(), db.DeleteAutopilotCollaboratorParams{
|
||
AutopilotID: ap.ID,
|
||
UserType: "member",
|
||
UserID: targetUUID,
|
||
}); err != nil {
|
||
writeError(w, http.StatusInternalServerError, "failed to revoke access")
|
||
return
|
||
}
|
||
|
||
h.publish(protocol.EventAutopilotUpdated, workspaceID, "member", uuidToString(actor.UserID), map[string]any{
|
||
"autopilot_id": uuidToString(ap.ID),
|
||
})
|
||
h.writeAutopilotCollaborators(w, r, ap.ID, http.StatusOK)
|
||
}
|
||
|
||
// ── Trigger management ──────────────────────────────────────────────────────
|
||
|
||
func (h *Handler) CreateAutopilotTrigger(w http.ResponseWriter, r *http.Request) {
|
||
autopilotID := chi.URLParam(r, "id")
|
||
workspaceID := h.resolveWorkspaceID(r)
|
||
|
||
ap, ok := h.loadAutopilotInWorkspace(w, r, autopilotID, workspaceID)
|
||
if !ok {
|
||
return
|
||
}
|
||
actor, ok := h.requireAutopilotWrite(w, r, ap, workspaceID)
|
||
if !ok {
|
||
return
|
||
}
|
||
// A new trigger changes what / when the rule fires — a substantive publish, so
|
||
// the acting member republishes the rule version ATOMICALLY with the trigger
|
||
// create (MUL-4302 §3.4). Resolved here so both the webhook and schedule create
|
||
// paths can write the version inside the same tx as the INSERT — a failed
|
||
// version write must roll the trigger back, never leave future dispatches
|
||
// attributed to the previous publisher.
|
||
//
|
||
// This is the most consequential stamp on the surface: the trigger's
|
||
// created_by is the immutable AUTHORIZATION principal every future firing
|
||
// acts as (MUL-6951). Taking it from the acting member is what stops an agent
|
||
// from minting a standing, permanent execution right for its runtime owner
|
||
// out of a one-off request (MUL-7108).
|
||
publisherID := actor.UserID
|
||
userID := uuidToString(publisherID)
|
||
|
||
var req CreateAutopilotTriggerRequest
|
||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||
writeError(w, http.StatusBadRequest, "invalid request body")
|
||
return
|
||
}
|
||
if req.Kind == "" {
|
||
writeError(w, http.StatusBadRequest, "kind is required")
|
||
return
|
||
}
|
||
if req.Kind != "schedule" && req.Kind != "webhook" {
|
||
// "api" kind is deprecated: it was reserved-but-inert (no scheduler,
|
||
// no ingress route), and the only way to actually fire one was via
|
||
// the manual /trigger endpoint — which already works regardless of
|
||
// trigger kind. Surface stragglers with 400 so callers move to
|
||
// schedule or webhook.
|
||
writeError(w, http.StatusBadRequest, "kind must be schedule or webhook")
|
||
return
|
||
}
|
||
if req.Kind == "schedule" && (req.CronExpression == nil || *req.CronExpression == "") {
|
||
writeError(w, http.StatusBadRequest, "cron_expression is required for schedule triggers")
|
||
return
|
||
}
|
||
if req.Kind == "webhook" && req.Timezone != nil && *req.Timezone != "" {
|
||
// Webhook triggers fire on demand from external POSTs — they have no
|
||
// next_run_at to compute, so a timezone is meaningless. Reject loudly
|
||
// instead of silently dropping the field.
|
||
writeError(w, http.StatusBadRequest, "timezone is not valid for webhook triggers")
|
||
return
|
||
}
|
||
if req.Kind != "webhook" && len(req.EventFilters) > 0 {
|
||
// event_filters narrows webhook ingress — it has no meaning for a
|
||
// schedule trigger and would otherwise be silently dropped.
|
||
writeError(w, http.StatusBadRequest, "event_filters is only valid for webhook triggers")
|
||
return
|
||
}
|
||
if err := validateWebhookEventFilters(req.EventFilters); err != nil {
|
||
writeError(w, http.StatusBadRequest, err.Error())
|
||
return
|
||
}
|
||
// Provider only applies to webhook triggers and the value space is
|
||
// closed — reject unknowns early so a typo on create doesn't quietly
|
||
// degrade into a "generic" trigger that bypasses provider-specific
|
||
// dedupe / signature behaviour.
|
||
provider := "generic"
|
||
if req.Provider != nil && *req.Provider != "" {
|
||
if req.Kind != "webhook" {
|
||
writeError(w, http.StatusBadRequest, "provider is only valid for webhook triggers")
|
||
return
|
||
}
|
||
if !isAllowedWebhookProvider(*req.Provider) {
|
||
writeError(w, http.StatusBadRequest, "provider must be generic or github")
|
||
return
|
||
}
|
||
provider = *req.Provider
|
||
}
|
||
|
||
if req.Timezone != nil && *req.Timezone != "" {
|
||
if err := service.ValidateTimezone(*req.Timezone); err != nil {
|
||
writeError(w, http.StatusBadRequest, err.Error())
|
||
return
|
||
}
|
||
}
|
||
|
||
// kind-specific normalization. Webhook triggers ignore cron/timezone/
|
||
// next_run_at — they're fired on demand.
|
||
var (
|
||
nextRunAt pgtype.Timestamptz
|
||
cronText pgtype.Text
|
||
tzText pgtype.Text
|
||
webhookToken pgtype.Text
|
||
)
|
||
switch req.Kind {
|
||
case "schedule":
|
||
cronText = ptrToText(req.CronExpression)
|
||
tzText = ptrToText(req.Timezone)
|
||
tz := "UTC"
|
||
if req.Timezone != nil && *req.Timezone != "" {
|
||
tz = *req.Timezone
|
||
}
|
||
t, err := computeNextRun(*req.CronExpression, tz)
|
||
if err != nil {
|
||
writeError(w, http.StatusBadRequest, err.Error())
|
||
return
|
||
}
|
||
nextRunAt = pgtype.Timestamptz{Time: t, Valid: true}
|
||
case "webhook":
|
||
// Mint the token BEFORE the INSERT so the row never exists in a
|
||
// half-written kind=webhook + webhook_token=NULL state. If the
|
||
// random token happens to collide with an existing unique-index
|
||
// entry (vanishingly unlikely with 256 bits but the retry keeps
|
||
// the failure mode obvious if RNG is degraded), we re-generate
|
||
// and re-INSERT — never UPDATE.
|
||
eventFiltersBytes, err := encodeWebhookEventFilters(req.EventFilters)
|
||
if err != nil {
|
||
writeError(w, http.StatusInternalServerError, "failed to encode event_filters")
|
||
return
|
||
}
|
||
trigger, err := h.createWebhookTriggerWithMintedToken(r, ap, ptrToText(req.Label), provider, eventFiltersBytes, publisherID)
|
||
if err != nil {
|
||
writeError(w, http.StatusInternalServerError, "failed to create trigger")
|
||
return
|
||
}
|
||
resp := h.triggerToResponse(trigger)
|
||
h.publish(protocol.EventAutopilotUpdated, workspaceID, "member", userID, map[string]any{
|
||
"autopilot_id": uuidToString(ap.ID),
|
||
"trigger": broadcastAutopilotTriggerResponse(resp),
|
||
})
|
||
writeJSON(w, http.StatusCreated, resp)
|
||
return
|
||
}
|
||
|
||
// Schedule create: write the trigger and republish the rule version atomically.
|
||
tx, err := h.TxStarter.Begin(r.Context())
|
||
if err != nil {
|
||
writeError(w, http.StatusInternalServerError, "failed to create trigger")
|
||
return
|
||
}
|
||
defer tx.Rollback(r.Context())
|
||
qtx := h.Queries.WithTx(tx)
|
||
|
||
trigger, err := qtx.CreateAutopilotTrigger(r.Context(), db.CreateAutopilotTriggerParams{
|
||
AutopilotID: ap.ID,
|
||
Kind: req.Kind,
|
||
Enabled: true,
|
||
CronExpression: cronText,
|
||
Timezone: tzText,
|
||
NextRunAt: nextRunAt,
|
||
Label: ptrToText(req.Label),
|
||
WebhookToken: webhookToken,
|
||
// published_by records who is currently responsible for this trigger's
|
||
// CONFIG: seeded to the creator, re-stamped to whoever later substantively
|
||
// edits it (MUL-4302). Since MUL-6951 it no longer decides anything about a
|
||
// run — neither authorization nor the task's accountable human — so an edit
|
||
// moves this column alone.
|
||
PublishedByType: pgtype.Text{String: "member", Valid: publisherID.Valid},
|
||
PublishedByID: publisherID,
|
||
// created_by is the AUTHORIZATION principal and is immutable: the run acts
|
||
// as this member forever, so no edit may move it (MUL-6951).
|
||
CreatedByType: pgtype.Text{String: "member", Valid: publisherID.Valid},
|
||
CreatedByID: publisherID,
|
||
})
|
||
if err != nil {
|
||
writeError(w, http.StatusInternalServerError, "failed to create trigger")
|
||
return
|
||
}
|
||
if err := h.recordAutopilotRuleVersion(r.Context(), qtx, ap, "member", publisherID); err != nil {
|
||
writeError(w, http.StatusInternalServerError, "failed to create trigger")
|
||
return
|
||
}
|
||
if err := tx.Commit(r.Context()); err != nil {
|
||
writeError(w, http.StatusInternalServerError, "failed to create trigger")
|
||
return
|
||
}
|
||
|
||
resp := h.triggerToResponse(trigger)
|
||
h.publish(protocol.EventAutopilotUpdated, workspaceID, "member", userID, map[string]any{
|
||
"autopilot_id": uuidToString(ap.ID),
|
||
"trigger": broadcastAutopilotTriggerResponse(resp),
|
||
})
|
||
writeJSON(w, http.StatusCreated, resp)
|
||
}
|
||
|
||
// createWebhookTriggerWithMintedToken atomically creates a webhook trigger
|
||
// with a freshly minted bearer token in the same INSERT. Avoids the older
|
||
// two-step (INSERT then UPDATE webhook_token) pattern which could leave a
|
||
// kind=webhook row with NULL webhook_token visible in the UI if the second
|
||
// statement failed.
|
||
//
|
||
// Each attempt runs in its OWN transaction so the trigger INSERT and the
|
||
// rule-version republish (a webhook trigger is a substantive change to what fires,
|
||
// MUL-4302 §3.4, published by publisherID) commit together — a version-write failure
|
||
// rolls the trigger back rather than leaving future dispatches attributed to the
|
||
// previous publisher. Retries on the unique-index collision case with a fresh token
|
||
// (the collided attempt's tx is already rolled back), so a vanishingly-rare RNG
|
||
// collision turns into a clean retry rather than a 500.
|
||
func (h *Handler) createWebhookTriggerWithMintedToken(
|
||
r *http.Request,
|
||
ap db.Autopilot,
|
||
label pgtype.Text,
|
||
provider string,
|
||
eventFilters []byte,
|
||
publisherID pgtype.UUID,
|
||
) (db.AutopilotTrigger, error) {
|
||
ctx := r.Context()
|
||
for attempt := 0; attempt < 3; attempt++ {
|
||
token, err := generateWebhookToken()
|
||
if err != nil {
|
||
return db.AutopilotTrigger{}, err
|
||
}
|
||
tx, err := h.TxStarter.Begin(ctx)
|
||
if err != nil {
|
||
return db.AutopilotTrigger{}, err
|
||
}
|
||
qtx := h.Queries.WithTx(tx)
|
||
trigger, err := qtx.CreateAutopilotTrigger(ctx, db.CreateAutopilotTriggerParams{
|
||
AutopilotID: ap.ID,
|
||
Kind: "webhook",
|
||
Enabled: true,
|
||
Label: label,
|
||
WebhookToken: pgtype.Text{String: token, Valid: true},
|
||
Provider: pgtype.Text{String: provider, Valid: provider != ""},
|
||
EventFilters: eventFilters,
|
||
// published_by records CONFIG responsibility only: seeded to the creator,
|
||
// re-stamped to a later substantive editor (MUL-4302). It has no bearing
|
||
// on the runs this trigger fires (MUL-6951).
|
||
PublishedByType: pgtype.Text{String: "member", Valid: publisherID.Valid},
|
||
PublishedByID: publisherID,
|
||
// Immutable authorization principal — see the schedule path above
|
||
// (MUL-6951).
|
||
CreatedByType: pgtype.Text{String: "member", Valid: publisherID.Valid},
|
||
CreatedByID: publisherID,
|
||
})
|
||
if err != nil {
|
||
tx.Rollback(ctx)
|
||
if isUniqueViolation(err) {
|
||
continue // token collision: retry with a fresh token
|
||
}
|
||
return db.AutopilotTrigger{}, err
|
||
}
|
||
if err := h.recordAutopilotRuleVersion(ctx, qtx, ap, "member", publisherID); err != nil {
|
||
tx.Rollback(ctx)
|
||
return db.AutopilotTrigger{}, err
|
||
}
|
||
if err := tx.Commit(ctx); err != nil {
|
||
return db.AutopilotTrigger{}, err
|
||
}
|
||
return trigger, nil
|
||
}
|
||
return db.AutopilotTrigger{}, fmt.Errorf("could not mint unique webhook token")
|
||
}
|
||
|
||
func isAllowedWebhookProvider(p string) bool {
|
||
switch p {
|
||
case "generic", "github":
|
||
return true
|
||
default:
|
||
return false
|
||
}
|
||
}
|
||
|
||
func isValidAutopilotAssigneeType(t string) bool {
|
||
switch t {
|
||
case "agent", "squad":
|
||
return true
|
||
default:
|
||
return false
|
||
}
|
||
}
|
||
|
||
// validateAutopilotAssigneeForSave checks that the assignee (agent or squad)
|
||
// exists in the given workspace and, when requireRuntime is true, that its
|
||
// effective Agent has a Runtime. It takes assignment locks through q so active
|
||
// saves and the caller's Autopilot write are serialized with Runtime teardown.
|
||
//
|
||
// At dispatch time the same checks (resolveAutopilotLeader + AgentReadiness)
|
||
// run again — they live there to handle "leader was online at save time but
|
||
// went offline by trigger time". Save-time validation exists so the user gets
|
||
// immediate feedback ("bind a runtime first") instead of discovering the
|
||
// Autopilot is inert at the next schedule tick.
|
||
func (h *Handler) validateAutopilotAssigneeForSave(
|
||
w http.ResponseWriter,
|
||
r *http.Request,
|
||
q *db.Queries,
|
||
assigneeType string,
|
||
assigneeID, workspaceID pgtype.UUID,
|
||
requireRuntime bool,
|
||
) bool {
|
||
switch assigneeType {
|
||
case "agent":
|
||
agent, err := q.LockAgentForAutopilotAssignment(r.Context(), db.LockAgentForAutopilotAssignmentParams{
|
||
ID: assigneeID,
|
||
WorkspaceID: workspaceID,
|
||
})
|
||
if err != nil {
|
||
writeError(w, http.StatusBadRequest, "assignee must be a valid agent in this workspace")
|
||
return false
|
||
}
|
||
if agent.ArchivedAt.Valid {
|
||
writeError(w, http.StatusUnprocessableEntity, "assignee agent is archived; pick a different agent")
|
||
return false
|
||
}
|
||
if requireRuntime && !agent.RuntimeID.Valid {
|
||
writeError(w, http.StatusUnprocessableEntity, "assignee agent needs a runtime before this autopilot can be active")
|
||
return false
|
||
}
|
||
return true
|
||
case "squad":
|
||
squad, err := q.LockSquadForAutopilotAssignment(r.Context(), db.LockSquadForAutopilotAssignmentParams{
|
||
ID: assigneeID,
|
||
WorkspaceID: workspaceID,
|
||
})
|
||
if err != nil {
|
||
writeError(w, http.StatusBadRequest, "assignee must be a valid squad in this workspace")
|
||
return false
|
||
}
|
||
// Archived squads must be rejected at save time: the dispatcher will
|
||
// otherwise produce an unbroken stream of skipped runs against a
|
||
// squad that can never be revived without an explicit un-archive.
|
||
// Pair with TransferSquadAutopilotsToLeader on DeleteSquad so any
|
||
// autopilot that survives the archive flips to assignee_type='agent'
|
||
// (the leader) and stops referencing the dead squad row.
|
||
if squad.ArchivedAt.Valid {
|
||
writeError(w, http.StatusUnprocessableEntity, "squad is archived; pick a different squad")
|
||
return false
|
||
}
|
||
leader, err := q.LockAgentForAutopilotAssignment(r.Context(), db.LockAgentForAutopilotAssignmentParams{
|
||
ID: squad.LeaderID,
|
||
WorkspaceID: workspaceID,
|
||
})
|
||
if err != nil {
|
||
writeError(w, http.StatusBadRequest, "squad leader agent not found")
|
||
return false
|
||
}
|
||
if leader.ArchivedAt.Valid {
|
||
writeError(w, http.StatusUnprocessableEntity, "squad leader is archived; pick a different squad or rotate the leader before assigning autopilot")
|
||
return false
|
||
}
|
||
if requireRuntime && !leader.RuntimeID.Valid {
|
||
writeError(w, http.StatusUnprocessableEntity, "squad leader needs a runtime before this autopilot can be active")
|
||
return false
|
||
}
|
||
// Private-leader gate: the member configuring the autopilot must have
|
||
// access to the private leader, same as validateAssigneePair.
|
||
actorType, actorID := h.resolveActor(r, requestUserID(r), util.UUIDToString(workspaceID))
|
||
if !h.canInvokeAgent(r.Context(), leader, actorType, actorID, h.invokeOriginatorFromRequest(r, actorType, actorID), util.UUIDToString(workspaceID)) {
|
||
writeError(w, http.StatusForbidden, "cannot assign autopilot to squad with private leader")
|
||
return false
|
||
}
|
||
return true
|
||
default:
|
||
writeError(w, http.StatusBadRequest, "assignee_type must be agent or squad")
|
||
return false
|
||
}
|
||
}
|
||
|
||
func (h *Handler) UpdateAutopilotTrigger(w http.ResponseWriter, r *http.Request) {
|
||
autopilotID := chi.URLParam(r, "id")
|
||
triggerID := chi.URLParam(r, "triggerId")
|
||
workspaceID := h.resolveWorkspaceID(r)
|
||
|
||
ap, ok := h.loadAutopilotInWorkspace(w, r, autopilotID, workspaceID)
|
||
if !ok {
|
||
return
|
||
}
|
||
actor, ok := h.requireAutopilotWrite(w, r, ap, workspaceID)
|
||
if !ok {
|
||
return
|
||
}
|
||
|
||
triggerUUID, ok := parseUUIDOrBadRequest(w, triggerID, "trigger id")
|
||
if !ok {
|
||
return
|
||
}
|
||
|
||
prev, err := h.Queries.GetAutopilotTrigger(r.Context(), triggerUUID)
|
||
if err != nil || uuidToString(prev.AutopilotID) != uuidToString(ap.ID) {
|
||
writeError(w, http.StatusNotFound, "trigger not found")
|
||
return
|
||
}
|
||
|
||
var req UpdateAutopilotTriggerRequest
|
||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||
writeError(w, http.StatusBadRequest, "invalid request body")
|
||
return
|
||
}
|
||
|
||
// Kind-specific validation. Mirrors the create-path discipline: cron
|
||
// and timezone only make sense on schedule triggers, so reject loudly
|
||
// rather than persisting fields that no code path reads. enabled and
|
||
// label remain valid on every kind.
|
||
if prev.Kind != "schedule" {
|
||
if req.CronExpression != nil {
|
||
writeError(w, http.StatusBadRequest, "cron_expression is only valid for schedule triggers")
|
||
return
|
||
}
|
||
if req.Timezone != nil {
|
||
writeError(w, http.StatusBadRequest, "timezone is only valid for schedule triggers")
|
||
return
|
||
}
|
||
}
|
||
|
||
params := db.UpdateAutopilotTriggerParams{
|
||
ID: prev.ID,
|
||
CronExpression: prev.CronExpression,
|
||
Timezone: prev.Timezone,
|
||
NextRunAt: prev.NextRunAt,
|
||
Label: prev.Label,
|
||
}
|
||
if req.Enabled != nil {
|
||
params.Enabled = pgtype.Bool{Bool: *req.Enabled, Valid: true}
|
||
}
|
||
if req.CronExpression != nil {
|
||
params.CronExpression = pgtype.Text{String: *req.CronExpression, Valid: true}
|
||
}
|
||
if req.Timezone != nil {
|
||
if *req.Timezone != "" {
|
||
if err := service.ValidateTimezone(*req.Timezone); err != nil {
|
||
writeError(w, http.StatusBadRequest, err.Error())
|
||
return
|
||
}
|
||
}
|
||
params.Timezone = pgtype.Text{String: *req.Timezone, Valid: true}
|
||
}
|
||
if req.Label != nil {
|
||
params.Label = pgtype.Text{String: *req.Label, Valid: true}
|
||
}
|
||
// Tri-state PATCH for event_filters. A nil pointer (field omitted or
|
||
// JSON null) leaves the existing row untouched — params.EventFilters
|
||
// stays unset and the COALESCE in the UPDATE preserves the previous
|
||
// value. A non-nil pointer is authoritative: an empty slice clears
|
||
// filters (encoded as the JSONB literal `[]` so COALESCE replaces
|
||
// rather than preserves), a populated slice replaces.
|
||
if req.EventFilters != nil {
|
||
if prev.Kind != "webhook" {
|
||
writeError(w, http.StatusBadRequest, "event_filters is only valid for webhook triggers")
|
||
return
|
||
}
|
||
if err := validateWebhookEventFilters(*req.EventFilters); err != nil {
|
||
writeError(w, http.StatusBadRequest, err.Error())
|
||
return
|
||
}
|
||
encoded, err := encodeWebhookEventFiltersAlways(*req.EventFilters)
|
||
if err != nil {
|
||
writeError(w, http.StatusInternalServerError, "failed to encode event_filters")
|
||
return
|
||
}
|
||
params.EventFilters = encoded
|
||
}
|
||
|
||
// Recompute next_run_at if cron or timezone changed.
|
||
cronExpr := prev.CronExpression.String
|
||
if req.CronExpression != nil {
|
||
cronExpr = *req.CronExpression
|
||
}
|
||
tz := "UTC"
|
||
if prev.Timezone.Valid {
|
||
tz = prev.Timezone.String
|
||
}
|
||
if req.Timezone != nil {
|
||
tz = *req.Timezone
|
||
}
|
||
if prev.Kind == "schedule" && cronExpr != "" {
|
||
t, err := computeNextRun(cronExpr, tz)
|
||
if err != nil {
|
||
writeError(w, http.StatusBadRequest, err.Error())
|
||
return
|
||
}
|
||
params.NextRunAt = pgtype.Timestamptz{Time: t, Valid: true}
|
||
}
|
||
|
||
userID := uuidToString(actor.UserID)
|
||
|
||
tx, err := h.TxStarter.Begin(r.Context())
|
||
if err != nil {
|
||
writeError(w, http.StatusInternalServerError, "failed to update trigger")
|
||
return
|
||
}
|
||
defer tx.Rollback(r.Context())
|
||
qtx := h.Queries.WithTx(tx)
|
||
|
||
trigger, err := qtx.UpdateAutopilotTrigger(r.Context(), params)
|
||
if err != nil {
|
||
writeError(w, http.StatusInternalServerError, "failed to update trigger")
|
||
return
|
||
}
|
||
|
||
// Only a substantive edit republishes the rule version and transfers this
|
||
// trigger's accountability to the editor. cron / timezone / enabled / event_filters
|
||
// change WHAT or WHEN the trigger fires; label is a cosmetic display field, and a
|
||
// no-op PATCH changes nothing — neither should move responsibility (MUL-4302; the
|
||
// over-transfer Elon flagged). Comparing the persisted before/after rows captures a
|
||
// real change and ignores label-only / no-op PATCHes (next_run_at is derived from
|
||
// cron/timezone, so it is not an independent signal).
|
||
triggerSubstantiveChange := prev.Enabled != trigger.Enabled ||
|
||
prev.CronExpression != trigger.CronExpression ||
|
||
prev.Timezone != trigger.Timezone ||
|
||
!bytes.Equal(prev.EventFilters, trigger.EventFilters)
|
||
if triggerSubstantiveChange {
|
||
if err := h.recordAutopilotRuleVersion(r.Context(), qtx, ap, "member", actor.UserID); err != nil {
|
||
writeError(w, http.StatusInternalServerError, "failed to update trigger")
|
||
return
|
||
}
|
||
// Responsibility for THIS trigger's runs transfers to the editor. Scoped to the
|
||
// single row so editing one trigger never reassigns another's accountability —
|
||
// the per-firing-trigger granularity the autopilot-scoped rule_version can't give.
|
||
if err := qtx.SetAutopilotTriggerPublisher(r.Context(), db.SetAutopilotTriggerPublisherParams{
|
||
ID: trigger.ID,
|
||
PublishedByType: pgtype.Text{String: "member", Valid: true},
|
||
PublishedByID: actor.UserID,
|
||
}); err != nil {
|
||
writeError(w, http.StatusInternalServerError, "failed to update trigger")
|
||
return
|
||
}
|
||
}
|
||
if err := tx.Commit(r.Context()); err != nil {
|
||
writeError(w, http.StatusInternalServerError, "failed to update trigger")
|
||
return
|
||
}
|
||
|
||
resp := h.triggerToResponse(trigger)
|
||
h.publish(protocol.EventAutopilotUpdated, workspaceID, "member", userID, map[string]any{
|
||
"autopilot_id": uuidToString(ap.ID),
|
||
"trigger": broadcastAutopilotTriggerResponse(resp),
|
||
})
|
||
writeJSON(w, http.StatusOK, resp)
|
||
}
|
||
|
||
func (h *Handler) DeleteAutopilotTrigger(w http.ResponseWriter, r *http.Request) {
|
||
autopilotID := chi.URLParam(r, "id")
|
||
triggerID := chi.URLParam(r, "triggerId")
|
||
workspaceID := h.resolveWorkspaceID(r)
|
||
|
||
autopilotUUID, ok := parseUUIDOrBadRequest(w, autopilotID, "autopilot id")
|
||
if !ok {
|
||
return
|
||
}
|
||
triggerUUID, ok := parseUUIDOrBadRequest(w, triggerID, "trigger id")
|
||
if !ok {
|
||
return
|
||
}
|
||
wsUUID, ok := parseUUIDOrBadRequest(w, workspaceID, "workspace id")
|
||
if !ok {
|
||
return
|
||
}
|
||
|
||
ap, err := h.Queries.GetAutopilotInWorkspace(r.Context(), db.GetAutopilotInWorkspaceParams{
|
||
ID: autopilotUUID,
|
||
WorkspaceID: wsUUID,
|
||
})
|
||
if err != nil {
|
||
writeError(w, http.StatusNotFound, "autopilot not found")
|
||
return
|
||
}
|
||
actor, ok := h.requireAutopilotWrite(w, r, ap, workspaceID)
|
||
if !ok {
|
||
return
|
||
}
|
||
|
||
trigger, err := h.Queries.GetAutopilotTrigger(r.Context(), triggerUUID)
|
||
if err != nil || uuidToString(trigger.AutopilotID) != uuidToString(autopilotUUID) {
|
||
writeError(w, http.StatusNotFound, "trigger not found")
|
||
return
|
||
}
|
||
|
||
userID := uuidToString(actor.UserID)
|
||
|
||
// Removing a trigger changes what fires — a substantive publish (MUL-4302 §3.4).
|
||
// Republish the rule version with this member as publisher, atomically with the
|
||
// delete.
|
||
tx, err := h.TxStarter.Begin(r.Context())
|
||
if err != nil {
|
||
writeError(w, http.StatusInternalServerError, "failed to delete trigger")
|
||
return
|
||
}
|
||
defer tx.Rollback(r.Context())
|
||
qtx := h.Queries.WithTx(tx)
|
||
|
||
if err := qtx.DeleteAutopilotTrigger(r.Context(), triggerUUID); err != nil {
|
||
writeError(w, http.StatusInternalServerError, "failed to delete trigger")
|
||
return
|
||
}
|
||
if err := h.recordAutopilotRuleVersion(r.Context(), qtx, ap, "member", actor.UserID); err != nil {
|
||
writeError(w, http.StatusInternalServerError, "failed to delete trigger")
|
||
return
|
||
}
|
||
if err := tx.Commit(r.Context()); err != nil {
|
||
writeError(w, http.StatusInternalServerError, "failed to delete trigger")
|
||
return
|
||
}
|
||
|
||
h.publish(protocol.EventAutopilotUpdated, workspaceID, "member", userID, map[string]any{
|
||
"autopilot_id": uuidToString(autopilotUUID),
|
||
"trigger_id": uuidToString(triggerUUID),
|
||
})
|
||
w.WriteHeader(http.StatusNoContent)
|
||
}
|
||
|
||
// RotateAutopilotTriggerWebhookToken issues a fresh bearer token for an
|
||
// existing webhook trigger. The old token stops working immediately because
|
||
// the unique-index lookup in the public ingress route is keyed on the
|
||
// current row value.
|
||
func (h *Handler) RotateAutopilotTriggerWebhookToken(w http.ResponseWriter, r *http.Request) {
|
||
autopilotID := chi.URLParam(r, "id")
|
||
triggerID := chi.URLParam(r, "triggerId")
|
||
workspaceID := h.resolveWorkspaceID(r)
|
||
|
||
ap, ok := h.loadAutopilotInWorkspace(w, r, autopilotID, workspaceID)
|
||
if !ok {
|
||
return
|
||
}
|
||
actor, ok := h.requireAutopilotWrite(w, r, ap, workspaceID)
|
||
if !ok {
|
||
return
|
||
}
|
||
|
||
triggerUUID, ok := parseUUIDOrBadRequest(w, triggerID, "trigger id")
|
||
if !ok {
|
||
return
|
||
}
|
||
prev, err := h.Queries.GetAutopilotTrigger(r.Context(), triggerUUID)
|
||
if err != nil || uuidToString(prev.AutopilotID) != uuidToString(ap.ID) {
|
||
writeError(w, http.StatusNotFound, "trigger not found")
|
||
return
|
||
}
|
||
if prev.Kind != "webhook" {
|
||
writeError(w, http.StatusBadRequest, "trigger is not a webhook trigger")
|
||
return
|
||
}
|
||
|
||
var rotated db.AutopilotTrigger
|
||
for attempt := 0; attempt < 3; attempt++ {
|
||
token, terr := generateWebhookToken()
|
||
if terr != nil {
|
||
writeError(w, http.StatusInternalServerError, "failed to generate webhook token")
|
||
return
|
||
}
|
||
rotated, err = h.Queries.RotateAutopilotTriggerWebhookToken(r.Context(), db.RotateAutopilotTriggerWebhookTokenParams{
|
||
ID: triggerUUID,
|
||
WebhookToken: pgtype.Text{String: token, Valid: true},
|
||
})
|
||
if err == nil {
|
||
break
|
||
}
|
||
if !isUniqueViolation(err) {
|
||
writeError(w, http.StatusInternalServerError, "failed to rotate webhook token")
|
||
return
|
||
}
|
||
}
|
||
if err != nil {
|
||
writeError(w, http.StatusInternalServerError, "failed to rotate webhook token")
|
||
return
|
||
}
|
||
|
||
resp := h.triggerToResponse(rotated)
|
||
h.publish(protocol.EventAutopilotUpdated, workspaceID, "member", uuidToString(actor.UserID), map[string]any{
|
||
"autopilot_id": uuidToString(ap.ID),
|
||
"trigger": broadcastAutopilotTriggerResponse(resp),
|
||
})
|
||
writeJSON(w, http.StatusOK, resp)
|
||
}
|
||
|
||
// SetAutopilotTriggerSigningSecret sets (or clears) the HMAC signing secret
|
||
// for a webhook trigger. Lives on its own endpoint so the secret value never
|
||
// shares a request body with any other field — keeping it out of generic
|
||
// request-body logs and audit captures that may include patch payloads.
|
||
//
|
||
// Empty body / empty `signing_secret` clears the secret and reverts the
|
||
// trigger to bearer-token-only authentication. The response carries
|
||
// `has_signing_secret` + `signing_secret_hint`; the secret itself is never
|
||
// echoed back, matching the GitHub / Stripe industry pattern.
|
||
func (h *Handler) SetAutopilotTriggerSigningSecret(w http.ResponseWriter, r *http.Request) {
|
||
autopilotID := chi.URLParam(r, "id")
|
||
triggerID := chi.URLParam(r, "triggerId")
|
||
workspaceID := h.resolveWorkspaceID(r)
|
||
|
||
ap, ok := h.loadAutopilotInWorkspace(w, r, autopilotID, workspaceID)
|
||
if !ok {
|
||
return
|
||
}
|
||
actor, ok := h.requireAutopilotWrite(w, r, ap, workspaceID)
|
||
if !ok {
|
||
return
|
||
}
|
||
triggerUUID, ok := parseUUIDOrBadRequest(w, triggerID, "trigger id")
|
||
if !ok {
|
||
return
|
||
}
|
||
prev, err := h.Queries.GetAutopilotTrigger(r.Context(), triggerUUID)
|
||
if err != nil || uuidToString(prev.AutopilotID) != uuidToString(ap.ID) {
|
||
writeError(w, http.StatusNotFound, "trigger not found")
|
||
return
|
||
}
|
||
if prev.Kind != "webhook" {
|
||
writeError(w, http.StatusBadRequest, "trigger is not a webhook trigger")
|
||
return
|
||
}
|
||
|
||
var req SetSigningSecretRequest
|
||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||
writeError(w, http.StatusBadRequest, "invalid request body")
|
||
return
|
||
}
|
||
secret := strings.TrimSpace(req.SigningSecret)
|
||
// 16 chars is the floor: enough to make brute force impractical for the
|
||
// SHA-256 HMAC but low enough not to reject providers that mint shorter
|
||
// keys (Slack signing secrets are 32 hex chars; GitHub recommends 32).
|
||
if secret != "" && len(secret) < 16 {
|
||
writeError(w, http.StatusBadRequest, "signing_secret must be at least 16 characters")
|
||
return
|
||
}
|
||
|
||
param := db.SetAutopilotTriggerSigningSecretParams{ID: triggerUUID}
|
||
if secret != "" {
|
||
param.SigningSecret = pgtype.Text{String: secret, Valid: true}
|
||
}
|
||
updated, err := h.Queries.SetAutopilotTriggerSigningSecret(r.Context(), param)
|
||
if err != nil {
|
||
writeError(w, http.StatusInternalServerError, "failed to update signing secret")
|
||
return
|
||
}
|
||
|
||
resp := h.triggerToResponse(updated)
|
||
// Publish the trigger update so the UI can refresh the has_signing_secret
|
||
// badge in real time. The event payload only carries the response shape,
|
||
// which excludes the signing secret and, on the broadcast copy, the webhook
|
||
// credential as well.
|
||
h.publish(protocol.EventAutopilotUpdated, workspaceID, "member", uuidToString(actor.UserID), map[string]any{
|
||
"autopilot_id": uuidToString(ap.ID),
|
||
"trigger": broadcastAutopilotTriggerResponse(resp),
|
||
})
|
||
writeJSON(w, http.StatusOK, resp)
|
||
}
|
||
|
||
// ── Runs ────────────────────────────────────────────────────────────────────
|
||
|
||
func (h *Handler) ListAutopilotRuns(w http.ResponseWriter, r *http.Request) {
|
||
autopilotID := chi.URLParam(r, "id")
|
||
workspaceID := h.resolveWorkspaceID(r)
|
||
|
||
autopilot, ok := h.loadAutopilotInWorkspace(w, r, autopilotID, workspaceID)
|
||
if !ok {
|
||
return
|
||
}
|
||
|
||
limit := int32(20)
|
||
offset := int32(0)
|
||
if l := r.URL.Query().Get("limit"); l != "" {
|
||
if v, err := strconv.Atoi(l); err == nil && v > 0 {
|
||
limit = int32(v)
|
||
}
|
||
}
|
||
if limit > 100 {
|
||
limit = 100
|
||
}
|
||
if o := r.URL.Query().Get("offset"); o != "" {
|
||
if v, err := strconv.Atoi(o); err == nil && v >= 0 {
|
||
offset = int32(v)
|
||
}
|
||
}
|
||
|
||
runs, err := h.Queries.ListAutopilotRuns(r.Context(), db.ListAutopilotRunsParams{
|
||
AutopilotID: autopilot.ID,
|
||
Limit: limit,
|
||
Offset: offset,
|
||
})
|
||
if err != nil {
|
||
writeError(w, http.StatusInternalServerError, "failed to list runs")
|
||
return
|
||
}
|
||
|
||
resp := make([]AutopilotRunResponse, len(runs))
|
||
for i, run := range runs {
|
||
// Omit trigger_payload in the list response — a webhook envelope
|
||
// can be up to 256 KiB and `limit` defaults to 20, so the full
|
||
// list would be a ~5 MB worst case. Detail dialog fetches the
|
||
// full payload from GetAutopilotRun.
|
||
resp[i] = runToResponseSlim(run)
|
||
}
|
||
writeJSON(w, http.StatusOK, map[string]any{"runs": resp, "total": len(resp)})
|
||
}
|
||
|
||
// GetAutopilotRun returns a single run including its full trigger_payload.
|
||
// Workspace scoping is enforced via loadAutopilotInWorkspace; the run is
|
||
// then re-checked to belong to that autopilot so a guessed runId from
|
||
// another workspace cannot leak data.
|
||
func (h *Handler) GetAutopilotRun(w http.ResponseWriter, r *http.Request) {
|
||
autopilotID := chi.URLParam(r, "id")
|
||
runID := chi.URLParam(r, "runId")
|
||
workspaceID := h.resolveWorkspaceID(r)
|
||
|
||
autopilot, ok := h.loadAutopilotInWorkspace(w, r, autopilotID, workspaceID)
|
||
if !ok {
|
||
return
|
||
}
|
||
|
||
runUUID, ok := parseUUIDOrBadRequest(w, runID, "run id")
|
||
if !ok {
|
||
return
|
||
}
|
||
|
||
run, err := h.Queries.GetAutopilotRun(r.Context(), runUUID)
|
||
if err != nil {
|
||
writeError(w, http.StatusNotFound, "run not found")
|
||
return
|
||
}
|
||
if uuidToString(run.AutopilotID) != uuidToString(autopilot.ID) {
|
||
// Guard against a runId from another autopilot being requested via
|
||
// this autopilot's URL — fail closed with 404 so the response shape
|
||
// matches the "not found" case and no information is leaked.
|
||
writeError(w, http.StatusNotFound, "run not found")
|
||
return
|
||
}
|
||
|
||
writeJSON(w, http.StatusOK, runToResponse(run))
|
||
}
|
||
|
||
// ── Manual trigger ──────────────────────────────────────────────────────────
|
||
|
||
// requireAutopilotTriggerInvoker resolves the human whose authority a manual
|
||
// "run now" spends and enforces that they may spend it. It is this endpoint's
|
||
// ONLY write gate — requireAutopilotWrite is deliberately not called alongside
|
||
// it, since both now judge the same human and requiring the grant twice would
|
||
// mean requiring it of two unrelated people (#8099, and see
|
||
// requireAutopilotActingMember for why the runtime owner is not one of them).
|
||
// On refusal it writes 403 and returns false; the caller must return early.
|
||
//
|
||
// The refusals keep their own codes: FormatError collapses every other 403 into
|
||
// generic "no access" copy, and these two are the ones an operator can act on —
|
||
// the CLI branches on them by code, never on the English sentence (#8078).
|
||
func (h *Handler) requireAutopilotTriggerInvoker(w http.ResponseWriter, r *http.Request, ap db.Autopilot, workspaceID string) (pgtype.UUID, bool) {
|
||
member, ok := h.requireAutopilotActingMember(w, r, workspaceID, autopilotTriggerRefusal)
|
||
if !ok {
|
||
return pgtype.UUID{}, false
|
||
}
|
||
if !h.memberCanWriteAutopilot(r.Context(), ap, member) {
|
||
// Deliberately the same sentence and code whether the ordering human is
|
||
// a non-member or simply lacks the grant: a caller must not be able to
|
||
// probe workspace membership through this endpoint.
|
||
writeErrorCode(w, http.StatusForbidden, autopilotTriggerRefusal.forbiddenCode, autopilotTriggerRefusal.forbiddenMsg)
|
||
return pgtype.UUID{}, false
|
||
}
|
||
return member.UserID, true
|
||
}
|
||
|
||
func (h *Handler) TriggerAutopilot(w http.ResponseWriter, r *http.Request) {
|
||
id := chi.URLParam(r, "id")
|
||
workspaceID := h.resolveWorkspaceID(r)
|
||
|
||
autopilot, ok := h.loadAutopilotInWorkspace(w, r, id, workspaceID)
|
||
if !ok {
|
||
return
|
||
}
|
||
// A manual "run now" is a direct human action, so the run is attributed
|
||
// direct_human to the human who authorized it (MUL-4302 §4) — the human this
|
||
// caller acts for, see requireAutopilotTriggerInvoker, which is this
|
||
// endpoint's only write gate. Authorization runs before the status check so an
|
||
// unauthorized caller cannot tell an inactive autopilot from an active one.
|
||
invokerUserID, ok := h.requireAutopilotTriggerInvoker(w, r, autopilot, workspaceID)
|
||
if !ok {
|
||
return
|
||
}
|
||
if autopilot.Status != "active" {
|
||
writeError(w, http.StatusBadRequest, "autopilot is not active")
|
||
return
|
||
}
|
||
|
||
idempotencyKey := strings.TrimSpace(r.Header.Get("Idempotency-Key"))
|
||
if len(idempotencyKey) > 255 {
|
||
writeError(w, http.StatusBadRequest, "Idempotency-Key is too long")
|
||
return
|
||
}
|
||
if idempotencyKey == "" {
|
||
idempotencyKey = service.NewRequestIdempotencyKey()
|
||
}
|
||
run, reasonCode, err := h.AutopilotService.DispatchAutopilotManualWithKey(r.Context(), autopilot, pgtype.UUID{}, nil, invokerUserID, idempotencyKey)
|
||
if err != nil {
|
||
var quotaErr *service.AutopilotQuotaExceededError
|
||
if errors.As(err, "aErr) {
|
||
retryAfter := int64(time.Until(quotaErr.ResetAt).Seconds())
|
||
if retryAfter < 1 {
|
||
retryAfter = 1
|
||
}
|
||
w.Header().Set("Retry-After", strconv.FormatInt(retryAfter, 10))
|
||
writeJSON(w, http.StatusTooManyRequests, map[string]any{
|
||
"reason_code": "quota_exceeded", "used": quotaErr.Used,
|
||
"reserved": quotaErr.Reserved, "limit": quotaErr.Limit,
|
||
"reset_at": quotaErr.ResetAt.UTC().Format(time.RFC3339),
|
||
})
|
||
return
|
||
}
|
||
// Everything past the quota branch is an unclassified internal failure
|
||
// whose chain carries pgx constraint/table names and internal ids. Any
|
||
// workspace member can reach "run now", so the detail stays in the log
|
||
// and the response is the same fixed 5xx string the rest of this file
|
||
// returns (MUL-6472).
|
||
slog.Error("trigger autopilot failed",
|
||
"error", err,
|
||
"autopilot_id", uuidToString(autopilot.ID),
|
||
)
|
||
writeError(w, http.StatusInternalServerError, "failed to trigger autopilot")
|
||
return
|
||
}
|
||
|
||
// Carry the typed admission reason (decided at its source, MUL-4525) straight
|
||
// into the response — no reverse-engineering from failure_reason text. The
|
||
// UI branches on run status + this code for the "run now" toast.
|
||
resp := runToResponse(*run)
|
||
if reasonCode != "" {
|
||
c := string(reasonCode)
|
||
resp.ReasonCode = &c
|
||
}
|
||
writeJSON(w, http.StatusOK, resp)
|
||
}
|
||
|
||
// GetAutopilotQuotaUsage exposes Cloud-provided interval facts plus durable
|
||
// server-owned blocked counts. When the gate is off or malformed, the service
|
||
// returns before any quota-table read.
|
||
func (h *Handler) GetAutopilotQuotaUsage(w http.ResponseWriter, r *http.Request) {
|
||
workspaceID, err := util.ParseUUID(h.resolveWorkspaceID(r))
|
||
if err != nil {
|
||
writeError(w, http.StatusBadRequest, "invalid workspace")
|
||
return
|
||
}
|
||
usage, err := h.AutopilotService.AutopilotQuotaUsage(r.Context(), workspaceID)
|
||
if err != nil {
|
||
writeError(w, http.StatusInternalServerError, "failed to load autopilot quota usage")
|
||
return
|
||
}
|
||
resp := AutopilotQuotaUsageResponse{Action: "off"}
|
||
if usage.Enabled {
|
||
resp.Action = usage.Action
|
||
resp.Used, resp.Reserved, resp.Total = usage.Used, usage.Reserved, usage.Total
|
||
resp.Limit, resp.Reached = usage.Limit, usage.Reached
|
||
resp.BlockedCounts = usage.BlockedCounts
|
||
if usage.PeriodStart != nil {
|
||
v := usage.PeriodStart.UTC().Format(time.RFC3339)
|
||
resp.PeriodStart = &v
|
||
}
|
||
if usage.PeriodEnd != nil {
|
||
v := usage.PeriodEnd.UTC().Format(time.RFC3339)
|
||
resp.PeriodEnd = &v
|
||
}
|
||
if usage.ResetAt != nil {
|
||
v := usage.ResetAt.UTC().Format(time.RFC3339)
|
||
resp.ResetAt = &v
|
||
}
|
||
}
|
||
writeJSON(w, http.StatusOK, resp)
|
||
}
|