From c48d169d3d192698fe5a8fa54569a44e83736746 Mon Sep 17 00:00:00 2001 From: J Date: Thu, 9 Jul 2026 20:47:08 +0800 Subject: [PATCH] feat(attribution): rule_owner versioning on trigger edits + system-pause/archive (MUL-4302) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- .../cmd/server/autopilot_failure_monitor.go | 12 ++ server/internal/handler/autopilot.go | 135 ++++++++++++------ .../service/attribution_stamp_test.go | 63 ++++++++ server/internal/service/autopilot.go | 44 ++++++ 4 files changed, 213 insertions(+), 41 deletions(-) diff --git a/server/cmd/server/autopilot_failure_monitor.go b/server/cmd/server/autopilot_failure_monitor.go index 3c0cc410e4..14d5f65f17 100644 --- 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 diff --git a/server/internal/handler/autopilot.go b/server/internal/handler/autopilot.go index 301da6d349..d8c8c4d678 100644 --- 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 } diff --git a/server/internal/service/attribution_stamp_test.go b/server/internal/service/attribution_stamp_test.go index 8acac16622..42c18be7c4 100644 --- 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 diff --git a/server/internal/service/autopilot.go b/server/internal/service/autopilot.go index 14a745e100..258dd6796c 100644 --- 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.