feat(attribution): rule_owner versioning on trigger edits + system-pause/archive (MUL-4302)

The final remaining Phase 1 item: substantive publishes beyond the autopilot row now republish the rule version, so a run's rule_owner accountable follows whoever last changed what the rule does. - Extracted the config-summary + insert into service.RecordAutopilotRuleVersion so the handler and the (different-package) failure monitor share one writer; the handler's recordAutopilotRuleVersion is now a thin wrapper. - Trigger edits: UpdateAutopilotTrigger and DeleteAutopilotTrigger republish the rule version with the acting member as publisher, ATOMICALLY (tx-wrapped mutation + version write, mirroring CreateAutopilot/UpdateAutopilot). CreateAutopilotTrigger republishes best-effort — the webhook path mints its token with a retry loop that cannot share one tx, and a create is usually initial setup already covered by v1; a failed write there is benign (active version stays the current publisher, the new trigger fires under it, no immediate daemon claim rides it). - Archive (DeleteAutopilot) republishes (member, status=archived), tx-wrapped. - System auto-pause (failure monitor) republishes with a 'system' publisher, best-effort — a background sweep to a non-dispatching state (a paused autopilot never dispatches; a later member resume supersedes). - RotateWebhookToken / SetSigningSecret deliberately do NOT version: they rotate credentials, not the rule's behavior (not §3.4 substantive). Semantics: a system-published (no-member) active version degrades dispatch to unattributed → owner_fallback, never fabricating a human. Tests: republish-reattributes (member A → member B supersedes → dispatch resolves to B; system publisher → unattributed). Full service/attribution/handler/migration/ scheduler/cmd suites pass on a DB migrated through 161; build/vet/gofmt clean. Also merges origin/main (unrelated frontend feature #5074). Co-authored-by: multica-agent <github@multica.ai>
2026-08-02 01:45:52 +02:00 · 2026-07-09 20:47:08 +08:00
parent 3d631f6658
commit c48d169d3d
4 changed files with 213 additions and 41 deletions
--- a/server/cmd/server/autopilot_failure_monitor.go
+++ b/server/cmd/server/autopilot_failure_monitor.go
@@ -14,6 +14,7 @@ import (
 	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgtype"
 	"github.com/multica-ai/multica/server/internal/events"
+	"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"
@@ -147,6 +148,17 @@ func tickAutopilotFailureMonitor(ctx context.Context, queries *db.Queries, bus *
 			continue
 		}

+		// A system auto-pause is a substantive status change (MUL-4302 §3.4).
+		// Record it as a rule-version publish with a 'system' publisher (no member
+		// actor). Best-effort: the monitor is a background sweep, a paused autopilot
+		// does not dispatch (so this version is never the active version at a real
+		// run — a later member resume would supersede it), and a failed write must
+		// not abort the sweep.
+		if verr := service.RecordAutopilotRuleVersion(ctx, queries, paused, "system", pgtype.UUID{}); verr != nil {
+			slog.Warn("autopilot failure monitor: record rule version failed",
+				"autopilot_id", util.UUIDToString(paused.ID), "error", verr)
+		}
+
 		failPct := 100.0
 		if c.TotalRuns > 0 {
 			failPct = math.Round(float64(c.FailedRuns)/float64(c.TotalRuns)*1000) / 10 // one decimal place
--- a/server/internal/handler/autopilot.go
+++ b/server/internal/handler/autopilot.go
@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
+	"log/slog"
 	"net/http"
 	"strconv"
 	"strings"
@@ -938,24 +939,12 @@ func (h *Handler) UpdateAutopilot(w http.ResponseWriter, r *http.Request) {
 	writeJSON(w, http.StatusOK, resp)
 }

-// autopilotRuleConfigSummary captures the substantive (accountability-bearing)
-// config of an autopilot at publish time, stored on each rule-version snapshot for
-// audit display (MUL-4302 §7). Cosmetic fields (title / description / issue title
-// template) are intentionally excluded — changing them does not transfer
-// accountability, so they neither appear here nor trigger a new version.
-type autopilotRuleConfigSummary struct {
-	AssigneeType  string `json:"assignee_type"`
-	AssigneeID    string `json:"assignee_id"`
-	Status        string `json:"status"`
-	ExecutionMode string `json:"execution_mode"`
-}
-
 // autopilotRuleSubstantiveChange reports whether a substantive (publish-worthy)
 // field of the autopilot ROW changed between prev and next: the execution target
 // (assignee), enabled-state (status: active/paused/archived), or execution mode
 // (MUL-4302 §3.4). Trigger-table edits (cron / webhook / event_filters) are also
-// substantive per the design but are not yet wired here — see the PR description's
-// remaining-work note.
+// substantive and republish the rule version in the trigger CRUD handlers
+// (Create/Update/Delete trigger); archive and system-pause do so in their own paths.
 func autopilotRuleSubstantiveChange(prev, next db.Autopilot) bool {
 	return prev.AssigneeType != next.AssigneeType ||
 		prev.AssigneeID != next.AssigneeID ||
@@ -964,30 +953,11 @@ func autopilotRuleSubstantiveChange(prev, next db.Autopilot) bool {
 }

 // recordAutopilotRuleVersion appends one rule-version snapshot for a substantive
-// publish (MUL-4302 §3.4), recording the publisher and the effective config. Runs
-// inside the caller's tx so the version is atomic with the autopilot write; a
-// failure surfaces to the caller rather than leaving a version-less publish that
-// would degrade every future run to unattributed.
+// 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 {
-	summary, err := json.Marshal(autopilotRuleConfigSummary{
-		AssigneeType:  ap.AssigneeType,
-		AssigneeID:    uuidToString(ap.AssigneeID),
-		Status:        ap.Status,
-		ExecutionMode: ap.ExecutionMode,
-	})
-	if err != nil {
-		return fmt.Errorf("marshal rule version config summary: %w", err)
-	}
-	if _, err := q.CreateAutopilotRuleVersion(ctx, db.CreateAutopilotRuleVersionParams{
-		AutopilotID:     ap.ID,
-		WorkspaceID:     ap.WorkspaceID,
-		PublishedByType: publishedByType,
-		PublishedByID:   publishedByID,
-		ConfigSummary:   summary,
-	}); err != nil {
-		return fmt.Errorf("create autopilot rule version: %w", err)
-	}
-	return nil
+	return service.RecordAutopilotRuleVersion(ctx, q, ap, publishedByType, publishedByID)
 }

 func (h *Handler) parseAutopilotProjectID(
@@ -1046,7 +1016,26 @@ func (h *Handler) DeleteAutopilot(w http.ResponseWriter, r *http.Request) {
 	// 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.
-	if err := h.Queries.ArchiveAutopilot(r.Context(), idUUID); err != nil {
+	// 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", parseUUID(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
 	}
@@ -1290,6 +1279,7 @@ func (h *Handler) CreateAutopilotTrigger(w http.ResponseWriter, r *http.Request)
 		}
 		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,
@@ -1315,6 +1305,7 @@ func (h *Handler) CreateAutopilotTrigger(w http.ResponseWriter, r *http.Request)

 	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,
@@ -1322,6 +1313,26 @@ 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
@@ -1556,14 +1567,37 @@ func (h *Handler) UpdateAutopilotTrigger(w http.ResponseWriter, r *http.Request)
 		params.NextRunAt = pgtype.Timestamptz{Time: t, Valid: true}
 	}

-	trigger, err := h.Queries.UpdateAutopilotTrigger(r.Context(), params)
+	userID, ok := requireUserID(w, r)
+	if !ok {
+		return
+	}
+
+	// Editing a trigger changes what / when the rule fires — a substantive publish
+	// (MUL-4302 §3.4) — so republish the rule version with this member as publisher,
+	// atomically with the trigger update (mirrors CreateAutopilot / UpdateAutopilot).
+	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
+	}
+	if err := h.recordAutopilotRuleVersion(r.Context(), qtx, ap, "member", parseUUID(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)
-	userID, _ := requireUserID(w, r)
 	h.publish(protocol.EventAutopilotUpdated, workspaceID, "member", userID, map[string]any{
 		"autopilot_id": uuidToString(ap.ID),
 		"trigger":      resp,
@@ -1612,7 +1646,26 @@ func (h *Handler) DeleteAutopilotTrigger(w http.ResponseWriter, r *http.Request)
 		return
 	}

-	if err := h.Queries.DeleteAutopilotTrigger(r.Context(), triggerUUID); err != nil {
+	// 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", parseUUID(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
 	}
--- a/server/internal/service/attribution_stamp_test.go
+++ b/server/internal/service/attribution_stamp_test.go
@@ -558,6 +558,69 @@ func TestEnqueueTaskForIssueAutopilotManualStampsDirectHuman(t *testing.T) {
 	}
 }

+// TestRecordAutopilotRuleVersionRepublishReattributes verifies the final Phase 1
+// item (MUL-4302 §3.4): republishing a rule (as a trigger edit / archive / system
+// pause does) appends a new version, the LATEST version is the active one, and
+// dispatch attribution follows it. So editing member A's autopilot as member B
+// re-attributes subsequent runs to B; a system pause records a 'system' publisher.
+func TestRecordAutopilotRuleVersionRepublishReattributes(t *testing.T) {
+	pool := newResolveOriginatorPool(t)
+	ctx := context.Background()
+	q := db.New(pool)
+	workspaceID, memberA, agentID, _ := seedAttributionFixture(t, pool)
+	autopilotID, _ := seedRunOnlyAutopilot(t, pool, workspaceID, agentID, memberA)
+
+	var memberB string
+	if err := pool.QueryRow(ctx, `INSERT INTO "user" (name, email) VALUES ('Editor', $1) RETURNING id`,
+		fmt.Sprintf("editor-%d@multica.test", time.Now().UnixNano())).Scan(&memberB); err != nil {
+		t.Fatalf("seed member B: %v", err)
+	}
+	t.Cleanup(func() { pool.Exec(context.Background(), `DELETE FROM "user" WHERE id = $1`, memberB) })
+	if _, err := pool.Exec(ctx, `INSERT INTO member (workspace_id, user_id, role) VALUES ($1, $2, 'member')`,
+		workspaceID, memberB); err != nil {
+		t.Fatalf("seed member B membership: %v", err)
+	}
+
+	ap, err := q.GetAutopilot(ctx, util.MustParseUUID(autopilotID))
+	if err != nil {
+		t.Fatalf("get autopilot: %v", err)
+	}
+	verParams := db.GetActiveAutopilotRuleVersionParams{WorkspaceID: ap.WorkspaceID, AutopilotID: ap.ID}
+
+	// v1: creator (member A) publishes; active + dispatch attribute to A.
+	if err := RecordAutopilotRuleVersion(ctx, q, ap, "member", util.MustParseUUID(memberA)); err != nil {
+		t.Fatalf("record v1: %v", err)
+	}
+	active, err := q.GetActiveAutopilotRuleVersion(ctx, verParams)
+	if err != nil || active.PublishedByType != "member" || active.PublishedByID.Bytes != util.MustParseUUID(memberA).Bytes {
+		t.Fatalf("v1 active = %+v (err %v), want member A", active, err)
+	}
+
+	// v2: member B republishes (e.g. edited a trigger) → latest wins.
+	if err := RecordAutopilotRuleVersion(ctx, q, ap, "member", util.MustParseUUID(memberB)); err != nil {
+		t.Fatalf("record v2: %v", err)
+	}
+	attr := ruleOwnerAttribution(ctx, q, ap.WorkspaceID, ap.ID, attribution.EvidenceAutopilotRun, ap.ID)
+	if attr.Source != attribution.SourceRuleOwner || attr.AccountableUserID.Bytes != util.MustParseUUID(memberB).Bytes {
+		t.Errorf("after republish, dispatch attribution = %+v, want rule_owner accountable = member B", attr)
+	}
+
+	// v3: system auto-pause records a 'system' publisher (no member id).
+	if err := RecordAutopilotRuleVersion(ctx, q, ap, "system", pgtype.UUID{}); err != nil {
+		t.Fatalf("record v3 (system): %v", err)
+	}
+	active, err = q.GetActiveAutopilotRuleVersion(ctx, verParams)
+	if err != nil || active.PublishedByType != "system" || active.PublishedByID.Valid {
+		t.Errorf("v3 active = %+v (err %v), want system publisher with NULL id", active, err)
+	}
+	// A system-published version has no member → dispatch degrades to unattributed
+	// (never fabricates a human).
+	sysAttr := ruleOwnerAttribution(ctx, q, ap.WorkspaceID, ap.ID, attribution.EvidenceAutopilotRun, ap.ID)
+	if sysAttr.Source != attribution.SourceUnattributed || sysAttr.AccountableUserID.Valid {
+		t.Errorf("system-published version must yield unattributed, got %+v", sysAttr)
+	}
+}
+
 // TestApplyAttributionFallbackRefusesOnMissingOwner: an unattributed run in an
 // OPEN (non-fail-closed) workspace whose agent has no valid owner cannot resolve an
 // accountable human via owner_fallback, so the enqueue is refused rather than
--- a/server/internal/service/autopilot.go
+++ b/server/internal/service/autopilot.go
@@ -47,6 +47,50 @@ func NewAutopilotService(q *db.Queries, tx TxStarter, bus *events.Bus, taskSvc *
 	return &AutopilotService{Queries: q, TxStarter: tx, Bus: bus, TaskSvc: taskSvc}
 }

+// autopilotRuleConfigSummary captures the substantive (accountability-bearing)
+// config of an autopilot at publish time, stored on each rule-version snapshot for
+// audit display (MUL-4302 §7). Cosmetic fields (title / description / issue title
+// template) are intentionally excluded — changing them does not transfer
+// accountability. Trigger config (cron / webhook / event_filters) lives in a
+// separate table and is not inlined here; a trigger edit still republishes the
+// rule (recording the editing member + timestamp), the summary just carries the
+// autopilot row's core config.
+type autopilotRuleConfigSummary struct {
+	AssigneeType  string `json:"assignee_type"`
+	AssigneeID    string `json:"assignee_id"`
+	Status        string `json:"status"`
+	ExecutionMode string `json:"execution_mode"`
+}
+
+// RecordAutopilotRuleVersion appends one rule-version snapshot for a substantive
+// publish (MUL-4302 §3.4), recording the publisher and the effective config. Shared
+// by the handler publish paths (create / update / trigger edits / archive, run in
+// their tx) and the failure monitor's system-pause (a different package). q is the
+// caller's *db.Queries (tx-scoped where the caller wants atomicity). publishedByType
+// is "member" (with the acting member id) or "system" (with an invalid id, e.g. the
+// auto-pause monitor).
+func RecordAutopilotRuleVersion(ctx context.Context, q *db.Queries, ap db.Autopilot, publishedByType string, publishedByID pgtype.UUID) error {
+	summary, err := json.Marshal(autopilotRuleConfigSummary{
+		AssigneeType:  ap.AssigneeType,
+		AssigneeID:    util.UUIDToString(ap.AssigneeID),
+		Status:        ap.Status,
+		ExecutionMode: ap.ExecutionMode,
+	})
+	if err != nil {
+		return fmt.Errorf("marshal rule version config summary: %w", err)
+	}
+	if _, err := q.CreateAutopilotRuleVersion(ctx, db.CreateAutopilotRuleVersionParams{
+		AutopilotID:     ap.ID,
+		WorkspaceID:     ap.WorkspaceID,
+		PublishedByType: publishedByType,
+		PublishedByID:   publishedByID,
+		ConfigSummary:   summary,
+	}); err != nil {
+		return fmt.Errorf("create autopilot rule version: %w", err)
+	}
+	return nil
+}
+
 // DispatchAutopilot is the core execution entry point.
 // It creates a run and either creates an issue or enqueues a direct agent task
 // depending on execution_mode.