feat: observe workflow inbox backlog

internal/observability/business.go

@@ -21,6 +21,10 @@ type BusinessSnapshot struct {
 	WorkflowStates map[string]uint64
 	// WorkflowOldestAgeSeconds 未终结流程中最早更新时间的年龄。
 	WorkflowOldestAgeSeconds float64
+	// InboxStates Weave 内部事件状态的有界计数。
+	InboxStates map[string]uint64
+	// InboxOldestAgeSeconds 待消费内部事件中最早可处理记录的年龄。
+	InboxOldestAgeSeconds float64
 	// Truncated 是否至少有一类状态达到采样上限。
 	Truncated bool
 	// RecentEvents 固定事件类别在最近窗口内的有界数量。
@@ -63,6 +67,9 @@ func (r *Registry) RecordBusinessSnapshot(snapshot BusinessSnapshot, started tim
 	for _, state := range workflowStates {
 		r.SetGauge("orbit_weave_workflows_state", float64(snapshot.WorkflowStates[state]), Labels{"state": state})
 	}
+	for _, state := range inboxStates {
+		r.SetGauge("orbit_weave_inbox_state", float64(snapshot.InboxStates[state]), Labels{"state": state})
+	}
 	for _, eventType := range eventTypes {
 		r.SetGauge("orbit_business_events_recent", float64(snapshot.RecentEvents[eventType]), Labels{"event": eventType})
 	}
@@ -73,6 +80,7 @@ func (r *Registry) RecordBusinessSnapshot(snapshot BusinessSnapshot, started tim
 	r.SetGauge("orbit_pulse_tasks_oldest_age_seconds", snapshot.TaskOldestAgeSeconds, nil)
 	r.SetGauge("orbit_result_delivery_oldest_age_seconds", snapshot.DeliveryOldestAgeSeconds, nil)
 	r.SetGauge("orbit_weave_workflow_oldest_age_seconds", snapshot.WorkflowOldestAgeSeconds, nil)
+	r.SetGauge("orbit_weave_inbox_oldest_age_seconds", snapshot.InboxOldestAgeSeconds, nil)
 	truncated := 0.0
 	if snapshot.Truncated {
 		truncated = 1

internal/observability/catalog.go

@@ -9,6 +9,7 @@ var components = []string{"workflow", "schedule", "execution_recovery"}
 var taskStates = []string{"WAITING", "LEASED", "RUNNING", "NEEDS_REVIEW", "SUCCEEDED", "FAILED", "CANCELLED"}
 var deliveryStates = []string{"PENDING", "LEASED", "ACKED", "EXPIRED"}
 var workflowStates = []string{"RUNNING", "WAITING", "CANCELLING", "NEEDS_REVIEW", "SUCCEEDED", "FAILED", "CANCELLED"}
+var inboxStates = []string{"PENDING", "LEASED", "PROCESSED", "QUARANTINED"}
 var eventTypes = []string{"task.accepted", "task.started", "task.succeeded", "task.failed", "task.needs_review", "task.requeued", "workflow.started", "workflow.advanced", "workflow.recovered", "workflow.blocked", "workflow.stopped"}
 
 // Every exported family is declared here, including each allowed label value.
@@ -43,6 +44,8 @@ func catalog() map[string]descriptor {
 	c["orbit_result_delivery_oldest_age_seconds"] = descriptor{gauge, "Age of the oldest sampled pending result delivery", nil}
 	c["orbit_weave_workflows_state"] = descriptor{gauge, "Bounded count of Weave instances by durable state", map[string][]string{"state": workflowStates}}
 	c["orbit_weave_workflow_oldest_age_seconds"] = descriptor{gauge, "Age of the oldest sampled unfinished Weave instance", nil}
+	c["orbit_weave_inbox_state"] = descriptor{gauge, "Bounded count of Weave internal inbox records by durable state", map[string][]string{"state": inboxStates}}
+	c["orbit_weave_inbox_oldest_age_seconds"] = descriptor{gauge, "Age of the oldest sampled pending Weave inbox record", nil}
 	c["orbit_business_events_recent"] = descriptor{gauge, "Bounded count of fixed event categories in the recent window", map[string][]string{"event": eventTypes}}
 	c["orbit_pulse_task_terminal_latency_seconds"] = descriptor{gauge, "Average sampled Pulse task create-to-terminal latency in seconds", nil}
 	c["orbit_pulse_task_terminal_latency_max_seconds"] = descriptor{gauge, "Maximum sampled Pulse task create-to-terminal latency in seconds", nil}

internal/observability/metrics_test.go

@@ -123,6 +123,8 @@ func TestBusinessSnapshotKeepsLastGoodStateOnFailure(t *testing.T) {
 		DeliveryOldestAgeSeconds:   8.25,
 		WorkflowStates:             map[string]uint64{"RUNNING": 1},
 		WorkflowOldestAgeSeconds:   4.75,
+		InboxStates:                map[string]uint64{"PENDING": 6},
+		InboxOldestAgeSeconds:      3.25,
 		RecentEvents:               map[string]uint64{"task.accepted": 5},
 		TaskTerminalLatencySeconds: 2.5, TaskTerminalLatencyMaxSeconds: 7.5,
 		WorkflowTerminalLatencySeconds: 3.5, WorkflowTerminalLatencyMaxSeconds: 9.5,
@@ -133,10 +135,12 @@ func TestBusinessSnapshotKeepsLastGoodStateOnFailure(t *testing.T) {
 		`orbit_pulse_tasks_state{state="WAITING"} 3`,
 		`orbit_result_deliveries_state{state="PENDING"} 2`,
 		`orbit_weave_workflows_state{state="RUNNING"} 1`,
+		`orbit_weave_inbox_state{state="PENDING"} 6`,
 		`orbit_business_events_recent{event="task.accepted"} 5`,
 		"orbit_pulse_task_terminal_latency_seconds 2.5",
 		"orbit_weave_workflow_terminal_latency_max_seconds 9.5",
 		"orbit_pulse_tasks_oldest_age_seconds 12.5",
+		"orbit_weave_inbox_oldest_age_seconds 3.25",
 		"orbit_business_snapshot_truncated 1",
 		`orbit_business_snapshot_total{outcome="success"} 1`,
 	} {

internal/storage/mysql/operational_metrics.go

@@ -25,6 +25,10 @@ type OperationalSnapshot struct {
 	WorkflowStates map[string]uint64
 	// WorkflowOldestAgeSeconds 是未终结流程中最早 updated_at 的年龄。
 	WorkflowOldestAgeSeconds float64
+	// InboxStates 是 Weave 内部事件按状态的有界计数。
+	InboxStates map[string]uint64
+	// InboxOldestAgeSeconds 是待消费内部事件中最早 available_at 的年龄。
+	InboxOldestAgeSeconds float64
 	// Truncated 表示至少一个状态达到有界采样上限。
 	Truncated bool
 	// RecentEvents 是固定事件类别在最近窗口内的有界数量。
@@ -43,6 +47,7 @@ var (
 	operationalTaskStates     = []string{"WAITING", "LEASED", "RUNNING", "NEEDS_REVIEW", "SUCCEEDED", "FAILED", "CANCELLED"}
 	operationalDeliveryStates = []string{"PENDING", "LEASED", "ACKED", "EXPIRED"}
 	operationalWorkflowStates = []string{"RUNNING", "WAITING", "CANCELLING", "NEEDS_REVIEW", "SUCCEEDED", "FAILED", "CANCELLED"}
+	operationalInboxStates    = []string{"PENDING", "LEASED", "PROCESSED", "QUARANTINED"}
 	operationalEventTypes     = []string{"task.accepted", "task.started", "task.succeeded", "task.failed", "task.needs_review", "task.requeued", "workflow.started", "workflow.advanced", "workflow.recovered", "workflow.blocked", "workflow.stopped"}
 )
 
@@ -68,10 +73,18 @@ func ReadOperationalSnapshot(ctx context.Context, db *sql.DB) (OperationalSnapsh
 	if err != nil {
 		return OperationalSnapshot{}, err
 	}
+	inbox, inboxTruncated, err := boundedStateCounts(ctx, db, "weave_inbox", "ix_inbox_due", operationalInboxStates)
+	if err != nil {
+		return OperationalSnapshot{}, err
+	}
 	taskAge, err := oldestAge(ctx, db, "pulse_tasks", "ix_task_observe_state_available", []string{"WAITING", "LEASED", "RUNNING", "NEEDS_REVIEW"}, "available_at", now)
 	if err != nil {
 		return OperationalSnapshot{}, err
 	}
+	inboxAge, err := oldestAge(ctx, db, "weave_inbox", "ix_inbox_due", []string{"PENDING", "LEASED"}, "available_at", now)
+	if err != nil {
+		return OperationalSnapshot{}, err
+	}
 	deliveryAge, err := oldestAge(ctx, db, "orbit_deliveries", "ix_delivery_observe_state_available", []string{"PENDING", "LEASED"}, "available_at", now)
 	if err != nil {
 		return OperationalSnapshot{}, err
@@ -96,10 +109,11 @@ func ReadOperationalSnapshot(ctx context.Context, db *sql.DB) (OperationalSnapsh
 		TaskStates: tasks, TaskOldestAgeSeconds: taskAge,
 		DeliveryStates: deliveries, DeliveryOldestAgeSeconds: deliveryAge,
 		WorkflowStates: workflows, WorkflowOldestAgeSeconds: workflowAge,
+		InboxStates: inbox, InboxOldestAgeSeconds: inboxAge,
 		RecentEvents:               events,
 		TaskTerminalLatencySeconds: taskLatency.average, TaskTerminalLatencyMaxSeconds: taskLatency.maximum,
 		WorkflowTerminalLatencySeconds: workflowLatency.average, WorkflowTerminalLatencyMaxSeconds: workflowLatency.maximum,
-		Truncated: taskTruncated || deliveryTruncated || workflowTruncated || eventTruncated || taskLatency.truncated || workflowLatency.truncated,
+		Truncated: taskTruncated || deliveryTruncated || workflowTruncated || inboxTruncated || eventTruncated || taskLatency.truncated || workflowLatency.truncated,
 	}, nil
 }
 

internal/storage/mysql/operational_metrics_integration_test.go

@@ -70,7 +70,7 @@ func TestIntegrationOperationalSnapshot(t *testing.T) {
 	if got := snapshot.RecentEvents["task.accepted"]; got < 1 {
 		t.Fatalf("accepted event was not observed: %#v", snapshot.RecentEvents)
 	}
-	if snapshot.TaskOldestAgeSeconds < 0 || snapshot.DeliveryOldestAgeSeconds < 0 || snapshot.WorkflowOldestAgeSeconds < 0 {
+	if snapshot.TaskOldestAgeSeconds < 0 || snapshot.DeliveryOldestAgeSeconds < 0 || snapshot.WorkflowOldestAgeSeconds < 0 || snapshot.InboxOldestAgeSeconds < 0 {
 		t.Fatalf("negative age in snapshot: %#v", snapshot)
 	}
 	for _, state := range []string{"WAITING", "LEASED", "RUNNING", "NEEDS_REVIEW", "SUCCEEDED", "FAILED", "CANCELLED"} {
@@ -84,6 +84,11 @@ func TestIntegrationOperationalSnapshot(t *testing.T) {
 	if _, ok := snapshot.WorkflowStates["RUNNING"]; !ok {
 		t.Fatal("missing running workflow state")
 	}
+	for _, state := range []string{"PENDING", "LEASED", "PROCESSED", "QUARANTINED"} {
+		if _, ok := snapshot.InboxStates[state]; !ok {
+			t.Fatalf("missing fixed inbox state %q", state)
+		}
+	}
 	f.exec("UPDATE pulse_tasks SET state='SUCCEEDED', completed_at=UTC_TIMESTAMP(6), updated_at=UTC_TIMESTAMP(6), revision=revision+1 WHERE namespace_id=? AND id=?", f.resources.NamespaceID, task.ID)
 	completed, err := ReadOperationalSnapshot(ctx, db)
 	if err != nil {