From 91c4268b1c80b529274431088f5c5e346ee53aa7 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 24 Jun 2026 06:37:54 +0000 Subject: [PATCH] fix(feed): dedup rule-driven counterparty/series rows on the timeline MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A retroactive (or sync-time) assign_counterparty / assign_series rule wrote both a `rule_applied` row (attributed to the rule) and a membership side-effect row (`counterparty_assigned` / `series_assigned`, attributed to the SystemActor "Breadbox"). The rule-source dedup in EnrichAnnotations only covered tag/category kinds, so the membership row escaped it and the same application surfaced as two adjacent feed events ("Netflix Counterparty ran a rule …" + "Breadbox assigned … to Netflix"). Mirror the category/tag pattern for membership kinds: - The rule-write paths (AssignCounterpartyFromRuleTx / AssignSeriesFromRuleTx) now stamp `source: "rule"` into the membership payload via a threaded `source` arg; user- and agent-authored assignments pass "" and survive. - isRuleSourceDuplicate drops counterparty_assigned/_unlinked and series_assigned/_unlinked rows carrying source="rule". - formatRuleSummary phrases the surviving rule_applied row for the counterparty / series action fields instead of falling back to "applied". The result: one rule_applied event, no duplicate "Breadbox …" row. Fixes #1915 Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01CvAru8AVLa95DGfZg4M1wj --- docs/activity-timeline.md | 16 +++- internal/service/annotations_enrich.go | 32 ++++++- internal/service/annotations_enrich_test.go | 88 +++++++++++++++++++ internal/service/counterparties.go | 25 ++++-- .../counterparties_integration_test.go | 59 +++++++++++++ internal/service/series.go | 25 ++++-- 6 files changed, 221 insertions(+), 24 deletions(-) diff --git a/docs/activity-timeline.md b/docs/activity-timeline.md index ce7cc1ed..b30a2716 100644 --- a/docs/activity-timeline.md +++ b/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` diff --git a/internal/service/annotations_enrich.go b/internal/service/annotations_enrich.go index 17ede630..6a98bc4e 100644 --- a/internal/service/annotations_enrich.go +++ b/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" } diff --git a/internal/service/annotations_enrich_test.go b/internal/service/annotations_enrich_test.go index b8871daf..992a6a0f 100644 --- a/internal/service/annotations_enrich_test.go +++ b/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"}, diff --git a/internal/service/counterparties.go b/internal/service/counterparties.go index da3e36c7..acaeeee1 100644 --- a/internal/service/counterparties.go +++ b/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, diff --git a/internal/service/counterparties_integration_test.go b/internal/service/counterparties_integration_test.go index 8ad00ed6..5745867d 100644 --- a/internal/service/counterparties_integration_test.go +++ b/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) { diff --git a/internal/service/series.go b/internal/service/series.go index 6e17fa78..4d9d1cc8 100644 --- a/internal/service/series.go +++ b/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,