fix(takserver): route TAK self-test through the real v1/v2 dispatch path (#6746)

Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com>

.skills/compose-ui/strings-index.txt

@@ -1699,9 +1699,12 @@ tak_server_status_unavailable
 tak_server_status_waiting
 tak_server_test_card_title
 tak_server_test_idle
+tak_server_test_protocol_v1_label
+tak_server_test_protocol_v2_label
 tak_server_test_result_bytes
+tak_server_test_result_expected_drop
 tak_server_test_result_unknown_error
-tak_server_test_results
+tak_server_test_results_v2
 tak_server_test_run
 tak_server_test_running
 tak_team

core/resources/src/commonMain/composeResources/values/strings.xml

@@ -1747,9 +1747,12 @@
     <string name="tak_server_status_waiting">Listening on 127.0.0.1:8089 — waiting for ATAK</string>
     <string name="tak_server_test_card_title">TAK Mesh Test (Debug)</string>
     <string name="tak_server_test_idle">Send all %1$d test fixtures to mesh</string>
+    <string name="tak_server_test_protocol_v1_label">V1 (legacy firmware &lt; 2.8.0 — PLI + chat only)</string>
+    <string name="tak_server_test_protocol_v2_label">V2 (firmware ≥ 2.8.0)</string>
     <string name="tak_server_test_result_bytes">%1$dB ✓</string>
+    <string name="tak_server_test_result_expected_drop">Not in v1 schema (expected)</string>
     <string name="tak_server_test_result_unknown_error">✗</string>
-    <string name="tak_server_test_results">%1$d passed, %2$d failed of %3$d/%4$d</string>
+    <string name="tak_server_test_results_v2">%1$d passed, %2$d expected drop, %3$d failed of %4$d/%5$d</string>
     <string name="tak_server_test_run">Run</string>
     <string name="tak_server_test_running">Running: %1$s</string>
     <string name="tak_team">Team Color</string>

core/takserver/src/commonMain/kotlin/org/meshtastic/core/takserver/TAKMeshIntegration.kt

@@ -47,6 +47,30 @@ import kotlin.random.Random
 import kotlin.time.Clock
 import kotlin.time.Duration.Companion.minutes
 
+/**
+ * Outcome of a single outbound CoT dispatch attempt via [TAKMeshIntegration.sendCoTToMeshV1] /
+ * [TAKMeshIntegration.sendCoTToMeshV2]. Exposed (internal) so [TakMeshTestRunner] can distinguish an intentional,
+ * schema-driven drop (e.g. v1's TAKPacket only representing PLI/GeoChat) from a genuine send failure, rather than
+ * reporting both as the same opaque "failed".
+ */
+internal sealed interface TakSendOutcome {
+    /** The CoT was successfully handed to [CommandSender.sendData] as [wireBytes] bytes. */
+    data class Sent(val wireBytes: Int) : TakSendOutcome
+
+    /**
+     * The CoT was never sent because the active protocol's schema/size limits can't represent it.
+     *
+     * @param schemaLimited true only when the drop is because the protocol's CoT type coverage doesn't include this
+     *   fixture's type (e.g. v1's TAKPacket only representing PLI/GeoChat) — a known, permanent limitation of that
+     *   protocol version. False for an oversize drop or any other reason: those are real MTU/size problems that happen
+     *   to hit a payload that *is* representable, and must not be reported as an expected/intentional limitation.
+     */
+    data class Dropped(val reason: String, val schemaLimited: Boolean) : TakSendOutcome
+
+    /** The CoT should have been sent but the attempt threw (radio disconnected, queue full, etc). */
+    data class Failed(val reason: String) : TakSendOutcome
+}
+
 /**
  * Bidirectional bridge between the local TAK server and the Meshtastic mesh network.
  *
@@ -64,6 +88,7 @@ import kotlin.time.Duration.Companion.minutes
  * from older nodes in mixed-firmware mesh deployments.
  */
 @OptIn(ExperimentalAtomicApi::class)
+@Suppress("TooManyFunctions")
 class TAKMeshIntegration(
     private val takServerManager: TAKServerManager,
     private val commandSender: CommandSender,
@@ -173,20 +198,29 @@ class TAKMeshIntegration(
         return Capabilities(fw).supportsTakV2
     }
 
-    private suspend fun sendCoTToMesh(cotMessage: CoTMessage) {
-        if (useTakV2()) {
-            sendCoTToMeshV2(cotMessage)
-        } else {
-            sendCoTToMeshV1(cotMessage)
-        }
+    private suspend fun sendCoTToMesh(cotMessage: CoTMessage): TakSendOutcome = if (useTakV2()) {
+        sendCoTToMeshV2(cotMessage)
+    } else {
+        sendCoTToMeshV1(cotMessage)
     }
 
+    /**
+     * Test-only entry point for [TakMeshTestRunner]'s debug self-test: dispatch [cotMessage] through the real
+     * production v1/v2 pipeline, but with the protocol explicitly forced via [forceV2] instead of read from the
+     * connected radio's firmware (as [useTakV2] does). This lets the self-test exercise BOTH real dispatch paths
+     * deterministically in a single run regardless of which firmware happens to be connected — closing the blind spot
+     * where the self-test always exercised the v2 pipeline even when a real connected radio on firmware < 2.8.0 would
+     * silently fall back to the much more limited v1 path.
+     */
+    internal suspend fun sendCoTToMeshForTest(cotMessage: CoTMessage, forceV2: Boolean): TakSendOutcome =
+        if (forceV2) sendCoTToMeshV2(cotMessage) else sendCoTToMeshV1(cotMessage)
+
     /**
      * v2 send path (firmware >= 2.8.0): SDK parser + zstd dictionary compression, full typed payload support
      * (DrawnShape, Marker, Route, Aircraft, Casevac, Emergency, Task, plus PLI / GeoChat). Wire format: `[flags
      * byte][zstd-compressed TAKPacketV2 protobuf]` on port 78 (ATAK_PLUGIN_V2).
      */
-    private suspend fun sendCoTToMeshV2(cotMessage: CoTMessage) {
+    private suspend fun sendCoTToMeshV2(cotMessage: CoTMessage): TakSendOutcome {
         // Prefer the sourceEventXml for shape/marker/route types — the SDK's
         // CotXmlParser extracts compact typed payloads (DrawnShape, Marker,
         // Route, etc.) that compress far better than raw_detail encoding.
@@ -236,14 +270,20 @@ class TAKMeshIntegration(
                                 }
                             }
                         }
-                        return
+                        return TakSendOutcome.Dropped(
+                            "Oversized (>${MAX_TAK_WIRE_PAYLOAD_BYTES}B)",
+                            schemaLimited = false,
+                        )
                     }
             } catch (e: Exception) {
                 Logger.w(e) { "SDK parser/compressor failed for ${cotMessage.type}, trying app conversion" }
                 val takPacketV2 = cotMessage.toTAKPacketV2()
                 if (takPacketV2 == null) {
                     Logger.w { "Cannot convert CoT type ${cotMessage.type} to TAKPacketV2, dropping" }
-                    return
+                    return TakSendOutcome.Dropped(
+                        "Cannot convert CoT type ${cotMessage.type} to TAKPacketV2",
+                        schemaLimited = false,
+                    )
                 }
                 try {
                     TakV2Compressor.compress(takPacketV2)
@@ -253,7 +293,7 @@ class TAKMeshIntegration(
                 }
             }
 
-        try {
+        return try {
             val dataPacket =
                 DataPacket(
                     to = NodeAddress.ID_BROADCAST,
@@ -262,6 +302,7 @@ class TAKMeshIntegration(
                 )
             commandSender.sendData(dataPacket)
             Logger.d { "Sent V2 to mesh: ${cotMessage.type} (${wirePayload.size} bytes)" }
+            TakSendOutcome.Sent(wirePayload.size)
         } catch (e: kotlin.coroutines.cancellation.CancellationException) {
             throw e
         } catch (e: Exception) {
@@ -269,6 +310,7 @@ class TAKMeshIntegration(
             Logger.e(e) {
                 "Failed to send TAKPacketV2 to mesh (${cotMessage.type}, ${wirePayload.size} bytes): ${e.message}"
             }
+            TakSendOutcome.Failed(e.message ?: "Send failed")
         }
     }
 
@@ -277,7 +319,7 @@ class TAKMeshIntegration(
      * compression. Only PLI and GeoChat payloads are supported by the v1 schema — shapes, markers, routes, casevac,
      * emergency, and task CoT events are dropped with a warning.
      */
-    private suspend fun sendCoTToMeshV1(cotMessage: CoTMessage) {
+    private suspend fun sendCoTToMeshV1(cotMessage: CoTMessage): TakSendOutcome {
         val takPacket =
             cotMessage.toTAKPacket()
                 ?: run {
@@ -286,7 +328,10 @@ class TAKMeshIntegration(
                             "in v1 TAKPacket schema (only PLI and GeoChat are supported). " +
                             "Upgrade radio firmware to >= 2.8.0 for full payload support."
                     }
-                    return
+                    return TakSendOutcome.Dropped(
+                        "Not representable in v1 TAKPacket schema (only PLI and GeoChat are supported)",
+                        schemaLimited = true,
+                    )
                 }
 
         val wirePayload = TAKPacket.ADAPTER.encode(takPacket)
@@ -295,10 +340,10 @@ class TAKMeshIntegration(
                 "Dropping oversized v1 TAK packet: type=${cotMessage.type} " +
                     "size=${wirePayload.size}B max=$MAX_TAK_WIRE_PAYLOAD_BYTES"
             }
-            return
+            return TakSendOutcome.Dropped("Oversized (>${MAX_TAK_WIRE_PAYLOAD_BYTES}B)", schemaLimited = false)
         }
 
-        try {
+        return try {
             val dataPacket =
                 DataPacket(
                     to = NodeAddress.ID_BROADCAST,
@@ -307,12 +352,14 @@ class TAKMeshIntegration(
                 )
             commandSender.sendData(dataPacket)
             Logger.d { "Sent V1 to mesh: ${cotMessage.type} (${wirePayload.size} bytes)" }
+            TakSendOutcome.Sent(wirePayload.size)
         } catch (e: kotlin.coroutines.cancellation.CancellationException) {
             throw e
         } catch (e: Exception) {
             Logger.e(e) {
                 "Failed to send v1 TAKPacket to mesh (${cotMessage.type}, ${wirePayload.size} bytes): ${e.message}"
             }
+            TakSendOutcome.Failed(e.message ?: "Send failed")
         }
     }
 

core/takserver/src/commonMain/kotlin/org/meshtastic/core/takserver/TakMeshTestRunner.kt

@@ -24,11 +24,15 @@ import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.asStateFlow
 import kotlinx.coroutines.sync.Mutex
-import okio.ByteString.Companion.toByteString
-import org.meshtastic.core.model.DataPacket
-import org.meshtastic.core.model.NodeAddress
-import org.meshtastic.core.repository.CommandSender
-import org.meshtastic.proto.PortNum
+
+/** Which outbound TAK protocol a [TakTestResult] was dispatched through. See [TAKMeshIntegration]. */
+enum class TakProtocol {
+    /** Firmware >= 2.8.0: zstd-compressed TAKPacketV2 on port 78, full typed CoT support. */
+    V2,
+
+    /** Firmware <= 2.7.x: bare-protobuf legacy TAKPacket on port 72, PLI + GeoChat only. */
+    V1,
+}
 
 /** Result of sending a single test fixture through the TAK mesh pipeline. */
 data class TakTestResult(
@@ -37,15 +41,26 @@ data class TakTestResult(
     val compressedBytes: Int,
     val passed: Boolean,
     val error: String? = null,
+    val protocol: TakProtocol = TakProtocol.V2,
+    // Mirrors TakSendOutcome.Dropped.schemaLimited: true only when this fixture was dropped because the active
+    // protocol's CoT type coverage doesn't include it (e.g. v1's TAKPacket only representing PLI/GeoChat) — a
+    // known, permanent limitation of that protocol, so this is expected/intentional behavior, not a self-test
+    // regression. False for an oversize drop (a real MTU problem, even if it happened on v1) or any failure —
+    // never set purely because the result's protocol is V1.
+    val expectedDrop: Boolean = false,
 )
 
 /**
- * Debug-only test runner that sends the SDK's CoT XML test fixtures through the real TAK mesh pipeline: strip → parse →
- * compress → send to mesh radio.
+ * Debug-only test runner that sends the SDK's CoT XML test fixtures through the REAL TAK mesh pipeline: strip → parse →
+ * dispatch, via [TAKMeshIntegration.sendCoTToMeshForTest]. Each fixture is run through BOTH the v2 (firmware >= 2.8.0)
+ * and v1 (legacy) dispatch paths so the self-test's pass/fail rate reflects what a real connected radio would actually
+ * do on either firmware generation, rather than always exercising the v2 pipeline regardless of the connected radio's
+ * firmware — see [TAKMeshIntegration.useTakV2].
  *
- * Paces sends by waiting [sendDelayMs] between each fixture to avoid flooding the radio's TX queue.
+ * Paces sends by waiting [sendDelayMs] between each successfully-sent fixture to avoid flooding the radio's TX queue.
+ * Fixtures that are dropped (not sent) incur no delay.
  */
-class TakMeshTestRunner(private val commandSender: CommandSender) {
+class TakMeshTestRunner(private val takMeshIntegration: TAKMeshIntegration) {
     private val _results = MutableStateFlow<List<TakTestResult>>(emptyList())
     val results: StateFlow<List<TakTestResult>> = _results.asStateFlow()
 
@@ -118,8 +133,8 @@ class TakMeshTestRunner(private val commandSender: CommandSender) {
     }
 
     /**
-     * Run all test fixtures sequentially, sending each through the mesh pipeline. Updates [results] and
-     * [currentFixture] as each fixture is processed.
+     * Run all test fixtures sequentially through both the v2 and v1 dispatch paths. Updates [results] and
+     * [currentFixture] as each fixture is processed. [results] accumulates v2 runs first, then v1 runs.
      */
     suspend fun runAll() {
         // Use tryLock to prevent concurrent test runs: if another coroutine is already
@@ -133,74 +148,95 @@ class TakMeshTestRunner(private val commandSender: CommandSender) {
 
             val allResults = mutableListOf<TakTestResult>()
 
-            for (name in FIXTURE_NAMES) {
-                _currentFixture.value = name
-                val result = runSingleFixture(name)
-                allResults.add(result)
-                _results.value = allResults.toList()
+            for (protocol in listOf(TakProtocol.V2, TakProtocol.V1)) {
+                for (name in FIXTURE_NAMES) {
+                    _currentFixture.value = "$name (${protocol.name.lowercase()})"
+                    val result = runSingleFixture(name, protocol)
+                    allResults.add(result)
+                    _results.value = allResults.toList()
 
-                if (result.passed) {
-                    // Wait for radio airtime + ACK before next send
-                    delay(SEND_DELAY_MS)
+                    if (result.passed) {
+                        // Wait for radio airtime + ACK before next send. Dropped fixtures never
+                        // reached the radio, so there's nothing to pace.
+                        delay(SEND_DELAY_MS)
+                    }
                 }
             }
 
             _currentFixture.value = null
 
-            val passed = allResults.count { it.passed }
-            val failed = allResults.size - passed
-            Logger.i { "TAK Mesh Test complete: $passed/${allResults.size} passed, $failed failed" }
+            val v2 = allResults.filter { it.protocol == TakProtocol.V2 }
+            val v1 = allResults.filter { it.protocol == TakProtocol.V1 }
+            val v2Passed = v2.count { it.passed }
+            val v1Passed = v1.count { it.passed }
+            val v1ExpectedDrops = v1.count { it.expectedDrop }
+            Logger.i {
+                "TAK Mesh Test complete: v2=$v2Passed/${v2.size} passed; " +
+                    "v1=$v1Passed/${v1.size} passed ($v1ExpectedDrops expected drops — " +
+                    "v1's legacy schema only supports PLI/GeoChat)"
+            }
         } finally {
             _isRunning.value = false
             runMutex.unlock()
         }
     }
 
-    private suspend fun runSingleFixture(name: String): TakTestResult {
+    private suspend fun runSingleFixture(name: String, protocol: TakProtocol): TakTestResult {
+        val forceV2 = protocol == TakProtocol.V2
+
         // Load fixture XML from bundled resources via platform-specific loader
         val xml =
             try {
                 loadTakFixtureXml(name)
             } catch (e: Exception) {
                 Logger.w(e) { "Failed to load fixture $name" }
-                return TakTestResult(name, 0, 0, false, "Load failed: ${e.message}")
+                return TakTestResult(name, 0, 0, false, "Load failed: ${e.message}", protocol)
             }
 
-        // Apply the same pipeline as TAKMeshIntegration.sendCoTToMesh()
-        val freshXml = TAKMeshIntegration.ensureMinimumStaleForMesh(xml)
-        val strippedXml = TAKMeshIntegration.stripNonEssentialElements(freshXml)
-
-        // Parse and compress via SDK
-        val wirePayload: ByteArray
-        try {
-            val result = TakSdkCompressor.compressCoT(strippedXml, MAX_TAK_WIRE_PAYLOAD_BYTES)
-            val compressed = result.wirePayload
-            if (compressed == null) {
-                Logger.w { "TAK Test: $name oversized even without remarks (xml=${xml.length}B)" }
-                return TakTestResult(name, xml.length, 0, false, "Oversized (>${MAX_TAK_WIRE_PAYLOAD_BYTES}B)")
+        // Parse into the same CoTMessage shape a real inbound TAK-client message would produce
+        // (sourceEventXml preserved) — see CoTXmlParser.buildCoTMessage.
+        val cotMessage =
+            CoTXmlParser(xml).parse().getOrElse { e ->
+                Logger.w(e) { "TAK Test: $name failed to parse as CoT XML: ${e.message}" }
+                return TakTestResult(name, xml.length, 0, false, "Parse failed: ${e.message}", protocol)
             }
-            wirePayload = compressed
-        } catch (e: Exception) {
-            Logger.w(e) { "TAK Test: $name compression failed: ${e.message}" }
-            return TakTestResult(name, xml.length, 0, false, "Compress failed: ${e.message}")
-        }
 
-        // Send to mesh
-        try {
-            val dataPacket =
-                DataPacket(
-                    to = NodeAddress.ID_BROADCAST,
-                    bytes = wirePayload.toByteString(),
-                    dataType = PortNum.ATAK_PLUGIN_V2.value,
-                )
-            commandSender.sendData(dataPacket)
-            Logger.i { "TAK Test: $name → ${wirePayload.size}B (xml=${xml.length}B)" }
-            return TakTestResult(name, xml.length, wirePayload.size, true)
+        // Dispatch through the REAL production pipeline (TAKMeshIntegration.sendCoTToMeshV1/V2),
+        // with the protocol forced rather than read from the connected radio's firmware.
+        return try {
+            when (val outcome = takMeshIntegration.sendCoTToMeshForTest(cotMessage, forceV2)) {
+                is TakSendOutcome.Sent -> {
+                    Logger.i { "TAK Test: $name → ${outcome.wireBytes}B (xml=${xml.length}B) via $protocol" }
+                    TakTestResult(name, xml.length, outcome.wireBytes, true, protocol = protocol)
+                }
+
+                is TakSendOutcome.Dropped -> {
+                    Logger.w { "TAK Test: $name dropped on $protocol: ${outcome.reason}" }
+                    TakTestResult(
+                        fixtureName = name,
+                        xmlBytes = xml.length,
+                        compressedBytes = 0,
+                        passed = false,
+                        error = outcome.reason,
+                        protocol = protocol,
+                        // Only a schema-coverage drop is an expected/intentional limitation (v1's legacy
+                        // schema legitimately supports only PLI/GeoChat). An oversize drop is a real MTU
+                        // problem regardless of which protocol hit it, and must never be mislabeled as
+                        // expected just because it happened on v1.
+                        expectedDrop = outcome.schemaLimited,
+                    )
+                }
+
+                is TakSendOutcome.Failed -> {
+                    Logger.w { "TAK Test: $name send failed on $protocol: ${outcome.reason}" }
+                    TakTestResult(name, xml.length, 0, false, outcome.reason, protocol)
+                }
+            }
         } catch (e: kotlin.coroutines.cancellation.CancellationException) {
             throw e
         } catch (e: Exception) {
             Logger.w(e) { "TAK Test: $name send failed: ${e.message}" }
-            return TakTestResult(name, xml.length, wirePayload.size, false, "Send failed: ${e.message}")
+            TakTestResult(name, xml.length, 0, false, "Send failed: ${e.message}", protocol)
         }
     }
 }

core/takserver/src/commonTest/kotlin/org/meshtastic/core/takserver/TAKMeshIntegrationTest.kt

@@ -48,6 +48,7 @@ import org.meshtastic.proto.TAKPacket
 import org.meshtastic.proto.User
 import kotlin.test.Test
 import kotlin.test.assertEquals
+import kotlin.test.assertNotNull
 import kotlin.test.assertTrue
 import kotlin.time.Clock
 import kotlin.time.Duration.Companion.minutes
@@ -331,6 +332,42 @@ class TAKMeshIntegrationTest {
         assertTrue(h.commandSender.sentPackets.isEmpty())
     }
 
+    // ── Dropped-outcome discrimination (regression: #6583 self-test blind spot) ────────
+
+    @Test
+    fun `v1 drop of an unsupported CoT type is schema-limited`() = runTest(UnconfinedTestDispatcher()) {
+        // a-h-G is a shape/marker type the legacy v1 TAKPacket schema has no field for at all —
+        // this is the "permanent limitation" case, not a size problem.
+        val h = TestHarness(nodeRepository = FakeNodeRepository(firmwareVersion = "2.7.0.0"))
+        val marker = CoTMessage(uid = "marker-1", type = "a-h-G", stale = Clock.System.now() + 5.minutes)
+
+        val outcome = h.integration.sendCoTToMeshForTest(marker, forceV2 = false)
+
+        val dropped = assertNotNull(outcome as? TakSendOutcome.Dropped, "expected a Dropped outcome, got $outcome")
+        assertTrue(dropped.schemaLimited, "an unsupported CoT type must be reported as schema-limited")
+    }
+
+    @Test
+    fun `v1 drop of an oversize but schema-representable PLI is NOT schema-limited`() =
+        runTest(UnconfinedTestDispatcher()) {
+            // a-f-G (PLI) IS representable in the v1 schema — this must be dropped for size, not
+            // mislabeled as an expected schema gap. This is the exact blind spot the self-test's
+            // expectedDrop flag has to avoid: an MTU problem hiding behind "expected" v1 behavior.
+            val h = TestHarness(nodeRepository = FakeNodeRepository(firmwareVersion = "2.7.0.0"))
+            val oversizePli =
+                CoTMessage(
+                    uid = "pli-1",
+                    type = "a-f-G-U-C",
+                    stale = Clock.System.now() + 5.minutes,
+                    contact = CoTContact(callsign = "X".repeat(500)),
+                )
+
+            val outcome = h.integration.sendCoTToMeshForTest(oversizePli, forceV2 = false)
+
+            val dropped = assertNotNull(outcome as? TakSendOutcome.Dropped, "expected a Dropped outcome, got $outcome")
+            assertTrue(!dropped.schemaLimited, "an oversize drop of a representable type must not be schema-limited")
+        }
+
     // ── GeoChat callsign enrichment ──────────────────────────────────────────
 
     @Test

core/takserver/src/commonTest/kotlin/org/meshtastic/core/takserver/TakMeshTestRunnerTest.kt

@@ -0,0 +1,180 @@
+/*
+ * Copyright (c) 2026 Meshtastic LLC
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+package org.meshtastic.core.takserver
+
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.test.UnconfinedTestDispatcher
+import kotlinx.coroutines.test.runTest
+import org.meshtastic.core.di.CoroutineDispatchers
+import org.meshtastic.core.repository.MeshConfigHandler
+import org.meshtastic.core.repository.RadioSessionContext
+import org.meshtastic.core.testing.FakeCommandSender
+import org.meshtastic.core.testing.FakeNodeRepository
+import org.meshtastic.core.testing.FakeServiceRepository
+import org.meshtastic.core.testing.FakeTakPrefs
+import org.meshtastic.core.testing.TestDataFactory
+import org.meshtastic.proto.Channel
+import org.meshtastic.proto.Config
+import org.meshtastic.proto.DeviceUIConfig
+import org.meshtastic.proto.LoRaRegionPresetMap
+import org.meshtastic.proto.LocalConfig
+import org.meshtastic.proto.LocalModuleConfig
+import org.meshtastic.proto.ModuleConfig
+import org.meshtastic.proto.PortNum
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertTrue
+
+/**
+ * Regression coverage for the self-test blind spot found while investigating #6583: [TakMeshTestRunner] must exercise
+ * the SAME real dispatch pipeline ([TAKMeshIntegration.sendCoTToMeshV1] / `sendCoTToMeshV2`) that live TAK traffic
+ * uses, not a hardcoded v2-only shortcut — and it must do so for BOTH protocol versions regardless of whichever
+ * firmware a test radio happens to report, since [TAKMeshIntegration.useTakV2] gates real traffic on the connected
+ * radio's firmware and older firmware silently falls back to the much more limited v1 schema.
+ */
+class TakMeshTestRunnerTest {
+
+    private class FakeMeshConfigHandler : MeshConfigHandler {
+        override val localConfig = MutableStateFlow(LocalConfig())
+        override val moduleConfig = MutableStateFlow(LocalModuleConfig())
+
+        override fun handleDeviceConfig(config: Config, session: RadioSessionContext) = true
+
+        override fun handleModuleConfig(config: ModuleConfig, session: RadioSessionContext) = true
+
+        override fun handleChannel(channel: Channel, session: RadioSessionContext) = true
+
+        override fun handleDeviceUIConfig(config: DeviceUIConfig, session: RadioSessionContext) = true
+
+        override fun handleRegionPresets(map: LoRaRegionPresetMap, session: RadioSessionContext) = true
+    }
+
+    private class Harness(firmwareVersion: String?) {
+        val serverManager = FakeTAKServerManager()
+        val commandSender = FakeCommandSender()
+        val serviceRepository = FakeServiceRepository()
+        val meshConfigHandler = FakeMeshConfigHandler()
+        val nodeRepository =
+            FakeNodeRepository().apply {
+                setMyNodeInfo(TestDataFactory.createMyNodeInfo(firmwareVersion = firmwareVersion))
+            }
+        val takPrefs = FakeTakPrefs()
+        private val dispatcher = UnconfinedTestDispatcher()
+        val broadcaster =
+            MeshToCotBroadcaster(
+                serverManager,
+                nodeRepository,
+                takPrefs,
+                CoroutineDispatchers(io = dispatcher, main = dispatcher, default = dispatcher),
+            )
+
+        val integration =
+            TAKMeshIntegration(
+                takServerManager = serverManager,
+                commandSender = commandSender,
+                serviceRepository = serviceRepository,
+                meshConfigHandler = meshConfigHandler,
+                nodeRepository = nodeRepository,
+                meshToCotBroadcaster = broadcaster,
+            )
+
+        val runner = TakMeshTestRunner(integration)
+    }
+
+    // ── runAll() covers both protocols ──────────────────────────────────────
+
+    @Test
+    fun `runAll exercises every fixture on both v1 and v2`() = runTest(UnconfinedTestDispatcher()) {
+        // Firmware is irrelevant here — the runner forces the protocol per pass rather than
+        // reading Capabilities.supportsTakV2, which is exactly the bug this test guards against.
+        val h = Harness(firmwareVersion = null)
+
+        h.runner.runAll()
+
+        val results = h.runner.results.value
+        assertEquals(TakMeshTestRunner.FIXTURE_NAMES.size * 2, results.size)
+        assertEquals(TakMeshTestRunner.FIXTURE_NAMES.size, results.count { it.protocol == TakProtocol.V2 })
+        assertEquals(TakMeshTestRunner.FIXTURE_NAMES.size, results.count { it.protocol == TakProtocol.V1 })
+    }
+
+    @Test
+    fun `v2 pass sends on ATAK_PLUGIN_V2 regardless of connected firmware`() = runTest(UnconfinedTestDispatcher()) {
+        // Connected "radio" reports legacy firmware — if the runner still read useTakV2() instead of
+        // forcing the protocol, this would incorrectly downgrade the v2 pass to v1.
+        val h = Harness(firmwareVersion = "2.7.0.0")
+
+        h.runner.runAll()
+
+        val v2Results = h.runner.results.value.filter { it.protocol == TakProtocol.V2 }
+        val v2Sent = h.commandSender.sentPackets.count { it.dataType == PortNum.ATAK_PLUGIN_V2.value }
+        assertEquals(v2Results.count { it.passed }, v2Sent, "every passing v2 result must correspond to a v2 send")
+        assertTrue(v2Sent > 0, "at least one fixture must round-trip through the real v2 pipeline")
+    }
+
+    @Test
+    fun `v1 pass sends on legacy ATAK_PLUGIN regardless of connected firmware`() = runTest(UnconfinedTestDispatcher()) {
+        // Connected "radio" reports v2-capable firmware — if the runner still read useTakV2() instead
+        // of forcing the protocol, this would incorrectly upgrade the v1 pass to v2, hiding the exact
+        // blind spot reported in #6583.
+        val h = Harness(firmwareVersion = "2.8.0.0")
+
+        h.runner.runAll()
+
+        val v1Results = h.runner.results.value.filter { it.protocol == TakProtocol.V1 }
+        val v1Sent = h.commandSender.sentPackets.count { it.dataType == PortNum.ATAK_PLUGIN.value }
+        assertEquals(v1Results.count { it.passed }, v1Sent, "every passing v1 result must correspond to a v1 send")
+        assertTrue(v1Sent > 0, "PLI/GeoChat fixtures must still round-trip through the real v1 pipeline")
+    }
+
+    @Test
+    fun `v1 pass reports non-PLI non-chat drops as expected not as failures`() = runTest(UnconfinedTestDispatcher()) {
+        val h = Harness(firmwareVersion = null)
+
+        h.runner.runAll()
+
+        val v1Results = h.runner.results.value.filter { it.protocol == TakProtocol.V1 }
+        // marker_spot.xml is type "b-m-p-s-m" — not representable in the legacy v1 TAKPacket
+        // schema (only PLI/"a-f-*" and chat/"b-t-f" are). This must show up as an *expected*
+        // drop, not an unlabeled self-test failure, per the v1 schema's documented limitations.
+        val markerResult = v1Results.single { it.fixtureName == "marker_spot.xml" }
+        assertTrue(markerResult.expectedDrop, "v1 dropping an unsupported CoT type is expected, not a bug")
+        assertTrue(!markerResult.passed)
+
+        // Every v1 drop must be accounted for: either the known schema-coverage limitation
+        // (expectedDrop) or an oversize drop, which is a real MTU problem and correctly NOT
+        // labeled expected even though it happened on v1 — see TakSendOutcome.Dropped.schemaLimited.
+        // A drop that is neither would be a genuine, unaccounted-for failure.
+        assertTrue(
+            v1Results.filter { !it.passed }.all { it.expectedDrop || it.error?.startsWith("Oversized") == true },
+            "every v1 drop in a clean run should be the known schema limitation or an oversize drop, " +
+                "not an unaccounted-for failure",
+        )
+    }
+
+    @Test
+    fun `v2 pass has no expected drops - v2 must support every fixture type`() = runTest(UnconfinedTestDispatcher()) {
+        val h = Harness(firmwareVersion = null)
+
+        h.runner.runAll()
+
+        val v2Results = h.runner.results.value.filter { it.protocol == TakProtocol.V2 }
+        assertTrue(
+            v2Results.none { it.expectedDrop },
+            "v2 has no known schema gaps, so nothing should be labeled expected",
+        )
+    }
+}

feature/settings/src/commonMain/kotlin/org/meshtastic/feature/settings/radio/component/TAKConfigItemList.kt

@@ -47,13 +47,13 @@ import androidx.compose.ui.graphics.Color
 import androidx.compose.ui.unit.dp
 import androidx.lifecycle.compose.collectAsStateWithLifecycle
 import kotlinx.coroutines.launch
+import org.jetbrains.compose.resources.StringResource
 import org.jetbrains.compose.resources.pluralStringResource
 import org.jetbrains.compose.resources.stringResource
 import org.koin.compose.koinInject
 import org.meshtastic.core.common.BuildConfigProvider
 import org.meshtastic.core.model.getColorFrom
 import org.meshtastic.core.model.getStringResFrom
-import org.meshtastic.core.repository.CommandSender
 import org.meshtastic.core.repository.TakPrefs
 import org.meshtastic.core.resources.Res
 import org.meshtastic.core.resources.back
@@ -79,15 +79,20 @@ import org.meshtastic.core.resources.tak_server_status_unavailable
 import org.meshtastic.core.resources.tak_server_status_waiting
 import org.meshtastic.core.resources.tak_server_test_card_title
 import org.meshtastic.core.resources.tak_server_test_idle
+import org.meshtastic.core.resources.tak_server_test_protocol_v1_label
+import org.meshtastic.core.resources.tak_server_test_protocol_v2_label
 import org.meshtastic.core.resources.tak_server_test_result_bytes
+import org.meshtastic.core.resources.tak_server_test_result_expected_drop
 import org.meshtastic.core.resources.tak_server_test_result_unknown_error
-import org.meshtastic.core.resources.tak_server_test_results
+import org.meshtastic.core.resources.tak_server_test_results_v2
 import org.meshtastic.core.resources.tak_server_test_run
 import org.meshtastic.core.resources.tak_server_test_running
 import org.meshtastic.core.resources.tak_team
 import org.meshtastic.core.takserver.TAKDataPackageGenerator
+import org.meshtastic.core.takserver.TAKMeshIntegration
 import org.meshtastic.core.takserver.TAKServerManager
 import org.meshtastic.core.takserver.TakMeshTestRunner
+import org.meshtastic.core.takserver.TakProtocol
 import org.meshtastic.core.takserver.TakTestResult
 import org.meshtastic.core.ui.component.DropDownPreference
 import org.meshtastic.core.ui.component.SwitchPreference
@@ -392,8 +397,8 @@ private fun TakMeshTestCard() {
     val buildConfig: BuildConfigProvider = koinInject()
     if (!buildConfig.isDebug) return
 
-    val commandSender: CommandSender = koinInject()
-    val testRunner = remember { TakMeshTestRunner(commandSender) }
+    val takMeshIntegration: TAKMeshIntegration = koinInject()
+    val testRunner = remember { TakMeshTestRunner(takMeshIntegration) }
     val results by testRunner.results.collectAsStateWithLifecycle()
     val isRunning by testRunner.isRunning.collectAsStateWithLifecycle()
     val currentFixture by testRunner.currentFixture.collectAsStateWithLifecycle()
@@ -418,8 +423,8 @@ internal fun TakMeshTestCardContent(
     onRunTests: () -> Unit,
 ) {
     val loadingLabel = stringResource(Res.string.tak_server_loading)
-    val passed = results.count { it.passed }
-    val failed = results.count { !it.passed }
+    val v2Results = results.filter { it.protocol == TakProtocol.V2 }
+    val v1Results = results.filter { it.protocol == TakProtocol.V1 }
 
     TitledCard(title = stringResource(Res.string.tak_server_test_card_title)) {
         Row(
@@ -437,25 +442,6 @@ internal fun TakMeshTestCardContent(
                     },
                     style = MaterialTheme.typography.bodyLarge,
                 )
-                if (results.isNotEmpty()) {
-                    Text(
-                        text =
-                        stringResource(
-                            Res.string.tak_server_test_results,
-                            passed,
-                            failed,
-                            results.size,
-                            fixtureCount,
-                        ),
-                        style = MaterialTheme.typography.bodySmall,
-                        color =
-                        if (failed > 0) {
-                            MaterialTheme.colorScheme.error
-                        } else {
-                            MaterialTheme.colorScheme.onSurfaceVariant
-                        },
-                    )
-                }
             }
             if (isRunning) {
                 CircularProgressIndicator()
@@ -467,7 +453,58 @@ internal fun TakMeshTestCardContent(
             }
         }
 
-        // Results list
+        if (v2Results.isNotEmpty()) {
+            TakProtocolResultsSection(
+                titleRes = Res.string.tak_server_test_protocol_v2_label,
+                results = v2Results,
+                fixtureCount = fixtureCount,
+            )
+        }
+        if (v1Results.isNotEmpty()) {
+            TakProtocolResultsSection(
+                titleRes = Res.string.tak_server_test_protocol_v1_label,
+                results = v1Results,
+                fixtureCount = fixtureCount,
+            )
+        }
+    }
+}
+
+/**
+ * One protocol's fixture run within the TAK self-test card: a summary line (passed / expected-limitation drops /
+ * unexpected failures) followed by the per-fixture rows. Expected v1 schema drops are surfaced distinctly from real
+ * failures so a low v1 pass rate reads as intentional legacy-schema behavior, not a self-test regression.
+ */
+@Composable
+private fun TakProtocolResultsSection(titleRes: StringResource, results: List<TakTestResult>, fixtureCount: Int) {
+    val passed = results.count { it.passed }
+    val expectedDrops = results.count { !it.passed && it.expectedDrop }
+    val unexpectedFailed = results.count { !it.passed && !it.expectedDrop }
+
+    Column {
+        HorizontalDivider()
+        Column(modifier = Modifier.fillMaxWidth().padding(horizontal = 16.dp, vertical = 8.dp)) {
+            Text(text = stringResource(titleRes), style = MaterialTheme.typography.titleSmall)
+            Text(
+                text =
+                stringResource(
+                    Res.string.tak_server_test_results_v2,
+                    passed,
+                    expectedDrops,
+                    unexpectedFailed,
+                    results.size,
+                    fixtureCount,
+                ),
+                style = MaterialTheme.typography.bodySmall,
+                color =
+                if (unexpectedFailed > 0) {
+                    MaterialTheme.colorScheme.error
+                } else {
+                    MaterialTheme.colorScheme.onSurfaceVariant
+                },
+            )
+        }
+
         for (result in results) {
             HorizontalDivider()
             Row(
@@ -482,13 +519,21 @@ internal fun TakMeshTestCardContent(
                 )
                 Text(
                     text =
-                    if (result.passed) {
-                        stringResource(Res.string.tak_server_test_result_bytes, result.compressedBytes)
-                    } else {
-                        result.error ?: stringResource(Res.string.tak_server_test_result_unknown_error)
+                    when {
+                        result.passed ->
+                            stringResource(Res.string.tak_server_test_result_bytes, result.compressedBytes)
+
+                        result.expectedDrop -> stringResource(Res.string.tak_server_test_result_expected_drop)
+
+                        else -> result.error ?: stringResource(Res.string.tak_server_test_result_unknown_error)
                     },
                     style = MaterialTheme.typography.bodySmall,
-                    color = if (result.passed) MaterialTheme.colorScheme.primary else MaterialTheme.colorScheme.error,
+                    color =
+                    when {
+                        result.passed -> MaterialTheme.colorScheme.primary
+                        result.expectedDrop -> MaterialTheme.colorScheme.onSurfaceVariant
+                        else -> MaterialTheme.colorScheme.error
+                    },
                 )
             }
         }