feat: add extend support for usage-based charges [credit_then_invoice settlement mode] (#4333)

Author: turip

.agents/skills/charges/SKILL.md

@@ -113,6 +113,8 @@ Do not emulate every billing rating mutator in the line mapper. Usage discounts
 Detailed-line expansion rules:
 
 - `meta.ExpandDetailedLines` is not standalone for charge reads; it requires `meta.ExpandRealizations`
+- usage-based deleted realization runs are hidden from `meta.ExpandRealizations` by default because their invoice/ledger effect has been cleaned up; request `meta.ExpandDeletedRealizations` together with `meta.ExpandRealizations` only when a caller must inspect cleanup state, such as frontend audit views or deletion tests
+- production code that sums or interprets realization history must explicitly skip runs with `DeletedAt != nil`, documenting the business reason at the guard, because deleted realizations are no longer effective billable history
 - usage-based detailed lines live under `RealizationRun.DetailedLines`
 - flat-fee detailed lines also live under `Charge.Realizations.DetailedLines`, not on the root charge
 - `mo.None()` means detailed lines were not expanded; present options mean expanded data, even when the underlying slice is nil/empty
@@ -149,6 +151,18 @@ Important rules:
 - usage-based invoice branches should read in this order: `started -> waiting_for_collection -> processing -> issuing -> completed`, then auto-advance out of the branch. Keep `invoice_issued` as the boundary between `processing` and `issuing`, run `FinalizeInvoiceRun(...)` from the `issuing` state, and let `completed` be the last branch-local status before `next` returns a partial invoice to `active` or moves a final invoice to `active.awaiting_payment_settlement`
 - when adding or renaming usage-based detailed statuses, remember that `status_detailed` is an Ent enum for `ChargeUsageBased`; run `make generate` so the generated enum validators and migrate schema include the new values before trusting state-machine changes
 
+Usage-based credit-then-invoice extension rules:
+
+- `PatchExtend` must represent a real extension: the new service period end must be after the persisted intent end; full service period and billing period ends may stay unchanged but must not move backwards.
+- Do not stretch an in-progress final realization run to cover the extension. There can only be one active run, and stretching the old final run would keep the run open across the extension and delay the standard invoice lifecycle.
+- If the current final realization run is still backed by a mutable invoice line, the extend flow should update the charge intent, emit an invoice-line delete patch for that mutable standard line, and let the billing invoice updater/line engine delete the invoice line and mark the run deleted. The charge should move back to `active` with `AdvanceAfter` set to the new service-period end so the extended tail can realize later.
+- While a mutable final invoice run is before invoice issuing (`active.final_realization.started`, `active.final_realization.waiting_for_collection`, or `active.final_realization.processing`), billing remains the owner of the ongoing invoice lifecycle. Extension may delete the mutable line, mark the current run deleted, and recreate a gathering line; a direct `AdvanceCharges(...)` before the new service-period end must be a no-op, and the replacement final run should be created by billing when the replacement gathering line is invoiced at the extended end.
+- Once a final invoice run reaches `active.final_realization.issuing` or `active.final_realization.completed`, extend is explicitly rejected by `UnsupportedExtendOperation` because invoice lifecycle callbacks or state-machine advancement still own those states. Subscription sync is expected to retry instead of moving the charge out of those states manually.
+- Extending from `active.awaiting_payment_settlement` is allowed: preserve the invoice line and ledger bookings, reclassify the old final run as partial, move the charge back to `active`, and create only a tail gathering line.
+- If the old final realization has already passed into immutable invoice territory, keep the invoice and ledger untouched. Reclassify the old final run as a partial invoice run and move the charge back to `active`; the extended tail will produce a new final run later. Immutable invoice cleanup should be surfaced as validation warnings by billing rather than reversing ledger bookings.
+- Pending gathering lines for the charge should be extended in place by charge ID. Existing standard invoice lines should only be deleted in the mutable-current-final case above.
+- These rules are intentionally about not blocking the invoice train: extension should not hold a standard invoice lifecycle open while waiting for newly extended usage windows to finish.
+
 Operational consequence:
 
 - adding a new charge engine enum without a registered implementation causes invoice collection to fail when billing resolves the line engine

.agents/skills/subscriptionsync/SKILL.md

@@ -142,11 +142,11 @@ Important routing rules:
 
 The `filterInScopeLines` function gates which target items enter reconciliation. It filters out non-billable items for every backend, and only invoicing-backed targets are additionally gated on `GetExpectedLine()`. This runs before any diffing so absent targets naturally produce delete/no-op outcomes.
 
-Current charge limitations:
-- charge collections only support create
-- delete/shrink/extend/prorate return unsupported errors
-
-This is intentional. If a test expects credit-only cancellation to fail on delete, that is current behavior, not a bug in the test.
+Charge-backed mutation notes:
+- charge collections support create/delete and period-shape changes
+- shrink and most extend operations are emitted as an emulated replacement: delete the existing charge and create a replacement charge from the target state
+- usage-based `credit_then_invoice` extend is emitted as a native charge patch so the usage-based state machine can preserve existing immutable invoice/ledger state and keep the invoice train unblocked
+- charge-backed explicit prorate still returns unsupported; charge domains own their own proration/materialization behavior
 
 ## Invoice vs Charge Semantics
 
@@ -160,8 +160,10 @@ Keep these separate:
 - Charge sync:
   - provisions charge intents directly
   - does not go through invoice-line rendering
-  - currently only create is supported
-  - invoice-style semantic proration is skipped in `diffItem(...)`; charges own their own proration logic
+  - supports create/delete and period-shape changes
+  - emits shrink and most extend operations as delete+create replacements
+  - emits usage-based `credit_then_invoice` extend as a native charge patch
+  - keeps explicit prorate unsupported because charges own their own proration logic
 
 Do not force charge behavior through invoice abstractions.
 

openmeter/billing/charges/flatfee/adapter.go

@@ -150,5 +150,9 @@ func validateExpands(expands meta.Expands) error {
 		return fmt.Errorf("%q requires %q", meta.ExpandDetailedLines, meta.ExpandRealizations)
 	}
 
+	if expands.Has(meta.ExpandDeletedRealizations) && !expands.Has(meta.ExpandRealizations) {
+		return fmt.Errorf("%q requires %q", meta.ExpandDeletedRealizations, meta.ExpandRealizations)
+	}
+
 	return nil
 }

openmeter/billing/charges/invoiceupdater/invoiceupdate.go

@@ -44,6 +44,10 @@ func (u *Updater) ApplyPatches(ctx context.Context, customerID customer.Customer
 		return fmt.Errorf("resolving gathering line deletes by charge ID: %w", err)
 	}
 
+	if err := u.resolveGatheringLineUpdatesByChargeID(ctx, customerID, &patchesParsed); err != nil {
+		return fmt.Errorf("resolving gathering line updates by charge ID: %w", err)
+	}
+
 	invoicesByID, err := u.listInvoicesByID(ctx, customerID.Namespace, lo.Keys(patchesParsed.updatedLinesByInvoiceID))
 	if err != nil {
 		return fmt.Errorf("listing invoices: %w", err)
@@ -203,6 +207,7 @@ type patchesParsed struct {
 	updatedLinesByInvoiceID map[string]invoicePatches
 
 	gatheringLineDeletesByChargeID []string
+	gatheringLineUpdatesByChargeID map[string]PatchUpdateGatheringLineByChargeID
 }
 
 type invoicePatches struct {
@@ -212,7 +217,8 @@ type invoicePatches struct {
 
 func (u *Updater) parsePatches(patches []Patch) (patchesParsed, error) {
 	parsed := patchesParsed{
-		updatedLinesByInvoiceID: make(map[string]invoicePatches),
+		updatedLinesByInvoiceID:        make(map[string]invoicePatches),
+		gatheringLineUpdatesByChargeID: make(map[string]PatchUpdateGatheringLineByChargeID),
 	}
 
 	for _, patch := range patches {
@@ -249,6 +255,13 @@ func (u *Updater) parsePatches(patches []Patch) (patchesParsed, error) {
 			}
 
 			parsed.gatheringLineDeletesByChargeID = append(parsed.gatheringLineDeletesByChargeID, deletePatch.ChargeID)
+		case PatchOpUpdateGatheringLineByChargeID:
+			updatePatch, err := patch.AsUpdateGatheringLineByChargeIDPatch()
+			if err != nil {
+				return patchesParsed{}, fmt.Errorf("getting gathering line update: %w", err)
+			}
+
+			parsed.gatheringLineUpdatesByChargeID[updatePatch.ChargeID] = updatePatch
 		default:
 			return patchesParsed{}, fmt.Errorf("unexpected patch operation: %s", patch.Op())
 		}
@@ -320,6 +333,53 @@ func (u *Updater) resolveGatheringLineDeletesByChargeID(ctx context.Context, cus
 	return nil
 }
 
+func (u *Updater) resolveGatheringLineUpdatesByChargeID(ctx context.Context, customerID customer.CustomerID, parsed *patchesParsed) error {
+	if len(parsed.gatheringLineUpdatesByChargeID) == 0 {
+		return nil
+	}
+
+	invoices, err := u.billingService.ListGatheringInvoices(ctx, billing.ListGatheringInvoicesInput{
+		Namespaces: []string{customerID.Namespace},
+		Customers:  []string{customerID.ID},
+		Expand: billing.GatheringInvoiceExpands{
+			billing.GatheringInvoiceExpandLines,
+		},
+	})
+	if err != nil {
+		return fmt.Errorf("listing gathering invoices: %w", err)
+	}
+
+	for _, invoice := range invoices.Items {
+		for _, line := range invoice.Lines.OrEmpty() {
+			if line.DeletedAt != nil || line.ChargeID == nil {
+				continue
+			}
+
+			updatePatch, ok := parsed.gatheringLineUpdatesByChargeID[*line.ChargeID]
+			if !ok {
+				continue
+			}
+
+			line.ServicePeriod.To = updatePatch.ServicePeriodTo
+			line.InvoiceAt = updatePatch.ServicePeriodTo
+			if line.Subscription != nil {
+				line.Subscription.BillingPeriod.To = updatePatch.ServicePeriodTo
+			}
+
+			genericLine, err := line.AsInvoiceLine().AsGenericLine()
+			if err != nil {
+				return fmt.Errorf("converting gathering line[%s] to generic line: %w", line.ID, err)
+			}
+
+			lineUpdates := parsed.updatedLinesByInvoiceID[invoice.ID]
+			lineUpdates.updatedLines = append(lineUpdates.updatedLines, genericLine)
+			parsed.updatedLinesByInvoiceID[invoice.ID] = lineUpdates
+		}
+	}
+
+	return nil
+}
+
 func (u *Updater) updateMutableStandardInvoice(ctx context.Context, invoice billing.StandardInvoice, linePatches invoicePatches) error {
 	updatedInvoice, err := u.billingService.UpdateStandardInvoice(ctx, billing.UpdateStandardInvoiceInput{
 		Invoice:             invoice.GetInvoiceID(),

openmeter/billing/charges/invoiceupdater/patch.go

@@ -3,6 +3,7 @@ package invoiceupdater
 import (
 	"fmt"
 	"log/slog"
+	"time"
 
 	"github.com/openmeterio/openmeter/openmeter/billing"
 )
@@ -14,6 +15,7 @@ const (
 	PatchOpLineDelete                    PatchOperation = "line_delete"
 	PatchOpLineUpdate                    PatchOperation = "line_update"
 	PatchOpDeleteGatheringLineByChargeID PatchOperation = "delete_gathering_line_by_charge_id"
+	PatchOpUpdateGatheringLineByChargeID PatchOperation = "update_gathering_line_by_charge_id"
 )
 
 type PatchLineCreate struct {
@@ -33,13 +35,19 @@ type PatchDeleteGatheringLineByChargeID struct {
 	ChargeID string
 }
 
+type PatchUpdateGatheringLineByChargeID struct {
+	ChargeID        string
+	ServicePeriodTo time.Time
+}
+
 type Patch struct {
 	op PatchOperation
 
 	createLinePatch                    PatchLineCreate
 	deleteLinePatch                    PatchLineDelete
 	updateLinePatch                    PatchLineUpdate
 	deleteGatheringLineByChargeIDPatch PatchDeleteGatheringLineByChargeID
+	updateGatheringLineByChargeIDPatch PatchUpdateGatheringLineByChargeID
 }
 
 func (p Patch) Op() PatchOperation {
@@ -78,6 +86,14 @@ func (p Patch) AsDeleteGatheringLineByChargeIDPatch() (PatchDeleteGatheringLineB
 	return p.deleteGatheringLineByChargeIDPatch, nil
 }
 
+func (p Patch) AsUpdateGatheringLineByChargeIDPatch() (PatchUpdateGatheringLineByChargeID, error) {
+	if p.op != PatchOpUpdateGatheringLineByChargeID {
+		return PatchUpdateGatheringLineByChargeID{}, fmt.Errorf("expected update gathering line by charge ID patch, got %s", p.op)
+	}
+
+	return p.updateGatheringLineByChargeIDPatch, nil
+}
+
 func NewDeleteLinePatch(lineID billing.LineID, invoiceID string) Patch {
 	return Patch{
 		op: PatchOpLineDelete,
@@ -97,6 +113,16 @@ func NewDeleteGatheringLineByChargeIDPatch(chargeID string) Patch {
 	}
 }
 
+func NewUpdateGatheringLineByChargeIDPatch(chargeID string, servicePeriodTo time.Time) Patch {
+	return Patch{
+		op: PatchOpUpdateGatheringLineByChargeID,
+		updateGatheringLineByChargeIDPatch: PatchUpdateGatheringLineByChargeID{
+			ChargeID:        chargeID,
+			ServicePeriodTo: servicePeriodTo,
+		},
+	}
+}
+
 func NewUpdateLinePatch(line billing.GenericInvoiceLine) Patch {
 	return Patch{
 		op: PatchOpLineUpdate,
@@ -125,6 +151,8 @@ func (p Patch) Log(logger *slog.Logger) {
 		logger.Info("update line patch", "line_id", p.updateLinePatch.TargetState.GetLineID().ID, "invoice_id", p.updateLinePatch.TargetState.GetInvoiceID(), "new_service_period_from", p.updateLinePatch.TargetState.GetServicePeriod().From, "new_service_period_to", p.updateLinePatch.TargetState.GetServicePeriod().To, "unique_reference_id", p.updateLinePatch.TargetState.GetChildUniqueReferenceID())
 	case PatchOpDeleteGatheringLineByChargeID:
 		logger.Info("delete gathering line by charge id patch", "charge_id", p.deleteGatheringLineByChargeIDPatch.ChargeID)
+	case PatchOpUpdateGatheringLineByChargeID:
+		logger.Info("update gathering line by charge id patch", "charge_id", p.updateGatheringLineByChargeIDPatch.ChargeID, "new_service_period_to", p.updateGatheringLineByChargeIDPatch.ServicePeriodTo)
 	default:
 		logger.Info("unknown patch operation", "operation", p.op)
 	}

openmeter/billing/charges/meta/charge.go

@@ -69,16 +69,18 @@ func (t ChargeType) Validate() error {
 type Expand string
 
 const (
-	ExpandRealizations  Expand = "realizations"
-	ExpandRealtimeUsage Expand = "realtime_usage"
-	ExpandDetailedLines Expand = "detailed_lines"
+	ExpandRealizations        Expand = "realizations"
+	ExpandRealtimeUsage       Expand = "realtime_usage"
+	ExpandDetailedLines       Expand = "detailed_lines"
+	ExpandDeletedRealizations Expand = "deleted_realizations"
 )
 
 func (e Expand) Values() []Expand {
 	return []Expand{
 		ExpandRealizations,
 		ExpandRealtimeUsage,
 		ExpandDetailedLines,
+		ExpandDeletedRealizations,
 	}
 }
 

openmeter/billing/charges/meta/patchextend.go

@@ -110,8 +110,8 @@ func (p PatchExtend) ValidateWith(intent Intent) error {
 		errs = append(errs, err)
 	}
 
-	if p.GetNewServicePeriodTo().Before(intent.ServicePeriod.To) {
-		errs = append(errs, fmt.Errorf("new service period to must be greater than or equal to existing service period to"))
+	if !p.GetNewServicePeriodTo().After(intent.ServicePeriod.To) {
+		errs = append(errs, fmt.Errorf("new service period to must be greater than existing service period to"))
 	}
 
 	if p.GetNewFullServicePeriodTo().Before(intent.FullServicePeriod.To) {

openmeter/billing/charges/meta/patchextend_test.go

@@ -0,0 +1,100 @@
+package meta
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/openmeterio/openmeter/pkg/timeutil"
+)
+
+func TestPatchExtendValidateWith(t *testing.T) {
+	base := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)
+
+	intent := Intent{
+		ServicePeriod: timeutil.ClosedPeriod{
+			From: base,
+			To:   base.AddDate(0, 1, 0),
+		},
+		FullServicePeriod: timeutil.ClosedPeriod{
+			From: base,
+			To:   base.AddDate(0, 2, 0),
+		},
+		BillingPeriod: timeutil.ClosedPeriod{
+			From: base,
+			To:   base.AddDate(0, 3, 0),
+		},
+	}
+
+	tests := []struct {
+		name    string
+		patch   PatchExtend
+		wantErr bool
+	}{
+		{
+			name: "allows service period extension with unchanged full service and billing periods",
+			patch: mustNewPatchExtend(t, NewPatchExtendInput{
+				NewServicePeriodTo:     intent.ServicePeriod.To.Add(time.Hour),
+				NewFullServicePeriodTo: intent.FullServicePeriod.To,
+				NewBillingPeriodTo:     intent.BillingPeriod.To,
+			}),
+		},
+		{
+			name: "rejects unchanged service period end",
+			patch: mustNewPatchExtend(t, NewPatchExtendInput{
+				NewServicePeriodTo:     intent.ServicePeriod.To,
+				NewFullServicePeriodTo: intent.FullServicePeriod.To,
+				NewBillingPeriodTo:     intent.BillingPeriod.To,
+			}),
+			wantErr: true,
+		},
+		{
+			name: "rejects earlier service period end",
+			patch: mustNewPatchExtend(t, NewPatchExtendInput{
+				NewServicePeriodTo:     intent.ServicePeriod.To.Add(-time.Hour),
+				NewFullServicePeriodTo: intent.FullServicePeriod.To,
+				NewBillingPeriodTo:     intent.BillingPeriod.To,
+			}),
+			wantErr: true,
+		},
+		{
+			name: "rejects earlier full service period end",
+			patch: mustNewPatchExtend(t, NewPatchExtendInput{
+				NewServicePeriodTo:     intent.ServicePeriod.To.Add(time.Hour),
+				NewFullServicePeriodTo: intent.FullServicePeriod.To.Add(-time.Hour),
+				NewBillingPeriodTo:     intent.BillingPeriod.To,
+			}),
+			wantErr: true,
+		},
+		{
+			name: "rejects earlier billing period end",
+			patch: mustNewPatchExtend(t, NewPatchExtendInput{
+				NewServicePeriodTo:     intent.ServicePeriod.To.Add(time.Hour),
+				NewFullServicePeriodTo: intent.FullServicePeriod.To,
+				NewBillingPeriodTo:     intent.BillingPeriod.To.Add(-time.Hour),
+			}),
+			wantErr: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := tt.patch.ValidateWith(intent)
+			if tt.wantErr {
+				require.Error(t, err)
+				return
+			}
+
+			require.NoError(t, err)
+		})
+	}
+}
+
+func mustNewPatchExtend(t *testing.T, input NewPatchExtendInput) PatchExtend {
+	t.Helper()
+
+	patch, err := NewPatchExtend(input)
+	require.NoError(t, err)
+	return patch
+}

openmeter/billing/charges/service/patch.go

@@ -18,12 +18,6 @@ func (s *service) ApplyPatches(ctx context.Context, input charges.ApplyPatchesIn
 	}
 
 	return transaction.RunWithNoValue(ctx, s.adapter, func(ctx context.Context) error {
-		// TODO[later]: Remove this once we have proper shrink/extend patching in place.
-		input, err := s.tmpMapShrinkExtendToCreateDelete(ctx, input)
-		if err != nil {
-			return err
-		}
-
 		if err := s.applyPatches(ctx, input.CustomerID, input.PatchesByChargeID); err != nil {
 			return err
 		}