diff --git a/src/apps/mobile/AGENTS.md b/src/apps/mobile/AGENTS.md index f8109397e9..151bbed465 100644 --- a/src/apps/mobile/AGENTS.md +++ b/src/apps/mobile/AGENTS.md @@ -119,6 +119,15 @@ isolation, run from `shared/` with the Android guide's Java runtime: `./gradlew :core-feature:jvmTest --tests '*promptSuggestion*' --tests '*SessionRecordReplicaTest*'`. These are simulated host streams, not evidence of a live remote connection. +For a delegation tool's spawned child — child session id retained on the launch +item, one stream per owner, steps attached to that owner's card, child gap +re-read, child failure before content, a child failure that keeps the session +readable instead of failing it, a late child error from a closed session, +unsupported host, cap and re-attach, session switch — run from `shared/` with the +Android guide's Java runtime: +`./gradlew :core-feature:jvmTest --tests '*SubagentBranchCoordinatorTest*'`. +A child transcript is read on demand and never stored on this device. + Native mobile surfaces are controllers. Desktop and CLI own execution, files, terminals and SSH credentials. Mobile workspace selection may use only connections already saved on the controlled runtime; it must not create an SSH connection or instantiate another runtime. Carry the selected connection identity through every operation, and report missing identity instead of falling back to local files. ## Offline Mini Apps diff --git a/src/apps/mobile/android/app/src/androidTest/kotlin/com/openbitfun/mobile/app/ChatMessageBubbleTest.kt b/src/apps/mobile/android/app/src/androidTest/kotlin/com/openbitfun/mobile/app/ChatMessageBubbleTest.kt index 694a2852ee..c01b5486f5 100644 --- a/src/apps/mobile/android/app/src/androidTest/kotlin/com/openbitfun/mobile/app/ChatMessageBubbleTest.kt +++ b/src/apps/mobile/android/app/src/androidTest/kotlin/com/openbitfun/mobile/app/ChatMessageBubbleTest.kt @@ -14,6 +14,7 @@ import androidx.compose.ui.test.onNodeWithText import androidx.compose.ui.test.performClick import androidx.test.platform.app.InstrumentationRegistry import com.openbitfun.mobile.app.ui.chat.ChatMessageBubble +import com.openbitfun.mobile.app.ui.chat.message.LocalReduceMotionOverride import com.openbitfun.mobile.app.ui.chat.message.SUBAGENT_GROUP_TEST_TAG import com.openbitfun.mobile.app.ui.chat.message.TYPING_DOTS_TEST_TAG import com.openbitfun.mobile.core.feature.session.ConversationRow @@ -255,22 +256,35 @@ class ChatMessageBubbleTest { composeRule.onNodeWithText("Working on it").assertIsDisplayed() } + /** + * One turn, drawn with animations removed. + * + * Three fixtures here render a running delegation card, and a running worker's + * state mark pulses forever: a compose test cannot idle on a clock that never + * stops, so without this the pre-existing assertions in this file would wait for + * an idle state that never arrives. The override changes only the movement — the + * reading printed on the card is the same — and + * `SubagentLivenessCardTest.aRunningWorkerMovesAndOnlyItsMarkMoves` is the test + * that measures the pulse, on a fixture that asks for it explicitly. + */ @androidx.compose.runtime.Composable private fun Bubble(row: ConversationRow) { - ChatMessageBubble( - row = row, - enabled = true, - onApproveTool = { _, _ -> }, - onRejectTool = { _, _ -> }, - onCancelTool = { _, _ -> }, - onAnswerTool = { _, _ -> }, - onAnswerToolStructured = { _, _ -> }, - onRetry = {}, - onOpenLink = { _, _ -> }, - previewingRemotePath = "", - previewLoading = false, - modifier = Modifier, - ) + androidx.compose.runtime.CompositionLocalProvider(LocalReduceMotionOverride provides true) { + ChatMessageBubble( + row = row, + enabled = true, + onApproveTool = { _, _ -> }, + onRejectTool = { _, _ -> }, + onCancelTool = { _, _ -> }, + onAnswerTool = { _, _ -> }, + onAnswerToolStructured = { _, _ -> }, + onRetry = {}, + onOpenLink = { _, _ -> }, + previewingRemotePath = "", + previewLoading = false, + modifier = Modifier, + ) + } } private fun row( diff --git a/src/apps/mobile/android/app/src/androidTest/kotlin/com/openbitfun/mobile/app/SubagentBranchCardTest.kt b/src/apps/mobile/android/app/src/androidTest/kotlin/com/openbitfun/mobile/app/SubagentBranchCardTest.kt new file mode 100644 index 0000000000..65e668417e --- /dev/null +++ b/src/apps/mobile/android/app/src/androidTest/kotlin/com/openbitfun/mobile/app/SubagentBranchCardTest.kt @@ -0,0 +1,276 @@ +package com.openbitfun.mobile.app + +import androidx.compose.foundation.layout.Column +import androidx.compose.foundation.layout.fillMaxSize +import androidx.compose.foundation.layout.padding +import androidx.compose.material3.MaterialTheme +import androidx.compose.material3.Surface +import androidx.compose.runtime.CompositionLocalProvider +import androidx.compose.ui.Modifier +import androidx.compose.ui.test.assertCountEquals +import androidx.compose.ui.test.assertIsDisplayed +import androidx.compose.ui.test.hasAnyAncestor +import androidx.compose.ui.test.hasTestTag +import androidx.compose.ui.test.hasText +import androidx.compose.ui.test.junit4.v2.createComposeRule +import androidx.compose.ui.test.onNodeWithTag +import androidx.compose.ui.test.performClick +import androidx.compose.ui.unit.dp +import androidx.test.platform.app.InstrumentationRegistry +import com.openbitfun.mobile.app.ui.chat.message.LocalReduceMotionOverride +import com.openbitfun.mobile.app.ui.chat.message.LocalSubagentAttach +import com.openbitfun.mobile.app.ui.chat.message.LocalSubagentFailures +import com.openbitfun.mobile.app.ui.chat.message.MessageBlockCallbacks +import com.openbitfun.mobile.app.ui.chat.message.MessageBlockList +import com.openbitfun.mobile.app.ui.chat.message.SUBAGENT_GROUP_TEST_TAG +import com.openbitfun.mobile.app.ui.theme.OpenBitFunTheme +import com.openbitfun.mobile.core.feature.session.MessageBlock +import com.openbitfun.mobile.core.feature.session.RemoteSessionFailureReason +import com.openbitfun.mobile.core.feature.workspace.RemoteFileDownloadUiState +import org.junit.Assert.assertEquals +import org.junit.Rule +import org.junit.Test + +private const val OWNER = "tool-owner" +private const val TITLE = "Audit the parser" +private const val STEP_ONE = "Read the parser module" +private const val STEP_TWO = "Run the parser tests" + +/** + * The delegation card as it is painted on a device. + * + * What this proves is the drawing path: a branch block — the shape the shared + * presentation produces for a launch item the coordinator decorated — goes through + * the real `MessageBlockList` -> `SubagentGroup` chain, so the reader really does see + * the worker's steps behind their own expansion, the worker's own lifecycle reading, + * the two "no steps" readings with their own copy, and the header's request to follow + * the worker again reaching the surface that asked for it. + * + * What it does not prove is where those steps came from. The child stream, the cap, + * the gap reset and the unsupported-host discovery are covered by the JVM tests in + * `core-feature` (`SubagentBranchCoordinatorTest`), because none of them is a drawing + * fact and all of them need a host to read from. No transport, no relay, no account + * and no paired session is involved here — the fixture is the coordinator's own + * output shape. + * + * Why the block list and not the whole conversation view: `ConversationView` takes a + * `RemoteSessionUiState.Ready`, and building one from an invented transcript would + * mean this test naming `core-domain` and `core-protocol` types, which the app module + * is not allowed to reach (`scripts/check-mobile-architecture.mjs`, rule 3). The two + * composition locals the card reads are therefore supplied here directly, exactly as + * `ConversationView` supplies them from the state it is handed; that wiring is + * compile-checked, and the store-side half of the same path is asserted in + * `SubagentBranchCoordinatorTest`. + * + * The launch item's status in each fixture is the reading the card is handed, not the + * spawn call's own answer: overriding the spawn's `completed` with the worker's + * `running` is the coordinator's job and is asserted there. What is asserted here is + * that the header prints the status it is given, that the steps are behind the + * reader's own expansion, and that a card with no readable worker says so in the + * reader's language instead of showing an invented step. + */ +class SubagentBranchCardTest { + @get:Rule + val composeRule = createComposeRule() + + @Test + fun aDelegationCardShowsTheWorkersStepsWhileTheChildIsStillRunning() { + val attached = mutableListOf() + setCard( + block = runningCard(children = listOf(step(STEP_ONE), step(STEP_TWO))), + onAttach = { attached += it }, + // The only fixture here whose worker is in flight, so the only one that + // pulses: pinned still so the card can be driven to idle. + reduceMotion = true, + ) + + composeRule.onNodeWithTag(SUBAGENT_GROUP_TEST_TAG).assertIsDisplayed() + assertCardShows(R.string.tool_phase_running) + assertCardHides(R.string.tool_phase_completed) + + // Collapsed by default: the worker's steps are one tap away, not on screen. + assertCardHidesText(STEP_ONE) + + openCard() + + assertCardShowsText(STEP_ONE) + assertCardShowsText(STEP_TWO) + // Opening the card is the gesture that asks the store to follow the worker + // again, and it names the launch item's own tool call — the key the store + // follows that child by. + assertEquals(listOf(OWNER), attached) + } + + @Test + fun aSettledChildKeepsItsStepsAndShowsTheSettledReading() { + setCard(block = runningCard(status = "completed", children = listOf(step(STEP_ONE), step(STEP_TWO)))) + + composeRule.onNodeWithTag(SUBAGENT_GROUP_TEST_TAG).assertIsDisplayed() + assertCardShows(R.string.tool_phase_completed) + assertCardHides(R.string.tool_phase_running) + + openCard() + + // A worker that reported is not a card that empties: its steps stay. + assertCardShowsText(STEP_ONE) + assertCardShowsText(STEP_TWO) + } + + @Test + fun aDesktopWithoutOnDemandStreamsSaysSoAndInventsNoSteps() { + setCard( + block = runningCard(status = "completed", children = emptyList()), + failures = mapOf(OWNER to RemoteSessionFailureReason.HOST_STREAM_UNSUPPORTED), + ) + + composeRule.onNodeWithTag(SUBAGENT_GROUP_TEST_TAG).assertIsDisplayed() + // Nothing was read about the worker, so the launch item's own status stands. + assertCardShows(R.string.tool_phase_completed) + + openCard() + + assertCardShows(R.string.subagent_steps_unsupported) + // The two "no steps" readings are not the same news, and this one is not a + // stream that merely failed. + assertCardHides(R.string.subagent_steps_unavailable) + // No step was made up to fill the card. + assertCardHides(R.string.chat_subagent_output) + assertCardHidesText(STEP_ONE) + } + + @Test + fun aChildStreamThatFailedBeforeAnyContentKeepsTheLaunchStatusAndSaysStepsAreUnavailable() { + setCard( + block = runningCard(status = "completed", children = emptyList()), + failures = mapOf(OWNER to RemoteSessionFailureReason.TRANSPORT), + ) + + composeRule.onNodeWithTag(SUBAGENT_GROUP_TEST_TAG).assertIsDisplayed() + assertCardShows(R.string.tool_phase_completed) + assertCardHides(R.string.tool_phase_running) + + openCard() + + assertCardShows(R.string.subagent_steps_unavailable) + assertCardHides(R.string.subagent_steps_unsupported) + assertCardHides(R.string.chat_subagent_output) + assertCardHidesText(STEP_ONE) + } + + // --- assertions ----------------------------------------------------------- + + /** + * Everything below is read *inside* the card. + * + * Several status words are shared with other surfaces ("Done", "Running"), so a + * whole-screen search would be asking a different question than the one these + * tests answer — and would break the day a "Done" button lands somewhere else. + */ + private fun inCard(text: String) = + hasText(text) and hasAnyAncestor(hasTestTag(SUBAGENT_GROUP_TEST_TAG)) + + private fun assertCardShows(resource: Int) { + composeRule.onNode(inCard(string(resource))).assertIsDisplayed() + } + + private fun assertCardHides(resource: Int) { + composeRule.onAllNodes(inCard(string(resource))).assertCountEquals(0) + } + + private fun assertCardShowsText(text: String) { + composeRule.onNode(inCard(text)).assertIsDisplayed() + } + + private fun assertCardHidesText(text: String) { + composeRule.onAllNodes(inCard(text)).assertCountEquals(0) + } + + /** The reader's own gesture: the card header is the only thing that opens it. */ + private fun openCard() { + composeRule.onNode(inCard(TITLE)).performClick() + composeRule.waitForIdle() + } + + // --- fixtures ------------------------------------------------------------- + + private fun setCard( + block: MessageBlock.Subagent, + failures: Map = emptyMap(), + onAttach: (String) -> Unit = {}, + /** + * The card's own reading is unchanged by this; only its pulse is. + * + * A running worker pulses forever, which is what a reader sees and what + * [SubagentLivenessCardTest] measures in pixels. Here the subject is the + * drawing path, not the movement, so the one fixture whose worker is still + * in flight pins the reading still: a compose test cannot idle on a clock + * that never stops. + */ + reduceMotion: Boolean = false, + ) { + composeRule.setContent { + OpenBitFunTheme(dark = false) { + CompositionLocalProvider( + LocalReduceMotionOverride provides reduceMotion, + LocalSubagentAttach provides onAttach, + LocalSubagentFailures provides failures, + ) { + Surface( + modifier = Modifier.fillMaxSize(), + color = MaterialTheme.colorScheme.background, + contentColor = MaterialTheme.colorScheme.onBackground, + ) { + Column(modifier = Modifier.fillMaxSize().padding(16.dp)) { + MessageBlockList( + blocks = listOf(block), + callbacks = callbacks(), + modifier = Modifier, + ) + } + } + } + } + } + } + + /** + * The card a decorated launch item draws. + * + * [status] is what the header prints; the owner tool call is the key the store + * follows that child by, and it is what opening the card names. [children] is + * what the decoration attaches: the worker's own process, in the worker's own + * order. + */ + private fun runningCard( + status: String = "running", + children: List, + ) = MessageBlock.Subagent( + id = "message-owner-subagent-0", + title = TITLE, + running = status in setOf("running", "tool_calling"), + text = "", + children = children, + status = status, + ownerToolCallId = OWNER, + ) + + private fun step(content: String) = MessageBlock.Text("$OWNER-$content", content, false) + + private fun callbacks() = MessageBlockCallbacks( + enabled = true, + onApproveTool = { _, _ -> }, + onRejectTool = { _, _ -> }, + onCancelTool = { _, _ -> }, + onAnswerTool = { _, _ -> }, + onAnswerToolStructured = { _, _ -> }, + onOpenLink = { _, _ -> }, + previewingRemotePath = "", + previewLoading = false, + download = RemoteFileDownloadUiState.None, + onDownloadFile = { _, _ -> }, + downloadEnabled = true, + ) + + private fun string(resource: Int): String = + InstrumentationRegistry.getInstrumentation().targetContext.getString(resource) +} diff --git a/src/apps/mobile/android/app/src/androidTest/kotlin/com/openbitfun/mobile/app/SubagentLivenessCardTest.kt b/src/apps/mobile/android/app/src/androidTest/kotlin/com/openbitfun/mobile/app/SubagentLivenessCardTest.kt new file mode 100644 index 0000000000..71577ce228 --- /dev/null +++ b/src/apps/mobile/android/app/src/androidTest/kotlin/com/openbitfun/mobile/app/SubagentLivenessCardTest.kt @@ -0,0 +1,309 @@ +package com.openbitfun.mobile.app + +import androidx.compose.foundation.layout.Column +import androidx.compose.foundation.layout.fillMaxSize +import androidx.compose.foundation.layout.padding +import androidx.compose.material3.MaterialTheme +import androidx.compose.material3.Surface +import androidx.compose.runtime.CompositionLocalProvider +import androidx.compose.ui.Modifier +import androidx.compose.ui.graphics.ImageBitmap +import androidx.compose.ui.graphics.toArgb +import androidx.compose.ui.graphics.toPixelMap +import androidx.compose.ui.test.assertCountEquals +import androidx.compose.ui.test.captureToImage +import androidx.compose.ui.test.hasText +import androidx.compose.ui.test.junit4.v2.createComposeRule +import androidx.compose.ui.test.onAllNodesWithTag +import androidx.compose.ui.test.onAllNodesWithText +import androidx.compose.ui.test.onNodeWithContentDescription +import androidx.compose.ui.test.onNodeWithTag +import androidx.compose.ui.test.performClick +import androidx.compose.ui.semantics.SemanticsProperties +import androidx.compose.ui.unit.dp +import androidx.test.platform.app.InstrumentationRegistry +import com.openbitfun.mobile.app.ui.chat.message.LocalReduceMotionOverride +import com.openbitfun.mobile.app.ui.chat.message.MessageBlockCallbacks +import com.openbitfun.mobile.app.ui.chat.message.MessageBlockList +import com.openbitfun.mobile.app.ui.chat.message.SUBAGENT_AVATAR_TEST_TAG +import com.openbitfun.mobile.app.ui.chat.message.SUBAGENT_GROUP_TEST_TAG +import com.openbitfun.mobile.app.ui.chat.message.SUBAGENT_STATE_MARK_TEST_TAG +import com.openbitfun.mobile.app.ui.theme.OpenBitFunTheme +import com.openbitfun.mobile.core.feature.session.MessageBlock +import com.openbitfun.mobile.core.feature.workspace.RemoteFileDownloadUiState +import org.junit.Assert.assertEquals +import org.junit.Assert.assertFalse +import org.junit.Assert.assertNotEquals +import org.junit.Assert.assertTrue +import org.junit.Rule +import org.junit.Test + +/** + * The delegation card as it is painted on a device. + * + * The JVM tests own the mapping (which host word means which reading) and the + * identity derivation. This file owns what a reader's eyes and a screen reader get: + * that the six readings are drawn differently, that two spawn cards with one title + * still look like two workers, that the header says what tapping it does, and that + * a running worker actually moves — measured in pixels, on the device, in a + * composition whose only input is written here. No session, no host, no account. + */ +class SubagentLivenessCardTest { + @get:Rule + val composeRule = createComposeRule() + + private val context = InstrumentationRegistry.getInstrumentation().targetContext + + @Test + fun everyReadingIsDrawnInItsOwnWords() { + // The card never ends, so the clock is driven by hand. + composeRule.mainClock.autoAdvance = false + setCards( + listOf( + card("a", "tool-a", "Audit the parser", "tool_calling"), + card("b", "tool-b", "Summarize the diff", "finishing"), + card("c", "tool-c", "Draft the release note", "queued"), + card("d", "tool-d", "Run the parser tests", "completed"), + card("e", "tool-e", "Rewrite the lexer", "cancelled"), + card("f", "tool-f", "Deploy the preview", "failed"), + ), + ) + composeRule.mainClock.advanceTimeByFrame() + + assertCardSays("Audit the parser", R.string.tool_phase_running) + assertCardSays("Summarize the diff", R.string.subagent_state_finishing) + assertCardSays("Draft the release note", R.string.tool_phase_waiting) + assertCardSays("Run the parser tests", R.string.tool_phase_completed) + assertCardSays("Rewrite the lexer", R.string.tool_phase_cancelled) + assertCardSays("Deploy the preview", R.string.tool_phase_failed) + } + + @Test + fun aQueuedWorkerIsNeverReportedAsDone() { + setCards(listOf(card("c", "tool-c", "Draft the release note", "queued"))) + + assertCardSays("Draft the release note", R.string.tool_phase_waiting) + // The reported symptom: a worker the host is still holding, printed as done. + composeRule.onAllNodesWithText(string(R.string.tool_phase_completed)).assertCountEquals(0) + } + + @Test + fun everyReadingHasItsOwnMarkAndNotOnlyItsOwnWord() { + composeRule.mainClock.autoAdvance = false + setCards( + listOf( + card("a", "tool-a", "Audit the parser", "tool_calling"), + card("b", "tool-b", "Summarize the diff", "finishing"), + card("c", "tool-c", "Draft the release note", "queued"), + card("d", "tool-d", "Run the parser tests", "completed"), + card("e", "tool-e", "Rewrite the lexer", "cancelled"), + card("f", "tool-f", "Deploy the preview", "failed"), + ), + ) + composeRule.mainClock.advanceTimeByFrame() + + // Same box, same density, six different pictures: the state is legible before + // the word is read, and no two readings share a mark. + val marks = composeRule.onAllNodesWithTag(SUBAGENT_STATE_MARK_TEST_TAG, useUnmergedTree = true) + val images = (0 until 6).map { marks[it].captureToImage() } + images.forEachIndexed { index, image -> + assertTrue("mark $index was not drawn", image.width > 0 && image.height > 0) + } + for (first in images.indices) { + for (second in first + 1 until images.size) { + assertNotEquals( + "readings $first and $second draw the same mark", + images[first].pixels(), + images[second].pixels(), + ) + } + } + } + + @Test + fun twoSpawnCardsWithOneTitleStillHaveTwoFaces() { + composeRule.mainClock.autoAdvance = false + setCards( + listOf( + card("a", "tool-owner", "AgentSpawn", "tool_calling"), + card("b", "tool-owner-2", "AgentSpawn", "tool_calling"), + ), + ) + composeRule.mainClock.advanceTimeByFrame() + + // The reported symptom: two spawn cards a reader could not tell apart. + assertEquals(2, composeRule.onAllNodesWithText("AgentSpawn").fetchSemanticsNodes().size) + val avatars = composeRule.onAllNodesWithTag(SUBAGENT_AVATAR_TEST_TAG, useUnmergedTree = true) + val first = avatars[0].captureToImage() + val second = avatars[1].captureToImage() + assertEquals(first.width, second.width) + assertEquals(first.height, second.height) + assertNotEquals("two workers were drawn with one face", first.pixels(), second.pixels()) + } + + @Test + fun theHeaderSaysWhatTappingItDoesAndTheAvatarSaysNothingTwice() { + setCards(listOf(card("c", "tool-c", "Draft the release note", "queued"))) + + // The avatar is decoration: the title already names the worker, so a screen + // reader that read the avatar too would say it twice. + val avatar = composeRule.onNodeWithTag(SUBAGENT_AVATAR_TEST_TAG, useUnmergedTree = true) + .fetchSemanticsNode() + assertFalse( + "the avatar must not repeat the name a screen reader already reads", + avatar.config.contains(SemanticsProperties.ContentDescription), + ) + + // The header is the only control that opens the steps, so it announces that. + composeRule.onNodeWithContentDescription(string(R.string.subagent_expand_steps)).assertExists() + composeRule.onNode(hasText("Draft the release note")).performClick() + composeRule.onNodeWithContentDescription(string(R.string.subagent_collapse_steps)).assertExists() + // And the state is announced with the name, in the same node. + composeRule.onNode( + hasText("Draft the release note") and hasText(string(R.string.tool_phase_waiting)), + ).assertExists() + } + + @Test + fun aRunningWorkerMovesAndOnlyItsMarkMoves() { + composeRule.mainClock.autoAdvance = false + setCards(listOf(card("a", "tool-a", "Audit the parser", "tool_calling"))) + composeRule.mainClock.advanceTimeByFrame() + + val card = composeRule.onNodeWithTag(SUBAGENT_GROUP_TEST_TAG) + val before = card.captureToImage() + val cardBounds = card.fetchSemanticsNode().boundsInRoot + val markBounds = composeRule + .onNodeWithTag(SUBAGENT_STATE_MARK_TEST_TAG, useUnmergedTree = true) + .fetchSemanticsNode().boundsInRoot + + // Half the pulse, which is what a second of watching looks like at 60fps. + composeRule.mainClock.advanceTimeBy(450) + val after = card.captureToImage() + + val changed = changedPixels(before, after) + assertTrue("a running worker must move", changed.isNotEmpty()) + // Everything that changed is the indicator: the name, the avatar and the + // chevron are the same pixels in both frames. + changed.forEach { (x, y) -> + val rootX = cardBounds.left + x + val rootY = cardBounds.top + y + assertTrue( + "pixel ($rootX,$rootY) changed outside the state mark $markBounds", + rootX >= markBounds.left - 1f && rootX <= markBounds.right + 1f && + rootY >= markBounds.top - 1f && rootY <= markBounds.bottom + 1f, + ) + } + } + + @Test + fun aReaderWhoRemovedAnimationsGetsTheSameCardTwice() { + composeRule.mainClock.autoAdvance = false + setCards(listOf(card("a", "tool-a", "Audit the parser", "tool_calling")), reduceMotion = true) + composeRule.mainClock.advanceTimeByFrame() + + val card = composeRule.onNodeWithTag(SUBAGENT_GROUP_TEST_TAG) + val before = card.captureToImage() + composeRule.mainClock.advanceTimeBy(450) + val after = card.captureToImage() + + // The reading is still there; only the movement is gone. + assertCardSays("Audit the parser", R.string.tool_phase_running) + assertTrue( + "reduced motion must not animate", + changedPixels(before, after).isEmpty(), + ) + } + + // --- assertions ----------------------------------------------------------- + + /** The name and the reading are announced together, in the card's own header. */ + private fun assertCardSays(title: String, word: Int) { + composeRule.onNode(hasText(title) and hasText(string(word))).assertExists() + } + + private fun string(resource: Int): String = context.getString(resource) + + // --- fixtures ------------------------------------------------------------- + + private fun setCards(cards: List, reduceMotion: Boolean = false) { + composeRule.setContent { + OpenBitFunTheme(dark = false) { + CompositionLocalProvider(LocalReduceMotionOverride provides reduceMotion) { + Surface( + modifier = Modifier.fillMaxSize(), + color = MaterialTheme.colorScheme.background, + contentColor = MaterialTheme.colorScheme.onBackground, + ) { + Column( + modifier = Modifier.fillMaxSize().padding(16.dp), + ) { + MessageBlockList( + blocks = cards, + callbacks = callbacks(), + modifier = Modifier, + ) + } + } + } + } + } + } + + /** + * One card, written here. + * + * [status] is the host word the transcript carries; [owner] is the launch item's + * tool call, which is what a worker's identity is derived from. The steps are + * what makes the card openable, and they are the same in every fixture because + * they are not what this file is about. + */ + private fun card(id: String, owner: String, title: String, status: String) = MessageBlock.Subagent( + id = id, + title = title, + running = status in setOf("running", "tool_calling", "queued", "finishing"), + text = "", + children = listOf( + MessageBlock.Text("$id-step-1", "Read the module", false), + MessageBlock.Text("$id-step-2", "Report what changed", false), + ), + status = status, + ownerToolCallId = owner, + ) + + private fun callbacks() = MessageBlockCallbacks( + enabled = true, + onApproveTool = { _, _ -> }, + onRejectTool = { _, _ -> }, + onCancelTool = { _, _ -> }, + onAnswerTool = { _, _ -> }, + onAnswerToolStructured = { _, _ -> }, + onOpenLink = { _, _ -> }, + previewingRemotePath = "", + previewLoading = false, + download = RemoteFileDownloadUiState.None, + onDownloadFile = { _, _ -> }, + downloadEnabled = true, + ) + + // --- pixel measurement ---------------------------------------------------- + + /** Every pixel that differs, as card-local coordinates. */ + private fun changedPixels(first: ImageBitmap, second: ImageBitmap): List> { + assertEquals("captures must be the same size", first.pixels().size, second.pixels().size) + val a = first.toPixelMap() + val b = second.toPixelMap() + val changed = mutableListOf>() + for (y in 0 until first.height) { + for (x in 0 until first.width) { + if (a[x, y].toArgb() != b[x, y].toArgb()) changed += x to y + } + } + return changed + } + + private fun ImageBitmap.pixels(): List { + val map = toPixelMap() + return (0 until height).flatMap { y -> (0 until width).map { x -> map[x, y].toArgb() } } + } +} diff --git a/src/apps/mobile/android/app/src/main/kotlin/com/openbitfun/mobile/app/ui/chat/ConversationView.kt b/src/apps/mobile/android/app/src/main/kotlin/com/openbitfun/mobile/app/ui/chat/ConversationView.kt index 3190c9ff2d..f8c3ab73a9 100644 --- a/src/apps/mobile/android/app/src/main/kotlin/com/openbitfun/mobile/app/ui/chat/ConversationView.kt +++ b/src/apps/mobile/android/app/src/main/kotlin/com/openbitfun/mobile/app/ui/chat/ConversationView.kt @@ -248,6 +248,17 @@ internal fun ConversationView( onIntent(RemoteSessionIntent.UpdateDraft(mergeComposerDraft(state.draft, text))) } + // Held once, not rebuilt per recomposition: a composition local is compared by + // identity, so a fresh lambda here would invalidate every reader of it — the whole + // block list — on every frame the conversation redraws. [onIntent] is read through + // a holder for the same reason: the store's callback may be a new lambda each + // composition, and following the worker on a card the reader opened is not worth a + // transcript rebuild. + val currentOnIntent by rememberUpdatedState(onIntent) + val attachSubagent: (String) -> Unit = remember { + { owner: String -> currentOnIntent(RemoteSessionIntent.EnsureSubagentAttached(owner)) } + } + androidx.compose.runtime.CompositionLocalProvider( com.openbitfun.mobile.app.ui.chat.tool.LocalPermissionMailbox provides state.permissionMailbox, com.openbitfun.mobile.app.ui.chat.tool.LocalPlanActions provides com.openbitfun.mobile.app.ui.chat.tool.PlanActions( @@ -255,6 +266,11 @@ internal fun ConversationView( enabled = !state.busy && activeTurn == null && phase == ConnectionPhase.CONNECTED, build = { plan -> onIntent(RemoteSessionIntent.BuildPlan(sessionId, plan.path, plan.name)) }, ), + // A delegation card's worker is followed on demand, so opening the card has + // to reach the store. Handed down as a local rather than threaded through + // every block renderer: only the subagent card ever asks. + com.openbitfun.mobile.app.ui.chat.message.LocalSubagentAttach provides attachSubagent, + com.openbitfun.mobile.app.ui.chat.message.LocalSubagentFailures provides state.subagentFailures, ) { androidx.compose.foundation.layout.BoxWithConstraints(modifier.fillMaxSize()) { val mailboxMaxHeight = maxHeight * 0.4f diff --git a/src/apps/mobile/android/app/src/main/kotlin/com/openbitfun/mobile/app/ui/chat/message/MessageBlockList.kt b/src/apps/mobile/android/app/src/main/kotlin/com/openbitfun/mobile/app/ui/chat/message/MessageBlockList.kt index c833398d88..7371bdbbf9 100644 --- a/src/apps/mobile/android/app/src/main/kotlin/com/openbitfun/mobile/app/ui/chat/message/MessageBlockList.kt +++ b/src/apps/mobile/android/app/src/main/kotlin/com/openbitfun/mobile/app/ui/chat/message/MessageBlockList.kt @@ -22,6 +22,7 @@ import androidx.compose.runtime.key import androidx.compose.runtime.saveable.rememberSaveable import androidx.compose.runtime.remember import androidx.compose.runtime.setValue +import androidx.compose.runtime.staticCompositionLocalOf import androidx.compose.ui.Alignment import androidx.compose.ui.Modifier import androidx.compose.ui.semantics.Role @@ -41,10 +42,33 @@ import com.openbitfun.mobile.app.ui.chat.tool.ToolDisclosure import com.openbitfun.mobile.app.ui.chat.tool.LocalToolDisclosure import com.openbitfun.mobile.core.feature.session.MessageBlock import com.openbitfun.mobile.core.feature.session.QuestionAnswer +import com.openbitfun.mobile.core.feature.session.RemoteSessionFailureReason import com.openbitfun.mobile.core.feature.workspace.RemoteFileDownloadUiState internal const val SUBAGENT_GROUP_TEST_TAG: String = "subagent-group" +/** + * Asks the session store to follow a delegation card's worker again. + * + * A spawned worker lives in its own session, and the store can hold only so many of + * those streams at once: past the cap the oldest running child gives up its slot, and + * a stream can also have failed. Opening the card is the reader's way of asking for + * its steps, so this is the one gesture that re-subscribes. Defaulted so the block + * list still draws outside a session (previews, tests). + */ +internal val LocalSubagentAttach = staticCompositionLocalOf<(String) -> Unit> { {} } + +/** + * Launch items whose worker's steps could not be read, by tool call id. + * + * The session stays readable when one child stream fails, so the failure is drawn on + * the card that owns it rather than replacing the conversation. The reason is carried + * because the two cases are not the same news: a stream that failed may be readable on + * the next attempt, while a desktop that has no on-demand streams never will be. + */ +internal val LocalSubagentFailures = + staticCompositionLocalOf> { emptyMap() } + /** What the app has to be handed to draw any block, gathered so nesting stays cheap. */ internal data class MessageBlockCallbacks( val enabled: Boolean, @@ -159,8 +183,24 @@ private fun MessageBlockView(block: MessageBlock, callbacks: MessageBlockCallbac private fun SubagentGroup(block: MessageBlock.Subagent, callbacks: MessageBlockCallbacks) { var expanded by rememberSaveable(block.id) { mutableStateOf(false) } val children = remember(block.children, block.running) { subagentChildren(block) } - val hasDetails = children.isNotEmpty() || block.text.isNotBlank() + // A card whose worker's steps are missing is still openable: opening it is how a + // reader asks for those steps again, and how the card says why they are missing. + // A worker that is still running and has published nothing yet is the same case — + // its branch may have lost its slot before its first record, and the card would + // otherwise be the one card that cannot be asked to follow it. + val stepsUnavailable = LocalSubagentFailures.current[block.ownerToolCallId] + val hasDetails = children.isNotEmpty() || block.text.isNotBlank() || stepsUnavailable != null || + (block.ownerToolCallId.isNotEmpty() && block.running) + val attach = LocalSubagentAttach.current val outline = MaterialTheme.colorScheme.outlineVariant + // Identity and lifecycle are read from the card's own data, never from a timer: + // the avatar is derived from the launch the host named, and the state is the + // status the transcript already carries. + val identity = remember(block.ownerToolCallId, block.id) { + subagentIdentity(subagentIdentityKey(block.ownerToolCallId, block.id)) + } + val liveness = remember(block.status) { subagentLiveness(block.status) } + val name = subagentDisplayName(block.title).ifEmpty { stringResource(R.string.chat_subagent) } Column( modifier = Modifier .fillMaxWidth() @@ -177,21 +217,20 @@ private fun SubagentGroup(block: MessageBlock.Subagent, callbacks: MessageBlockC value = expanded, role = Role.Button, enabled = hasDetails, - onValueChange = { expanded = it }, + onValueChange = { open -> + expanded = open + // First expansion is the reader asking to see this worker's + // work, so it re-attaches a branch that a failed or evicted + // stream left with nothing. + if (open) attach(block.ownerToolCallId) + }, ), horizontalArrangement = Arrangement.spacedBy(8.dp), verticalAlignment = Alignment.CenterVertically, ) { - if (hasDetails) androidx.compose.material3.Icon( - painter = androidx.compose.ui.res.painterResource( - if (expanded) R.drawable.ic_symbol_chevron_down else R.drawable.ic_symbol_chevron_right, - ), - contentDescription = null, - tint = MaterialTheme.colorScheme.onSurfaceVariant, - modifier = Modifier.width(18.dp), - ) + SubagentAvatar(identity, Modifier) Text( - block.title.ifEmpty { stringResource(R.string.chat_subagent) }, + name, fontSize = 13.sp, fontWeight = FontWeight.Medium, color = MaterialTheme.colorScheme.onSurface, @@ -199,12 +238,20 @@ private fun SubagentGroup(block: MessageBlock.Subagent, callbacks: MessageBlockC maxLines = 1, overflow = androidx.compose.ui.text.style.TextOverflow.Ellipsis, ) - Text(stringResource(when { - subagentFailed(block.status) -> R.string.tool_phase_failed - block.running -> R.string.tool_phase_running - else -> R.string.tool_phase_completed - }), style = MaterialTheme.typography.labelSmall, - color = if (subagentFailed(block.status)) MaterialTheme.colorScheme.error else MaterialTheme.colorScheme.onSurfaceVariant) + SubagentStateChip(liveness) + // The card header is the only control that opens the worker's steps, so + // it is the one that has to say what tapping it does. A card with nothing + // to open draws no chevron, and so has nothing to announce. + if (hasDetails) androidx.compose.material3.Icon( + painter = androidx.compose.ui.res.painterResource( + if (expanded) R.drawable.ic_symbol_chevron_down else R.drawable.ic_symbol_chevron_right, + ), + contentDescription = stringResource( + if (expanded) R.string.subagent_collapse_steps else R.string.subagent_expand_steps, + ), + tint = MaterialTheme.colorScheme.onSurfaceVariant, + modifier = Modifier.width(16.dp), + ) } if (expanded && hasDetails) { Column( @@ -221,6 +268,21 @@ private fun SubagentGroup(block: MessageBlock.Subagent, callbacks: MessageBlockC .padding(start = 12.dp), verticalArrangement = Arrangement.spacedBy(6.dp), ) { + if (children.isEmpty() && stepsUnavailable != null) { + // A desktop without on-demand streams is not going to answer a + // second time, so its copy says so instead of inviting a retry. + Text( + stringResource( + if (stepsUnavailable == RemoteSessionFailureReason.HOST_STREAM_UNSUPPORTED) { + R.string.subagent_steps_unsupported + } else { + R.string.subagent_steps_unavailable + }, + ), + style = MaterialTheme.typography.bodySmall, + color = MaterialTheme.colorScheme.onSurfaceVariant, + ) + } if (children.isEmpty() && block.text.isNotBlank()) { Text( subagentPreview(block.text), diff --git a/src/apps/mobile/android/app/src/main/kotlin/com/openbitfun/mobile/app/ui/chat/message/ProcessPresentation.kt b/src/apps/mobile/android/app/src/main/kotlin/com/openbitfun/mobile/app/ui/chat/message/ProcessPresentation.kt index 278540ed47..545ac2d86b 100644 --- a/src/apps/mobile/android/app/src/main/kotlin/com/openbitfun/mobile/app/ui/chat/message/ProcessPresentation.kt +++ b/src/apps/mobile/android/app/src/main/kotlin/com/openbitfun/mobile/app/ui/chat/message/ProcessPresentation.kt @@ -47,9 +47,6 @@ internal fun subagentChildren(block: MessageBlock.Subagent): List } } -internal fun subagentFailed(status: String): Boolean = status.lowercase() in - setOf("failed", "error", "timeout", "cancelled", "canceled", "rejected") - internal fun subagentPreview(text: String): String { val trimmed = text.trim() val count = trimmed.codePointCount(0, trimmed.length) diff --git a/src/apps/mobile/android/app/src/main/kotlin/com/openbitfun/mobile/app/ui/chat/message/SubagentAvatar.kt b/src/apps/mobile/android/app/src/main/kotlin/com/openbitfun/mobile/app/ui/chat/message/SubagentAvatar.kt new file mode 100644 index 0000000000..e2946e2518 --- /dev/null +++ b/src/apps/mobile/android/app/src/main/kotlin/com/openbitfun/mobile/app/ui/chat/message/SubagentAvatar.kt @@ -0,0 +1,267 @@ +package com.openbitfun.mobile.app.ui.chat.message + +import android.provider.Settings +import androidx.compose.animation.core.RepeatMode +import androidx.compose.animation.core.animateFloat +import androidx.compose.animation.core.infiniteRepeatable +import androidx.compose.animation.core.rememberInfiniteTransition +import androidx.compose.animation.core.tween +import androidx.compose.foundation.background +import androidx.compose.foundation.border +import androidx.compose.foundation.layout.Arrangement +import androidx.compose.foundation.layout.Box +import androidx.compose.foundation.layout.Row +import androidx.compose.foundation.layout.size +import androidx.compose.foundation.shape.CircleShape +import androidx.compose.foundation.shape.RoundedCornerShape +import androidx.compose.material3.Icon +import androidx.compose.material3.MaterialTheme +import androidx.compose.material3.Text +import androidx.compose.runtime.Composable +import androidx.compose.runtime.getValue +import androidx.compose.runtime.remember +import androidx.compose.runtime.staticCompositionLocalOf +import androidx.compose.ui.Alignment +import androidx.compose.ui.Modifier +import androidx.compose.ui.graphics.Color +import androidx.compose.ui.graphics.graphicsLayer +import androidx.compose.ui.platform.LocalContext +import androidx.compose.ui.platform.testTag +import androidx.compose.ui.res.painterResource +import androidx.compose.ui.res.stringResource +import androidx.compose.ui.text.font.FontWeight +import androidx.compose.ui.unit.dp +import com.openbitfun.mobile.app.R +import com.openbitfun.mobile.app.ui.theme.OpenBitFunEaseInOut +import com.openbitfun.mobile.app.ui.theme.openBitFunColors + +internal const val SUBAGENT_AVATAR_TEST_TAG: String = "subagent-avatar" +internal const val SUBAGENT_STATE_TEST_TAG: String = "subagent-state" +internal const val SUBAGENT_STATE_MARK_TEST_TAG: String = "subagent-state-mark" +internal const val SUBAGENT_STATE_LABEL_TEST_TAG: String = "subagent-state-label" + +/** + * Forces the reduced-motion reading for a composition, for fixtures and tests. + * + * Null means "ask the platform", which is what every production surface does. + */ +internal val LocalReduceMotionOverride = staticCompositionLocalOf { null } + +/** + * Whether the reader asked the system to remove animations. + * + * The window recomposer already scales finite Compose animations by this setting, + * but the pulse below never ends, so a zero scale would leave it parked at + * whichever value it happened to reach rather than stopping it. Reading the scale + * once per composition is what turns the pulse off outright. A change to the + * setting is picked up the next time a card is composed, which is when it matters. + */ +@Composable +internal fun rememberReduceMotion(): Boolean { + val context = LocalContext.current + return remember(context) { + runCatching { + Settings.Global.getFloat(context.contentResolver, Settings.Global.ANIMATOR_DURATION_SCALE, 1f) == 0f + }.getOrDefault(false) + } +} + +/** + * A worker's identity, drawn: one glyph on one semantic tint. + * + * Ported in spirit from the desktop's `SubagentAvatar` — the same worker keeps the + * same face — without its authored artwork: a phone card is 24dp and this client + * has no brand SVGs of its own, so the identity is composed from existing symbol + * assets and existing colour roles. It is decorative: the card's title already + * names the worker, and a screen reader that read the avatar too would say it + * twice. + */ +@Composable +internal fun SubagentAvatar(identity: SubagentIdentity, modifier: Modifier = Modifier) { + val tint = subagentIdentityTint(identity.tint) + val shape = RoundedCornerShape(6.dp) + Box( + modifier = modifier + .size(24.dp) + .testTag(SUBAGENT_AVATAR_TEST_TAG) + // A tinted field rather than a solid one: the glyph has to stay legible + // in both themes, and the card behind it is the only thing composited. + .background(tint.copy(alpha = IDENTITY_FILL_ALPHA), shape) + .border(1.dp, tint.copy(alpha = IDENTITY_LINE_ALPHA), shape), + contentAlignment = Alignment.Center, + ) { + Icon( + painter = painterResource(SUBAGENT_GLYPHS[identity.glyph % SUBAGENT_GLYPHS.size]), + contentDescription = null, + tint = tint, + modifier = Modifier.size(15.dp), + ) + } +} + +/** + * The worker's lifecycle, drawn next to its name. + * + * The word carries the state and the mark reinforces it: the palette's own status + * greens and ambers are below the 3:1 a graphic needs against the light card, so + * nothing here is legible by colour alone. A worker that is still taking its turn + * pulses; one that is finishing keeps its pulse in a ring; one that is queued, + * done, cancelled, failed, or unknown holds still. The pulse is the only moving + * pixel on the card, and it is drawn inside its own 10dp box, so a frame captured + * a second later differs there and nowhere else. + */ +@Composable +internal fun SubagentStateChip(liveness: SubagentLiveness, modifier: Modifier = Modifier) { + val reduceMotion = LocalReduceMotionOverride.current ?: rememberReduceMotion() + val color = subagentStateColor(liveness) + val pulse = liveness.inFlight && !reduceMotion + Row( + modifier = modifier.testTag(SUBAGENT_STATE_TEST_TAG), + horizontalArrangement = Arrangement.spacedBy(4.dp), + verticalAlignment = Alignment.CenterVertically, + ) { + // One fixed box for every reading, so the six marks are the same size and + // can be compared, captured, and swapped without moving the row. + Box( + modifier = Modifier.size(MARK_SIZE).testTag(SUBAGENT_STATE_MARK_TEST_TAG), + contentAlignment = Alignment.Center, + ) { + when (liveness) { + // A shape, not a dot: a terminal state is not a heartbeat. + SubagentLiveness.COMPLETED -> StateGlyph(R.drawable.ic_symbol_checkmark_circle_fill, color) + SubagentLiveness.ERROR -> StateGlyph(R.drawable.ic_symbol_xmark, color) + SubagentLiveness.UNKNOWN -> StateGlyph(R.drawable.ic_symbol_questionmark_circle, color) + else -> StateDot( + color = color, + ring = liveness == SubagentLiveness.FINISHING || liveness == SubagentLiveness.CANCELLED, + pulsing = pulse, + ) + } + } + Text( + stringResource(subagentStateLabel(liveness)), + style = MaterialTheme.typography.labelSmall, + fontWeight = if (liveness.inFlight) FontWeight.Medium else FontWeight.Normal, + color = subagentStateLabelColor(liveness), + modifier = Modifier.testTag(SUBAGENT_STATE_LABEL_TEST_TAG), + ) + } +} + +@Composable +private fun StateGlyph(icon: Int, tint: Color) { + Icon( + painter = painterResource(icon), + contentDescription = null, + tint = tint, + modifier = Modifier.size(12.dp), + ) +} + +@Composable +private fun StateDot(color: Color, ring: Boolean, pulsing: Boolean) { + // The pulse runs from the composition's own clock, never from data: no state + // is re-read, no command is sent, and the value dies with the card. + val pulse = if (pulsing) { + val transition = rememberInfiniteTransition(label = "subagent-state") + val value by transition.animateFloat( + initialValue = 0f, + targetValue = 1f, + animationSpec = infiniteRepeatable( + animation = tween(PULSE_MS, easing = OpenBitFunEaseInOut), + repeatMode = RepeatMode.Reverse, + ), + label = "subagent-state-pulse", + ) + value + } else { + 1f + } + val mark = if (ring) { + Modifier.border(1.5.dp, color, CircleShape) + } else { + Modifier.background(color, CircleShape) + } + Box( + modifier = Modifier + .size(DOT_SIZE) + .graphicsLayer { + // Inside the dot's own bounds, so the pulse can never move a + // neighbour or push the row's layout around. + scaleX = PULSE_MIN_SCALE + (1f - PULSE_MIN_SCALE) * pulse + scaleY = scaleX + alpha = PULSE_MIN_ALPHA + (1f - PULSE_MIN_ALPHA) * pulse + } + .then(mark), + ) +} + +@Composable +private fun subagentIdentityTint(index: Int): Color = when (index % SUBAGENT_TINT_COUNT) { + 0 -> MaterialTheme.colorScheme.tertiary + 1 -> MaterialTheme.colorScheme.onSurface + else -> MaterialTheme.colorScheme.onSurfaceVariant +} + +@Composable +private fun subagentStateColor(liveness: SubagentLiveness): Color = when (liveness) { + SubagentLiveness.RUNNING, SubagentLiveness.FINISHING -> openBitFunColors.statusSuccess + SubagentLiveness.WAITING -> openBitFunColors.statusWarning + SubagentLiveness.COMPLETED -> MaterialTheme.colorScheme.tertiary + SubagentLiveness.CANCELLED -> MaterialTheme.colorScheme.onSurfaceVariant + SubagentLiveness.ERROR -> MaterialTheme.colorScheme.error + SubagentLiveness.UNKNOWN -> MaterialTheme.colorScheme.outline +} + +/** + * The word for a reading, in the reader's language. + * + * Reuses the phase words the rest of the surface already speaks with, so "Running" + * means the same thing on a card as on the tool row inside it. + */ +private fun subagentStateLabel(liveness: SubagentLiveness): Int = when (liveness) { + SubagentLiveness.RUNNING -> R.string.tool_phase_running + SubagentLiveness.FINISHING -> R.string.subagent_state_finishing + SubagentLiveness.WAITING -> R.string.tool_phase_waiting + SubagentLiveness.COMPLETED -> R.string.tool_phase_completed + SubagentLiveness.CANCELLED -> R.string.tool_phase_cancelled + SubagentLiveness.ERROR -> R.string.tool_phase_failed + SubagentLiveness.UNKNOWN -> R.string.subagent_state_unknown +} + +/** + * The label's colour. + * + * In-flight readings are inked rather than tinted: the palette's status colours do + * not reach the 4.5:1 small text needs on the light card, and a status a reader + * cannot read is worse than one they have to look at twice. + */ +@Composable +private fun subagentStateLabelColor(liveness: SubagentLiveness): Color = when (liveness) { + SubagentLiveness.RUNNING, SubagentLiveness.FINISHING -> MaterialTheme.colorScheme.onSurface + SubagentLiveness.ERROR -> MaterialTheme.colorScheme.error + else -> MaterialTheme.colorScheme.onSurfaceVariant +} + +/** + * The worker glyphs, in the order [subagentIdentity] indexes them. + * + * Existing symbol assets, not new artwork: a delegated worker is a robot, a + * person, code, a tool, or a stack of steps, and which one a card gets is derived + * from its own key rather than from what the worker happens to be doing. + */ +private val SUBAGENT_GLYPHS = intArrayOf( + R.drawable.ic_symbol_robot, + R.drawable.ic_symbol_person, + R.drawable.ic_symbol_code_square, + R.drawable.ic_symbol_wrench_and_screwdriver, + R.drawable.ic_symbol_rectangle_stack, +) + +private const val IDENTITY_FILL_ALPHA = 0.12f +private const val IDENTITY_LINE_ALPHA = 0.30f +private val MARK_SIZE = 12.dp +private val DOT_SIZE = 8.dp +private const val PULSE_MS = 900 +private const val PULSE_MIN_SCALE = 0.62f +private const val PULSE_MIN_ALPHA = 0.45f diff --git a/src/apps/mobile/android/app/src/main/kotlin/com/openbitfun/mobile/app/ui/chat/message/SubagentLiveness.kt b/src/apps/mobile/android/app/src/main/kotlin/com/openbitfun/mobile/app/ui/chat/message/SubagentLiveness.kt new file mode 100644 index 0000000000..63dc2ee77a --- /dev/null +++ b/src/apps/mobile/android/app/src/main/kotlin/com/openbitfun/mobile/app/ui/chat/message/SubagentLiveness.kt @@ -0,0 +1,132 @@ +package com.openbitfun.mobile.app.ui.chat.message + +/** + * What a delegation card says about the worker it launched. + * + * The six readings are the ones the desktop card draws, and each of them is a + * fact the host already published — this file only decides which word it is. It + * exists because the card used to fold every status into two: anything not in the + * running vocabulary was printed as "Done", so a worker that was calling a tool, + * waiting in a queue, or wrapping its turn up was reported as finished. + * + * [UNKNOWN] is not a seventh state of the design: it is what is drawn when the + * host sent a status this client cannot read. Guessing "completed" there is the + * one answer that is always wrong when it is wrong, so the card says it does not + * know instead. + */ +internal enum class SubagentLiveness { + RUNNING, + FINISHING, + WAITING, + COMPLETED, + CANCELLED, + ERROR, + UNKNOWN, + ; + + /** Whether the worker is still taking its turn, i.e. the card should move. */ + val inFlight: Boolean get() = this == RUNNING || this == FINISHING +} + +/** + * The reading for one host status. + * + * The vocabulary is the host's, lowercased: the child's own turn status when its + * stream could be read, and the launching item's status otherwise. Anything not + * listed here is [SubagentLiveness.UNKNOWN] on purpose — see the enum's note. + */ +internal fun subagentLiveness(status: String): SubagentLiveness = when (status.trim().lowercase()) { + // Started and not finished. `tool_calling` and `pending_confirmation` are the + // states a worker spends most of its life in, and the ones the card used to + // print as "Done". + "running", "streaming", "active", "inprogress", "in_progress", "preparing", "starting", + "thinking", "tool_calling", "pending_confirmation", "capability_required", + -> SubagentLiveness.RUNNING + + // Its last token is written and it is closing the turn out. + "finishing", "finalizing", "wrapping_up", + -> SubagentLiveness.FINISHING + + // Accepted but not started: the host is holding it, so nothing is running yet. + "queued", "pending", "waiting", "blocked", "deferred", "scheduled", + -> SubagentLiveness.WAITING + + "completed", "complete", "success", "succeeded", "done", "finished", + -> SubagentLiveness.COMPLETED + + "cancelled", "canceled", "aborted", "interrupted", "stopped", "skipped", + -> SubagentLiveness.CANCELLED + + "failed", "error", "timeout", "timed_out", "rejected", "crashed", "denied", + -> SubagentLiveness.ERROR + + else -> SubagentLiveness.UNKNOWN +} + +/** + * The one identity a card is drawn with: a glyph and a tint, both derived from a + * stable key. + * + * The desktop draws twenty authored robots picked by session id; a phone card is + * 24dp and the same reader has to tell two delegations apart at a glance, so this + * client uses the same idea with its own vocabulary: one of five worker glyphs and + * one of three semantic tints. Both are hash-derived, so a worker looks the same + * every time its card is drawn, and two launches of different workers almost never + * look alike. Collisions are allowed — this is identity, not a key. + */ +internal data class SubagentIdentity(val glyph: Int, val tint: Int) + +/** How many distinct glyphs [subagentIdentity] can pick from. */ +internal const val SUBAGENT_GLYPH_COUNT: Int = 5 + +/** How many semantic tints [subagentIdentity] can pick from. */ +internal const val SUBAGENT_TINT_COUNT: Int = 3 + +/** + * What a card is identified by. + * + * The launch item's tool call id is the host's own name for this delegation, and + * it is the key this client follows the child session by; a block the host + * published without one falls back to the position id the transcript gave it. + * Neither is invented from the worker's name or the tool that launched it. + */ +internal fun subagentIdentityKey(ownerToolCallId: String, blockId: String): String = + ownerToolCallId.trim().ifEmpty { blockId.trim() } + +/** + * The identity for [key], stable across redraws and across restarts. + * + * Two seeds are hashed rather than two slices of one hash, so the glyph and the + * tint vary independently instead of moving together. + */ +internal fun subagentIdentity(key: String): SubagentIdentity { + val seed = key.trim() + if (seed.isEmpty()) return SubagentIdentity(glyph = 0, tint = 0) + return SubagentIdentity( + glyph = (fnv1a("$SUBAGENT_IDENTITY_VERSION:glyph:$seed") ushr 1) % SUBAGENT_GLYPH_COUNT, + tint = (fnv1a("$SUBAGENT_IDENTITY_VERSION:tint:$seed") ushr 1) % SUBAGENT_TINT_COUNT, + ) +} + +/** + * The worker's name: the delegation's own words when it has them. + * + * A worker's name is the brief it was spawned with, which the shared presentation + * already extracted from the launch item; this client never substitutes a tool + * name for it. Blank means the host gave none, and the card falls back to its own + * word for "subagent". + */ +internal fun subagentDisplayName(title: String): String = title.trim() + +/** Versioned so a future change to the pools does not silently repaint old cards. */ +private const val SUBAGENT_IDENTITY_VERSION = "subagent-identity-v1" + +/** FNV-1a, 32-bit: the whole point is that it is identical on every run and device. */ +private fun fnv1a(text: String): Int { + var hash = 0x811C9DC5.toInt() + text.forEach { char -> + hash = hash xor char.code + hash *= 0x01000193 + } + return hash +} diff --git a/src/apps/mobile/android/app/src/main/kotlin/com/openbitfun/mobile/app/ui/theme/Theme.kt b/src/apps/mobile/android/app/src/main/kotlin/com/openbitfun/mobile/app/ui/theme/Theme.kt index 678d089ad3..2e144d8fe8 100644 --- a/src/apps/mobile/android/app/src/main/kotlin/com/openbitfun/mobile/app/ui/theme/Theme.kt +++ b/src/apps/mobile/android/app/src/main/kotlin/com/openbitfun/mobile/app/ui/theme/Theme.kt @@ -124,10 +124,13 @@ private val DarkScheme = darkColorScheme( * * [success] is the connection dot's "connected" green — it is not `primary`, * because primary here is near-black ink and a black dot reads as "off". + * [statusWarning] is the same family's amber, used where a state is "held" rather + * than working or finished (a queued delegation). */ internal data class OpenBitFunColors( val transparent: Color, val statusSuccess: Color, + val statusWarning: Color, val shellScrim: Color, val mediaBackground: Color, val mediaScrim: Color, @@ -192,6 +195,7 @@ internal data class CodeSyntaxColors( private val LightExtras = OpenBitFunColors( transparent = LightTokens.Transparent, statusSuccess = LightTokens.StatusSuccess, + statusWarning = LightTokens.StatusWarning, shellScrim = LightTokens.ShellScrim, mediaBackground = LightTokens.MediaBackground, mediaScrim = LightTokens.MediaScrim, @@ -232,6 +236,7 @@ private val LightExtras = OpenBitFunColors( private val DarkExtras = OpenBitFunColors( transparent = DarkTokens.Transparent, statusSuccess = DarkTokens.StatusSuccess, + statusWarning = DarkTokens.StatusWarning, shellScrim = DarkTokens.ShellScrim, mediaBackground = DarkTokens.MediaBackground, mediaScrim = DarkTokens.MediaScrim, diff --git a/src/apps/mobile/android/app/src/main/kotlin/com/openbitfun/mobile/app/viewmodel/AccountViewModel.kt b/src/apps/mobile/android/app/src/main/kotlin/com/openbitfun/mobile/app/viewmodel/AccountViewModel.kt index 230db1802d..ad6d8a9dd4 100644 --- a/src/apps/mobile/android/app/src/main/kotlin/com/openbitfun/mobile/app/viewmodel/AccountViewModel.kt +++ b/src/apps/mobile/android/app/src/main/kotlin/com/openbitfun/mobile/app/viewmodel/AccountViewModel.kt @@ -153,7 +153,10 @@ internal class AccountViewModel(application: Application) : AndroidViewModel(app activeTarget = target if (target == null) return - remoteStore = store.createSessionStore(viewModelScope)?.also { created -> + // Android draws the delegation card: opening one asks for its worker to be + // followed again, and a worker that cannot be read is reported on that card. + // A platform that has not wired both stays on the plain transcript. + remoteStore = store.createSessionStore(viewModelScope, followSubagentBranches = true)?.also { created -> remoteJob = viewModelScope.launch { created.state.collect { _remoteState.value = it completionNotifier.observe(it, target) diff --git a/src/apps/mobile/android/app/src/main/res/values-zh/strings_subagent.xml b/src/apps/mobile/android/app/src/main/res/values-zh/strings_subagent.xml new file mode 100644 index 0000000000..b299024fde --- /dev/null +++ b/src/apps/mobile/android/app/src/main/res/values-zh/strings_subagent.xml @@ -0,0 +1,15 @@ + + + + 无法读取该子代理的步骤 + 该桌面端不支持共享子代理步骤 + 收尾中 + 状态未知 + 展开子代理步骤 + 收起子代理步骤 + diff --git a/src/apps/mobile/android/app/src/main/res/values/strings_subagent.xml b/src/apps/mobile/android/app/src/main/res/values/strings_subagent.xml new file mode 100644 index 0000000000..060c5aaa9b --- /dev/null +++ b/src/apps/mobile/android/app/src/main/res/values/strings_subagent.xml @@ -0,0 +1,25 @@ + + + + + Worker steps unavailable + + This desktop cannot share worker steps + + Finishing + + Status unknown + + Show worker steps + Hide worker steps + diff --git a/src/apps/mobile/android/app/src/test/kotlin/com/openbitfun/mobile/app/ui/chat/ProcessPresentationTest.kt b/src/apps/mobile/android/app/src/test/kotlin/com/openbitfun/mobile/app/ui/chat/ProcessPresentationTest.kt index 3f08a1ccff..b24ed22d67 100644 --- a/src/apps/mobile/android/app/src/test/kotlin/com/openbitfun/mobile/app/ui/chat/ProcessPresentationTest.kt +++ b/src/apps/mobile/android/app/src/test/kotlin/com/openbitfun/mobile/app/ui/chat/ProcessPresentationTest.kt @@ -20,10 +20,6 @@ class ProcessPresentationTest { assertTrue(preview.endsWith("…")) assertEquals("answer", subagentPreview(" answer ")) } - @Test fun statusPreservesFailuresInsteadOfCallingThemDone() { - listOf("failed", "ERROR", "timeout", "cancelled", "canceled", "rejected").forEach { assertTrue(subagentFailed(it)) } - assertFalse(subagentFailed("completed")) - } private fun tool(id: String, phase: ToolPhase = ToolPhase.COMPLETED) = ToolCard( id, "Read", phase, ToolKind.DOCUMENT, ToolOperation.READ_FILE, "file.kt", "", "", "", "", null, emptySet(), diff --git a/src/apps/mobile/android/app/src/test/kotlin/com/openbitfun/mobile/app/ui/chat/SubagentLivenessTest.kt b/src/apps/mobile/android/app/src/test/kotlin/com/openbitfun/mobile/app/ui/chat/SubagentLivenessTest.kt new file mode 100644 index 0000000000..d175ba7205 --- /dev/null +++ b/src/apps/mobile/android/app/src/test/kotlin/com/openbitfun/mobile/app/ui/chat/SubagentLivenessTest.kt @@ -0,0 +1,108 @@ +package com.openbitfun.mobile.app.ui.chat + +import com.openbitfun.mobile.app.ui.chat.message.* +import org.junit.Assert.* +import org.junit.Test + +/** + * What a delegation card is allowed to say about its worker. + * + * The card used to fold the host's vocabulary into two readings, and everything it + * did not recognise became "Done". These are the facts that replaced that: which + * host word means which of the six readings, that an unreadable status is not + * guessed at, and that the identity a worker is drawn with is stable and derived + * from the delegation itself. + */ +class SubagentLivenessTest { + @Test fun aWorkerThatIsStillTakingItsTurnIsNeverReportedAsDone() { + val inFlight = listOf( + "running", "streaming", "active", "inprogress", "in_progress", "preparing", + "starting", "thinking", "tool_calling", "pending_confirmation", "capability_required", + ) + inFlight.forEach { assertEquals("status=$it", SubagentLiveness.RUNNING, subagentLiveness(it)) } + // The host's own case and padding are not a different state. + assertEquals(SubagentLiveness.RUNNING, subagentLiveness(" Tool_Calling ")) + assertTrue(SubagentLiveness.RUNNING.inFlight) + } + + @Test fun aTurnThatIsWrappingUpIsItsOwnReadingAndStillInFlight() { + listOf("finishing", "finalizing", "wrapping_up").forEach { + assertEquals("status=$it", SubagentLiveness.FINISHING, subagentLiveness(it)) + } + assertTrue(SubagentLiveness.FINISHING.inFlight) + } + + @Test fun aQueuedWorkerIsWaitingRatherThanRunningOrDone() { + listOf("queued", "pending", "waiting", "blocked", "deferred", "scheduled").forEach { + assertEquals("status=$it", SubagentLiveness.WAITING, subagentLiveness(it)) + } + // Accepted but not started: nothing is running, so nothing may pulse. + assertFalse(SubagentLiveness.WAITING.inFlight) + } + + @Test fun terminalOutcomesKeepTheirOwnReading() { + listOf("completed", "complete", "success", "succeeded", "done", "finished").forEach { + assertEquals("status=$it", SubagentLiveness.COMPLETED, subagentLiveness(it)) + } + listOf("cancelled", "canceled", "aborted", "interrupted", "stopped", "skipped").forEach { + assertEquals("status=$it", SubagentLiveness.CANCELLED, subagentLiveness(it)) + } + listOf("failed", "error", "timeout", "timed_out", "rejected", "crashed", "denied").forEach { + assertEquals("status=$it", SubagentLiveness.ERROR, subagentLiveness(it)) + } + listOf(SubagentLiveness.COMPLETED, SubagentLiveness.CANCELLED, SubagentLiveness.ERROR).forEach { + assertFalse(it.inFlight) + } + } + + @Test fun aStatusThisClientCannotReadIsSaidSoInsteadOfGuessedAsDone() { + // The one answer that is always wrong when it is wrong is the invented one. + listOf("", " ", "paused-by-policy", "reticulating", "in progress").forEach { + assertEquals("status=$it", SubagentLiveness.UNKNOWN, subagentLiveness(it)) + } + assertFalse(SubagentLiveness.UNKNOWN.inFlight) + } + + @Test fun aWorkerIsIdentifiedByItsOwnDelegationAndNotByItsLabel() { + // The launch item's tool call is the host's name for this delegation; a block + // the host published without one falls back to its position. + assertEquals("tool-abc", subagentIdentityKey(" tool-abc ", "message-1-subagent-0")) + assertEquals("message-1-subagent-0", subagentIdentityKey("", "message-1-subagent-0")) + assertEquals("message-1-subagent-0", subagentIdentityKey(" ", " message-1-subagent-0 ")) + } + + @Test fun theSameDelegationIsAlwaysDrawnTheSameWay() { + val first = subagentIdentity("tool-abc") + repeat(5) { assertEquals(first, subagentIdentity("tool-abc")) } + // Padding is not a different worker. + assertEquals(first, subagentIdentity(" tool-abc ")) + // Pinned: the derivation is a pure function of the key, so a worker keeps its + // face across redraws, restarts and devices. Changing this value repaints + // every card that already exists, which is why it is asserted here. + assertEquals(SubagentIdentity(glyph = 3, tint = 2), subagentIdentity("tool-owner")) + assertEquals(SubagentIdentity(glyph = 0, tint = 0), subagentIdentity("")) + // Pools are bounded, so a card can never index outside them. + listOf("tool-abc", "tool-def", "tool-1", "message-9-subagent-3", "session-42").forEach { key -> + val identity = subagentIdentity(key) + assertTrue(identity.glyph in 0 until SUBAGENT_GLYPH_COUNT) + assertTrue(identity.tint in 0 until SUBAGENT_TINT_COUNT) + } + } + + @Test fun twoDelegationsOfTheSameTitleDoNotShareOneFace() { + // The reported symptom: two spawn cards with the same words and no way to + // tell them apart. Collisions are allowed by design, but the launches a + // screen actually shows have to differ. + val keys = listOf("tool-owner", "tool-owner-2", "tool-owner-3", "tool-owner-4") + val faces = keys.map(::subagentIdentity) + assertEquals(faces.size, faces.distinct().size) + assertNotEquals(subagentIdentity("tool-owner"), subagentIdentity("tool-owner-2")) + } + + @Test fun aWorkersNameIsItsOwnWordsOrNothing() { + assertEquals("Audit the parser", subagentDisplayName(" Audit the parser \n")) + assertEquals("", subagentDisplayName(" ")) + // Never the tool that launched it, and never a name this client invented. + assertEquals("agentspawn", subagentDisplayName("agentspawn")) + } +} diff --git a/src/apps/mobile/shared/core-feature/src/commonMain/kotlin/com/openbitfun/mobile/core/feature/account/AccountStore.kt b/src/apps/mobile/shared/core-feature/src/commonMain/kotlin/com/openbitfun/mobile/core/feature/account/AccountStore.kt index c23845e7de..dfe491387d 100644 --- a/src/apps/mobile/shared/core-feature/src/commonMain/kotlin/com/openbitfun/mobile/core/feature/account/AccountStore.kt +++ b/src/apps/mobile/shared/core-feature/src/commonMain/kotlin/com/openbitfun/mobile/core/feature/account/AccountStore.kt @@ -165,6 +165,29 @@ public class AccountStore internal constructor( ).also { it.bindCatalog(catalogChanges(current, target, transport)) } } + /** + * The same store, told whether the platform wants delegated children followed. + * + * A delegation tool's child keeps its work in its own session, and drawing that + * work inside the card that launched it is a behaviour a surface has to be able to + * finish: ask for a re-attach when a branch lost its slot, and show why a worker + * could not be read. Callers that can do neither leave it off — the overload above + * does, and so does every client that has not wired the card — and then the store + * publishes the host's own transcript, item for item, exactly as before. + */ + public fun createSessionStore(scope: CoroutineScope, followSubagentBranches: Boolean): RemoteSessionStore? { + val current = session ?: return null + val target = current.targetDeviceId?.let(::authorizedDeviceId) ?: return null + val transport = backend.transport(current, target) + return RemoteSessionStore.create( + scope, + transport, + deviceKey = target, + persistence = persistence, + followSubagentBranches = followSubagentBranches, + ).also { it.bindCatalog(catalogChanges(current, target, transport)) } + } + public fun createWorkspaceStore(scope: CoroutineScope): RemoteWorkspaceStore? { val current = session ?: return null val target = current.targetDeviceId?.let(::authorizedDeviceId) ?: return null diff --git a/src/apps/mobile/shared/core-feature/src/commonMain/kotlin/com/openbitfun/mobile/core/feature/session/MessageBlockPresentation.kt b/src/apps/mobile/shared/core-feature/src/commonMain/kotlin/com/openbitfun/mobile/core/feature/session/MessageBlockPresentation.kt index 32cda2ecac..ef25583ef7 100644 --- a/src/apps/mobile/shared/core-feature/src/commonMain/kotlin/com/openbitfun/mobile/core/feature/session/MessageBlockPresentation.kt +++ b/src/apps/mobile/shared/core-feature/src/commonMain/kotlin/com/openbitfun/mobile/core/feature/session/MessageBlockPresentation.kt @@ -54,10 +54,28 @@ public sealed interface MessageBlock { public val children: List, /** Preserve host status; running alone loses failure and queued states. */ public val status: String, + /** + * The tool call that launched this worker, when the item named one. + * + * [id] is this block's position in the transcript, which is what a lazy list + * keys on; the launch item's own identity is what the session store follows a + * spawned child by, so a card that wants its worker's steps read again names + * this. Empty for a branch the host published inside its own stream. + */ + public val ownerToolCallId: String, ) : MessageBlock { // Keep source compatibility without default arguments on the exported feature API. public constructor(id: String, title: String, running: Boolean, text: String, children: List) : this(id, title, running, text, children, if (running) "running" else "completed") + + public constructor( + id: String, + title: String, + running: Boolean, + text: String, + children: List, + status: String, + ) : this(id, title, running, text, children, status, "") } } @@ -190,6 +208,7 @@ private fun walk(items: List, path: String, streaming: text = subagentBody(entry), children = walk(entry.subItems.orEmpty().map { it.copy(isSubagent = false) }, id, running), status = status, + ownerToolCallId = entry.tool?.id.orEmpty(), ) return@forEachIndexed } @@ -323,7 +342,15 @@ private fun plainSubagentLabel(raw: String): String { return text } -private val SUBAGENT_RUNNING = setOf("running", "active", "preparing", "pending", "queued") +/** + * Statuses that mean the worker still owes work, whatever word the host used. + * + * This is the block's `running` flag, not the card's reading: it says the worker has + * work coming, so its scope is still open and its card can still be asked for its + * steps. A queued worker is in here for that reason — it is not working yet, and the + * card says so from its own status word, but it has not finished either. + */ +private val SUBAGENT_RUNNING = setOf("running", "active", "preparing", "pending", "queued", "waiting") private const val TITLE_LIMIT = 80 private val TEXT_TYPES = setOf("text", "message", "") diff --git a/src/apps/mobile/shared/core-feature/src/commonMain/kotlin/com/openbitfun/mobile/core/feature/session/RemoteSessionStore.kt b/src/apps/mobile/shared/core-feature/src/commonMain/kotlin/com/openbitfun/mobile/core/feature/session/RemoteSessionStore.kt index aa2f1ecd89..853dbc8513 100644 --- a/src/apps/mobile/shared/core-feature/src/commonMain/kotlin/com/openbitfun/mobile/core/feature/session/RemoteSessionStore.kt +++ b/src/apps/mobile/shared/core-feature/src/commonMain/kotlin/com/openbitfun/mobile/core/feature/session/RemoteSessionStore.kt @@ -89,6 +89,19 @@ public class RemoteSessionStore internal constructor( private val transport: RemoteCommandTransport, private val deviceKey: String? = null, private val persistence: MobilePersistenceStores? = null, + /** + * Whether this store follows the child sessions a delegation tool names. + * + * Following a child means opening a second session stream and drawing its records + * inside the card that launched it, and that behaviour has two halves a surface + * has to be able to finish: the card has to be able to ask for a re-attach, and it + * has to be able to show why a worker could not be read. A client missing either + * half would hold a released branch's stale prefix with no way back, and record + * failures nobody can see, so the follower is opt-in rather than assumed. With the + * flag off no child stream is opened at all and the published transcript is the + * host's own, item for item — which is what every platform but Android ships. + */ + private val followSubagentBranches: Boolean = false, ) { private var catalogSubscription: Job? = null private var catalogRefresh: Job? = null @@ -228,6 +241,7 @@ public class RemoteSessionStore internal constructor( is RemoteSessionIntent.RespondPermission -> permissionMailbox.respond(intent.requestId, intent.approve, intent.updatedInput) RemoteSessionIntent.RefreshPermissionMailbox -> permissionMailbox.invalidate() is RemoteSessionIntent.SetForeground -> setForeground(intent.active) + is RemoteSessionIntent.EnsureSubagentAttached -> ensureSubagentAttached(intent.ownerToolCallId) RemoteSessionIntent.Load, RemoteSessionIntent.Refresh -> load(current?.query.orEmpty(), current?.agentFilter ?: SessionAgentFilter.ALL) RemoteSessionIntent.LoadMore -> loadMore() @@ -1057,6 +1071,15 @@ public class RemoteSessionStore internal constructor( private var sessionHistoryHasMore = false private var sessionRecords: SessionRecordReplica? = null + /** + * Streams the open session's spawned children into the cards that launched them. + * + * Null whenever no session is open: a child stream belongs to the session on + * screen, and one that outlived its parent would keep publishing records into a + * transcript nobody is reading. + */ + private var subagents: SubagentBranchCoordinator? = null + private fun currentPromptSuggestion(sessionId: String?): String? = sessionRecords?.takeIf { it.sessionId == sessionId }?.promptSuggestion() private var transcriptWrite: Job? = null @@ -1115,15 +1138,78 @@ public class RemoteSessionStore internal constructor( markConnected() } + /** + * Asks for one delegation card's worker to be followed again. + * + * What a reader opening a card asks for, and the only way back for a branch the + * concurrency cap released: past the cap a still-running child gives up its slot + * to the newest launch, and without this the card would keep the prefix it + * published before the slot was taken. It reads nothing itself — the child's + * stream is the coordinator's — and it cannot fail the session on screen. + */ + public fun ensureSubagentAttached(ownerToolCallId: String) { + if (ownerToolCallId.isEmpty()) return + val retried = subagents?.ensureAttached(ownerToolCallId) == true + // What was reported about this card describes a stream that is being read + // again, so the report is dropped — but only when a read really happened: a + // host that cannot serve child streams keeps saying so instead of silently + // losing the only explanation the card had. + if (retried) clearSubagentFailure(ownerToolCallId) + } + + /** + * Records that one delegation card's child session could not be read. + * + * Deliberately not [handleFailure]: a child stream failing is not this + * conversation failing. The parent transcript is the host's answer for the + * session the reader opened and stays on screen, the launching item keeps its own + * status, and nothing about the connection phase or the composer changes because + * one worker's stream is unavailable. + */ + private fun reportSubagentFailure(ownerToolCallId: String, error: Throwable) { + val current = _state.value as? RemoteSessionUiState.Ready ?: return + if (ownerToolCallId.isEmpty()) return + val reason = remoteSessionFailure(error).reason + if (current.subagentFailures[ownerToolCallId] == reason) return + _state.value = current.copy(subagentFailures = current.subagentFailures + (ownerToolCallId to reason)) + } + + private fun clearSubagentFailure(ownerToolCallId: String) { + val current = _state.value as? RemoteSessionUiState.Ready ?: return + if (ownerToolCallId !in current.subagentFailures) return + _state.value = current.copy(subagentFailures = current.subagentFailures - ownerToolCallId) + } + + /** A card that is showing its worker's steps has nothing left to report. */ + private fun reconcileSubagentFailures(messages: List) { + val current = _state.value as? RemoteSessionUiState.Ready ?: return + if (current.subagentFailures.isEmpty()) return + val attached = messages.flatMap { it.items.orEmpty() } + .filter { !it.subItems.isNullOrEmpty() } + .mapNotNull { it.tool?.id } + .toSet() + if (attached.isEmpty()) return + val remaining = current.subagentFailures - attached + if (remaining.size != current.subagentFailures.size) { + _state.value = current.copy(subagentFailures = remaining) + } + } + private fun subscribeSessionUpdates(sessionId: String): CompletableDeferred { val initialHistory = CompletableDeferred() permissionMailbox.select(sessionId) messageQueue.select(sessionId, "dialog_queue_v1" in hostCapabilities) + // Switching sessions closes the previous session's child streams here, not in + // the cancelled job: cancellation is cooperative, and a child stream that + // outlived its parent even briefly would publish into the new session. + subagents?.endSession() + subagents = null sessionUpdates?.cancel() transcriptWrite?.cancel() forgetWrittenTranscript() sessionHistoryHasMore = false sessionUpdates = scope.launch { + var branches: SubagentBranchCoordinator? = null try { val source = transport as? RemoteSessionStreamTransport ?: error("Durable session transport unavailable") // A host that has not advertised `host_stream_v1` cannot serve @@ -1146,7 +1232,15 @@ public class RemoteSessionStore internal constructor( */ fun render(): ChatSyncPhase { timelineStore.clearActiveTurn() - val messages = records.messages() + // A delegation tool's spawned child keeps its work in its own + // session, so the transcript that is published is the host's own + // plus whatever each launched child has said so far, attached to + // the item that launched it. + val messages = subagents?.decorate(records.messages()) ?: records.messages() + // A card that is drawing its worker's steps has nothing left to + // report, so a failure recorded before the branch was read again + // stops being shown. + reconcileSubagentFailures(messages) val active = messages.lastOrNull()?.takeIf { it.role == "assistant" && it.status == "streaming" } timelineStore.setPersistedMessages(if (active == null) messages else messages.dropLast(1)) timelineStore.setActiveTurn(active) @@ -1162,6 +1256,29 @@ public class RemoteSessionStore internal constructor( timelineStore.setSyncPhase(phase) return phase } + /** + * Follows the children this session's own transcript names, where the + * platform opted in with [followSubagentBranches]. + * + * Created here because a child stream belongs to the session on + * screen: it is bound to `sessionId` before the parent stream is + * subscribed, and closed in this job's `finally`. Child records + * reach the screen through the same publish the parent's own + * records use, so the UI needs no new state to draw them. + */ + val coordinator = if (followSubagentBranches) SubagentBranchCoordinator( + scope = scope, + source = source, + publish = { + if (caughtUp && !replayingHistory) { + render() + publishDurableTimeline() + } + }, + onError = { ownerToolCallId, error -> reportSubagentFailure(ownerToolCallId, error) }, + ).also { it.beginSession(sessionId) } else null + branches = coordinator + subagents = coordinator source.subscribe(sessionId, { handleFailure(it, _state.value as? RemoteSessionUiState.Ready) }, { @@ -1245,6 +1362,12 @@ public class RemoteSessionStore internal constructor( initialHistory.completeExceptionally(error) handleFailure(error, _state.value as? RemoteSessionUiState.Ready) } finally { + // The parent stream ending ends every child stream with it, so nothing + // keeps pushing records into a transcript that is no longer read. The + // identity guard keeps a job cancelled during a session switch from + // closing the session that replaced it. + branches?.endSession() + if (subagents === branches) subagents = null initialHistory.cancel() } } @@ -2245,6 +2368,9 @@ public class RemoteSessionStore internal constructor( * Keep foreground host probes for idle open conversations as well as lists. */ private fun setForeground(active: Boolean) { messageQueue.setForeground(active) + // Child streams follow the app's visibility too: a worker's steps are worth + // nothing to a backgrounded phone, and its card keeps what it published. + subagents?.setForeground(active) if (active && healthWork?.isActive == true) return val generation = ++healthGeneration healthWork?.cancel() @@ -2292,15 +2418,21 @@ public class RemoteSessionStore internal constructor( private const val FILTER_PAGE_SIZE: Int = 100 private const val DIRECTORY_WORKSPACE_PAGE_SIZE: Int = 50 - internal fun create(scope: CoroutineScope, transport: RemoteCommandTransport): RemoteSessionStore = - RemoteSessionStore(scope, transport) + internal fun create( + scope: CoroutineScope, + transport: RemoteCommandTransport, + followSubagentBranches: Boolean = false, + ): RemoteSessionStore = + RemoteSessionStore(scope, transport, followSubagentBranches = followSubagentBranches) internal fun create( scope: CoroutineScope, transport: RemoteCommandTransport, deviceKey: String?, persistence: MobilePersistenceStores?, - ): RemoteSessionStore = RemoteSessionStore(scope, transport, deviceKey, persistence) + followSubagentBranches: Boolean = false, + ): RemoteSessionStore = + RemoteSessionStore(scope, transport, deviceKey, persistence, followSubagentBranches) } @@ -2327,12 +2459,24 @@ private fun toPersisted(sessionId: String, m: ChatMessage): PersistedRemoteMessa turnId = m.turnId, detail = m.detail, tools = m.tools, - items = m.items, + items = m.items?.map(::withoutChildSteps), images = m.images, error = m.error, )), ) +/** + * One item of the host's transcript, without a child session's steps. + * + * A spawned child's work is drawn from that child's own stream on demand, so what + * is stored here is the launching item and nothing else: the steps were never the + * parent's transcript, and a cached copy would be a snapshot of a worker that is + * still moving. The child session id stays on the tool, so re-opening the session + * from this device's cache still finds the branch and reads it again. + */ +internal fun withoutChildSteps(item: ChatMessageItemResponse): ChatMessageItemResponse = + if (item.tool?.subagentSessionId != null && item.subItems != null) item.copy(subItems = null) else item + private fun toChatMessage(m: PersistedRemoteMessage): ChatMessage { val payload = try { STORE_JSON.decodeFromString(m.payloadJson) diff --git a/src/apps/mobile/shared/core-feature/src/commonMain/kotlin/com/openbitfun/mobile/core/feature/session/RemoteSessionUiState.kt b/src/apps/mobile/shared/core-feature/src/commonMain/kotlin/com/openbitfun/mobile/core/feature/session/RemoteSessionUiState.kt index 82d9b5ae27..77c79909ec 100644 --- a/src/apps/mobile/shared/core-feature/src/commonMain/kotlin/com/openbitfun/mobile/core/feature/session/RemoteSessionUiState.kt +++ b/src/apps/mobile/shared/core-feature/src/commonMain/kotlin/com/openbitfun/mobile/core/feature/session/RemoteSessionUiState.kt @@ -297,7 +297,31 @@ public sealed interface RemoteSessionUiState { /** Host hint for the latest completed turn; never persisted with the transcript. */ public val promptSuggestion: String?, public val messageQueue: MessageQueueUiState, + /** + * Delegation cards whose child session could not be read, by owner tool call id. + * + * A spawned child lives in its own session, and that stream can fail on its + * own: a host that serves this conversation perfectly well may refuse one + * child's `read_stream`, or drop it, or never answer it. That says nothing + * about the conversation on screen, which is still the host's own transcript, + * so it is reported here — on a session that stays [Ready] — instead of + * replacing the readable conversation with a session-level failure. The card + * keeps the launching item's own status and can say that its steps are + * unavailable; the entry disappears when the branch is read again. + */ + public val subagentFailures: Map, ) : RemoteSessionUiState { + /** Preserve the existing Swift initializer when adding the per-card delegation failures. */ + public constructor( + sessions: List, selectedSessionId: String?, timeline: ChatTimelineState?, busy: Boolean, + permissionMode: SessionPermissionMode?, permissionModeFailure: PermissionModeFailure?, query: String, + agentFilter: SessionAgentFilter, hasMore: Boolean, hasMoreMessages: Boolean, + modelCatalog: RemoteModelCatalog?, modelCatalogFailure: ModelCatalogFailure?, draft: String, + revision: Long, lastSentMessage: SentChatMessage?, permissionMailbox: PermissionMailboxUiState, + historyLoadState: HistoryLoadState, promptSuggestion: String?, messageQueue: MessageQueueUiState, + ) : this(sessions, selectedSessionId, timeline, busy, permissionMode, permissionModeFailure, query, + agentFilter, hasMore, hasMoreMessages, modelCatalog, modelCatalogFailure, draft, revision, + lastSentMessage, permissionMailbox, historyLoadState, promptSuggestion, messageQueue, emptyMap()) public constructor( sessions: List, selectedSessionId: String?, timeline: ChatTimelineState?, busy: Boolean, permissionMode: SessionPermissionMode?, permissionModeFailure: PermissionModeFailure?, query: String, @@ -431,6 +455,15 @@ public sealed interface RemoteSessionIntent { /** Native lifecycle controls foreground connection health probes. */ public data class SetForeground(public val active: Boolean) : RemoteSessionIntent + /** + * A delegation card was opened and wants its worker's own steps. + * + * The card may have lost its stream to the concurrency cap, never had one, or been + * released while the app was in the background; opening it is the one gesture that + * asks for the child's stream again. It changes nothing else about the session. + */ + public data class EnsureSubagentAttached(public val ownerToolCallId: String) : RemoteSessionIntent + public data object Load : RemoteSessionIntent public data object Refresh : RemoteSessionIntent diff --git a/src/apps/mobile/shared/core-feature/src/commonMain/kotlin/com/openbitfun/mobile/core/feature/session/SessionRecordReplica.kt b/src/apps/mobile/shared/core-feature/src/commonMain/kotlin/com/openbitfun/mobile/core/feature/session/SessionRecordReplica.kt index ecee7252a2..6b5b21ca9c 100644 --- a/src/apps/mobile/shared/core-feature/src/commonMain/kotlin/com/openbitfun/mobile/core/feature/session/SessionRecordReplica.kt +++ b/src/apps/mobile/shared/core-feature/src/commonMain/kotlin/com/openbitfun/mobile/core/feature/session/SessionRecordReplica.kt @@ -221,7 +221,11 @@ internal class SessionRecordReplica(val sessionId: String) { val data = item.getValue("data").jsonObject val type = item.string("type") val result = data["toolResult"] as? JsonObject - ChatMessageItemResponse(type = type, content = data.string("content"), isSubagent = data["isSubagentItem"]?.jsonPrimitive?.booleanOrNull == true || data.string("subagentSessionId").isNotEmpty(), + // The child session a delegation tool spawned. It marks the item as a + // subagent branch and is also the stream id that branch is read from, + // so it is kept on the mapped tool instead of being read and dropped. + val childSessionId = data.string("subagentSessionId").takeIf { it.isNotEmpty() } + ChatMessageItemResponse(type = type, content = data.string("content"), isSubagent = data["isSubagentItem"]?.jsonPrimitive?.booleanOrNull == true || childSessionId != null, tool = if (type != "tool") null else RemoteToolStatusResponse( id = (data["toolCall"] as? JsonObject)?.string("id")?.takeIf { it.isNotEmpty() } ?: data.string("id"), name = data.string("toolName"), status = data.string("status").takeIf { it.isNotEmpty() } @@ -230,7 +234,8 @@ internal class SessionRecordReplica(val sessionId: String) { errorPreview = result?.string("error"), startMs = data["startTime"]?.jsonPrimitive?.longOrNull, durationMs = data["durationMs"]?.jsonPrimitive?.longOrNull - ?: result?.get("durationMs")?.jsonPrimitive?.longOrNull)) + ?: result?.get("durationMs")?.jsonPrimitive?.longOrNull, + subagentSessionId = childSessionId)) } val superseded = rendered.mapNotNull { it.tool }.filter { it.status in setOf("completed", "failed", "cancelled", "rejected", "skipped") } if (superseded.isNotEmpty()) { diff --git a/src/apps/mobile/shared/core-feature/src/commonMain/kotlin/com/openbitfun/mobile/core/feature/session/SubagentBranchCoordinator.kt b/src/apps/mobile/shared/core-feature/src/commonMain/kotlin/com/openbitfun/mobile/core/feature/session/SubagentBranchCoordinator.kt new file mode 100644 index 0000000000..57449a32c7 --- /dev/null +++ b/src/apps/mobile/shared/core-feature/src/commonMain/kotlin/com/openbitfun/mobile/core/feature/session/SubagentBranchCoordinator.kt @@ -0,0 +1,837 @@ +package com.openbitfun.mobile.core.feature.session + +import com.openbitfun.mobile.core.domain.ChatMessage +import com.openbitfun.mobile.core.domain.ToolNamePolicy +import com.openbitfun.mobile.core.protocol.ChatMessageItemResponse +import com.openbitfun.mobile.core.protocol.RemoteToolStatusResponse +import com.openbitfun.mobile.core.transport.HostStreamUnsupportedException +import com.openbitfun.mobile.core.transport.RemoteSessionStreamTransport +import com.openbitfun.mobile.core.transport.STREAM_EVENT_GAP +import com.openbitfun.mobile.core.transport.STREAM_EVENT_HISTORY_STARTED +import com.openbitfun.mobile.core.transport.STREAM_EVENT_READY +import com.openbitfun.mobile.core.transport.STREAM_EVENT_RESUMED +import kotlinx.coroutines.CancellationException +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Job +import kotlinx.coroutines.NonCancellable +import kotlinx.coroutines.launch +import kotlinx.coroutines.withContext +import kotlinx.serialization.json.JsonObject +import kotlinx.serialization.json.jsonPrimitive + +/** + * How many of a session's children are followed, and how many wait for a slot. + * + * Read by focused tests; the transcript itself never needs it. + */ +internal data class SubagentFollowState( + val branches: Int, + val subscribed: Int, + val waiting: Int, +) + +/** + * How many child sessions one parent session follows at once. + * + * Every child is another paged stream read over the same device RPC lane as the + * parent session, and a delegation wider than this is past what a phone screen + * presents at once. The cap never costs a card its content: a child that holds + * no slot keeps the branch it already published, and takes a slot as soon as one + * frees or the reader opens its card. + */ +internal const val MAX_CONCURRENT_CHILD_STREAMS: Int = 8 + +/** + * How many distinct child sessions one parent session tracks over its lifetime. + * + * A long-lived session can name more workers than any reader will scroll through, + * and each tracked child holds its own reduced transcript, so the map is bounded + * rather than growing with the session's history. Both the tracked branches and the + * launch items pointing at them are bounded by this number: one owner is one entry, + * and a session that keeps launching workers would otherwise grow that index with + * its whole history. Past the bound, a later launch is not followed: its card keeps + * the launching item's own status and result, which is what every card showed + * before child streams existed. The number is far above the concurrent cap on + * purpose — this bounds memory, not attention. + */ +internal const val MAX_TRACKED_CHILD_BRANCHES: Int = 64 + +/** + * Streams a delegation tool's child sessions into the branch that launched them. + * + * A host keeps a spawned child agent in its own hidden session: the parent record + * names that session on the launching tool item, and nothing the child does + * appears in the parent's own stream. The only place the child's steps exist is + * its stream, so the branch content is grouped by the subscription the records + * arrived on — the child session id the host named — and never by a marker on a + * record. Behavioural reference: HarmonyOS + * `harmonyos/entry/src/main/ets/services/SubagentStreamCoordinator.ets`. + * + * The parent session's replica stays the single author of the parent transcript: + * this coordinator only attaches each child's own projection as the branch items + * of the item that owns it, and [MessageBlockPresentation] then draws that item as + * the card it already knew how to draw. A host that names no child session + * produces no branch here at all, so an older desktop keeps the plain tool rows it + * has always sent. + * + * Two facts about a branch are read from two different places, and are kept apart + * on purpose: + * + * - The launching item's own status is the spawn call's outcome. A host answers an + * asynchronous `AgentSpawn` as soon as the worker starts, so that status says + * nothing about whether the worker is still working. + * - The card's lifecycle is the worker's, so it is derived from the child's own + * stream here, and only that reading is what the card is fed. + * + * Nothing is persisted: a child transcript is reconstructible by asking the host + * for the child's stream again, and this change deliberately keeps child steps out + * of the phone's stored parent transcript (HarmonyOS does cache them). + * + * A child stream that cannot be read is not a failure of the session on screen. The + * parent transcript is still the host's answer for the conversation the reader + * opened, so a child's error is reported as a per-card fact and the session stays + * readable; the reference client is explicit about the same thing + * (`SubagentStreamCoordinator.ets`, a child stream that cannot be opened does not + * fail the session). + * + * Threading: every member runs on the store's dispatcher — the same one that + * renders the parent transcript and collects the parent stream — so branch state + * needs no locking, and a child record can never interleave with a render. + */ +internal class SubagentBranchCoordinator( + private val scope: CoroutineScope, + private val source: RemoteSessionStreamTransport?, + private val publish: () -> Unit, + /** + * Reports one child's unreadable stream, keyed by the launch item it belongs to. + * + * Deliberately not the parent session's failure channel: the conversation is + * still readable, so a child's stream cannot be allowed to replace it. + */ + private val onError: (ownerToolCallId: String, error: Throwable) -> Unit, +) { + /** One child session whose records are drawn inside the branch that launched it. */ + private class Branch( + val sessionId: String, + var ownerToolCallId: String, + var replica: SessionRecordReplica = SessionRecordReplica(sessionId), + var job: Job? = null, + /** Identifies the current subscription, so a late `finally` cannot clear a newer one. */ + var subscription: Int = 0, + var subscribed: Boolean = false, + /** False until the child's first page settled, so its backfill is one publish. */ + var caughtUp: Boolean = false, + /** True when the child's stream failed before it published anything. */ + var streamFailed: Boolean = false, + var items: List = emptyList(), + /** The worker's own lifecycle: running, completed, failed, or empty for "nothing to say". */ + var status: String = "", + /** Whether [status] came from the child's own stream rather than from silence. */ + var liveStatus: Boolean = false, + var signature: String = "", + /** How many times this branch's content or status changed. */ + var changes: Int = 0, + /** Monotonic stamp of when this child was last seen working; decides release order. */ + var lastActivity: Long = 0, + ) + + private var parentSessionId: String = "" + private val branches = LinkedHashMap() + /** Owner tool call id -> child session id, i.e. which item a child belongs to. */ + private val owners = mutableMapOf() + /** Child session ids holding no slot, oldest first; newest is served first. */ + private val waiters = ArrayDeque() + /** Monotonic stamp backing [Branch.lastActivity], so release order needs no clock. */ + private var activitySeq: Long = 0 + /** True once this host answered a child stream with "no on-demand streams". */ + private var unsupported: Boolean = false + private var foreground: Boolean = true + /** Set while [decorate] runs, so a publish cannot re-enter the render it came from. */ + private var decorating: Boolean = false + private var publishPending: Boolean = false + + /** + * Binds the coordinator to the parent session that is being opened. + * + * Nothing survives a session switch: another session's children are not this + * session's branches, and their streams would keep pushing records into a + * transcript that is no longer on screen. + */ + fun beginSession(parentSessionId: String) { + detach() + this.parentSessionId = parentSessionId + } + + /** Closes every child stream; the parent session's own transcript is untouched. */ + fun endSession() { + detach() + } + + /** + * The parent transcript with every named child's work attached to its owner. + * + * Discovery happens here and only here: a child is opened when the parent + * transcript itself names one, so nothing is probed speculatively and no + * amount of tool ordering is treated as a subagent. + * + * Messages are returned unchanged when no branch applies, so a session without + * spawned children pays nothing beyond one map lookup per tool item. + */ + fun decorate(messages: List): List { + if (parentSessionId.isEmpty()) return messages + decorating = true + try { + observe(messages) + if (branches.isEmpty()) return messages + var decoratedAny = false + val decorated = messages.map { message -> + val items = message.items + if (items.isNullOrEmpty()) return@map message + var attached = 0 + val nextItems = items.mapIndexed { index, item -> + val childSessionId = owners[item.tool?.id.orEmpty()] ?: return@mapIndexed item + val branch = branches[childSessionId] ?: return@mapIndexed item + val replaced = ownerItem(item, branch, index, items) ?: return@mapIndexed item + attached += 1 + replaced + } + if (attached == 0) message else { + decoratedAny = true + message.copy(items = nextItems) + } + } + return if (decoratedAny) decorated else messages + } finally { + decorating = false + if (publishPending) { + publishPending = false + publish() + } + } + } + + /** + * Makes sure the child behind one launching item is being followed. + * + * This is what a reader opening a card asks for. The branch may have lost its + * slot to the cap, never had one, or been released when the phone left the + * foreground; either way an open card has to show the worker's own steps rather + * than a prefix that stopped where its slot did. + * + * Returns true when a stream was actually (re)opened. False means nothing was + * read — the child is unknown, already followed, or this host cannot serve + * child streams at all — and the caller must not treat that as a retry. + */ + fun ensureAttached(ownerToolCallId: String): Boolean { + if (parentSessionId.isEmpty()) return false + val childSessionId = owners[ownerToolCallId] ?: return false + val branch = branches[childSessionId] ?: return false + if (branch.subscribed) return false + if (source == null || unsupported) return false + acquireSubscriber(branch) + return branch.subscribed + } + + /** + * Follows children only while the session is on screen. + * + * A child stream is another device RPC lane, and nothing is looking at its card + * once the app is in the background. What the branch published stays attached, + * so the card keeps its steps; the stream is re-opened on the way back in, and a + * child that published while this phone was away replays from its newest page. + * + * A settled worker is not re-opened: it has nothing left to publish (that is what + * [isSettled] reads), so following it again would spend one of the eight child + * lanes on a card that cannot change — and a phone that leaves and returns a few + * times would spend them on the oldest finished workers rather than on the ones + * still working. + */ + fun setForeground(active: Boolean) { + if (active == foreground) return + foreground = active + if (!active) { + branches.values.filter { it.subscribed }.forEach { releaseSubscription(it, wait = false) } + return + } + branches.values.filterNot(::isSettled).sortedByDescending { it.lastActivity }.forEach { acquireSubscriber(it) } + } + + /** What this coordinator currently follows, for focused tests. */ + fun followState(): SubagentFollowState = + SubagentFollowState(branches.size, subscribedCount(), waiters.size) + + /** + * The children the transcript itself names. + * + * Discovery reads the rendered parent transcript rather than the raw records, + * because the item is where the host writes the child session id and the item is + * what survives both a live record and a re-render of the same transcript. + * + * On this client [decorate] is called from one place only — `render()` in + * `RemoteSessionStore`, over the messages the open session's replica just + * produced. The transcript restored from this device's cache is deliberately not + * decorated: it still carries the launch item's child id, but it is this device's + * copy of a session the host has not answered for yet, and opening a child stream + * against it would follow a worker for a session the host may have moved on from. + * The card is served as soon as the host's own transcript is rendered, which is + * when the launching item is the host's word rather than the cache's. + * + * A host that starts publishing a subagent's process inside its own stream — the + * older shape — is honoured from that moment on: a branch already followed for + * that owner is released and forgotten, so the transcript's own grouping draws it + * and this coordinator stops rewriting the item. + */ + private fun observe(messages: List) { + messages.forEach { message -> + val items = message.items.orEmpty() + items.forEachIndexed { index, item -> + val tool = item.tool ?: return@forEachIndexed + val ownerToolCallId = tool.id.orEmpty() + // A host that publishes a subagent's process inside its own stream + // needs no child stream at all: those records are already here, and + // opening the child would read the same work twice. + if (parentCarriesBranchItems(items, index)) { + dropBranch(owners[ownerToolCallId].orEmpty()) + return@forEachIndexed + } + openBranch(tool.subagentSessionId.orEmpty(), ownerToolCallId) + } + } + } + + /** + * Stops following one child, keeping nothing about it. + * + * Used when the parent takes the branch over: its own records are the process + * now, so the stream is released and the owner mapping is dropped, which is what + * stops the item from being rewritten with a branch that no longer speaks for it. + */ + private fun dropBranch(childSessionId: String) { + if (childSessionId.isEmpty()) return + val branch = branches.remove(childSessionId) ?: return + releaseSubscription(branch, wait = false) + forgetWaiter(branch) + owners.entries.removeAll { it.value == childSessionId } + drainWaiters() + } + + /** + * One launching item, as the card reads it. + * + * Two things are replaced and nothing else: the status the card draws is the + * worker's own lifecycle, and the branch's items are the worker's own process. + * A host that publishes the process inside its own stream keeps this item on + * the path it has always used, and needs no branch attached here. + */ + private fun ownerItem( + item: ChatMessageItemResponse, + branch: Branch, + ownerIndex: Int, + items: List, + ): ChatMessageItemResponse? { + val tool = if (branch.status.isNotEmpty()) item.tool?.copy(status = branch.status) else item.tool + val subItems = if (branch.items.isEmpty() || parentCarriesBranchItems(items, ownerIndex)) { + item.subItems + } else { + branch.items + } + if (tool == item.tool && subItems == item.subItems) return null + return item.copy(tool = tool, subItems = subItems) + } + + /** + * Whether the parent stream itself already carries this owner's process. + * + * Older hosts published a subagent's records inside the parent stream and + * marked them, and the transcript grouping already attaches those records to + * the item that launched them. Attaching the child session's records here as + * well would draw every step twice, so a marked sibling record — up to the next + * launch item, which owns its own records — keeps that host on the path it has + * always used. + */ + private fun parentCarriesBranchItems(items: List, ownerIndex: Int): Boolean { + for (index in ownerIndex + 1 until items.size) { + val entry = items[index] + if (entry.isSubagent != true) continue + val tool = entry.tool + if (tool != null && isOwnerTool(tool)) return false + return true + } + return false + } + + /** Whether an item is one of the tools that launches a delegation. */ + private fun isOwnerTool(tool: RemoteToolStatusResponse): Boolean = + ToolNamePolicy.isTask(tool) || ToolNamePolicy.normalized(tool) == "agentspawn" + + private fun openBranch(childSessionId: String, ownerToolCallId: String) { + // A host that names no child session, or names this very session, has no + // branch to open: the launch item stays the plain tool row it always was. + if (childSessionId.isEmpty() || ownerToolCallId.isEmpty()) return + if (parentSessionId == childSessionId) return + if (unsupported) return + // One subscription per owner. The same launch is republished by every + // parent record, and that is not a reason to take a slot back from a worker + // that is using one. + if (owners.containsKey(ownerToolCallId)) return + // Owners and branches are bounded by the same number: an owner is one entry + // pointing at a branch, and a session that keeps launching workers would + // otherwise grow this map with its whole history. A launch past the bound is + // not followed; its card keeps the launching item's own status. + if (owners.size >= MAX_TRACKED_CHILD_BRANCHES || branches.size >= MAX_TRACKED_CHILD_BRANCHES) return + val resumed = branches[childSessionId] + if (resumed != null) { + // The same worker, named by a launch this session has not seen before: a + // reusable child that was continued. It is one branch — its steps are the + // worker's, not the launch's — but the launch happening now is what its + // records have to land under, and a branch the cap released has to be + // followed again, or the turn it is taking would never arrive. + owners[ownerToolCallId] = childSessionId + resumed.ownerToolCallId = ownerToolCallId + resumed.lastActivity = ++activitySeq + if (!resumed.subscribed) acquireSubscriber(resumed) + return + } + val branch = Branch(sessionId = childSessionId, ownerToolCallId = ownerToolCallId, lastActivity = ++activitySeq) + branches[childSessionId] = branch + owners[ownerToolCallId] = childSessionId + acquireSubscriber(branch) + } + + /** + * Opens a child's stream if this session can follow one more child. + * + * Past the cap a slot is taken from another child, oldest activity first and a + * settled child before a working one: a worker that already reported is the one + * whose stream can be released for nothing, and a working one that loses its + * slot keeps the branch it published and waits to be re-attached. + */ + private fun acquireSubscriber(branch: Branch) { + if (branch.subscribed || source == null || unsupported) return + if (subscribedCount() >= MAX_CONCURRENT_CHILD_STREAMS) { + val victim = releaseCandidate(branch) + if (victim == null) { + waitForSlot(branch) + return + } + releaseSubscription(victim, wait = true) + } + forgetWaiter(branch) + subscribe(branch) + // Subscribing decides what can be said about the worker: a branch opened + // over the cap has no reading until this point. + refresh(branch) + } + + private fun subscribe(branch: Branch) { + val source = source ?: return + val token = ++branch.subscription + branch.subscribed = true + branch.job = scope.launch { + try { + source.subscribe( + branch.sessionId, + { error -> childStreamFailed(branch.sessionId, token, error) }, + { childCaughtUp(branch.sessionId) }, + ).collect { event -> applyChildEvent(branch.sessionId, event) } + } catch (cancelled: CancellationException) { + throw cancelled + } catch (error: Throwable) { + childStreamFailed(branch.sessionId, token, error) + } finally { + // Cancellation is what releases the child's flow, and the host + // stream's own cleanup runs under `NonCancellable`; this marks the + // slot free on the same terms, so the cap cannot drift. + withContext(NonCancellable) { + val current = branches[branch.sessionId] + if (current === branch && branch.subscription == token) { + branch.subscribed = false + branch.job = null + } + } + } + } + } + + /** Closes one child's stream, keeping the branch and everything it published. */ + private fun releaseSubscription(branch: Branch, wait: Boolean) { + branch.job?.cancel() + branch.job = null + branch.subscription += 1 + branch.subscribed = false + if (wait) waitForSlot(branch) + } + + /** + * Which child gives up its slot, or null when every other child is already + * waiting. + * + * The most recently active children stay followed — a card the reader is + * watching is the one that just changed — and a settled child goes before a + * working one, because releasing it cannot lose a step. + */ + private fun releaseCandidate(except: Branch): Branch? { + var settled: Branch? = null + var oldest: Branch? = null + branches.values.forEach { branch -> + if (branch === except || !branch.subscribed) return@forEach + val quietest = oldest + if (quietest == null || branch.lastActivity < quietest.lastActivity) oldest = branch + val settledFirst = settled + if (isSettled(branch) && (settledFirst == null || branch.lastActivity < settledFirst.lastActivity)) settled = branch + } + return settled ?: oldest + } + + private fun waitForSlot(branch: Branch) { + if (branch.sessionId !in waiters) waiters.addLast(branch.sessionId) + } + + private fun forgetWaiter(branch: Branch) { + waiters.removeAll { it == branch.sessionId } + } + + /** + * Follows waiting children as slots free, newest first. + * + * Newest first because a launch the reader just watched appear is the card most + * likely to be on screen; the older ones keep their branch and are served when + * their turn comes or when their card is opened. + */ + private fun drainWaiters() { + var budget = waiters.size + while (budget > 0 && subscribedCount() < MAX_CONCURRENT_CHILD_STREAMS) { + budget -= 1 + val sessionId = waiters.removeLastOrNull() ?: return + val branch = branches[sessionId] ?: continue + if (branch.subscribed) continue + acquireSubscriber(branch) + } + } + + private fun subscribedCount(): Int = branches.values.count { it.subscribed } + + /** + * A child that has settled: its own stream read a turn of it and no turn of it + * is in progress. + * + * Only the stream makes a branch settled. Silence is not a reading, so a child + * whose stream never published anything is never treated as finished. + */ + private fun isSettled(branch: Branch): Boolean = + branch.caughtUp && branch.liveStatus && branch.status.isNotEmpty() && branch.status !in UNSETTLED + + /** + * One record of one child session. + * + * Records are deduplicated by identity and revision inside the child's own + * replica, which is the same replica the session on screen uses, so a replayed + * page cannot publish a step twice. Until the child's first page has settled the + * parent is not republished at all: a child that finished before its branch was + * opened backfills hundreds of records, and each of them republishing the parent + * transcript would repaint the whole conversation per record for a branch nobody + * has seen yet. + */ + private fun applyChildEvent(childSessionId: String, event: JsonObject) { + val branch = branches[childSessionId] ?: return + if (parentSessionId.isEmpty()) return + // One subscription carries one session's records, so a record stamped with + // another session id is a broken contract, not a step of this worker. The + // parent stream fails loud on the same mismatch, and a child reduced into + // the wrong branch would put another agent's work on this card. + check(event["session_id"]?.jsonPrimitive?.content == childSessionId) { "Session binding mismatch" } + when (event["event"]?.jsonPrimitive?.content) { + STREAM_EVENT_GAP -> { + // The host restarted this child's stream: everything reduced so far + // came from its previous epoch, so it is dropped before the replay. + // Only this child is affected — the parent's own epoch is the + // parent replica's business, and another child's stream is another + // subscription. + branch.replica = SessionRecordReplica(childSessionId) + branch.caughtUp = false + branch.streamFailed = false + refresh(branch) + return + } + "session-record" -> { + // The replica drops a record it already holds at that revision, so a + // replayed page leaves the branch unchanged and publishes nothing. + val payload = event["payload"] as? JsonObject ?: return + branch.replica.apply(payload) + } + "agentic://tool-event" -> { + val payload = event["payload"] as? JsonObject ?: return + branch.replica.applyControl(payload) + } + STREAM_EVENT_READY, STREAM_EVENT_HISTORY_STARTED, STREAM_EVENT_RESUMED -> return + else -> return + } + branch.streamFailed = false + branch.lastActivity = ++activitySeq + if (!branch.caughtUp) return + refresh(branch) + } + + private fun childCaughtUp(childSessionId: String) { + val branch = branches[childSessionId] ?: return + branch.caughtUp = true + branch.streamFailed = false + refresh(branch) + } + + /** + * A child stream that failed or could not be opened. + * + * A child stream that failed before it published anything says nothing about the + * worker, so no lifecycle is invented for it: the card keeps the launching item's + * own status, and that child is not read again on its own (only a card being + * opened asks for it again). + * + * A stream that died *after* publishing steps is a different case, and the branch + * is marked failed either way. The steps it published stay — they are the worker's + * own work and dropping them would empty a card the reader was reading — but a + * worker whose last turn was still in progress when its stream stopped is no longer + * vouched for by anything: that `running` is a snapshot of a dead stream, not + * evidence the worker is alive, and a card that keeps saying "running" forever is + * the worst of the available readings. Nothing terminal is invented in its place; + * see [childStatus]. + * + * The failure is reported to this coordinator's own channel, never to the parent + * session's failure state. A host that serves this conversation perfectly well can + * still refuse one child's `read_stream`, and losing the readable transcript + * behind a session-level error would be a lie about what is on screen. A host that + * answers "no on-demand streams" at all is recorded once — no further child stream + * is opened for it — and every child stream is released. + * + * Only a failure belonging to the session still on screen is reported: an error + * that arrives after the session was closed, switched away from, or re-subscribed + * describes a stream nobody is looking at any more. + */ + private fun childStreamFailed(childSessionId: String, token: Int, error: Throwable) { + if (parentSessionId.isEmpty()) return + val branch = branches[childSessionId] ?: return + if (branch.subscription != token) return + // Whatever the branch published, a dead stream stops being a reading of the + // worker, so the derived status is recomputed rather than left as it was. + branch.streamFailed = true + refresh(branch) + if (error is HostStreamUnsupportedException) { + if (unsupported) return + unsupported = true + branches.values.filter { it.subscribed }.forEach { releaseSubscription(it, wait = false) } + } + onError(branch.ownerToolCallId, error) + } + + /** + * Republishes a branch that changed in any way the card can show. + * + * The items are what the card draws; the status is what its header says, and a + * child's turn can settle without publishing another item, so both are compared + * before the parent transcript is republished. + */ + private fun refresh(branch: Branch) { + val wasSettled = isSettled(branch) + val messages = branch.replica.messages() + val items = branchItems(messages) + val signature = branchSignature(items) + val live = turnStatus(messages) + val status = childStatus(branch, live) + branch.liveStatus = live.isNotEmpty() + if (signature == branch.signature && status == branch.status) return + branch.signature = signature + branch.items = items + branch.status = status + branch.changes += 1 + requestPublish() + // A settled worker will not publish again, so it costs nothing to stop + // following it. It is the transition that releases: a worker that settled, + // was continued and then settled again keeps its stream until the cap wants + // the slot back, which is where a settled branch is the first one to be taken. + if (!wasSettled && isSettled(branch) && branch.subscribed) { + releaseSubscription(branch, wait = false) + drainWaiters() + } + } + + /** + * Publishes once, and never from inside the render that caused the change. + * + * The store's publish re-renders the parent transcript, which decorates again; + * publishing from within [decorate] would recurse through that render. The + * pending publish is flushed as soon as the decoration finishes, so a branch + * opened by discovery still reaches the screen in the same turn. + */ + private fun requestPublish() { + if (decorating) { + publishPending = true + return + } + publish() + } + + /** + * The worker's own lifecycle, as its stream describes it. + * + * A child session is an ordinary session with its own turns, so the reading is + * the child's own: a turn of it still in progress means the worker is working, + * and the last turn's outcome is what a finished worker reported. Multi-turn + * children — a worker continued after it answered once — are running again the + * moment any of their turns is. + * + * Nothing is invented from what is missing. Before the child has said anything + * at all the worker was started and is not finished, so it reads as running; a + * stream that failed before publishing anything says nothing, and the launching + * item's own status stands. + * + * A dead stream makes no claim, which is why it wins over an in-progress turn: + * "the last turn I saw was still going" describes a stream that has stopped, not + * a worker that is alive, and nothing will ever settle it. The card therefore + * falls back to the launching item's own status — the host's word about the + * spawn, which this client has no right to overrule — rather than inventing a + * `failed` the host never sent. What the child did report before the stream died + * is a real outcome and is kept: a settled worker does not become unsettled + * because a stream ended afterwards, and its steps stay on the card either way. + */ + private fun childStatus(branch: Branch, live: String): String = when { + branch.streamFailed && live in UNSETTLED -> "" + live.isNotEmpty() -> live + branch.streamFailed -> "" + else -> RUNNING + } + + /** + * A child transcript's lifecycle reading, or "" when it holds no turn yet. + * + * The child's turn status is the host's own vocabulary, and only part of it + * means "finished". A turn that is starting, thinking, calling a tool, or + * waiting on a permission answer is still a turn in progress, and a card that + * read those as an outcome would print "Done" over a worker that was + * demonstrably still working. A turn the host accepted but has not started is + * [WAITING]: not an outcome, and not a worker at work either. The newest + * assistant message decides which reading applies; anything this client does + * not recognise is read as an outcome, which is the pre-existing behaviour. + */ + private fun turnStatus(messages: List): String { + var sawTurn = false + var running = false + var lastStatus = "" + var lastOutcome = "" + messages.forEach { message -> + if (message.role != "assistant") return@forEach + sawTurn = true + val status = message.status.orEmpty().lowercase() + lastStatus = status + if (status in RUNNING_STATUSES) { + running = true + return@forEach + } + if (status in FINISHING_STATUSES || status in WAITING_STATUSES) return@forEach + lastOutcome = if (status in FAILED_STATUSES) "failed" else "completed" + } + if (!sawTurn) return "" + return when { + // A turn of this worker is going, whatever any other turn of it says. + running -> RUNNING + lastStatus in FINISHING_STATUSES -> FINISHING + lastStatus in WAITING_STATUSES -> WAITING + else -> lastOutcome + } + } + + /** + * The process a child session publishes, in its own order. + * + * A child session is an ordinary session: it has its own turns and its own user + * message (the brief it was spawned with), which the card already shows as the + * branch title, so only the assistant's items are branch content. + */ + private fun branchItems(messages: List): List { + val items = mutableListOf() + messages.forEach { message -> + if (message.role != "assistant") return@forEach + items += message.items.orEmpty() + } + return items + } + + /** Everything about a branch that can change what the card draws. */ + private fun branchSignature(items: List): String = buildString { + append(items.size) + items.forEach { item -> + val tool = item.tool + append('\u0001') + append(item.type.orEmpty()) + append('\u0000') + append(item.content.orEmpty().length) + append('\u0000') + if (tool != null) { + append(tool.id.orEmpty()); append(':') + append(tool.name.orEmpty()); append(':') + append(tool.status.orEmpty()) + } + } + } + + private fun detach() { + branches.values.forEach { branch -> + branch.job?.cancel() + branch.job = null + branch.subscription += 1 + branch.subscribed = false + } + branches.clear() + owners.clear() + waiters.clear() + parentSessionId = "" + publishPending = false + decorating = false + } + + private companion object { + /** Host vocabulary, not user-visible copy: presentation maps it to its own words. */ + const val RUNNING = "running" + + /** + * The worker's turn is wrapping up: still its turn, no longer its work. + * + * Kept apart from [RUNNING] because the card says something different for + * it, and because a branch that reports it is not settled either. + */ + const val FINISHING = "finishing" + + /** + * The worker's turn is accepted but has not started: nothing of it is running. + * + * A queued turn is neither an outcome nor a worker at work. Reading it as + * [RUNNING] made the card say "Running" and pulse over a worker the host was + * still holding; reading it as an outcome printed "Done" before it had begun. + * It is kept out of the settled readings too: the turn it was queued for is + * still going to be published on this branch's stream. + */ + const val WAITING = "waiting" + + /** Readings that mean the worker still owes work, so a branch that says one is not settled. */ + val UNSETTLED = setOf(RUNNING, FINISHING, WAITING) + + /** + * Host vocabularies for "this turn is still going". + * + * `streaming` is here because the replica renders an in-progress turn as a + * streaming message, and a child transcript is read through that same replica. + * The rest are the host's own in-progress states: a turn that started, is + * thinking, is calling a tool, or is waiting on an answer has not finished, + * and reading any of them as an outcome is what printed "Done" over a + * worker that was still working. + */ + val RUNNING_STATUSES = setOf( + "running", "streaming", "active", "inprogress", "in_progress", "preparing", + "starting", "thinking", "tool_calling", "pending_confirmation", "capability_required", + ) + + /** Host vocabularies for "this turn was accepted and has not started yet". */ + val WAITING_STATUSES = setOf("queued", "pending", "waiting", "blocked", "deferred", "scheduled") + val FINISHING_STATUSES = setOf("finishing", "finalizing", "wrapping_up") + val FAILED_STATUSES = setOf("failed", "error", "timeout", "cancelled", "canceled", "rejected") + } +} diff --git a/src/apps/mobile/shared/core-feature/src/commonTest/kotlin/com/openbitfun/mobile/core/feature/session/MessageBlockPresentationTest.kt b/src/apps/mobile/shared/core-feature/src/commonTest/kotlin/com/openbitfun/mobile/core/feature/session/MessageBlockPresentationTest.kt index 5b9de92922..e7485d1cce 100644 --- a/src/apps/mobile/shared/core-feature/src/commonTest/kotlin/com/openbitfun/mobile/core/feature/session/MessageBlockPresentationTest.kt +++ b/src/apps/mobile/shared/core-feature/src/commonTest/kotlin/com/openbitfun/mobile/core/feature/session/MessageBlockPresentationTest.kt @@ -68,13 +68,15 @@ class MessageBlockPresentationTest { @Test fun subagentStatusPreservesHostVocabularyAndOwnActiveScope() { - for (status in listOf("running", "active", "preparing", "pending", "queued", "failed", "error", "timeout", "cancelled", "canceled", "rejected", "completed")) { + for (status in listOf("running", "active", "preparing", "pending", "queued", "waiting", "failed", "error", "timeout", "cancelled", "canceled", "rejected", "completed")) { val blocks = messageBlocks(message(items = listOf( item(type = "subagent", tool = tool("task", name = "Task", status = status)), item(type = "thinking", content = "child").copy(isSubagent = true), item(type = "text", content = "Parent continues"))), true) val child = blocks.first() as MessageBlock.Subagent - val running = status in listOf("running", "active", "preparing", "pending", "queued") + // `waiting` is the coordinator's word for a queued turn: not working, but + // still owed, so the scope stays open the way it does for `queued`. + val running = status in listOf("running", "active", "preparing", "pending", "queued", "waiting") assertEquals(status, child.status) assertEquals(running, child.running) assertEquals(running, (child.children.single() as MessageBlock.Thinking).streaming) diff --git a/src/apps/mobile/shared/core-feature/src/commonTest/kotlin/com/openbitfun/mobile/core/feature/session/SubagentBranchCoordinatorTest.kt b/src/apps/mobile/shared/core-feature/src/commonTest/kotlin/com/openbitfun/mobile/core/feature/session/SubagentBranchCoordinatorTest.kt new file mode 100644 index 0000000000..dee10ad4ad --- /dev/null +++ b/src/apps/mobile/shared/core-feature/src/commonTest/kotlin/com/openbitfun/mobile/core/feature/session/SubagentBranchCoordinatorTest.kt @@ -0,0 +1,1126 @@ +package com.openbitfun.mobile.core.feature.session + +import com.openbitfun.mobile.core.domain.ChatMessage +import com.openbitfun.mobile.core.domain.ChatTimelineStore +import com.openbitfun.mobile.core.feature.connection.ConnectionPhase +import com.openbitfun.mobile.core.feature.session.RemoteSessionFailureReason +import com.openbitfun.mobile.core.protocol.ChatMessageItemResponse +import com.openbitfun.mobile.core.protocol.CommandStatus +import com.openbitfun.mobile.core.protocol.RelayJson +import com.openbitfun.mobile.core.protocol.RemoteCommand +import com.openbitfun.mobile.core.protocol.RemoteToolStatusResponse +import com.openbitfun.mobile.core.transport.HostStreamUnsupportedException +import com.openbitfun.mobile.core.transport.RemoteCommandTransport +import com.openbitfun.mobile.core.transport.RemoteSessionStreamTransport +import com.openbitfun.mobile.core.transport.STREAM_EVENT_GAP +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.MutableSharedFlow +import kotlinx.coroutines.flow.flow +import kotlinx.coroutines.test.runCurrent +import kotlinx.coroutines.test.runTest +import kotlinx.serialization.DeserializationStrategy +import kotlinx.serialization.json.JsonObject +import kotlinx.serialization.json.buildJsonObject +import kotlinx.serialization.json.jsonObject +import kotlinx.serialization.json.put +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertIs +import kotlin.test.assertNull +import kotlin.test.assertSame +import kotlin.test.assertTrue + +private const val PARENT = "s-parent" +private const val CHILD = "s-child" +private const val OWNER = "tool-owner" + +/** + * A delegation tool's child session, drawn inside the card that launched it. + * + * Every host here is simulated: these tests say what this client does with the + * records a host publishes, and nothing about what a live host publishes. The + * reference for the behaviour is the HarmonyOS coordinator; the grouping rule it + * implements is the one asserted here — a child's steps belong to the launch item + * the parent transcript names, and nothing else is ever probed. + */ +@OptIn(ExperimentalCoroutinesApi::class) +class SubagentBranchCoordinatorTest { + + /** + * The child session id the host stamped on the launching item has to survive + * mapping: it is the only key a reader has to open the child's stream. + */ + @Test + fun aChildSessionIdSurvivesTheParentTranscriptAndMarksTheBranch() { + val replica = SessionRecordReplica(PARENT) + replica.apply(launchEvent(PARENT, "t-parent", OWNER, CHILD, revision = 1).getValue("payload").jsonObject) + val item = replica.messages().flatMap { it.items.orEmpty() }.single { it.tool != null } + + assertEquals(CHILD, item.tool?.subagentSessionId) + assertEquals(OWNER, item.tool?.id) + assertEquals(true, item.isSubagent) + } + + /** + * One subscription per owner. The parent republishes the same launch on every + * record it receives, and that is not a reason to take a slot back from a + * worker that is using one. + */ + @Test + fun oneSubscriptionPerOwnerEvenWhenTheLaunchIsRestated() = runTest { + val host = FakeChildStreams() + val coordinator = SubagentBranchCoordinator(backgroundScope, host, publish = {}, onError = { _, _ -> }) + coordinator.beginSession(PARENT) + val parent = listOf(parentMessage(launchItem(OWNER, CHILD))) + + coordinator.decorate(parent) + runCurrent() + coordinator.decorate(parent) + coordinator.decorate(parent) + runCurrent() + + assertEquals(listOf(CHILD), host.attempts) + assertEquals(1, host.openCount) + assertEquals(1, coordinator.followState().branches) + } + + /** + * The card shows the worker's own steps and the worker's own lifecycle. The + * launch item's status is the spawn call's outcome, which says nothing about + * whether the worker is still working. + */ + @Test + fun theChildsStepsAttachToTheOwnerItemAndItsStatusFollowsTheChildsTurns() = runTest { + val host = FakeChildStreams() + val coordinator = SubagentBranchCoordinator(backgroundScope, host, publish = {}, onError = { _, _ -> }) + coordinator.beginSession(PARENT) + val parent = listOf(parentMessage(launchItem(OWNER, CHILD, status = "completed"))) + + var decorated = coordinator.decorate(parent).single().items!!.single() + runCurrent() + // Started and not finished: the worker is running, whatever the spawn answered. + assertEquals("running", decorated.tool?.status) + + host.push(CHILD, richRecord(CHILD, "t-child", 0, 1, "inprogress", "step one")) + runCurrent() + decorated = coordinator.decorate(parent).single().items!!.single() + assertEquals(listOf("step one"), decorated.subItems?.map { it.content }) + assertEquals("running", decorated.tool?.status) + + host.push(CHILD, richRecord(CHILD, "t-child", 0, 2, "completed", "step one")) + runCurrent() + decorated = coordinator.decorate(parent).single().items!!.single() + assertEquals(listOf("step one"), decorated.subItems?.map { it.content }) + assertEquals("completed", decorated.tool?.status) + coordinator.endSession() + } + + /** + * A worker that is still taking its turn is never reported as finished. + * + * The host's turn vocabulary is wider than `running`: a turn that started, is + * thinking, is calling a tool, or is waiting on a permission answer has not + * produced an outcome, and reading any of them as one is what printed "Done" + * over a worker that was still working. The card is handed the reading, so the + * reading is what is asserted here. + */ + @Test + fun aChildTurnThatIsStillTakingItsTurnIsNotReportedAsFinished() = runTest { + val host = FakeChildStreams() + val coordinator = SubagentBranchCoordinator(backgroundScope, host, publish = {}, onError = { _, _ -> }) + coordinator.beginSession(PARENT) + val parent = listOf(parentMessage(launchItem(OWNER, CHILD, status = "completed"))) + + coordinator.decorate(parent) + runCurrent() + + // Each of these is a turn in progress, published by the host in that word. + var revision = 0L + for (status in listOf("starting", "thinking", "tool_calling", "pending_confirmation", "capability_required")) { + revision += 1 + host.push(CHILD, richRecord(CHILD, "t-child", 0, revision, status, "step one")) + runCurrent() + assertEquals("running", coordinator.decorate(parent).single().items!!.single().tool?.status) + } + + revision += 1 + host.push(CHILD, richRecord(CHILD, "t-child", 0, revision, "completed", "step one")) + runCurrent() + assertEquals("completed", coordinator.decorate(parent).single().items!!.single().tool?.status) + coordinator.endSession() + } + + /** + * A queued turn is its own reading: not running, not done. + * + * The host accepts a turn before it starts one, and says so with `queued` or + * `pending`. The card drew that as "Running" — and, since it is in flight for + * the drawing, pulsed over a worker that had not begun — while an older reading + * of the same word printed "Done". It is neither: the reading is passed through + * as `waiting`, and the branch keeps its stream, because the turn it was queued + * for is still going to be published on it. + */ + @Test + fun aChildTurnThatIsOnlyQueuedIsReportedAsWaitingAndKeepsItsStream() = runTest { + val host = FakeChildStreams() + val coordinator = SubagentBranchCoordinator(backgroundScope, host, publish = {}, onError = { _, _ -> }) + coordinator.beginSession(PARENT) + val parent = listOf(parentMessage(launchItem(OWNER, CHILD, status = "completed"))) + + coordinator.decorate(parent) + runCurrent() + + var revision = 0L + for (status in listOf("queued", "pending", "waiting")) { + revision += 1 + host.push(CHILD, richRecord(CHILD, "t-child", 0, revision, status, "step one")) + runCurrent() + assertEquals("waiting", coordinator.decorate(parent).single().items!!.single().tool?.status) + } + + // Accepted, not started: releasing the stream here would drop the turn the + // host queued, so the branch keeps its slot. + assertEquals(1, host.openCount) + + // The work it was queued for still arrives on the same stream, and is the + // reading from then on. + revision += 1 + host.push(CHILD, richRecord(CHILD, "t-child", 0, revision, "tool_calling", "step one")) + runCurrent() + assertEquals("running", coordinator.decorate(parent).single().items!!.single().tool?.status) + + revision += 1 + host.push(CHILD, richRecord(CHILD, "t-child", 0, revision, "completed", "step one")) + runCurrent() + assertEquals("completed", coordinator.decorate(parent).single().items!!.single().tool?.status) + coordinator.endSession() + } + + /** + * A turn that is wrapping up is its own reading, and it is not an outcome. + * + * The host says `finishing` between the last token and the result. Collapsing + * that into `completed` told the reader the worker was done while it still had + * work to close out, so the reading is passed through and the branch keeps its + * stream until the turn actually settles. + */ + @Test + fun aChildTurnThatIsFinishingIsReportedAsFinishingAndKeepsItsStream() = runTest { + val host = FakeChildStreams() + val coordinator = SubagentBranchCoordinator(backgroundScope, host, publish = {}, onError = { _, _ -> }) + coordinator.beginSession(PARENT) + val parent = listOf(parentMessage(launchItem(OWNER, CHILD, status = "completed"))) + + coordinator.decorate(parent) + runCurrent() + host.push(CHILD, richRecord(CHILD, "t-child", 0, 1, "finishing", "step one")) + runCurrent() + + assertEquals("finishing", coordinator.decorate(parent).single().items!!.single().tool?.status) + // Still the worker's turn: releasing the stream here would drop what it has + // left to publish, so the branch keeps its slot. + assertEquals(1, host.openCount) + + host.push(CHILD, richRecord(CHILD, "t-child", 0, 2, "completed", "step one")) + runCurrent() + assertEquals("completed", coordinator.decorate(parent).single().items!!.single().tool?.status) + coordinator.endSession() + } + + /** + * Older hosts published a subagent's records inside the parent stream and + * marked them, and the transcript grouping already attaches those to the item + * that launched them. Attaching the child session as well would draw every + * step twice, so nothing is attached and no child stream is opened. + */ + @Test + fun aHostThatMarksItsOwnSiblingRecordsKeepsThePathItAlwaysUsed() = runTest { + val host = FakeChildStreams() + val coordinator = SubagentBranchCoordinator(backgroundScope, host, publish = {}, onError = { _, _ -> }) + coordinator.beginSession(PARENT) + val parent = listOf( + parentMessage( + launchItem(OWNER, CHILD, status = "completed"), + ChatMessageItemResponse(type = "text", content = "step one", isSubagent = true), + ), + ) + + val decorated = coordinator.decorate(parent).single() + runCurrent() + + assertTrue(host.attempts.isEmpty()) + assertNull(decorated.items!!.first().subItems) + assertEquals("completed", decorated.items!!.first().tool?.status) + coordinator.endSession() + } + + /** + * A child stream that restarts drops only that child's replica, and the replay + * that follows is the same steps, not a second copy of them. + */ + @Test + fun aChildGapReReadsThatChildOnceWithoutDuplicatingItsSteps() = runTest { + val host = FakeChildStreams() + val coordinator = SubagentBranchCoordinator(backgroundScope, host, publish = {}, onError = { _, _ -> }) + coordinator.beginSession(PARENT) + val parent = listOf(parentMessage(launchItem(OWNER, CHILD))) + + coordinator.decorate(parent) + runCurrent() + host.push(CHILD, richRecord(CHILD, "t-child", 0, 1, "inprogress", "step one")) + runCurrent() + assertEquals(1, coordinator.decorate(parent).single().items!!.single().subItems?.size) + + host.push(CHILD, buildJsonObject { put("session_id", CHILD); put("event", STREAM_EVENT_GAP); put("payload", buildJsonObject {}) }) + runCurrent() + // The host restarted the child's stream, so what was reduced is stale until + // the replay arrives; the branch is empty in between, not wrong. + assertNull(coordinator.decorate(parent).single().items!!.single().subItems) + + host.push(CHILD, richRecord(CHILD, "t-child", 0, 1, "inprogress", "step one")) + host.catchUp(CHILD) + runCurrent() + val decorated = coordinator.decorate(parent).single().items!!.single() + assertEquals(listOf("step one"), decorated.subItems?.map { it.content }) + assertEquals(1, host.attempts.count { it == CHILD }) + coordinator.endSession() + } + + /** + * A child stream that failed before it published anything says nothing about + * the worker: no lifecycle is invented, the launch item's own status stands, + * and the failure is reported once, keyed by the card that owns it — never as a + * failure of the conversation around it. + */ + @Test + fun aChildStreamThatFailsBeforeAnyContentKeepsTheLaunchStatusAndReportsOnce() = runTest { + val host = FakeChildStreams() + val surfaced = mutableListOf() + val coordinator = SubagentBranchCoordinator(backgroundScope, host, publish = {}, onError = { _, error -> surfaced += error }) + coordinator.beginSession(PARENT) + val parent = listOf(parentMessage(launchItem(OWNER, CHILD, status = "completed"))) + + coordinator.decorate(parent) + runCurrent() + host.fail(CHILD, IllegalStateException("stream dropped")) + runCurrent() + + val decorated = coordinator.decorate(parent).single().items!!.single() + assertEquals("completed", decorated.tool?.status) + assertNull(decorated.subItems) + assertEquals(1, surfaced.size) + assertIs(surfaced.single()) + coordinator.endSession() + } + + /** + * A child stream that dies after publishing steps keeps them but stops being a + * reading of the worker. + * + * The steps are the worker's own work, so dropping them would empty a card the + * reader was reading. The `running` is not: it came from a turn of a stream that + * has stopped, and nothing will ever settle it, so the card falls back to the + * launching item's own status — the host's word about the spawn — instead of + * claiming forever that a dead worker is still working. No `failed` is invented, + * and the failure is reported once on the card that owns it. + */ + @Test + fun aChildStreamThatDiesAfterPublishingStepsKeepsThemAndStopsClaimingToBeRunning() = runTest { + val host = FakeChildStreams() + val surfaced = mutableListOf() + val coordinator = SubagentBranchCoordinator(backgroundScope, host, publish = {}, onError = { _, error -> surfaced += error }) + coordinator.beginSession(PARENT) + val parent = listOf(parentMessage(launchItem(OWNER, CHILD, status = "completed"))) + + coordinator.decorate(parent) + runCurrent() + host.push(CHILD, richRecord(CHILD, "t-child", 0, 1, "inprogress", "step one")) + runCurrent() + // The worker's turn is still in progress, so the card reads as running. + assertEquals("running", coordinator.decorate(parent).single().items!!.single().tool?.status) + + host.fail(CHILD, IllegalStateException("stream dropped")) + runCurrent() + + val card = coordinator.decorate(parent).single().items!!.single() + assertEquals(listOf("step one"), card.subItems?.map { it.content }) + assertEquals("completed", card.tool?.status) + assertEquals(1, surfaced.size) + assertIs(surfaced.single()) + coordinator.endSession() + } + + /** + * A record belongs to the session that published it. + * + * One subscription carries one child's records, so a record stamped with another + * session id is a broken contract. Reducing it would put another agent's work on + * this card, so it is rejected the way the parent stream rejects the same + * mismatch — loud, and contained to the card that owns the stream. + */ + @Test + fun aChildRecordStampedWithAnotherSessionIsRejectedInsteadOfReduced() = runTest { + val host = FakeChildStreams() + val surfaced = mutableListOf() + val coordinator = SubagentBranchCoordinator(backgroundScope, host, publish = {}, onError = { _, error -> surfaced += error }) + coordinator.beginSession(PARENT) + val parent = listOf(parentMessage(launchItem(OWNER, CHILD))) + + coordinator.decorate(parent) + runCurrent() + host.push(CHILD, richRecord("s-someone-else", "t-other", 0, 1, "inprogress", "another agent's work")) + runCurrent() + + val card = coordinator.decorate(parent).single().items!!.single() + assertNull(card.subItems) + assertEquals(1, surfaced.size) + assertIs(surfaced.single()) + coordinator.endSession() + } + + /** + * A host without on-demand streams is recorded once and then left alone: no + * per-card retry, no command it would fail to parse, and no child read again. + */ + @Test + fun aHostWithoutOnDemandStreamsIsReportedOnceAndNoFurtherChildIsRead() = runTest { + val host = FakeChildStreams().apply { failSubscribe = { HostStreamUnsupportedException() } } + val surfaced = mutableListOf() + val coordinator = SubagentBranchCoordinator(backgroundScope, host, publish = {}, onError = { _, error -> surfaced += error }) + coordinator.beginSession(PARENT) + val parent = listOf(parentMessage(launchItem(OWNER, CHILD), launchItem("tool-owner-2", "s-child-2"))) + + coordinator.decorate(parent) + runCurrent() + coordinator.decorate(parent) + runCurrent() + + assertEquals(listOf(CHILD), host.attempts) + assertEquals(1, surfaced.size) + assertIs(surfaced.single()) + assertEquals(0, coordinator.followState().subscribed) + coordinator.endSession() + } + + /** + * Nothing keeps pushing once the session is gone: a switch closes every child + * stream, and so does closing the session outright. + */ + @Test + fun switchingOrEndingTheSessionClosesEveryChildStream() = runTest { + val host = FakeChildStreams() + val coordinator = SubagentBranchCoordinator(backgroundScope, host, publish = {}, onError = { _, _ -> }) + coordinator.beginSession(PARENT) + val parent = listOf(parentMessage(launchItem(OWNER, CHILD), launchItem("tool-owner-2", "s-child-2"))) + + coordinator.decorate(parent) + runCurrent() + assertEquals(2, host.openCount) + + coordinator.beginSession("s-other") + runCurrent() + assertEquals(0, host.openCount) + assertEquals(listOf(CHILD, "s-child-2"), host.closed) + assertEquals(0, coordinator.followState().branches) + + coordinator.beginSession(PARENT) + coordinator.decorate(parent) + runCurrent() + assertEquals(2, host.openCount) + coordinator.endSession() + runCurrent() + assertEquals(0, host.openCount) + } + + /** + * The cap bounds how many children are followed at once, and a card that is + * opened is always served: a settled worker frees its slot to the child that + * was waiting, and an open card can re-attach a branch the cap released. + */ + @Test + fun theConcurrencyCapIsRespectedAndAnOpenedCardCanAlwaysReattach() = runTest { + val host = FakeChildStreams() + val coordinator = SubagentBranchCoordinator(backgroundScope, host, publish = {}, onError = { _, _ -> }) + coordinator.beginSession(PARENT) + val owners = (0 until MAX_CONCURRENT_CHILD_STREAMS + 1).map { index -> "owner-$index" } + val children = owners.map { owner -> "child-$owner" } + val parent = listOf(parentMessage(*owners.indices.map { launchItem(owners[it], children[it]) }.toTypedArray())) + + coordinator.decorate(parent) + runCurrent() + assertEquals(MAX_CONCURRENT_CHILD_STREAMS, host.openCount) + assertEquals(1, coordinator.followState().waiting) + // The launch that had been quiet longest gave up its slot for the newest one, + // so the branch waiting for a slot is the first child, not the last. + assertTrue(children[0] !in host.opened) + assertTrue(children.last() in host.opened) + + // A worker that reported costs nothing to stop following, and the child that + // was waiting takes the slot it frees. + host.push(children[1], richRecord(children[1], "t-child", 0, 1, "completed", "done")) + runCurrent() + assertEquals(MAX_CONCURRENT_CHILD_STREAMS, host.openCount) + assertEquals(0, coordinator.followState().waiting) + assertTrue(children[0] in host.opened) + + // The released branch keeps its steps and is followed again when its card is + // opened, which is what an expansion asks for. + val attemptsBefore = host.attempts.size + coordinator.ensureAttached(owners[1]) + runCurrent() + assertEquals(attemptsBefore + 1, host.attempts.size) + assertEquals(MAX_CONCURRENT_CHILD_STREAMS, host.openCount) + coordinator.endSession() + } + + /** + * A worker that is still running when the cap takes its slot keeps the steps it + * published, and opening its card follows it again. + * + * This is the case the cap cannot avoid: past the bound some card has to be + * released, and a running worker is the one that cannot be dropped silently. The + * branch keeps its prefix, the card stays readable, and the reader's own gesture — + * opening it — is what resumes the stream and brings the branch up to date. + */ + @Test + fun aRunningChildEvictedByTheCapIsFollowedAgainWhenItsCardIsOpened() = runTest { + val host = FakeSubagentHost() + val reported = mutableListOf() + val coordinator = SubagentBranchCoordinator(backgroundScope, host, publish = {}, onError = { _, error -> reported += error }) + coordinator.beginSession(PARENT) + val owners = (0 until MAX_CONCURRENT_CHILD_STREAMS + 1).map { index -> "owner-$index" } + val children = owners.map { owner -> "child-$owner" } + val first = listOf(parentMessage(launchItem(owners[0], children[0]))) + // Two steps are two items, and `richRecord` names an item after its turn, so a + // second step is a second turn — which is also how a continued worker reads. + val stepOne = richRecord(children[0], "t-child-1", 0, 1, "inprogress", "step one") + val stepTwo = richRecord(children[0], "t-child-2", 1, 2, "inprogress", "step two") + + host.replay(children[0], stepOne) + coordinator.decorate(first) + runCurrent() + assertEquals(1, coordinator.decorate(first).single().items!!.single().subItems?.size) + + // Every remaining launch takes a slot, and the cap takes this worker's — the + // oldest still running one. + val all = listOf(parentMessage(*owners.indices.map { launchItem(owners[it], children[it]) }.toTypedArray())) + coordinator.decorate(all) + runCurrent() + assertEquals(MAX_CONCURRENT_CHILD_STREAMS, host.openCount) + val attemptsBefore = host.attempts.count { it == children[0] } + + // The worker kept working while its slot was gone; a host replays that page. + host.replay(children[0], stepOne, stepTwo) + coordinator.ensureAttached(owners[0]) + runCurrent() + + assertEquals(attemptsBefore + 1, host.attempts.count { it == children[0] }) + assertEquals(emptyList(), reported) + val steps = coordinator.decorate(all).single().items!!.first().subItems?.map { it.content } + assertEquals(listOf("step one", "step two"), steps) + assertEquals(MAX_CONCURRENT_CHILD_STREAMS, host.openCount) + coordinator.endSession() + } + + /** + * A host that starts publishing a worker's process inside its own stream is + * honoured from that moment on. + * + * The older shape needs no child stream: those marked records are already grouped + * under the launch item by the transcript itself. Following the child as well + * would draw every step twice, so the stream is released and the item stops being + * rewritten. + */ + @Test + fun aHostThatStartsPublishingMarkedSiblingsStopsBeingFollowed() = runTest { + val host = FakeChildStreams() + val coordinator = SubagentBranchCoordinator(backgroundScope, host, publish = {}, onError = { _, _ -> }) + coordinator.beginSession(PARENT) + val named = listOf(parentMessage(launchItem(OWNER, CHILD, status = "completed"))) + + coordinator.decorate(named) + runCurrent() + host.push(CHILD, richRecord(CHILD, "t-child", 0, 1, "inprogress", "step one")) + runCurrent() + assertEquals( + listOf("step one"), + coordinator.decorate(named).single().items!!.single().subItems?.map { it.content }, + ) + assertEquals(1, host.openCount) + + // The same host now carries that worker's process in its own stream, marked. + val marked = listOf( + parentMessage( + launchItem(OWNER, CHILD, status = "completed"), + ChatMessageItemResponse(type = "text", content = "step one", isSubagent = true), + ), + ) + val takenOver = coordinator.decorate(marked).single().items!!.first() + runCurrent() + + assertNull(takenOver.subItems) + assertEquals("completed", takenOver.tool?.status) + assertEquals(0, host.openCount) + assertTrue(CHILD in host.closed) + assertEquals(0, coordinator.followState().branches) + coordinator.endSession() + } + + /** + * A backgrounded app follows nothing: the branch keeps what it published, and + * coming back re-opens the streams that were released. + */ + @Test + fun leavingTheForegroundReleasesChildStreamsAndReturningReattachesThem() = runTest { + val host = FakeChildStreams() + val coordinator = SubagentBranchCoordinator(backgroundScope, host, publish = {}, onError = { _, _ -> }) + coordinator.beginSession(PARENT) + val parent = listOf(parentMessage(launchItem(OWNER, CHILD))) + + coordinator.decorate(parent) + runCurrent() + assertEquals(1, host.openCount) + + coordinator.setForeground(false) + runCurrent() + assertEquals(0, host.openCount) + assertEquals(1, coordinator.followState().branches) + + coordinator.setForeground(true) + runCurrent() + assertEquals(1, host.openCount) + coordinator.endSession() + } + + /** + * Opening a conversation follows the child its own transcript names, and the + * decorated transcript is what the store publishes — the UI needs no new state. + */ + @Test + fun openingASessionFollowsTheChildItsTranscriptNames() = runTest { + val host = FakeSubagentHost().apply { replay(PARENT, launchEvent(PARENT, "t-parent", OWNER, CHILD, revision = 1)) } + val store = RemoteSessionStore(this, host, followSubagentBranches = true) + + store.dispatch(RemoteSessionIntent.Open(PARENT)) + runCurrent() + assertEquals(listOf(PARENT, CHILD), host.attempts) + + host.push(CHILD, richRecord(CHILD, "t-child", 0, 1, "inprogress", "step one")) + runCurrent() + + val owner = assertIs(store.state.value).timeline!! + .persistedMessages.flatMap { it.items.orEmpty() }.single { it.tool?.id == OWNER } + assertEquals(listOf("step one"), owner.subItems?.map { it.content }) + assertEquals("running", owner.tool?.status) + store.stop() + } + + /** + * The transcript this coordinator publishes is the card the surface draws. + * + * The coordinator attaches a child's items to the launch item; what a reader + * actually sees is what the presentation layer makes of that transcript. This + * walks the same handover the app does — timeline state, then rows, then + * blocks — so the branch is known to become a subagent block that still names + * the launch item's tool call (the key a card re-attaches by), still reads as + * the worker's lifecycle, and still carries the worker's steps. The device + * test then only has to prove the drawing. + */ + @Test + fun theDecoratedTranscriptBecomesTheSubagentCardTheSurfaceDraws() = runTest { + val host = FakeChildStreams() + val coordinator = SubagentBranchCoordinator(backgroundScope, host, publish = {}, onError = { _, _ -> }) + coordinator.beginSession(PARENT) + // The launch carries the task's own description, which is what the card is + // titled with — the same string the device test clicks to open it. + val launched = launchItem(OWNER, CHILD, status = "completed") + val parent = listOf(parentMessage(launched.copy(tool = launched.tool?.copy(inputPreview = "Audit the parser")))) + + coordinator.decorate(parent) + runCurrent() + host.push(CHILD, richRecord(CHILD, "t-child", 0, 1, "inprogress", "step one")) + runCurrent() + + val timeline = ChatTimelineStore() + timeline.reset(PARENT) + timeline.setPersistedMessages(coordinator.decorate(parent)) + val card = timeline.snapshot().conversationRows() + .flatMap { it.blocks } + .filterIsInstance() + .single() + + assertEquals("Audit the parser", card.title) + assertEquals(OWNER, card.ownerToolCallId) + assertEquals("running", card.status) + assertEquals(listOf("step one"), card.children.filterIsInstance().map { it.text }) + coordinator.endSession() + } + + /** + * A conversation that is no longer on screen does not keep reading its workers, + * and neither does a closed store. + */ + @Test + fun switchingOrClosingTheConversationStopsEveryChildStream() = runTest { + val host = FakeSubagentHost().apply { replay(PARENT, launchEvent(PARENT, "t-parent", OWNER, CHILD, revision = 1)) } + val store = RemoteSessionStore(this, host, followSubagentBranches = true) + + store.dispatch(RemoteSessionIntent.Open(PARENT)) + runCurrent() + assertEquals(2, host.openCount) + + store.dispatch(RemoteSessionIntent.Open("s-other")) + runCurrent() + assertEquals(1, host.openCount) + assertTrue(CHILD in host.closed) + + store.dispatch(RemoteSessionIntent.Open(PARENT)) + runCurrent() + assertEquals(2, host.openCount) + store.stop() + runCurrent() + assertEquals(0, host.openCount) + } + + /** + * A host that cannot serve one worker's stream is not a host that cannot serve + * this conversation. + * + * The parent transcript is the host's own answer for the session the reader + * opened, so it stays on screen, the launch item keeps the status the host gave + * it, no lifecycle is invented from a stream that said nothing, and the failure + * is reported once on the card that owns it. Reading is abandoned for that host, + * so nothing else is opened and no command it would fail to parse is sent. + */ + @Test + fun anUnsupportedChildStreamLeavesTheConversationReadableAndIsReportedOnItsCard() = runTest { + val host = FakeSubagentHost().apply { replay(PARENT, launchEvent(PARENT, "t-parent", OWNER, CHILD, revision = 1)) } + val store = RemoteSessionStore(this, host, followSubagentBranches = true) + + store.dispatch(RemoteSessionIntent.Open(PARENT)) + runCurrent() + // The conversation plus the child its transcript names. + assertEquals(2, host.openCount) + val attemptsBeforeFailure = host.attempts.size + + host.fail(CHILD, HostStreamUnsupportedException()) + runCurrent() + + // The session is still the readable conversation it was, not a failure page. + val ready = assertIs(store.state.value) + assertEquals(ConnectionPhase.CONNECTED, store.connectionPhase.value) + assertEquals( + mapOf(OWNER to RemoteSessionFailureReason.HOST_STREAM_UNSUPPORTED), + ready.subagentFailures, + ) + val owner = ready.timeline!!.persistedMessages.flatMap { it.items.orEmpty() }.single { it.tool?.id == OWNER } + // The launch item's own status stands: the stream said nothing about the worker. + assertEquals("completed", owner.tool?.status) + assertNull(owner.subItems) + + // Read once, and never again — not for this child, and not for one the parent + // names afterwards, because following children is off for this host. + assertEquals(1, host.attempts.count { it == CHILD }) + host.push(PARENT, launchEvent(PARENT, "t-parent", "tool-owner-2", "s-child-2", revision = 2)) + runCurrent() + assertEquals(attemptsBeforeFailure, host.attempts.size) + assertEquals(1, host.openCount) + // Reported once: a second failure on the same card adds nothing. + host.fail(CHILD, HostStreamUnsupportedException()) + runCurrent() + assertEquals( + mapOf(OWNER to RemoteSessionFailureReason.HOST_STREAM_UNSUPPORTED), + assertIs(store.state.value).subagentFailures, + ) + store.stop() + } + + /** + * A child's failure belongs to the session that spawned it. + * + * The coordinator of a conversation that is no longer on screen is detached with + * its streams, so an error arriving late describes something nobody is reading and + * must not mark the conversation now on screen — which is a different session — + * as failed. + */ + @Test + fun aLateChildErrorFromAClosedSessionLeavesTheNewSessionUntouched() = runTest { + val host = FakeSubagentHost().apply { + replay(PARENT, launchEvent(PARENT, "t-parent", OWNER, CHILD, revision = 1)) + replay("s-other", richRecord("s-other", "t-other", 0, 1, "completed", "unrelated")) + } + val store = RemoteSessionStore(this, host, followSubagentBranches = true) + + store.dispatch(RemoteSessionIntent.Open(PARENT)) + runCurrent() + assertEquals(listOf(PARENT, CHILD), host.attempts) + + store.dispatch(RemoteSessionIntent.Open("s-other")) + runCurrent() + val before = assertIs(store.state.value) + + // The old session's child stream fails after the reader moved on. + host.fail(CHILD, IllegalStateException("stream dropped")) + runCurrent() + + val after = assertIs(store.state.value) + assertEquals("s-other", after.selectedSessionId) + assertTrue(after.subagentFailures.isEmpty()) + assertEquals(before.timeline, after.timeline) + assertEquals(ConnectionPhase.CONNECTED, store.connectionPhase.value) + store.stop() + } + + /** + * A child transcript is drawn from the child's stream on demand, so it is never + * written into this device's copy of the parent's transcript. + */ + @Test + fun aChildsStepsAreNeverStoredInTheParentsTranscript() { + val child = ChatMessageItemResponse(type = "text", content = "step one") + val item = launchItem(OWNER, CHILD).copy(subItems = listOf(child)) + + val stored = withoutChildSteps(item) + + assertEquals(CHILD, stored.tool?.subagentSessionId) + assertNull(stored.subItems) + // An ordinary tool row is stored exactly as the host sent it. + assertEquals(launchItem("tool-read", ""), withoutChildSteps(launchItem("tool-read", ""))) + } + + /** + * A card stops reporting its worker once that worker's stream is read again — and + * only then. + * + * The report describes an attempt, not the worker, so it is dropped when a read + * really happened. The case that matters is the one the cap documents: the branch + * lost its stream (here because the phone left the foreground), the card said it + * could not be read, and opening the card follows the worker again. + */ + @Test + fun aCardStopsReportingItsWorkerOnceTheStreamIsReadAgain() = runTest { + val host = FakeSubagentHost().apply { replay(PARENT, launchEvent(PARENT, "t-parent", OWNER, CHILD, revision = 1)) } + val store = RemoteSessionStore(this, host, followSubagentBranches = true) + + store.dispatch(RemoteSessionIntent.Open(PARENT)) + runCurrent() + assertEquals(1, host.attempts.count { it == CHILD }) + + host.fail(CHILD, IllegalStateException("stream dropped")) + runCurrent() + assertEquals( + mapOf(OWNER to RemoteSessionFailureReason.TRANSPORT), + assertIs(store.state.value).subagentFailures, + ) + + // The branch gives up its lane, and its card says why it has nothing to show. + store.dispatch(RemoteSessionIntent.SetForeground(false)) + runCurrent() + + store.dispatch(RemoteSessionIntent.EnsureSubagentAttached(OWNER)) + runCurrent() + + // The read really happened, so the report was dropped with it. + assertEquals(2, host.attempts.count { it == CHILD }) + assertTrue(assertIs(store.state.value).subagentFailures.isEmpty()) + store.stop() + } + + /** + * The other half of the guard above: no read, no clearing. + * + * A host that cannot serve child streams is never read again, so the one sentence + * that explains the card — this desktop cannot share worker steps — has to survive + * the only gesture that could take it away. + */ + @Test + fun openingACardWhoseHostCannotServeChildStreamsKeepsTheExplanationItHas() = runTest { + val host = FakeSubagentHost().apply { replay(PARENT, launchEvent(PARENT, "t-parent", OWNER, CHILD, revision = 1)) } + val store = RemoteSessionStore(this, host, followSubagentBranches = true) + + store.dispatch(RemoteSessionIntent.Open(PARENT)) + runCurrent() + host.fail(CHILD, HostStreamUnsupportedException()) + runCurrent() + val attempts = host.attempts.size + assertEquals( + mapOf(OWNER to RemoteSessionFailureReason.HOST_STREAM_UNSUPPORTED), + assertIs(store.state.value).subagentFailures, + ) + + store.dispatch(RemoteSessionIntent.EnsureSubagentAttached(OWNER)) + runCurrent() + + // Nothing was read, so nothing about the card changed. + assertEquals(attempts, host.attempts.size) + assertEquals( + mapOf(OWNER to RemoteSessionFailureReason.HOST_STREAM_UNSUPPORTED), + assertIs(store.state.value).subagentFailures, + ) + store.stop() + } + + /** + * A card that is already drawing its worker's steps has nothing left to report. + * + * A stream can die after it published a turn. The steps stay — they are the + * worker's own work — and the failure recorded for that card is dropped as soon as + * the transcript carrying those steps is rendered again: the card is no longer in + * the state the report described. + */ + @Test + fun aCardThatIsShowingItsWorkersStepsStopsReportingAFailure() = runTest { + val host = FakeSubagentHost().apply { replay(PARENT, launchEvent(PARENT, "t-parent", OWNER, CHILD, revision = 1)) } + val store = RemoteSessionStore(this, host, followSubagentBranches = true) + + store.dispatch(RemoteSessionIntent.Open(PARENT)) + runCurrent() + host.push(CHILD, richRecord(CHILD, "t-child", 0, 1, "inprogress", "step one")) + runCurrent() + host.fail(CHILD, IllegalStateException("stream dropped")) + runCurrent() + assertEquals( + mapOf(OWNER to RemoteSessionFailureReason.TRANSPORT), + assertIs(store.state.value).subagentFailures, + ) + + // The parent transcript is rendered again while this card already has steps. + host.push(PARENT, launchEvent(PARENT, "t-parent", "tool-owner-2", "s-child-2", revision = 2)) + runCurrent() + + val ready = assertIs(store.state.value) + assertTrue(ready.subagentFailures.isEmpty()) + val owner = ready.timeline!!.persistedMessages.flatMap { it.items.orEmpty() }.single { it.tool?.id == OWNER } + // The report is gone; the work it reported on is not. + assertEquals(listOf("step one"), owner.subItems?.map { it.content }) + store.stop() + } + + /** + * Following delegated children is opted in by the platform, not assumed. + * + * A client that has not wired the card's expansion and its per-card failure surface + * would otherwise open a child stream it can never re-open and record a failure it + * can never show. With the flag off — which is every platform but Android — the + * store publishes the host's own transcript item for item and nothing about a + * delegation is attempted. + */ + @Test + fun subagentBranchesAreOffByDefaultAndLeaveTheTranscriptUntouched() = runTest { + val host = FakeSubagentHost().apply { + replay(PARENT, launchEvent(PARENT, "t-parent", OWNER, CHILD, revision = 1)) + // A child stream that would fail loudly if it were ever asked for. + failSubscribe = { if (it == CHILD) IllegalStateException("stream dropped") else null } + } + val store = RemoteSessionStore(this, host) + + store.dispatch(RemoteSessionIntent.Open(PARENT)) + runCurrent() + + // (i) The parent's own stream, and no attempt at the child it names. + assertEquals(listOf(PARENT), host.attempts) + + // (ii) The launch item is the host's item: the child id it named is kept, no + // child item is attached, and the status is the one the host sent. + val published = assertIs(store.state.value).timeline!! + .persistedMessages.flatMap { it.items.orEmpty() }.single { it.tool?.id == OWNER } + val replica = SessionRecordReplica(PARENT) + replica.apply(launchEvent(PARENT, "t-parent", OWNER, CHILD, revision = 1).getValue("payload").jsonObject) + val hostOwnItem = replica.messages().flatMap { it.items.orEmpty() }.single { it.tool?.id == OWNER } + assertEquals(hostOwnItem, published) + assertEquals(CHILD, published.tool?.subagentSessionId) + assertEquals("completed", published.tool?.status) + assertNull(published.subItems) + + // (iii) The card's expansion is inert: no stream, and not even a new state + // object for the surface to redraw. + val before = store.state.value + val attempts = host.attempts.size + store.dispatch(RemoteSessionIntent.EnsureSubagentAttached(OWNER)) + runCurrent() + assertEquals(attempts, host.attempts.size) + assertSame(before, store.state.value) + + // (iv) Nothing can be reported about a stream that was never opened. + assertTrue(assertIs(store.state.value).subagentFailures.isEmpty()) + store.stop() + } + + // --- fixtures ------------------------------------------------------------- + + private fun parentMessage(vararg items: ChatMessageItemResponse) = ChatMessage( + id = "m-parent", role = "assistant", text = "", status = "completed", renderVersion = null, + turnId = "t-parent", detail = null, timestamp = null, thinking = null, tools = null, + items = items.toList(), images = null, error = null, + ) + + private fun launchItem(ownerToolCallId: String, childSessionId: String, status: String = "running") = + ChatMessageItemResponse( + type = "tool", + tool = RemoteToolStatusResponse( + id = ownerToolCallId, + name = "task", + status = status, + subagentSessionId = childSessionId.takeIf { it.isNotEmpty() }, + ), + isSubagent = childSessionId.isNotEmpty(), + ) + + /** A parent record for the tool item that launched [child]. */ + private fun launchEvent( + session: String, + turn: String, + ownerToolCallId: String, + child: String, + revision: Long, + ): JsonObject = buildJsonObject { + put("session_id", session) + put("event", "session-record") + put("payload", buildJsonObject { + put("sessionId", session) + put("id", "item/$ownerToolCallId") + put("revision", revision) + put("turn", buildJsonObject { + put("sessionId", session); put("turnId", turn); put("turnIndex", 0); put("status", "completed") + put("userMessage", buildJsonObject { put("id", "${turn}_user"); put("content", "delegate this"); put("timestamp", 1) }) + }) + put("round", buildJsonObject { put("id", "$turn-round"); put("turnId", turn); put("roundIndex", 0) }) + put("item", buildJsonObject { + put("type", "tool") + put("data", buildJsonObject { + put("id", ownerToolCallId); put("orderIndex", 0); put("timestamp", 1) + put("toolName", "task"); put("status", "completed") + put("toolCall", buildJsonObject { put("id", ownerToolCallId) }) + put("subagentSessionId", child) + }) + }) + }) + } +} + +/** + * Host streams for one parent session and the children it names. + * + * Each session is its own flow, so a child's records can only arrive on the + * subscription the coordinator opened for that child — which is the grouping rule + * being asserted. + */ +private class FakeChildStreams : RemoteSessionStreamTransport { + /** Every `subscribe` call, including one that fails before returning a flow. */ + val attempts = mutableListOf() + /** Subscriptions that started collecting, in order. */ + val opened = mutableListOf() + /** Subscriptions whose flow ended, in order. */ + val closed = mutableListOf() + val openCount: Int get() = opened.size - closed.size + + private val channels = mutableMapOf>() + private val caughtUp = mutableMapOf Unit>() + private val errors = mutableMapOf Unit>() + + /** When set, opening a stream for that session fails instead. */ + var failSubscribe: ((String) -> Throwable?)? = null + + fun push(sessionId: String, event: JsonObject) { + channels.getOrPut(sessionId) { MutableSharedFlow(extraBufferCapacity = 16) }.tryEmit(event) + } + + fun catchUp(sessionId: String) { + caughtUp[sessionId]?.invoke() + } + + fun fail(sessionId: String, error: Throwable) { + errors[sessionId]?.invoke(error) + } + + override suspend fun subscribe( + sessionId: String, + onError: (Throwable) -> Unit, + onCaughtUp: () -> Unit, + ): Flow { + attempts += sessionId + failSubscribe?.invoke(sessionId)?.let { throw it } + return flow { + opened += sessionId + errors[sessionId] = onError + caughtUp[sessionId] = onCaughtUp + try { + onCaughtUp() + channels.getOrPut(sessionId) { MutableSharedFlow(extraBufferCapacity = 16) }.collect { emit(it) } + } finally { + closed += sessionId + } + } + } +} + +/** + * A simulated host for one open conversation: the parent stream and the children + * it names are served from the same device, as a real host does. + */ +private class FakeSubagentHost : RemoteCommandTransport, RemoteSessionStreamTransport { + val attempts = mutableListOf() + val opened = mutableListOf() + val closed = mutableListOf() + val openCount: Int get() = opened.size - closed.size + + private val replayEvents = mutableMapOf>() + private val channels = mutableMapOf>() + private val caughtUp = mutableMapOf Unit>() + private val errors = mutableMapOf Unit>() + + var failSubscribe: ((String) -> Throwable?)? = null + + fun replay(sessionId: String, vararg events: JsonObject) { + replayEvents[sessionId] = events.toList() + } + + fun push(sessionId: String, event: JsonObject) { + channels.getOrPut(sessionId) { MutableSharedFlow(extraBufferCapacity = 16) }.tryEmit(event) + } + + fun fail(sessionId: String, error: Throwable) { + errors[sessionId]?.invoke(error) + } + + override suspend fun loadOlder(sessionId: String) = Unit + + override suspend fun subscribe( + sessionId: String, + onError: (Throwable) -> Unit, + onCaughtUp: () -> Unit, + ): Flow { + attempts += sessionId + failSubscribe?.invoke(sessionId)?.let { throw it } + return flow { + opened += sessionId + errors[sessionId] = onError + caughtUp[sessionId] = onCaughtUp + try { + replayEvents[sessionId].orEmpty().forEach { emit(it) } + onCaughtUp() + channels.getOrPut(sessionId) { MutableSharedFlow(extraBufferCapacity = 16) }.collect { emit(it) } + } finally { + closed += sessionId + } + } + } + + override suspend fun send( + deserializer: DeserializationStrategy, + command: RemoteCommand, + timeoutMs: Long, + ): T { + val json = when (command.cmd) { + "get_workspace_info" -> """{"resp":"ok","path":"/repo","capabilities":["host_stream_v1"]}""" + "get_permission_mode" -> """{"resp":"ok","mode":"ask"}""" + "get_model_catalog" -> """{"resp":"ok","catalog":{"version":0,"models":[],"default_models":{}}}""" + "list_sessions" -> """{"resp":"ok","sessions":[],"has_more":false}""" + "get_session_messages" -> """{"resp":"ok","messages":[],"has_more":false}""" + "poll_session" -> """{"resp":"ok","version":1,"changed":false,"session_state":"idle"}""" + else -> """{"resp":"ok"}""" + } + return RelayJson.decodeFromString(deserializer, json) + } +} diff --git a/src/apps/mobile/shared/core-protocol/src/commonMain/kotlin/com/openbitfun/mobile/core/protocol/MessageDtos.kt b/src/apps/mobile/shared/core-protocol/src/commonMain/kotlin/com/openbitfun/mobile/core/protocol/MessageDtos.kt index dd4ca1a6f5..03bbff98ec 100644 --- a/src/apps/mobile/shared/core-protocol/src/commonMain/kotlin/com/openbitfun/mobile/core/protocol/MessageDtos.kt +++ b/src/apps/mobile/shared/core-protocol/src/commonMain/kotlin/com/openbitfun/mobile/core/protocol/MessageDtos.kt @@ -20,6 +20,17 @@ public data class RemoteToolStatusResponse( @SerialName("error_preview") val errorPreview: String? = null, @SerialName("exit_code") val exitCode: Int? = null, @SerialName("plan") val plan: JsonElement? = null, + /** + * The hidden session a delegation tool spawned, when the host named one. + * + * A host keeps a spawned child agent in its own session, so the child's work + * is in that session's stream and nowhere in the parent transcript. The + * launching item is the only place the id is written, which is why it has to + * survive mapping: it is the key every reader needs to open the child. + * Optional — an older host names no child and keeps the plain tool row it + * has always sent. + */ + @SerialName("subagent_session_id") val subagentSessionId: String? = null, ) @Serializable