fix(node): stop rendering saved-on-phone nodes as freshly heard

core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/MeshConfigFlowManagerImpl.kt

@@ -298,8 +298,13 @@ class MeshConfigFlowManagerImpl(
             nodeManager.applyTrustedIdentityMigrations(removedNums)
         }
 
+        // Exactly what this session's Stage 2 handshake downloaded (+ the local node) — the UI's "is this node part
+        // of the connected radio's own NodeDB right now" signal (#6263). Published before setNodeDbReady(true) so no
+        // reader can observe "ready" without an up-to-date snapshot to compare against.
+        val currentSessionNodeNums = entities.mapTo(mutableSetOf()) { it.num }.apply { add(info.myNodeNum) }
         val published =
             runForSession(session) {
+                nodeManager.publishCurrentSessionNodeNums(session.generation, currentSessionNodeNums)
                 nodeManager.setNodeDbReady(true)
                 nodeManager.setAllowNodeDbWrites(true)
                 serviceStateWriter.setConnectionState(ConnectionState.Connected)

core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/NodeManagerImpl.kt

@@ -317,18 +317,60 @@ class NodeManagerImpl(
 
     override fun clearConnectionIdentity() {
         _connectionIdentity.value = null
+        currentSessionNodeNumsGeneration.value = NO_SESSION_NODE_NUMS_GENERATION
+        _currentSessionNodeNums.value = null
     }
 
     override fun clearStaleConnectionIdentity(activeSessionGeneration: Long) {
         _connectionIdentity.updateStateFlow { identity ->
             identity?.takeIf { it.sessionGeneration == activeSessionGeneration }
         }
+        // Same reconciliation as connectionIdentity: a snapshot already published for the active generation must
+        // survive a delayed boundary collector from that same generation (RadioControllerImpl's sessionGeneration
+        // collector can fire after installAndPublishNodeDatabase already published for the new session).
+        if (currentSessionNodeNumsGeneration.value != activeSessionGeneration) {
+            _currentSessionNodeNums.value = null
+        }
     }
 
     override fun publishConnectionIdentity(sessionGeneration: Long, address: String, nodeNum: Int, deviceId: String?) {
         _connectionIdentity.value = ConnectionIdentity(sessionGeneration, address, nodeNum, deviceId)
     }
 
+    private val currentSessionNodeNumsGeneration = atomic(NO_SESSION_NODE_NUMS_GENERATION)
+    private val _currentSessionNodeNums = MutableStateFlow<Set<Int>?>(null)
+    override val currentSessionNodeNums: StateFlow<Set<Int>?> = _currentSessionNodeNums
+
+    override fun publishCurrentSessionNodeNums(sessionGeneration: Long, nodeNums: Set<Int>) {
+        currentSessionNodeNumsGeneration.value = sessionGeneration
+        _currentSessionNodeNums.value = nodeNums
+    }
+
+    /**
+     * Extends the current session's membership set with [nodeNum] after the radio forwarded live traffic for it, so a
+     * node that announces itself *after* the Stage 2 snapshot is not mistaken for locally-retained history (#6263).
+     *
+     * A null [session] means the mutation did not come from the radio at all (an optimistic admin projection, a
+     * shared-contact import, a locally applied fixed position) and must not claim session membership. Everything else
+     * is decided inside the [updateStateFlow] lambda so a concurrent session boundary makes this a no-op rather than
+     * resurrecting a number into the wrong session's set:
+     * - no snapshot published yet → nothing to extend (leaving it null keeps every row unbadged, per the contract);
+     * - the published snapshot belongs to a different generation → this packet is from a superseded session;
+     * - already a member → return the same instance, so the hot inbound path neither copies a set of up to
+     *   [MAX_IN_MEMORY_NODES] entries nor emits a conflated no-change update.
+     */
+    private fun noteHeardInSession(nodeNum: Int, session: RadioSessionContext?) {
+        if (session == null) return
+        _currentSessionNodeNums.updateStateFlow { current ->
+            when {
+                current == null -> null
+                currentSessionNodeNumsGeneration.value != session.generation -> current
+                nodeNum in current -> current
+                else -> current + nodeNum
+            }
+        }
+    }
+
     override val firmwareEdition = MutableStateFlow<FirmwareEdition?>(null)
 
     override fun setFirmwareEdition(edition: FirmwareEdition?) {
@@ -401,6 +443,9 @@ class NodeManagerImpl(
          * legitimately busy mesh never reaches it and only sustained novel-`from` traffic does.
          */
         const val MAX_IN_MEMORY_NODES = 2_000
+
+        /** Sentinel for [currentSessionNodeNumsGeneration] meaning "no snapshot published yet this process". */
+        private const val NO_SESSION_NODE_NUMS_GENERATION = -1L
     }
 
     override fun loadCachedNodeDB() {
@@ -475,6 +520,8 @@ class NodeManagerImpl(
         myDeviceId.value = null
         firmwareEdition.value = null
         _connectionIdentity.value = null
+        currentSessionNodeNumsGeneration.value = NO_SESSION_NODE_NUMS_GENERATION
+        _currentSessionNodeNums.value = null
     }
 
     override fun getMyNodeInfo(): MyNodeInfo? {
@@ -590,7 +637,13 @@ class NodeManagerImpl(
         session: RadioSessionContext? = null,
         transform: (Node) -> Node,
     ): NodeStateChange? = updateNodeState(nodeNum, channel, transform).also { change ->
-        if (change != null && shouldPersist(change.next)) {
+        if (change == null) return@also
+        // A committed session-scoped update means the radio just forwarded traffic for this node — position,
+        // telemetry, node status, PaxCounter, admin reply — so it belongs to this session's membership set even if
+        // Stage 2 never listed it (#6263). A null change means the update was refused (retired-absent number) and
+        // proves nothing. Retries are impossible here: updateNodeState commits at most once.
+        noteHeardInSession(nodeNum, session)
+        if (shouldPersist(change.next)) {
             radioInterfaceService.launchSessionWork(scope, session) { persistLatestNode(nodeNum) }
         }
     }
@@ -668,6 +721,9 @@ class NodeManagerImpl(
                         " key=$keyStr canonical=$canonicalNum" +
                         " decision=${transition.decision} notify=${transition.notifyNode != null}"
                 }
+                // Only the winning CAS may claim session membership: a packet that lost the race breaks out of this
+                // loop and is logged as discarded, and an entry-point add would have already polluted the set (#6263).
+                transition.sessionMemberNodeNum?.let { noteHeardInSession(it, session) }
                 applyReceivedUserEffects(transition, session)
                 return
             }
@@ -835,6 +891,14 @@ class NodeManagerImpl(
         val notifyNode: Node?,
         /** Retired number to reactivate after a validated, genuinely new identity claims the vacant slot. */
         val unretireNodeNum: Int? = null,
+        /**
+         * Number under which [after] actually keys the node this packet was attributed to, to be recorded as a member
+         * of the live connection session (#6263). Deliberately not always `fromNum`: a stale noncanonical presentation
+         * yields to the canonical row, so the *canonical* number is the row the traffic evidences. Null when the
+         * reduction suppressed or ignored the packet — a suppressed replay proves nothing about session membership, and
+         * claiming `fromNum` there would badge a slot held by an entirely different identity.
+         */
+        val sessionMemberNodeNum: Int? = null,
         val decision: ReceivedUserDecision,
     )
 
@@ -932,6 +996,7 @@ class NodeManagerImpl(
                 after = afterRemovals.put(fromNum, transformed, preferredNum = fromNum),
                 upsertNode = transformed,
                 notifyNode = null,
+                sessionMemberNodeNum = fromNum,
                 decision = ReceivedUserDecision.LOCAL_UPDATE,
             )
         }
@@ -985,6 +1050,7 @@ class NodeManagerImpl(
                     after = afterRemovals.put(fromNum, transformed, preferredNum = fromNum),
                     upsertNode = null,
                     notifyNode = null,
+                    sessionMemberNodeNum = fromNum,
                     decision = ReceivedUserDecision.CANONICAL_DUPLICATE_RECONCILED,
                 )
             }
@@ -998,6 +1064,9 @@ class NodeManagerImpl(
                     after = after,
                     upsertNode = null,
                     notifyNode = null,
+                    // The identity lives at canonicalNum (that is what `canonicalNum in otherSameKeyNums` asserts), so
+                    // that — not the yielding fromNum slot — is the row this traffic proves the radio just relayed.
+                    sessionMemberNodeNum = canonicalNum,
                     decision = ReceivedUserDecision.STALE_PRESENTATION_REMOVED,
                 )
             }
@@ -1026,6 +1095,7 @@ class NodeManagerImpl(
                     after = before.put(fromNum, transformed, preferredNum = fromNum),
                     upsertNode = null,
                     notifyNode = null,
+                    sessionMemberNodeNum = fromNum,
                     decision = ReceivedUserDecision.AMBIGUOUS_DUPLICATE_UPDATED,
                 )
             }
@@ -1068,6 +1138,7 @@ class NodeManagerImpl(
             upsertNode = transformed.takeIf { persist },
             notifyNode = notify,
             unretireNodeNum = unretireNodeNum,
+            sessionMemberNodeNum = fromNum,
             decision = decision,
         )
     }

core/data/src/commonTest/kotlin/org/meshtastic/core/data/manager/MeshConfigFlowManagerImplTest.kt

@@ -627,6 +627,45 @@ class MeshConfigFlowManagerImplTest {
         verifySuspend { connectionManager.onNodeDbReady() }
     }
 
+    @Test
+    fun `Stage 2 complete publishes exact session membership before readiness`() = testScope.runTest {
+        val firstNum = 100
+        val secondNum = 200
+        val firstNode = org.meshtastic.core.testing.TestDataFactory.createTestNode(num = firstNum)
+        val secondNode = org.meshtastic.core.testing.TestDataFactory.createTestNode(num = secondNum)
+        every { nodeManager.nodeDBbyNodeNum } returns mapOf(firstNum to firstNode, secondNum to secondNode)
+        val callOrder = mutableListOf<String>()
+        every { nodeManager.publishCurrentSessionNodeNums(any(), any()) } calls
+            {
+                callOrder.add("publishSessionNodeNums")
+            }
+        every { nodeManager.setNodeDbReady(true) } calls { callOrder.add("nodeDbReady") }
+
+        handleMyInfo(protoMyNodeInfo)
+        advanceUntilIdle()
+        manager.handleLocalMetadata(metadata)
+        advanceUntilIdle()
+        manager.handleConfigComplete(HandshakeConstants.CONFIG_NONCE)
+        advanceTimeBy(STAGE_TRANSITION_ADVANCE_MS)
+        runCurrent()
+        manager.handleNodeInfo(NodeInfo(num = firstNum))
+        manager.handleNodeInfo(NodeInfo(num = secondNum))
+        manager.handleConfigComplete(HandshakeConstants.NODE_INFO_NONCE)
+        advanceUntilIdle()
+
+        // Exactly the downloaded set plus the local node — not the entire (possibly larger, locally-retained)
+        // nodeDBbyNodeNum, which would defeat the badge's purpose of flagging rows the radio did NOT just report.
+        verify {
+            nodeManager.publishCurrentSessionNodeNums(
+                activeSession.generation,
+                setOf(myNodeNum, firstNum, secondNum),
+            )
+        }
+        // Published strictly before setNodeDbReady(true), so no reader can observe "ready" against a stale/absent
+        // session snapshot.
+        assertEquals(listOf("publishSessionNodeNums", "nodeDbReady"), callOrder)
+    }
+
     @Test
     fun `Stage 2 applies trusted migrations before readiness and replay`() = testScope.runTest {
         val retiredNum = 456