fix: backfill triggers on unchanged re-apply; test coverage

Addresses PR review feedback on the trigger-registration fix.

- workflow.Save reconciles triggers on the unchanged-content path too, so
  workflows applied before this fix (definition present, workflow_triggers
  empty) register their triggers on a plain re-apply instead of requiring
  a byte-level YAML change. Reconciliation only runs when the stored rows
  have actually drifted from the latest version's declared triggers, so
  steady-state GitOps re-applies don't churn rows (which matters for the
  email poller, whose IMAP connections key off trigger IDs).
- Backfilled rows inherit the workflow's enabled/disabled state, so a
  disabled workflow doesn't start firing on re-apply.
- Tests: assert webhook path/secret persistence, add backfill regression
  (including no-churn on repeat apply) and disabled-state preservation.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WnKgCRfMvFcZAkoVbr8k79

Author: claude

.changeset/fix-trigger-registration.md

@@ -2,4 +2,4 @@
 "@mantle/engine": patch
 ---
 
-Register workflow triggers on apply. Cron, webhook, and email triggers declared in a workflow's YAML were parsed and validated but never written to `workflow_triggers`, so they never fired. `workflow.Save` now reconciles the table in the same transaction as the definition insert (covering both `mantle apply` and GitOps sync), pruned/disabled workflows have their triggers disabled so they stop firing, and the email poller periodically reloads so new email triggers activate without a server restart.
+Register workflow triggers on apply. Cron, webhook, and email triggers declared in a workflow's YAML were parsed and validated but never written to `workflow_triggers`, so they never fired. `workflow.Save` now reconciles the table in the same transaction as the definition insert (covering both `mantle apply` and GitOps sync). Re-applying an unchanged workflow backfills its triggers if the table has drifted (so upgrading and re-applying activates triggers for workflows created before this fix, without churning rows in steady state). Pruned/disabled workflows have their triggers disabled so they stop firing, and the email poller periodically reloads so new email triggers activate without a server restart.

packages/engine/internal/workflow/store.go

@@ -30,6 +30,13 @@ func Save(ctx context.Context, database *sql.DB, result *ParseResult, rawContent
 		return 0, fmt.Errorf("checking latest hash: %w", err)
 	}
 	if latestHash == hash {
+		// Content unchanged, so no new version — but still backfill triggers
+		// if the table drifted from the definition (e.g. a workflow applied
+		// before trigger registration existed). Otherwise upgrading and
+		// re-applying would leave triggers inert until a byte-level change.
+		if err := reconcileTriggersIfDrifted(ctx, database, teamID, name, result.Workflow.Triggers); err != nil {
+			return 0, err
+		}
 		return 0, nil
 	}
 
@@ -60,8 +67,9 @@ func Save(ctx context.Context, database *sql.DB, result *ParseResult, rawContent
 
 	// Register the workflow's declared triggers so the cron/webhook/email
 	// schedulers actually fire it. Done in the same transaction as the
-	// definition insert so version and triggers commit atomically.
-	if err := syncTriggersTx(ctx, tx, teamID, name, newVersion, result.Workflow.Triggers); err != nil {
+	// definition insert so version and triggers commit atomically. A new
+	// version is always active, so its triggers are enabled.
+	if err := syncTriggersTx(ctx, tx, teamID, name, newVersion, result.Workflow.Triggers, true); err != nil {
 		return 0, err
 	}
 

packages/engine/internal/workflow/triggers.go

@@ -15,8 +15,10 @@ import (
 //
 // It runs inside Save's transaction so a new definition version and its
 // triggers are committed atomically — a workflow is never left registered at
-// one version with triggers pointing at another.
-func syncTriggersTx(ctx context.Context, tx *sql.Tx, teamID, name string, version int, triggers []Trigger) error {
+// one version with triggers pointing at another. enabled sets the rows'
+// enabled flag: true for an active workflow, false when backfilling a
+// currently-disabled one so it does not start firing.
+func syncTriggersTx(ctx context.Context, tx *sql.Tx, teamID, name string, version int, triggers []Trigger, enabled bool) error {
 	// A new version supersedes the previous one, so replace whatever was
 	// registered before rather than trying to diff. Scoped to team so we
 	// never touch another tenant's rows.
@@ -29,11 +31,12 @@ func syncTriggersTx(ctx context.Context, tx *sql.Tx, teamID, name string, versio
 	for _, t := range triggers {
 		if _, err := tx.ExecContext(ctx,
 			`INSERT INTO workflow_triggers
-			 (workflow_name, workflow_version, type, schedule, path, secret, team_id, mailbox, folder, filter, poll_interval)
-			 VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)`,
+			 (workflow_name, workflow_version, type, schedule, path, secret, team_id, mailbox, folder, filter, poll_interval, enabled)
+			 VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)`,
 			name, version, t.Type,
 			nullableString(t.Schedule), nullableString(t.Path), nullableString(t.Secret), teamID,
 			nullableString(t.Mailbox), nullableString(t.Folder), nullableString(t.Filter), nullableString(t.PollInterval),
+			enabled,
 		); err != nil {
 			return fmt.Errorf("registering %s trigger for %q: %w", t.Type, name, err)
 		}
@@ -42,6 +45,56 @@ func syncTriggersTx(ctx context.Context, tx *sql.Tx, teamID, name string, versio
 	return nil
 }
 
+// reconcileTriggersIfDrifted backfills a workflow's triggers on the
+// unchanged-content apply path. Save early-returns when the content hash is
+// unchanged, so without this a workflow applied before trigger registration
+// existed (definition present, workflow_triggers empty) would never register
+// its triggers unless the user made a byte-level YAML change. It reconciles
+// only when the stored rows are actually out of sync with the latest version's
+// declared triggers, so steady-state GitOps re-applies stay churn-free (which
+// matters for the email poller, whose IMAP connections key off trigger IDs).
+// The backfilled rows inherit the workflow's current enabled/disabled state.
+func reconcileTriggersIfDrifted(ctx context.Context, database *sql.DB, teamID, name string, desired []Trigger) error {
+	// Latest version and whether it is currently active (not disabled).
+	var version int
+	var active bool
+	err := database.QueryRowContext(ctx,
+		`SELECT version, disabled_at IS NULL FROM workflow_definitions
+		 WHERE name = $1 AND team_id = $2 ORDER BY version DESC LIMIT 1`,
+		name, teamID).Scan(&version, &active)
+	if err == sql.ErrNoRows {
+		return nil // no definition to reconcile against
+	}
+	if err != nil {
+		return fmt.Errorf("loading latest version for %q: %w", name, err)
+	}
+
+	// Already in sync when the row count matches and every row points at the
+	// latest version. Covers both the empty-table (pre-fix) case and rows
+	// left at a stale version.
+	var total, atVersion int
+	if err := database.QueryRowContext(ctx,
+		`SELECT COUNT(*), COUNT(*) FILTER (WHERE workflow_version = $3)
+		 FROM workflow_triggers WHERE workflow_name = $1 AND team_id = $2`,
+		name, teamID, version).Scan(&total, &atVersion); err != nil {
+		return fmt.Errorf("checking trigger drift for %q: %w", name, err)
+	}
+	if total == len(desired) && atVersion == len(desired) {
+		return nil
+	}
+
+	tx, err := database.BeginTx(ctx, nil)
+	if err != nil {
+		return fmt.Errorf("starting transaction: %w", err)
+	}
+	defer tx.Rollback() //nolint:errcheck
+
+	if err := syncTriggersTx(ctx, tx, teamID, name, version, desired, active); err != nil {
+		return err
+	}
+	return tx.Commit()
+}
+
 // nullableString maps an empty string to NULL so that nullable TEXT columns
 // receive NULL rather than an empty string. Reads COALESCE these back to
 // their defaults.

packages/engine/internal/workflow/triggers_test.go

@@ -97,6 +97,128 @@ func TestSave_RegistersTriggers(t *testing.T) {
 		t.Errorf("email fields = (%q,%q,%q,%q), want (inbox-cred,INBOX,unseen,30s)",
 			mailbox, folder, filter, poll)
 	}
+
+	// The webhook trigger row must carry its path and secret.
+	var path, secret string
+	if err := database.QueryRowContext(ctx,
+		`SELECT COALESCE(path,''), COALESCE(secret,'') FROM workflow_triggers
+		 WHERE workflow_name = 'triggered-wf' AND type = 'webhook'`,
+	).Scan(&path, &secret); err != nil {
+		t.Fatalf("querying webhook trigger: %v", err)
+	}
+	if path != "/incoming" || secret != "shh" {
+		t.Errorf("webhook fields = (%q,%q), want (/incoming,shh)", path, secret)
+	}
+}
+
+// triggerIDs returns the sorted IDs of a workflow's trigger rows.
+func triggerIDs(t *testing.T, database *sql.DB, name string) []string {
+	t.Helper()
+	rows, err := database.QueryContext(context.Background(),
+		`SELECT id::text FROM workflow_triggers WHERE workflow_name = $1 ORDER BY id`, name)
+	if err != nil {
+		t.Fatalf("querying trigger ids: %v", err)
+	}
+	defer rows.Close()
+	var ids []string
+	for rows.Next() {
+		var id string
+		if err := rows.Scan(&id); err != nil {
+			t.Fatalf("scanning id: %v", err)
+		}
+		ids = append(ids, id)
+	}
+	return ids
+}
+
+func equalStrings(a, b []string) bool {
+	if len(a) != len(b) {
+		return false
+	}
+	for i := range a {
+		if a[i] != b[i] {
+			return false
+		}
+	}
+	return true
+}
+
+// TestSave_BackfillsTriggersOnUnchangedApply is the regression test for the
+// upgrade path: a workflow whose definition already exists but has no
+// workflow_triggers rows (applied before trigger registration existed) must
+// get its triggers backfilled when identical content is re-applied — without
+// churning rows on subsequent no-op applies.
+func TestSave_BackfillsTriggersOnUnchangedApply(t *testing.T) {
+	database := setupTestDB(t)
+	ctx := context.Background()
+
+	result, _ := ParseBytes(triggeredWorkflowYAML)
+	if _, err := Save(ctx, database, result, triggeredWorkflowYAML); err != nil {
+		t.Fatalf("initial Save: %v", err)
+	}
+
+	// Simulate a pre-fix workflow: definition present, trigger rows absent.
+	if _, err := database.ExecContext(ctx,
+		`DELETE FROM workflow_triggers WHERE workflow_name = 'triggered-wf'`); err != nil {
+		t.Fatalf("clearing triggers: %v", err)
+	}
+	if got := countTriggers(t, database, "triggered-wf", false); got != 0 {
+		t.Fatalf("precondition: triggers = %d, want 0", got)
+	}
+
+	// Re-applying identical content must backfill the triggers (version stays 0).
+	version, err := Save(ctx, database, result, triggeredWorkflowYAML)
+	if err != nil {
+		t.Fatalf("backfill Save: %v", err)
+	}
+	if version != 0 {
+		t.Errorf("version = %d, want 0 (unchanged content)", version)
+	}
+	if got := countTriggers(t, database, "triggered-wf", true); got != 3 {
+		t.Fatalf("backfilled enabled triggers = %d, want 3", got)
+	}
+
+	// A further identical apply must be a true no-op: rows already in sync,
+	// so their IDs must not churn (the email poller keys off trigger IDs).
+	ids1 := triggerIDs(t, database, "triggered-wf")
+	if _, err := Save(ctx, database, result, triggeredWorkflowYAML); err != nil {
+		t.Fatalf("no-op Save: %v", err)
+	}
+	ids2 := triggerIDs(t, database, "triggered-wf")
+	if !equalStrings(ids1, ids2) {
+		t.Errorf("triggers churned on no-op apply: %v -> %v", ids1, ids2)
+	}
+}
+
+// TestSave_BackfillPreservesDisabledState verifies that backfilling triggers
+// for a currently-disabled workflow inserts them disabled, so a pruned
+// workflow does not start firing on a re-apply.
+func TestSave_BackfillPreservesDisabledState(t *testing.T) {
+	database := setupTestDB(t)
+	ctx := context.Background()
+
+	result, _ := ParseBytes(triggeredWorkflowYAML)
+	if _, err := Save(ctx, database, result, triggeredWorkflowYAML); err != nil {
+		t.Fatalf("initial Save: %v", err)
+	}
+	if err := Disable(ctx, database, "triggered-wf"); err != nil {
+		t.Fatalf("Disable: %v", err)
+	}
+	// Drop the rows to force a backfill on the next apply.
+	if _, err := database.ExecContext(ctx,
+		`DELETE FROM workflow_triggers WHERE workflow_name = 'triggered-wf'`); err != nil {
+		t.Fatalf("clearing triggers: %v", err)
+	}
+
+	if _, err := Save(ctx, database, result, triggeredWorkflowYAML); err != nil {
+		t.Fatalf("backfill Save: %v", err)
+	}
+	if got := countTriggers(t, database, "triggered-wf", false); got != 3 {
+		t.Errorf("backfilled rows = %d, want 3", got)
+	}
+	if got := countTriggers(t, database, "triggered-wf", true); got != 0 {
+		t.Errorf("enabled backfilled rows = %d, want 0 (workflow is disabled)", got)
+	}
 }
 
 // TestSave_ReplacesTriggersOnNewVersion verifies a new version supersedes the