From bebb403f211aae38ed40787d5aa2057c9bbb1779 Mon Sep 17 00:00:00 2001 From: J Date: Thu, 9 Jul 2026 23:39:21 +0800 Subject: [PATCH] fix(attribution): make trigger-create rule-version republish atomic (MUL-4302) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses Elon's final Phase 1 blocking finding: CreateAutopilotTrigger recorded the rule-version republish best-effort AFTER the trigger insert. If member B added a schedule/webhook trigger to member A's autopilot and the version write failed, future schedule/webhook dispatches would keep attributing to A — violating the rule_owner invariant that the last member to substantively change the rule owns future runs ("no immediate daemon claim" doesn't save it, since the miss surfaces at the LATER trigger firing). Both create paths now write the version in the SAME tx as the trigger INSERT: - schedule create: wrap CreateAutopilotTrigger + recordAutopilotRuleVersion in one tx. - webhook create: each mint-with-retry attempt runs in its own tx (insert + version commit together; a token collision rolls that attempt back and retries with a fresh token; a version-write failure rolls the trigger back). Passes ap + the acting member id into the helper. - removed the best-effort recordTriggerRuleVersionBestEffort helper (and the now-unused slog import). Test: TestCreateTrigger_RepublishesRuleVersionAtomically drives both create paths through the handler and asserts a rule version is published by the acting member. Existing webhook/trigger/archive handler tests still pass. Also merges origin/main (unrelated avatar feature #5074). Full service/attribution/handler/migration/ scheduler/cmd suites pass on a DB migrated through 161; build/vet/gofmt clean. Co-authored-by: multica-agent --- server/internal/handler/autopilot.go | 94 ++++++++++++------- .../handler/autopilot_webhook_handler_test.go | 58 +++++++++++- 2 files changed, 114 insertions(+), 38 deletions(-) diff --git a/server/internal/handler/autopilot.go b/server/internal/handler/autopilot.go index d8c8c4d678..852955b01a 100644 --- a/server/internal/handler/autopilot.go +++ b/server/internal/handler/autopilot.go @@ -5,7 +5,6 @@ import ( "encoding/json" "fmt" "io" - "log/slog" "net/http" "strconv" "strings" @@ -1174,6 +1173,17 @@ func (h *Handler) CreateAutopilotTrigger(w http.ResponseWriter, r *http.Request) if !h.requireAutopilotWrite(w, r, ap, workspaceID) { 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. + userID, ok := requireUserID(w, r) + if !ok { + return + } + publisherID := parseUUID(userID) var req CreateAutopilotTriggerRequest if err := json.NewDecoder(r.Body).Decode(&req); err != nil { @@ -1272,14 +1282,12 @@ func (h *Handler) CreateAutopilotTrigger(w http.ResponseWriter, r *http.Request) writeError(w, http.StatusInternalServerError, "failed to encode event_filters") return } - trigger, err := h.createWebhookTriggerWithMintedToken(r, ap.ID, ptrToText(req.Label), provider, eventFiltersBytes) + 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) - userID, _ := requireUserID(w, r) - h.recordTriggerRuleVersionBestEffort(r.Context(), ap, userID) h.publish(protocol.EventAutopilotUpdated, workspaceID, "member", userID, map[string]any{ "autopilot_id": uuidToString(ap.ID), "trigger": resp, @@ -1288,7 +1296,16 @@ func (h *Handler) CreateAutopilotTrigger(w http.ResponseWriter, r *http.Request) return } - trigger, err := h.Queries.CreateAutopilotTrigger(r.Context(), db.CreateAutopilotTriggerParams{ + // 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, @@ -1302,10 +1319,16 @@ func (h *Handler) CreateAutopilotTrigger(w http.ResponseWriter, r *http.Request) 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) - userID, _ := requireUserID(w, r) - h.recordTriggerRuleVersionBestEffort(r.Context(), ap, userID) h.publish(protocol.EventAutopilotUpdated, workspaceID, "member", userID, map[string]any{ "autopilot_id": uuidToString(ap.ID), "trigger": resp, @@ -1313,48 +1336,40 @@ func (h *Handler) CreateAutopilotTrigger(w http.ResponseWriter, r *http.Request) writeJSON(w, http.StatusCreated, resp) } -// recordTriggerRuleVersionBestEffort republishes an autopilot's rule version for a -// trigger CREATE (a substantive change to what fires, MUL-4302 §3.4). Unlike the -// trigger update/delete paths — which wrap the mutation and the version write in one -// tx — create is best-effort: the webhook create path mints its token with a retry -// loop that cannot share a single tx, and a create is often part of initial setup -// (already covered by the autopilot-create v1). A failed version write here is -// benign: the autopilot's active version stays the current publisher, the new -// trigger still fires under it, and no immediate daemon claim rides this (unlike the -// manual-rerun lineage race). Only a member actor publishes; an unresolved user is a -// no-op rather than a bogus 'system' publish. -func (h *Handler) recordTriggerRuleVersionBestEffort(ctx context.Context, ap db.Autopilot, userID string) { - uid, err := util.ParseUUID(userID) - if err != nil { - return - } - if err := h.recordAutopilotRuleVersion(ctx, h.Queries, ap, "member", uid); err != nil { - slog.Warn("create trigger: record rule version failed", "autopilot_id", uuidToString(ap.ID), "error", err) - } -} - // 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. // -// Retries on the unique-index collision case so a vanishingly-rare RNG +// 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, - autopilotID pgtype.UUID, + 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 } - trigger, err := h.Queries.CreateAutopilotTrigger(r.Context(), db.CreateAutopilotTriggerParams{ - AutopilotID: autopilotID, + 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, @@ -1362,12 +1377,21 @@ func (h *Handler) createWebhookTriggerWithMintedToken( Provider: pgtype.Text{String: provider, Valid: provider != ""}, EventFilters: eventFilters, }) - if err == nil { - return trigger, nil - } - if !isUniqueViolation(err) { + 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") } diff --git a/server/internal/handler/autopilot_webhook_handler_test.go b/server/internal/handler/autopilot_webhook_handler_test.go index 32d790f8d0..fc48d16edc 100644 --- a/server/internal/handler/autopilot_webhook_handler_test.go +++ b/server/internal/handler/autopilot_webhook_handler_test.go @@ -76,6 +76,58 @@ func createWebhookTriggerViaHandler(t *testing.T, autopilotID string) AutopilotT return resp } +// TestCreateTrigger_RepublishesRuleVersionAtomically verifies Elon's final Phase 1 +// must-fix: creating a trigger (a substantive change to what fires, MUL-4302 §3.4) +// republishes the autopilot's rule version with the acting member as publisher, +// written atomically in the same tx as the trigger INSERT — for BOTH the webhook +// create path (mint-with-retry, whole attempt wrapped in a tx) and the schedule path. +func TestCreateTrigger_RepublishesRuleVersionAtomically(t *testing.T) { + agentID := createWebhookTestAgent(t, "TriggerVersion Agent") + apID := createWebhookTestAutopilot(t, agentID, "active", "run_only") + ctx := context.Background() + verParams := db.GetActiveAutopilotRuleVersionParams{ + WorkspaceID: parseUUID(testWorkspaceID), + AutopilotID: parseUUID(apID), + } + + // The test autopilot is inserted directly (no v1), so no rule version exists yet. + if _, err := testHandler.Queries.GetActiveAutopilotRuleVersion(ctx, verParams); err == nil { + t.Fatal("expected no rule version before any trigger is created") + } + + // Webhook create (mint-with-retry) republishes atomically. + createWebhookTriggerViaHandler(t, apID) + ver, err := testHandler.Queries.GetActiveAutopilotRuleVersion(ctx, verParams) + if err != nil { + t.Fatalf("webhook trigger create must republish a rule version: %v", err) + } + if ver.PublishedByType != "member" || uuidToString(ver.PublishedByID) != testUserID { + t.Errorf("webhook version published_by = %s/%s, want member/%s", ver.PublishedByType, uuidToString(ver.PublishedByID), testUserID) + } + + // Schedule create appends a fresh version, also published by the acting member. + w := httptest.NewRecorder() + req := newRequest("POST", "/api/autopilots/"+apID+"/triggers", map[string]any{ + "kind": "schedule", + "cron_expression": "0 0 * * *", + }) + req = withURLParam(req, "id", apID) + testHandler.CreateAutopilotTrigger(w, req) + if w.Code != http.StatusCreated { + t.Fatalf("schedule CreateAutopilotTrigger: got %d body=%s", w.Code, w.Body.String()) + } + ver2, err := testHandler.Queries.GetActiveAutopilotRuleVersion(ctx, verParams) + if err != nil { + t.Fatalf("schedule trigger create must republish a rule version: %v", err) + } + if ver2.ID.Bytes == ver.ID.Bytes { + t.Error("schedule create must append a NEW rule version, not reuse the webhook one") + } + if ver2.PublishedByType != "member" || uuidToString(ver2.PublishedByID) != testUserID { + t.Errorf("schedule version published_by = %s/%s, want member/%s", ver2.PublishedByType, uuidToString(ver2.PublishedByID), testUserID) + } +} + // createWebhookTriggerWithFilters builds the request body with a real JSON // array — the same shape the frontend sends. Earlier revisions of this // helper marshaled the filters separately and assigned the resulting @@ -111,7 +163,7 @@ func TestWebhookHandler_FiltersUndeclaredEvent(t *testing.T) { }) w := postWebhook(t, *trig.WebhookToken, map[string]any{ - "action": "in_progress", + "action": "in_progress", "workflow_run": map[string]any{"id": 123}, }, map[string]string{"X-GitHub-Event": "workflow_run"}) if w.Code != http.StatusOK { @@ -149,7 +201,7 @@ func TestWebhookHandler_AllowsDeclaredEvent(t *testing.T) { }) w := postWebhook(t, *trig.WebhookToken, map[string]any{ - "action": "completed", + "action": "completed", "workflow_run": map[string]any{"id": 123}, }, map[string]string{"X-GitHub-Event": "workflow_run"}) if w.Code != http.StatusOK { @@ -170,7 +222,7 @@ func TestWebhookHandler_EmptyFiltersAllowsAll(t *testing.T) { trig := createWebhookTriggerViaHandler(t, apID) w := postWebhook(t, *trig.WebhookToken, map[string]any{ - "action": "in_progress", + "action": "in_progress", "workflow_run": map[string]any{"id": 123}, }, map[string]string{"X-GitHub-Event": "workflow_run"}) if w.Code != http.StatusOK {