Fix conversation captions, streamed meanings, and final usage recovery

apps/android/app/src/androidTest/java/chat/mural/CaptionParityTest.kt

@@ -18,6 +18,7 @@ import okhttp3.HttpUrl.Companion.toHttpUrl
 import okhttp3.MediaType.Companion.toMediaType
 import okhttp3.ResponseBody.Companion.toResponseBody
 import okio.Buffer
+import okio.buffer
 import kotlinx.serialization.json.*
 import org.junit.*
 import org.junit.Assert.*
@@ -37,6 +38,7 @@ class CaptionParityTest {
     private var originalHasKey = false
     private val requests = LinkedBlockingQueue<String>()
     @Volatile private var responseGate: CountDownLatch? = null
+    @Volatile private var streamGate: CountDownLatch? = null
     @Volatile private var responseCode = 200
     @Volatile private var response = "This word is explained in the context of the sentence."
     private val recording get() = InstrumentationRegistry.getArguments().getString("record") == "true"
@@ -65,6 +67,30 @@ class CaptionParityTest {
                     }) })
                     put("usage", buildJsonObject { put("input_tokens", 0); put("output_tokens", 0) })
                 }
+                if (Json.parseToJsonElement(body).jsonObject["stream"] == JsonPrimitive(true) && status == 200) {
+                    val partial = buildJsonObject { put("type", "response.output_text.delta"); put("delta", reply.take(5)) }
+                    val completion = buildJsonObject { put("type", "response.completed"); put("response", payload) }
+                    val prefix = Buffer().writeUtf8("data: $partial\n\n")
+                    val suffix = Buffer().writeUtf8("data: $completion\n\n")
+                    val gate = streamGate
+                    val source = object : okio.Source {
+                        override fun read(sink: Buffer, byteCount: Long): Long {
+                            if (!prefix.exhausted()) return prefix.read(sink, byteCount)
+                            gate?.await(15, TimeUnit.SECONDS)
+                            return suffix.read(sink, byteCount)
+                        }
+                        override fun timeout() = okio.Timeout.NONE
+                        override fun close() { gate?.countDown() }
+                    }
+                    val streamBody = object : ResponseBody() {
+                        private val buffered = source.buffer()
+                        override fun contentType() = "text/event-stream".toMediaType()
+                        override fun contentLength() = -1L
+                        override fun source() = buffered
+                    }
+                    return@addInterceptor Response.Builder().request(chain.request()).protocol(Protocol.HTTP_1_1).code(200).message("OK")
+                        .header("Content-Type", "text/event-stream").body(streamBody).build()
+                }
                 Response.Builder().request(chain.request()).protocol(Protocol.HTTP_1_1).code(status).message(if (status == 200) "OK" else "Fixture error")
                     .body(payload.toString().toResponseBody("application/json".toMediaType())).build()
             }.build()
@@ -81,6 +107,7 @@ class CaptionParityTest {
     }
     @After fun restore() {
         responseGate?.countDown()
+        streamGate?.countDown()
         finishRecording()
         if (!::vm.isInitialized || !::original.isInitialized) return
         compose.runOnIdle {
@@ -205,6 +232,44 @@ class CaptionParityTest {
         }
     }
 
+    @Test fun everyLanguagePreservesProviderFragmentsAndDisplaysMeaningBeforeCompletion() {
+        val samples = mapOf(
+            "nb" to listOf("Hygg", "elig! Jeg liker fri", "luftsliv."),
+            "en" to listOf("That is inter", "esting."),
+            "es" to listOf("Me gusta apren", "der espa", "ñol."),
+            "fr" to listOf("Aujourd", "’hui, c’est inté", "ressant."),
+            "de" to listOf("Das ist eine Sprach", "lern", "anwendung."),
+            "it" to listOf("È una conver", "sazione interes", "sante."),
+            "pt" to listOf("Estou apren", "dendo portu", "guês."),
+            "zh" to listOf("我", "喜欢", "学习", "中文。"))
+        assertEquals(LanguageRegistry.all.map { it.id }.toSet(), samples.keys)
+        for ((language, parts) in samples) {
+            show(language, "", "")
+            val gate = CountDownLatch(1); streamGate = gate; response = "Meaning for $language"
+            compose.runOnIdle {
+                vm.updatePreferences(vm.archive.preferences.copy(meaningVisible = false))
+                state("session", SessionRecord(languageID = language, title = "Synthetic stream verification"))
+                val handle = MuralViewModel::class.java.getDeclaredMethod("handle", JsonObject::class.java).apply { isAccessible = true }
+                parts.forEachIndexed { index, part -> handle.invoke(vm, buildJsonObject {
+                    put("type", "session.output_transcript.delta"); put("delta", part)
+                    put("start_ms", index * 100); put("end_ms", (index + 1) * 100); put("event_id", "$language-$index")
+                }) }
+                vm.updatePreferences(vm.archive.preferences.copy(meaningVisible = true))
+                MuralViewModel::class.java.getDeclaredMethod("scheduleTranslation", Boolean::class.javaPrimitiveType)
+                    .apply { isAccessible = true }.invoke(vm, true)
+            }
+            compose.onNodeWithTag("target-caption").assertTextEquals(parts.joinToString(""))
+            compose.waitUntil(10_000) { vm.meaning == response.take(5) }
+            compose.onNodeWithTag("meaning-caption").assertTextEquals(response.take(5))
+            compose.runOnIdle { assertTrue(vm.session!!.translations.isEmpty()) }
+            gate.countDown()
+            compose.waitUntil(10_000) { vm.meaning == response }
+            compose.runOnIdle { assertEquals(listOf(response), vm.session!!.translations.values.toList()) }
+            capture("stream-$language")
+            streamGate = null
+        }
+    }
+
     @Test fun endingKeepsOnlyTheSpecificEndReasonNotice() {
         for (reason in listOf("Inactivity", "Time limit", "Ended by you")) {
             show("es", "Hola.", "Hello.")

apps/android/app/src/main/java/chat/mural/MuralViewModel.kt

@@ -200,13 +200,14 @@ class MuralViewModel(application: Application) : AndroidViewModel(application) {
     private var assessmentJob: Job? = null
     private var actionJob: Job? = null
     private val meanings = MeaningController(viewModelScope, canRetryFailure = HostedHelperRetry::canRetryAtBoundary,
-        retryDelay = HostedHelperRetry::automaticDelay) { request ->
+        retryDelay = HostedHelperRetry::automaticDelay, stream = { request, onText -> translateMeaning(request, onText) }) { request -> translateMeaning(request) }
+    private suspend fun translateMeaning(request: MeaningRequest, onText: ((String) -> Unit)? = null): MeaningResult {
         if (archive.preferences.aiConsentVersion != 1) throw IllegalStateException("AI processing consent is required.")
         val module = LanguageRegistry.get(request.learningLanguageID) ?: throw IllegalStateException("Unsupported language.")
         if (request.sessionID in hostedSessionIDs && request.translationInput.toByteArray(Charsets.UTF_8).size > 24_576) throw MeaningInputLimitException()
         val result = teaching(request.sessionID, HelperPurpose.MEANING, request.cacheKey,
-            TeachingPolicy.translation(module, request.meaningLanguage), request.translationInput)
-        MeaningResult(result.text, result.usage.input, result.usage.output)
+            TeachingPolicy.translation(module, request.meaningLanguage), request.translationInput, onText = onText)
+        return MeaningResult(result.text, result.usage.input, result.usage.output)
     }
     private val finalAssessments = FinalAssessmentQueue(viewModelScope) { snapshot, passage -> requestAssessment(snapshot, passage) }
     private val languageDetector = LanguageDetector(application)
@@ -494,13 +495,14 @@ class MuralViewModel(application: Application) : AndroidViewModel(application) {
 
     /** The creating session, rather than the currently selected settings option, chooses every helper. */
     private suspend fun teaching(localID: String?, purpose: HelperPurpose, logicalID: String,
-        instructions: String, input: String, schema: JsonObject? = null, search: Boolean = false): APIResult {
+        instructions: String, input: String, schema: JsonObject? = null, search: Boolean = false, onText: ((String) -> Unit)? = null): APIResult {
         if (archive.preferences.aiConsentVersion != 1) throw HostedFailure.Unavailable
         if (localID != null && localID in hostedSessionIDs) {
-            return hostedBindings.respond(localID, purpose, logicalID, instructions, input, schema, search)
+            return hostedBindings.respond(localID, purpose, logicalID, instructions, input, schema, search, onText)
         }
         if (localID == null && conversationProvider == ConversationProvider.HOSTED_MINUTES) throw HostedFailure.Unavailable
-        return api.respond(instructions, input, schema, search, purpose)
+        return if (onText != null && purpose == HelperPurpose.MEANING) api.streamMeaning(instructions, input, onText)
+        else api.respond(instructions, input, schema, search, purpose)
     }
     private fun helperContext(snapshot: SessionRecord, passage: Passage? = null): String =
         if (snapshot.id in hostedSessionIDs) ConversationHistory.helperContext(snapshot, passage)
@@ -899,7 +901,7 @@ class MuralViewModel(application: Application) : AndroidViewModel(application) {
         val result = teaching(snapshot.id, HelperPurpose.ASSESSMENT, passage.revisionKey,
             TeachingPolicy.assessment(module), helperContext(snapshot, passage), assessmentSchema(module.id))
         val decoded = json.decodeFromString<AssessmentResponse>(result.text)
-        val proposal = Assessment(passage.id, passage.revisionKey, decoded.outcome, decoded.suggestedLevel, decoded.nextGoal, decoded.capability, decoded.words, context = snapshot.themeID ?: "free")
+        val proposal = Assessment(passage.id, passage.revisionKey, decoded.outcome, decoded.suggestedLevel, decoded.nextGoal, decoded.capability, decoded.words, context = snapshot.themeID ?: "free", textAssemblyVersion = 2)
         return FinalAssessmentResult(snapshot.id, snapshot.languageID, proposal, result.usage.input, result.usage.output, result.usage.searches)
     }
     /** Hosted finalization awaits the original helper for its remaining lease window, without entering BYOK recovery. */
@@ -940,14 +942,12 @@ class MuralViewModel(application: Application) : AndroidViewModel(application) {
                 updated.assessments.removeAll { it.passageID == valid.passageID }; updated.assessments += valid
                 addUsage(updated, APIUsage(result.inputTokens, result.outputTokens, result.searchCalls)); save(updated)
                 if (session?.id == updated.id) session = clone(updated)
-                if (state == "active" && session?.id == snapshot.id) {
-                    val progress = learner
+                if (state == "active" && session?.id == snapshot.id &&
+                    session?.passages?.lastOrNull { it.speaker == Speaker.user }?.revisionKey == passage.revisionKey) {
+                    // Keep assessment notes in learning records; injecting them during speech can make the voice read them aloud.
                     if (voiceSession && conversationPace.observe(valid, passage, snapshot.languageID)) {
                         command("instructions", conversationPace.instruction)
                     }
-                    val targetLanguage = LanguageRegistry.get(snapshot.languageID)?.name ?: language.name
-                    val revisit = progress.words.filter { it.dueAt < nowSeconds() }.take(3).joinToString(", ") { it.lemma }
-                    command("thinking", "Teaching context, not spoken text: challenge ${progress.challenge}/5 in $targetLanguage. Next goal: ${progress.nextGoal}. Revisit naturally: $revisit.")
                 }
             } catch (_: CancellationException) { } catch (_: Exception) { /* No unverified progress. */ }
         }
@@ -980,7 +980,8 @@ class MuralViewModel(application: Application) : AndroidViewModel(application) {
             newSession(false); state = "active"; startDurationChecks()
         }
         val id = session!!.id; val token = generation
-        val offset = ((nowSeconds() - session!!.startedAt) * 1000).toInt().coerceAtLeast(0)
+        val offset = if (voiceSession) session!!.nextTypedVoiceOffsetMS
+            else ((nowSeconds() - session!!.startedAt) * 1000).toInt().coerceAtLeast(0)
         val fragment = Fragment(speaker = Speaker.user, text = clean, startMS = offset, endMS = offset + 1, meaningVisible = archive.preferences.meaningVisible, typed = true)
         val draft = clone(session!!).also { it.append(fragment) }
         if (voiceSession) { activity.learnerEngaged(activityNow()); inactivitySeconds = null }; working = true

apps/android/app/src/main/java/chat/mural/core/ConversationProviders.kt

@@ -91,7 +91,10 @@ object HostedHelperRetry {
 class HostedConversationBindings(private val scope: CoroutineScope, private val now: () -> Long = System::currentTimeMillis) {
     class Lease(val serverID: String, val teaching: TeachingClient, val close: suspend () -> Unit,
         val status: suspend () -> HostedSessionStatus, val deadlineMilliseconds: Long = Long.MAX_VALUE)
-    private class Attempt(val result: Deferred<APIResult>, var usageDelivered: Boolean = false)
+    private class Attempt(val result: Deferred<APIResult>, var usageDelivered: Boolean = false) {
+        var partial: String? = null
+        val listeners = mutableSetOf<(String) -> Unit>()
+    }
     private class Binding(val ownerID: String, val lease: Lease) {
         var endedAt: Long? = null
         var confirmedClosed = false
@@ -132,39 +135,48 @@ class HostedConversationBindings(private val scope: CoroutineScope, private val
     /** Each logical automatic request owns one deferred result, including an uncertain failure.
      * Cancelling a UI waiter cannot create a second provider bill or discard the request identity. */
     suspend fun respond(localID: String, purpose: HelperPurpose, logicalID: String,
-        instructions: String, input: String, schema: JsonObject? = null, search: Boolean = false): APIResult {
+        instructions: String, input: String, schema: JsonObject? = null, search: Boolean = false, onText: ((String) -> Unit)? = null): APIResult {
         prune()
         val binding = bindings[localID] ?: throw HostedFailure.Unavailable
         if (binding.disabled) throw HostedFailure.Unavailable
         val key = "${purpose.wireValue}:$logicalID"
         binding.attempts[key]?.let {
-            try { return deliver(binding, key, it) }
+            try { return deliver(binding, key, it, onText) }
             catch (failure: Exception) { if (!HostedHelperRetry.isConfirmedNotAdmitted(failure)) throw failure }
         }
         if (binding.attempts.size >= 128) throw HostedFailure.Unavailable
+        lateinit var attempt: Attempt
         val request = helpersScope.async(start = CoroutineStart.LAZY) {
             val ended = binding.endedAt
             if (ended != null) {
                 if (purpose !in POST_END_PURPOSES || now() - ended !in 0 until POST_END_MILLIS) throw HostedFailure.Unavailable
                 if (!closeAndConfirm(localID)) throw HostedFailure.Unconfirmed
                 if (now() - ended !in 0 until POST_END_MILLIS) throw HostedFailure.Unavailable
             }
-            binding.lease.teaching.respond(instructions, input, schema, search, purpose)
+            if (onText != null && purpose == HelperPurpose.MEANING) {
+                binding.lease.teaching.streamMeaning(instructions, input) { partial ->
+                    attempt.partial = partial
+                    attempt.listeners.toList().forEach { it(partial) }
+                }
+            } else binding.lease.teaching.respond(instructions, input, schema, search, purpose)
         }
-        val attempt = Attempt(request)
+        attempt = Attempt(request)
         binding.attempts[key] = attempt
         request.start()
-        return deliver(binding, key, attempt)
+        return deliver(binding, key, attempt, onText)
     }
 
-    private suspend fun deliver(binding: Binding, key: String, attempt: Attempt): APIResult {
-        val result = try { attempt.result.await() }
+    private suspend fun deliver(binding: Binding, key: String, attempt: Attempt, onText: ((String) -> Unit)?): APIResult {
+        val result = try {
+            if (onText != null) { attempt.listeners.add(onText); attempt.partial?.let(onText) }
+            attempt.result.await()
+        }
         catch (failure: Exception) {
             // A concurrent waiter must not remove a later successful retry for this logical request.
             if (HostedHelperRetry.isConfirmedNotAdmitted(failure) && binding.attempts[key] === attempt)
                 binding.attempts.remove(key)
             throw failure
-        }
+        } finally { if (onText != null) attempt.listeners.remove(onText) }
         return if (attempt.usageDelivered) result.copy(usage = APIUsage())
         else { attempt.usageDelivered = true; result }
     }

apps/android/app/src/main/java/chat/mural/core/LearningEngine.kt

@@ -20,16 +20,18 @@ object LearningEngine {
         val passage=session.passages.firstOrNull { it.id==proposal.passageID && it.speaker==Speaker.user } ?: return null
         if (passage.revisionKey != proposal.revisionKey || proposal.suggestedLevel !in 0..5 || proposal.words.size>12) return null
         val allowed=passage.fragments.map { it.id }.toSet()
+        val join: (List<String>) -> String = if (proposal.textAssemblyVersion == null) Passage::legacyJoin else Passage::join
+        val evidenceText = join(passage.fragments.map { it.text })
         val words=proposal.words.mapNotNull { word ->
             if (word.language != session.languageID || word.sourceIDs.isEmpty() || !allowed.containsAll(word.sourceIDs) ||
                 !word.confidence.isFinite() || word.confidence !in 0.8..1.0 || word.lemma.isEmpty() || word.lemma.length>=100 ||
                 word.meaning.isEmpty() || word.meaning.length>=180 || word.form.isEmpty() || word.quote.isEmpty() ||
-                !passage.text.containsCanonical(word.quote) || !word.quote.containsCanonical(word.form)) return@mapNotNull null
-            val refs=Passage.join(passage.fragments.filter { word.sourceIDs.contains(it.id) }.map { it.text })
+                !evidenceText.containsCanonical(word.quote) || !word.quote.containsCanonical(word.form)) return@mapNotNull null
+            val refs=join(passage.fragments.filter { word.sourceIDs.contains(it.id) }.map { it.text })
             if (!refs.containsCanonical(word.quote)) return@mapNotNull null
             var out=word
             if (out.kind==EvidenceKind.independent) {
-                val modeled=session.passages.any { p -> p.speaker==Speaker.assistant && p.startMS<=passage.startMS && passage.startMS-p.endMS<90000 && p.text.containsCanonical(word.form) }
+                val modeled=session.passages.any { p -> p.speaker==Speaker.assistant && p.startMS<=passage.startMS && passage.startMS-p.endMS<90000 && join(p.fragments.map { it.text }).containsCanonical(word.form) }
                 if (passage.fragments.any { it.meaningVisible || it.typed } || modeled) out=out.copy(kind=EvidenceKind.assisted)
             }
             out