Checkpoint completed transcript turns during live sessions

app/src/androidTest/java/dev/threadline/ConnectionFormRetentionTest.kt

@@ -584,6 +584,10 @@ class ConnectionFormRetentionTest {
             .performClick()
         compose.waitForIdle()
         compose.onNodeWithText("printf saved").assertExists()
+        compose.onNodeWithText(
+            "While connected, only completed commands are saved. " +
+                "A crash can lose the unfinished command.",
+        ).assertExists()
         compose.onNodeWithTag(TranscriptHistoryTags.OUTPUT_PREFIX + "command-session-1")
             .assertExists()
         compose.onNodeWithText("Succeeded · exit 0").assertExists()
@@ -609,6 +613,25 @@ class ConnectionFormRetentionTest {
         compose.onAllNodesWithTag(TranscriptHistoryTags.OPEN).assertCountEquals(0)
     }
 
+    @Test
+    fun transcriptSaveFailureDoesNotClaimTheSessionEnded() {
+        compose.setContent {
+            MaterialTheme {
+                TranscriptHistorySection(
+                    sessions = emptyList(),
+                    saveFailed = true,
+                    onLoad = { error("No saved session") },
+                    onDelete = {},
+                    onClearAll = {},
+                )
+            }
+        }
+        compose.onNodeWithText(
+            "The latest transcript could not be saved. " +
+                "Recent commands may be missing from history.",
+        ).assertExists()
+    }
+
     @Test
     fun savedKeySelectionLoadsCredentialAndWipesTemporaryPassphrase() {
         val savedKey = ImportedPrivateKeyMetadata(
@@ -1338,7 +1361,7 @@ private fun transcriptSummary(
     port = 2222,
     username = "threadline",
     startedAtMillis = 1,
-    endedAtMillis = 2,
+    savedAtMillis = 2,
     turnsTruncated = false,
     turnCount = 1,
 )

app/src/androidTest/java/dev/threadline/core/session/AndroidTranscriptCheckpointProcessDeathTest.kt

@@ -0,0 +1,138 @@
+package dev.threadline.core.session
+
+import android.os.Bundle
+import android.os.Process
+import androidx.test.ext.junit.runners.AndroidJUnit4
+import androidx.test.platform.app.InstrumentationRegistry
+import dev.threadline.SessionRuntime
+import dev.threadline.core.model.ConnectionRequest
+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.SessionState
+import dev.threadline.core.shell.CommandSubmissionResult
+import dev.threadline.core.shell.StructuredShellState
+import dev.threadline.core.transcript.CommandStatus
+import java.io.File
+import kotlinx.coroutines.awaitCancellation
+import kotlinx.coroutines.flow.filterIsInstance
+import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.withTimeout
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertTrue
+import org.junit.Assume.assumeTrue
+import org.junit.Test
+import org.junit.runner.RunWith
+
+@RunWith(AndroidJUnit4::class)
+class AndroidTranscriptCheckpointProcessDeathTest {
+    // The fixture runner kills the preparing process, then verifies in a new one.
+    @Test
+    fun completedHistorySurvivesProcessDeathWithoutResumingConnection() = runBlocking {
+        val arguments = InstrumentationRegistry.getArguments()
+        val phase = arguments.getString("threadlineCheckpointPhase")
+        assumeTrue("Run through test-android-transcript-process-death.sh", phase != null)
+        val ephemeral = arguments.getString("threadlineEphemeral") == "true"
+        val manager = SessionRuntime.manager
+        val history = SessionRuntime.transcriptHistory
+
+        assertEquals(SessionState.Disconnected, manager.state.value)
+        assertEquals(StructuredShellState.Inactive, manager.structuredState.value)
+        assertTrue(manager.transcriptState.value.turns.isEmpty())
+
+        if (phase == "verify") {
+            val sessions = history.sessions.first()
+            if (ephemeral) {
+                assertTrue(sessions.isEmpty())
+            } else {
+                val saved = history.load(sessions.single().id)
+                val turn = saved.turns.single().turn
+                assertEquals("printf 'checkpoint-completed\\n'", turn.command)
+                assertEquals("checkpoint-completed\n", turn.output.plainText)
+                assertEquals(CommandStatus.SUCCEEDED, turn.status)
+                assertTrue(turn.completedAtMillis != null)
+            }
+            history.clearAll()
+            return@runBlocking
+        }
+
+        assertEquals("prepare", phase)
+        history.clearAll()
+        val context = InstrumentationRegistry.getInstrumentation().targetContext
+        val keyFile = File(context.filesDir, "threadline-fixture-private-key")
+        val keyBytes = keyFile.readBytes()
+        keyFile.delete()
+        val credential = try {
+            SessionCredential.PrivateKey.from(keyBytes, passphrase = null)
+        } finally {
+            keyBytes.fill(0)
+        }
+        try {
+            assertTrue(
+                manager.prepareConnection(
+                    ConnectionRequest(
+                        profile = HostProfile(
+                            displayName = "Checkpoint fixture",
+                            endpoint = HostEndpoint("10.0.2.2", 2222),
+                            username = "threadline",
+                        ),
+                        credential = credential,
+                        ephemeral = ephemeral,
+                    ),
+                ),
+            )
+            assertTrue(manager.connectPrepared())
+            val connection = withTimeout(20_000) {
+                manager.state.first {
+                    it is SessionState.AwaitingHostKey || it is SessionState.Connected ||
+                        it is SessionState.Failed
+                }
+            }
+            if (connection is SessionState.AwaitingHostKey) {
+                assertEquals(
+                    arguments.getString("threadlineFixtureHostKeyFingerprint"),
+                    connection.prompt.fingerprint,
+                )
+                assertTrue(manager.resolveHostKey(HostKeyDecision.ACCEPT_AND_SAVE))
+            }
+            withTimeout(20_000) {
+                manager.state.filterIsInstance<SessionState.Connected>().first()
+                manager.structuredState.filterIsInstance<StructuredShellState.Ready>().first()
+            }
+        } finally {
+            credential.clear()
+        }
+
+        val completed = manager.submitCommand("printf 'checkpoint-completed\\n'")
+            as CommandSubmissionResult.Accepted
+        withTimeout(20_000) {
+            manager.structuredState.filterIsInstance<StructuredShellState.Ready>()
+                .first { it.lastCommand?.id == completed.commandId }
+            if (!ephemeral) history.sessions.first { it.singleOrNull()?.turnCount == 1 }
+        }
+        val unfinished = manager.submitCommand("printf 'checkpoint-unfinished\\n'; sleep 120")
+            as CommandSubmissionResult.Accepted
+        withTimeout(20_000) {
+            manager.transcriptState.first { transcript ->
+                transcript.turns.firstOrNull { it.id == unfinished.commandId }
+                    ?.output?.plainText == "checkpoint-unfinished\n"
+            }
+        }
+        assertTrue(manager.state.value is SessionState.Connected)
+        assertTrue(!manager.transcriptSaveFailed.value)
+        val saved = history.sessions.first()
+        assertEquals(if (ephemeral) 0 else 1, saved.size)
+        if (!ephemeral) assertEquals(1, history.load(saved.single().id).turns.size)
+
+        InstrumentationRegistry.getInstrumentation().sendStatus(
+            0,
+            Bundle().apply {
+                putString("stream", "THREADLINE_CHECKPOINT_READY pid=${Process.myPid()}\n")
+            },
+        )
+        // No disconnect or database close: the host must kill this live process.
+        withTimeout(60_000) { awaitCancellation() }
+    }
+}

app/src/androidTest/java/dev/threadline/data/transcript/RoomTranscriptHistoryStoreTest.kt

@@ -45,7 +45,7 @@ class RoomTranscriptHistoryStoreTest {
     }
 
     @Test
-    fun archiveSurvivesReopenWithOrderedChunksAndBoundedNewestData() = runBlocking {
+    fun checkpointsAndFinalArchiveSurviveReopenWithBoundedNewestData() = runBlocking {
         val store = store()
         val retainedOutput =
             "x".repeat(RoomTranscriptHistoryStore.OUTPUT_CHUNK_CHARACTERS - 1) +
@@ -80,13 +80,17 @@ class RoomTranscriptHistoryStoreTest {
                 },
             )
         }
+        turns.indices.forEach { index ->
+            store.save(archive("persistent-session", index.toLong(), turns.take(index + 1)))
+        }
         store.save(archive("persistent-session", 100, turns))
 
         database.close()
         database = persistentDatabase()
         val reopenedStore = store()
         val summary = reopenedStore.sessions.first().single()
         assertEquals("persistent-session", summary.id)
+        assertEquals(100L, summary.savedAtMillis)
         assertEquals(RoomTranscriptHistoryStore.MAXIMUM_TURNS_PER_SESSION, summary.turnCount)
         assertTrue(summary.turnsTruncated)
 
@@ -129,6 +133,12 @@ class RoomTranscriptHistoryStoreTest {
         assertNull(database.transcriptArchives().findSession("session-0"))
         assertEquals("session-20", retained.first().id)
 
+        store.save(archive("session-20", 21, listOf(turn(20), turn(21))))
+        assertEquals(RoomTranscriptHistoryStore.MAXIMUM_SESSIONS, store.sessions.first().size)
+        assertEquals(2, store.load("session-20").turns.size)
+        assertEquals(2, database.transcriptArchives().findTurns("session-20").size)
+        assertEquals(2, database.transcriptArchives().findChunks("session-20").size)
+
         store.delete("session-10")
         assertNull(database.transcriptArchives().findSession("session-10"))
         assertTrue(database.transcriptArchives().findSession("session-11") != null)
@@ -164,7 +174,7 @@ class RoomTranscriptHistoryStoreTest {
 
     private fun archive(
         id: String,
-        endedAtMillis: Long,
+        savedAtMillis: Long,
         turns: List<CommandTurn>,
     ) = TranscriptSessionArchive(
         id = id,
@@ -174,7 +184,7 @@ class RoomTranscriptHistoryStoreTest {
             username = "threadline",
         ),
         startedAtMillis = 1,
-        endedAtMillis = endedAtMillis,
+        savedAtMillis = savedAtMillis,
         transcript = CommandTranscriptState(turns = turns),
     )
 

app/src/main/java/dev/threadline/TranscriptHistory.kt

@@ -68,7 +68,8 @@ internal fun TranscriptHistorySection(
     if (saveFailed || error != null) {
         Card(modifier = Modifier.fillMaxWidth()) {
             Text(
-                text = error ?: "The last session ended, but its transcript could not be saved.",
+                text = error ?: "The latest transcript could not be saved. " +
+                    "Recent commands may be missing from history.",
                 color = MaterialTheme.colorScheme.error,
                 modifier = Modifier.padding(16.dp),
             )
@@ -120,7 +121,7 @@ internal fun TranscriptHistorySection(
                                     fontFamily = FontFamily.Monospace,
                                 )
                                 Text(
-                                    "${formatTranscriptTimestamp(session.endedAtMillis)} · " +
+                                    "${formatTranscriptTimestamp(session.savedAtMillis)} · " +
                                         "${session.turnCount} " +
                                         if (session.turnCount == 1) "turn" else "turns",
                                     style = MaterialTheme.typography.bodySmall,
@@ -222,12 +223,17 @@ internal fun TranscriptHistorySection(
                                 style = MaterialTheme.typography.bodySmall,
                             )
                             Text(
-                                "Ended " +
-                                    formatTranscriptTimestamp(session.summary.endedAtMillis) +
+                                "Saved " +
+                                    formatTranscriptTimestamp(session.summary.savedAtMillis) +
                                     ". Saved output is plain text; terminal styling and " +
                                     "links are inert.",
                                 style = MaterialTheme.typography.bodySmall,
                             )
+                            Text(
+                                "While connected, only completed commands are saved. " +
+                                    "A crash can lose the unfinished command.",
+                                style = MaterialTheme.typography.bodySmall,
+                            )
                         }
                     }
                     if (session.summary.turnsTruncated) item {

app/src/main/java/dev/threadline/core/session/SessionManager.kt

@@ -58,6 +58,8 @@ import kotlinx.coroutines.flow.debounce
 import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.flow.stateIn
 import kotlinx.coroutines.launch
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
 
 enum class ReplySubmissionResult {
     SENT,
@@ -89,6 +91,7 @@ class SessionManager(
     private val decisionLock = Any()
     private val structuredLock = Any()
     private val transcriptArchiveLock = Any()
+    private val transcriptSaveMutex = Mutex()
     private val resizeRequests = MutableSharedFlow<TerminalSize>(
         extraBufferCapacity = 1,
         onBufferOverflow = BufferOverflow.DROP_OLDEST,
@@ -667,13 +670,16 @@ class SessionManager(
                 }
 
                 is ProtocolStreamItem.Lifecycle -> {
+                    var commandCompleted = false
                     val pendingBytes = synchronized(structuredLock) {
                         val pending = if (item.event is ShellLifecycleEvent.CommandStarted) {
                             internalInputEchoFilter.cancel()
                         } else {
                             byteArrayOf()
                         }
                         if (structuredContext === context) {
+                            commandCompleted = item.event is ShellLifecycleEvent.CommandEnded &&
+                                commandTranscript.state.value.activeCommandId == item.event.commandId
                             commandTranscript.lifecycle(item.event)
                             val next = structuredStateMachine.apply(
                                 StructuredShellEvent.Lifecycle(item.event),
@@ -692,6 +698,7 @@ class SessionManager(
                         pending
                     }
                     if (pendingBytes.isNotEmpty()) terminal.receive(pendingBytes)
+                    if (commandCompleted) checkpointTranscriptSession(context)
                 }
             }
         }
@@ -738,24 +745,53 @@ class SessionManager(
         synchronized(transcriptArchiveLock) {
             val active = activeTranscriptSession ?: return@synchronized null
             activeTranscriptSession = null
-            val startedAtMillis = active.startedAtMillis
-            if (
-                active.ephemeral ||
-                startedAtMillis == null ||
-                commandTranscript.state.value.turns.isEmpty()
-            ) {
-                return@synchronized null
+            transcriptArchive(active, completedOnly = false)
+        }
+
+    private suspend fun checkpointTranscriptSession(context: StructuredShellContext) {
+        transcriptSaveMutex.withLock {
+            // Capture after acquiring the save lock so a closed or replaced session
+            // cannot enqueue an older snapshot after its final archive.
+            val archive = synchronized(structuredLock) {
+                if (structuredContext !== context) return@synchronized null
+                synchronized(transcriptArchiveLock) {
+                    activeTranscriptSession?.let { transcriptArchive(it, completedOnly = true) }
+                }
             }
-            TranscriptSessionArchive(
-                id = active.id,
-                profile = active.profile,
-                startedAtMillis = startedAtMillis,
-                endedAtMillis = clockMillis(),
-                transcript = commandTranscript.state.value,
-            )
+            saveTranscriptArchive(archive)
         }
+    }
+
+    private fun transcriptArchive(
+        active: ActiveTranscriptSession,
+        completedOnly: Boolean,
+    ): TranscriptSessionArchive? {
+        val startedAtMillis = active.startedAtMillis ?: return null
+        if (active.ephemeral) return null
+        val transcript = commandTranscript.state.value.let { state ->
+            if (completedOnly) {
+                state.copy(
+                    turns = state.turns.filter { it.completedAtMillis != null },
+                    activeCommandId = null,
+                )
+            } else {
+                state
+            }
+        }
+        if (transcript.turns.isEmpty()) return null
+        return TranscriptSessionArchive(
+            id = active.id,
+            profile = active.profile,
+            startedAtMillis = startedAtMillis,
+            savedAtMillis = clockMillis(),
+            transcript = transcript,
+        )
+    }
+
+    private suspend fun persistTranscriptArchive(archive: TranscriptSessionArchive?) =
+        transcriptSaveMutex.withLock { saveTranscriptArchive(archive) }
 
-    private suspend fun persistTranscriptArchive(archive: TranscriptSessionArchive?) {
+    private suspend fun saveTranscriptArchive(archive: TranscriptSessionArchive?) {
         if (archive == null) return
         try {
             transcriptArchiveSink.save(archive)