Merge pull request #922 from warmbly/devin/1791625283-preserve-typed-cli-monitoring

feat: preserve typed monitoring identities and count semantics in redacted admin CLI output

cmd/warmblyctl/remote_output.go

@@ -94,6 +94,75 @@ MANAGE_TESTERS MANAGE_WARMUP_BANS REVIEW_APPEALS public dedicated shared global
 
 var remoteSemver = regexp.MustCompile(`^v?[0-9]{1,5}\.[0-9]{1,5}\.[0-9]{1,5}(-[a-z0-9.-]{1,30})?$`)
 
+var remoteMonitoringIDs = remoteWordSet(`
+send_safety sends campaign_reservations dispatch sync mailbox_errors worker_samples
+jobs dead_letters webhooks notifications result_effects arrivals workers domain_auth
+warmup_loading backend_event_broker worker_command_broker send_wait_attribution tracking_broker
+arrival_aging arrival_due arrival_leased arrival_pending arrival_stage_0 arrival_stage_1
+arrival_stage_2 arrival_stage_3 arrival_wait backfill_complete backfill_pending
+backfill_running backfill_stale_running campaign_reserved campaign_reserved_aging
+confirmed_other_1h confirmed_warmup_1h confirmed_warmup_proxy_1h credential_errors
+dispatch_due dispatch_old_failure dispatch_persistent dlq_due dlq_exhausted dlq_replayed
+dlq_unscheduled dlq_wait effects_leased effects_pending effects_repeated effects_unleased
+failed_results_1h failed_results_24h future_schedule holds_authentication holds_conflict
+holds_permanent holds_unknown inactive_credential_errors job_error_old job_error_recent
+job_future job_ok job_overdue job_running notification_aging notification_due
+notification_exhausted notification_retry notification_sending notification_settled_24h
+notification_stale notification_wait original_slot_overdue_1h other_mailbox_errors
+reservations_5_to_30m reservations_over_30m reservations_under_5m retry_wait send_cooldowns
+send_paused sync_deferred sync_eligible sync_errors sync_future_clock sync_held_stale
+sync_hold_burst sync_hold_daily sync_hold_hourly sync_hold_org_daily sync_hold_other
+sync_hold_priority_daily sync_missing sync_safety_reports sync_stale sync_state_rows
+sync_unassigned task_failures_recent unknown_results unscheduled_pending webhook_abandoned_24h
+webhook_auto_disabled webhook_delivered_24h webhook_drop_windows webhook_due webhook_endpoint_old
+webhook_endpoint_persistent webhook_endpoint_recent webhook_retry webhook_stale_claim webhook_wait
+worker_sample_auth_errors_1h worker_sample_rate_errors_1h worker_sample_sends_1h worker_samples_missing
+domain_blocked domain_failing domain_future_clock domain_passing_fresh domain_stale domain_unknown
+loading_old loading_persistent loading_recent workers_missing_heartbeat worker_missing_heartbeat
+send_budget_wait send_daily_limit_wait send_schedule_reason tracking_queue
+broker_queue committed_lag members
+`)
+
+var remoteMonitoringUnits = remoteWordSet(`
+arrivals deliveries effects endpoints jobs mailboxes messages records reports reservations
+rows tasks windows workers members
+`)
+
+var remoteMonitoringReasons = remoteWordSet(`
+scope_limit sample_limit query_failed schema_absent no_registered_scopes
+dependency_missing resource_absent permission_denied timeout collection_failed
+unsupported no_recent_evidence
+`)
+
+func safeRemoteMonitoringString(key, value string) string {
+	if value == "" {
+		return ""
+	}
+	switch key {
+	case "id":
+		if remoteMonitoringIDs[value] {
+			return value
+		}
+	case "unit":
+		if remoteMonitoringUnits[value] {
+			return value
+		}
+	case "severity":
+		if value == "info" || value == "warning" || value == "critical" {
+			return value
+		}
+	case "reason":
+		if remoteMonitoringReasons[value] {
+			return value
+		}
+	case "version":
+		if value == "1" {
+			return value
+		}
+	}
+	return "[REDACTED]"
+}
+
 func decodeRemoteJSON(payload []byte) (any, error) {
 	dec := json.NewDecoder(bytes.NewReader(payload))
 	dec.UseNumber()
@@ -164,6 +233,7 @@ func safeRemoteString(key, value string) any {
 	if value == "" {
 		return ""
 	}
+
 	if (key == "cursor" || key == "next_cursor" || key == "prev_cursor") && len(value) <= 512 {
 		if _, err := paging.DecodeOffsetCursor(value); err == nil {
 			return value
@@ -212,6 +282,15 @@ func redactRemoteJSON(key string, value any) any {
 				redacted++
 				continue
 			}
+			if (key == "sources" || key == "metrics") && (name == "id" || name == "unit" || name == "severity" || name == "reason") ||
+				key == "" && name == "version" && v["sources"] != nil {
+				if value, ok := item.(string); ok {
+					out[name] = safeRemoteMonitoringString(name, value)
+				} else {
+					out[name] = "[REDACTED]"
+				}
+				continue
+			}
 			out[name] = redactRemoteJSON(name, item)
 		}
 		if redacted > 0 {

cmd/warmblyctl/remote_test.go

@@ -479,6 +479,116 @@ func TestRemoteRedactionPreservesExplicitUnavailableGenericErrorCoverage(t *test
 	}
 }
 
+func TestRemoteRedactionPreservesTypedMonitoring(t *testing.T) {
+	value, err := decodeRemoteJSON([]byte(`{"version":"1","coverage":"partial","sources":[{"id":"sends","reason":"scope_limit","metrics":[{"id":"failed_results_1h","unit":"tasks","severity":"warning","condition":"recent_failure","count":7,"title":"fixture-private-title","note":"fixture-private-note"},{"id":"unknown_results","unit":"tasks","severity":"warning","condition":"recovery_unverified","count":null},{"id":"fresh","unit":"fresh","severity":"fresh","reason":"fresh"},{"id":"future_metric_fixture","unit":"future_unit_fixture","severity":"future_severity_fixture","reason":"fixture-private-reason"}]},{"id":"fresh","reason":"fresh"}],"payload":{"token":"fixture-private-token"}}`))
+	if err != nil {
+		t.Fatal(err)
+	}
+	got := redactRemoteJSON("", value).(map[string]any)
+	source := got["sources"].([]any)[0].(map[string]any)
+	metrics := source["metrics"].([]any)
+	first := metrics[0].(map[string]any)
+	if got["version"] != "1" || got["coverage"] != "partial" || source["id"] != "sends" || source["reason"] != "scope_limit" || first["id"] != "failed_results_1h" || first["unit"] != "tasks" || first["severity"] != "warning" || first["condition"] != "recent_failure" || first["count"] != json.Number("7") {
+		t.Fatal("typed monitoring identity, severity, units or count discarded")
+	}
+	unknown := metrics[1].(map[string]any)
+	if unknown["id"] != "unknown_results" || unknown["count"] != nil {
+		t.Fatal("unmeasured metric converted to zero or its identity discarded")
+	}
+	for _, candidate := range metrics[2:] {
+		for _, field := range []string{"id", "unit", "severity", "reason"} {
+			if candidate.(map[string]any)[field] != "[REDACTED]" {
+				t.Fatalf("unknown %s exposed", field)
+			}
+		}
+	}
+	if unknownSource := got["sources"].([]any)[1].(map[string]any); unknownSource["id"] != "[REDACTED]" || unknownSource["reason"] != "[REDACTED]" {
+		t.Fatal("unknown source identity or reason exposed")
+	}
+	var output bytes.Buffer
+	if err := writeRemoteJSON(&output, got); err != nil {
+		t.Fatal(err)
+	}
+	if strings.Contains(output.String(), "fixture-private") || strings.Contains(output.String(), "future_") {
+		t.Fatal("free text or unknown enum escaped monitoring projection")
+	}
+	for _, value := range []string{"failed_results_1h", "domain_auth", "warning", "critical", "tasks", "scope_limit", "1"} {
+		if safeRemoteString("name", value) != "[REDACTED]" {
+			t.Fatalf("monitoring value %q accepted outside its typed field", value)
+		}
+	}
+	if got := redactRemoteJSON("", map[string]any{"version": "fresh", "sources": []any{}}).(map[string]any); got["version"] != "[REDACTED]" {
+		t.Fatal("unknown monitoring schema version escaped")
+	}
+	if got := redactRemoteJSON("", map[string]any{"status": "fresh", "id": "pending"}).(map[string]any); got["status"] != "fresh" || got["id"] != "pending" {
+		t.Fatal("other admin projections lost their legacy enum treatment")
+	}
+}
+
+func TestRemoteMonitoringProjectionMatchesCompiledCollectors(t *testing.T) {
+	for _, path := range []string{
+		"../../internal/repository/pg_monitoring.go",
+		"../../internal/repository/pg_monitoring_state.go",
+		"../../internal/repository/pg_monitoring_broker.go",
+		"../../internal/app/monitoring/unobserved.go",
+		"../../internal/infrastructure/eventbus/diagnostics.go",
+	} {
+		file, err := parser.ParseFile(token.NewFileSet(), path, nil, 0)
+		if err != nil {
+			t.Fatal(err)
+		}
+		check := func(key string, node ast.Node) {
+			t.Helper()
+			literal, ok := node.(*ast.BasicLit)
+			if !ok || literal.Kind != token.STRING {
+				return
+			}
+			value, err := strconv.Unquote(literal.Value)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if got := safeRemoteMonitoringString(key, value); got != value {
+				t.Errorf("%s: compiled monitoring %s=%q is redacted", path, key, value)
+			}
+		}
+		ast.Inspect(file, func(node ast.Node) bool {
+			switch n := node.(type) {
+			case *ast.ValueSpec:
+				if len(n.Names) == 1 && n.Names[0].Name == "monitoringSectionOrder" {
+					for _, value := range n.Values[0].(*ast.CompositeLit).Elts {
+						check("id", value)
+					}
+				}
+			case *ast.KeyValueExpr:
+				if key, ok := n.Key.(*ast.Ident); ok {
+					switch field := strings.ToLower(key.Name); field {
+					case "id", "unit", "severity", "reason":
+						if key.Name == "ID" && strings.HasSuffix(path, "pg_monitoring_broker.go") {
+							if literal, ok := n.Value.(*ast.BasicLit); ok && literal.Value == `"worker_events"` {
+								if safeRemoteString("scope_id", "worker_events") != "worker_events" {
+									t.Error("worker-event scope ID is redacted")
+								}
+								break
+							}
+						}
+						check(field, n.Value)
+					}
+				}
+			case *ast.AssignStmt:
+				for i, left := range n.Lhs {
+					if field, ok := left.(*ast.SelectorExpr); ok && i < len(n.Rhs) {
+						switch key := strings.ToLower(field.Sel.Name); key {
+						case "id", "unit", "severity", "reason":
+							check(key, n.Rhs[i])
+						}
+					}
+				}
+			}
+			return true
+		})
+	}
+}
+
 func TestRemoteMalformedInputAndWriteSafeguards(t *testing.T) {
 	for _, input := range []string{`{`, `{"enabled":true} {}`, `{"enabled":true,"enabled":false}`, strings.Repeat("[", 66) + "0" + strings.Repeat("]", 66)} {
 		if _, err := decodeRemoteJSON([]byte(input)); err == nil {

docs/content/docs/development/warmblyctl.mdx

@@ -203,6 +203,8 @@ Some catalog entries are discoverable but deliberately unavailable: browser devi
 
 Remote success is one deterministic JSON value on stdout. Progress, approval codes, help and machine-readable errors are on stderr. Object keys are sorted by the JSON encoder; backend list ordering is preserved. Redaction projects only allowlisted diagnostic fields, canonical UUIDs, timestamps, typed enums, known broker scope names, validated cursors, numbers and booleans. Unknown fields are omitted and counted in `_redacted_fields`; arbitrary strings become `[REDACTED]`, including addresses and display names. No provider tokens, passwords, body, header, payload or free-form error subtree is exposed. The existing local emergency and public API commands keep their own documented output behavior unchanged.
 
+Monitoring preserves the compiled source and metric IDs, count units, severity, schema version and bounded coverage reasons so an AI client can identify each observation without free-form labels. New or unrecognized identifiers still fail closed; update the CLI when the backend adds a monitoring metric. Titles, notes, mailbox addresses and provider error text remain redacted.
+
 Errors preserve safe server `code`, HTTP `status` and `request_id`, but never raw backend messages, response bodies, response headers or transport errors. A numeric `retry_after_seconds` can accompany rate limiting.
 
 | Exit | Meaning |