feat: subscription sync patch targeting (#4576)

Author: turip

.agents/skills/billing/SKILL.md

@@ -202,6 +202,8 @@ The same registry also owns the single `CreateLineRouter`. Billing's default rou
 - `ChangeSourceAPIRequest`: user/API-originated line edits. Billing diffs the original invoice against the edited invoice with `diffMutableInvoiceLines`, applies API edits through line engines with `applyAPIInvoiceLineEdits`, and rebuilds the invoice from unchanged lines plus line-engine outputs.
 - `ChangeSourceSystem`: system-originated edits from billing/charges/subscription sync. Billing still computes the line diff, but does not invoke API create/update callbacks. For standard invoices only, deleted lines are dispatched through `OnMutableStandardLinesDeletedBySystem` because charge line updaters currently rely on that cleanup notification. Gathering invoices do not emit system delete callbacks.
 
+Subscription-sync charge patches can update or delete invoice lines as a side effect of reconciling charges. Those invoice updates must use `ChangeSourceSystem`: API edit callbacks are for user-initiated invoice-line edits, while system deletes use `OnMutableStandardLinesDeletedBySystem` so charge line engines can clean up detached line-backed runs.
+
 The API edit path treats the diff as the owner of invoice lines while engines run:
 
 - `applyAPIInvoiceLineEdits` clones the edited invoice and calls `UnsetLines()` before line-engine dispatch so the edited invoice is only the header/context, not a competing source of line truth.

.agents/skills/charges/SKILL.md

@@ -116,6 +116,14 @@ Intent layer access rules:
 - Subscription sync and repair code targets subscription-owned source state. It should use `GetBaseIntent()` or base-specific helpers, not effective getters, so user/API overrides do not feed back into the subscription target state.
 - Adapter writes should persist the base intent with `GetBaseIntent()` and persist override fields only through the override-specific adapter paths. Do not derive base persistence values from effective intent accessors.
 
+Patch target rules:
+
+- Charge patch payloads (`PatchShrink`, `PatchExtend`, `PatchDelete`) must carry an explicit `meta.ChangeTarget`. Do not rely on a zero/default target.
+- Subscription sync emits patches against `meta.ChangeTargetBase` because it reconciles subscription-owned source state, not user/API overrides.
+- State machines should mutate the requested patch target, then reconcile customer-facing invoice artifacts from the effective intent. If a base-target patch hits a charge with an active override, updating the base layer should not rewrite effective invoice state.
+- Delete patches follow the same target rule: deleting the base layer while an override is active should mark the base intent deleted but leave the effective override-backed charge behavior intact.
+- Charge listing for subscription-sync persisted-state loading must filter on base-intent deletion state, not effective `DeletedAt`; otherwise an active override can hide a subscription-owned base charge that sync still needs to reconcile.
+
 ## Usage-Based Invoice Line Mapping
 
 Usage-based charge realization runs are not billing standard lines. When mapping a run back to `billing.StandardLine` in `openmeter/billing/charges/usagebased/service/linemapper.go`, preserve the billing-facing semantics:

.coderabbit.yaml

@@ -39,6 +39,16 @@ reviews:
 
         Only flag `AdvanceAfter` if the specific code path demonstrably persists an invalid value or violates an explicit flat-fee domain invariant in the current diff.
 
+    - path: "openmeter/billing/charges/**/service/*.go"
+      instructions: |
+        For charge state-machine handlers in this path, do not claim that an in-memory mutation is not persisted until you trace the full patch execution and persistence flow in the current code.
+
+        When a handler is invoked from `service.TriggerPatch`, follow this call chain before raising a persistence finding: `service.TriggerPatch -> stateMachine.FireAndActivate -> chargestatemachine.Machine.FireAndActivate`.
+
+        In this flow, mutations made by the handler are persisted after the handler returns via `Persistence.UpdateBase(ctx, m.Charge.GetBase())`.
+
+        Before reporting that a mutation is "only in memory" or "not saved", verify whether the handler mutates the base/intent state that `UpdateBase(...)` persists. If that persistence path exists in reachable current code, omit the finding. If you cannot verify a missing persistence path from the diff and reachable current code, omit the finding.
+
     - path: "**/*.tsp"
       instructions: |
         Review the TypeSpec code for conformity with TypeSpec best practices. When recommending changes also consider the fact that multiple codegeneration toolchains depend on the TypeSpec code, each of which have their idiosyncrasies and bugs.

openmeter/billing/charges/adapter/search.go

@@ -53,7 +53,11 @@ func (a *adapter) ListCharges(ctx context.Context, input charges.ListChargesInpu
 			Where(dbchargessearchv1.Namespace(input.Namespace))
 
 		if !input.IncludeDeleted {
-			query = query.Where(dbchargessearchv1.DeletedAtIsNil())
+			if input.DeletedAtFilter == charges.ListChargesDeletedAtFilterBaseIntent {
+				query = query.Where(dbchargessearchv1.BaseIntentDeletedAtIsNil())
+			} else {
+				query = query.Where(dbchargessearchv1.DeletedAtIsNil())
+			}
 		}
 
 		if len(input.CustomerIDs) > 0 {

openmeter/billing/charges/adapter/search_test.go

@@ -85,7 +85,7 @@ func (s *ListCustomersToAdvanceSuite) createCustomer(namespace string) string {
 }
 
 // insertFlatFeeCharge inserts a minimal flat fee charge row for testing the search view.
-func (s *ListCustomersToAdvanceSuite) insertFlatFeeCharge(namespace, customerID string, status meta.ChargeStatus, advanceAfter *time.Time) {
+func (s *ListCustomersToAdvanceSuite) insertFlatFeeCharge(namespace, customerID string, status meta.ChargeStatus, advanceAfter *time.Time) string {
 	s.T().Helper()
 
 	now := time.Now().UTC().Truncate(time.Microsecond)
@@ -121,8 +121,62 @@ func (s *ListCustomersToAdvanceSuite) insertFlatFeeCharge(namespace, customerID
 		create = create.SetAdvanceAfter(*advanceAfter)
 	}
 
-	_, err := create.Save(context.Background())
+	charge, err := create.Save(context.Background())
 	s.Require().NoError(err)
+
+	return charge.ID
+}
+
+func (s *ListCustomersToAdvanceSuite) TestListChargesDeletedAtFilter() {
+	ctx := s.T().Context()
+	ns := "test-list-charges-deleted-at-filter"
+	now := time.Now().UTC().Truncate(time.Microsecond)
+	deletedAt := now.Add(time.Minute)
+
+	customerID := s.createCustomer(ns)
+
+	liveChargeID := s.insertFlatFeeCharge(ns, customerID, meta.ChargeStatusActive, nil)
+	overrideDeletedChargeID := s.insertFlatFeeCharge(ns, customerID, meta.ChargeStatusActive, nil)
+	baseDeletedChargeID := s.insertFlatFeeCharge(ns, customerID, meta.ChargeStatusActive, nil)
+
+	_, err := s.dbClient.ChargeFlatFee.UpdateOneID(overrideDeletedChargeID).
+		SetDeletedAt(deletedAt).
+		Save(ctx)
+	s.Require().NoError(err)
+
+	_, err = s.dbClient.ChargeFlatFee.UpdateOneID(baseDeletedChargeID).
+		SetDeletedAt(deletedAt).
+		SetIntentDeletedAt(deletedAt).
+		Save(ctx)
+	s.Require().NoError(err)
+
+	listIDs := func(input charges.ListChargesInput) []string {
+		s.T().Helper()
+
+		input.Namespace = ns
+		input.ChargeTypes = []meta.ChargeType{meta.ChargeTypeFlatFee}
+
+		result, err := s.adapter.ListCharges(ctx, input)
+		s.Require().NoError(err)
+
+		out := make([]string, 0, len(result.Items))
+		for _, item := range result.Items {
+			out = append(out, item.ID.ID)
+		}
+
+		return out
+	}
+
+	s.ElementsMatch([]string{liveChargeID}, listIDs(charges.ListChargesInput{}))
+	s.ElementsMatch([]string{liveChargeID}, listIDs(charges.ListChargesInput{
+		DeletedAtFilter: charges.ListChargesDeletedAtFilterEffective,
+	}))
+	s.ElementsMatch([]string{liveChargeID, overrideDeletedChargeID}, listIDs(charges.ListChargesInput{
+		DeletedAtFilter: charges.ListChargesDeletedAtFilterBaseIntent,
+	}))
+	s.ElementsMatch([]string{liveChargeID, overrideDeletedChargeID, baseDeletedChargeID}, listIDs(charges.ListChargesInput{
+		IncludeDeleted: true,
+	}))
 }
 
 func (s *ListCustomersToAdvanceSuite) TestReturnsOnlyEligibleCustomers() {

openmeter/billing/charges/flatfee/service/creditheninvoice.go

@@ -15,6 +15,7 @@ import (
 	"github.com/openmeterio/openmeter/openmeter/billing/charges/meta"
 	"github.com/openmeterio/openmeter/openmeter/billing/charges/models/payment"
 	"github.com/openmeterio/openmeter/openmeter/productcatalog"
+	"github.com/openmeterio/openmeter/pkg/clock"
 	"github.com/openmeterio/openmeter/pkg/models"
 	"github.com/openmeterio/openmeter/pkg/statelessx"
 	"github.com/openmeterio/openmeter/pkg/timeutil"
@@ -26,10 +27,12 @@ type CreditThenInvoiceStateMachine struct {
 
 type periodPatch interface {
 	Op() meta.PatchType
+	GetTarget() meta.ChangeTarget
 	GetNewServicePeriodTo() time.Time
 	GetNewFullServicePeriodTo() time.Time
 	GetNewBillingPeriodTo() time.Time
 	GetNewInvoiceAt() time.Time
+	ValidateWith(meta.IntentMutableFields) error
 }
 
 var (
@@ -77,7 +80,7 @@ func (s *CreditThenInvoiceStateMachine) configureStates() {
 			flatfee.StatusActive,
 			statelessx.BoolFn(s.IsInsideServicePeriodAndNonZeroAmount),
 		).
-		Permit(meta.TriggerDelete, flatfee.StatusDeleted).
+		InternalTransition(meta.TriggerDelete, statelessx.WithParameters(s.DeleteCharge)).
 		InternalTransition(meta.TriggerExtend, statelessx.WithParameters(s.ExtendCharge)).
 		InternalTransition(meta.TriggerShrink, statelessx.WithParameters(s.ShrinkCharge)).
 		OnActive(s.AdvanceAfterServicePeriodFrom)
@@ -88,63 +91,77 @@ func (s *CreditThenInvoiceStateMachine) configureStates() {
 		// operational state.
 		Permit(meta.TriggerNext, flatfee.StatusFinal, statelessx.BoolFn(s.IsZeroAmount)).
 		Permit(meta.TriggerFinalInvoiceCreated, flatfee.StatusActiveRealizationStarted).
-		Permit(meta.TriggerDelete, flatfee.StatusDeleted).
+		InternalTransition(meta.TriggerDelete, statelessx.WithParameters(s.DeleteCharge)).
 		InternalTransition(meta.TriggerExtend, statelessx.WithParameters(s.ExtendCharge)).
 		InternalTransition(meta.TriggerShrink, statelessx.WithParameters(s.ShrinkCharge)).
 		OnActive(s.AdvanceAfterServicePeriodTo)
 
 	s.Configure(flatfee.StatusActiveRealizationStarted).
 		Permit(meta.TriggerNext, flatfee.StatusActiveRealizationWaitingForCollection).
-		Permit(meta.TriggerDelete, flatfee.StatusDeleted).
+		InternalTransition(meta.TriggerDelete, statelessx.WithParameters(s.DeleteCharge)).
 		InternalTransition(meta.TriggerExtend, statelessx.WithParameters(s.ExtendCharge)).
 		InternalTransition(meta.TriggerShrink, statelessx.WithParameters(s.ShrinkCharge)).
 		OnEntryFrom(meta.TriggerFinalInvoiceCreated, statelessx.WithParameters(s.StartRealization))
 
 	s.Configure(flatfee.StatusActiveRealizationWaitingForCollection).
 		Permit(meta.TriggerCollectionCompleted, flatfee.StatusActiveRealizationProcessing).
-		Permit(meta.TriggerDelete, flatfee.StatusDeleted).
+		InternalTransition(meta.TriggerDelete, statelessx.WithParameters(s.DeleteCharge)).
 		InternalTransition(meta.TriggerExtend, statelessx.WithParameters(s.ExtendCharge)).
 		InternalTransition(meta.TriggerShrink, statelessx.WithParameters(s.ShrinkCharge))
 
 	s.Configure(flatfee.StatusActiveRealizationProcessing).
 		Permit(meta.TriggerInvoiceIssued, flatfee.StatusActiveRealizationIssuing).
-		Permit(meta.TriggerDelete, flatfee.StatusDeleted).
+		InternalTransition(meta.TriggerDelete, statelessx.WithParameters(s.DeleteCharge)).
 		InternalTransition(meta.TriggerExtend, statelessx.WithParameters(s.ExtendCharge)).
 		InternalTransition(meta.TriggerShrink, statelessx.WithParameters(s.ShrinkCharge))
 
 	s.Configure(flatfee.StatusActiveRealizationIssuing).
 		Permit(meta.TriggerNext, flatfee.StatusActiveRealizationCompleted).
-		Permit(meta.TriggerDelete, flatfee.StatusDeleted).
+		InternalTransition(meta.TriggerDelete, statelessx.WithParameters(s.DeleteCharge)).
 		InternalTransition(meta.TriggerExtend, statelessx.WithParameters(s.UnsupportedExtendOperation)).
 		InternalTransition(meta.TriggerShrink, statelessx.WithParameters(s.UnsupportedShrinkOperation)).
 		OnEntryFrom(meta.TriggerInvoiceIssued, statelessx.WithParameters(s.AccrueInvoiceUsage))
 
 	s.Configure(flatfee.StatusActiveRealizationCompleted).
 		Permit(meta.TriggerNext, flatfee.StatusActiveAwaitingPaymentSettlement).
-		Permit(meta.TriggerDelete, flatfee.StatusDeleted).
+		InternalTransition(meta.TriggerDelete, statelessx.WithParameters(s.DeleteCharge)).
 		InternalTransition(meta.TriggerExtend, statelessx.WithParameters(s.UnsupportedExtendOperation)).
 		InternalTransition(meta.TriggerShrink, statelessx.WithParameters(s.UnsupportedShrinkOperation))
 
 	s.Configure(flatfee.StatusActiveAwaitingPaymentSettlement).
 		Permit(meta.TriggerNext, flatfee.StatusFinal, statelessx.BoolFn(s.AreAllPaymentsSettled)).
 		Permit(meta.TriggerAllPaymentsSettled, flatfee.StatusFinal, statelessx.BoolFn(s.AreAllPaymentsSettled)).
-		Permit(meta.TriggerDelete, flatfee.StatusDeleted).
+		InternalTransition(meta.TriggerDelete, statelessx.WithParameters(s.DeleteCharge)).
 		InternalTransition(meta.TriggerExtend, statelessx.WithParameters(s.ExtendCharge)).
 		InternalTransition(meta.TriggerShrink, statelessx.WithParameters(s.ShrinkCharge))
 
 	s.Configure(flatfee.StatusFinal).
-		Permit(meta.TriggerDelete, flatfee.StatusDeleted).
+		InternalTransition(meta.TriggerDelete, statelessx.WithParameters(s.DeleteCharge)).
 		InternalTransition(meta.TriggerExtend, statelessx.WithParameters(s.ExtendCharge)).
 		InternalTransition(meta.TriggerShrink, statelessx.WithParameters(s.ShrinkCharge)).
 		OnActive(s.ClearAdvanceAfter)
 
 	s.Configure(flatfee.StatusDeleted).
 		InternalTransition(meta.TriggerExtend, statelessx.WithParameters(s.UnsupportedExtendOperation)).
-		InternalTransition(meta.TriggerShrink, statelessx.WithParameters(s.UnsupportedShrinkOperation)).
-		OnEntry(statelessx.WithParameters(s.DeleteCharge))
+		InternalTransition(meta.TriggerShrink, statelessx.WithParameters(s.UnsupportedShrinkOperation))
 }
 
-func (s *CreditThenInvoiceStateMachine) DeleteCharge(ctx context.Context, _ meta.PatchDeletePolicy) error {
+func (s *CreditThenInvoiceStateMachine) DeleteCharge(ctx context.Context, patch meta.PatchDelete) error {
+	if err := s.Charge.Intent.Mutate(patch.GetTarget(), func(fields flatfee.IntentMutableFields) (flatfee.IntentMutableFields, error) {
+		fields.IntentDeletedAt = lo.ToPtr(clock.Now())
+		return fields, nil
+	}); err != nil {
+		return fmt.Errorf("mutating %s intent deleted at: %w", patch.GetTarget(), err)
+	}
+
+	if patch.GetTarget() == meta.ChangeTargetBase && s.Charge.Intent.HasOverrideLayer() {
+		// Subscription sync targets the base intent. When an override is active,
+		// the customer-facing charge and invoice history remain owned by the override.
+		return nil
+	}
+
+	s.Charge.Status = flatfee.StatusDeleted
+
 	patches := []invoiceupdater.Patch{
 		invoiceupdater.NewDeleteGatheringLineByChargeIDPatch(s.Charge.ID),
 	}
@@ -180,43 +197,57 @@ func (s *CreditThenInvoiceStateMachine) DeleteCharge(ctx context.Context, _ meta
 }
 
 func (s *CreditThenInvoiceStateMachine) ExtendCharge(ctx context.Context, patch meta.PatchExtend) error {
-	if err := patch.ValidateWith(s.Charge.Intent.GetEffectiveMetaIntentMutableFields()); err != nil {
-		return fmt.Errorf("validate extend patch: %w", err)
-	}
-
 	invoicingStateInput, err := s.applyPeriodPatch(patch)
 	if err != nil {
 		return err
 	}
 
+	if !invoicingStateInput.ShouldReconcile {
+		return nil
+	}
+
 	return s.reconcileInvoicingState(ctx, invoicingStateInput)
 }
 
 func (s *CreditThenInvoiceStateMachine) ShrinkCharge(ctx context.Context, patch meta.PatchShrink) error {
-	if err := patch.ValidateWith(s.Charge.Intent.GetEffectiveMetaIntentMutableFields()); err != nil {
-		return fmt.Errorf("validate shrink patch: %w", err)
-	}
-
 	invoicingStateInput, err := s.applyPeriodPatch(patch)
 	if err != nil {
 		return err
 	}
 
+	if !invoicingStateInput.ShouldReconcile {
+		return nil
+	}
+
 	return s.reconcileInvoicingState(ctx, invoicingStateInput)
 }
 
 func (s *CreditThenInvoiceStateMachine) applyPeriodPatch(patch periodPatch) (reconcileInvoicingStateInput, error) {
-	oldAmountAfterProration := s.Charge.State.AmountAfterProration
+	targetIntent, err := s.Charge.Intent.GetIntentForTarget(patch.GetTarget())
+	if err != nil {
+		return reconcileInvoicingStateInput{}, fmt.Errorf("getting %s intent: %w", patch.GetTarget(), err)
+	}
 
+	if err := patch.ValidateWith(targetIntent.IntentMutableFields.IntentMutableFields); err != nil {
+		return reconcileInvoicingStateInput{}, fmt.Errorf("validate %s patch: %w", patch.Op(), err)
+	}
 	intent := s.Charge.Intent
-	if err := intent.MutateEffective(func(fields flatfee.IntentMutableFields) (flatfee.IntentMutableFields, error) {
+	if err := intent.Mutate(patch.GetTarget(), func(fields flatfee.IntentMutableFields) (flatfee.IntentMutableFields, error) {
 		fields.ServicePeriod.To = patch.GetNewServicePeriodTo()
 		fields.FullServicePeriod.To = patch.GetNewFullServicePeriodTo()
 		fields.BillingPeriod.To = patch.GetNewBillingPeriodTo()
 		fields.InvoiceAt = patch.GetNewInvoiceAt()
 		return fields, nil
 	}); err != nil {
-		return reconcileInvoicingStateInput{}, fmt.Errorf("mutating effective intent: %w", err)
+		return reconcileInvoicingStateInput{}, fmt.Errorf("mutating %s intent: %w", patch.GetTarget(), err)
+	}
+
+	s.Charge.Intent = intent
+
+	if patch.GetTarget() == meta.ChangeTargetBase && s.Charge.Intent.HasOverrideLayer() {
+		// Subscription sync targets the base intent. When an override is active,
+		// the customer-facing invoice remains owned by the override layer.
+		return reconcileInvoicingStateInput{}, nil
 	}
 
 	amountAfterProration, err := intent.CalculateAmountAfterProration()
@@ -225,10 +256,11 @@ func (s *CreditThenInvoiceStateMachine) applyPeriodPatch(patch periodPatch) (rec
 	}
 
 	return reconcileInvoicingStateInput{
+		ShouldReconcile:         true,
 		Op:                      patch.Op(),
 		Period:                  intent.GetEffectiveServicePeriod(),
 		Intent:                  intent,
-		OldAmountAfterProration: oldAmountAfterProration,
+		OldAmountAfterProration: s.Charge.State.AmountAfterProration,
 		NewAmountAfterProration: amountAfterProration,
 	}, nil
 }
@@ -304,6 +336,7 @@ func (s *CreditThenInvoiceStateMachine) AreAllPaymentsSettled() bool {
 }
 
 type reconcileInvoicingStateInput struct {
+	ShouldReconcile         bool
 	Op                      meta.PatchType
 	Period                  timeutil.ClosedPeriod
 	Intent                  flatfee.OverridableIntent