From 9af99d9e4d27ad2ff3328e23b1e4d67e116ac334 Mon Sep 17 00:00:00 2001 From: Ross Tomsic Date: Thu, 1 Oct 2026 15:11:25 -0400 Subject: [PATCH 1/2] Offer explicit fresh-shell recovery after session loss --- .../dev/threadline/SessionRecoveryTest.kt | 292 ++++++++++++++++++ .../AndroidStructuredShellIntegrationTest.kt | 164 ++++++++++ .../main/java/dev/threadline/MainActivity.kt | 97 +++++- .../java/dev/threadline/TranscriptScreen.kt | 97 +++++- .../core/diagnostics/DiagnosticReport.kt | 1 + .../core/model/SessionErrorPresentation.kt | 6 + .../threadline/core/model/SessionModels.kt | 18 ++ .../threadline/core/session/SessionManager.kt | 178 ++++++----- .../core/ssh/ConnectBotSshClientAdapter.kt | 55 +++- .../threadline/core/ssh/SshClientAdapter.kt | 2 + .../threadline/service/SshSessionService.kt | 1 + .../core/diagnostics/DiagnosticReportTest.kt | 1 + .../model/SessionErrorPresentationTest.kt | 2 + .../core/session/SessionManagerIoTest.kt | 151 ++++++++- docs/STATUS.md | 10 +- fixtures/openssh/README.md | 5 + gradle/libs.versions.toml | 2 +- gradle/verification-metadata.xml | 133 ++++++++ 18 files changed, 1115 insertions(+), 100 deletions(-) create mode 100644 app/src/androidTest/java/dev/threadline/SessionRecoveryTest.kt diff --git a/app/src/androidTest/java/dev/threadline/SessionRecoveryTest.kt b/app/src/androidTest/java/dev/threadline/SessionRecoveryTest.kt new file mode 100644 index 0000000..192f099 --- /dev/null +++ b/app/src/androidTest/java/dev/threadline/SessionRecoveryTest.kt @@ -0,0 +1,292 @@ +package dev.threadline + +import androidx.compose.material3.MaterialTheme +import androidx.compose.runtime.CompositionLocalProvider +import androidx.compose.runtime.mutableStateOf +import androidx.compose.ui.platform.LocalDensity +import androidx.compose.ui.test.assertIsDisplayed +import androidx.compose.ui.test.assertIsSelected +import androidx.compose.ui.test.junit4.v2.createComposeRule +import androidx.compose.ui.test.onNodeWithTag +import androidx.compose.ui.test.onNodeWithText +import androidx.compose.ui.test.performClick +import androidx.compose.ui.test.performScrollTo +import androidx.compose.ui.test.performTextInput +import androidx.compose.ui.unit.Density +import dev.threadline.core.model.ConnectionRequest +import dev.threadline.core.model.ConnectionTarget +import dev.threadline.core.model.HostEndpoint +import dev.threadline.core.model.HostProfile +import dev.threadline.core.model.SessionCredential +import dev.threadline.core.model.SessionError +import dev.threadline.core.transcript.CommandTranscript +import dev.threadline.core.transcript.CommandTranscriptState +import dev.threadline.core.shell.CommandId +import dev.threadline.core.shell.ShellLifecycleEvent +import dev.threadline.core.shell.StructuredShellState +import dev.threadline.core.shell.StructuredShellUnavailableReason +import dev.threadline.core.shell.CommandSubmissionResult +import dev.threadline.core.shell.CommandSubmissionRejection +import dev.threadline.data.identity.IdentityAuthenticationMethod +import dev.threadline.data.identity.SshIdentity +import dev.threadline.data.key.ImportedPrivateKeyMetadata +import org.junit.Assert.assertEquals +import org.junit.Assert.assertNull +import org.junit.Rule +import org.junit.Test + +class SessionRecoveryTest { + @get:Rule + val compose = createComposeRule() + + @Test + fun shellEndKeepsOutputReadOnlyAndRequiresAnExplicitFreshShellAtLargeFontScale() { + val transcript = CommandTranscript() + transcript.commandSubmitted(CommandId("ended-command"), "exit 7", directoryAtStart = "/tmp") + transcript.lifecycle(ShellLifecycleEvent.CommandOutputStarted(CommandId("ended-command"))) + transcript.consumeOutput("partial output\n".encodeToByteArray()) + transcript.sessionDisconnected() + val snapshot = transcript.state.value + var reconnects = 0 + compose.setContent { + val density = LocalDensity.current + CompositionLocalProvider(LocalDensity provides Density(density.density, 2f)) { + MaterialTheme { + EndedSessionScreen( + target = target(), + error = SessionError.ShellEnded(7), + transcript = snapshot, + onReconnect = { reconnects += 1 }, + onOpenHome = {}, + ) + } + } + } + compose.onNodeWithText("Shell ended").assertExists() + compose.onNodeWithText("The remote shell ended with exit status 7.").assertExists() + compose.onNodeWithText(FreshShellNotice).assertExists() + compose.onNodeWithTag(TranscriptTags.output("ended-command")).performScrollTo() + compose.onNodeWithText("partial output\n").assertIsDisplayed() + compose.onNodeWithTag(TranscriptTags.COMPOSER).assertDoesNotExist() + compose.onNodeWithText("Rerun").assertDoesNotExist() + compose.onNodeWithTag(TranscriptTags.cardActions("ended-command")).performScrollTo().performClick() + compose.onNodeWithText("Edit command").assertDoesNotExist() + compose.onNodeWithText("Copy output").performClick() + compose.runOnIdle { assertEquals(0, reconnects) } + compose.onNodeWithTag(TranscriptTags.RECONNECT).performScrollTo().performClick() + compose.runOnIdle { assertEquals(1, reconnects) } + } + + @Test + fun connectionLossShowsNetworkRecoveryAndKeepsHomeAvailable() { + var homeOpened = false + compose.setContent { + MaterialTheme { + EndedSessionScreen( + target = target(), + error = SessionError.ConnectionLost, + transcript = CommandTranscriptState(), + onReconnect = {}, + onOpenHome = { homeOpened = true }, + ) + } + } + compose.onNodeWithText("Connection lost").assertExists() + compose.onNodeWithText("Shell ended").assertDoesNotExist() + compose.onNodeWithText("Check the device network, VPN, and server status before reconnecting.") + .assertExists() + compose.onNodeWithText("Home").performClick() + compose.runOnIdle { assertEquals(true, homeOpened) } + } + + @Test + fun integrationDowngradeKeepsRawTerminalAccessWithoutAnEndedSessionPrompt() { + compose.setContent { + MaterialTheme { + ConnectedSessionScreen( + displayName = "Downgrade fixture", + structuredShell = StructuredShellState.Unavailable( + StructuredShellUnavailableReason.BOOTSTRAP_TIMED_OUT, + ), + transcript = CommandTranscriptState(), + onSubmit = { CommandSubmissionResult.Rejected(CommandSubmissionRejection.NOT_READY) }, + onControlC = {}, + onDisconnect = {}, + rawTerminal = { modifier, _ -> + androidx.compose.material3.Text("Live raw terminal", modifier) + }, + ) + } + } + compose.onNodeWithText("Transcript unavailable ยท terminal available").assertExists() + compose.onNodeWithText("Live raw terminal").assertIsDisplayed() + compose.onNodeWithTag(TranscriptTags.ENDED_SESSION).assertDoesNotExist() + compose.onNodeWithTag(TranscriptTags.RECONNECT).assertDoesNotExist() + } + + @Test + fun freshKeyConnectionKeepsTheOriginalEndpointUsernameAndKeyChoice() { + val draft = mutableStateOf(ConnectionFormDraft.emptyDefaults()) + var loads = 0 + var prepared: ConnectionRequest? = null + compose.setContent { + MaterialTheme { + HostForm( + draft = draft.value, + onDraftChange = { draft.value = it }, + sessionError = null, + reconnectTarget = target(usesPrivateKey = true), + importedPrivateKeys = listOf(key("original-key"), key("replacement-key")), + sshIdentities = listOf(identity( + method = IdentityAuthenticationMethod.IMPORTED_PRIVATE_KEY, + username = "edited-user", + privateKeyId = "replacement-key", + )), + onLoadPrivateKey = { id, _ -> + assertEquals("original-key", id) + loads += 1 + SessionCredential.PrivateKey.from(byteArrayOf(1, 2), null) + }, + onPrepared = { prepared = it; true }, + ) + } + } + compose.onNodeWithText("Complete sign-in").assertExists() + compose.onNodeWithText("chosen-user@fixture.test:2222").assertExists() + compose.onNodeWithTag(ConnectionFormTags.SAVED_KEY_PREFIX + "original-key").assertIsSelected() + compose.runOnIdle { assertEquals(0, loads); assertNull(prepared) } + compose.onNodeWithTag(ConnectionFormTags.CONNECT).performScrollTo().performClick() + compose.waitForIdle() + compose.runOnIdle { + val request = requireNotNull(prepared) + assertEquals(1, loads) + assertEquals(target().profile, request.profile) + assertEquals("chosen-identity", request.identityId) + assertEquals("original-key", request.importedPrivateKeyId) + assertEquals(true, request.ephemeral) + request.credential.clear() + } + } + + @Test + fun freshConnectionKeepsItsKeyChoiceThroughNotificationPermissionRecovery() { + val draft = mutableStateOf(ConnectionFormDraft.emptyDefaults()) + val permission = mutableStateOf(SessionNotificationPermissionState.DENIED) + var prepared: ConnectionRequest? = null + compose.setContent { + MaterialTheme { + HostForm( + draft = draft.value, + onDraftChange = { draft.value = it }, + sessionError = null, + reconnectTarget = target(usesPrivateKey = true), + notificationPermissionState = permission.value, + importedPrivateKeys = listOf(key("original-key")), + onLoadPrivateKey = { id, _ -> + assertEquals("original-key", id) + SessionCredential.PrivateKey.from(byteArrayOf(1, 2), null) + }, + onPrepared = { prepared = it; true }, + ) + } + } + compose.onNodeWithTag(ConnectionFormTags.CONNECT).assertDoesNotExist() + compose.runOnIdle { permission.value = SessionNotificationPermissionState.GRANTED } + compose.onNodeWithTag(ConnectionFormTags.SAVED_KEY_PREFIX + "original-key").assertIsSelected() + compose.onNodeWithTag(ConnectionFormTags.CONNECT).performScrollTo().performClick() + compose.waitForIdle() + compose.runOnIdle { + assertEquals("original-key", requireNotNull(prepared).importedPrivateKeyId) + requireNotNull(prepared).credential.clear() + } + } + + @Test + fun freshSavedPasswordConnectionRequiresANewUnlock() { + val draft = mutableStateOf(ConnectionFormDraft.emptyDefaults()) + var unlocks = 0 + var prepared: ConnectionRequest? = null + compose.setContent { + MaterialTheme { + HostForm( + draft = draft.value, + onDraftChange = { draft.value = it }, + sessionError = null, + reconnectTarget = target(), + sshIdentities = listOf(identity(IdentityAuthenticationMethod.PASSWORD)), + savedPasswordSupported = true, + onLoadSavedSshPassword = { id -> + assertEquals("chosen-identity", id) + unlocks += 1 + SessionCredential.Password.from("fresh-password".toCharArray()) + }, + onPrepared = { prepared = it; true }, + ) + } + } + compose.runOnIdle { assertEquals(0, unlocks); assertNull(prepared) } + compose.onNodeWithTag(ConnectionFormTags.CONNECT).performScrollTo().performClick() + compose.waitForIdle() + compose.runOnIdle { + assertEquals(1, unlocks) + assertEquals(target().profile, requireNotNull(prepared).profile) + requireNotNull(prepared).credential.clear() + } + } + + @Test + fun freshConnectionClearsAnEarlierPasswordAndRequiresNewEntry() { + val draft = mutableStateOf(ConnectionFormDraft.fixtureDefaults()) + val recovery = mutableStateOf(null) + var prepared: ConnectionRequest? = null + compose.setContent { + MaterialTheme { + HostForm( + draft = draft.value, + onDraftChange = { draft.value = it }, + sessionError = null, + reconnectTarget = recovery.value, + onPrepared = { prepared = it; true }, + ) + } + } + compose.onNodeWithTag(ConnectionFormTags.PASSWORD).performScrollTo().performTextInput("old-password") + compose.runOnIdle { recovery.value = target() } + compose.onNodeWithTag(ConnectionFormTags.CONNECT).performScrollTo().performClick() + compose.onNodeWithText("Enter the password.").assertExists() + compose.runOnIdle { assertNull(prepared) } + compose.onNodeWithTag(ConnectionFormTags.PASSWORD).performTextInput("fresh-password") + compose.onNodeWithTag(ConnectionFormTags.CONNECT).performScrollTo().performClick() + compose.waitForIdle() + compose.runOnIdle { + val password = requireNotNull(prepared).credential as SessionCredential.Password + assertEquals("fresh-password", password.characters.concatToString()) + password.clear() + } + } + + private fun target(usesPrivateKey: Boolean = false) = ConnectionTarget( + profile = HostProfile("Recovery fixture", HostEndpoint("fixture.test", 2222), "chosen-user"), + usesPrivateKey = usesPrivateKey, + ephemeral = true, + identityId = "chosen-identity", + importedPrivateKeyId = "original-key".takeIf { usesPrivateKey }, + privateKeyUri = null, + ) + + private fun identity( + method: IdentityAuthenticationMethod, + username: String = "chosen-user", + privateKeyId: String? = null, + ) = SshIdentity( + id = "chosen-identity", label = "Chosen identity", username = username, + authenticationMethod = method, importedPrivateKeyId = privateKeyId, + createdAtMillis = 1, updatedAtMillis = 1, + hasSavedPassword = method == IdentityAuthenticationMethod.PASSWORD, + ) + + private fun key(id: String) = ImportedPrivateKeyMetadata( + id = id, displayName = id, format = "OpenSSH", keyType = "ssh-ed25519", + publicKeyFingerprint = "fixture fingerprint", createdAtMillis = 1, + ) +} diff --git a/app/src/androidTest/java/dev/threadline/core/session/AndroidStructuredShellIntegrationTest.kt b/app/src/androidTest/java/dev/threadline/core/session/AndroidStructuredShellIntegrationTest.kt index 0319808..f4f2521 100644 --- a/app/src/androidTest/java/dev/threadline/core/session/AndroidStructuredShellIntegrationTest.kt +++ b/app/src/androidTest/java/dev/threadline/core/session/AndroidStructuredShellIntegrationTest.kt @@ -9,6 +9,7 @@ import dev.threadline.core.model.HostEndpoint import dev.threadline.core.model.HostKeyDecision import dev.threadline.core.model.HostProfile import dev.threadline.core.model.SessionCredential +import dev.threadline.core.model.SessionError import dev.threadline.core.model.SessionState import dev.threadline.core.model.TerminalSize import dev.threadline.core.shell.CommandId @@ -16,6 +17,7 @@ import dev.threadline.core.shell.CommandSubmissionRejection import dev.threadline.core.shell.CommandSubmissionResult import dev.threadline.core.shell.CompletedCommand import dev.threadline.core.shell.StructuredShellState +import dev.threadline.core.shell.StructuredShellUnavailableReason import dev.threadline.core.ssh.AndroidSshCryptoProvider import dev.threadline.core.ssh.ConnectBotSshClientAdapter import dev.threadline.core.ssh.HostKeyAlgorithmPolicy @@ -36,6 +38,7 @@ import java.io.ByteArrayOutputStream import java.io.File import java.security.KeyStore import java.util.concurrent.CopyOnWriteArrayList +import java.util.concurrent.atomic.AtomicInteger import kotlinx.coroutines.delay import kotlinx.coroutines.flow.filterIsInstance import kotlinx.coroutines.flow.first @@ -50,6 +53,165 @@ import org.junit.runner.RunWith @RunWith(AndroidJUnit4::class) class AndroidStructuredShellIntegrationTest { + @Test + fun shellExitExecStrictFailureAndTransportLossRequireExplicitFreshConnections() = runBlocking { + val arguments = InstrumentationRegistry.getArguments() + val password = arguments.getString(PASSWORD_ARGUMENT) + assumeTrue("No fixture password supplied", !password.isNullOrEmpty()) + val context = InstrumentationRegistry.getInstrumentation().targetContext + val preferences = context.getSharedPreferences("android_recovery_known_hosts", Context.MODE_PRIVATE) + preferences.edit().clear().commit() + val database = Room.inMemoryDatabaseBuilder(context, ThreadlineDatabase::class.java).build() + val history = RoomTranscriptHistoryStore(database.transcriptArchives()) + val adapter = RecordingSshClientAdapter(ConnectBotSshClientAdapter( + HostKeyAlgorithmPolicy.overrideWhenEd25519Unavailable(AndroidSshCryptoProvider.install()), + )) + val ids = AtomicInteger() + val manager = SessionManager( + adapter = adapter, + knownHostStore = RoomKnownHostStore(database.knownHosts(), preferences), + terminal = NoOpTerminal, + transcriptArchiveSink = history, + transcriptSessionIdFactory = { "recovery-session-${ids.incrementAndGet()}" }, + ) + val profile = HostProfile( + "Recovery fixture", + HostEndpoint(arguments.getString(HOST_ARGUMENT) ?: DEFAULT_HOST, + arguments.getString(PORT_ARGUMENT)?.toIntOrNull() ?: DEFAULT_PORT), + arguments.getString(USER_ARGUMENT) ?: DEFAULT_USER, + ) + suspend fun connect(first: Boolean = false) { + val request = ConnectionRequest( + profile, SessionCredential.Password.from(requireNotNull(password).toCharArray()), + identityId = "fixture-identity", + ) + assertTrue(manager.prepareConnection(request)) + assertTrue(manager.connectPrepared()) + if (first) { + withTimeout(CONNECTION_TIMEOUT_MILLIS) { + manager.state.filterIsInstance().first() + } + assertTrue(manager.resolveHostKey(HostKeyDecision.ACCEPT_AND_SAVE)) + } + withTimeout(CONNECTION_TIMEOUT_MILLIS) { + manager.structuredState.filterIsInstance().first() + } + assertTrue(manager.transcriptState.value.turns.isEmpty()) + } + try { + connect(first = true) + val cases = listOf( + "exit 7" to SessionError.ShellEnded(7), + "exec bash -c 'printf exec-output; exit 6'" to SessionError.ShellEnded(6), + "set -e; false" to SessionError.ShellEnded(1), + "kill -KILL \"\$PPID\"" to SessionError.ConnectionLost, + ) + List(5) { + "printf '%131072s' ''; kill -KILL \"\$PPID\"" to SessionError.ConnectionLost + } + cases.forEachIndexed { index, (command, expected) -> + assertSuccessful(manager, + "cd /tmp; export THREADLINE_RECOVERY=old; " + + "alias recovery_alias='printf old'; recovery_function() { printf old; }") + val ended = accepted(manager.submitCommand(command)) + val failure = withTimeout(COMMAND_TIMEOUT_MILLIS) { + manager.state.filterIsInstance().first() + } + assertEquals(expected, failure.error) + assertEquals(profile, manager.connectionTarget.value?.profile) + assertEquals("fixture-identity", manager.connectionTarget.value?.identityId) + assertEquals(CommandStatus.DISCONNECTED, + manager.transcriptState.value.turns.last().status) + assertEquals(ended.commandId, manager.transcriptState.value.turns.last().id) + val sends = adapter.sent.size + val attempts = adapter.attempts.get() + delay(200) + assertEquals(sends, adapter.sent.size) + assertEquals(attempts, adapter.attempts.get()) + connect() + assertEquals(attempts + 1, adapter.attempts.get()) + assertEquals(sends + 1, adapter.sent.size) + assertEquals(index + 2, ids.get()) + val archived = history.load("recovery-session-${index + 1}") + assertEquals(command, archived.turns.last().turn.command) + assertEquals(CommandStatus.DISCONNECTED, archived.turns.last().turn.status) + assertEquals(0, execute(manager, + "test \"\$PWD\" = \"\$HOME\" && test \"\${THREADLINE_RECOVERY-unset}\" = unset && " + + "! alias recovery_alias >/dev/null 2>&1 && ! declare -F recovery_function && " + + "printf 'fresh-state\\n'").exitStatus) + } + } finally { + manager.disconnect() + withTimeout(CONNECTION_TIMEOUT_MILLIS) { manager.state.first { it is SessionState.Disconnected } } + database.close() + preferences.edit().clear().commit() + } + } + + @Test + fun bootstrapDowngradeKeepsTheFixturePtyLive() = runBlocking { + val arguments = InstrumentationRegistry.getArguments() + val password = arguments.getString(PASSWORD_ARGUMENT) + assumeTrue("No fixture password supplied", !password.isNullOrEmpty()) + val context = InstrumentationRegistry.getInstrumentation().targetContext + val preferences = context.getSharedPreferences("android_downgrade_known_hosts", Context.MODE_PRIVATE) + preferences.edit().clear().commit() + val database = Room.inMemoryDatabaseBuilder(context, ThreadlineDatabase::class.java).build() + val delegate = ConnectBotSshClientAdapter( + HostKeyAlgorithmPolicy.overrideWhenEd25519Unavailable(AndroidSshCryptoProvider.install()), + ) + val adapter = object : SshClientAdapter { + override suspend fun connect( + request: ConnectionRequest, + verifier: ServerHostKeyVerifier, + initialSize: TerminalSize, + onStage: (dev.threadline.core.model.ConnectionStage) -> Unit, + ): LiveSshSession { + val session = delegate.connect(request, verifier, initialSize, onStage) + var firstSend = true + return object : LiveSshSession by session { + override suspend fun send(bytes: ByteArray) { + if (firstSend) { + firstSend = false + session.send("printf 'raw-only-fixture\\n'\r".encodeToByteArray()) + } else session.send(bytes) + } + } + } + } + val terminal = RecordingTerminal() + val manager = SessionManager(adapter, RoomKnownHostStore(database.knownHosts(), preferences), + terminal, bootstrapTimeoutMillis = 1_000) + try { + manager.prepareConnection(ConnectionRequest( + HostProfile("Downgrade fixture", HostEndpoint( + arguments.getString(HOST_ARGUMENT) ?: DEFAULT_HOST, + arguments.getString(PORT_ARGUMENT)?.toIntOrNull() ?: DEFAULT_PORT), + arguments.getString(USER_ARGUMENT) ?: DEFAULT_USER), + SessionCredential.Password.from(requireNotNull(password).toCharArray()), + )) + manager.connectPrepared() + withTimeout(CONNECTION_TIMEOUT_MILLIS) { + manager.state.filterIsInstance().first() + } + manager.resolveHostKey(HostKeyDecision.ACCEPT_AND_SAVE) + val downgrade = withTimeout(CONNECTION_TIMEOUT_MILLIS) { + manager.structuredState.filterIsInstance().first() + } + assertEquals(StructuredShellUnavailableReason.BOOTSTRAP_TIMED_OUT, downgrade.reason) + assertTrue(manager.state.value is SessionState.Connected) + manager.send("printf 'live-%s\\n' terminal\r".encodeToByteArray()) + withTimeout(COMMAND_TIMEOUT_MILLIS) { + while (!terminal.text().contains("live-terminal")) delay(10) + } + assertTrue(manager.state.value is SessionState.Connected) + } finally { + manager.disconnect() + withTimeout(CONNECTION_TIMEOUT_MILLIS) { manager.state.first { it is SessionState.Disconnected } } + database.close() + preferences.edit().clear().commit() + } + } + @Test fun repliesReachRunningCommandAndOnlyRemoteEchoEntersSavedOutput() = runBlocking { val arguments = InstrumentationRegistry.getArguments() @@ -707,6 +869,7 @@ private class RecordingSshClientAdapter( private val delegate: SshClientAdapter, ) : SshClientAdapter { val sent = CopyOnWriteArrayList() + val attempts = AtomicInteger() override suspend fun connect( request: ConnectionRequest, @@ -714,6 +877,7 @@ private class RecordingSshClientAdapter( initialSize: TerminalSize, onStage: (dev.threadline.core.model.ConnectionStage) -> Unit, ): LiveSshSession { + attempts.incrementAndGet() val session = delegate.connect(request, verifier, initialSize, onStage) return object : LiveSshSession by session { override suspend fun send(bytes: ByteArray) { diff --git a/app/src/main/java/dev/threadline/MainActivity.kt b/app/src/main/java/dev/threadline/MainActivity.kt index 3a1ca22..50ddbd8 100644 --- a/app/src/main/java/dev/threadline/MainActivity.kt +++ b/app/src/main/java/dev/threadline/MainActivity.kt @@ -82,6 +82,7 @@ import androidx.lifecycle.Lifecycle import androidx.lifecycle.compose.LifecycleEventEffect import androidx.lifecycle.compose.collectAsStateWithLifecycle import dev.threadline.core.model.ConnectionRequest +import dev.threadline.core.model.ConnectionTarget import dev.threadline.core.model.HostEndpoint import dev.threadline.core.model.HostKeyDecision import dev.threadline.core.model.HostKeyPrompt @@ -300,6 +301,7 @@ private fun ThreadlineApp() { val transcriptSessions by SessionRuntime.transcriptHistory.sessions .collectAsStateWithLifecycle(initialValue = emptyList()) val transcriptSaveFailed by manager.transcriptSaveFailed.collectAsStateWithLifecycle() + val connectionTarget by manager.connectionTarget.collectAsStateWithLifecycle() val state = snapshot.connection var connectionDraft by rememberSaveable(stateSaver = ConnectionFormDraft.Saver) { mutableStateOf(ConnectionFormDraft.emptyDefaults()) @@ -318,6 +320,8 @@ private fun ThreadlineApp() { } var selectedHostProfileId by rememberSaveable { mutableStateOf(null) } var showConnectedSession by rememberSaveable { mutableStateOf(true) } + var showRecovery by rememberSaveable { mutableStateOf(true) } + var reconnectRequested by rememberSaveable { mutableStateOf(false) } val connectedSessionStateHolder = rememberSaveableStateHolder() var diagnosticGeneratedAtMillis by remember { mutableStateOf(null) } val diagnosticEnvironment = remember(context) { androidDiagnosticEnvironment(context) } @@ -383,10 +387,20 @@ private fun ThreadlineApp() { val canShowIntroduction = state is SessionState.Disconnected || state is SessionState.Failed || state is SessionState.Connected && !showConnectedSession + val endedError = (state as? SessionState.Failed)?.error?.takeIf { + it is SessionError.ShellEnded || it == SessionError.ConnectionLost + } val initialHomeTask = when { - state is SessionState.Failed -> HomeTask.CONNECTION + reconnectRequested -> HomeTask.CONNECTION + state is SessionState.Failed && endedError == null -> HomeTask.CONNECTION else -> homeTaskAfterIntroduction } + LaunchedEffect(state) { + if (state is SessionState.Connecting || state is SessionState.Connected) { + showRecovery = true + reconnectRequested = false + } + } LaunchedEffect(showIntroduction) { if (!showIntroduction) homeTaskAfterIntroduction = HomeTask.OVERVIEW } @@ -397,6 +411,23 @@ private fun ThreadlineApp() { showIntroduction = false }, ) + } else if (endedError != null && connectionTarget != null && showRecovery) { + EndedSessionScreen( + target = requireNotNull(connectionTarget), + error = endedError, + transcript = snapshot.transcript, + onReconnect = { + showRecovery = false + reconnectRequested = true + selectedHostProfileId = null + }, + onOpenHome = { + showRecovery = false + reconnectRequested = false + homeTaskAfterIntroduction = HomeTask.OVERVIEW + }, + onOpenDiagnostics = openDiagnostics, + ) } else if (state is SessionState.Connected && showConnectedSession) { connectedSessionStateHolder.SaveableStateProvider("active-session") { ConnectedSessionScreen( @@ -421,7 +452,10 @@ private fun ThreadlineApp() { onDraftChange = { connectionDraft = it }, sessionError = (current as? SessionState.Failed) ?.error - ?.takeUnless { it == SessionError.NotificationPermissionRequired }, + ?.takeUnless { + it == SessionError.NotificationPermissionRequired || endedError != null + }, + reconnectTarget = connectionTarget.takeIf { reconnectRequested }, activeSessionDisplayName = (current as? SessionState.Connected)?.displayName, connectionEnabled = current !is SessionState.Connected, initialTask = initialHomeTask, @@ -716,6 +750,7 @@ internal fun HostForm( draft: ConnectionFormDraft, onDraftChange: (ConnectionFormDraft) -> Unit, sessionError: SessionError?, + reconnectTarget: ConnectionTarget? = null, activeSessionDisplayName: String? = null, connectionEnabled: Boolean = true, initialTask: HomeTask = HomeTask.CONNECTION, @@ -836,7 +871,7 @@ internal fun HostForm( val privateKeyBringIntoViewRequester = remember { BringIntoViewRequester() } val keyPassphraseFocusRequester = remember { FocusRequester() } val selectedHostProfile = hostProfiles.firstOrNull { it.id == selectedHostProfileId } - val isCredentialEntry = credentialOnly && selectedHostProfile != null + val isCredentialEntry = credentialOnly val selectedPreferredIdentity = sshIdentities.firstOrNull { it.id == selectedPreferredIdentityId } @@ -864,19 +899,45 @@ internal fun HostForm( } } - fun clearSessionCredentialInputs() { + fun clearSessionCredentialInputs(clearKeySelection: Boolean = true) { passwordState = TextFieldState() keyPassphrase = "" - selectedKeyUri = null - selectedSavedKeyId = null + if (clearKeySelection) { + selectedKeyUri = null + selectedSavedKeyId = null + } savePrivateKey = false validationError = null connectionPreparationError = null } + LaunchedEffect(reconnectTarget) { + val target = reconnectTarget ?: return@LaunchedEffect + clearSessionCredentialInputs() + selectedPreferredIdentityId = target.identityId + selectedSavedKeyId = target.importedPrivateKeyId + selectedKeyUri = target.privateKeyUri + credentialOnly = true + savedTask = HomeTask.CONNECTION.name + onDraftChange( + ConnectionFormDraft( + displayName = target.profile.displayName, + hostname = target.profile.endpoint.hostname, + port = target.profile.endpoint.port.toString(), + username = target.profile.username, + authenticationMode = if (target.usesPrivateKey) { + AuthenticationMode.PRIVATE_KEY + } else { + AuthenticationMode.PASSWORD + }, + ephemeral = target.ephemeral, + ), + ) + } + LaunchedEffect(notificationPermissionState) { if (!notificationPermissionState.allowsCredentialEntry) { - clearSessionCredentialInputs() + clearSessionCredentialInputs(clearKeySelection = reconnectTarget == null) } } @@ -1069,6 +1130,8 @@ internal fun HostForm( ), credential = credential, ephemeral = false, + identityId = identity.id, + importedPrivateKeyId = identity.importedPrivateKeyId, ) if (onPrepared(request)) { transferred = true @@ -1190,7 +1253,11 @@ internal fun HostForm( } HomeTaskHeader( - title = selectedHostProfile?.displayName ?: "New connection", + title = if (isCredentialEntry) { + draft.displayName + } else { + selectedHostProfile?.displayName ?: "New connection" + }, onBack = { clearSessionCredentialInputs() formError = null @@ -1204,15 +1271,16 @@ internal fun HostForm( ) Text( if (isCredentialEntry) { - val profile = requireNotNull(selectedHostProfile) - "${selectedPreferredIdentity?.username ?: profile.username}@" + - "${profile.hostname}:${profile.port}" + "${draft.username}@${draft.hostname}:${draft.port}" } else { "Threadline connects directly to the SSH endpoint you enter and verifies " + "the server before signing in." }, style = MaterialTheme.typography.bodyMedium, ) + if (isCredentialEntry && reconnectTarget != null) { + Text(FreshShellNotice, style = MaterialTheme.typography.bodySmall) + } if (isCredentialEntry) { TextButton( onClick = { credentialOnly = false }, @@ -1794,6 +1862,13 @@ internal fun HostForm( profile = profile, credential = credential, ephemeral = draft.ephemeral, + identityId = selectedPreferredIdentityId, + importedPrivateKeyId = selectedSavedKeyId.takeIf { + draft.authenticationMode == AuthenticationMode.PRIVATE_KEY + }, + privateKeyUri = selectedKeyUri.takeIf { + draft.authenticationMode == AuthenticationMode.PRIVATE_KEY + }, ) if (onPrepared(request)) { credentialTransferred = true diff --git a/app/src/main/java/dev/threadline/TranscriptScreen.kt b/app/src/main/java/dev/threadline/TranscriptScreen.kt index 2f982aa..1f6d6d2 100644 --- a/app/src/main/java/dev/threadline/TranscriptScreen.kt +++ b/app/src/main/java/dev/threadline/TranscriptScreen.kt @@ -3,6 +3,7 @@ package dev.threadline import android.content.ClipData import android.content.ClipboardManager import android.content.Context +import androidx.activity.compose.BackHandler import androidx.compose.foundation.background import androidx.compose.foundation.interaction.collectIsDraggedAsState import androidx.compose.foundation.horizontalScroll @@ -99,6 +100,9 @@ import dev.threadline.core.terminal.TerminalKey import dev.threadline.core.terminal.TerminalModifiers import dev.threadline.core.transcript.AnsiColor import dev.threadline.core.transcript.CommandOutput +import dev.threadline.core.model.ConnectionTarget +import dev.threadline.core.model.SessionError +import dev.threadline.core.model.presentation import dev.threadline.core.transcript.CommandStatus import dev.threadline.core.transcript.CommandTranscriptState import dev.threadline.core.transcript.CommandTurn @@ -108,6 +112,8 @@ import kotlinx.coroutines.delay import org.connectbot.terminal.Terminal internal object TranscriptTags { + const val ENDED_SESSION = "session-ended" + const val RECONNECT = "session-reconnect" const val TRANSCRIPT = "session-transcript" const val COMPOSER = "command-composer" const val SEND = "command-send" @@ -164,6 +170,86 @@ private val terminalExtraKeys = listOf( private val CompactRawSessionHeight = 360.dp +internal const val FreshShellNotice = + "A fresh connection starts a new shell. The old directory, exports, aliases, functions, " + + "and remote processes are not restored. No command is replayed. " + + "Remote processes may still be running on the server." + +@Composable +internal fun EndedSessionScreen( + target: ConnectionTarget, + error: SessionError, + transcript: CommandTranscriptState, + onReconnect: () -> Unit, + onOpenHome: () -> Unit, + onOpenDiagnostics: () -> Unit = {}, +) { + BackHandler(onBack = onOpenHome) + val presentation = error.presentation() + val uriHandler = LocalUriHandler.current + Scaffold( + topBar = { + Column(Modifier.statusBarsPadding()) { + Text( + target.profile.displayName, + style = MaterialTheme.typography.titleLarge, + modifier = Modifier.padding(16.dp).semantics { heading() }, + ) + Row { + TextButton(onClick = onOpenHome) { Text("Home") } + TextButton(onClick = onOpenDiagnostics) { Text("Diagnostics") } + } + HorizontalDivider() + } + }, + ) { padding -> + LazyColumn( + modifier = Modifier.padding(padding).fillMaxSize() + .testTag(TranscriptTags.ENDED_SESSION), + contentPadding = androidx.compose.foundation.layout.PaddingValues(16.dp), + verticalArrangement = Arrangement.spacedBy(12.dp), + ) { + item { + Column( + verticalArrangement = Arrangement.spacedBy(8.dp), + modifier = Modifier.semantics { liveRegion = LiveRegionMode.Assertive }, + ) { + Text(presentation.title, style = MaterialTheme.typography.titleMedium) + Text(presentation.message) + Text(presentation.recovery, style = MaterialTheme.typography.bodySmall) + Text(FreshShellNotice, style = MaterialTheme.typography.bodySmall) + Text( + "The server identity is verified again. Enter or unlock credentials again.", + style = MaterialTheme.typography.bodySmall, + ) + Button( + onClick = onReconnect, + modifier = Modifier.testTag(TranscriptTags.RECONNECT), + ) { + Text("Start fresh shell") + } + } + } + items(transcript.turns, key = { it.id.value }) { turn -> + CommandCard( + turn = turn, + canSubmit = false, + canReply = false, + onReply = { _, _ -> ReplySubmissionResult.NOT_RUNNING }, + onStop = {}, + onDisconnect = {}, + onEdit = {}, + onRerun = {}, + onOpenUrl = uriHandler::openUri, + onOpenTerminal = {}, + clockMillis = System::currentTimeMillis, + readOnly = true, + ) + } + } + } +} + @Composable internal fun ConnectedSessionScreen( displayName: String, @@ -1015,6 +1101,7 @@ private fun CommandCard( onOpenTerminal: () -> Unit, clockMillis: () -> Long, modifier: Modifier = Modifier, + readOnly: Boolean = false, ) { val context = LocalContext.current var expanded by rememberSaveable(turn.id.value) { mutableStateOf(false) } @@ -1081,7 +1168,11 @@ private fun CommandCard( } if (turn.output.approximate) { Text( - "Transcript rendering is approximate; open Terminal for the exact view.", + if (readOnly) { + "Transcript rendering is approximate." + } else { + "Transcript rendering is approximate; open Terminal for the exact view." + }, style = MaterialTheme.typography.labelSmall, ) } @@ -1136,7 +1227,7 @@ private fun CommandCard( } } - else -> { + else -> if (!readOnly) { TextButton(onClick = onRerun, enabled = canSubmit) { Text("Rerun") } } } @@ -1153,7 +1244,7 @@ private fun CommandCard( expanded = cardActionsExpanded, onDismissRequest = { cardActionsExpanded = false }, ) { - if (!turn.status.isActive()) { + if (!readOnly && !turn.status.isActive()) { DropdownMenuItem( text = { Text("Edit command") }, onClick = { diff --git a/app/src/main/java/dev/threadline/core/diagnostics/DiagnosticReport.kt b/app/src/main/java/dev/threadline/core/diagnostics/DiagnosticReport.kt index 1929411..9ebecaf 100644 --- a/app/src/main/java/dev/threadline/core/diagnostics/DiagnosticReport.kt +++ b/app/src/main/java/dev/threadline/core/diagnostics/DiagnosticReport.kt @@ -219,6 +219,7 @@ private fun SessionError.diagnosticCode(): String = when (this) { SessionError.NetworkUnreachable -> "network_unreachable" SessionError.ProtocolMismatch -> "protocol_mismatch" SessionError.ConnectionLost -> "connection_lost" + is SessionError.ShellEnded -> "shell_ended" SessionError.InputBackpressure -> "input_backpressure" SessionError.TerminalRendererFailed -> "terminal_renderer_failed" SessionError.NotificationPermissionRequired -> "notification_permission_required" diff --git a/app/src/main/java/dev/threadline/core/model/SessionErrorPresentation.kt b/app/src/main/java/dev/threadline/core/model/SessionErrorPresentation.kt index 8e06ecf..e41eebc 100644 --- a/app/src/main/java/dev/threadline/core/model/SessionErrorPresentation.kt +++ b/app/src/main/java/dev/threadline/core/model/SessionErrorPresentation.kt @@ -119,6 +119,12 @@ fun SessionError.presentation(): SessionErrorPresentation = when (this) { action = SessionErrorAction.REVIEW_SERVER, ) + is SessionError.ShellEnded -> SessionErrorPresentation( + title = "Shell ended", + message = userMessage, + recovery = "Start a fresh shell to continue. The last command may have ended the shell.", + ) + SessionError.InputBackpressure -> SessionErrorPresentation( title = "Session input stopped", message = userMessage, diff --git a/app/src/main/java/dev/threadline/core/model/SessionModels.kt b/app/src/main/java/dev/threadline/core/model/SessionModels.kt index 930cd9a..428e7b6 100644 --- a/app/src/main/java/dev/threadline/core/model/SessionModels.kt +++ b/app/src/main/java/dev/threadline/core/model/SessionModels.kt @@ -65,11 +65,23 @@ class ConnectionRequest( val profile: HostProfile, val credential: SessionCredential, val ephemeral: Boolean = false, + val identityId: String? = null, + val importedPrivateKeyId: String? = null, + val privateKeyUri: String? = null, ) { override fun toString(): String = "ConnectionRequest(profile=$profile, credential=$credential, ephemeral=$ephemeral)" } +data class ConnectionTarget( + val profile: HostProfile, + val usesPrivateKey: Boolean, + val ephemeral: Boolean, + val identityId: String?, + val importedPrivateKeyId: String?, + val privateKeyUri: String?, +) + data class TerminalSize( val rows: Int, val columns: Int, @@ -170,6 +182,12 @@ sealed interface SessionError { override val userMessage: String = "The SSH connection was lost." } + data class ShellEnded(val exitStatus: Long? = null) : SessionError { + override val userMessage: String = exitStatus?.let { + "The remote shell ended with exit status $it." + } ?: "The remote shell or SSH channel ended without an exit status." + } + data object InputBackpressure : SessionError { override val userMessage: String = "Local input was stopped because the SSH send queue filled." diff --git a/app/src/main/java/dev/threadline/core/session/SessionManager.kt b/app/src/main/java/dev/threadline/core/session/SessionManager.kt index 00ff46b..8d6d235 100644 --- a/app/src/main/java/dev/threadline/core/session/SessionManager.kt +++ b/app/src/main/java/dev/threadline/core/session/SessionManager.kt @@ -1,6 +1,7 @@ package dev.threadline.core.session import dev.threadline.core.model.ConnectionRequest +import dev.threadline.core.model.ConnectionTarget import dev.threadline.core.model.HostProfile import dev.threadline.core.model.HostKeyDecision import dev.threadline.core.model.HostKeyPrompt @@ -55,7 +56,7 @@ import kotlinx.coroutines.flow.SharingStarted import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.combine import kotlinx.coroutines.flow.debounce -import kotlinx.coroutines.flow.first +import kotlinx.coroutines.flow.firstOrNull import kotlinx.coroutines.flow.stateIn import kotlinx.coroutines.launch import kotlinx.coroutines.sync.Mutex @@ -105,6 +106,8 @@ class SessionManager( val transcriptState: StateFlow = commandTranscript.state private val mutableTranscriptSaveFailed = MutableStateFlow(false) val transcriptSaveFailed: StateFlow = mutableTranscriptSaveFailed + private val mutableConnectionTarget = MutableStateFlow(null) + val connectionTarget: StateFlow = mutableConnectionTarget val snapshot: StateFlow = combine( state, structuredState, @@ -126,6 +129,7 @@ class SessionManager( private var outputJob: Job? = null private var disconnectMonitorJob: Job? = null private var disconnectJob: Job? = null + private var cleanupJob: Job? = null private var bootstrapTimeoutJob: Job? = null private var activeTranscriptSession: ActiveTranscriptSession? = null @@ -143,9 +147,10 @@ class SessionManager( scope.launch { inputRequests.consumeEach { input -> if ( + liveSession !== input.session || input.commandId != null && - (structuredState.value as? StructuredShellState.Running) - ?.activeCommand?.id != input.commandId + (structuredState.value as? StructuredShellState.Running) + ?.activeCommand?.id != input.commandId ) { input.bytes.fill(0) return@consumeEach @@ -158,12 +163,9 @@ class SessionManager( input.session.send(input.bytes) } .onFailure { - internalInputEchoFilter.cancel() - if ( - liveSession === input.session && - state.value is SessionState.Connected - ) { - failSession(SessionError.ConnectionLost) + if (liveSession === input.session) { + internalInputEchoFilter.cancel() + failSession(SessionError.ConnectionLost, input.session) } } } finally { @@ -175,10 +177,11 @@ class SessionManager( resizeRequests .debounce(100) .collect { size -> - val accepted = runCatching { liveSession?.resize(size) ?: true } + val session = liveSession ?: return@collect + val accepted = runCatching { session.resize(size) } .getOrDefault(false) - if (!accepted && state.value is SessionState.Connected) { - failSession(SessionError.PtyResizeRejected) + if (!accepted) { + failSession(SessionError.PtyResizeRejected, session) } } } @@ -192,8 +195,8 @@ class SessionManager( fun prepareConnection(request: ConnectionRequest): Boolean = synchronized(pendingLock) { if ( - state.value !is SessionState.Disconnected && - state.value !is SessionState.Failed + disconnectJob?.isActive == true || + state.value !is SessionState.Disconnected && state.value !is SessionState.Failed ) { request.credential.clear() return@synchronized false @@ -204,25 +207,33 @@ class SessionManager( true } - fun connectPrepared(): Boolean { - val request = synchronized(pendingLock) { - pendingRequest.also { pendingRequest = null } - } ?: return false - - synchronized(transcriptArchiveLock) { - activeTranscriptSession = ActiveTranscriptSession( - id = transcriptSessionIdFactory(), - profile = request.profile, - ephemeral = request.ephemeral, - ) + fun connectPrepared(): Boolean = synchronized(pendingLock) { + val request = pendingRequest ?: return@synchronized false + pendingRequest = null + if (disconnectJob?.isActive == true || + state.value !is SessionState.Disconnected && state.value !is SessionState.Failed + ) { + request.credential.clear() + return@synchronized false } stateMachine.apply(SessionEvent.ConnectRequested(request.profile.displayName)) - commandTranscript.reset() - resetStructuredShell() - internalInputEchoFilter.reset() - terminal.clear() - connectJob = scope.launch { establish(request) } - return true + val previousCleanup = cleanupJob + connectJob = scope.launch { + previousCleanup?.join() + synchronized(transcriptArchiveLock) { + activeTranscriptSession = ActiveTranscriptSession( + id = transcriptSessionIdFactory(), + profile = request.profile, + ephemeral = request.ephemeral, + ) + } + commandTranscript.reset() + resetStructuredShell() + internalInputEchoFilter.reset() + terminal.clear() + establish(request) + }.also { job -> job.invokeOnCompletion { request.credential.clear() } } + true } fun cancelPrepared(error: SessionError) { @@ -230,7 +241,7 @@ class SessionManager( pendingRequest?.credential?.clear() pendingRequest = null } - stateMachine.apply(SessionEvent.Failed(error)) + failSession(error) } fun resolveHostKey(decision: HostKeyDecision): Boolean { @@ -365,10 +376,12 @@ class SessionManager( resizeRequests.tryEmit(size) } - fun disconnect() { - if (disconnectJob?.isActive == true) return + fun disconnect() = synchronized(pendingLock) { + if (disconnectJob?.isActive == true) return@synchronized + stateMachine.apply(SessionEvent.DisconnectRequested) + val previousCleanup = cleanupJob disconnectJob = scope.launch { - stateMachine.apply(SessionEvent.DisconnectRequested) + previousCleanup?.join() synchronized(decisionLock) { hostKeyDecision?.complete(HostKeyDecision.REJECT) } @@ -388,25 +401,7 @@ class SessionManager( } fun onServiceDestroyed() { - scope.launch { - synchronized(decisionLock) { - hostKeyDecision?.complete(HostKeyDecision.REJECT) - } - connectJob?.cancelAndJoin() - outputJob?.cancelAndJoin() - disconnectMonitorJob?.cancelAndJoin() - bootstrapTimeoutJob?.cancelAndJoin() - val session = liveSession - liveSession = null - commandTranscript.sessionDisconnected() - val archive = closeTranscriptSession() - resetStructuredShell() - runCatching { session?.disconnect() } - persistTranscriptArchive(archive) - if (state.value !is SessionState.Failed) { - stateMachine.apply(SessionEvent.Disconnected) - } - } + if (state.value !is SessionState.Failed) disconnect() } private suspend fun establish(request: ConnectionRequest) { @@ -436,11 +431,21 @@ class SessionManager( } } } - liveSession = session - markTranscriptSessionConnected() - startStructuredShell(session) - stateMachine.apply(SessionEvent.ShellReady(terminal.size)) - startSessionJobs(session) + synchronized(pendingLock) { + liveSession = session + mutableConnectionTarget.value = ConnectionTarget( + profile = request.profile, + usesPrivateKey = request.credential is SessionCredential.PrivateKey, + ephemeral = request.ephemeral, + identityId = request.identityId, + importedPrivateKeyId = request.importedPrivateKeyId, + privateKeyUri = request.privateKeyUri, + ) + markTranscriptSessionConnected() + startStructuredShell(session) + stateMachine.apply(SessionEvent.ShellReady(terminal.size)) + startSessionJobs(session) + } } catch (failure: SshAdapterException) { failSession(failure.error) } finally { @@ -474,6 +479,9 @@ class SessionManager( profile = request.profile, credential = attemptCredential, ephemeral = request.ephemeral, + identityId = request.identityId, + importedPrivateKeyId = request.importedPrivateKeyId, + privateKeyUri = request.privateKeyUri, ) return try { @@ -550,28 +558,22 @@ class SessionManager( outputJob = scope.launch { try { for (bytes in session.output) { + if (liveSession !== session) break val terminalBytes = internalInputEchoFilter.consume(bytes) if (terminalBytes.isNotEmpty()) terminal.receive(terminalBytes) processStructuredOutput(bytes) } - failIfUnexpectedDisconnect() + failSession(session.outputEndError(), session) } catch (cancelled: CancellationException) { throw cancelled } catch (_: Exception) { - if (state.value is SessionState.Connected) { - failSession(SessionError.TerminalRendererFailed) - } + failSession(SessionError.TerminalRendererFailed, session) } } disconnectMonitorJob = scope.launch { - session.disconnects.first() - failIfUnexpectedDisconnect() - } - } - - private fun failIfUnexpectedDisconnect() { - if (state.value is SessionState.Connected) { - failSession(SessionError.ConnectionLost) + if (session.disconnects.firstOrNull() != null) { + failSession(SessionError.ConnectionLost, session) + } } } @@ -727,13 +729,33 @@ class SessionManager( } } - private fun failSession(error: SessionError) { - commandTranscript.sessionDisconnected() - val archive = closeTranscriptSession() - scope.launch { persistTranscriptArchive(archive) } - resetStructuredShell() - stateMachine.apply(SessionEvent.Failed(error)) - } + private fun failSession(error: SessionError, expectedSession: LiveSshSession? = null) = + synchronized(pendingLock) { + if (expectedSession != null && + (liveSession !== expectedSession || state.value !is SessionState.Connected) + ) return@synchronized + val session = liveSession + liveSession = null + val connecting = connectJob + val output = outputJob + val monitor = disconnectMonitorJob + val previousCleanup = cleanupJob + connecting?.cancel() + output?.cancel() + monitor?.cancel() + commandTranscript.sessionDisconnected() + val archive = closeTranscriptSession() + resetStructuredShell() + cleanupJob = scope.launch { + previousCleanup?.join() + connecting?.join() + output?.join() + monitor?.join() + runCatching { session?.disconnect() } + persistTranscriptArchive(archive) + } + stateMachine.apply(SessionEvent.Failed(error)) + } private fun markTranscriptSessionConnected() = synchronized(transcriptArchiveLock) { activeTranscriptSession = activeTranscriptSession?.copy( diff --git a/app/src/main/java/dev/threadline/core/ssh/ConnectBotSshClientAdapter.kt b/app/src/main/java/dev/threadline/core/ssh/ConnectBotSshClientAdapter.kt index 62fdf0e..4c6d603 100644 --- a/app/src/main/java/dev/threadline/core/ssh/ConnectBotSshClientAdapter.kt +++ b/app/src/main/java/dev/threadline/core/ssh/ConnectBotSshClientAdapter.kt @@ -13,7 +13,16 @@ import java.nio.channels.UnresolvedAddressException import kotlinx.coroutines.CancellationException import kotlinx.coroutines.channels.ReceiveChannel import kotlinx.coroutines.flow.Flow -import kotlinx.coroutines.flow.map +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.CoroutineStart +import kotlinx.coroutines.Deferred +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.async +import kotlinx.coroutines.cancel +import kotlinx.coroutines.flow.first +import kotlinx.coroutines.flow.flow +import kotlinx.coroutines.withTimeoutOrNull import org.connectbot.sshlib.AuthResult import org.connectbot.sshlib.ConnectResult import org.connectbot.sshlib.HostKeyVerifier @@ -21,6 +30,7 @@ import org.connectbot.sshlib.PublicKey import org.connectbot.sshlib.SshClient import org.connectbot.sshlib.SshClientConfig import org.connectbot.sshlib.SshSession +import org.connectbot.sshlib.SessionExit class ConnectBotSshClientAdapter( private val hostKeyAlgorithmsOverride: String? = null, @@ -35,6 +45,9 @@ class ConnectBotSshClientAdapter( host = request.profile.endpoint.hostname port = request.profile.endpoint.port preferPasswordAuth = true + autoDisconnectOnLastChannelClose = false + // Preserve the existing bound on unread remote output after the library update. + sessionWindowSize = 64 * 1024 hostKeyAlgorithmsOverride?.let { hostKeyAlgorithms = it } hostKeyVerifier = object : HostKeyVerifier { override suspend fun verify(key: PublicKey): Boolean = @@ -42,7 +55,13 @@ class ConnectBotSshClientAdapter( } } val client = SshClient(config) + val monitorScope = CoroutineScope(SupervisorJob() + Dispatchers.IO) + val connectionEnded = monitorScope.async(start = CoroutineStart.UNDISPATCHED) { + client.disconnectedFlow.first() + Unit + } var session: SshSession? = null + var transferred = false try { when (val result = client.connect()) { @@ -91,7 +110,8 @@ class ConnectBotSshClientAdapter( throw SshAdapterException(SessionError.ShellRejected) } - return ConnectBotLiveSession(client, session) + transferred = true + return ConnectBotLiveSession(client, session, connectionEnded, monitorScope) } catch (cancelled: CancellationException) { cleanUp(client, session) throw cancelled @@ -103,6 +123,7 @@ class ConnectBotSshClientAdapter( throw SshAdapterException(transportSessionError(unexpected), unexpected) } finally { request.credential.clear() + if (!transferred) monitorScope.cancel() } } @@ -157,9 +178,29 @@ private const val MAX_CAUSE_DEPTH = 8 private class ConnectBotLiveSession( private val client: SshClient, private val session: SshSession, + private val connectionEnded: Deferred, + private val monitorScope: CoroutineScope, ) : LiveSshSession { override val output: ReceiveChannel = session.stdout - override val disconnects: Flow = client.disconnectedFlow.map { } + override val disconnects: Flow = flow { + connectionEnded.await() + // A reported shell exit takes precedence over a following transport shutdown. + if (session.exitInfo.await() == null) emit(Unit) + } + + override suspend fun outputEndError(): SessionError = + // sshlib resolves unknown exit info before forwarding transport failure. + // Bound the wait because EOF alone need not include an exit report or CLOSE. + withTimeoutOrNull(1_000) { + when (val exit = session.exitInfo.await()) { + is SessionExit.Status -> SessionError.ShellEnded(exit.code) + is SessionExit.Signal -> SessionError.ShellEnded() + null -> { + connectionEnded.await() + SessionError.ConnectionLost + } + } + } ?: SessionError.ShellEnded() override suspend fun send(bytes: ByteArray) = session.write(bytes) @@ -172,7 +213,11 @@ private class ConnectBotLiveSession( ) override suspend fun disconnect() { - runCatching { session.close() } - client.disconnect() + try { + runCatching { session.close() } + client.disconnect() + } finally { + monitorScope.cancel() + } } } diff --git a/app/src/main/java/dev/threadline/core/ssh/SshClientAdapter.kt b/app/src/main/java/dev/threadline/core/ssh/SshClientAdapter.kt index 1bae6c4..bd1ee19 100644 --- a/app/src/main/java/dev/threadline/core/ssh/SshClientAdapter.kt +++ b/app/src/main/java/dev/threadline/core/ssh/SshClientAdapter.kt @@ -18,6 +18,8 @@ interface LiveSshSession { val output: ReceiveChannel val disconnects: Flow + suspend fun outputEndError(): SessionError = SessionError.ShellEnded() + suspend fun send(bytes: ByteArray) suspend fun resize(size: TerminalSize): Boolean diff --git a/app/src/main/java/dev/threadline/service/SshSessionService.kt b/app/src/main/java/dev/threadline/service/SshSessionService.kt index 7e24530..b5ef0ca 100644 --- a/app/src/main/java/dev/threadline/service/SshSessionService.kt +++ b/app/src/main/java/dev/threadline/service/SshSessionService.kt @@ -97,6 +97,7 @@ class SshSessionService : Service() { SessionState.Disconnected, is SessionState.Failed, -> { + handlesSessionCommand = false stopForeground(STOP_FOREGROUND_REMOVE) stopSelf() } diff --git a/app/src/test/java/dev/threadline/core/diagnostics/DiagnosticReportTest.kt b/app/src/test/java/dev/threadline/core/diagnostics/DiagnosticReportTest.kt index 4de72d9..f19ae32 100644 --- a/app/src/test/java/dev/threadline/core/diagnostics/DiagnosticReportTest.kt +++ b/app/src/test/java/dev/threadline/core/diagnostics/DiagnosticReportTest.kt @@ -138,6 +138,7 @@ class DiagnosticReportTest { SessionError.NetworkUnreachable to "network_unreachable", SessionError.ProtocolMismatch to "protocol_mismatch", SessionError.ConnectionLost to "connection_lost", + SessionError.ShellEnded(1) to "shell_ended", SessionError.InputBackpressure to "input_backpressure", SessionError.TerminalRendererFailed to "terminal_renderer_failed", SessionError.NotificationPermissionRequired to "notification_permission_required", diff --git a/app/src/test/java/dev/threadline/core/model/SessionErrorPresentationTest.kt b/app/src/test/java/dev/threadline/core/model/SessionErrorPresentationTest.kt index 6e96b33..7064860 100644 --- a/app/src/test/java/dev/threadline/core/model/SessionErrorPresentationTest.kt +++ b/app/src/test/java/dev/threadline/core/model/SessionErrorPresentationTest.kt @@ -86,6 +86,8 @@ class SessionErrorPresentationTest { SessionError.NetworkUnreachable, SessionError.ProtocolMismatch, SessionError.ConnectionLost, + SessionError.ShellEnded(), + SessionError.ShellEnded(1), SessionError.InputBackpressure, SessionError.TerminalRendererFailed, SessionError.NotificationPermissionRequired, diff --git a/app/src/test/java/dev/threadline/core/session/SessionManagerIoTest.kt b/app/src/test/java/dev/threadline/core/session/SessionManagerIoTest.kt index bac5d99..3af4271 100644 --- a/app/src/test/java/dev/threadline/core/session/SessionManagerIoTest.kt +++ b/app/src/test/java/dev/threadline/core/session/SessionManagerIoTest.kt @@ -47,6 +47,152 @@ import java.util.concurrent.CopyOnWriteArrayList import java.util.concurrent.atomic.AtomicInteger class SessionManagerIoTest { + @Test + fun `fresh connection waits for cleanup uses a new archive and never replays commands`() = runBlocking { + val releaseDisconnect = CompletableDeferred() + val old = RecordingSession(releaseDisconnect) + val fresh = RecordingSession() + val sessions = ArrayDeque(listOf(old, fresh)) + val attempts = AtomicInteger() + val adapter = object : SshClientAdapter { + override suspend fun connect( + request: ConnectionRequest, + verifier: ServerHostKeyVerifier, + initialSize: TerminalSize, + onStage: (dev.threadline.core.model.ConnectionStage) -> Unit, + ): LiveSshSession { + attempts.incrementAndGet() + return sessions.removeFirst() + } + } + val nonce = SessionNonce("0123456789abcdef0123456789abcdef") + val freshNonce = SessionNonce("abcdef0123456789abcdef0123456789") + val nonces = ArrayDeque(listOf(nonce, freshNonce)) + val ids = ArrayDeque(listOf("bootstrap-old", "completed-old", "exit-old", "bootstrap-new", "command-new")) + val archiveIds = ArrayDeque(listOf("old-session", "new-session")) + val archives = RecordingTranscriptArchiveSink() + val manager = SessionManager( + adapter, EmptyKnownHostStore, FakeTerminal, + sessionNonceFactory = { nonces.removeFirst() }, + commandIdFactory = { CommandId(ids.removeFirst()) }, + transcriptSessionIdFactory = { archiveIds.removeFirst() }, + transcriptArchiveSink = archives, + ) + val profile = HostProfile("Fixture", HostEndpoint("fixture.test", 2222), "threadline") + fun request() = ConnectionRequest( + profile = profile, + credential = SessionCredential.PrivateKey.from(byteArrayOf(1, 2), null), + identityId = "chosen-identity", + importedPrivateKeyId = "chosen-key", + ) + try { + assertTrue(manager.prepareConnection(request())) + assertTrue(manager.connectPrepared()) + old.awaitSent(1) + old.output.send(lifecycleBytes(nonce, CommandId("bootstrap-old"), 0, "/tmp")) + withTimeout(5_000) { manager.structuredState.first { it is StructuredShellState.Ready } } + manager.submitCommand("printf old") + old.awaitSent(2) + old.output.send(lifecycleBytes(nonce, CommandId("completed-old"), 0, "/tmp", "old output")) + withTimeout(5_000) { + manager.structuredState.first { + it is StructuredShellState.Ready && it.lastCommand?.id == CommandId("completed-old") + } + } + manager.submitCommand("exit") + old.awaitSent(3) + old.output.close() + withTimeout(5_000) { manager.state.first { it is SessionState.Failed } } + assertEquals(listOf("printf old", "exit"), manager.transcriptState.value.turns.map { it.command }) + assertEquals(CommandStatus.DISCONNECTED, manager.transcriptState.value.turns.last().status) + assertEquals("chosen-identity", manager.connectionTarget.value?.identityId) + assertEquals("chosen-key", manager.connectionTarget.value?.importedPrivateKeyId) + assertTrue(requireNotNull(manager.connectionTarget.value).usesPrivateKey) + assertEquals(1, attempts.get()) + + manager.cancelPrepared(SessionError.ServiceStartFailed) + assertTrue(manager.prepareConnection(request())) + assertTrue(manager.connectPrepared()) + delay(100) + assertEquals(1, attempts.get()) + releaseDisconnect.complete(Unit) + fresh.awaitSent(1) + fresh.output.send(lifecycleBytes(freshNonce, CommandId("bootstrap-new"), 0, "/home/threadline")) + withTimeout(5_000) { manager.structuredState.first { it is StructuredShellState.Ready } } + old.disconnects.emit(Unit) + delay(100) + assertTrue(manager.state.value is SessionState.Connected) + assertTrue(manager.transcriptState.value.turns.isEmpty()) + assertEquals(1, fresh.sent.size) + assertTrue(!fresh.sent.single().decodeToString().contains("printf old")) + assertEquals(2, attempts.get()) + + manager.submitCommand("printf fresh") + fresh.awaitSent(2) + fresh.output.send(lifecycleBytes(freshNonce, CommandId("command-new"), 0, "/home/threadline", "fresh output")) + withTimeout(5_000) { while (archives.archives.none { it.id == "new-session" }) delay(10) } + assertEquals(setOf("old-session", "new-session"), archives.archives.map { it.id }.toSet()) + assertEquals(2, archives.archives.last { it.id == "old-session" }.transcript.turns.size) + assertEquals("printf fresh", archives.archives.last { it.id == "new-session" }.transcript.turns.single().command) + } finally { + releaseDisconnect.complete(Unit) + manager.disconnect() + withTimeout(5_000) { manager.state.first { it is SessionState.Disconnected } } + } + } + + @Test + fun `fresh connection verifies the host key again and blocks a changed key`() = runBlocking { + val original = byteArrayOf(1, 2, 3) + val replacement = byteArrayOf(9, 9, 9) + val session = RecordingSession() + val adapter = HostKeyCheckingAdapter(listOf(original, original, replacement), session) + val store = MutableKnownHostStore() + val manager = SessionManager(adapter, store, FakeTerminal) + try { + manager.prepareConnection(fixtureRequest()) + manager.connectPrepared() + withTimeout(5_000) { manager.state.first { it is SessionState.AwaitingHostKey } } + manager.resolveHostKey(HostKeyDecision.ACCEPT_AND_SAVE) + withTimeout(5_000) { manager.state.first { it is SessionState.Connected } } + session.output.close() + withTimeout(5_000) { manager.state.first { it is SessionState.Failed } } + val freshRequest = fixtureRequest() + manager.prepareConnection(freshRequest) + manager.connectPrepared() + val failed = withTimeout(5_000) { + manager.state.filterIsInstance().first() + } + assertTrue(failed.error is SessionError.HostKeyChanged) + assertEquals(3, adapter.attempts) + assertArrayEquals(original, requireNotNull(store.record).key.encoded) + withTimeout(5_000) { while (!freshRequest.credential.isCleared()) delay(10) } + } finally { + manager.disconnect() + withTimeout(5_000) { manager.state.first { it is SessionState.Disconnected } } + } + } + + @Test + fun `ended shell releases its transport before another connection`() = runBlocking { + val session = RecordingSession() + val manager = SessionManager(ImmediateAdapter(session), EmptyKnownHostStore, FakeTerminal) + try { + manager.prepareConnection(fixtureRequest()) + manager.connectPrepared() + withTimeout(5_000) { manager.state.first { it is SessionState.Connected } } + session.output.close() + val failed = withTimeout(5_000) { + manager.state.filterIsInstance().first() + } + assertEquals(SessionError.ShellEnded(), failed.error) + withTimeout(5_000) { session.disconnected.await() } + } finally { + manager.disconnect() + withTimeout(5_000) { manager.state.first { it is SessionState.Disconnected } } + } + } + @Test fun `unknown host saves and reconnects after deferred decision`() = runBlocking { val key = byteArrayOf(1, 2, 3, 4) @@ -996,7 +1142,9 @@ private fun SessionCredential.isCleared(): Boolean = when (this) { keyBytes.all { it == 0.toByte() } && passphrase?.all { it == '\u0000' } != false } -private class RecordingSession : LiveSshSession { +private class RecordingSession( + private val releaseDisconnect: CompletableDeferred? = null, +) : LiveSshSession { override val output = Channel(Channel.BUFFERED) override val disconnects = MutableSharedFlow() val sent = CopyOnWriteArrayList() @@ -1014,6 +1162,7 @@ private class RecordingSession : LiveSshSession { override suspend fun resize(size: TerminalSize): Boolean = true override suspend fun disconnect() { + releaseDisconnect?.await() disconnected.complete(Unit) } } diff --git a/docs/STATUS.md b/docs/STATUS.md index 51c7c3c..df1fdeb 100644 --- a/docs/STATUS.md +++ b/docs/STATUS.md @@ -1,6 +1,6 @@ # Threadline current status -Updated: 2026-09-25 +Updated: 2026-10-01 This is the canonical execution-status page. `PROJECT_SPEC.md` remains the normative product and technical specification. Dated investigations are historical evidence for the boundary they @@ -31,6 +31,14 @@ without a dependency may be worked independently. Signed builds continue through direct invited sharing; public release needs a separate decision. No feature in this roadmap is implemented merely because its issue is open. +Fresh-shell recovery (#50) keeps the available transcript when a shell or channel ends or the +connection is lost. Starting a fresh shell repeats host-key verification and credential entry or +unlock, with the original endpoint and identity selection. It creates a new session without +restoring shell state or replaying commands. Integration downgrade keeps the live raw terminal. +The recovery fixture exposed an uncaught receive-window write failure in sshlib 0.4.2 (#73). +The adapter uses 0.5.0 with its receive window explicitly kept at 64 KiB; repeated transport-loss +and existing SSH/PTY, trust, imported-key, and transcript fixture paths passed on Android. + ## Phase 5 checkpoint (historical) Five Phase 5 slices were implemented before its original exit criterion was diff --git a/fixtures/openssh/README.md b/fixtures/openssh/README.md index 33a512e..4409f81 100644 --- a/fixtures/openssh/README.md +++ b/fixtures/openssh/README.md @@ -73,6 +73,11 @@ nonzero exit even though cases are skipped during ordinary connected-test runs because neither fixture credential is supplied. +Recovery cases cover shell `exit`, `exec`, persistent strict-mode failure, and abrupt transport +loss. Each fresh connection verifies the server again, gets its own archive, and starts without +the previous directory, exports, aliases, functions, or command replay. A deliberately missed +bootstrap also proves that the connected raw PTY remains usable. + To run the opt-in Phase 5 large-output profile against the production SSH, session, transcript, and terminal path: diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml index 1b48502..15d5640 100644 --- a/gradle/libs.versions.toml +++ b/gradle/libs.versions.toml @@ -7,7 +7,7 @@ coreKtx = "1.19.0" lifecycle = "2.11.0" coroutines = "1.11.0" material = "1.14.0" -sshlib = "0.4.2" +sshlib = "0.5.0" termlib = "0.1.0" conscrypt = "2.6.1" slf4j = "2.0.18" diff --git a/gradle/verification-metadata.xml b/gradle/verification-metadata.xml index 0d93206..e6ad041 100644 --- a/gradle/verification-metadata.xml +++ b/gradle/verification-metadata.xml @@ -1322,6 +1322,11 @@ + + + + + @@ -1330,11 +1335,24 @@ + + + + + + + + + + + + + @@ -1343,6 +1361,14 @@ + + + + + + + + @@ -2760,6 +2786,11 @@ + + + + + @@ -2768,11 +2799,24 @@ + + + + + + + + + + + + + @@ -2781,11 +2825,24 @@ + + + + + + + + + + + + + @@ -2794,6 +2851,14 @@ + + + + + + + + @@ -3356,6 +3421,14 @@ + + + + + + + + @@ -3364,6 +3437,14 @@ + + + + + + + + @@ -3869,6 +3950,14 @@ + + + + + + + + @@ -4264,6 +4353,11 @@ + + + + + @@ -4272,11 +4366,24 @@ + + + + + + + + + + + + + @@ -4285,6 +4392,14 @@ + + + + + + + + @@ -4362,11 +4477,24 @@ + + + + + + + + + + + + + @@ -4385,6 +4513,11 @@ + + + + + From 522baa92ec7a5a2b7907d4121c70b0f1320f6f69 Mon Sep 17 00:00:00 2001 From: Ross Tomsic Date: Thu, 1 Oct 2026 15:38:53 -0400 Subject: [PATCH 2/2] Discard abandoned input before starting a fresh shell --- .../threadline/core/session/SessionManager.kt | 7 +++- .../core/session/SessionManagerIoTest.kt | 42 +++++++++++++++++++ 2 files changed, 48 insertions(+), 1 deletion(-) diff --git a/app/src/main/java/dev/threadline/core/session/SessionManager.kt b/app/src/main/java/dev/threadline/core/session/SessionManager.kt index 8d6d235..02d1d80 100644 --- a/app/src/main/java/dev/threadline/core/session/SessionManager.kt +++ b/app/src/main/java/dev/threadline/core/session/SessionManager.kt @@ -220,6 +220,11 @@ class SessionManager( val previousCleanup = cleanupJob connectJob = scope.launch { previousCleanup?.join() + // Abandoned input must not consume the fresh shell's bootstrap slot. + while (true) { + val input = inputRequests.tryReceive().getOrNull() ?: break + input.bytes.fill(0) + } synchronized(transcriptArchiveLock) { activeTranscriptSession = ActiveTranscriptSession( id = transcriptSessionIdFactory(), @@ -255,7 +260,7 @@ class SessionManager( inputRequests.trySend(SessionInput(session, bytes.copyOf())).isFailure && state.value is SessionState.Connected ) { - failSession(SessionError.InputBackpressure) + failSession(SessionError.InputBackpressure, session) } } diff --git a/app/src/test/java/dev/threadline/core/session/SessionManagerIoTest.kt b/app/src/test/java/dev/threadline/core/session/SessionManagerIoTest.kt index 3af4271..fc41451 100644 --- a/app/src/test/java/dev/threadline/core/session/SessionManagerIoTest.kt +++ b/app/src/test/java/dev/threadline/core/session/SessionManagerIoTest.kt @@ -47,6 +47,43 @@ import java.util.concurrent.CopyOnWriteArrayList import java.util.concurrent.atomic.AtomicInteger class SessionManagerIoTest { + @Test + fun `input backpressure discards queued old input before a fresh shell`() = runBlocking { + val old = RecordingSession(releaseSend = CompletableDeferred()) + val fresh = RecordingSession() + val sessions = ArrayDeque(listOf(old, fresh)) + val adapter = object : SshClientAdapter { + override suspend fun connect( + request: ConnectionRequest, + verifier: ServerHostKeyVerifier, + initialSize: TerminalSize, + onStage: (dev.threadline.core.model.ConnectionStage) -> Unit, + ): LiveSshSession = sessions.removeFirst() + } + val manager = SessionManager(adapter, EmptyKnownHostStore, FakeTerminal) + try { + manager.prepareConnection(fixtureRequest()) + manager.connectPrepared() + withTimeout(5_000) { manager.state.first { it is SessionState.Connected } } + withTimeout(5_000) { old.sendStarted.await() } + repeat(257) { manager.send("queued-old".encodeToByteArray()) } + val failed = withTimeout(5_000) { + manager.state.filterIsInstance().first() + } + assertEquals(SessionError.InputBackpressure, failed.error) + withTimeout(5_000) { old.disconnected.await() } + manager.prepareConnection(fixtureRequest()) + manager.connectPrepared() + fresh.awaitSent(1) + assertTrue(manager.state.value is SessionState.Connected) + assertEquals(1, old.sent.size) + assertTrue(!fresh.sent.single().decodeToString().contains("queued-old")) + } finally { + manager.disconnect() + withTimeout(5_000) { manager.state.first { it is SessionState.Disconnected } } + } + } + @Test fun `fresh connection waits for cleanup uses a new archive and never replays commands`() = runBlocking { val releaseDisconnect = CompletableDeferred() @@ -1144,13 +1181,17 @@ private fun SessionCredential.isCleared(): Boolean = when (this) { private class RecordingSession( private val releaseDisconnect: CompletableDeferred? = null, + private val releaseSend: CompletableDeferred? = null, ) : LiveSshSession { override val output = Channel(Channel.BUFFERED) override val disconnects = MutableSharedFlow() val sent = CopyOnWriteArrayList() val disconnected = CompletableDeferred() + val sendStarted = CompletableDeferred() override suspend fun send(bytes: ByteArray) { + sendStarted.complete(Unit) + releaseSend?.await() delay(1) sent += bytes.copyOf() } @@ -1163,6 +1204,7 @@ private class RecordingSession( override suspend fun disconnect() { releaseDisconnect?.await() + releaseSend?.complete(Unit) disconnected.complete(Unit) } }