fix(attribution): make trigger-create rule-version republish atomic (MUL-4302)

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 <github@multica.ai>

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")
 }

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 {