feat: add taxonomy lifecycle observability

.env.example

@@ -42,6 +42,7 @@ PORT=8080
 # Default: info
 # Valid values: debug, info, warn, error
 LOG_LEVEL=info
+# LOG_FORMAT=json
 
 # Taxonomy service integration (optional; beta)
 # TAXONOMY_SERVICE_URL is the internal URL Hub uses to call the standalone taxonomy service.

cmd/api/app.go

@@ -230,10 +230,6 @@ func NewApp(cfg *config.Config, db *pgxpool.Pool) (*App, error) {
 		}
 	}
 
-	// Install TraceContextHandler unconditionally so request_id (and trace_id/span_id when tracing is on) appear in logs.
-	defaultHandler := slog.Default().Handler()
-	slog.SetDefault(slog.New(observability.NewTraceContextHandler(defaultHandler)))
-
 	if tracerProvider != nil {
 		otel.SetTracerProvider(tracerProvider)
 	}
@@ -449,11 +445,17 @@ func NewApp(cfg *config.Config, db *pgxpool.Pool) (*App, error) {
 		taxonomyStarter = taxonomyClient
 	}
 
+	var taxonomyMetrics observability.TaxonomyMetrics
+	if metrics != nil {
+		taxonomyMetrics = metrics.Taxonomy
+	}
+
 	taxonomyService := service.NewTaxonomyService(service.NewTaxonomyServiceParams{
 		Repo:                  taxonomyRepo,
 		Starter:               taxonomyStarter,
 		EmbeddingModel:        taxonomyEmbeddingModel,
 		MinimumEmbeddingCount: cfg.Taxonomy.MinimumEmbeddedRecords,
+		Metrics:               taxonomyMetrics,
 	})
 	taxonomyHandler := handlers.NewTaxonomyHandler(taxonomyService)
 	feedbackRecordsHandler := handlers.NewFeedbackRecordsHandler(feedbackRecordsService)
@@ -620,7 +622,7 @@ func (a *App) Run(ctx context.Context) error {
 	// Reap taxonomy runs orphaned in a non-terminal state, but only when the taxonomy service is wired
 	// (no runs exist otherwise, so the sweep would be pointless).
 	if a.taxonomyRepo != nil && (a.cfg.Taxonomy.ServiceURL != "" || a.cfg.Taxonomy.ServiceToken != "") {
-		go runTaxonomyRunReaper(ctx, a.taxonomyRepo,
+		go runTaxonomyRunReaper(ctx, a.taxonomyRepo, taxonomyMetricsFromAggregate(a.metrics),
 			a.cfg.Taxonomy.StuckRunTimeout.Duration(), a.cfg.Taxonomy.ReaperInterval.Duration())
 	}
 
@@ -821,7 +823,8 @@ const stuckTaxonomyRunMessage = "taxonomy run timed out without completing"
 // forever in the UI and generation can be retried. Idempotent: the repository's status filter skips
 // runs that finished on their own between sweeps.
 func runTaxonomyRunReaper(
-	ctx context.Context, repo *repository.TaxonomyRepository, timeout, interval time.Duration,
+	ctx context.Context, repo *repository.TaxonomyRepository, metrics observability.TaxonomyMetrics,
+	timeout, interval time.Duration,
 ) {
 	// A non-positive interval panics time.NewTicker, and a non-positive timeout would reap active
 	// runs (cutoff would be now or in the future). Either is misconfiguration — disable the reaper.
@@ -836,10 +839,32 @@ func runTaxonomyRunReaper(
 	defer ticker.Stop()
 
 	reap := func() {
-		failed, err := repo.FailStuckRuns(ctx, timeout, stuckTaxonomyRunMessage,
+		reaped, err := repo.FailStuckRuns(ctx, timeout, stuckTaxonomyRunMessage,
 			models.TaxonomyRunFailureCodeInternalError)
-		if failed > 0 {
-			slog.InfoContext(ctx, "taxonomy stuck-run reaper failed stalled runs", "count", failed)
+		for _, run := range reaped {
+			slog.ErrorContext(ctx, "taxonomy stuck-run reaper failed stalled run",
+				"event", "hub.taxonomy.run.reaped", "run_id", run.ID, "tenant_id", run.TenantID,
+				"scope_type", run.ScopeType, "source_type", run.SourceType,
+				"source_id", run.SourceID, "field_id", run.FieldID,
+				"failure_code", models.TaxonomyRunFailureCodeInternalError)
+		}
+
+		if len(reaped) > 0 && metrics != nil {
+			metrics.RecordRunsReaped(ctx, int64(len(reaped)))
+
+			for _, run := range reaped {
+				metrics.RecordRunOutcome(ctx, string(models.TaxonomyRunStatusFailed),
+					string(models.TaxonomyRunFailureCodeInternalError), string(run.ScopeType))
+
+				started := run.CreatedAt
+				if run.StartedAt != nil {
+					started = *run.StartedAt
+				}
+
+				duration := max(run.FinishedAt.Sub(started), 0)
+				metrics.RecordRunDuration(ctx, duration, string(models.TaxonomyRunStatusFailed),
+					string(run.ScopeType))
+			}
 		}
 
 		if err != nil {
@@ -859,6 +884,14 @@ func runTaxonomyRunReaper(
 	}
 }
 
+func taxonomyMetricsFromAggregate(metrics *observability.Metrics) observability.TaxonomyMetrics {
+	if metrics == nil {
+		return nil
+	}
+
+	return metrics.Taxonomy
+}
+
 // shutdownObservability shuts down tracer and meter providers. Logs secondary errors, returns the first.
 func shutdownObservability(ctx context.Context, tracer *sdktrace.TracerProvider, meter *sdkmetric.MeterProvider) error {
 	var first error

cmd/api/main.go

@@ -27,20 +27,20 @@ func main() {
 func run() int {
 	cfg, err := config.Load()
 	if err != nil {
-		observability.SetupLogging("info")
+		observability.SetupLogging("info", "text")
 		slog.Error("Failed to load configuration", "error", err)
 
 		return exitFailure
 	}
 
 	if cfg.Server.HubAPIKey == "" {
-		observability.SetupLogging(cfg.Server.LogLevel)
+		observability.SetupLogging(cfg.Server.LogLevel, cfg.Server.LogFormat)
 		slog.Error("API_KEY is required for hub-api")
 
 		return exitFailure
 	}
 
-	observability.SetupLogging(cfg.Server.LogLevel)
+	observability.SetupLogging(cfg.Server.LogLevel, cfg.Server.LogFormat)
 
 	ctx := context.Background()
 

cmd/worker/main.go

@@ -29,13 +29,13 @@ func run() int {
 	cfg, err := config.Load()
 	if err != nil {
 		// Configure logging before reporting the failure, so a broken config still produces output.
-		observability.SetupLogging("info")
+		observability.SetupLogging("info", "text")
 		slog.Error("Failed to load configuration", "error", err)
 
 		return exitFailure
 	}
 
-	observability.SetupLogging(cfg.Server.LogLevel)
+	observability.SetupLogging(cfg.Server.LogLevel, cfg.Server.LogFormat)
 
 	if cfg.Database.URL == "" || cfg.Database.URL == config.DefaultDatabaseURL {
 		slog.Error("DATABASE_URL must be set explicitly for hub-worker (do not use the default test URL)")

internal/api/handlers/taxonomy_internal_handler.go

@@ -17,8 +17,7 @@ type TaxonomyInternalService interface {
 	FailRun(
 		ctx context.Context,
 		runID uuid.UUID,
-		message string,
-		errorCode models.TaxonomyRunFailureCode,
+		req models.TaxonomyRunFailedRequest,
 	) (*models.TaxonomyRun, error)
 	Heartbeat(ctx context.Context, runID uuid.UUID) error
 }
@@ -119,7 +118,7 @@ func (h *TaxonomyInternalHandler) FailRun(w http.ResponseWriter, r *http.Request
 		return
 	}
 
-	result, err := h.service.FailRun(r.Context(), runID, req.Error, req.ErrorCode)
+	result, err := h.service.FailRun(r.Context(), runID, req)
 	if err != nil {
 		respondTaxonomyError(w, r, err)
 

internal/config/config.go

@@ -71,6 +71,7 @@ type ServerConfig struct {
 	HubAPIKey       string      `env:"API_KEY"`
 	PublicBaseURL   string      `env:"PUBLIC_BASE_URL"`
 	LogLevel        string      `env:"LOG_LEVEL"                env-default:"info"`
+	LogFormat       string      `env:"LOG_FORMAT"               env-default:"text"`
 	ShutdownTimeout DurationSec `env:"SHUTDOWN_TIMEOUT_SECONDS" env-default:"30"`
 }
 
@@ -339,6 +340,10 @@ func applyDefaults(cfg *Config) {
 		cfg.Server.LogLevel = "info"
 	}
 
+	if cfg.Server.LogFormat == "" {
+		cfg.Server.LogFormat = "text"
+	}
+
 	const defaultShutdownSec = 30
 	if cfg.Server.ShutdownTimeout.Duration() == 0 {
 		cfg.Server.ShutdownTimeout = DurationSec(time.Duration(defaultShutdownSec) * time.Second)

internal/models/taxonomy.go

@@ -251,8 +251,26 @@ type TaxonomyRunResultRequest struct {
 
 // TaxonomyRunFailedRequest records a taxonomy run failure.
 type TaxonomyRunFailedRequest struct {
-	Error     string                 `json:"error"                validate:"required,no_null_bytes,min=1,max=2000"`
-	ErrorCode TaxonomyRunFailureCode `json:"error_code,omitempty" validate:"omitempty,oneof=insufficient_data service_unavailable generation_failed invalid_output internal_error"` //nolint:lll // Validator oneof values are space-delimited.
+	Error       string                         `json:"error"                 validate:"required,no_null_bytes,min=1,max=2000"`
+	ErrorCode   TaxonomyRunFailureCode         `json:"error_code,omitempty"  validate:"omitempty,oneof=insufficient_data service_unavailable generation_failed invalid_output internal_error"` //nolint:lll // Validator oneof values are space-delimited.
+	Diagnostics *TaxonomyRunFailureDiagnostics `json:"diagnostics,omitempty" validate:"omitempty"`
+}
+
+// TaxonomyRunFailureDiagnostics contains bounded, non-sensitive compute diagnostics. It is stored
+// inside the existing metrics JSON column, keeping the database and public API schema unchanged.
+type TaxonomyRunFailureDiagnostics struct {
+	Phase              string             `json:"phase,omitempty"                   validate:"omitempty,oneof=input_fetch input_validation clustering evidence_selection cluster_labeling taxonomy_generation payload_validation persistence unknown"`                                                        //nolint:lll
+	FailureReason      string             `json:"failure_reason,omitempty"          validate:"omitempty,oneof=provider_authentication provider_rate_limit provider_timeout provider_unavailable provider_response invalid_output validation_failed insufficient_data hub_unavailable internal_error unknown"` //nolint:lll
+	Provider           string             `json:"provider,omitempty"                validate:"omitempty,oneof=openai bedrock vertex unknown"`
+	Model              string             `json:"model,omitempty"                   validate:"omitempty,no_null_bytes,max=255"`
+	ProviderAdapter    string             `json:"provider_adapter,omitempty"        validate:"omitempty,no_null_bytes,max=100"`
+	AdapterVersion     string             `json:"adapter_version,omitempty"         validate:"omitempty,no_null_bytes,max=100"`
+	ProviderSDKVersion string             `json:"provider_sdk_version,omitempty"    validate:"omitempty,no_null_bytes,max=100"`
+	LLMAttempts        *int               `json:"llm_attempts,omitempty"            validate:"omitempty,min=0,max=1000"`
+	InputTokens        *int64             `json:"input_tokens,omitempty"            validate:"omitempty,min=0"`
+	OutputTokens       *int64             `json:"output_tokens,omitempty"           validate:"omitempty,min=0"`
+	TotalTokens        *int64             `json:"total_tokens,omitempty"            validate:"omitempty,min=0"`
+	PhaseDurations     map[string]float64 `json:"phase_durations_seconds,omitempty" validate:"omitempty,max=8,dive,keys,oneof=input_fetch input_validation clustering evidence_selection cluster_labeling taxonomy_generation payload_validation persistence,endkeys,gte=0"` //nolint:lll
 }
 
 // RenameTaxonomyNodeRequest renames a generated taxonomy node.

internal/observability/aggregate.go

@@ -22,6 +22,7 @@ type Metrics struct {
 	EnrichmentClear EnrichmentClearMetrics
 	// EnrichmentBacklog gauges the aggregate eligible-but-unenriched record count per enrichment.
 	EnrichmentBacklog EnrichmentBacklogMetrics
+	Taxonomy          TaxonomyMetrics
 }
 
 // NewMetrics creates EventMetrics, WebhookMetrics, EmbeddingMetrics, TranslationMetrics, and CacheMetrics from the given meter.
@@ -77,6 +78,11 @@ func NewMetrics(meter metric.Meter) (*Metrics, error) {
 		return nil, fmt.Errorf("enrichment backlog metrics: %w", err)
 	}
 
+	taxonomy, err := NewTaxonomyMetrics(meter)
+	if err != nil {
+		return nil, fmt.Errorf("taxonomy metrics: %w", err)
+	}
+
 	return &Metrics{
 		Events:            events,
 		Webhooks:          webhooks,
@@ -87,5 +93,6 @@ func NewMetrics(meter metric.Meter) (*Metrics, error) {
 		Cache:             cache,
 		EnrichmentClear:   enrichmentClear,
 		EnrichmentBacklog: enrichmentBacklog,
+		Taxonomy:          taxonomy,
 	}, nil
 }

internal/observability/logging.go

@@ -3,6 +3,7 @@ package observability
 import (
 	"context"
 	"fmt"
+	"io"
 	"log/slog"
 	"os"
 	"strings"
@@ -14,7 +15,7 @@ import (
 // unrecognized (including the empty string) so a typo degrades to the default rather than silencing
 // logs.
 func ParseLogLevel(level string) slog.Level {
-	switch strings.ToLower(level) {
+	switch strings.ToLower(strings.TrimSpace(level)) {
 	case "debug":
 		return slog.LevelDebug
 	case "info":
@@ -28,12 +29,10 @@ func ParseLogLevel(level string) slog.Level {
 	}
 }
 
-// SetupLogging installs the default slog handler at the given level. Both binaries call it during
-// startup: hub-api and hub-worker have to agree on log format and honor the same LOG_LEVEL, or a
-// failure that only reproduces in one of them is harder to read than it needs to be.
-func SetupLogging(level string) {
-	opts := &slog.HandlerOptions{Level: ParseLogLevel(level)}
-	slog.SetDefault(slog.New(slog.NewTextHandler(os.Stdout, opts)))
+// SetupLogging installs the default slog handler at the given level and format. Both binaries call
+// it during startup so hub-api and hub-worker honor the same LOG_LEVEL and LOG_FORMAT behavior.
+func SetupLogging(level, format string) {
+	slog.SetDefault(slog.New(NewLogHandler(os.Stdout, level, format)))
 }
 
 // requestIDKey is the context key for the request ID (X-Request-ID).
@@ -99,3 +98,14 @@ func (h *TraceContextHandler) WithAttrs(attrs []slog.Attr) slog.Handler {
 func (h *TraceContextHandler) WithGroup(name string) slog.Handler {
 	return &TraceContextHandler{inner: h.inner.WithGroup(name)}
 }
+
+// NewLogHandler creates a structured JSON or human-readable text slog handler.
+// Text remains the default for local and backwards-compatible deployments.
+func NewLogHandler(output io.Writer, level, format string) slog.Handler {
+	options := &slog.HandlerOptions{Level: ParseLogLevel(level)}
+	if strings.EqualFold(strings.TrimSpace(format), "json") {
+		return NewTraceContextHandler(slog.NewJSONHandler(output, options))
+	}
+
+	return NewTraceContextHandler(slog.NewTextHandler(output, options))
+}

internal/observability/logging_test.go

@@ -1,10 +1,44 @@
 package observability
 
 import (
+	"bytes"
+	"context"
+	"encoding/json"
 	"log/slog"
+	"strings"
 	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
+func TestNewLogHandlerJSONIncludesRequestCorrelation(t *testing.T) {
+	var output bytes.Buffer
+
+	logger := slog.New(NewLogHandler(&output, "info", "json"))
+	ctx := context.WithValue(context.Background(), RequestIDKey, "request-1")
+
+	logger.InfoContext(ctx, "taxonomy lifecycle", "event", "hub.taxonomy.run.started")
+
+	assert.Equal(t, 1, strings.Count(output.String(), `"request_id"`))
+
+	var record map[string]any
+	require.NoError(t, json.Unmarshal(output.Bytes(), &record))
+	assert.Equal(t, "request-1", record["request_id"])
+	assert.Equal(t, "hub.taxonomy.run.started", record["event"])
+}
+
+func TestNewLogHandlerDefaultsToText(t *testing.T) {
+	var output bytes.Buffer
+
+	logger := slog.New(NewLogHandler(&output, "info", ""))
+
+	logger.Info("taxonomy lifecycle", "event", "hub.taxonomy.run.started")
+
+	assert.True(t, strings.HasPrefix(output.String(), "time="))
+	assert.Contains(t, output.String(), "event=hub.taxonomy.run.started")
+}
+
 // ParseLogLevel backs LOG_LEVEL for both hub-api and hub-worker, so an unrecognized value must
 // degrade to info rather than silencing logs — a config typo that hid worker output would defeat the
 // point of honoring the level at all.

internal/observability/names.go

@@ -54,6 +54,12 @@ const (
 
 	MetricNameCacheHits   = "hub_cache_hits_total"
 	MetricNameCacheMisses = "hub_cache_misses_total"
+
+	MetricNameTaxonomyRunsStarted   = "hub_taxonomy_runs_started_total"
+	MetricNameTaxonomyRunOutcomes   = "hub_taxonomy_outcomes_total"
+	MetricNameTaxonomyRunDuration   = "hub_taxonomy_run_duration_seconds"
+	MetricNameTaxonomyDispatchError = "hub_taxonomy_dispatch_errors_total"
+	MetricNameTaxonomyRunsReaped    = "hub_taxonomy_runs_reaped_total"
 )
 
 // Attribute keys.
@@ -70,7 +76,9 @@ const (
 	// AttrEnrichment labels the enrichment-backlog gauge; values are the fixed enrichment types
 	// (translation, sentiment, emotions). tenant_id is deliberately NOT a label — the gauge is
 	// aggregated across all tenants to keep cardinality bounded.
-	AttrEnrichment = "enrichment"
+	AttrEnrichment  = "enrichment"
+	AttrFailureCode = "failure_code"
+	AttrScopeType   = "scope_type"
 )
 
 // AllowedEventTypes returns event type strings allowed for metric attributes (bounded cardinality).