Merge pull request #2466 from vitorpamplona/claude/fix-marmot-add-member-j5dx2

Fix MLS commit framing for wire protocol compliance

commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/marmot/MarmotManager.kt

@@ -179,7 +179,10 @@ class MarmotManager(
         }
 
         val commitResult = groupManager.addMember(nostrGroupId, keyPackageBytes)
-        val commitEvent = outboundProcessor.buildCommitEvent(nostrGroupId, commitResult.commitBytes)
+        val commitEvent = outboundProcessor.buildCommitEvent(nostrGroupId, commitResult.framedCommitBytes)
+        // We've already applied this commit locally; prevent the relay-echoed
+        // copy from being re-applied by the inbound pipeline.
+        inboundProcessor.markEventProcessed(commitEvent.signedEvent.id)
 
         val welcomeDelivery =
             welcomeSender.wrapWelcome(
@@ -266,7 +269,9 @@ class MarmotManager(
         targetLeafIndex: Int,
     ): OutboundGroupEvent {
         val commitResult = groupManager.removeMember(nostrGroupId, targetLeafIndex)
-        return outboundProcessor.buildCommitEvent(nostrGroupId, commitResult.commitBytes)
+        val commitEvent = outboundProcessor.buildCommitEvent(nostrGroupId, commitResult.framedCommitBytes)
+        inboundProcessor.markEventProcessed(commitEvent.signedEvent.id)
+        return commitEvent
     }
 
     /**
@@ -278,7 +283,9 @@ class MarmotManager(
         metadata: MarmotGroupData,
     ): OutboundGroupEvent {
         val commitResult = groupManager.updateGroupExtensions(nostrGroupId, listOf(metadata.toExtension()))
-        return outboundProcessor.buildCommitEvent(nostrGroupId, commitResult.commitBytes)
+        val commitEvent = outboundProcessor.buildCommitEvent(nostrGroupId, commitResult.framedCommitBytes)
+        inboundProcessor.markEventProcessed(commitEvent.signedEvent.id)
+        return commitEvent
     }
 
     // --- KeyPackage Management ---

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/marmot/MarmotInboundProcessor.kt

@@ -287,6 +287,31 @@ class MarmotInboundProcessor(
             WelcomeResult.Error("Failed to process Welcome: ${e.message}", e)
         }
 
+    /**
+     * Mark a kind:445 event id as already processed so that a later relay
+     * echo of the same event is treated as a [GroupEventResult.Duplicate]
+     * instead of being re-applied.
+     *
+     * Callers should invoke this right after publishing a commit (e.g. from
+     * [com.vitorpamplona.amethyst.commons.marmot.MarmotManager.addMember])
+     * because `group.addMember` / `group.commit` have already advanced the
+     * local epoch. Reprocessing the same commit bytes would otherwise fail
+     * with a confirmation-tag / transcript mismatch.
+     */
+    suspend fun markEventProcessed(eventId: HexKey) {
+        processedIdsMutex.withLock {
+            processedEventIds.add(eventId)
+            if (processedEventIds.size > MAX_PROCESSED_IDS) {
+                val iterator = processedEventIds.iterator()
+                val toRemove = processedEventIds.size - MAX_PROCESSED_IDS
+                repeat(toRemove) {
+                    iterator.next()
+                    iterator.remove()
+                }
+            }
+        }
+    }
+
     /**
      * Resolve any pending commit conflicts for a given epoch.
      *

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/marmot/mls/group/MlsGroup.kt

@@ -31,6 +31,9 @@ import com.vitorpamplona.quartz.marmot.mls.crypto.X25519
 import com.vitorpamplona.quartz.marmot.mls.framing.ContentType
 import com.vitorpamplona.quartz.marmot.mls.framing.MlsMessage
 import com.vitorpamplona.quartz.marmot.mls.framing.PrivateMessage
+import com.vitorpamplona.quartz.marmot.mls.framing.PublicMessage
+import com.vitorpamplona.quartz.marmot.mls.framing.Sender
+import com.vitorpamplona.quartz.marmot.mls.framing.SenderType
 import com.vitorpamplona.quartz.marmot.mls.framing.WireFormat
 import com.vitorpamplona.quartz.marmot.mls.messages.Commit
 import com.vitorpamplona.quartz.marmot.mls.messages.CommitResult
@@ -502,7 +505,10 @@ class MlsGroup private constructor(
         val newConfirmedTranscriptHash = MlsCryptoProvider.hash(confirmedInput.toByteArray())
 
         val newTreeHash = tree.treeHash()
-        val newEpoch = groupContext.epoch + 1
+        val oldEpoch = groupContext.epoch
+        val preCommitGroupId = groupContext.groupId
+        val committerLeafIndex = myLeafIndex
+        val newEpoch = oldEpoch + 1
 
         groupContext =
             groupContext.copy(
@@ -544,7 +550,20 @@ class MlsGroup private constructor(
         sentKeys.clear()
 
         val commitBytes = commit.toTlsBytes()
-        return CommitResult(commitBytes, welcomeBytes, null)
+        val framedCommitBytes =
+            framePublicMessageCommit(
+                groupId = preCommitGroupId,
+                epoch = oldEpoch,
+                senderLeafIndex = committerLeafIndex,
+                commitBytes = commitBytes,
+                confirmationTag = confirmationTag,
+            )
+        return CommitResult(
+            commitBytes = commitBytes,
+            welcomeBytes = welcomeBytes,
+            groupInfoBytes = null,
+            framedCommitBytes = framedCommitBytes,
+        )
     }
 
     // --- Message Encryption ---
@@ -1475,6 +1494,38 @@ class MlsGroup private constructor(
         private const val REUSE_GUARD_LENGTH = 4
         private const val RATCHET_TREE_EXTENSION_TYPE = 0x0001
 
+        /**
+         * Wrap a raw [Commit] (as [commitBytes]) in an MlsMessage(PublicMessage(...))
+         * envelope so it can be published on the wire (RFC 9420 §6 / §6.2).
+         *
+         * The receiver uses the sender's leaf index and the confirmation_tag from
+         * the [PublicMessage] header to drive [MlsGroup.processCommit]. The
+         * `signature` and `membership_tag` opaque fields are intentionally empty —
+         * the current implementation does not verify them on inbound commits,
+         * but the TLS structure must still be present so decoding succeeds.
+         */
+        internal fun framePublicMessageCommit(
+            groupId: ByteArray,
+            epoch: Long,
+            senderLeafIndex: Int,
+            commitBytes: ByteArray,
+            confirmationTag: ByteArray,
+        ): ByteArray {
+            val publicMessage =
+                PublicMessage(
+                    groupId = groupId,
+                    epoch = epoch,
+                    sender = Sender(SenderType.MEMBER, senderLeafIndex),
+                    authenticatedData = ByteArray(0),
+                    contentType = ContentType.COMMIT,
+                    content = commitBytes,
+                    signature = ByteArray(0),
+                    confirmationTag = confirmationTag,
+                    membershipTag = ByteArray(0),
+                )
+            return MlsMessage.fromPublicMessage(publicMessage).toTlsBytes()
+        }
+
         /**
          * Build ConfirmedTranscriptHashInput (RFC 9420 Section 8.2) — static version
          * usable from both instance methods and companion object factory methods.

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/marmot/mls/messages/Commit.kt

@@ -91,11 +91,22 @@ data class UpdatePath(
 
 /**
  * Result of creating a Commit: the MLS messages to distribute.
+ *
+ * [commitBytes] is the raw TLS-encoded [Commit] struct (RFC 9420 §12.4), useful
+ * for unit tests and the [com.vitorpamplona.quartz.marmot.mls.group.MlsGroup.processCommit]
+ * entry point. For on-the-wire distribution, callers MUST publish
+ * [framedCommitBytes] (the MlsMessage(PublicMessage(FramedContent(commit))) envelope)
+ * so that receivers can parse the sender's leaf index and confirmation tag.
  */
 data class CommitResult(
     val commitBytes: ByteArray,
     val welcomeBytes: ByteArray?,
     val groupInfoBytes: ByteArray?,
+    /**
+     * Fully-framed commit ready for the MIP-03 outer ChaCha20 encryption.
+     * Wire format: MlsMessage(version=mls10, wireFormat=mls_public_message, payload=PublicMessage(...)).
+     */
+    val framedCommitBytes: ByteArray = commitBytes,
 ) {
     override fun equals(other: Any?): Boolean {
         if (this === other) return true

quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/marmot/MarmotPipelineTest.kt

@@ -225,6 +225,87 @@ class MarmotPipelineTest {
         }
     }
 
+    @Test
+    fun testAddMemberCommitIsFramedAsPublicMessage() {
+        // Regression: addMember's outbound commit used to carry the raw Commit TLS
+        // bytes instead of an MlsMessage(PublicMessage(commit)) envelope, which
+        // caused receivers to fail parsing with "Unsupported MLS version: …"
+        // when the first two bytes of the Commit struct were read as the
+        // MlsMessage version field.
+        runBlocking {
+            val manager = createGroupManager()
+            manager.createGroup(groupId, "alice".encodeToByteArray())
+            val group = manager.getGroup(groupId)!!
+            val bobBundle = group.createKeyPackage("bob".encodeToByteArray(), ByteArray(0))
+
+            val commitResult = manager.addMember(groupId, bobBundle.keyPackage.toTlsBytes())
+
+            // The framedCommitBytes must decode as an MlsMessage(PublicMessage(commit))
+            val framed = commitResult.framedCommitBytes
+            val mlsMessage =
+                com.vitorpamplona.quartz.marmot.mls.framing.MlsMessage
+                    .decodeTls(
+                        com.vitorpamplona.quartz.marmot.mls.codec
+                            .TlsReader(framed),
+                    )
+            assertEquals(
+                com.vitorpamplona.quartz.marmot.mls.framing.WireFormat.PUBLIC_MESSAGE,
+                mlsMessage.wireFormat,
+            )
+
+            val publicMessage =
+                com.vitorpamplona.quartz.marmot.mls.framing.PublicMessage
+                    .decodeTls(
+                        com.vitorpamplona.quartz.marmot.mls.codec
+                            .TlsReader(mlsMessage.payload),
+                    )
+            assertEquals(
+                com.vitorpamplona.quartz.marmot.mls.framing.ContentType.COMMIT,
+                publicMessage.contentType,
+            )
+            assertEquals(
+                com.vitorpamplona.quartz.marmot.mls.framing.SenderType.MEMBER,
+                publicMessage.sender.senderType,
+            )
+            assertNotNull(publicMessage.confirmationTag, "confirmation_tag must be present on a commit")
+            // The PublicMessage.content carries the raw Commit struct.
+            kotlin.test.assertContentEquals(commitResult.commitBytes, publicMessage.content)
+        }
+    }
+
+    @Test
+    fun testAddMemberCommitEventDecryptsToFramedMlsMessage() {
+        // End-to-end variant: the kind:445 content returned by buildCommitEvent,
+        // when ChaCha20-Poly1305 decrypted with the current exporter key, must
+        // yield an MlsMessage whose wire format is PUBLIC_MESSAGE (not a raw
+        // Commit struct).
+        runBlocking {
+            val manager = createGroupManager()
+            manager.createGroup(groupId, "alice".encodeToByteArray())
+            val group = manager.getGroup(groupId)!!
+            val bobBundle = group.createKeyPackage("bob".encodeToByteArray(), ByteArray(0))
+
+            val commitResult = manager.addMember(groupId, bobBundle.keyPackage.toTlsBytes())
+
+            val outbound = MarmotOutboundProcessor(manager)
+            val outboundResult = outbound.buildCommitEvent(groupId, commitResult.framedCommitBytes)
+            val event = outboundResult.signedEvent
+
+            val exporterKey = manager.exporterSecret(groupId)
+            val mlsBytes = GroupEventEncryption.decrypt(event.content, exporterKey)
+            val mlsMessage =
+                com.vitorpamplona.quartz.marmot.mls.framing.MlsMessage
+                    .decodeTls(
+                        com.vitorpamplona.quartz.marmot.mls.codec
+                            .TlsReader(mlsBytes),
+                    )
+            assertEquals(
+                com.vitorpamplona.quartz.marmot.mls.framing.WireFormat.PUBLIC_MESSAGE,
+                mlsMessage.wireFormat,
+            )
+        }
+    }
+
     @Test
     fun testSubscriptionManagerSyncWithGroupManager() {
         runBlocking {