mirror of
https://github.com/vitorpamplona/amethyst.git
synced 2026-09-14 00:55:08 +00:00
feat(marmot): advertise every agent-stream role and render previews on Android
Our leaf advertised `receive` only, which was honest while nothing could originate a stream and is not any more. It now advertises `receive`, `send` and `fanout` — the same set MDK puts on every KeyPackage it publishes — so a group requiring any of them admits us. A capability is a claim about what the client supports, not a duty to stream: a member that never originates one is a quiet member, not a broken one. The role-gate test asserted the old behaviour, so it was testing our own capability set rather than the gate. It now builds a deliberately reduced leaf and checks that THAT is refused, which keeps working whatever we go on to advertise; a second case pins the new fact that we fill every role the profile defines. `MarmotAgentStreamWatcher` in commons follows the newest kind:1200 in a group, folds the QUIC records behind it under the receive discipline, and settles the result against the durable kind:9 — confirmed when the transcript agrees, dropped when it does not, because a disagreement means we rendered something the publisher never sent. Resolving the final message lives here rather than in the UI so a front end only has to say "the feed moved", and so the whole decision is testable without a UI. Android shows it as an italic, labelled row between the transcript and the composer. Provisional content has to look provisional: preview text is not durable history until the final message vouches for it, and the row disappears the moment it is confirmed or contradicted. Progress and status records render as separate chrome, never as answer text, which is what the spec requires of them. Every failure path ends as "no preview" rather than as a broken group: no stream, no broker candidate, an unreachable broker, an unimplemented stream type, or a platform with no QUIC at all. `receive` explicitly does not require the QUIC data plane. The desktop app has no Marmot chat screen to render into — its chat UI is NIP-17 only — so there is nothing to wire there yet. The watcher is in commons and speaks only quartz's transport port, so desktop inherits it the day that screen exists. Interop unchanged at 19 of 19 with all three roles advertised. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016kCuA6tc4JQzHPCDd39GHq
This commit is contained in:
Claude
@@ -402,6 +402,10 @@ dependencies {
|
||||
implementation(project(":quartz"))
|
||||
implementation(project(":commons"))
|
||||
implementation(project(":nestsClient"))
|
||||
// Agent text stream previews: the raw-QUIC binding plus the QUIC
|
||||
// stack under it (for the certificate validator it requires).
|
||||
implementation(project(":marmotQuic"))
|
||||
implementation(project(":quic"))
|
||||
implementation(project(":nappletHost"))
|
||||
// Compose Multiplatform resources runtime, so app-side screens that share a
|
||||
// string with a commons renderer can read commons' generated `Res` directly
|
||||
|
||||
@@ -175,6 +175,7 @@ import com.vitorpamplona.amethyst.ui.actions.NewMessageTagger
|
||||
import com.vitorpamplona.amethyst.ui.navigation.bottombars.BottomBarEntry
|
||||
import com.vitorpamplona.amethyst.ui.navigation.bottombars.NavBarItem
|
||||
import com.vitorpamplona.amethyst.ui.screen.loggedIn.EventProcessor
|
||||
import com.vitorpamplona.marmotquic.QuicAgentTextStreamTransport
|
||||
import com.vitorpamplona.quartz.buzz.threading.buzzThread
|
||||
import com.vitorpamplona.quartz.buzz.threading.buzzThreadReply
|
||||
import com.vitorpamplona.quartz.buzz.threading.buzzThreadRoot
|
||||
@@ -204,6 +205,7 @@ import com.vitorpamplona.quartz.experimental.profileGallery.fromEvent
|
||||
import com.vitorpamplona.quartz.experimental.profileGallery.hash
|
||||
import com.vitorpamplona.quartz.experimental.profileGallery.image
|
||||
import com.vitorpamplona.quartz.experimental.profileGallery.mimeType
|
||||
import com.vitorpamplona.quartz.marmot.appComponents.agentTextStream.transport.MarmotQuicTransport
|
||||
import com.vitorpamplona.quartz.marmot.mip00KeyPackages.KeyPackageEvent
|
||||
import com.vitorpamplona.quartz.marmot.mls.group.MlsGroupStateStore
|
||||
import com.vitorpamplona.quartz.nip01Core.core.Address
|
||||
@@ -340,6 +342,7 @@ import com.vitorpamplona.quartz.utils.RandomInstance
|
||||
import com.vitorpamplona.quartz.utils.TimeUtils
|
||||
import com.vitorpamplona.quartz.utils.ciphers.AESGCM
|
||||
import com.vitorpamplona.quartz.utils.containsAny
|
||||
import com.vitorpamplona.quic.tls.JdkCertificateValidator
|
||||
import kotlinx.coroutines.CoroutineScope
|
||||
import kotlinx.coroutines.DelicateCoroutinesApi
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
@@ -953,6 +956,24 @@ class Account(
|
||||
)
|
||||
}
|
||||
|
||||
/**
|
||||
* Raw QUIC for agent text stream previews (`transports/quic.md`).
|
||||
*
|
||||
* Only the live preview needs it. A device that cannot open a QUIC
|
||||
* connection still participates fully — it reads every stream's
|
||||
* authoritative kind:9 like ordinary chat — which is why this is a
|
||||
* separate optional piece rather than part of [marmotManager].
|
||||
*/
|
||||
val marmotStreamTransport: MarmotQuicTransport by lazy {
|
||||
QuicAgentTextStreamTransport(
|
||||
parentScope = scope,
|
||||
// Preview brokers are commonly self-signed and the binding expects
|
||||
// that; the platform trust store is still the default answer, and
|
||||
// a deployment that pins does it here.
|
||||
certificateValidator = JdkCertificateValidator(),
|
||||
)
|
||||
}
|
||||
|
||||
val paymentTargetsState = NipA3PaymentTargetsState(signer, cache, scope, settings)
|
||||
|
||||
val bolt12OfferList = Bolt12OfferListState(signer, cache, scope, settings)
|
||||
|
||||
+112
@@ -0,0 +1,112 @@
|
||||
/*
|
||||
* Copyright (c) 2025 Vitor Pamplona
|
||||
*
|
||||
* Permission is hereby granted, free of charge, to any person obtaining a copy of
|
||||
* this software and associated documentation files (the "Software"), to deal in
|
||||
* the Software without restriction, including without limitation the rights to use,
|
||||
* copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
|
||||
* Software, and to permit persons to whom the Software is furnished to do so,
|
||||
* subject to the following conditions:
|
||||
*
|
||||
* The above copyright notice and this permission notice shall be included in all
|
||||
* copies or substantial portions of the Software.
|
||||
*
|
||||
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
|
||||
* FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
|
||||
* COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
|
||||
* AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
|
||||
* WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
|
||||
*/
|
||||
package com.vitorpamplona.amethyst.ui.screen.loggedIn.chats.marmotGroup
|
||||
|
||||
import androidx.compose.animation.AnimatedVisibility
|
||||
import androidx.compose.foundation.background
|
||||
import androidx.compose.foundation.layout.Column
|
||||
import androidx.compose.foundation.layout.Row
|
||||
import androidx.compose.foundation.layout.fillMaxWidth
|
||||
import androidx.compose.foundation.layout.padding
|
||||
import androidx.compose.foundation.shape.RoundedCornerShape
|
||||
import androidx.compose.material3.MaterialTheme
|
||||
import androidx.compose.material3.Text
|
||||
import androidx.compose.runtime.Composable
|
||||
import androidx.compose.ui.Alignment
|
||||
import androidx.compose.ui.Modifier
|
||||
import androidx.compose.ui.draw.clip
|
||||
import androidx.compose.ui.text.font.FontStyle
|
||||
import androidx.compose.ui.unit.dp
|
||||
import com.vitorpamplona.amethyst.commons.marmot.AgentStreamPreview
|
||||
import com.vitorpamplona.quartz.marmot.appComponents.agentTextStream.PreviewStatus
|
||||
|
||||
/**
|
||||
* The live agent-preview row, shown between the transcript and the composer.
|
||||
*
|
||||
* The whole point of this row is that it is NOT the transcript. Preview text
|
||||
* is provisional until the durable kind:9 lands and its transcript hash agrees
|
||||
* with what we folded — every record can open individually and the stream
|
||||
* still be wrong, if one was dropped, reordered or injected. So it renders in
|
||||
* italic on a tinted surface with an explicit label, and it disappears the
|
||||
* moment the real message arrives (confirmed) or is contradicted (dropped).
|
||||
*
|
||||
* `ProgressDelta` and `Status` records never reach [AgentStreamPreview.text] —
|
||||
* the spec keeps them out of preview text, notifications, indexes and
|
||||
* automation input — so they render here only as a separate, quieter line.
|
||||
*/
|
||||
@Composable
|
||||
fun AgentStreamPreviewBanner(
|
||||
preview: AgentStreamPreview?,
|
||||
modifier: Modifier = Modifier,
|
||||
) {
|
||||
// An aborted preview produces no durable text at all: the publisher
|
||||
// withdrew it, so there is nothing honest left to show.
|
||||
val visible = preview != null && preview.status != PreviewStatus.ABORTED && !preview.isConfirmed
|
||||
|
||||
AnimatedVisibility(visible = visible) {
|
||||
if (preview == null) return@AnimatedVisibility
|
||||
Column(
|
||||
modifier =
|
||||
modifier
|
||||
.fillMaxWidth()
|
||||
.padding(horizontal = 10.dp, vertical = 4.dp)
|
||||
.clip(RoundedCornerShape(8.dp))
|
||||
.background(MaterialTheme.colorScheme.surfaceVariant)
|
||||
.padding(horizontal = 10.dp, vertical = 6.dp),
|
||||
) {
|
||||
Row(verticalAlignment = Alignment.CenterVertically) {
|
||||
Text(
|
||||
text =
|
||||
when (preview.status) {
|
||||
PreviewStatus.UNVERIFIABLE -> "Live preview (incomplete)"
|
||||
else -> "Live preview"
|
||||
},
|
||||
style = MaterialTheme.typography.labelSmall,
|
||||
color = MaterialTheme.colorScheme.onSurfaceVariant,
|
||||
)
|
||||
preview.statusLabel?.let {
|
||||
Text(
|
||||
text = " · $it",
|
||||
style = MaterialTheme.typography.labelSmall,
|
||||
color = MaterialTheme.colorScheme.onSurfaceVariant,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
if (preview.text.isNotEmpty()) {
|
||||
Text(
|
||||
text = preview.text,
|
||||
style = MaterialTheme.typography.bodyMedium,
|
||||
fontStyle = FontStyle.Italic,
|
||||
color = MaterialTheme.colorScheme.onSurfaceVariant,
|
||||
)
|
||||
}
|
||||
|
||||
preview.progressLabel?.let {
|
||||
Text(
|
||||
text = it,
|
||||
style = MaterialTheme.typography.labelSmall,
|
||||
color = MaterialTheme.colorScheme.onSurfaceVariant,
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
+31
@@ -44,8 +44,10 @@ import androidx.compose.ui.Modifier
|
||||
import androidx.compose.ui.graphics.Color
|
||||
import androidx.compose.ui.platform.LocalContext
|
||||
import androidx.compose.ui.unit.dp
|
||||
import androidx.lifecycle.compose.collectAsStateWithLifecycle
|
||||
import androidx.lifecycle.viewmodel.compose.viewModel
|
||||
import com.vitorpamplona.amethyst.R
|
||||
import com.vitorpamplona.amethyst.commons.marmot.MarmotAgentStreamWatcher
|
||||
import com.vitorpamplona.amethyst.commons.resources.Res
|
||||
import com.vitorpamplona.amethyst.commons.resources.marmot_group_default_name
|
||||
import com.vitorpamplona.amethyst.ui.actions.MentionPreservingInputTransformation
|
||||
@@ -76,6 +78,7 @@ import com.vitorpamplona.quartz.nip01Core.core.HexKey
|
||||
import kotlinx.collections.immutable.ImmutableList
|
||||
import kotlinx.collections.immutable.persistentListOf
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
import kotlinx.coroutines.flow.MutableStateFlow
|
||||
import kotlinx.coroutines.launch
|
||||
|
||||
@Composable
|
||||
@@ -120,6 +123,32 @@ fun MarmotGroupChatView(
|
||||
}
|
||||
}
|
||||
|
||||
// The live agent-preview watcher. It follows the newest kind:1200 in the
|
||||
// group and folds the QUIC records behind it; a group with no stream, no
|
||||
// broker candidate or no reachable broker simply never shows a preview,
|
||||
// and the durable kind:9 still arrives as ordinary chat either way.
|
||||
val marmot = accountViewModel.account.marmotManager
|
||||
val streamScope = rememberCoroutineScope()
|
||||
val streamWatcher =
|
||||
remember(nostrGroupId, marmot) {
|
||||
marmot?.let {
|
||||
MarmotAgentStreamWatcher(it, accountViewModel.account.marmotStreamTransport, streamScope)
|
||||
}
|
||||
}
|
||||
val streamPreview by (streamWatcher?.preview ?: remember { MutableStateFlow(null) }).collectAsStateWithLifecycle()
|
||||
|
||||
// Re-check on every feed change: a kind:1200 arrives as an ordinary group
|
||||
// message, so "the feed moved" is exactly when a new stream may have been
|
||||
// anchored. watchLatest is idempotent for a stream already being followed.
|
||||
val feedState by feedViewModel.feedState.feedContent.collectAsStateWithLifecycle()
|
||||
LaunchedEffect(feedState, streamWatcher) {
|
||||
streamWatcher?.watchLatest(nostrGroupId)
|
||||
}
|
||||
|
||||
DisposableEffect(streamWatcher) {
|
||||
onDispose { streamWatcher?.stop() }
|
||||
}
|
||||
|
||||
Column(Modifier.fillMaxHeight()) {
|
||||
Column(
|
||||
modifier =
|
||||
@@ -137,6 +166,8 @@ fun MarmotGroupChatView(
|
||||
)
|
||||
}
|
||||
|
||||
AgentStreamPreviewBanner(streamPreview)
|
||||
|
||||
Spacer(modifier = DoubleVertSpacer)
|
||||
|
||||
MarmotGroupMessageComposer(
|
||||
|
||||
+257
@@ -0,0 +1,257 @@
|
||||
/*
|
||||
* Copyright (c) 2025 Vitor Pamplona
|
||||
*
|
||||
* Permission is hereby granted, free of charge, to any person obtaining a copy of
|
||||
* this software and associated documentation files (the "Software"), to deal in
|
||||
* the Software without restriction, including without limitation the rights to use,
|
||||
* copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
|
||||
* Software, and to permit persons to whom the Software is furnished to do so,
|
||||
* subject to the following conditions:
|
||||
*
|
||||
* The above copyright notice and this permission notice shall be included in all
|
||||
* copies or substantial portions of the Software.
|
||||
*
|
||||
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
|
||||
* FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
|
||||
* COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
|
||||
* AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
|
||||
* WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
|
||||
*/
|
||||
package com.vitorpamplona.amethyst.commons.marmot
|
||||
|
||||
import com.vitorpamplona.quartz.marmot.appComponents.agentTextStream.AgentTextStreamFinal
|
||||
import com.vitorpamplona.quartz.marmot.appComponents.agentTextStream.AgentTextStreamStart
|
||||
import com.vitorpamplona.quartz.marmot.appComponents.agentTextStream.AgentTextStreamSubscriber
|
||||
import com.vitorpamplona.quartz.marmot.appComponents.agentTextStream.PreviewStatus
|
||||
import com.vitorpamplona.quartz.marmot.appComponents.agentTextStream.transport.MarmotQuicTransport
|
||||
import com.vitorpamplona.quartz.nip01Core.core.Event
|
||||
import com.vitorpamplona.quartz.nip01Core.core.HexKey
|
||||
import com.vitorpamplona.quartz.nip01Core.core.hexToByteArray
|
||||
import com.vitorpamplona.quartz.utils.Log
|
||||
import kotlinx.coroutines.CoroutineScope
|
||||
import kotlinx.coroutines.Job
|
||||
import kotlinx.coroutines.flow.MutableStateFlow
|
||||
import kotlinx.coroutines.flow.StateFlow
|
||||
import kotlinx.coroutines.flow.asStateFlow
|
||||
import kotlinx.coroutines.launch
|
||||
import kotlinx.coroutines.sync.Mutex
|
||||
import kotlinx.coroutines.sync.withLock
|
||||
|
||||
/**
|
||||
* What a front end renders for one live agent text stream.
|
||||
*
|
||||
* [isConfirmed] is the only thing that licenses showing this as ordinary
|
||||
* content. Until the durable kind:9 lands and its transcript matches what we
|
||||
* folded, this is provisional and a renderer MUST make it visibly distinct —
|
||||
* every record can open individually and the stream still be wrong, if one was
|
||||
* dropped, reordered or injected.
|
||||
*/
|
||||
class AgentStreamPreview(
|
||||
val streamId: HexKey,
|
||||
val startEventId: HexKey,
|
||||
/** The account that anchored the stream — not necessarily the group's agent. */
|
||||
val author: HexKey,
|
||||
val text: String,
|
||||
val status: PreviewStatus,
|
||||
/** Latest `Status` label, for chrome. Never part of the answer text. */
|
||||
val statusLabel: String? = null,
|
||||
/** Latest `ProgressDelta`, for chrome. Never part of the answer text. */
|
||||
val progressLabel: String? = null,
|
||||
val isConfirmed: Boolean = false,
|
||||
)
|
||||
|
||||
/**
|
||||
* Watches one Marmot group for an agent text stream and exposes it as UI state.
|
||||
*
|
||||
* The live preview is a progressive enhancement, and every failure here is
|
||||
* meant to look like "no preview" rather than like a broken group: a group
|
||||
* with no broker candidate, a candidate that will not connect, a platform with
|
||||
* no QUIC at all, or a stream type we do not implement all end the same way —
|
||||
* the group still works and the final kind:9 still arrives as normal chat.
|
||||
*
|
||||
* [transport] is null on a platform that cannot open a raw QUIC connection.
|
||||
* That is a supported configuration, not a degraded one: `receive` explicitly
|
||||
* does not require implementing the QUIC data plane.
|
||||
*/
|
||||
class MarmotAgentStreamWatcher(
|
||||
private val marmot: MarmotManager,
|
||||
private val transport: MarmotQuicTransport?,
|
||||
private val scope: CoroutineScope,
|
||||
) {
|
||||
private val mutable = MutableStateFlow<AgentStreamPreview?>(null)
|
||||
val preview: StateFlow<AgentStreamPreview?> = mutable.asStateFlow()
|
||||
|
||||
private val mutex = Mutex()
|
||||
private var job: Job? = null
|
||||
private var watchingStreamId: HexKey? = null
|
||||
private var subscriber: AgentTextStreamSubscriber? = null
|
||||
|
||||
/**
|
||||
* Start (or keep) watching the newest agent text stream in [nostrGroupId].
|
||||
*
|
||||
* Idempotent: calling it again for a stream already being watched does
|
||||
* nothing, so a front end can call it on every feed update.
|
||||
*/
|
||||
suspend fun watchLatest(nostrGroupId: HexKey) {
|
||||
// Resolve first: the durable message may already be in the log (a
|
||||
// catch-up sync delivers the whole stream at once), and a preview we
|
||||
// can no longer improve should be settled before we open a socket.
|
||||
resolveAgainstStoredFinal(nostrGroupId)
|
||||
|
||||
if (transport == null) return
|
||||
val anchor = findLatestStart(nostrGroupId) ?: return
|
||||
val (startEvent, start) = anchor
|
||||
|
||||
// A stream type or route we do not implement is not an error: ignore
|
||||
// the live route and let the final message do its job.
|
||||
if (!start.isTextProfile || !start.isQuicRoute || start.brokerCandidates.isEmpty()) return
|
||||
|
||||
mutex.withLock {
|
||||
if (watchingStreamId == start.streamId) return@withLock
|
||||
job?.cancel()
|
||||
watchingStreamId = start.streamId
|
||||
mutable.value = null
|
||||
job = scope.launch { follow(nostrGroupId, startEvent, start) }
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* A durable kind:9 closed a stream out. Confirms the preview when our fold
|
||||
* agrees with it, and drops the preview when it does not — a disagreement
|
||||
* means we rendered something the publisher did not send, so the durable
|
||||
* message is the only thing that should remain on screen.
|
||||
*/
|
||||
fun onFinal(
|
||||
streamId: HexKey,
|
||||
transcriptHash: HexKey,
|
||||
chunkCount: Long,
|
||||
) {
|
||||
val current = mutable.value ?: return
|
||||
if (!current.streamId.equals(streamId, ignoreCase = true)) return
|
||||
val folded = subscriber
|
||||
val matches = folded != null && folded.matchesFinal(transcriptHash.hexToByteArray(), chunkCount)
|
||||
mutable.value = if (matches) AgentStreamPreviewCopy.confirmed(current) else null
|
||||
if (!matches) {
|
||||
Log.d("MarmotAgentStreamWatcher") {
|
||||
"stream ${streamId.take(8)}… did not match its final message — dropping the preview"
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Apply the durable kind:9 for the stream being previewed, if the group's
|
||||
* log already holds it.
|
||||
*
|
||||
* A front end only has to say "the feed moved"; deciding whether a preview
|
||||
* is confirmed, contradicted or still pending is this class's job, and
|
||||
* keeping it here is what makes it testable without a UI.
|
||||
*/
|
||||
private suspend fun resolveAgainstStoredFinal(nostrGroupId: HexKey) {
|
||||
val current = mutable.value ?: return
|
||||
for (line in marmot.loadStoredMessages(nostrGroupId)) {
|
||||
val parsed = Event.fromJsonOrNull(line) ?: continue
|
||||
if (parsed.kind != AgentTextStreamStart.FINAL_KIND_TEXT) continue
|
||||
val final = AgentTextStreamFinal.fromTags(parsed.tags) ?: continue
|
||||
if (!final.streamId.equals(current.streamId, ignoreCase = true)) continue
|
||||
onFinal(final.streamId, final.transcriptHash, final.chunkCount)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
/** Stop watching and clear the preview. */
|
||||
fun stop() {
|
||||
job?.cancel()
|
||||
job = null
|
||||
watchingStreamId = null
|
||||
subscriber = null
|
||||
mutable.value = null
|
||||
}
|
||||
|
||||
private suspend fun follow(
|
||||
nostrGroupId: HexKey,
|
||||
startEvent: Event,
|
||||
start: AgentTextStreamStart,
|
||||
) {
|
||||
val quic = transport ?: return
|
||||
// The epoch that DELIVERED the anchor, not the group's current one —
|
||||
// the record key context binds it, and a commit landing in between
|
||||
// would otherwise derive a key nobody else is using.
|
||||
val epoch = marmot.storedEpochs(nostrGroupId)[startEvent.id]
|
||||
val crypto =
|
||||
try {
|
||||
marmot.agentTextStreamCrypto(
|
||||
nostrGroupId = nostrGroupId,
|
||||
streamId = start.streamId.hexToByteArray(),
|
||||
startEventId = startEvent.id.hexToByteArray(),
|
||||
senderPubKey = startEvent.pubKey,
|
||||
epoch = epoch,
|
||||
)
|
||||
} catch (e: Exception) {
|
||||
Log.w("MarmotAgentStreamWatcher", "cannot derive stream keys for $nostrGroupId", e)
|
||||
return
|
||||
}
|
||||
|
||||
val folding = AgentTextStreamSubscriber(crypto)
|
||||
subscriber = folding
|
||||
|
||||
// "A receiver tries advertised candidates in listed order"; the first
|
||||
// that yields the matching stream wins, and one that fails is simply
|
||||
// skipped.
|
||||
for (candidate in start.brokerCandidates) {
|
||||
val stream =
|
||||
try {
|
||||
quic.subscribe(candidate, start.streamId.hexToByteArray(), startEvent.id.hexToByteArray())
|
||||
} catch (e: Exception) {
|
||||
Log.d("MarmotAgentStreamWatcher") { "candidate $candidate unusable: ${e.message}" }
|
||||
continue
|
||||
}
|
||||
try {
|
||||
stream.incoming().collect { record ->
|
||||
folding.accept(record)
|
||||
mutable.value =
|
||||
AgentStreamPreview(
|
||||
streamId = start.streamId,
|
||||
startEventId = startEvent.id,
|
||||
author = startEvent.pubKey,
|
||||
text = folding.previewText,
|
||||
status = folding.status,
|
||||
statusLabel = folding.latestStatus,
|
||||
progressLabel = folding.latestProgress,
|
||||
isConfirmed = false,
|
||||
)
|
||||
}
|
||||
} catch (e: Exception) {
|
||||
Log.d("MarmotAgentStreamWatcher") { "stream from $candidate ended: ${e.message}" }
|
||||
} finally {
|
||||
runCatching { stream.close() }
|
||||
}
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
/** Newest kind:1200 in the group's decrypted log, with its own event. */
|
||||
private suspend fun findLatestStart(nostrGroupId: HexKey): Pair<Event, AgentTextStreamStart>? {
|
||||
var best: Pair<Event, AgentTextStreamStart>? = null
|
||||
for (line in marmot.loadStoredMessages(nostrGroupId)) {
|
||||
val parsed = Event.fromJsonOrNull(line) ?: continue
|
||||
val start = AgentTextStreamStart.fromTags(parsed.kind, parsed.tags) ?: continue
|
||||
if (best == null || parsed.createdAt >= best.first.createdAt) best = parsed to start
|
||||
}
|
||||
return best
|
||||
}
|
||||
}
|
||||
|
||||
private object AgentStreamPreviewCopy {
|
||||
fun confirmed(p: AgentStreamPreview) =
|
||||
AgentStreamPreview(
|
||||
streamId = p.streamId,
|
||||
startEventId = p.startEventId,
|
||||
author = p.author,
|
||||
text = p.text,
|
||||
status = p.status,
|
||||
statusLabel = p.statusLabel,
|
||||
progressLabel = p.progressLabel,
|
||||
isConfirmed = true,
|
||||
)
|
||||
}
|
||||
+337
@@ -0,0 +1,337 @@
|
||||
/*
|
||||
* Copyright (c) 2025 Vitor Pamplona
|
||||
*
|
||||
* Permission is hereby granted, free of charge, to any person obtaining a copy of
|
||||
* this software and associated documentation files (the "Software"), to deal in
|
||||
* the Software without restriction, including without limitation the rights to use,
|
||||
* copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
|
||||
* Software, and to permit persons to whom the Software is furnished to do so,
|
||||
* subject to the following conditions:
|
||||
*
|
||||
* The above copyright notice and this permission notice shall be included in all
|
||||
* copies or substantial portions of the Software.
|
||||
*
|
||||
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
|
||||
* FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
|
||||
* COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
|
||||
* AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
|
||||
* WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
|
||||
*/
|
||||
package com.vitorpamplona.amethyst.commons.marmot
|
||||
|
||||
import com.vitorpamplona.quartz.marmot.appComponents.agentTextStream.AgentTextStreamPublisher
|
||||
import com.vitorpamplona.quartz.marmot.appComponents.agentTextStream.AgentTextStreamRecordV1
|
||||
import com.vitorpamplona.quartz.marmot.appComponents.agentTextStream.InMemoryAgentTextStreamSequenceStore
|
||||
import com.vitorpamplona.quartz.marmot.appComponents.agentTextStream.PreviewStatus
|
||||
import com.vitorpamplona.quartz.marmot.appComponents.agentTextStream.transport.MarmotQuicException
|
||||
import com.vitorpamplona.quartz.marmot.appComponents.agentTextStream.transport.MarmotQuicStream
|
||||
import com.vitorpamplona.quartz.marmot.appComponents.agentTextStream.transport.MarmotQuicTransport
|
||||
import com.vitorpamplona.quartz.marmot.mip00KeyPackages.KeyPackageBundleStore
|
||||
import com.vitorpamplona.quartz.marmot.mip01Groups.MarmotGroupData
|
||||
import com.vitorpamplona.quartz.marmot.mls.group.MarmotMessageStore
|
||||
import com.vitorpamplona.quartz.marmot.mls.group.MlsGroupStateStore
|
||||
import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair
|
||||
import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerInternal
|
||||
import kotlinx.coroutines.CompletableDeferred
|
||||
import kotlinx.coroutines.flow.Flow
|
||||
import kotlinx.coroutines.flow.MutableSharedFlow
|
||||
import kotlinx.coroutines.flow.first
|
||||
import kotlinx.coroutines.runBlocking
|
||||
import kotlinx.coroutines.withTimeout
|
||||
import kotlin.test.Test
|
||||
import kotlin.test.assertEquals
|
||||
import kotlin.test.assertNotNull
|
||||
import kotlin.test.assertNull
|
||||
import kotlin.test.assertTrue
|
||||
|
||||
/**
|
||||
* The front end's half of an agent text stream: notice the kind:1200 that
|
||||
* arrived in a group, render the live preview it points at, and hand back to
|
||||
* the durable kind:9 when it lands.
|
||||
*
|
||||
* Everything the renderer needs to be honest is decided here — whether the
|
||||
* preview may be shown as confirmed, and whether it turned out to be the
|
||||
* stream the publisher actually sent.
|
||||
*/
|
||||
class MarmotAgentStreamWatcherTest {
|
||||
private val nostrGroupId = "c".repeat(64)
|
||||
|
||||
/** A transport whose records the test pushes by hand. */
|
||||
private class FakeTransport : MarmotQuicTransport {
|
||||
val records = MutableSharedFlow<AgentTextStreamRecordV1>(replay = 32)
|
||||
val subscribed = CompletableDeferred<String>()
|
||||
var failEveryCandidate = false
|
||||
|
||||
override suspend fun publish(
|
||||
candidate: String,
|
||||
streamId: ByteArray,
|
||||
startEventId: ByteArray,
|
||||
): MarmotQuicStream = error("the watcher never publishes")
|
||||
|
||||
override suspend fun subscribe(
|
||||
candidate: String,
|
||||
streamId: ByteArray,
|
||||
startEventId: ByteArray,
|
||||
): MarmotQuicStream {
|
||||
if (failEveryCandidate) {
|
||||
throw MarmotQuicException(MarmotQuicException.Kind.HandshakeFailed, "no route to $candidate")
|
||||
}
|
||||
if (!subscribed.isCompleted) subscribed.complete(candidate)
|
||||
return object : MarmotQuicStream {
|
||||
override suspend fun send(record: AgentTextStreamRecordV1) = error("read only")
|
||||
|
||||
override fun incoming(): Flow<AgentTextStreamRecordV1> = records
|
||||
|
||||
override suspend fun finish() = Unit
|
||||
|
||||
override suspend fun close() = Unit
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private fun manager() =
|
||||
MarmotManager(
|
||||
NostrSignerInternal(KeyPair()),
|
||||
WatcherStateStore(),
|
||||
WatcherMessageStore(),
|
||||
WatcherBundleStore(),
|
||||
)
|
||||
|
||||
private suspend fun aGroupWithAStream(
|
||||
manager: MarmotManager,
|
||||
brokers: List<String> = listOf("quic://broker.invalid:4450"),
|
||||
): Pair<String, String> {
|
||||
manager.createGroup(
|
||||
nostrGroupId,
|
||||
MarmotGroupData(nostrGroupId = nostrGroupId, name = "stream group", relays = listOf("wss://relay.invalid")),
|
||||
)
|
||||
val streamId = "a".repeat(64)
|
||||
val start = manager.buildAgentStreamStart(nostrGroupId, streamId, brokers)
|
||||
return streamId to start.innerEvent.id
|
||||
}
|
||||
|
||||
@Test
|
||||
fun aPreviewAppearsAsRecordsArriveAndIsNeverShownAsConfirmed() =
|
||||
runBlocking {
|
||||
val manager = manager()
|
||||
val transport = FakeTransport()
|
||||
val (streamId, startEventId) = aGroupWithAStream(manager)
|
||||
val watcher = MarmotAgentStreamWatcher(manager, transport, this)
|
||||
|
||||
watcher.watchLatest(nostrGroupId)
|
||||
withTimeout(5_000) { transport.subscribed.await() }
|
||||
|
||||
val crypto = manager.agentTextStreamCrypto(nostrGroupId, hex(streamId), hex(startEventId))
|
||||
val publisher = AgentTextStreamPublisher.open(crypto, InMemoryAgentTextStreamSequenceStore())
|
||||
transport.records.emit(publisher.publish(AgentTextStreamRecordV1.TYPE_TEXT_DELTA, "half an ".encodeToByteArray()))
|
||||
transport.records.emit(publisher.publish(AgentTextStreamRecordV1.TYPE_TEXT_DELTA, "answer".encodeToByteArray()))
|
||||
|
||||
val preview = withTimeout(5_000) { watcher.preview.first { it?.text == "half an answer" } }
|
||||
assertNotNull(preview)
|
||||
assertEquals(streamId, preview.streamId)
|
||||
assertEquals(PreviewStatus.LIVE, preview.status)
|
||||
assertTrue(
|
||||
!preview.isConfirmed,
|
||||
"a live preview is provisional — a renderer must be able to tell it apart from durable content",
|
||||
)
|
||||
watcher.stop()
|
||||
}
|
||||
|
||||
@Test
|
||||
fun theFinalMessageConfirmsAPreviewThatMatchesIt() =
|
||||
runBlocking {
|
||||
val manager = manager()
|
||||
val transport = FakeTransport()
|
||||
val (streamId, startEventId) = aGroupWithAStream(manager)
|
||||
val watcher = MarmotAgentStreamWatcher(manager, transport, this)
|
||||
watcher.watchLatest(nostrGroupId)
|
||||
withTimeout(5_000) { transport.subscribed.await() }
|
||||
|
||||
val crypto = manager.agentTextStreamCrypto(nostrGroupId, hex(streamId), hex(startEventId))
|
||||
val publisher = AgentTextStreamPublisher.open(crypto, InMemoryAgentTextStreamSequenceStore())
|
||||
transport.records.emit(publisher.publish(AgentTextStreamRecordV1.TYPE_TEXT_DELTA, "the answer".encodeToByteArray()))
|
||||
withTimeout(5_000) { watcher.preview.first { it?.text == "the answer" } }
|
||||
|
||||
watcher.onFinal(streamId, publisher.transcript.hash.asHex(), publisher.transcript.chunkCount)
|
||||
val confirmed = assertNotNull(withTimeout(5_000) { watcher.preview.first { it?.isConfirmed == true } })
|
||||
assertEquals("the answer", confirmed.text)
|
||||
watcher.stop()
|
||||
}
|
||||
|
||||
@Test
|
||||
fun aFinalThatDisagreesDiscardsThePreviewInsteadOfShowingIt() =
|
||||
runBlocking {
|
||||
val manager = manager()
|
||||
val transport = FakeTransport()
|
||||
val (streamId, startEventId) = aGroupWithAStream(manager)
|
||||
val watcher = MarmotAgentStreamWatcher(manager, transport, this)
|
||||
watcher.watchLatest(nostrGroupId)
|
||||
withTimeout(5_000) { transport.subscribed.await() }
|
||||
|
||||
val crypto = manager.agentTextStreamCrypto(nostrGroupId, hex(streamId), hex(startEventId))
|
||||
val publisher = AgentTextStreamPublisher.open(crypto, InMemoryAgentTextStreamSequenceStore())
|
||||
transport.records.emit(publisher.publish(AgentTextStreamRecordV1.TYPE_TEXT_DELTA, "tampered".encodeToByteArray()))
|
||||
withTimeout(5_000) { watcher.preview.first { it?.text == "tampered" } }
|
||||
|
||||
// A transcript that does not match means records were dropped,
|
||||
// reordered or injected even though each one opened. The durable
|
||||
// kind:9 is the answer; the preview must go.
|
||||
watcher.onFinal(streamId, "0".repeat(64), 1)
|
||||
withTimeout(5_000) { watcher.preview.first { it == null } }
|
||||
watcher.stop()
|
||||
}
|
||||
|
||||
@Test
|
||||
fun aFinalAlreadyInTheLogSettlesThePreviewWithoutTheUiSayingSo() =
|
||||
runBlocking {
|
||||
val manager = manager()
|
||||
val transport = FakeTransport()
|
||||
val (streamId, startEventId) = aGroupWithAStream(manager)
|
||||
val watcher = MarmotAgentStreamWatcher(manager, transport, this)
|
||||
watcher.watchLatest(nostrGroupId)
|
||||
withTimeout(5_000) { transport.subscribed.await() }
|
||||
|
||||
val crypto = manager.agentTextStreamCrypto(nostrGroupId, hex(streamId), hex(startEventId))
|
||||
val publisher = AgentTextStreamPublisher.open(crypto, InMemoryAgentTextStreamSequenceStore())
|
||||
transport.records.emit(publisher.publish(AgentTextStreamRecordV1.TYPE_TEXT_DELTA, "done".encodeToByteArray()))
|
||||
withTimeout(5_000) { watcher.preview.first { it?.text == "done" } }
|
||||
|
||||
// The durable message lands in the group log the ordinary way. A
|
||||
// front end only reports "the feed moved"; the watcher does the
|
||||
// rest.
|
||||
manager.buildAgentStreamFinal(
|
||||
nostrGroupId,
|
||||
streamId,
|
||||
publisher.transcript.hash.asHex(),
|
||||
publisher.transcript.chunkCount,
|
||||
"done",
|
||||
)
|
||||
watcher.watchLatest(nostrGroupId)
|
||||
|
||||
val confirmed = assertNotNull(withTimeout(5_000) { watcher.preview.first { it?.isConfirmed == true } })
|
||||
assertEquals("done", confirmed.text)
|
||||
watcher.stop()
|
||||
}
|
||||
|
||||
@Test
|
||||
fun aGroupWithNoBrokerCandidateShowsNoPreviewAtAll() =
|
||||
runBlocking {
|
||||
val manager = manager()
|
||||
val transport = FakeTransport()
|
||||
aGroupWithAStream(manager, brokers = emptyList())
|
||||
val watcher = MarmotAgentStreamWatcher(manager, transport, this)
|
||||
|
||||
watcher.watchLatest(nostrGroupId)
|
||||
// Zero candidates is valid: the preview is simply unavailable and
|
||||
// every member still gets the final message.
|
||||
assertNull(watcher.preview.value)
|
||||
watcher.stop()
|
||||
}
|
||||
|
||||
@Test
|
||||
fun anUnreachableBrokerLeavesTheGroupUsableWithoutAPreview() =
|
||||
runBlocking {
|
||||
val manager = manager()
|
||||
val transport = FakeTransport().also { it.failEveryCandidate = true }
|
||||
aGroupWithAStream(manager)
|
||||
val watcher = MarmotAgentStreamWatcher(manager, transport, this)
|
||||
|
||||
watcher.watchLatest(nostrGroupId)
|
||||
assertNull(
|
||||
watcher.preview.value,
|
||||
"a candidate that will not connect is skipped, not fatal",
|
||||
)
|
||||
watcher.stop()
|
||||
}
|
||||
|
||||
@Test
|
||||
fun aPlatformWithoutQuicSimplyNeverPreviews() =
|
||||
runBlocking {
|
||||
val manager = manager()
|
||||
aGroupWithAStream(manager)
|
||||
val watcher = MarmotAgentStreamWatcher(manager, transport = null, scope = this)
|
||||
|
||||
watcher.watchLatest(nostrGroupId)
|
||||
assertNull(watcher.preview.value)
|
||||
watcher.stop()
|
||||
}
|
||||
|
||||
private fun hex(s: String) = ByteArray(s.length / 2) { ((s[it * 2].digitToInt(16) shl 4) or s[it * 2 + 1].digitToInt(16)).toByte() }
|
||||
|
||||
private fun ByteArray.asHex() = joinToString("") { (it.toInt() and 0xff).toString(16).padStart(2, '0') }
|
||||
}
|
||||
|
||||
private class WatcherStateStore : MlsGroupStateStore {
|
||||
private val states = mutableMapOf<String, ByteArray>()
|
||||
private val retained = mutableMapOf<String, List<ByteArray>>()
|
||||
|
||||
override suspend fun save(
|
||||
nostrGroupId: String,
|
||||
state: ByteArray,
|
||||
) {
|
||||
states[nostrGroupId] = state
|
||||
}
|
||||
|
||||
override suspend fun load(nostrGroupId: String): ByteArray? = states[nostrGroupId]
|
||||
|
||||
override suspend fun delete(nostrGroupId: String) {
|
||||
states.remove(nostrGroupId)
|
||||
retained.remove(nostrGroupId)
|
||||
}
|
||||
|
||||
override suspend fun listGroups(): List<String> = states.keys.toList()
|
||||
|
||||
override suspend fun saveRetainedEpochs(
|
||||
nostrGroupId: String,
|
||||
retainedSecrets: List<ByteArray>,
|
||||
) {
|
||||
retained[nostrGroupId] = retainedSecrets
|
||||
}
|
||||
|
||||
override suspend fun loadRetainedEpochs(nostrGroupId: String): List<ByteArray> = retained[nostrGroupId] ?: emptyList()
|
||||
}
|
||||
|
||||
private class WatcherMessageStore : MarmotMessageStore {
|
||||
private val messages = mutableMapOf<String, MutableList<String>>()
|
||||
private val epochs = mutableMapOf<String, MutableMap<String, Long>>()
|
||||
|
||||
override suspend fun appendMessage(
|
||||
nostrGroupId: String,
|
||||
innerEventJson: String,
|
||||
) {
|
||||
val log = messages.getOrPut(nostrGroupId) { mutableListOf() }
|
||||
if (innerEventJson !in log) log.add(innerEventJson)
|
||||
}
|
||||
|
||||
override suspend fun loadMessages(nostrGroupId: String): List<String> = messages[nostrGroupId]?.toList() ?: emptyList()
|
||||
|
||||
override suspend fun delete(nostrGroupId: String) {
|
||||
messages.remove(nostrGroupId)
|
||||
epochs.remove(nostrGroupId)
|
||||
}
|
||||
|
||||
override suspend fun recordEpoch(
|
||||
nostrGroupId: String,
|
||||
innerEventId: String,
|
||||
epoch: Long,
|
||||
) {
|
||||
epochs.getOrPut(nostrGroupId) { mutableMapOf() }[innerEventId] = epoch
|
||||
}
|
||||
|
||||
override suspend fun loadEpochs(nostrGroupId: String): Map<String, Long> = epochs[nostrGroupId]?.toMap() ?: emptyMap()
|
||||
}
|
||||
|
||||
private class WatcherBundleStore : KeyPackageBundleStore {
|
||||
private var snapshot: ByteArray? = null
|
||||
|
||||
override suspend fun save(snapshot: ByteArray) {
|
||||
this.snapshot = snapshot
|
||||
}
|
||||
|
||||
override suspend fun load(): ByteArray? = snapshot
|
||||
|
||||
override suspend fun delete() {
|
||||
snapshot = null
|
||||
}
|
||||
}
|
||||
@@ -76,9 +76,10 @@ amy marmot stream finish GID --stream-id … --transcript-hash … --chunk-count
|
||||
|
||||
## Not done
|
||||
|
||||
- The GUIs do not originate or render a stream yet, which is why the `send`
|
||||
(`0xF2D2`) and `fanout` (`0xF2D4`) role capabilities stay unadvertised — a
|
||||
role is a promise to the whole group.
|
||||
- The Android GUI renders previews but does not originate a stream — that is
|
||||
an agent's job, and no agent runs in the app yet. Only `amy` publishes one.
|
||||
- The desktop app has no Marmot chat screen at all, so there is nothing to
|
||||
render a preview into. The watcher it would use already lives in `commons`.
|
||||
- The direct path (`marmot.quic_stream.v1`) is unimplemented. v1 defines no
|
||||
start-payload candidate format for it, so it is only reachable with an
|
||||
endpoint known out of band.
|
||||
|
||||
@@ -695,12 +695,11 @@ test we have.
|
||||
in-memory `AgentTextStreamSequenceStore` exists; a platform-backed one lands
|
||||
with the transport that needs it.
|
||||
|
||||
We still do NOT advertise `send` (`0xF2D2`) or `fanout` (`0xF2D4`) in
|
||||
KeyPackage capabilities. Everything behind them now works end to end, but the
|
||||
roles are a promise to a whole group and the GUIs do not yet originate or
|
||||
render a stream — only the CLI does. A group whose policy requires `send` is
|
||||
refused at join rather than joined into a state every peer would reject us
|
||||
from.
|
||||
Our leaf now advertises all three roles — `receive` (`0xF2D1`), `send`
|
||||
(`0xF2D2`) and `fanout` (`0xF2D4`) — which is the same set MDK puts on every
|
||||
KeyPackage it publishes, so a group requiring any of them admits us. A
|
||||
capability is a claim about what the client supports, not a duty to stream: a
|
||||
member that never originates one is a quiet member, not a broken one.
|
||||
|
||||
- **The QUIC transport binding is implemented and verified against MDK's
|
||||
broker.** `transports/quic.md` is a RAW QUIC binding — its own ALPNs
|
||||
@@ -723,6 +722,19 @@ test we have.
|
||||
start-payload candidate format for it, so it is only usable with an
|
||||
out-of-band endpoint.
|
||||
|
||||
- **The Android GUI renders live previews.** `MarmotAgentStreamWatcher` in
|
||||
`commons` follows the newest kind:1200 in a group, folds the QUIC records
|
||||
behind it under the receive discipline, and settles the preview against the
|
||||
durable kind:9 — confirmed when the transcript agrees, dropped when it does
|
||||
not, because a disagreement means we rendered something the publisher did not
|
||||
send. It is deliberately in `commons` and speaks only quartz's transport
|
||||
port, so it is testable without a UI and the desktop app inherits it the day
|
||||
it grows a Marmot chat screen. Android's chat view shows it as an italic,
|
||||
labelled row between the transcript and the composer: provisional content has
|
||||
to look provisional. Every failure path — no stream, no candidate, an
|
||||
unreachable broker, a platform with no QUIC at all — ends as "no preview",
|
||||
never as a broken group.
|
||||
|
||||
- **The feature is wired end to end, both directions, against MDK.**
|
||||
`amy marmot stream start|send|watch|finish` mints the kind-1200 anchor,
|
||||
pushes records through a broker, folds a preview under the receive discipline
|
||||
|
||||
@@ -3442,7 +3442,21 @@ class MlsGroup private constructor(
|
||||
listOf(
|
||||
AppDataDictionary.EXTENSION_TYPE,
|
||||
MarmotGroupData.EXTENSION_ID_INT,
|
||||
// All three agent-stream roles, matching what MDK puts
|
||||
// on every KeyPackage it publishes. `receive` is the
|
||||
// baseline compatibility role; `send` says we can
|
||||
// originate preview records, which we can now that the
|
||||
// publisher, the raw-QUIC binding and the app wiring
|
||||
// exist; `fanout` says records may be forwarded on our
|
||||
// behalf, which is what using a broker at all means.
|
||||
//
|
||||
// A capability is only a claim about what we support,
|
||||
// not a duty to stream: a group that requires `send`
|
||||
// wants members that COULD originate, and a member that
|
||||
// never does is a quiet member, not a broken one.
|
||||
AgentTextStreamRoles.RECEIVE_CAPABILITY,
|
||||
AgentTextStreamRoles.SEND_CAPABILITY,
|
||||
AgentTextStreamRoles.FANOUT_CAPABILITY,
|
||||
),
|
||||
proposals = listOf(APP_DATA_UPDATE_PROPOSAL_TYPE, SELF_REMOVE_PROPOSAL_TYPE),
|
||||
)
|
||||
|
||||
+6
-3
@@ -107,12 +107,13 @@ class CurrentProfileGroupFactoryTest {
|
||||
assertContentEquals(ByteArray(0), kpDictionary[AppComponentIds.LAST_RESORT_KEY_PACKAGE])
|
||||
|
||||
// Capabilities advertise the draft extension the current profile
|
||||
// needs, plus the legacy 0xF2EE group-data extension and the
|
||||
// agent-text-stream RECEIVE role.
|
||||
// needs, plus the legacy 0xF2EE group-data extension and all three
|
||||
// agent-text-stream roles — the same set MDK puts on every
|
||||
// KeyPackage it publishes.
|
||||
//
|
||||
// The extra entries are deliberate and are NOT drift from the MDK
|
||||
// reference. A capability says "this client can handle it", and a
|
||||
// group that REQUIRES 0xF2EE (legacy) or 0xF2D1 (any group MDK
|
||||
// group that REQUIRES 0xF2EE (legacy) or a role (any group MDK
|
||||
// creates) refuses to add a leaf that does not advertise it — so
|
||||
// without these a current-profile KeyPackage would be un-addable
|
||||
// to every legacy group that already exists and to every group MDK
|
||||
@@ -123,6 +124,8 @@ class CurrentProfileGroupFactoryTest {
|
||||
AppDataDictionary.EXTENSION_TYPE,
|
||||
MarmotGroupData.EXTENSION_ID_INT,
|
||||
AgentTextStreamRoles.RECEIVE_CAPABILITY,
|
||||
AgentTextStreamRoles.SEND_CAPABILITY,
|
||||
AgentTextStreamRoles.FANOUT_CAPABILITY,
|
||||
),
|
||||
kp.leafNode.capabilities.extensions,
|
||||
)
|
||||
|
||||
+76
-5
@@ -23,6 +23,12 @@ package com.vitorpamplona.quartz.marmot.mls.group
|
||||
import com.vitorpamplona.quartz.marmot.appComponents.CurrentProfileGroupFactory
|
||||
import com.vitorpamplona.quartz.marmot.appComponents.GroupProfileV1
|
||||
import com.vitorpamplona.quartz.marmot.appComponents.agentTextStream.AgentTextStreamQuicPolicyV1
|
||||
import com.vitorpamplona.quartz.marmot.appComponents.agentTextStream.AgentTextStreamRoles
|
||||
import com.vitorpamplona.quartz.marmot.mls.crypto.Ed25519
|
||||
import com.vitorpamplona.quartz.marmot.mls.crypto.Ed25519KeyPair
|
||||
import com.vitorpamplona.quartz.marmot.mls.messages.KeyPackageBundle
|
||||
import com.vitorpamplona.quartz.marmot.mls.tree.Capabilities
|
||||
import com.vitorpamplona.quartz.nip01Core.core.hexToByteArray
|
||||
import com.vitorpamplona.quartz.nip01Core.core.toHexKey
|
||||
import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair
|
||||
import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerInternal
|
||||
@@ -88,12 +94,19 @@ class CurrentProfileWelcomeTest {
|
||||
|
||||
/**
|
||||
* The `0x8006` policy names MLS leaf capabilities every member must
|
||||
* advertise. Our current-profile leaf advertises `receive` only, so a
|
||||
* group that also requires `send` must be refused at join rather than
|
||||
* joined into a state where every commit we make is rejected by peers.
|
||||
* advertise, and a group that requires one we do not advertise has to be
|
||||
* refused at join — joining anyway lands us in a group where peers reject
|
||||
* every commit we make.
|
||||
*
|
||||
* The gate is tested against a deliberately reduced leaf rather than
|
||||
* against our own KeyPackage, because our own now advertises all three
|
||||
* defined roles (see [aJoinerFillsEveryRoleTheProfileDefines]) and so
|
||||
* cannot fail this check. Testing "we refuse what we cannot fill" through
|
||||
* our own capability set would silently stop testing anything the moment
|
||||
* that set changed — which is exactly what happened here.
|
||||
*/
|
||||
@Test
|
||||
fun aJoinerRefusesAGroupWhoseStreamRolesItCannotFill() =
|
||||
fun aJoinerRefusesAGroupWhoseStreamRolesItsLeafDoesNotAdvertise() =
|
||||
runBlocking<Unit> {
|
||||
val group =
|
||||
aGroup(
|
||||
@@ -105,7 +118,7 @@ class CurrentProfileWelcomeTest {
|
||||
paddingBucketBytes = 0,
|
||||
),
|
||||
)
|
||||
val invitee = CurrentProfileGroupFactory.createKeyPackage(signer(0x44))
|
||||
val invitee = receiveOnlyKeyPackage(signer(0x44))
|
||||
group.proposeAdd(invitee.keyPackage.toTlsBytes())
|
||||
val welcome = assertNotNull(group.commit().welcomeBytes)
|
||||
|
||||
@@ -116,6 +129,64 @@ class CurrentProfileWelcomeTest {
|
||||
)
|
||||
}
|
||||
|
||||
/**
|
||||
* Our published leaf advertises `receive`, `send` AND `fanout` — the same
|
||||
* set MDK puts on every KeyPackage — so a group that requires any of them
|
||||
* admits us.
|
||||
*/
|
||||
@Test
|
||||
fun aJoinerFillsEveryRoleTheProfileDefines() =
|
||||
runBlocking<Unit> {
|
||||
val group =
|
||||
aGroup(
|
||||
AgentTextStreamQuicPolicyV1(
|
||||
requiredMemberRoles = AgentTextStreamRoles.MASK,
|
||||
allowedMemberRoles = AgentTextStreamRoles.MASK,
|
||||
maxPlaintextFrameLen = 4096,
|
||||
replayTtlSecs = 0,
|
||||
paddingBucketBytes = 0,
|
||||
),
|
||||
)
|
||||
val invitee = CurrentProfileGroupFactory.createKeyPackage(signer(0x66))
|
||||
group.proposeAdd(invitee.keyPackage.toTlsBytes())
|
||||
val welcome = assertNotNull(group.commit().welcomeBytes)
|
||||
|
||||
val joined = MlsGroup.processWelcome(welcome, invitee)
|
||||
assertEquals(nostrGroupId.toHexKey(), joined.currentNostrGroupId())
|
||||
}
|
||||
|
||||
/** A current-profile leaf with the `send` and `fanout` roles stripped. */
|
||||
private suspend fun receiveOnlyKeyPackage(signer: NostrSignerInternal): KeyPackageBundle {
|
||||
val full = CurrentProfileGroupFactory.createKeyPackage(signer)
|
||||
val reduced =
|
||||
MlsGroup.currentProfileLeafCapabilities().let {
|
||||
Capabilities(
|
||||
extensions = it.extensions.filterNot { ext -> ext == AgentTextStreamRoles.SEND_CAPABILITY || ext == AgentTextStreamRoles.FANOUT_CAPABILITY },
|
||||
proposals = it.proposals,
|
||||
)
|
||||
}
|
||||
val identity = signer.pubKey.hexToByteArray()
|
||||
// The SAME signature keypair: the leaf's account identity proof covers
|
||||
// its own signature key, so a fresh one would fail proof validation
|
||||
// before the role gate is ever reached and the test would pass for the
|
||||
// wrong reason.
|
||||
val leafKeys =
|
||||
Ed25519KeyPair(
|
||||
privateKey = full.signaturePrivateKey,
|
||||
publicKey = Ed25519.publicFromPrivate(full.signaturePrivateKey),
|
||||
)
|
||||
return MlsGroup
|
||||
.create(identity)
|
||||
.createKeyPackage(
|
||||
identity = identity,
|
||||
signingKey = full.signaturePrivateKey,
|
||||
leafSignatureKeyPair = leafKeys,
|
||||
leafExtensions = full.keyPackage.leafNode.extensions,
|
||||
capabilities = reduced,
|
||||
keyPackageExtensions = full.keyPackage.extensions,
|
||||
)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun aJoinerAcceptsAGroupRequiringOnlyTheReceiveRole() =
|
||||
runBlocking<Unit> {
|
||||
|
||||
Reference in New Issue
Block a user