fix(telemetry): report unpaired events, and retract migration 000155's reconciler promise

internal/infra/http/handler/runtime_telemetry_handler.go

@@ -61,8 +61,26 @@ type ingestRequest struct {
 }
 
 type ingestResponse struct {
-	Accepted int      `json:"accepted"`
-	Rejected int      `json:"rejected"`
+	Accepted int `json:"accepted"`
+	Rejected int `json:"rejected"`
+
+	// Unpaired counts ACCEPTED events that carried no endpoint_asset_id.
+	// They are stored and the IOC correlator still matches them, because it
+	// keys on values inside the event. They are invisible to every
+	// asset-scoped read: Stage-4 detection correlation's heuristic fallback
+	// and the per-asset Stage-6 dashboards.
+	//
+	// This is permanent, not a pending state. There is no server-side way to
+	// fill it in later — `agents` has no asset column and `assets` has no
+	// agent column, and only the producer knows which endpoint an event
+	// describes anyway (a forwarder reports on many hosts). Migration 000155
+	// once promised a nightly reconciler; it was never written and could not
+	// have been.
+	//
+	// Reported so a producer sees the degradation on the response it already
+	// reads, rather than discovering months later that half the feature never
+	// applied to its data.
+	Unpaired int      `json:"unpaired"`
 	Errors   []string `json:"errors,omitempty"`
 }
 
@@ -188,6 +206,9 @@ func (h *RuntimeTelemetryHandler) Ingest(w http.ResponseWriter, r *http.Request)
 			continue
 		}
 		resp.Accepted++
+		if ev.EndpointAssetID == "" {
+			resp.Unpaired++
+		}
 
 		if eventID, parseErr := shared.IDFromString(eventIDStr); parseErr == nil {
 			accepted = append(accepted, iocapp.TelemetryEvent{
@@ -198,6 +219,20 @@ func (h *RuntimeTelemetryHandler) Ingest(w http.ResponseWriter, r *http.Request)
 		}
 	}
 
+	// Surface the degradation in the logs too. A producer that never sends
+	// endpoint_asset_id gets a fully successful 200 with a healthy accepted
+	// count, and would have no reason to suspect that asset-scoped correlation
+	// silently does not apply to any of its data.
+	if resp.Unpaired > 0 {
+		h.logger.Warn("runtime telemetry accepted without an endpoint asset link",
+			"tenant_id", agt.TenantID.String(),
+			"agent_id", agt.ID.String(),
+			"unpaired", resp.Unpaired,
+			"accepted", resp.Accepted,
+			"impact", "invisible to asset-scoped detection correlation and per-asset dashboards; "+
+				"the producer must supply endpoint_asset_id, the server cannot infer it")
+	}
+
 	// B6 wire: ONE batch correlate call for the whole accepted slice.
 	// Correlator dedups candidates internally and runs a single
 	// FindActiveByValues query; errors here are logged but never block

internal/infra/http/handler/runtime_telemetry_unpaired_test.go

@@ -0,0 +1,212 @@
+package handler
+
+import (
+	"bytes"
+	"context"
+	"database/sql"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"testing"
+	"time"
+
+	_ "github.com/lib/pq"
+
+	"github.com/openctemio/api/pkg/domain/agent"
+	"github.com/openctemio/api/pkg/domain/shared"
+	"github.com/openctemio/api/pkg/logger"
+)
+
+// endpoint_asset_id is nullable, and nothing will ever fill it in later.
+//
+// Migration 000155 promised "a nightly reconciler job pairs events with assets
+// by agent_id". That job was never written and could not be: `agents` has no
+// asset column, `assets` has no agent column, and there is no join table — so
+// there is no key to pair BY. It is also the wrong idea, because only the
+// producer knows which endpoint an event describes; an EDR/XDR forwarder
+// reports on many hosts.
+//
+// So an event without an asset link is permanently invisible to every
+// asset-scoped read — Stage-4 detection correlation's heuristic fallback and
+// the per-asset Stage-6 dashboards — while the response says "accepted" and the
+// IOC correlator, which keys on values inside the event, still works. That is a
+// half-working feature that looks fully working.
+//
+// These tests pin the counter that makes it visible.
+
+func openTelemetryDB(t *testing.T) *sql.DB {
+	t.Helper()
+
+	dbURL := os.Getenv("DATABASE_URL")
+	if dbURL == "" {
+		t.Skip("DATABASE_URL not set; skipping telemetry ingest tests")
+	}
+	db, err := sql.Open("postgres", dbURL)
+	if err != nil {
+		t.Skipf("open db: %v", err)
+	}
+	t.Cleanup(func() { _ = db.Close() })
+	if err := db.PingContext(context.Background()); err != nil {
+		t.Skipf("cannot reach DATABASE_URL: %v", err)
+	}
+	return db
+}
+
+func seedTelemetryTenant(t *testing.T, db *sql.DB) shared.ID {
+	t.Helper()
+
+	id := shared.NewID()
+	if _, err := db.ExecContext(context.Background(),
+		`INSERT INTO tenants (id, name, slug) VALUES ($1, $2, $3)`,
+		id.String(), "telemetry unpaired test", "tel-"+id.String()); err != nil {
+		t.Fatalf("seed tenant: %v", err)
+	}
+	t.Cleanup(func() {
+		_, _ = db.ExecContext(context.Background(), `DELETE FROM tenants WHERE id = $1`, id.String())
+	})
+	return id
+}
+
+func seedTelemetryAsset(t *testing.T, db *sql.DB, tenantID shared.ID) shared.ID {
+	t.Helper()
+
+	id := shared.NewID()
+	if _, err := db.ExecContext(context.Background(),
+		`INSERT INTO assets (id, tenant_id, name, asset_type) VALUES ($1, $2, $3, 'host')`,
+		id.String(), tenantID.String(), "host-"+id.String()); err != nil {
+		t.Fatalf("seed asset: %v", err)
+	}
+	return id
+}
+
+// ingestEvents posts a batch as an authenticated tenant agent and returns the
+// decoded response.
+func ingestEvents(t *testing.T, db *sql.DB, tenantID shared.ID, events []map[string]any) ingestResponse {
+	t.Helper()
+
+	body, err := json.Marshal(map[string]any{"events": events})
+	if err != nil {
+		t.Fatalf("marshal body: %v", err)
+	}
+
+	h := NewRuntimeTelemetryHandler(db, logger.NewNop())
+
+	r := httptest.NewRequest(http.MethodPost, "/api/v1/telemetry-events", bytes.NewReader(body))
+	tid := tenantID
+	agt := &agent.Agent{ID: shared.NewID(), TenantID: &tid, Status: agent.AgentStatusActive}
+	r = r.WithContext(context.WithValue(r.Context(), agentContextKey, agt))
+
+	w := httptest.NewRecorder()
+	h.Ingest(w, r)
+
+	// 202 on a fully-accepted batch, 207 when some events were rejected.
+	if w.Code != http.StatusAccepted && w.Code != http.StatusMultiStatus {
+		t.Fatalf("ingest returned %d: %s", w.Code, w.Body.String())
+	}
+
+	var resp ingestResponse
+	if err := json.NewDecoder(w.Body).Decode(&resp); err != nil {
+		t.Fatalf("decode response: %v", err)
+	}
+	t.Cleanup(func() {
+		_, _ = db.ExecContext(context.Background(),
+			`DELETE FROM runtime_telemetry_events WHERE tenant_id = $1`, tenantID.String())
+	})
+	return resp
+}
+
+func event(assetID string) map[string]any {
+	e := map[string]any{
+		"event_type":  "network_connect",
+		"observed_at": time.Now().UTC().Format(time.RFC3339),
+	}
+	if assetID != "" {
+		e["endpoint_asset_id"] = assetID
+	}
+	return e
+}
+
+// The core case: a producer that never sends endpoint_asset_id gets a fully
+// successful response. Without the counter there is nothing in that response to
+// tell it half the feature does not apply.
+func TestIngest_ReportsUnpairedEvents(t *testing.T) {
+	db := openTelemetryDB(t)
+	tenantID := seedTelemetryTenant(t, db)
+
+	resp := ingestEvents(t, db, tenantID, []map[string]any{
+		event(""), event(""), event(""),
+	})
+
+	if resp.Accepted != 3 {
+		t.Fatalf("accepted = %d, want 3 (errors: %v)", resp.Accepted, resp.Errors)
+	}
+	if resp.Rejected != 0 {
+		t.Fatalf("rejected = %d, want 0: unpaired events are stored, not refused", resp.Rejected)
+	}
+	if resp.Unpaired != 3 {
+		t.Fatalf("unpaired = %d, want 3: the response claims full success while every "+
+			"event is invisible to asset-scoped correlation", resp.Unpaired)
+	}
+}
+
+// A producer doing it right must not be told it has a problem.
+func TestIngest_PairedEventsAreNotCountedUnpaired(t *testing.T) {
+	db := openTelemetryDB(t)
+	tenantID := seedTelemetryTenant(t, db)
+	assetID := seedTelemetryAsset(t, db, tenantID)
+
+	resp := ingestEvents(t, db, tenantID, []map[string]any{
+		event(assetID.String()), event(assetID.String()),
+	})
+
+	if resp.Accepted != 2 {
+		t.Fatalf("accepted = %d, want 2 (errors: %v)", resp.Accepted, resp.Errors)
+	}
+	if resp.Unpaired != 0 {
+		t.Fatalf("unpaired = %d, want 0: these events carry a valid asset link", resp.Unpaired)
+	}
+}
+
+// A mixed batch is the realistic case — the count must be per-event, not a
+// boolean about the batch.
+func TestIngest_CountsUnpairedPerEvent(t *testing.T) {
+	db := openTelemetryDB(t)
+	tenantID := seedTelemetryTenant(t, db)
+	assetID := seedTelemetryAsset(t, db, tenantID)
+
+	resp := ingestEvents(t, db, tenantID, []map[string]any{
+		event(assetID.String()), event(""), event(assetID.String()), event(""),
+	})
+
+	if resp.Accepted != 4 {
+		t.Fatalf("accepted = %d, want 4 (errors: %v)", resp.Accepted, resp.Errors)
+	}
+	if resp.Unpaired != 2 {
+		t.Fatalf("unpaired = %d, want 2", resp.Unpaired)
+	}
+}
+
+// Unpaired counts ACCEPTED events only. A rejected event was never stored, so
+// counting it here would overstate the gap and send a producer looking for a
+// configuration problem that is really a validation error.
+func TestIngest_RejectedEventsAreNotCountedUnpaired(t *testing.T) {
+	db := openTelemetryDB(t)
+	tenantID := seedTelemetryTenant(t, db)
+
+	resp := ingestEvents(t, db, tenantID, []map[string]any{
+		{"observed_at": time.Now().UTC().Format(time.RFC3339)}, // no event_type -> rejected
+		event(""), // accepted, unpaired
+	})
+
+	if resp.Rejected != 1 {
+		t.Fatalf("rejected = %d, want 1", resp.Rejected)
+	}
+	if resp.Accepted != 1 {
+		t.Fatalf("accepted = %d, want 1", resp.Accepted)
+	}
+	if resp.Unpaired != 1 {
+		t.Fatalf("unpaired = %d, want 1: a rejected event was never stored and must "+
+			"not be reported as an unpaired one", resp.Unpaired)
+	}
+}

migrations/000155_runtime_telemetry.up.sql

@@ -14,9 +14,27 @@
 --   - properties JSONB holds event-specific fields. Kept intentionally
 --     schemaless so agents on different OSes (Windows EDR, Linux
 --     osquery, …) can emit without a wire-format migration every time.
---   - endpoint_asset_id is nullable — during onboarding the agent may
---     not yet know its asset UUID. A nightly reconciler job pairs
---     events with assets by agent_id.
+--   - endpoint_asset_id is nullable — during onboarding the producer may
+--     not yet know its asset UUID.
+--
+--     CORRECTION (2026-08-04): an earlier version of this comment promised
+--     "a nightly reconciler job pairs events with assets by agent_id".
+--     No such job was ever written, and it cannot be: there is no join
+--     key. `agents` has no asset column, `assets` has no agent column,
+--     and there is no join table — so there is nothing to pair BY.
+--
+--     It is also the wrong idea. Only the producer knows which endpoint
+--     an event describes. An EDR/XDR forwarder reports on MANY hosts, so
+--     even the emitting agent's own hostname is not the answer. The
+--     server cannot infer this after the fact.
+--
+--     So a NULL here is permanent. Such an event is still stored and
+--     still matched by the IOC correlator (which keys on values inside
+--     the event, not on the asset), but it is invisible to every
+--     asset-scoped read: Stage-4 detection correlation's heuristic
+--     fallback and the per-asset Stage-6 dashboards. The ingest response
+--     reports these as `unpaired` so a producer sees the degradation
+--     instead of silently losing half the feature.
 --   - (tenant_id, observed_at) compound index — all downstream reads
 --     are per-tenant, time-ordered.