feat(ledger): add accounting diagnostics and balance telemetry

app/common/billing.go

@@ -250,6 +250,7 @@ func NewBillingRegistry(
 
 	if creditsConfig.Enabled {
 		chargesRegistry, err = newChargesRegistry(
+			metricMeter,
 			db,
 			logger,
 			locker,

app/common/charges.go

@@ -4,6 +4,8 @@ import (
 	"fmt"
 	"log/slog"
 
+	"go.opentelemetry.io/otel/metric"
+
 	"github.com/openmeterio/openmeter/app/config"
 	"github.com/openmeterio/openmeter/openmeter/billing"
 	"github.com/openmeterio/openmeter/openmeter/billing/charges"
@@ -64,6 +66,7 @@ func NewChargesMetaAdapter(
 }
 
 func NewChargesCollectorService(
+	metricMeter metric.Meter,
 	logger *slog.Logger,
 	db *entdb.Client,
 	ledgerService ledger.Ledger,
@@ -74,6 +77,7 @@ func NewChargesCollectorService(
 	breakageService ledgerbreakage.Service,
 ) (ledgercollector.Service, error) {
 	collectorService, err := ledgercollector.NewService(ledgercollector.Config{
+		Meter:   metricMeter,
 		Logger:  logger,
 		Advance: advanceService,
 		Ledger:  ledgerService,
@@ -435,6 +439,7 @@ func NewRecognizerService(
 // newChargesRegistry constructs the full charges stack.
 // Private: must be initialized via NewBillingRegistry.
 func newChargesRegistry(
+	metricMeter metric.Meter,
 	db *entdb.Client,
 	logger *slog.Logger,
 	locker *lockr.Locker,
@@ -479,6 +484,7 @@ func newChargesRegistry(
 
 	transactionManager := enttx.NewCreator(db)
 	collectorService, err := ledgercollector.NewService(ledgercollector.Config{
+		Meter:   metricMeter,
 		Logger:  logger,
 		Advance: advanceService,
 		Ledger:  ledgerService,

openmeter/billing/README.md

@@ -200,6 +200,11 @@ earlier attempt remain available to idempotent retry handling.
 
 ## Validation issues alongside successful results
 
+`models.Error` carries codes and diagnostic attributes but remains a system error.
+`models.ValidationIssue` explicitly opts into validation handling. Their shared
+metadata does not make them interchangeable; only an explicit
+`WrapAsValidationIssue` conversion changes a system error into a billing issue.
+
 Typed validation issues can accompany usable results through an `error` return.
 Classify that error before discarding the result. Its meaning depends on the
 boundary:

openmeter/billing/charges/service/base_test.go

@@ -9,6 +9,7 @@ import (
 	"github.com/alpacahq/alpacadecimal"
 	"github.com/invopop/gobl/currency"
 	"github.com/samber/lo"
+	metricnoop "go.opentelemetry.io/otel/metric/noop"
 
 	"github.com/openmeterio/openmeter/openmeter/billing"
 	"github.com/openmeterio/openmeter/openmeter/billing/charges"
@@ -187,6 +188,7 @@ func (s *BaseSuite) SetupSuite() {
 		advanceService := advancetestutils.NewService(s.T(), ledgerDeps, breakageService)
 
 		collectorService, err := ledgercollector.NewService(ledgercollector.Config{
+			Meter:              metricnoop.NewMeterProvider().Meter("test"),
 			Logger:             omtestutils.NewDiscardLogger(s.T()),
 			Advance:            advanceService,
 			Ledger:             ledgerDeps.HistoricalLedger,

openmeter/billing/validationissue.go

@@ -529,14 +529,33 @@ func toValidationIssue(err error, fieldPrefix string, component ComponentName, m
 				Attributes: mergedAttributes,
 			},
 		}, nil
+	case models.Error:
+		// Generic errors remain system errors unless explicitly converted with WrapAsValidationIssue.
+		if ordinaryErrorSeverity == "" {
+			return nil, err
+		}
+
+		// Conversion is explicit at this boundary; keep the structured context.
+		severity := models.ErrorSeverityCritical
+		if ordinaryErrorSeverity == ValidationIssueSeverityWarning {
+			severity = models.ErrorSeverityWarning
+		}
+
+		return modelErrorAsBillingValidationIssue(errT.WithSeverity(severity), fieldPrefix, component, messagePrefix, attributes)
+	case *models.Error:
+		if errT == nil {
+			return nil, nil
+		}
+
+		return toValidationIssue(*errT, fieldPrefix, component, messagePrefix, attributes, ordinaryErrorSeverity)
 	case models.ValidationIssue:
-		return modelValidationIssueAsBillingValidationIssue(errT, fieldPrefix, component, messagePrefix, attributes)
+		return modelErrorAsBillingValidationIssue(models.Error(errT), fieldPrefix, component, messagePrefix, attributes)
 	case *models.ValidationIssue:
 		if errT == nil {
 			return nil, nil
 		}
 
-		return modelValidationIssueAsBillingValidationIssue(*errT, fieldPrefix, component, messagePrefix, attributes)
+		return modelErrorAsBillingValidationIssue(models.Error(*errT), fieldPrefix, component, messagePrefix, attributes)
 	}
 
 	switch errT := err.(type) {
@@ -578,7 +597,7 @@ func toValidationIssue(err error, fieldPrefix string, component ComponentName, m
 	}
 }
 
-func modelValidationIssueAsBillingValidationIssue(issue models.ValidationIssue, fieldPrefix string, component ComponentName, messagePrefix string, attributes models.Annotations) ([]ValidationIssue, error) {
+func modelErrorAsBillingValidationIssue(issue models.Error, fieldPrefix string, component ComponentName, messagePrefix string, attributes models.Annotations) ([]ValidationIssue, error) {
 	issueComponent := component
 	if issueComponent == "" {
 		issueComponent = ComponentName(issue.Component())

openmeter/billing/validationissue_test.go

@@ -872,3 +872,43 @@ func TestValidationWithMessage(t *testing.T) {
 		require.NoError(t, ValidationWithMessage(nil, "unused", models.Attributes{"unused": true}))
 	})
 }
+
+func TestStructuredSystemErrorClassification(t *testing.T) {
+	validation := models.NewValidationIssue("invoice_invalid", "invoice cannot proceed")
+	system := models.NewError("ledger_failed", "accounting failed", models.WithCause(validation)).
+		WithAttr("reason", "missing_source")
+
+	for _, tc := range []struct {
+		name string
+		err  error
+	}{
+		{name: "value", err: system},
+		{name: "pointer", err: &system},
+		{name: "formatted wrapper", err: fmt.Errorf("collect: %w", system)},
+		{name: "component wrapper", err: ValidationWithComponent("ledger", system)},
+		{name: "field wrapper", err: ValidationWithFieldPrefix("lines", system)},
+		{name: "attribute wrapper", err: ValidationWithAttributes(models.Annotations{"operation": "collect"}, system)},
+		{name: "joined with warning", err: errors.Join(NewValidationWarning("warning", "warning"), system)},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			issues, err := ToValidationIssues(tc.err)
+			require.ErrorIs(t, err, system)
+			require.Empty(t, issues)
+			require.False(t, IsValidationIssueOnly(tc.err))
+		})
+	}
+}
+
+func TestExplicitStructuredErrorConversionPreservesContext(t *testing.T) {
+	system := models.NewError("provider_failed", "provider rejected invoice", models.WithComponent("provider")).
+		WithAttr("reason", "missing_address")
+
+	issues, err := ToValidationIssues(WrapAsValidationIssue(fmt.Errorf("sync: %w", system), WithWarningSeverity()))
+	require.NoError(t, err)
+	require.Len(t, issues, 1)
+	require.Equal(t, "provider_failed", issues[0].Code)
+	require.Equal(t, system.Message(), issues[0].Message)
+	require.Equal(t, ComponentName("provider"), issues[0].Component)
+	require.Equal(t, ValidationIssueSeverityWarning, issues[0].Severity)
+	require.Equal(t, "missing_address", issues[0].Attributes["reason"])
+}

openmeter/billing/worker/subscriptionsync/service/sync_credittheninvoice_test.go

@@ -13,6 +13,7 @@ import (
 	"github.com/samber/mo"
 	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
+	metricnoop "go.opentelemetry.io/otel/metric/noop"
 
 	"github.com/openmeterio/openmeter/openmeter/billing"
 	"github.com/openmeterio/openmeter/openmeter/billing/charges"
@@ -86,6 +87,7 @@ func (s *CreditThenInvoiceTestSuite) SetupSuite() {
 	advanceService := advancetestutils.NewService(s.T(), ledgerDeps, breakageService)
 
 	collectorService, err := ledgercollector.NewService(ledgercollector.Config{
+		Meter:   metricnoop.NewMeterProvider().Meter("test"),
 		Logger:  logger,
 		Advance: advanceService,
 		Ledger:  ledgerDeps.HistoricalLedger,

openmeter/ledger/README.md

@@ -87,6 +87,17 @@ persist the returned group reference with its own lifecycle state. Ledger
 annotations and entry identity preserve accounting meaning and provenance; they
 are not operation idempotency keys.
 
+## Error diagnostics
+
+Accounting guards use fixed diagnostic reasons, such as
+`ledger_breakage_release_exceeds_open_amount`. Add them with
+`models.NewError` and its `reason` attribute. These remain system errors;
+use `models.NewValidationIssue` only for domain validation.
+Existing validation issues keep their codes and `reason` attributes.
+The shared error handler logs both as `error_codes` and `error_reasons`, including
+through wrapping and joined errors. Keep IDs and amounts in the message, not the
+reason. Direct logger calls do not extract this metadata.
+
 ## Collection provenance
 
 Entry provenance carries three distinct identities:

openmeter/ledger/advance/service/attribution.go

@@ -12,6 +12,7 @@ import (
 	"github.com/openmeterio/openmeter/openmeter/ledger"
 	"github.com/openmeterio/openmeter/pkg/cmpx"
 	"github.com/openmeterio/openmeter/pkg/currencyx"
+	"github.com/openmeterio/openmeter/pkg/models"
 )
 
 // advanceAttribution keeps the amounts and original routes needed to attribute
@@ -113,7 +114,7 @@ func allocateAccruedBackedAdvanceAttributions(
 				}
 			})
 			if allocation.Amount.Sub(consumed).IsPositive() {
-				return nil, fmt.Errorf("advance attribution allocation %s exceeds remaining receivable for spend charge", allocation.Amount.String())
+				return nil, models.NewError(ledger.ErrCodeAdvanceFailed, fmt.Sprintf("advance attribution allocation %s exceeds remaining receivable for spend charge", allocation.Amount.String()), models.WithAttribute("reason", "ledger_advance_attribution_exceeds_receivable"))
 			}
 
 			attributions = append(attributions, allocated...)

openmeter/ledger/advance/service/legacy_correction.go

@@ -61,7 +61,7 @@ func (s *service) PlanLegacyCorrection(ctx context.Context, input advance.Legacy
 		}
 
 		if attribution == nil {
-			return out, fmt.Errorf("backing group %s has no receivable attribution matching the collected source", group.ID().ID)
+			return out, models.NewError(ledger.ErrCodeAdvanceFailed, fmt.Sprintf("backing group %s has no receivable attribution matching the collected source", group.ID().ID), models.WithAttribute("reason", "ledger_advance_receivable_attribution_missing"))
 		}
 
 		out.LegacyCorrections = append(out.LegacyCorrections, transactions.CorrectionInput{
@@ -272,5 +272,5 @@ func (s *service) backfilledCreditReissueRoute(group ledger.TransactionGroup) (b
 		}, nil
 	}
 
-	return backfilledCreditReissueRouteResult{}, fmt.Errorf("backing transaction group %s does not contain a known cost basis route", group.ID().ID)
+	return backfilledCreditReissueRouteResult{}, models.NewError(ledger.ErrCodeAdvanceFailed, fmt.Sprintf("backing transaction group %s does not contain a known cost basis route", group.ID().ID), models.WithAttribute("reason", "ledger_advance_cost_basis_route_missing"))
 }

openmeter/ledger/breakage/breakage_impacts.go

@@ -94,7 +94,7 @@ func (s *service) ListExpiredBreakageImpacts(ctx context.Context, input ListExpi
 
 		reference := subAccount.Route().Currency.Clone()
 		if reference.GetCode() != record.Currency {
-			return ListExpiredBreakageImpactsResult{}, fmt.Errorf("breakage FBO sub-account %s currency %s does not match record currency %s", record.FBOSubAccountID, reference.GetCode(), record.Currency)
+			return ListExpiredBreakageImpactsResult{}, models.NewError(ledger.ErrCodeBreakageFailed, fmt.Sprintf("breakage FBO sub-account %s currency %s does not match record currency %s", record.FBOSubAccountID, reference.GetCode(), record.Currency), models.WithAttribute("reason", "ledger_breakage_currency_mismatch"))
 		}
 
 		currencyReferences[record.FBOSubAccountID] = reference
@@ -150,11 +150,11 @@ func (s *service) ListExpiredBreakageImpacts(ctx context.Context, input ListExpi
 		}
 
 		if group.amount.IsNegative() {
-			return ListExpiredBreakageImpactsResult{}, fmt.Errorf("expired breakage amount is negative for %s %s", group.expiresAt, group.currency)
+			return ListExpiredBreakageImpactsResult{}, models.NewError(ledger.ErrCodeBreakageFailed, fmt.Sprintf("expired breakage amount is negative for %s %s", group.expiresAt, group.currency), models.WithAttribute("reason", "ledger_breakage_expired_amount_negative"))
 		}
 
 		if group.cursorID.ID == "" {
-			return ListExpiredBreakageImpactsResult{}, fmt.Errorf("expired breakage impact has no plan record for %s %s", group.expiresAt, group.currency)
+			return ListExpiredBreakageImpactsResult{}, models.NewError(ledger.ErrCodeBreakageFailed, fmt.Sprintf("expired breakage impact has no plan record for %s %s", group.expiresAt, group.currency), models.WithAttribute("reason", "ledger_breakage_plan_missing"))
 		}
 
 		annotations := models.Annotations{

openmeter/ledger/breakage/service.go

@@ -146,7 +146,7 @@ func (i PlanIssuanceInput) Validate() error {
 	}
 
 	if immediateReleaseAmount.GreaterThan(i.Amount) {
-		errs = append(errs, errors.New("immediate release amount cannot exceed amount"))
+		errs = append(errs, models.NewError(ledger.ErrCodeBreakageFailed, "immediate release amount cannot exceed amount", models.WithAttribute("reason", "ledger_breakage_release_exceeds_issuance")))
 	}
 
 	if err := i.Currency.Validate(); err != nil {
@@ -234,7 +234,7 @@ func (i ReleasePlanInput) Validate() error {
 	}
 
 	if i.Amount.GreaterThan(i.Plan.OpenAmount) {
-		errs = append(errs, errors.New("release amount cannot exceed open plan amount"))
+		errs = append(errs, models.NewError(ledger.ErrCodeBreakageFailed, "release amount cannot exceed open plan amount", models.WithAttribute("reason", "ledger_breakage_release_exceeds_open_amount")))
 	}
 
 	if i.Plan.FBOAddress == nil {
@@ -291,7 +291,7 @@ func (i ReopenReleaseInput) Validate() error {
 	}
 
 	if i.Amount.GreaterThan(i.Release.OpenAmount) {
-		errs = append(errs, errors.New("reopen amount cannot exceed open release amount"))
+		errs = append(errs, models.NewError(ledger.ErrCodeBreakageFailed, "reopen amount cannot exceed open release amount", models.WithAttribute("reason", "ledger_breakage_reopen_exceeds_open_amount")))
 	}
 
 	if i.Release.FBOAddress == nil {
@@ -654,7 +654,7 @@ func (s *service) ListExpiredRecords(ctx context.Context, input ListExpiredRecor
 // Deferred ledger FKs reject discarded postings at the enclosing transaction's commit.
 func (s *service) recordPosting(ctx context.Context, input PostingInput, record Record, posting ledger.TransactionInput) (ledger.TransactionInput, error) {
 	if _, err := transaction.GetDriverFromContext(ctx); err != nil {
-		return nil, fmt.Errorf("breakage requires the caller's transaction: %w", err)
+		return nil, models.NewError(ledger.ErrCodeBreakageFailed, fmt.Sprintf("breakage requires the caller's transaction: %v", err), models.WithAttribute("reason", "ledger_breakage_transaction_required"), models.WithCause(err))
 	}
 
 	posting, err := ledger.PreassignIDs(posting)
@@ -777,7 +777,7 @@ func (s *service) resolveBreakageTemplate(
 	}
 
 	if len(inputs) != 1 {
-		return nil, fmt.Errorf("expected one breakage transaction input, got %d", len(inputs))
+		return nil, models.NewError(ledger.ErrCodeBreakageFailed, fmt.Sprintf("expected one breakage transaction input, got %d", len(inputs)), models.WithAttribute("reason", "ledger_breakage_transaction_count_invalid"))
 	}
 
 	kind, err := breakageKindForTemplate(template)

openmeter/ledger/breakage/service_test.go

@@ -240,6 +240,11 @@ func TestBreakageRequiresCallerTransaction(t *testing.T) {
 
 	inputs, err := service.PlanIssuance(t.Context(), expiringIssuance(env, ulid.Make().String()))
 	require.ErrorContains(t, err, "requires the caller's transaction")
+
+	var diagnostic models.Error
+	require.ErrorAs(t, err, &diagnostic)
+	require.Equal(t, "ledger_breakage_transaction_required", diagnostic.Attributes()["reason"])
+
 	require.Nil(t, inputs)
 	requireNoBreakageRows(t, env)
 }
@@ -295,3 +300,40 @@ func requireNoBreakageRows(t *testing.T, env *ledgertestutils.IntegrationEnv) {
 	require.NoError(t, err)
 	require.Zero(t, count)
 }
+
+func TestBreakageReleaseLimitReportsReason(t *testing.T) {
+	env, service := newBreakageEnv(t)
+
+	// given: a plan for ten credits, created in the caller's transaction.
+	err := transaction.RunWithNoValue(t.Context(), enttx.NewCreator(env.DB), func(ctx context.Context) error {
+		_, err := service.PlanIssuance(ctx, expiringIssuance(env, ulid.Make().String()))
+		require.NoError(t, err)
+
+		plans, err := service.ListPlans(ctx, breakage.ListPlansInput{
+			CustomerID: env.CustomerID,
+			Currency:   env.Currency,
+			AsOf:       env.Now(),
+		})
+		require.NoError(t, err)
+		require.Len(t, plans, 1)
+
+		// when: consumption tries to release more than the plan holds.
+		_, err = service.ReleasePlan(ctx, breakage.ReleasePlanInput{
+			PostingInput: breakage.PostingInput{TransactionGroupID: ulid.Make().String()},
+			Plan:         plans[0],
+			Amount:       alpacadecimal.NewFromInt(11),
+			SourceKind:   breakage.SourceKindUsage,
+		})
+
+		// then: callers can classify the rejection without parsing its message.
+		require.ErrorContains(t, err, "release amount cannot exceed open plan amount")
+
+		var diagnostic models.Error
+		require.ErrorAs(t, err, &diagnostic)
+		require.Equal(t, "ledger_breakage_release_exceeds_open_amount", diagnostic.Attributes()["reason"])
+
+		return err
+	})
+	require.Error(t, err)
+	requireNoBreakageRows(t, env)
+}

openmeter/ledger/chargeadapter/flatfee_test.go

@@ -8,6 +8,7 @@ import (
 	"github.com/alpacahq/alpacadecimal"
 	"github.com/oklog/ulid/v2"
 	"github.com/stretchr/testify/require"
+	metricnoop "go.opentelemetry.io/otel/metric/noop"
 
 	"github.com/openmeterio/openmeter/openmeter/billing"
 	chargeflatfee "github.com/openmeterio/openmeter/openmeter/billing/charges/flatfee"
@@ -721,6 +722,7 @@ func newFlatFeeHandlerTestEnv(t *testing.T) *flatFeeHandlerTestEnv {
 	advanceService := advancetestutils.NewService(t, base.Deps, breakageService)
 
 	collectorService, err := ledgercollector.NewService(ledgercollector.Config{
+		Meter:              metricnoop.NewMeterProvider().Meter("test"),
 		Logger:             omtestutils.NewDiscardLogger(t),
 		Advance:            advanceService,
 		Ledger:             base.Deps.HistoricalLedger,

openmeter/ledger/chargeadapter/origin_test.go

@@ -13,6 +13,7 @@ import (
 	"github.com/samber/lo"
 	"github.com/samber/mo"
 	"github.com/stretchr/testify/require"
+	metricnoop "go.opentelemetry.io/otel/metric/noop"
 
 	"github.com/openmeterio/openmeter/openmeter/billing/charges/creditpurchase"
 	"github.com/openmeterio/openmeter/openmeter/billing/charges/legacylineage"
@@ -59,6 +60,7 @@ func newOriginTestEnv(t *testing.T, custom bool) *originTestEnv {
 	advanceService := advancetestutils.NewService(t, base.Deps, base.breakage)
 
 	collect, err := collector.NewService(collector.Config{
+		Meter:              metricnoop.NewMeterProvider().Meter("test"),
 		Logger:             omtestutils.NewDiscardLogger(t),
 		Advance:            advanceService,
 		Ledger:             base.Deps.HistoricalLedger,