Add noop-monitor backend for convos connection monitoring

examples/streaming-daemon-go/README.md

@@ -14,7 +14,7 @@ This daemon receives client connections from Apache via `mod_socket_handoff` usi
 
 ## Features
 
-- **Backend plugin architecture** - Extensible backends (langgraph, mock, openai, typing) with Caddy-style `init()` registration
+- **Backend plugin architecture** - Extensible backends (langgraph, mock, noop-monitor, openai, typing) with Caddy-style `init()` registration
 - **YAML configuration** - Optional config file with flag overrides
 - **Goroutine-per-connection** - Lightweight concurrency for high throughput
 - **SCM_RIGHTS fd receiving** - Portable buffer sizing with `syscall.CmsgSpace`
@@ -103,6 +103,7 @@ The daemon supports multiple streaming backends via a plugin architecture. Backe
 |---------|-------------|----------|
 | `langgraph` | LangGraph Platform API | Stateful agents with conversation threads |
 | `mock` | Fixed demo messages with configurable delay | Testing, benchmarking |
+| `noop-monitor` | Holds the connection open, sends nothing | Connection volume/concurrency/duration measurement |
 | `openai` | OpenAI-compatible streaming API | GPT-4, Groq, Ollama, any OpenAI-compatible API |
 | `typing` | Character-by-character typewriter effect | Demos, fortune integration |
 
@@ -224,6 +225,23 @@ Streams characters one at a time with realistic typing delays. Uses `/usr/games/
 ./streaming-daemon -backend typing
 ```
 
+### Noop Monitor Backend
+
+Holds a handed-off connection open and sends only periodic `: ping` SSE keepalives — no LLM, no message data, no per-user logging. Its purpose is measurement: connection volume, concurrency, and duration are read straight from this daemon's Prometheus metrics. Intended to run as a dedicated monitor instance on its own socket and metrics port.
+
+```bash
+./streaming-daemon -backend noop-monitor -socket /run/streaming-daemon/convos-monitor.sock
+```
+
+Config (`backend.noop_monitor.ping_interval_ms`, default `25000`) sets the keepalive interval; the keepalive doubles as client-disconnect detection since a write-only stream only learns the client left when a write fails. `server.max_stream_duration_ms` optionally caps connection lifetime (`0` = hold indefinitely).
+
+Besides the daemon-level metrics, it exposes two low-cardinality series:
+
+| Metric | Type | Labels | Meaning |
+|--------|------|--------|---------|
+| `noop_monitor_active` | gauge | `source` | Currently held connections by surface (`detail`, `message_list`, `other`) |
+| `noop_monitor_closed_total` | counter | `source`, `reason` | Closed connections by surface and reason (`client_disconnect`, `max_lifetime`, `shutdown`) |
+
 ### Adding New Backends
 
 To add a new backend (e.g., Anthropic):

examples/streaming-daemon-go/backends/backend.go

@@ -70,18 +70,25 @@ type HandoffData struct {
 	TestPattern string    `json:"test_pattern,omitempty"` // For testing: passed to backend as X-Test-Pattern header
 
 	// LangGraph-specific fields
-	ThreadID       string         `json:"thread_id,omitempty"`        // For stateful runs (uses /threads/{id}/runs/stream)
-	AssistantID    string         `json:"assistant_id,omitempty"`     // Override default assistant ID
-	StreamMode     []string       `json:"stream_mode,omitempty"`      // Stream modes (e.g., ["messages", "updates", "custom"])
-	LangGraphInput map[string]any `json:"langgraph_input,omitempty"`  // Custom input fields (seller_id, shop_id, etc.)
-	Profile        string         `json:"profile,omitempty"`          // Named LangGraph profile from config
-	LangGraphURL   string         `json:"langgraph_url,omitempty"`    // Per-request API base URL override
+	ThreadID       string         `json:"thread_id,omitempty"`         // For stateful runs (uses /threads/{id}/runs/stream)
+	AssistantID    string         `json:"assistant_id,omitempty"`      // Override default assistant ID
+	StreamMode     []string       `json:"stream_mode,omitempty"`       // Stream modes (e.g., ["messages", "updates", "custom"])
+	LangGraphInput map[string]any `json:"langgraph_input,omitempty"`   // Custom input fields (seller_id, shop_id, etc.)
+	Profile        string         `json:"profile,omitempty"`           // Named LangGraph profile from config
+	LangGraphURL   string         `json:"langgraph_url,omitempty"`     // Per-request API base URL override
 	LangGraphKey   string         `json:"langgraph_api_key,omitempty"` // Per-request API key override
-	LG             string         `json:"lg,omitempty"`               // Compact: "profile|url|key" (pipe-delimited, empty segment = no override for that position)
+	LG             string         `json:"lg,omitempty"`                // Compact: "profile|url|key" (pipe-delimited, empty segment = no override for that position)
 
 	// Per-request backend selection (overrides the daemon's default provider)
 	Backend string `json:"backend,omitempty"`
 
+	// Convos connection-monitor fields (noop-monitor backend).
+	// Source labels the originating surface ("detail"|"message_list") for
+	// per-surface metrics. ConnectionID is a correlation id for optional
+	// per-connection debug logging only; neither has a functional consumer.
+	Source       string `json:"source,omitempty"`
+	ConnectionID string `json:"connection_id,omitempty"`
+
 	// Image handoff fields (for multimodal requests)
 	// Legacy single-image fields (deprecated, use Images/ImagePaths instead)
 	ImagePath     string `json:"image_path,omitempty"`      // Deprecated: use ImagePaths

examples/streaming-daemon-go/backends/noop_monitor.go

@@ -0,0 +1,110 @@
+package backends
+
+import (
+	"context"
+	"errors"
+	"log/slog"
+	"net"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+
+	"examples/config"
+)
+
+// Backend-owned metrics for the connection monitor. Kept low-cardinality:
+// source is clamped to {detail, message_list, other} and reason to a fixed
+// set, so these are safe to expose per-label.
+var (
+	monitorActive = promauto.NewGaugeVec(prometheus.GaugeOpts{
+		Name: "noop_monitor_active",
+		Help: "Currently held monitor connections, by originating surface",
+	}, []string{"source"})
+
+	monitorClosed = promauto.NewCounterVec(prometheus.CounterOpts{
+		Name: "noop_monitor_closed_total",
+		Help: "Monitor connections closed, by surface and reason",
+	}, []string{"source", "reason"})
+)
+
+// NoopMonitor holds a handed-off client connection open, sending only periodic
+// SSE keepalive comments, so connection volume/concurrency/duration can be read
+// straight from this daemon's Prometheus metrics. It never streams message data.
+type NoopMonitor struct {
+	pingInterval time.Duration
+}
+
+func init() {
+	Register(&NoopMonitor{})
+}
+
+func (n *NoopMonitor) Name() string {
+	return "noop-monitor"
+}
+
+func (n *NoopMonitor) Description() string {
+	return "Holds the connection open, sends nothing (convos connection monitor)"
+}
+
+func (n *NoopMonitor) Init(cfg *config.BackendConfig) error {
+	n.pingInterval = 25 * time.Second
+	if cfg != nil && cfg.NoopMonitor.PingIntervalMs > 0 {
+		n.pingInterval = time.Duration(cfg.NoopMonitor.PingIntervalMs) * time.Millisecond
+	}
+	slog.Info("noop-monitor backend initialized", "ping_interval", n.pingInterval)
+	return nil
+}
+
+// Stream holds the connection open, writing a `: ping` keepalive every
+// pingInterval. A write-only stream over SOCK_SEQPACKET only learns the client
+// left when a write fails, so the keepalive doubles as disconnect detection.
+// A client disconnect is normal, not an error, so it returns (0, nil).
+func (n *NoopMonitor) Stream(ctx context.Context, conn net.Conn, handoff HandoffData) (int64, error) {
+	source := normalizeSource(handoff.Source)
+
+	RecordBackendRequest("noop-monitor")
+	monitorActive.WithLabelValues(source).Inc()
+	defer monitorActive.WithLabelValues(source).Dec()
+
+	start := time.Now()
+	reason := "shutdown"
+	ticker := time.NewTicker(n.pingInterval)
+	defer ticker.Stop()
+
+loop:
+	for {
+		select {
+		case <-ctx.Done():
+			if errors.Is(context.Cause(ctx), context.DeadlineExceeded) {
+				reason = "max_lifetime"
+			}
+			break loop
+		case <-ticker.C:
+			_ = conn.SetWriteDeadline(time.Now().Add(WriteTimeout))
+			if _, err := conn.Write(pingMsg); err != nil {
+				reason = "client_disconnect"
+				break loop
+			}
+		}
+	}
+
+	RecordBackendDuration("noop-monitor", time.Since(start).Seconds())
+	monitorClosed.WithLabelValues(source, reason).Inc()
+	return 0, nil
+}
+
+// pingMsg is an SSE comment line; clients ignore it, but the write attempt is
+// how a write-only stream detects that the client has gone away.
+var pingMsg = []byte(": ping\n\n")
+
+// normalizeSource clamps the client-influenced source label to a known set so a
+// malformed or unexpected value can't blow up Prometheus label cardinality.
+func normalizeSource(source string) string {
+	switch source {
+	case "detail", "message_list":
+		return source
+	default:
+		return "other"
+	}
+}

examples/streaming-daemon-go/backends/noop_monitor_test.go

@@ -0,0 +1,124 @@
+package backends
+
+import (
+	"bytes"
+	"context"
+	"io"
+	"net"
+	"testing"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus/testutil"
+)
+
+// runMonitor runs Stream against a net.Pipe and returns once it exits.
+// The drain callback receives the client end so a test can either read the
+// pings or close the connection to simulate a disconnect.
+func runMonitor(t *testing.T, ctx context.Context, n *NoopMonitor, h HandoffData, drain func(client net.Conn)) {
+	t.Helper()
+	client, server := net.Pipe()
+	defer client.Close()
+	defer server.Close()
+
+	drain(client)
+
+	done := make(chan struct{})
+	go func() {
+		n.Stream(ctx, server, h)
+		close(done)
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("Stream did not return within timeout")
+	}
+}
+
+func TestNoopMonitorClientDisconnect(t *testing.T) {
+	before := testutil.ToFloat64(monitorClosed.WithLabelValues("detail", "client_disconnect"))
+	activeBefore := testutil.ToFloat64(monitorActive.WithLabelValues("detail"))
+
+	n := &NoopMonitor{pingInterval: 5 * time.Millisecond}
+	runMonitor(t, context.Background(), n, HandoffData{Source: "detail"}, func(client net.Conn) {
+		client.Close() // client is gone; the first ping write fails
+	})
+
+	if got := testutil.ToFloat64(monitorClosed.WithLabelValues("detail", "client_disconnect")); got != before+1 {
+		t.Errorf("client_disconnect counter = %v, want %v", got, before+1)
+	}
+	if got := testutil.ToFloat64(monitorActive.WithLabelValues("detail")); got != activeBefore {
+		t.Errorf("active gauge = %v, want %v (should return to baseline)", got, activeBefore)
+	}
+}
+
+func TestNoopMonitorShutdown(t *testing.T) {
+	before := testutil.ToFloat64(monitorClosed.WithLabelValues("detail", "shutdown"))
+
+	ctx, cancel := context.WithCancel(context.Background())
+	// Long ping interval so no write happens before the cancel is observed.
+	n := &NoopMonitor{pingInterval: 10 * time.Second}
+	go func() {
+		time.Sleep(10 * time.Millisecond)
+		cancel()
+	}()
+	runMonitor(t, ctx, n, HandoffData{Source: "detail"}, func(client net.Conn) {
+		go io.Copy(io.Discard, client)
+	})
+	cancel()
+
+	if got := testutil.ToFloat64(monitorClosed.WithLabelValues("detail", "shutdown")); got != before+1 {
+		t.Errorf("shutdown counter = %v, want %v", got, before+1)
+	}
+}
+
+func TestNoopMonitorMaxLifetime(t *testing.T) {
+	before := testutil.ToFloat64(monitorClosed.WithLabelValues("detail", "max_lifetime"))
+
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond)
+	defer cancel()
+	n := &NoopMonitor{pingInterval: 10 * time.Second}
+	runMonitor(t, ctx, n, HandoffData{Source: "detail"}, func(client net.Conn) {
+		go io.Copy(io.Discard, client)
+	})
+
+	if got := testutil.ToFloat64(monitorClosed.WithLabelValues("detail", "max_lifetime")); got != before+1 {
+		t.Errorf("max_lifetime counter = %v, want %v", got, before+1)
+	}
+}
+
+func TestNoopMonitorWritesPing(t *testing.T) {
+	n := &NoopMonitor{pingInterval: 5 * time.Millisecond}
+	client, server := net.Pipe()
+	defer client.Close()
+	defer server.Close()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	go func() {
+		n.Stream(ctx, server, HandoffData{Source: "message_list"})
+	}()
+
+	got := make([]byte, len(pingMsg))
+	_ = client.SetReadDeadline(time.Now().Add(2 * time.Second))
+	if _, err := io.ReadFull(client, got); err != nil {
+		t.Fatalf("reading ping: %v", err)
+	}
+	if !bytes.Equal(got, pingMsg) {
+		t.Errorf("first write = %q, want %q", got, pingMsg)
+	}
+	cancel()
+}
+
+func TestNormalizeSource(t *testing.T) {
+	tests := map[string]string{
+		"detail":       "detail",
+		"message_list": "message_list",
+		"":             "other",
+		"bogus":        "other",
+	}
+	for in, want := range tests {
+		if got := normalizeSource(in); got != want {
+			t.Errorf("normalizeSource(%q) = %q, want %q", in, got, want)
+		}
+	}
+}

examples/streaming-daemon-go/config/config.go

@@ -27,7 +27,7 @@ type ServerConfig struct {
 	MaxConnections      int    `yaml:"max_connections"`
 	MaxStreamDurationMs int    `yaml:"max_stream_duration_ms"`
 	PprofAddr           string `yaml:"pprof_addr"`
-	MemLimit            string `yaml:"mem_limit"`   // Soft memory limit, e.g. "768MiB", "1GiB"
+	MemLimit            string `yaml:"mem_limit"`  // Soft memory limit, e.g. "768MiB", "1GiB"
 	GCPercent           int    `yaml:"gc_percent"` // GOGC value; 0 = not set (use -gc-percent flag for GOGC=0)
 	DataDir             string `yaml:"data_dir"`   // Allowed directory for attachment file reads (default: /run/handoff-data)
 }
@@ -38,10 +38,11 @@ type BackendConfig struct {
 	DefaultModel string `yaml:"default_model"`
 
 	// Backend-specific configuration sections
-	OpenAI    OpenAIConfig    `yaml:"openai"`
-	LangGraph LangGraphConfig `yaml:"langgraph"`
-	Mock      MockConfig      `yaml:"mock"`
-	Typing    TypingConfig    `yaml:"typing"`
+	OpenAI      OpenAIConfig      `yaml:"openai"`
+	LangGraph   LangGraphConfig   `yaml:"langgraph"`
+	Mock        MockConfig        `yaml:"mock"`
+	Typing      TypingConfig      `yaml:"typing"`
+	NoopMonitor NoopMonitorConfig `yaml:"noop_monitor"`
 }
 
 // OpenAIConfig contains OpenAI-compatible API settings.
@@ -92,6 +93,11 @@ type TypingConfig struct {
 	DefaultModel string `yaml:"default_model"`
 }
 
+// NoopMonitorConfig contains noop-monitor backend settings.
+type NoopMonitorConfig struct {
+	PingIntervalMs int `yaml:"ping_interval_ms"` // keepalive interval; 0 = default (25000)
+}
+
 // MetricsConfig contains Prometheus metrics settings.
 type MetricsConfig struct {
 	Enabled    bool   `yaml:"enabled"`
@@ -207,6 +213,11 @@ func (c *Config) Validate() error {
 		return fmt.Errorf("backend.mock.message_delay_ms must be non-negative")
 	}
 
+	// Validate noop-monitor backend
+	if c.Backend.NoopMonitor.PingIntervalMs < 0 {
+		return fmt.Errorf("backend.noop_monitor.ping_interval_ms must be non-negative")
+	}
+
 	return nil
 }
 
@@ -297,6 +308,9 @@ func Default() *Config {
 			Mock: MockConfig{
 				MessageDelayMs: 50,
 			},
+			NoopMonitor: NoopMonitorConfig{
+				PingIntervalMs: 25000,
+			},
 		},
 		Metrics: MetricsConfig{
 			Enabled:    true,

examples/streaming-daemon-go/config/example.yaml

@@ -64,6 +64,11 @@ backend:
   # Typing backend configuration
   typing: {}
 
+  # Noop-monitor backend configuration (connection monitor).
+  # Holds connections open sending only keepalives; measurement is via metrics.
+  noop_monitor:
+    ping_interval_ms: 25000  # Keepalive interval (also the disconnect-detection cadence)
+
 metrics:
   enabled: true
   listen_addr: 127.0.0.1:9090