fix(feed): dedup rule-driven counterparty/series rows on the timeline

docs/activity-timeline.md

@@ -397,13 +397,21 @@ Enrichment is a pure transformation in `internal/service/annotations_enrich.go`
 that runs on every list of annotations before they reach the UI (or MCP). It
 does three things:
 
-1. **Drop rule-source structural rows.** When a rule fires during sync, the
-   sync engine writes a `rule_applied` annotation **and** a structural
-   side-effect row (`tag_added` / `category_set`) carrying
+1. **Drop rule-source structural rows.** When a rule fires (during sync or a
+   retroactive apply), the engine writes a `rule_applied` annotation **and** a
+   structural side-effect row — `tag_added` / `tag_removed` / `category_set`,
+   or a `counterparty_assigned` / `counterparty_unlinked` /
+   `series_assigned` / `series_unlinked` membership row — carrying
    `payload.source = "rule"`. The `rule_applied` row is the canonical audit
    record; its rule-source siblings are noise. `isRuleSourceDuplicate`
    filters them out. Comments and `rule_applied` itself are never deduped
-   here — only structural side-effects flagged with `source: "rule"`.
+   here — only structural side-effects flagged with `source: "rule"`. The
+   membership kinds were added after the original dedup and were initially
+   omitted, which surfaced a rule-driven counterparty/series assignment as
+   two adjacent feed rows (the `rule_applied` row plus a system-actor
+   "Breadbox assigned …" row) — issue #1915. When you add a new rule-write
+   path, stamp `source: "rule"` on its side-effect rows **and** add the kind
+   to `isRuleSourceDuplicate`.
 
 2. **Drop adjacent same-actor comment-vs-tag-note duplicates.** The MCP
    `update_transactions` tool can write a `tag_added` with `payload.note`

internal/service/annotations_enrich.go

@@ -110,15 +110,32 @@ func EnrichAnnotations(in []Annotation, opts EnrichOptions) []Annotation {
 // the raw slug round-trips unchanged.
 func identityDisplay(s string) string { return s }
 
+// annotationSourceRule marks a structural annotation (tag / category /
+// counterparty / series membership) written as a side-effect of a rule firing.
+// EnrichAnnotations drops these in favour of the parent rule_applied row, which
+// is the canonical audit record. Rule-write paths stamp it into the payload;
+// user- and agent-authored writes leave it absent so those events survive.
+const annotationSourceRule = "rule"
+
 // isRuleSourceDuplicate reports whether an annotation row was written as a
 // side-effect of a rule application and therefore duplicates a parent
-// rule_applied row. Only tag_added / tag_removed / category_set rows can
-// carry source="rule" today; comment and rule_applied are never deduped here.
+// rule_applied row. tag_added / tag_removed / category_set and the
+// counterparty / series membership kinds can carry source="rule"; comment and
+// rule_applied are never deduped here.
+//
+// The membership kinds were added after the original dedup (see the
+// *_annotations_series_kinds.sql / *_annotations_counterparty_kinds.sql
+// migrations) and were previously omitted — that gap is what surfaced a
+// rule-driven assignment as two adjacent feed events (the rule_applied row plus
+// a "Breadbox assigned …" membership row). Keeping the list in lock-step with
+// the rule-write paths is what collapses them back into one.
 func isRuleSourceDuplicate(a Annotation) bool {
 	switch a.Kind {
-	case "tag_added", "tag_removed", "category_set":
+	case "tag_added", "tag_removed", "category_set",
+		"counterparty_assigned", "counterparty_unlinked",
+		"series_assigned", "series_unlinked":
 		source, _ := a.Payload["source"].(string)
-		return source == "rule"
+		return source == annotationSourceRule
 	}
 	return false
 }
@@ -420,6 +437,13 @@ func formatRuleSummary(ruleName, field, value string, categoryDisplay func(strin
 		verb = "added tag " + value
 	case "comment":
 		verb = "added a comment"
+	case "counterparty":
+		// The membership row (which carries the counterparty name) is deduped
+		// away in favour of this rule_applied row, so phrase it generically —
+		// the rule name already names the counterparty in practice.
+		verb = "set the counterparty"
+	case "series":
+		verb = "linked a recurring series"
 	default:
 		verb = "applied"
 	}

internal/service/annotations_enrich_test.go

@@ -593,6 +593,94 @@ func TestEnrichAnnotations_CounterpartyMembershipSummaries(t *testing.T) {
 	}
 }
 
+// A rule-driven counterparty/series assignment writes BOTH a rule_applied row
+// and a membership side-effect row stamped source="rule". Enrichment drops the
+// membership row in favour of the parent rule_applied row, so the feed and the
+// per-transaction timeline show one event instead of two (issue #1915). This is
+// the membership twin of TestEnrichAnnotations_DropsRuleSourcedTagAdded.
+func TestEnrichAnnotations_DropsRuleSourcedMembership(t *testing.T) {
+	ruleID := "rule-cp"
+	rows := []Annotation{
+		{
+			ID:        "cp-membership",
+			Kind:      "counterparty_assigned",
+			ActorName: "Breadbox", // SystemActor() — the mislabelled "Breadbox assigned …" row
+			ActorType: "system",
+			Payload: map[string]interface{}{
+				"counterparty_id":   "CPX12345",
+				"counterparty_name": "Netflix",
+				"source":            "rule",
+			},
+			CreatedAt: "2026-04-04T12:00:00Z",
+		},
+		{
+			ID:        "series-membership",
+			Kind:      "series_assigned",
+			ActorName: "Breadbox",
+			ActorType: "system",
+			Payload: map[string]interface{}{
+				"series_id":   "SER12345",
+				"series_name": "Netflix",
+				"source":      "rule",
+			},
+			CreatedAt: "2026-04-04T12:00:01Z",
+		},
+		{
+			ID:        "cp-rule",
+			Kind:      "rule_applied",
+			ActorName: "Netflix Counterparty",
+			ActorType: "system",
+			ActorID:   &ruleID,
+			RuleID:    &ruleID,
+			Payload: map[string]interface{}{
+				"rule_name":    "Netflix Counterparty",
+				"action_field": "counterparty",
+				"action_value": "CPX12345",
+				"applied_by":   "retroactive",
+			},
+			CreatedAt: "2026-04-04T12:00:02Z",
+		},
+	}
+	out := EnrichAnnotations(rows, EnrichOptions{})
+	if len(out) != 1 {
+		t.Fatalf("len = %d, want 1 (rule_applied only)", len(out))
+	}
+	if out[0].Kind != "rule_applied" {
+		t.Errorf("survivor kind = %q, want rule_applied", out[0].Kind)
+	}
+	// The surviving rule_applied sentence names the action without needing the
+	// deduped membership row.
+	wantSummary := `Rule "Netflix Counterparty" set the counterparty retroactively`
+	if out[0].Summary != wantSummary {
+		t.Errorf("Summary = %q, want %q", out[0].Summary, wantSummary)
+	}
+}
+
+// A user- or agent-authored counterparty/series assignment (no source="rule")
+// is a first-class event and must survive enrichment — only rule side-effects
+// are deduped.
+func TestEnrichAnnotations_KeepsUserAuthoredMembership(t *testing.T) {
+	actorID := "user-1"
+	rows := []Annotation{
+		{
+			ID:        "cp-user",
+			Kind:      "counterparty_assigned",
+			ActorName: "Alice",
+			ActorType: "user",
+			ActorID:   &actorID,
+			Payload:   map[string]interface{}{"counterparty_id": "CPX12345", "counterparty_name": "Netflix"},
+			CreatedAt: "2026-04-04T12:00:00Z",
+		},
+	}
+	out := EnrichAnnotations(rows, EnrichOptions{})
+	if len(out) != 1 {
+		t.Fatalf("len = %d, want 1 (user membership survives)", len(out))
+	}
+	if out[0].Summary != "Alice set the counterparty to Netflix" {
+		t.Errorf("Summary = %q", out[0].Summary)
+	}
+}
+
 func TestEnrichAnnotations_PreservesOrder(t *testing.T) {
 	rows := []Annotation{
 		{ID: "a", Kind: "comment", ActorName: "alice", Payload: map[string]interface{}{"content": "1"}, CreatedAt: "2026-04-01T00:00:00Z"},

internal/service/counterparties.go

@@ -75,7 +75,7 @@ func (s *Service) createCounterpartyByName(ctx context.Context, in AssignCounter
 	}
 
 	if len(memberIDs) > 0 {
-		if err := linkCounterpartyAndAnnotate(ctx, tx, qtx, row, memberIDs, actor); err != nil {
+		if err := linkCounterpartyAndAnnotate(ctx, tx, qtx, row, memberIDs, actor, ""); err != nil {
 			return nil, err
 		}
 	}
@@ -118,7 +118,7 @@ func (s *Service) AssignCounterpartyFromRuleTx(ctx context.Context, tx pgx.Tx, t
 		return nil // nothing actionable
 	}
 
-	return linkCounterpartyAndAnnotate(ctx, tx, qtx, row, []pgtype.UUID{txnID}, SystemActor())
+	return linkCounterpartyAndAnnotate(ctx, tx, qtx, row, []pgtype.UUID{txnID}, SystemActor(), annotationSourceRule)
 }
 
 // resolveOrCreateCounterpartyByName returns the oldest live counterparty with the
@@ -143,8 +143,10 @@ func resolveOrCreateCounterpartyByName(ctx context.Context, qtx *db.Queries, nam
 
 // linkCounterpartyAndAnnotate links members to a counterparty (NULL-fill only)
 // and emits a counterparty_assigned annotation for the charges this call actually
-// linked. The caller owns the surrounding transaction.
-func linkCounterpartyAndAnnotate(ctx context.Context, tx pgx.Tx, qtx *db.Queries, cp db.Counterparty, memberIDs []pgtype.UUID, actor Actor) error {
+// linked. The caller owns the surrounding transaction. `source` flows to the
+// annotation payload — the rule path passes annotationSourceRule so the row is
+// deduped against its parent rule_applied row; imperative callers pass "".
+func linkCounterpartyAndAnnotate(ctx context.Context, tx pgx.Tx, qtx *db.Queries, cp db.Counterparty, memberIDs []pgtype.UUID, actor Actor, source string) error {
 	if len(memberIDs) == 0 {
 		return nil
 	}
@@ -161,7 +163,7 @@ func linkCounterpartyAndAnnotate(ctx context.Context, tx pgx.Tx, qtx *db.Queries
 		return fmt.Errorf("link counterparty members: %w", err)
 	}
 	if len(newlyLinked) > 0 {
-		if err := emitCounterpartyMembershipAnnotations(ctx, qtx, annotationKindCounterpartyAssigned, newlyLinked, cp.ShortID, cp.Name, actor); err != nil {
+		if err := emitCounterpartyMembershipAnnotations(ctx, qtx, annotationKindCounterpartyAssigned, newlyLinked, cp.ShortID, cp.Name, actor, source); err != nil {
 			return fmt.Errorf("emit counterparty_assigned annotations: %w", err)
 		}
 	}
@@ -195,7 +197,7 @@ func (s *Service) linkCounterpartyMembers(ctx context.Context, idOrShort string,
 		}
 		return nil, fmt.Errorf("get counterparty: %w", err)
 	}
-	if err := linkCounterpartyAndAnnotate(ctx, tx, qtx, row, memberIDs, actor); err != nil {
+	if err := linkCounterpartyAndAnnotate(ctx, tx, qtx, row, memberIDs, actor, ""); err != nil {
 		return nil, err
 	}
 	if err := tx.Commit(ctx); err != nil {
@@ -253,7 +255,7 @@ func (s *Service) UnlinkCounterpartyTransactions(ctx context.Context, idOrShort
 			ErrInvalidParameter, len(memberIDs)-int(detached), len(memberIDs))
 	}
 
-	if err := emitCounterpartyMembershipAnnotations(ctx, qtx, annotationKindCounterpartyUnlinked, memberIDs, row.ShortID, row.Name, actor); err != nil {
+	if err := emitCounterpartyMembershipAnnotations(ctx, qtx, annotationKindCounterpartyUnlinked, memberIDs, row.ShortID, row.Name, actor, ""); err != nil {
 		return nil, fmt.Errorf("emit counterparty_unlinked annotations: %w", err)
 	}
 
@@ -526,7 +528,11 @@ func unlinkedCounterpartySubset(ctx context.Context, tx pgx.Tx, ids []pgtype.UUI
 // counterparty_unlinked annotation per affected transaction, atomic with the
 // surrounding tx. The payload carries the counterparty short_id + name so the
 // timeline can render and deep-link the sentence.
-func emitCounterpartyMembershipAnnotations(ctx context.Context, q *db.Queries, kind string, txnIDs []pgtype.UUID, cpShortID, cpName string, actor Actor) error {
+// When source is non-empty it is stamped into the payload (the rule paths pass
+// annotationSourceRule so the row is recognised as a rule side-effect and
+// deduped against the parent rule_applied row by EnrichAnnotations; user- and
+// agent-authored calls pass "" so their events survive).
+func emitCounterpartyMembershipAnnotations(ctx context.Context, q *db.Queries, kind string, txnIDs []pgtype.UUID, cpShortID, cpName string, actor Actor, source string) error {
 	if len(txnIDs) == 0 {
 		return nil
 	}
@@ -537,6 +543,9 @@ func emitCounterpartyMembershipAnnotations(ctx context.Context, q *db.Queries, k
 		"counterparty_id":   cpShortID,
 		"counterparty_name": cpName,
 	}
+	if source != "" {
+		payload["source"] = source
+	}
 	for _, txnID := range txnIDs {
 		if err := writeAnnotation(ctx, q, writeAnnotationParams{
 			TransactionID: txnID,

internal/service/counterparties_integration_test.go

@@ -240,6 +240,65 @@ func TestApplyRuleRetroactively_AssignCounterpartyByShortID(t *testing.T) {
 	}
 }
 
+// TestApplyRuleRetroactively_AssignCounterparty_FeedSingleEvent is the
+// regression test for issue #1915: a retroactive assign_counterparty rule must
+// surface as ONE feed event (the rule_applied row), not two. Before the fix the
+// rule wrote both a rule_applied row (attributed to the rule) AND a
+// counterparty_assigned side-effect row (attributed to SystemActor "Breadbox"),
+// and the latter escaped the rule-source dedup — so the feed showed the same
+// retroactive application twice ("… ran a rule …" + "Breadbox assigned …").
+func TestApplyRuleRetroactively_AssignCounterparty_FeedSingleEvent(t *testing.T) {
+	svc, queries, _ := newService(t)
+	ctx := context.Background()
+	acctID := seedTxnFixture(t, queries)
+	// Three charges so the bucket clears the default bulk-action threshold.
+	testutil.MustCreateTransaction(t, queries, acctID, "NETFLIX.COM 1", "Netflix", 1599, "2026-03-15")
+	testutil.MustCreateTransaction(t, queries, acctID, "NETFLIX.COM 2", "Netflix", 1599, "2026-04-15")
+	testutil.MustCreateTransaction(t, queries, acctID, "NETFLIX.COM 3", "Netflix", 1599, "2026-05-15")
+	actor := service.Actor{Type: "user", ID: "u1", Name: "Tester"}
+
+	cp, err := svc.AssignCounterparty(ctx, service.AssignCounterpartyInput{Name: "Netflix", CreateIfMissing: true}, actor)
+	if err != nil {
+		t.Fatalf("create counterparty: %v", err)
+	}
+	rule, err := svc.CreateTransactionRule(ctx, service.CreateTransactionRuleParams{
+		Name:       "Netflix Counterparty",
+		Conditions: service.Condition{Field: "provider_name", Op: "contains", Value: "Netflix"},
+		Actions:    []service.RuleAction{{Type: "assign_counterparty", CounterpartyShortID: cp.ShortID}},
+		Actor:      actor,
+	})
+	if err != nil {
+		t.Fatalf("create rule: %v", err)
+	}
+	if _, err := svc.ApplyRuleRetroactively(ctx, rule.ID); err != nil {
+		t.Fatalf("ApplyRuleRetroactively: %v", err)
+	}
+
+	events, err := svc.ListFeedEvents(ctx, service.FeedEventsParams{})
+	if err != nil {
+		t.Fatalf("ListFeedEvents: %v", err)
+	}
+
+	// Exactly one bulk_action event survives — the rule_applied bucket. The
+	// counterparty_assigned side-effect (system "Breadbox" actor) is deduped.
+	bulk := countEventsByType(events, "bulk_action")
+	if bulk != 1 {
+		for _, ev := range events {
+			if ev.Type == "bulk_action" {
+				t.Logf("bulk_action kind=%q actor=%q", ev.BulkAction.Kind, ev.BulkAction.ActorName)
+			}
+		}
+		t.Fatalf("bulk_action events = %d, want 1 (rule_applied only; membership side-effect deduped)", bulk)
+	}
+	ev := findEventByType(events, "bulk_action").BulkAction
+	if ev.Kind != "rule_applied" {
+		t.Errorf("surviving bulk_action kind = %q, want rule_applied", ev.Kind)
+	}
+	if ev.ActorName == "Breadbox" {
+		t.Errorf("surviving event attributed to %q — the system membership row should have been deduped", ev.ActorName)
+	}
+}
+
 // TestApplyRuleRetroactively_AssignCounterpartyByName covers resolve-or-create by
 // name on the single-rule retroactive path.
 func TestApplyRuleRetroactively_AssignCounterpartyByName(t *testing.T) {

internal/service/series.go

@@ -135,7 +135,7 @@ func (s *Service) mintSeriesByName(ctx context.Context, in AssignSeriesInput, ac
 	}
 
 	if len(memberIDs) > 0 {
-		if err := backLinkMembers(ctx, tx, qtx, seriesID, memberIDs, actor); err != nil {
+		if err := backLinkMembers(ctx, tx, qtx, seriesID, memberIDs, actor, ""); err != nil {
 			return nil, err
 		}
 	}
@@ -185,7 +185,7 @@ func (s *Service) AssignSeriesFromRuleTx(ctx context.Context, tx pgx.Tx, txnID p
 		return nil // nothing actionable
 	}
 
-	if err := backLinkMembers(ctx, tx, qtx, seriesID, []pgtype.UUID{txnID}, SystemActor()); err != nil {
+	if err := backLinkMembers(ctx, tx, qtx, seriesID, []pgtype.UUID{txnID}, SystemActor(), annotationSourceRule); err != nil {
 		return err
 	}
 	return nil
@@ -194,8 +194,10 @@ func (s *Service) AssignSeriesFromRuleTx(ctx context.Context, tx pgx.Tx, txnID p
 // backLinkMembers back-links members to a series (NULL-fill only) and emits a
 // series_assigned annotation for the charges this call actually linked. Shared
 // by the mint, link, and rule paths. The caller owns the surrounding
-// transaction.
-func backLinkMembers(ctx context.Context, tx pgx.Tx, qtx *db.Queries, seriesID pgtype.UUID, memberIDs []pgtype.UUID, actor Actor) error {
+// transaction. `source` flows to the annotation payload — the rule path passes
+// annotationSourceRule so the row is deduped against its parent rule_applied
+// row; imperative callers pass "".
+func backLinkMembers(ctx context.Context, tx pgx.Tx, qtx *db.Queries, seriesID pgtype.UUID, memberIDs []pgtype.UUID, actor Actor, source string) error {
 	if len(memberIDs) == 0 {
 		return nil
 	}
@@ -217,7 +219,7 @@ func backLinkMembers(ctx context.Context, tx pgx.Tx, qtx *db.Queries, seriesID p
 		if err != nil {
 			return fmt.Errorf("reload series for annotation: %w", err)
 		}
-		if err := emitSeriesMembershipAnnotations(ctx, qtx, annotationKindSeriesAssigned, newlyLinked, row.ShortID, row.Name, actor); err != nil {
+		if err := emitSeriesMembershipAnnotations(ctx, qtx, annotationKindSeriesAssigned, newlyLinked, row.ShortID, row.Name, actor, source); err != nil {
 			return fmt.Errorf("emit series_assigned annotations: %w", err)
 		}
 	}
@@ -250,7 +252,7 @@ func (s *Service) linkSeriesMembers(ctx context.Context, idOrShort string, membe
 		}
 		return nil, fmt.Errorf("get series: %w", err)
 	}
-	if err := backLinkMembers(ctx, tx, qtx, row.ID, memberIDs, actor); err != nil {
+	if err := backLinkMembers(ctx, tx, qtx, row.ID, memberIDs, actor, ""); err != nil {
 		return nil, err
 	}
 	if err := tx.Commit(ctx); err != nil {
@@ -364,7 +366,7 @@ func (s *Service) UnlinkSeriesTransactions(ctx context.Context, idOrShort string
 			ErrInvalidParameter, len(memberIDs)-int(detached), len(memberIDs))
 	}
 
-	if err := emitSeriesMembershipAnnotations(ctx, qtx, annotationKindSeriesUnlinked, memberIDs, row.ShortID, row.Name, actor); err != nil {
+	if err := emitSeriesMembershipAnnotations(ctx, qtx, annotationKindSeriesUnlinked, memberIDs, row.ShortID, row.Name, actor, ""); err != nil {
 		return nil, fmt.Errorf("emit series_unlinked annotations: %w", err)
 	}
 
@@ -602,7 +604,11 @@ func unlinkedMemberSubset(ctx context.Context, tx pgx.Tx, ids []pgtype.UUID) ([]
 // annotation per affected transaction, atomic with the surrounding tx (q is the
 // tx-scoped *db.Queries). The payload carries the series short_id + name so the
 // timeline can render and deep-link the sentence.
-func emitSeriesMembershipAnnotations(ctx context.Context, q *db.Queries, kind string, txnIDs []pgtype.UUID, seriesShortID, seriesName string, actor Actor) error {
+// When source is non-empty it is stamped into the payload (the rule path passes
+// annotationSourceRule so the row is deduped against the parent rule_applied row
+// by EnrichAnnotations; user- and agent-authored calls pass "" so their events
+// survive).
+func emitSeriesMembershipAnnotations(ctx context.Context, q *db.Queries, kind string, txnIDs []pgtype.UUID, seriesShortID, seriesName string, actor Actor, source string) error {
 	if len(txnIDs) == 0 {
 		return nil
 	}
@@ -613,6 +619,9 @@ func emitSeriesMembershipAnnotations(ctx context.Context, q *db.Queries, kind st
 		"series_id":   seriesShortID,
 		"series_name": seriesName,
 	}
+	if source != "" {
+		payload["source"] = source
+	}
 	for _, txnID := range txnIDs {
 		if err := writeAnnotation(ctx, q, writeAnnotationParams{
 			TransactionID: txnID,