fix(marmot): frame outbound commits as MlsMessage(PublicMessage(...))

addMember/removeMember/updateGroupMetadata were publishing the raw TLS
Commit struct inside the kind:445 outer ChaCha20 layer, so receivers
that started with MlsMessage.decodeTls read the first two bytes of the
Commit as the MLS protocol version and aborted with "Unsupported MLS
version: <garbage>" (e.g. 16704 = 0x4140).

Wrap commits produced by MlsGroup.commit in a PublicMessage carrying
the sender leaf index and confirmation_tag, then in an MlsMessage
envelope, and expose those bytes via CommitResult.framedCommitBytes so
MarmotManager uses them instead of the raw commit bytes. The internal
CommitResult.commitBytes field stays raw so MlsGroup.processCommit and
existing tests that exercise it directly keep working.

Also mark the committer's own published kind:445 as already processed
in MarmotInboundProcessor so that the relay echo of our own commit is
treated as a duplicate rather than re-applied on top of the already-
advanced local epoch.

https://claude.ai/code/session_01NR6JgYRimVb142T2nkChuh

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 {