fix(observers): an observer's own transmissions are not receptions (#667)

cmd/ingestor/neighbor_builder.go

@@ -198,6 +198,7 @@ func (s *Store) buildAndPersistNeighborEdges(trust *packetpath.TrustConfig) (int
 	JOIN transmissions t ON t.id = o.transmission_id
 	LEFT JOIN observers obs ON obs.rowid = o.observer_idx
 	WHERE o.timestamp > ?
+	AND o.direction IS NOT 'tx' -- the observer's own transmission, not a reception (#667)
 	ORDER BY o.timestamp
 	LIMIT ?`, watermarkEpoch, neighborBuilderMaxBatch)
 	if err != nil {

cmd/ingestor/neighbor_builder_test.go

@@ -332,3 +332,55 @@ func TestNeighborEdgesBuilderPathTrustAllowsTwoByte(t *testing.T) {
 		t.Fatalf("2-byte hop must still produce the edge, got %d", got)
 	}
 }
+
+// TestNeighborEdgesBuilderSkipsOwnTransmissions pins #667: an observation
+// with direction 'tx' is the observer's own radio transmitting, not a
+// reception, so it must not produce an edge.
+func TestNeighborEdgesBuilderSkipsOwnTransmissions(t *testing.T) {
+	store, err := OpenStore(filepath.Join(t.TempDir(), "tx.db"))
+	if err != nil {
+		t.Fatalf("OpenStore: %v", err)
+	}
+	defer store.Close()
+
+	if _, err := store.db.Exec(
+		`INSERT INTO nodes (public_key, name) VALUES (?, ?), (?, ?)`,
+		"aaaaaaaaaa", "from-node",
+		"bbbbbbbbbb", "first-hop",
+	); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := store.db.Exec(`INSERT INTO observers (id, name) VALUES (?, ?)`, "obs-1", "observer-1"); err != nil {
+		t.Fatal(err)
+	}
+	var obsRowid int64
+	if err := store.db.QueryRow(`SELECT rowid FROM observers WHERE id = ?`, "obs-1").Scan(&obsRowid); err != nil {
+		t.Fatal(err)
+	}
+	res, err := store.db.Exec(
+		`INSERT INTO transmissions (raw_hex, hash, first_seen, route_type, payload_type, payload_version, decoded_json, from_pubkey)
+		 VALUES (?, ?, ?, ?, ?, ?, ?, ?)`,
+		"", "h-tx", "2026-01-01T00:00:00Z", 0, payloadADVERT, 0, "{}", "aaaaaaaaaa",
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+	txID, _ := res.LastInsertId()
+	if _, err := store.db.Exec(
+		`INSERT INTO observations (transmission_id, observer_idx, direction, path_json, timestamp) VALUES (?, ?, 'tx', ?, ?)`,
+		txID, obsRowid, `["bb"]`, int64(1735689600),
+	); err != nil {
+		t.Fatal(err)
+	}
+
+	if _, err := store.buildAndPersistNeighborEdges(trustAllPrefixes()); err != nil {
+		t.Fatalf("buildAndPersistNeighborEdges: %v", err)
+	}
+	var got int
+	if err := store.db.QueryRow(`SELECT COUNT(*) FROM neighbor_edges`).Scan(&got); err != nil {
+		t.Fatal(err)
+	}
+	if got != 0 {
+		t.Fatalf("a tx observation produced %d neighbor_edges rows, want 0", got)
+	}
+}

cmd/server/direct_heard.go

@@ -157,7 +157,7 @@ func buildDirectHeardIndex(packets []*StoreTx, pm *prefixMap) directHeardIndex {
 	idx := make(directHeardIndex, 256)
 	for _, tx := range packets {
 		for _, obs := range tx.Observations {
-			if obs.ObserverID == "" {
+			if obs.ObserverID == "" || obs.isOwnTransmission() {
 				continue
 			}
 			pk := directHeardNode(tx, obs, pm)

cmd/server/direct_heard_test.go

@@ -308,3 +308,29 @@ func TestBuildDirectObserverRows_CanRelayTriState(t *testing.T) {
 		t.Fatalf("repeater must report can_relay true, got %v", got["repeater"])
 	}
 }
+
+// #667: a "tx" observation is the observer's own radio transmitting, not a
+// reception. When a repeater-observer forwards a flood it appends its own
+// hash first (firmware Mesh.cpp), so the tx copy ends on the observer itself,
+// and its own advert goes out with an empty path. Counted as receptions, both
+// made the observer the top entry of its own "Heard By" card.
+func TestBuildDirectHeardIndex_SkipsOwnTransmissions(t *testing.T) {
+	forwarded := dhTx(RouteFlood, PayloadTXT_MSG, "")
+	forwarded.Observations = []*StoreObs{
+		{ObserverID: dhUnique, ObserverName: "self", Direction: "tx", PathJSON: `["A433","BB11"]`},
+		{ObserverID: dhUnique, ObserverName: "self", Direction: "rx", PathJSON: `["A433"]`},
+	}
+	ownAdvert := dhTx(RouteFlood, PayloadADVERT, `{"type":"ADVERT","pubKey":"`+dhUnique+`"}`)
+	ownAdvert.Observations = []*StoreObs{
+		{ObserverID: dhUnique, ObserverName: "self", Direction: "tx", PathJSON: `[]`},
+	}
+
+	idx := buildDirectHeardIndex([]*StoreTx{forwarded, ownAdvert}, buildPrefixMap(directHeardNodes))
+
+	if got := idx[dhUnique]; len(got) != 0 {
+		t.Fatalf("own transmissions credited as receptions: %#v", got)
+	}
+	if a := idx[dhCollideA][dhUnique]; a == nil || a.Count != 1 {
+		t.Fatalf("rx copy of the forwarded packet = %#v, want Count 1", a)
+	}
+}

cmd/server/neighbor_graph.go

@@ -328,6 +328,9 @@ func BuildFromStoreWithOptions(store *PacketStore, opts BuildOptions) *NeighborG
 		}
 
 		for _, obs := range p.observations {
+			if obs.isOwnTransmission() {
+				continue
+			}
 			path := parsePathJSON(obs.PathJSON)
 			observerPK := cachedToLower(lowerCache, obs.ObserverID)
 

cmd/server/neighbor_graph_test.go

@@ -1019,3 +1019,37 @@ func TestBuildNeighborGraph_CountsByMode(t *testing.T) {
 		t.Errorf("CountsByMode[3] = %d, want 1", got)
 	}
 }
+
+// #667: the tx copy of a packet the observer forwarded carries the same
+// originator↔path[0] evidence as its rx copy, so counting it doubles the edge.
+// A tx row with an empty path is not a zero-hop hearing either.
+func TestBuildNeighborGraph_SkipsOwnTransmissions(t *testing.T) {
+	nodes := []nodeInfo{
+		{Role: "repeater", PublicKey: "aaaa1111", Name: "NodeX"},
+		{Role: "repeater", PublicKey: "r1aabbcc", Name: "R1"},
+		{Role: "repeater", PublicKey: "obs00001", Name: "Observer"},
+	}
+	rx := ngMakeObs("obs00001", `["r1aa"]`, nowStr, nil)
+	rx.Direction = "rx"
+	tx := ngMakeObs("obs00001", `["r1aa","obs0"]`, nowStr, nil)
+	tx.Direction = "tx"
+	emptyTx := ngMakeObs("obs00001", `[]`, nowStr, nil)
+	emptyTx.Direction = "tx"
+
+	store := ngTestStore(nodes, []*StoreTx{
+		ngMakeTx(1, 4, ngFromNodeJSON("aaaa1111"), []*StoreObs{rx, tx}),
+		ngMakeTx(2, 4, ngFromNodeJSON("aaaa1111"), []*StoreObs{emptyTx}),
+	})
+	g := BuildFromStore(store)
+
+	for _, e := range g.AllEdges() {
+		if (e.NodeA == "aaaa1111" && e.NodeB == "r1aabbcc") || (e.NodeA == "r1aabbcc" && e.NodeB == "aaaa1111") {
+			if e.Count != 1 {
+				t.Errorf("X↔R1 count = %d, want 1 (the tx copy must not count)", e.Count)
+			}
+		}
+		if e.NodeA == "aaaa1111" && e.NodeB == "obs00001" || e.NodeA == "obs00001" && e.NodeB == "aaaa1111" {
+			t.Errorf("empty-path tx row produced a zero-hop edge X↔Observer")
+		}
+	}
+}

cmd/server/store.go

@@ -75,6 +75,11 @@ type StoreTx struct {
 	accountedBytes int64
 }
 
+// isOwnTransmission reports whether the observation is the observer's own
+// radio transmitting (direction "tx" from the MQTT bridge) rather than a
+// reception. Aggregates of who heard whom must skip it (#667).
+func (o *StoreObs) isOwnTransmission() bool { return o.Direction == "tx" }
+
 // StoreObs is a lean in-memory observation (no duplication of transmission fields).
 type StoreObs struct {
 	ID             int