diff --git a/amethyst/build.gradle.kts b/amethyst/build.gradle.kts index 29d5e686d9..6cf575a8ce 100644 --- a/amethyst/build.gradle.kts +++ b/amethyst/build.gradle.kts @@ -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 diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/model/Account.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/model/Account.kt index b9575826f1..d9f9dcb97b 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/model/Account.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/model/Account.kt @@ -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) diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chats/marmotGroup/AgentStreamPreviewBanner.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chats/marmotGroup/AgentStreamPreviewBanner.kt new file mode 100644 index 0000000000..e561072f7a --- /dev/null +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chats/marmotGroup/AgentStreamPreviewBanner.kt @@ -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, + ) + } + } + } +} diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chats/marmotGroup/MarmotGroupChatView.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chats/marmotGroup/MarmotGroupChatView.kt index 60cd8649fc..6075939f97 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chats/marmotGroup/MarmotGroupChatView.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chats/marmotGroup/MarmotGroupChatView.kt @@ -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( diff --git a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/marmot/MarmotAgentStreamWatcher.kt b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/marmot/MarmotAgentStreamWatcher.kt new file mode 100644 index 0000000000..3b4a1e1381 --- /dev/null +++ b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/marmot/MarmotAgentStreamWatcher.kt @@ -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(null) + val preview: StateFlow = 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? { + var best: Pair? = 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, + ) +} diff --git a/commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/marmot/MarmotAgentStreamWatcherTest.kt b/commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/marmot/MarmotAgentStreamWatcherTest.kt new file mode 100644 index 0000000000..ca6a582a5d --- /dev/null +++ b/commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/marmot/MarmotAgentStreamWatcherTest.kt @@ -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(replay = 32) + val subscribed = CompletableDeferred() + 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 = 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 = listOf("quic://broker.invalid:4450"), + ): Pair { + 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() + private val retained = mutableMapOf>() + + 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 = states.keys.toList() + + override suspend fun saveRetainedEpochs( + nostrGroupId: String, + retainedSecrets: List, + ) { + retained[nostrGroupId] = retainedSecrets + } + + override suspend fun loadRetainedEpochs(nostrGroupId: String): List = retained[nostrGroupId] ?: emptyList() +} + +private class WatcherMessageStore : MarmotMessageStore { + private val messages = mutableMapOf>() + private val epochs = mutableMapOf>() + + 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 = 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 = 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 + } +} diff --git a/marmotQuic/README.md b/marmotQuic/README.md index 3f9c1e1837..b0ac4608e5 100644 --- a/marmotQuic/README.md +++ b/marmotQuic/README.md @@ -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. diff --git a/quartz/plans/2026-09-08-marmot-spec-resync.md b/quartz/plans/2026-09-08-marmot-spec-resync.md index c6865b72cf..3ad2d44bc5 100644 --- a/quartz/plans/2026-09-08-marmot-spec-resync.md +++ b/quartz/plans/2026-09-08-marmot-spec-resync.md @@ -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 diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/marmot/mls/group/MlsGroup.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/marmot/mls/group/MlsGroup.kt index 09ff4aeb05..bd24d33d1a 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/marmot/mls/group/MlsGroup.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/marmot/mls/group/MlsGroup.kt @@ -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), ) diff --git a/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/marmot/appComponents/CurrentProfileGroupFactoryTest.kt b/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/marmot/appComponents/CurrentProfileGroupFactoryTest.kt index 96cdd0a214..d99adc5d01 100644 --- a/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/marmot/appComponents/CurrentProfileGroupFactoryTest.kt +++ b/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/marmot/appComponents/CurrentProfileGroupFactoryTest.kt @@ -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, ) diff --git a/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/marmot/mls/group/CurrentProfileWelcomeTest.kt b/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/marmot/mls/group/CurrentProfileWelcomeTest.kt index 476770671a..8b4288df3c 100644 --- a/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/marmot/mls/group/CurrentProfileWelcomeTest.kt +++ b/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/marmot/mls/group/CurrentProfileWelcomeTest.kt @@ -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 { 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 { + 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 {